diff options
Diffstat (limited to 'tests/conftest.py')
| -rw-r--r-- | tests/conftest.py | 301 |
1 files changed, 272 insertions, 29 deletions
diff --git a/tests/conftest.py b/tests/conftest.py index 25eb457..7a74fa0 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,13 +1,28 @@ +from __future__ import annotations + import os -from typing import TYPE_CHECKING +import subprocess +import sys +from collections.abc import Callable, Generator +from datetime import UTC, datetime +from pathlib import Path +from random import randint +from typing import TYPE_CHECKING, Any from uuid import uuid4 -from dotenv import load_dotenv + import pytest -from generalresearchutils.pg_helper import PostgresConfig -from tests import generate_amt_id -from _pytest.config import Config -from jb.decorators import CLIENT_CONFIG +import redis +from fastapi.testclient import TestClient +from generalresearch.models.custom_types import InternalHostname, PostgresDict +from generalresearch.pg_helper import PostgresConfig, PostgresDsn +from generalresearch.redis_helper import RedisConfig from mypy_boto3_mturk import MTurkClient +from pydantic import TypeAdapter +from pytest import TempPathFactory + +from jb.decorators import CLIENT_CONFIG, get_redis_config +from jb.main import app +from tests import generate_amt_id if TYPE_CHECKING: from jb.settings import Settings @@ -75,53 +90,281 @@ def pe_id() -> str: @pytest.fixture(scope="session") -def env_file_path(pytestconfig: Config) -> str: - root_path = pytestconfig.rootpath - env_path = os.path.join(root_path, ".env.test") +def settings() -> Settings: + from jb.settings import Settings as JBSettings - if os.path.exists(env_path): - load_dotenv(dotenv_path=env_path, override=True) + return JBSettings() - return env_path + +# --- Database Connectors --- @pytest.fixture(scope="session") -def settings(env_file_path: str) -> "Settings": - from jb.settings import Settings as JBSettings +def postgres_instance(settings: Settings) -> Generator[PostgresDsn]: + """Create a ephemeral postgresql instance for us to use during pytest. - s = JBSettings(_env_file=env_file_path) + This does not create any tables, or schema definitions within the instance. + What this does is simply: - return s + 1. Create a database on a known, consistent, staging or unittest + defined Postgres server. + 2. Return the PostgresDsn of that table -# --- Database Connectors --- + 3. On shutdown, go ahead and delete that database after the + tests have finished. + """ + + msg = "Must define Postgres test settings" + assert settings.testing_postgres, msg + assert settings.testing_postgres_user, msg + assert settings.testing_postgres_pass, msg + + db_uri, db_user, db_pass = ( + settings.testing_postgres, + settings.testing_postgres_user, + settings.testing_postgres_pass, + ) + + # Connect to default DB to create the new one + from psycopg import connect + from psycopg.sql import SQL, Identifier + + now = datetime.now(UTC) + ts: str = now.strftime("%Y-%m-%d") + db_name = f"unittest-{ts}-{uuid4().hex[:6]}" + + db_path_connect = f"postgres://{db_user}:{db_pass}@{db_uri}" + db_path = f"{db_path_connect}/{db_name}" + + # The DATABASE does NOT yet exist on the Postgres SERVER, thus + # we first must connect only to the SERVER (eg: default postgres path used) + conn = connect(f"{db_path_connect}/postgres") + conn.autocommit = True + cur = conn.cursor() + cur.execute(SQL("CREATE DATABASE {}").format(Identifier(db_name))) + cur.close() + conn.close() + + yield PostgresDsn(db_path) + + # Teardown: drop the DB after the session + conn = connect(f"{db_path_connect}/postgres") + conn.autocommit = True + cur = conn.cursor() + cur.execute(SQL("DROP DATABASE {} WITH (FORCE)").format(Identifier(db_name))) + cur.close() + conn.close() @pytest.fixture(scope="session") -def redis(settings: "Settings"): - from generalresearchutils.redis_helper import RedisConfig +def django_settings_file( + postgres_instance_dict: PostgresDict, + tmp_path_factory: TempPathFactory, +) -> Callable[..., tuple[str, Path]]: + + def _inner(extra_installed_apps: list[str] | None = None) -> tuple[str, Path]: + installed_apps = [ + "django.contrib.postgres", + "django.contrib.contenttypes", + ] + (extra_installed_apps or []) + + settings_dir = tmp_path_factory.mktemp("django-settings") + settings_module = "test_settings" + + settings_content = f"""DATABASES = {{ + "default": {{ + "ENGINE": "django.db.backends.postgresql", + "NAME": {postgres_instance_dict["name"]!r}, + "USER": {postgres_instance_dict["username"]!r}, + "PASSWORD": {postgres_instance_dict["password"]!r}, + "HOST": {postgres_instance_dict["host"]!r}, + "PORT": {postgres_instance_dict["port"]!r}, + }} +}} +INSTALLED_APPS = {installed_apps!r} +DEFAULT_AUTO_FIELD = "django.db.models.BigAutoField" +LANGUAGE_CODE = "en-us" +TIME_ZONE = "UTC" +USE_I18N = True +USE_L10N = True +USE_TZ = True +""" + settings_file_path = settings_dir / f"{settings_module}.py" + settings_file_path.write_text(settings_content, encoding="utf-8") + + return settings_module, settings_dir + + return _inner - redis_config = RedisConfig( - dsn=settings.redis, - decode_responses=True, - socket_timeout=settings.redis_timeout, - socket_connect_timeout=settings.redis_timeout, + +@pytest.fixture(scope="session") +def postgres_instance_dict( + postgres_instance: PostgresDsn, +) -> Generator[PostgresDict]: + host = postgres_instance.hosts()[0] + assert host is not None + + msg = "Must have full Postgres details" + assert host["host"], msg + assert host["username"], msg + assert host["password"], msg + + assert postgres_instance.path + + yield PostgresDict( + username=host["username"], + password=host["password"], + host=host["host"], + name=postgres_instance.path.lstrip("/"), + port=5432, ) - return redis_config.create_redis_client() @pytest.fixture(scope="session") -def pg_config(settings: "Settings") -> PostgresConfig: +def postgres_instance_host( + postgres_instance_dict: PostgresDict, +) -> Generator[InternalHostname]: + adapter = TypeAdapter(InternalHostname) + value = adapter.validate_python(postgres_instance_dict["host"]) + yield value + + +@pytest.fixture(scope="session") +def django_db_factory( + postgres_instance: PostgresDsn, + django_settings_file: Callable[..., tuple[str, Path]], + postgres_instance_dict: PostgresDict, + tmp_path_factory: TempPathFactory, +) -> Callable[..., PostgresDsn | None]: + + _ran = {} + + def _inner( + django_project: str = "carer.mtwerk", + ) -> PostgresDsn | None: + + if _ran.get(django_project, False): + print(f"Already ran django_db_factory:{django_project}") + return postgres_instance + _ran[django_project] = True + + _cwd = None + _manage_path = "generalresearch.thl_django.app.manage" + _settings_module, _settings_dir = django_settings_file( + extra_installed_apps=[ + "carer.mtwerk", + ], + ) + + pythonpath = str(_settings_dir) + if existing_pythonpath := os.environ.get("PYTHONPATH"): + pythonpath += os.pathsep + existing_pythonpath + + env = { + **os.environ, + "DJANGO_SETTINGS_MODULE": _settings_module, + "PYTHONPATH": pythonpath, + } + + # we check right after. if we check now, we won't print if bad + res1 = subprocess.run( # noqa: PLW1510 + [ + sys.executable, + "-m", + _manage_path, + "makemigrations", + f"--settings={_settings_module}", + ], + cwd=str(_cwd) if _cwd is not None else None, + env=env, + capture_output=True, + text=True, + ) + + if res1.returncode != 0: + print("STDOUT:", res1.stdout) + print("STDERR:", res1.stderr) + res1.check_returncode() + + res2 = subprocess.run( # noqa: PLW1510 + [ + sys.executable, + "-m", + _manage_path, + "migrate", + f"--settings={_settings_module}", + ], + env=env, + cwd=str(_cwd) if _cwd is not None else None, + capture_output=True, + text=True, + ) + + if res2.returncode != 0: + print("STDOUT:", res2.stdout) + print("STDERR:", res2.stderr) + res2.check_returncode() + + # 3. Return the Dsn so the factory gives a way to connect + return postgres_instance + + return _inner + + +@pytest.fixture(scope="session") +def pg_config(django_db_factory: Callable[..., PostgresDsn]) -> PostgresConfig: + _dsn = django_db_factory() + return PostgresConfig( - dsn=settings.amt_jb_db, + dsn=_dsn, connect_timeout=1, - statement_timeout=1, + statement_timeout=5, ) +# --- Redis --- + + +@pytest.fixture(scope="session") +def redis_config_db() -> str: + # need to update 'databases' in /etc/redis/redis.conf + # or this won't work and you'll have no indication why ... + return str(randint(99, 1_023)) + + +@pytest.fixture(scope="session") +def redis_config(settings: Settings, redis_config_db: str) -> Generator[RedisConfig]: + assert "unittest" in str(settings.testing_redis) or "127.0.0.1" in str( + settings.testing_redis + ) + + uri = f"redis://{settings.testing_redis}/{redis_config_db}" + + res = subprocess.run( + ["redis-cli", "-u", uri, "SET", "jenkins_lock", "1", "NX", "EX", "3600"], + check=True, + text=True, + capture_output=True, + ) + + if res.stdout.strip() != "OK": + raise ValueError("Redis already locked... aborting.") + + yield RedisConfig( + dsn=uri, + decode_responses=True, + socket_timeout=settings.redis_timeout, + socket_connect_timeout=settings.redis_timeout, + ) + + r = redis.from_url(uri) + r.flushdb() + + # --- Connectors --- @pytest.fixture(scope="session") -def amt_client(settings: "Settings") -> MTurkClient: +def amt_client(settings: Settings) -> MTurkClient: import boto3 client = boto3.client( |