🏷️ add types to tests

Author: techouse

requirements_dev.txt

@@ -13,12 +13,16 @@ python-slugify>=7.0.0
 types-python-slugify
 simplejson>=3.19.1
 types-simplejson
-sqlalchemy<2.0.0
+sqlalchemy>=2.0.0
 sqlalchemy-utils
+types-sqlalchemy-utils
 tox
 tqdm>=4.65.0
 types-tqdm
 packaging>=23.1
 tabulate
 types-tabulate
-typing_extensions
+typing_extensions
+requests
+types-requests
+mypy>=1.3.0

tests/conftest.py

@@ -1,28 +1,38 @@
 import json
+import os
 import socket
+import typing as t
 from codecs import open
-from collections import namedtuple
-from contextlib import closing, contextmanager
+from contextlib import contextmanager
 from os.path import abspath, dirname, isfile, join
+from pathlib import Path
 from random import choice
 from string import ascii_lowercase, ascii_uppercase, digits
 from time import sleep
 
 import docker
 import mysql.connector
 import pytest
+from _pytest._py.path import LocalPath
+from _pytest.config import Config
+from _pytest.config.argparsing import Parser
+from _pytest.legacypath import TempdirFactory
 from click.testing import CliRunner
+from docker import DockerClient
 from docker.errors import NotFound
-from mysql.connector import errorcode
+from docker.models.containers import Container
+from faker import Faker
+from mysql.connector import CMySQLConnection, MySQLConnection, errorcode
+from mysql.connector.pooling import PooledMySQLConnection
 from requests import HTTPError
 from sqlalchemy.exc import IntegrityError
+from sqlalchemy.orm import Session
 from sqlalchemy_utils import database_exists, drop_database
 
