diff options
Diffstat (limited to 'test_utils/conftest.py')
| -rw-r--r-- | test_utils/conftest.py | 205 |
1 files changed, 103 insertions, 102 deletions
diff --git a/test_utils/conftest.py b/test_utils/conftest.py index 232c1fc..9c80065 100644 --- a/test_utils/conftest.py +++ b/test_utils/conftest.py @@ -1,22 +1,20 @@ import os import shutil -import sys from datetime import datetime, timedelta, timezone from os.path import join as pjoin from pathlib import Path from typing import TYPE_CHECKING, Callable, Generator from uuid import uuid4 -import django import pytest import redis from _pytest.config import Config -from django.conf import settings as django_settings -from django.core.management import call_command from dotenv import load_dotenv -from pydantic import MariaDBDsn, PostgresDsn +from pydantic import MariaDBDsn, PostgresDsn, TypeAdapter +from pydantic_core import MultiHostHost from redis import Redis +from generalresearch.models.custom_types import InternalHostname, PostgresDict from generalresearch.pg_helper import PostgresConfig from generalresearch.redis_helper import RedisConfig from generalresearch.sql_helper import SqlHelper @@ -28,7 +26,7 @@ if TYPE_CHECKING: @pytest.fixture(scope="session") -def env_file_path(pytestconfig: Config) -> str: +def env_file_path(pytestconfig: Config) -> Path: root_path = pytestconfig.rootpath env_file = ".env.test" @@ -40,18 +38,16 @@ def env_file_path(pytestconfig: Config) -> str: for env_path in candidates: if os.path.exists(env_path): load_dotenv(dotenv_path=env_path, override=True) - return os.path.normpath(env_path) + return Path(os.path.normpath(env_path)) raise AssertionError(f"No .env.test file found in: {', '.join(candidates)}") @pytest.fixture(scope="session") -def settings(env_file_path: str) -> "GRLBaseSettings": +def settings(env_file_path: Path) -> "GRLBaseSettings": from generalresearch.config import GRLBaseSettings - print(f"{env_file_path=}") - - s = GRLBaseSettings(_env_file=env_file_path) + s = GRLBaseSettings() if s.thl_mkpl_rr_db is not None: if s.spectrum_rw_db is None: @@ -71,14 +67,28 @@ def settings(env_file_path: str) -> "GRLBaseSettings": def postgres_instance(settings: "GRLBaseSettings") -> Generator[PostgresDsn]: """Create a ephemeral postgresql instance for us to use during pytest. - This is simplified, and only based off a single host. We don't want to - create multiple migrated tmp databases for each rw/rr/ro connection + 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. """ - assert settings.thl_web_rw_db - # assert settings.thl_web_rw_db.host + msg = "Must define Postgres test settings" + assert settings.testing_postgres, msg + assert settings.testing_postgres_user, msg + assert settings.testing_postgres_pass, msg - dsn: PostgresDsn = settings.thl_web_rw_db + 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 @@ -86,40 +96,74 @@ def postgres_instance(settings: "GRLBaseSettings") -> Generator[PostgresDsn]: now = datetime.now(timezone.utc) ts: str = now.strftime("%Y-%m-%d") - db_name = f"unittest-{ts}-{uuid4().hex[:6]}" - print("XXX", str(dsn)) - conn = connect(str(dsn)) + + 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() - host = dsn.hosts()[0] - db_url = ( - f"postgres://{host['username']}:{host['password']}@{host['host']}/{db_name}" - ) - - yield PostgresDsn(db_url) + yield PostgresDsn(db_path) # Teardown: drop the DB after the session - conn = connect(str(dsn)) + conn = connect(f"{db_path_connect}/postgres") conn.autocommit = True cur = conn.cursor() - # cur.execute(SQL("DROP DATABASE {}").format(Identifier(db_name))) + cur.execute(SQL("DROP DATABASE {} WITH (FORCE)").format(Identifier(db_name))) cur.close() conn.close() @pytest.fixture(scope="session") -def django_db_setup(settings: "GRLBaseSettings") -> Callable[..., None]: +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, postgres_instance_dict: PostgresDict +) -> Callable[..., PostgresDsn]: - def _inner(): + import django + from django.apps import apps + from django.conf import settings as django_settings + from django.core.management import call_command - assert settings.thl_web_rw_db - dsn: PostgresDsn = settings.thl_web_rw_db - host = dsn.hosts()[0] + def _inner(django_project: str = "generalresearch.thl_django"): # 1. Bootstrapping Django settings if not django_settings.configured: @@ -127,93 +171,70 @@ def django_db_setup(settings: "GRLBaseSettings") -> Callable[..., None]: DATABASES={ "default": { "ENGINE": "django.db.backends.postgresql", - # PostgresDsn stores path as "/dbname" - "NAME": str(dsn.path).lstrip("/"), - "USER": host["username"], - "PASSWORD": host["password"], - "HOST": host["host"], - "PORT": "5432", + "NAME": postgres_instance_dict["name"], + "USER": postgres_instance_dict["username"], + "PASSWORD": postgres_instance_dict["password"], + "HOST": postgres_instance_dict["host"], + "PORT": postgres_instance_dict["port"], } }, INSTALLED_APPS=[ "django.contrib.postgres", "django.contrib.contenttypes", - "generalresearch.thl_django", + django_project, ], ) django.setup() - from django.apps import apps - - for model in apps.get_models(): - print(f"Discovered model: {model._meta.label}") + # for model in apps.get_models(): + # print(f"Discovered model: {model._meta.label}") # 2. Run migrations directly during fixture activation call_command("migrate") + # 3. Return the Dsn so the factory gives a way to connect + return postgres_instance + return _inner @pytest.fixture(scope="session") -def thl_web_rr( - settings: "GRLBaseSettings", postgres_instance: PostgresDsn, django_db_setup -) -> PostgresConfig: - dsn = settings.thl_web_rr_db - assert dsn - assert dsn.path - - if dsn.path not in ["/", "/postgres"]: - assert "/unittest-" in dsn.path - - db_path = postgres_instance.path - host = dsn.hosts()[0] - db_url = f"postgres://{host['username']}:{host['password']}@{host['host']}{db_path}" - - # Run Migrations now. - django_db_setup() +def thl_web_rr(django_db_factory: Callable[..., PostgresDsn]) -> PostgresConfig: return PostgresConfig( - dsn=PostgresDsn(db_url), + dsn=django_db_factory("generalresearch.thl_django"), connect_timeout=1, statement_timeout=5, ) @pytest.fixture(scope="session") -def thl_web_rw( - settings: "GRLBaseSettings", postgres_instance: PostgresDsn, django_db_setup -) -> PostgresConfig: - dsn = settings.thl_web_rw_db - assert dsn - assert dsn.path +def thl_web_rw(thl_web_rr: PostgresConfig) -> PostgresConfig: + return thl_web_rr - if dsn.path not in ["/", "/postgres"]: - assert "/unittest-" in dsn.path - db_path = postgres_instance.path - host = dsn.hosts()[0] - db_url = f"postgres://{host['username']}:{host['password']}@{host['host']}{db_path}" - - # Run Migrations now. - django_db_setup() +@pytest.fixture(scope="session") +def gr_db(django_db_factory: Callable[..., PostgresDsn]) -> PostgresConfig: return PostgresConfig( - dsn=PostgresDsn(db_url), + dsn=django_db_factory("gr_carer"), connect_timeout=1, statement_timeout=5, ) @pytest.fixture(scope="session") -def gr_db(settings: "GRLBaseSettings") -> PostgresConfig: - dsn = settings.gr_db - assert dsn - assert dsn.path +def grliq_db(postgres_instance: PostgresDsn) -> PostgresConfig: - if dsn.path not in ["/", "/postgres"]: - assert "/unittest-" in dsn.path + # test_words = {"localhost", "127.0.0.1", "unittest", "grliq-test"} + # assert any(w in str(postgres_config.dsn) for w in test_words), "check grliq postgres_config" + # assert "grliqdeceezpocymo" not in str(postgres_config.dsn), "check grliq postgres_config" - return PostgresConfig(dsn=settings.gr_db, connect_timeout=5, statement_timeout=2) + return PostgresConfig( + dsn=postgres_instance, + connect_timeout=1, + statement_timeout=5, + ) @pytest.fixture(scope="session") @@ -234,26 +255,6 @@ def spectrum_rw(settings: "GRLBaseSettings") -> SqlHelper: @pytest.fixture(scope="session") -def grliq_db(settings: "GRLBaseSettings") -> PostgresConfig: - dsn = settings.grliq_db - assert dsn - assert dsn.path - - if dsn.path not in ["/", "/postgres"]: - assert "/unittest-" in dsn.path - - # test_words = {"localhost", "127.0.0.1", "unittest", "grliq-test"} - # assert any(w in str(postgres_config.dsn) for w in test_words), "check grliq postgres_config" - # assert "grliqdeceezpocymo" not in str(postgres_config.dsn), "check grliq postgres_config" - - return PostgresConfig( - dsn=settings.grliq_db, - connect_timeout=2, - statement_timeout=2, - ) - - -@pytest.fixture(scope="session") def thl_redis(settings: "GRLBaseSettings") -> "Redis": # todo: this should get replaced with redisconfig (in most places) # I'm not sure where this would be? in the domain name? |