diff options
Diffstat (limited to 'test_utils/conftest.py')
| -rw-r--r-- | test_utils/conftest.py | 401 |
1 files changed, 167 insertions, 234 deletions
diff --git a/test_utils/conftest.py b/test_utils/conftest.py index 232c1fc..378b9cc 100644 --- a/test_utils/conftest.py +++ b/test_utils/conftest.py @@ -1,34 +1,31 @@ +from __future__ import annotations + import os import shutil +import stat +import subprocess import sys +import tempfile 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 typing import 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 redis import Redis +from pydantic import MariaDBDsn, PostgresDsn, TypeAdapter +from generalresearch.config import GRLBaseSettings +from generalresearch.currency import USDCent +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 -if TYPE_CHECKING: - from generalresearch.config import GRLBaseSettings - from generalresearch.currency import USDCent - from generalresearch.models.thl.session import Status - @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 +37,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: @@ -68,17 +63,31 @@ def settings(env_file_path: str) -> "GRLBaseSettings": @pytest.fixture(scope="session") -def postgres_instance(settings: "GRLBaseSettings") -> Generator[PostgresDsn]: +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,138 +95,167 @@ 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, + ) - def _inner(): - assert settings.thl_web_rw_db - dsn: PostgresDsn = settings.thl_web_rw_db - host = dsn.hosts()[0] +@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 - # 1. Bootstrapping Django settings - if not django_settings.configured: - django_settings.configure( - 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", - } - }, - INSTALLED_APPS=[ - "django.contrib.postgres", - "django.contrib.contenttypes", - "generalresearch.thl_django", - ], - ) - django.setup() - from django.apps import apps +# @pytest.fixture(scope="session") +# def git_key_path(settings: GRLBaseSettings) -> Path: +# return Path('/tmp/') - for model in apps.get_models(): - print(f"Discovered model: {model._meta.label}") - # 2. Run migrations directly during fixture activation - call_command("migrate") +@pytest.fixture(scope="session") +def git_key_path( + settings: GRLBaseSettings, +) -> Generator[Path]: - return _inner + assert settings.git_creds + with tempfile.NamedTemporaryFile(mode="w", delete=False, suffix="_id_rsa") as f: + f.write(settings.git_creds) + key_path = f.name + os.chmod(key_path, stat.S_IRUSR | stat.S_IWUSR) -@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 + yield Path(key_path) + + os.unlink(key_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}" +@pytest.fixture(scope="session") +def gr_repo(git_key_path: Path) -> Callable[..., Path]: + repo_url = "ssh://code.g-r-l.com/general-research/gr-carer.git" + repo_path = Path("/tmp/gr-carer") + + def _inner() -> Path: + ssh_cmd = ( + f"ssh -i {git_key_path} " + "-o IdentitiesOnly=yes " + "-o StrictHostKeyChecking=no " # or accept-new, see note below + ) + env = {"GIT_SSH_COMMAND": ssh_cmd} + + if repo_path.exists(): + subprocess.run(["git", "-C", str(repo_path), "pull"], check=True, env=env) + else: + subprocess.run( + ["git", "clone", "--depth", "1", repo_url, str(repo_path)], + check=True, + env=env, + ) - # Run Migrations now. - django_db_setup() + return repo_path - return PostgresConfig( - dsn=PostgresDsn(db_url), - connect_timeout=1, - statement_timeout=5, - ) + return _inner @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 django_db_factory( + postgres_instance: PostgresDsn, + postgres_instance_dict: PostgresDict, + gr_repo: Callable[..., Path], +) -> Callable[..., PostgresDsn]: - if dsn.path not in ["/", "/postgres"]: - assert "/unittest-" in dsn.path + import django + from django.conf import settings as django_settings + from django.core.management import call_command - db_path = postgres_instance.path - host = dsn.hosts()[0] - db_url = f"postgres://{host['username']}:{host['password']}@{host['host']}{db_path}" + def _inner(django_project: str = "generalresearch.thl_django"): - # Run Migrations now. - django_db_setup() + if "gr" in django_project: + # We need model files that are NOT in this repo. + gr_path = gr_repo() + sys.path.insert(0, str(gr_path)) - return PostgresConfig( - dsn=PostgresDsn(db_url), - connect_timeout=1, - statement_timeout=5, - ) + print(sys.path) + # 1. Bootstrapping Django settings + if not django_settings.configured: + django_settings.configure( + DATABASES={ + "default": { + "ENGINE": "django.db.backends.postgresql", + "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", + django_project, + ], + ) + django.setup() -@pytest.fixture(scope="session") -def gr_db(settings: "GRLBaseSettings") -> PostgresConfig: - dsn = settings.gr_db - assert dsn - assert dsn.path + # for model in apps.get_models(): + # print(f"Discovered model: {model._meta.label}") - if dsn.path not in ["/", "/postgres"]: - assert "/unittest-" in dsn.path + # 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 PostgresConfig(dsn=settings.gr_db, connect_timeout=5, statement_timeout=2) + return _inner @pytest.fixture(scope="session") -def spectrum_rw(settings: "GRLBaseSettings") -> SqlHelper: +def spectrum_rw(settings: GRLBaseSettings) -> SqlHelper: dsn = settings.spectrum_rw_db assert dsn assert dsn.path @@ -233,114 +271,15 @@ 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? - assert "unittest" in str(settings.thl_redis) or "127.0.0.1" in str( - settings.thl_redis - ) - - return redis.Redis.from_url( - **{ - "url": str(settings.thl_redis), - "decode_responses": True, - "socket_timeout": settings.redis_timeout, - "socket_connect_timeout": settings.redis_timeout, - } - ) - - -@pytest.fixture(scope="session") -def thl_redis_config(settings: "GRLBaseSettings") -> RedisConfig: - assert "unittest" in str(settings.thl_redis) or "127.0.0.1" in str( - settings.thl_redis - ) - return RedisConfig( - dsn=settings.thl_redis, - decode_responses=True, - socket_timeout=settings.redis_timeout, - socket_connect_timeout=settings.redis_timeout, - ) - - -@pytest.fixture(scope="session") -def gr_redis_config(settings: "GRLBaseSettings") -> "RedisConfig": - assert "unittest" in str(settings.gr_redis) or "127.0.0.1" in str(settings.gr_redis) - - return RedisConfig( - dsn=settings.gr_redis, - decode_responses=True, - socket_timeout=settings.redis_timeout, - socket_connect_timeout=settings.redis_timeout, - ) - - -@pytest.fixture(scope="session") -def gr_redis(settings: "GRLBaseSettings") -> "Redis": - assert "unittest" in str(settings.gr_redis) or "127.0.0.1" in str(settings.gr_redis) - return redis.Redis.from_url( - **{ - "url": str(settings.gr_redis), - "decode_responses": True, - "socket_timeout": settings.redis_timeout, - "socket_connect_timeout": settings.redis_timeout, - } - ) - - -@pytest.fixture -def gr_redis_async(settings: "GRLBaseSettings"): - assert "unittest" in str(settings.gr_redis) or "127.0.0.1" in str(settings.gr_redis) - - import redis.asyncio as redis_async - - return redis_async.Redis.from_url( - str(settings.gr_redis), - decode_responses=True, - socket_timeout=0.20, - socket_connect_timeout=0.20, - ) - - # === Random helpers === @pytest.fixture -def start() -> "datetime": - from datetime import datetime, timezone - +def start() -> datetime: return datetime(year=1900, month=1, day=1, tzinfo=timezone.utc) @pytest.fixture -def wall_status(request) -> "Status": - from generalresearch.models.thl.session import Status - - return request.param if hasattr(request, "wall_status") else Status.COMPLETE - - -@pytest.fixture def utc_now() -> datetime: return datetime.now(tz=timezone.utc) @@ -351,30 +290,22 @@ def utc_hour_ago() -> datetime: @pytest.fixture -def utc_day_ago() -> "datetime": - from datetime import datetime, timedelta, timezone - +def utc_day_ago() -> datetime: return datetime.now(tz=timezone.utc) - timedelta(hours=24) @pytest.fixture -def utc_90days_ago() -> "datetime": - from datetime import datetime, timedelta, timezone - +def utc_90days_ago() -> datetime: return datetime.now(tz=timezone.utc) - timedelta(days=90) @pytest.fixture -def utc_60days_ago() -> "datetime": - from datetime import datetime, timedelta, timezone - +def utc_60days_ago() -> datetime: return datetime.now(tz=timezone.utc) - timedelta(days=60) @pytest.fixture -def utc_30days_ago() -> "datetime": - from datetime import datetime, timedelta, timezone - +def utc_30days_ago() -> datetime: return datetime.now(tz=timezone.utc) - timedelta(days=30) @@ -424,6 +355,8 @@ def delete_df_collection( ) case _: + assert coll.data_type + thl_web_rw.execute_write( query=f"DELETE FROM {coll.data_type.value};", ) @@ -435,23 +368,23 @@ def delete_df_collection( @pytest.fixture(scope="function") -def amount_1(request) -> "USDCent": - from generalresearch.currency import USDCent - +def amount_1() -> USDCent: return USDCent(1) @pytest.fixture(scope="function") -def amount_100(request) -> "USDCent": - from generalresearch.currency import USDCent - +def amount_100() -> USDCent: return USDCent(100) -def clear_directory(path: Path): - for entry in os.listdir(path): +def clear_directory(path: Path | str): + dir_path = Path(path) + + for entry in os.listdir(dir_path): + full_path = os.path.join(path, entry) if os.path.isfile(full_path) or os.path.islink(full_path): os.unlink(full_path) # remove file or symlink + elif os.path.isdir(full_path): shutil.rmtree(full_path) # remove folder |