-from .database import Database
-from .factories import ArticleFactory, AuthorFactory, CrazyNameFactory, ImageFactory, MiscFactory, TagFactory
+from . import database, factories, models
 
 
-def pytest_addoption(parser):
+def pytest_addoption(parser: "Parser"):
     parser.addoption(
         "--mysql-user",
         dest="mysql_user",
@@ -78,9 +88,9 @@ def pytest_addoption(parser):
 
 
 @pytest.fixture(scope="session", autouse=True)
-def cleanup_hanged_docker_containers():
+def cleanup_hanged_docker_containers() -> None:
     try:
-        client = docker.from_env()
+        client: DockerClient = docker.from_env()
         for container in client.containers.list():
             if container.name == "pytest_mysql_to_sqlite3":
                 container.kill()
@@ -89,9 +99,9 @@ def cleanup_hanged_docker_containers():
         pass
 
 
-def pytest_keyboard_interrupt():
+def pytest_keyboard_interrupt() -> None:
     try:
-        client = docker.from_env()
+        client: DockerClient = docker.from_env()
         for container in client.containers.list():
             if container.name == "pytest_mysql_to_sqlite3":
                 container.kill()
@@ -103,17 +113,17 @@ def pytest_keyboard_interrupt():
 class Helpers:
     @staticmethod
     @contextmanager
-    def not_raises(exception):
+    def not_raises(exception: t.Type[Exception]) -> t.Generator:
         try:
             yield
         except exception:
             raise pytest.fail("DID RAISE {0}".format(exception))
 
     @staticmethod
     @contextmanager
-    def session_scope(db):
+    def session_scope(db: database.Database) -> t.Generator:
         """Provide a transactional scope around a series of operations."""
-        session = db.Session()
+        session: Session = db.Session()
         try:
             yield session
             session.commit()
@@ -125,29 +135,37 @@ def session_scope(db):
 
 
 @pytest.fixture
-def helpers():
+def helpers() -> t.Type[Helpers]:
     return Helpers
 
 
 @pytest.fixture()
-def sqlite_database(tmpdir):
-    db_name = "".join(choice(ascii_uppercase + ascii_lowercase + digits) for _ in range(32))
-    return str(tmpdir.join("{}.sqlite3".format(db_name)))
+def sqlite_database(tmpdir: LocalPath) -> t.Union[str, Path, "os.PathLike[t.Any]"]:
+    db_name: str = "".join(choice(ascii_uppercase + ascii_lowercase + digits) for _ in range(32))
+    return Path(tmpdir.join(Path("{}.sqlite3".format(db_name))))
 
 
-def is_port_in_use(port, host="0.0.0.0"):
+def is_port_in_use(port: int, host: str = "0.0.0.0") -> bool:
     with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
         return s.connect_ex((host, port)) == 0
 
 
-@pytest.fixture(scope="session")
-def mysql_credentials(pytestconfig):
-    MySQLCredentials = namedtuple("MySQLCredentials", ["user", "password", "host", "port", "database"])
+class MySQLCredentials(t.NamedTuple):
+    """MySQL credentials."""
 
-    db_credentials_file = abspath(join(dirname(__file__), "db_credentials.json"))
+    user: str
+    password: str
+    host: str
+    port: int
+    database: str
+
+
+@pytest.fixture(scope="session")
+def mysql_credentials(pytestconfig: Config) -> MySQLCredentials:
+    db_credentials_file: str = abspath(join(dirname(__file__), "db_credentials.json"))
     if isfile(db_credentials_file):
         with open(db_credentials_file, "r", "utf-8") as fh:
-            db_credentials = json.load(fh)
+            db_credentials: t.Dict[str, t.Any] = json.load(fh)
             return MySQLCredentials(
                 user=db_credentials["mysql_user"],
                 password=db_credentials["mysql_password"],
@@ -156,16 +174,13 @@ def mysql_credentials(pytestconfig):
                 port=db_credentials["mysql_port"],
             )
 
-    port = pytestconfig.getoption("mysql_port") or 3306
+    port: int = pytestconfig.getoption("mysql_port") or 3306
     if pytestconfig.getoption("use_docker"):
         while is_port_in_use(port, pytestconfig.getoption("mysql_host")):
             if port >= 2**16 - 1:
                 pytest.fail(
                     "No ports appear to be available on the host {}".format(pytestconfig.getoption("mysql_host"))
                 )
-                raise ConnectionError(
-                    "No ports appear to be available on the host {}".format(pytestconfig.getoption("mysql_host"))
-                )
             port += 1
 
     return MySQLCredentials(
@@ -178,11 +193,11 @@ def mysql_credentials(pytestconfig):
 
 
 @pytest.fixture(scope="session")
-def mysql_instance(mysql_credentials, pytestconfig):
-    container = None
-    mysql_connection = None
-    mysql_available = False
-    mysql_connection_retries = 15  # failsafe
+def mysql_instance(mysql_credentials: MySQLCredentials, pytestconfig: Config) -> t.Iterator[MySQLConnection]:
+    container: t.Optional[Container] = None
+    mysql_connection: t.Optional[t.Union[PooledMySQLConnection, MySQLConnection, CMySQLConnection]] = None
+    mysql_available: bool = False
+    mysql_connection_retries: int = 15  # failsafe
 
     db_credentials_file = abspath(join(dirname(__file__), "db_credentials.json"))
     if isfile(db_credentials_file):
@@ -198,7 +213,6 @@ def mysql_instance(mysql_credentials, pytestconfig):
             client = docker.from_env()
         except Exception as err:
             pytest.fail(str(err))
-            raise
 
         docker_mysql_image = pytestconfig.getoption("docker_mysql_image") or "mysql:latest"
 
@@ -208,7 +222,6 @@ def mysql_instance(mysql_credentials, pytestconfig):
                 client.images.pull(docker_mysql_image)
             except (HTTPError, NotFound) as err:
                 pytest.fail(str(err))
-                raise
 
         container = client.containers.run(
             image=docker_mysql_image,
@@ -256,17 +269,22 @@ def mysql_instance(mysql_credentials, pytestconfig):
         if not mysql_available and mysql_connection_retries <= 0:
             raise ConnectionAbortedError("Maximum MySQL connection retries exhausted! Are you sure MySQL is running?")
 
-    yield
+    yield  # type: ignore[misc]
 
-    if use_docker:
+    if use_docker and container is not None:
         container.kill()
 
 
 @pytest.fixture(scope="session")
-def mysql_database(tmpdir_factory, mysql_instance, mysql_credentials, _session_faker):
-    temp_image_dir = tmpdir_factory.mktemp("images")
-
-    db = Database(
+def mysql_database(
+    tmpdir_factory: TempdirFactory,
+    mysql_instance: MySQLConnection,
+    mysql_credentials: MySQLCredentials,
+    _session_faker: Faker,
+) -> t.Iterator[database.Database]:
+    temp_image_dir: LocalPath = tmpdir_factory.mktemp("images")
+
+    db: database.Database = database.Database(
         "mysql+mysqldb://{user}:{password}@{host}:{port}/{database}".format(
             user=mysql_credentials.user,
             password=mysql_credentials.password,
@@ -278,13 +296,13 @@ def mysql_database(tmpdir_factory, mysql_instance, mysql_credentials, _session_f
 
     with Helpers.session_scope(db) as session:
         for _ in range(_session_faker.pyint(min_value=12, max_value=24)):
-            article = ArticleFactory()
-            article.authors.append(AuthorFactory())
-            article.tags.append(TagFactory())
-            article.misc.append(MiscFactory())
+            article: models.Article = factories.ArticleFactory()
+            article.authors.append(factories.AuthorFactory())
+            article.tags.append(factories.TagFactory())
+            article.misc.append(factories.MiscFactory())
             for _ in range(_session_faker.pyint(min_value=1, max_value=4)):
                 article.images.append(
-                    ImageFactory(
+                    factories.ImageFactory(
                         path=join(
                             str(temp_image_dir),
                             _session_faker.year(),
@@ -297,7 +315,7 @@ def mysql_database(tmpdir_factory, mysql_instance, mysql_credentials, _session_f
             session.add(article)
 
         for _ in range(_session_faker.pyint(min_value=12, max_value=24)):
-            session.add(CrazyNameFactory())
+            session.add(factories.CrazyNameFactory())
         try:
             session.commit()
         except IntegrityError:
@@ -310,5 +328,5 @@ def mysql_database(tmpdir_factory, mysql_instance, mysql_credentials, _session_f
 
 
 @pytest.fixture()
-def cli_runner():
+def cli_runner() -> t.Iterator[CliRunner]:
     yield CliRunner()

tests/database.py

@@ -1,17 +1,19 @@
+import typing as t
 from datetime import datetime, timedelta
 from decimal import Decimal
 
 import simplejson as json
 from sqlalchemy import create_engine
+from sqlalchemy.engine import Engine
 from sqlalchemy.orm import sessionmaker
 from sqlalchemy_utils import create_database, database_exists
 
 from .models import Base
 
 
 class Database:
-    engine = None
-    Session = None
+    engine: Engine
+    Session: sessionmaker
 
     def __init__(self, database_uri):
         self.Session = sessionmaker()
@@ -21,15 +23,15 @@ def __init__(self, database_uri):
         self._create_db_tables()
         self.Session.configure(bind=self.engine)
 
-    def _create_db_tables(self):
+    def _create_db_tables(self) -> None:
         Base.metadata.create_all(self.engine)
 
     @classmethod
-    def dumps(cls, data):
+    def dumps(cls, data: t.Any) -> str:
         return json.dumps(data, default=cls.json_serializer)
 
     @staticmethod
-    def json_serializer(data):
+    def json_serializer(data: t.Any) -> t.Optional[str]:
         if isinstance(data, datetime):
             return data.isoformat()
         if isinstance(data, Decimal):
@@ -38,3 +40,4 @@ def json_serializer(data):
             hours, remainder = divmod(data.total_seconds(), 3600)
             minutes, seconds = divmod(remainder, 60)
             return "{:02}:{:02}:{:02}".format(int(hours), int(minutes), int(seconds))
+        return None