path: root/tests/conftest.py

from __future__ import annotations

import os
import subprocess
import sys
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 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
from tests import generate_amt_id

if TYPE_CHECKING:
    from jb.settings import Settings


pytest_plugins = [
    "tests.fixtures.amt",
    "tests.fixtures.flow",
    "tests.fixtures.http",
    "tests.fixtures.managers",
    "tests.fixtures.models",
]


# --- IDs and Identifiers ---


@pytest.fixture
def amt_hit_id() -> str:
    return generate_amt_id()


@pytest.fixture
def amt_hit_type_id() -> str:
    return generate_amt_id()


@pytest.fixture
def amt_assignment_id() -> str:
    return generate_amt_id()


@pytest.fixture
def amt_worker_id() -> str:
    return generate_amt_id(length=21)


@pytest.fixture
def amt_group_id() -> str:
    return generate_amt_id()


@pytest.fixture
def tsid() -> str:
    return uuid4().hex


@pytest.fixture
def tsid1() -> str:
    return uuid4().hex


@pytest.fixture
def tsid2() -> str:
    return uuid4().hex


@pytest.fixture
def pe_id() -> str:
    # payout event / cashout request UUID
    return uuid4().hex


# --- Settings ---


@pytest.fixture(scope="session")
def settings() -> Settings:
    from jb.settings import Settings as JBSettings

    return JBSettings()


# --- Database Connectors ---


@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,
    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=_dsn,
        connect_timeout=1,
        statement_timeout=5,
    )


# --- Redis ---


@pytest.fixture(scope="session")
def redis(settings: Settings):
    from generalresearch.redis_helper import RedisConfig

    redis_config = RedisConfig(
        dsn=settings.testing_redis,
        decode_responses=True,
        socket_timeout=settings.redis_timeout,
        socket_connect_timeout=settings.redis_timeout,
    )
    return redis_config.create_redis_client()


# --- Connectors ---
@pytest.fixture(scope="session")
def amt_client(settings: Settings) -> MTurkClient:
    import boto3

    client = boto3.client(
        service_name="mturk",
        region_name="us-east-1",
        endpoint_url=str(settings.amt_endpoint),
        aws_access_key_id=settings.amt_access_id,
        aws_secret_access_key=settings.amt_secret_key,
        config=CLIENT_CONFIG,
    )

    # Confirm we're only using the Sandbox for any unittests
    assert "mturk-requester-sandbox" in client.meta.endpoint_url

    return client