diff options
Diffstat (limited to 'tests/conftest.py')
| -rw-r--r-- | tests/conftest.py | 153 |
1 files changed, 143 insertions, 10 deletions
diff --git a/tests/conftest.py b/tests/conftest.py index 457f6a3..ca6661a 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -3,15 +3,17 @@ from __future__ import annotations import os import subprocess import sys -from collections.abc import Callable +from collections.abc import Callable, Generator +from datetime import UTC, datetime from pathlib import Path from typing import TYPE_CHECKING from uuid import uuid4 import pytest -from generalresearch.models.custom_types import PostgresDict +from generalresearch.models.custom_types import InternalHostname, PostgresDict from generalresearch.pg_helper import PostgresConfig, PostgresDsn from mypy_boto3_mturk import MTurkClient +from pydantic import TypeAdapter from pytest import TempPathFactory from jb.decorators import CLIENT_CONFIG @@ -22,7 +24,6 @@ if TYPE_CHECKING: pytest_plugins = [ - "test_utils.conftest", "tests.fixtures.amt", "tests.fixtures.flow", "tests.fixtures.http", @@ -94,9 +95,139 @@ def settings() -> Settings: @pytest.fixture(scope="session") +def postgres_instance(settings: Settings) -> Generator[PostgresDsn]: + """Create a ephemeral postgresql instance for us to use during pytest. + + This does not create any tables, or schema definitions within the instance. + What this does is simply: + + 1. Create a database on a known, consistent, staging or unittest + defined Postgres server. + + 2. Return the PostgresDsn of that table + + 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 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 + + +@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, + ) + + +@pytest.fixture(scope="session") +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, - gr_repo: Callable[..., Path], django_settings_file: Callable[..., tuple[str, Path]], postgres_instance_dict: PostgresDict, tmp_path_factory: TempPathFactory, @@ -105,7 +236,7 @@ def django_db_factory( _ran = {} def _inner( - django_project: str = "generalresearch.thl_django", + django_project: str = "carer.mtwerk", ) -> PostgresDsn | None: if _ran.get(django_project, False): @@ -117,7 +248,7 @@ def django_db_factory( _manage_path = "generalresearch.thl_django.app.manage" _settings_module, _settings_dir = django_settings_file( extra_installed_apps=[ - "generalresearch.thl_django", + "carer.mtwerk", ], ) @@ -177,11 +308,13 @@ def django_db_factory( @pytest.fixture(scope="session") -def pg_config(settings: Settings) -> PostgresConfig: +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, ) @@ -193,7 +326,7 @@ def redis(settings: Settings): from generalresearch.redis_helper import RedisConfig redis_config = RedisConfig( - dsn=settings.redis, + dsn=settings.testing_redis, decode_responses=True, socket_timeout=settings.redis_timeout, socket_connect_timeout=settings.redis_timeout, |