ss-tools/backend/tests/integration/conftest.py

# #region IntegrationTestConftest [C:3] [TYPE Module] [SEMANTICS test, conftest, testcontainers, postgres, integration]
# @BRIEF Shared pytest fixtures for integration tests using Testcontainers PostgreSQL.
# @RELATION BINDS_TO -> [TranslationJob]
# @RELATION BINDS_TO -> [TranslationRun]
# @RELATION BINDS_TO -> [TerminologyDictionary]
# @PRE Docker daemon is running and accessible.
# @POST PostgreSQL container is started, tables created, session available per test.
# @RATIONALE
#   Architecture:
#   - Testcontainers spins up a real PostgreSQL 16 container per session.
#   - Each test gets an isolated session with transaction rollback.
#   - FK constraints, JSON columns, indexes — all behave like production.
#
#   Why Testcontainers over SQLite:
#   - SQLite silently ignores FK violations in some edge cases.
#   - JSON column behavior differs (PostgreSQL JSONB vs SQLite JSON text).
#   - Index behavior, constraint names, and error messages differ.
#   - Production parity — catches bugs that SQLite tests miss.
#
# @REJECTED
#   Always-on PostgreSQL — requires manual setup, not portable.
#   Docker Compose for tests — slower startup, harder isolation.
#   Shared container across tests — state leakage between tests.
import os
import sys
import time
from pathlib import Path

import pytest


def pytest_collection_modifyitems(config, items):
    if not config.getoption("--run-integration"):
        skip_integration = pytest.mark.skip(reason="use --run-integration to run")
        for item in items:
            # Only skip items under tests/integration/ — the global hook
            # applies to ALL items, not just integration tests.
            if "/integration/" in item.nodeid:
                item.add_marker(skip_integration)


import requests
from sqlalchemy import create_engine, event, text
from sqlalchemy.orm import Session, sessionmaker

sys.path.insert(0, str(Path(__file__).parent.parent.parent / "src"))

from src.core.database import Base

# Import all models to ensure tables are created
from src.models.translate import (  # noqa: F401
    DictionaryEntry,
    MetricSnapshot,
    TerminologyDictionary,
    TranslationBatch,
    TranslationEvent,
    TranslationJob,
    TranslationJobDictionary,
    TranslationLanguage,
    TranslationPreviewLanguage,
    TranslationPreviewRecord,
    TranslationPreviewSession,
    TranslationRecord,
    TranslationRun,
    TranslationRunLanguageStats,
    TranslationSchedule,
)
from src.models.llm import (  # noqa: F401
    LLMProvider,
    ValidationPolicy,
    ValidationRun,
    ValidationRecord,
    ValidationSource,
)
from src.models.maintenance import (  # noqa: F401
    MaintenanceEvent,
    MaintenanceDashboardBanner,
    MaintenanceDashboardState,
    MaintenanceSettings,
    MaintenanceEventStatus,
    MaintenanceDashboardBannerStatus,
    MaintenanceDashboardStateStatus,
    DashboardScope,
)
from src.models.task import (  # noqa: F401
    TaskRecord,
    TaskLogRecord,
)
from src.models.mapping import Environment  # noqa: F401


# #region postgres_container [C:2] [TYPE Fixture]
# @BRIEF Session-scoped Testcontainers PostgreSQL container.
@pytest.fixture(scope="session")
def postgres_container():
    """Start a PostgreSQL 16 container for the test session."""
    from testcontainers.postgres import PostgresContainer

    with PostgresContainer(
        image="postgres:16-alpine",
        username="test",
        password="test",
        dbname="test_translate",
    ) as pg:
        yield pg


# #endregion postgres_container


# #region pg_engine [C:2] [TYPE Fixture]
# @BRIEF SQLAlchemy engine connected to the Testcontainers PostgreSQL.
@pytest.fixture(scope="session")
def pg_engine(postgres_container):
    """Create engine and tables once per session."""
    url = postgres_container.get_connection_url()
    engine = create_engine(url, echo=False)

    # Create all tables
    Base.metadata.create_all(bind=engine)

    yield engine

    # Cleanup
    Base.metadata.drop_all(bind=engine)
    engine.dispose()


# #endregion pg_engine


# #region db_session [C:2] [TYPE Fixture]
# @BRIEF Per-test database session with transaction rollback for isolation.
@pytest.fixture
def db_session(pg_engine):
    """Provide a transactional database session that rolls back after each test.

    This ensures test isolation — each test starts with a clean database state.
    Uses connection-level transaction to wrap all operations.
    """
    connection = pg_engine.connect()
    transaction = connection.begin()

    session = sessionmaker(bind=connection)()

    # Join the outer transaction
    session.begin_nested()

    @event.listens_for(session, "after_transaction_end")
    def restart_savepoint(session, transaction):
        if transaction.nested and not transaction._parent.nested:
            session.begin_nested()

    try:
        yield session
    finally:
        session.close()
        transaction.rollback()
        connection.close()


# #endregion db_session


# #region mock_config_manager [C:2] [TYPE Fixture]
# @BRIEF Mock ConfigManager for integration tests.
@pytest.fixture
def mock_config_manager():
    """Provide a mock ConfigManager that returns test environment."""
    from unittest.mock import MagicMock
    from src.core.config_models import Environment

    config_manager = MagicMock()
    config_manager.get_environments.return_value = [
        Environment(
            id="test_env",
            name="Test Environment",
            url="http://superset:8088",
            username="admin",
            password="admin",
        )
    ]
    config_manager.get_environment.return_value = Environment(
        id="test_env",
        name="Test Environment",
        url="http://superset:8088",
        username="admin",
        password="admin",
    )
    config_manager.get_config.return_value = MagicMock()
    return config_manager


# #endregion mock_config_manager


# #region verify_postgres_features [C:2] [TYPE Function]
# @BRIEF Helper to verify PostgreSQL-specific features are working.
def verify_postgres_features(session: Session) -> dict:
    """Verify that PostgreSQL-specific features are available.

    Returns a dict with feature availability status.
    """
    features = {}

    # Check JSON support
    try:
        result = session.execute(text("SELECT '{}'::jsonb")).scalar()
        features["jsonb"] = result is not None
    except Exception:
        features["jsonb"] = False

    # Check FK constraints
    try:
        result = session.execute(text("SELECT COUNT(*) FROM information_schema.table_constraints WHERE constraint_type = 'FOREIGN KEY'")).scalar()
        features["foreign_keys"] = result > 0
    except Exception:
        features["foreign_keys"] = False

    # Check indexes
    try:
        result = session.execute(text("SELECT COUNT(*) FROM pg_indexes WHERE schemaname = 'public'")).scalar()
        features["indexes"] = result > 0
    except Exception:
        features["indexes"] = False

    return features


# #endregion verify_postgres_features


# ── TLS / PKI Fixtures ────────────────────────────────────────────────


# #region ca_chain [C:2] [TYPE Fixture]
# @BRIEF Session-scoped — generates 3-tier PKI (Root CA → Intermediate CA → Server cert).
# @POST Returns dict with root_crt, intermediate_crt, server_crt, server_key,
#   server_key_encrypted, server_key_passphrase, fullchain PEM strings.
@pytest.fixture(scope="session")
def ca_chain():
    """Generate 3-tier PKI: Root CA → Intermediate CA → Server certificate (SAN: localhost)."""
    from cryptography import x509
    from cryptography.x509.oid import NameOID
    from cryptography.hazmat.primitives import hashes, serialization
    from cryptography.hazmat.primitives.asymmetric import rsa
    import datetime

    def _gen_key():
        return rsa.generate_private_key(public_exponent=65537, key_size=2048)

    def _gen_cert_pem(subject_name, issuer_name, subject_key, issuer_key, is_ca=False, san_dns=None, san_ip=None):
        now = datetime.datetime.now(datetime.timezone.utc)
        builder = x509.CertificateBuilder()
        builder = builder.subject_name(
            x509.Name(
                [
                    x509.NameAttribute(NameOID.COMMON_NAME, subject_name),
                ]
            )
        )
        builder = builder.issuer_name(
            x509.Name(
                [
                    x509.NameAttribute(NameOID.COMMON_NAME, issuer_name),
                ]
            )
        )
        builder = builder.not_valid_before(now - datetime.timedelta(hours=1))
        builder = builder.not_valid_after(now + datetime.timedelta(days=365))
        builder = builder.serial_number(x509.random_serial_number())
        builder = builder.public_key(subject_key.public_key())

        if is_ca:
            builder = builder.add_extension(
                x509.BasicConstraints(ca=True, path_length=None),
                critical=True,
            )
            builder = builder.add_extension(
                x509.KeyUsage(
                    key_cert_sign=True,
                    crl_sign=True,
                    digital_signature=False,
                    content_commitment=False,
                    key_encipherment=False,
                    data_encipherment=False,
                    key_agreement=False,
                    encipher_only=False,
                    decipher_only=False,
                ),
                critical=True,
            )
        else:
            # Server cert
            builder = builder.add_extension(
                x509.BasicConstraints(ca=False, path_length=None),
                critical=True,
            )
            sans: list[x509.GeneralName] = []
            if san_dns:
                sans.append(x509.DNSName(san_dns))
            if san_ip:
                sans.append(x509.IPAddress(san_ip))
            builder = builder.add_extension(
                x509.SubjectAlternativeName(sans),
                critical=False,
            )

        # Subject Key Identifier — required for OpenSSL 3.x capath chain building
        builder = builder.add_extension(
            x509.SubjectKeyIdentifier.from_public_key(subject_key.public_key()),
            critical=False,
        )
        # Authority Key Identifier — identifies the issuer's key (needed for non-root certs)
        if subject_name != issuer_name:
            builder = builder.add_extension(
                x509.AuthorityKeyIdentifier.from_issuer_public_key(issuer_key.public_key()),
                critical=False,
            )

        cert = builder.sign(issuer_key, hashes.SHA256())
        return cert.public_bytes(serialization.Encoding.PEM).decode()

    root_key = _gen_key()
    root_crt = _gen_cert_pem("Test Root CA", "Test Root CA", root_key, root_key, is_ca=True)

    intermediate_key = _gen_key()
    intermediate_crt = _gen_cert_pem(
        "Test Intermediate CA",
        "Test Root CA",
        intermediate_key,
        root_key,
        is_ca=True,
    )

    server_key = _gen_key()
    import ipaddress

    server_crt = _gen_cert_pem(
        "localhost",
        "Test Intermediate CA",
        server_key,
        intermediate_key,
        is_ca=False,
        san_dns="localhost",
        san_ip=ipaddress.IPv4Address("127.0.0.1"),
    )

    # Full chain for server: server PEM + intermediate PEM (order: leaf first)
    fullchain = server_crt + "\n" + intermediate_crt

    # Passphrase for encrypted key tests
    server_key_passphrase = "test-passphrase-12345"
    server_key_encrypted = server_key.private_bytes(
        encoding=serialization.Encoding.PEM,
        format=serialization.PrivateFormat.TraditionalOpenSSL,
        encryption_algorithm=serialization.BestAvailableEncryption(server_key_passphrase.encode()),
    ).decode()

    return {
        "root_crt": root_crt,
        "root_key": root_key.private_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PrivateFormat.TraditionalOpenSSL,
            encryption_algorithm=serialization.NoEncryption(),
        ).decode(),
        "intermediate_crt": intermediate_crt,
        "intermediate_key": intermediate_key.private_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PrivateFormat.TraditionalOpenSSL,
            encryption_algorithm=serialization.NoEncryption(),
        ).decode(),
        "server_crt": server_crt,
        "server_key": server_key.private_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PrivateFormat.TraditionalOpenSSL,
            encryption_algorithm=serialization.NoEncryption(),
        ).decode(),
        "server_key_encrypted": server_key_encrypted,
        "server_key_passphrase": server_key_passphrase,
        "fullchain": fullchain,
    }


# #endregion ca_chain


# ── Superset Integration Test Fixtures ────────────────────────────────


# #region superset_db_url [C:2] [TYPE Fixture]
# @BRIEF Session-scoped — creates a separate 'superset_meta' database in the Postgres container.
@pytest.fixture(scope="session")
def superset_db_url(postgres_container):
    """Create a dedicated database for Superset metadata inside the shared Postgres."""
    import psycopg2

    conn = psycopg2.connect(
        host=postgres_container.get_container_host_ip(),
        port=postgres_container.get_exposed_port(5432),
        user="test",
        password="test",
        dbname="test_translate",
    )
    conn.autocommit = True

    with conn.cursor() as cur:
        cur.execute("DROP DATABASE IF EXISTS superset_meta")
        cur.execute("CREATE DATABASE superset_meta")

    conn.close()

    # Build a Superset-compatible SQLALCHEMY_DATABASE_URI.
    # CRITICAL: cannot use "localhost" because from inside the Superset container,
    # "localhost" is the container itself. Must use the Docker bridge IP of the
    # Postgres container so that the Superset container can reach it.
    pg_container = postgres_container.get_wrapped_container()
    pg_container.reload()
    pg_ip = pg_container.attrs["NetworkSettings"]["Networks"]["bridge"]["IPAddress"]
    pg_port = 5432  # internal port, not exposed port
    superset_url = f"postgresql+psycopg2://test:test@{pg_ip}:{pg_port}/superset_meta"
    return superset_url


# #endregion superset_db_url


# #region superset_secret_key [C:1] [TYPE Fixture]
# @BRIEF Session-scoped — deterministic secret key for the Superset instance.
@pytest.fixture(scope="session")
def superset_secret_key():
    return "test-superset-secret-key-for-integration-tests-xyz"


# #endregion superset_secret_key


# #region superset_admin_password [C:1] [TYPE Fixture]
# @BRIEF Password for the pre-created admin user in the Superset container.
@pytest.fixture(scope="session")
def superset_admin_password():
    return "admin123"


# #endregion superset_admin_password


# #region superset_container [C:3] [TYPE Fixture]
# @BRIEF Session-scoped Apache Superset container — db upgrade → admin → init → web server.
#   Can be parametrized with {"tls": True} to enable HTTPS with custom 3-tier PKI.
# @PRE Docker daemon is running. superset_db_url points to an empty Postgres database.
# @POST Superset web server is listening on port 8088 (HTTP or HTTPS). Admin user created.
# @SIDE_EFFECT Spins up a short-lived init container, then a long-lived web container.
#   In TLS mode: writes server cert + key into container via base64, runs with --cert/--key.
# @RATIONALE Two-container approach (init + web) avoids race conditions where the web
#   server starts before migrations have completed. The init container runs `superset db
#   upgrade`, creates the admin user, and runs `superset init`, then exits. The web
#   container then starts against the fully initialized database.
#   TLS parametrization allows testing corporate SSL certificate chain with the
#   same Superset instance, using a dynamically generated 3-tier PKI.
# @REJECTED Single container with sequential exec calls — cannot change entrypoint
#   from init to web server within the same container lifecycle.
# @REJECTED Using `latest` tag — version drift breaks tests silently.
@pytest.fixture(scope="session")
def superset_container(request, superset_db_url, superset_secret_key, superset_admin_password):
    """Start an Apache Superset instance for integration testing.

    Parametrize with indirect=True, {"tls": True} to enable HTTPS with custom CA chain.
    """
    from testcontainers.core.container import DockerContainer
    import base64

    superset_image = "apache/superset:4.1.2"

    # Check if TLS was requested via parametrization
    tls_config = getattr(request, "param", None)
    use_tls = tls_config is not None and tls_config.get("tls", False)

    if use_tls:
        ca_chain = request.getfixturevalue("ca_chain")
        # Base64-encode certs to avoid shell escaping issues in heredoc
        fullchain_b64 = base64.b64encode(ca_chain["fullchain"].encode()).decode()

        if tls_config.get("encrypted_key", False):
            key_pem = ca_chain["server_key_encrypted"]
            key_pass = ca_chain["server_key_passphrase"]
            key_b64 = base64.b64encode(key_pem.encode()).decode()
            _write_certs = f"echo '{fullchain_b64}' | base64 -d > /tmp/server.crt && echo '{key_b64}' | base64 -d > /tmp/server.key"
            _decrypt_key = f"export SSL_KEY_PASSPHRASE='{key_pass}' && openssl rsa -in /tmp/server.key -passin env:SSL_KEY_PASSPHRASE -out /tmp/server.key 2>/dev/null; if [ $? -ne 0 ]; then   echo 'ERROR: Failed to decrypt private key with SSL_KEY_PASSPHRASE'; exit 1; fi"
        else:
            key_b64 = base64.b64encode(ca_chain["server_key"].encode()).decode()
            _write_certs = f"echo '{fullchain_b64}' | base64 -d > /tmp/server.crt && echo '{key_b64}' | base64 -d > /tmp/server.key"
            _decrypt_key = ""

        _protocol = "https"
        _superset_run = f"superset run -h 0.0.0.0 -p 8088 --cert /tmp/server.crt --key /tmp/server.key --with-threads"
    else:
        _write_certs = ""
        _decrypt_key = ""
        _protocol = "http"
        _superset_run = "superset run -h 0.0.0.0 -p 8088 --with-threads"

    # Superset ignores SQLALCHEMY_DATABASE_URI env var — we must create
    # superset_config.py and point SUPERSET_CONFIG_PATH to it.
    _bootstrap_config = 'cat > /tmp/superset_config.py << "PYEOF"\nimport os\nSQLALCHEMY_DATABASE_URI = os.environ.get("SUPERSET_DB_URI", "sqlite://")\nPYEOF\n'
    _export_config = "export SUPERSET_CONFIG_PATH=/tmp/superset_config.py"

    # ── Step 1: Init container ──
    init = DockerContainer(superset_image)
    init.with_env("SUPERSET_SECRET_KEY", superset_secret_key)
    init.with_env("SUPERSET_DB_URI", superset_db_url)
    init.with_env("SUPERSET_LOAD_EXAMPLES", "no")
    init.with_env("FLASK_APP", "superset")
    # Note: init container does NOT need TLS certs — only runs db upgrade + init
    init.with_command(f"sh -c 'python -m pip install psycopg2-binary -q && {_bootstrap_config}{_export_config} && superset db upgrade && superset fab create-admin   --username admin --firstname Admin --lastname User   --email admin@test.local --password {superset_admin_password} && superset init'")

    try:
        init.start()
        # Block until the init command completes (container exits)
        init.get_wrapped_container().wait()
        exit_code = init.get_wrapped_container().attrs["State"]["ExitCode"]
        if exit_code != 0:
            logs = init.get_logs()
            init.stop()
            raise RuntimeError(f"Superset init container failed (exit {exit_code}):\n{logs}")
    finally:
        init.stop()

    # ── Step 2: Web server container ──
    superset = DockerContainer(superset_image)
    superset.with_env("SUPERSET_SECRET_KEY", superset_secret_key)
    superset.with_env("SUPERSET_DB_URI", superset_db_url)
    superset.with_env("SUPERSET_LOAD_EXAMPLES", "no")
    superset.with_env("FLASK_APP", "superset")

    # Build web server command: optionally write certs, decrypt key, then start superset
    if use_tls and _write_certs:
        _decrypt_prefix = f"{_decrypt_key} && " if _decrypt_key else ""
        web_cmd = f"sh -c 'python -m pip install psycopg2-binary -q && {_write_certs} && {_decrypt_prefix}{_bootstrap_config}{_export_config} && {_superset_run}'"
    resp.raise_for_status()
    data = resp.json()
    access_token = data["access_token"]
    assert access_token, "No access_token in JWT response"
    return {"Authorization": f"Bearer {access_token}"}


# #endregion superset_jwt_headers


# #region superset_env [C:2] [TYPE Fixture]
# @BRIEF Function-scoped — real Environment pointing to the running Superset container.
@pytest.fixture
def superset_env(superset_url, superset_admin_password):
    """Create a real Environment config pointing to the Superset container."""
    from src.core.config_models import Environment

    return Environment(
        id="test_env",
        name="Test Superset Container",
        url=superset_url,
        username="admin",
        password=superset_admin_password,
        verify_ssl=False,
        timeout=30,
    )


# #endregion superset_env


# #region superset_client [C:2] [TYPE Fixture]
# @BRIEF Function-scoped — real SupersetClient connected to the container.
@pytest.fixture
async def superset_client(superset_env):
    """Create and authenticate a SupersetClient against the real container."""
    from src.core.superset_client import SupersetClient

    client = SupersetClient(superset_env)
    await client.authenticate()
    return client


# #endregion superset_client


# ── TLS / Custom CA Fixtures ─────────────────────────────────────


# #region install_custom_ca [C:2] [TYPE Fixture]
# @BRIEF Function-scoped — installs Test Root CA into system trust store.
#   Tries update-ca-certificates (root required), falls back to SSL_CERT_DIR.
# @POST Root CA accessible via ssl.create_default_context().
# @SIDE_EFFECT Modifies /usr/local/share/ca-certificates/custom/ and runs update-ca-certificates.
#   On fallback: sets SSL_CERT_DIR env var, creates temp dir with hash symlinks.
@pytest.fixture
def install_custom_ca(ca_chain):
    """Install the custom Root CA into the system trust store.

    Strategy 1 (prod-like): write .crt → update-ca-certificates (requires root).
    Strategy 2 (fallback): SSL_CERT_DIR with hash symlinks.
    """
    import subprocess
    import tempfile
    from pathlib import Path

    ca_pem = ca_chain["root_crt"]
    used_update_ca = False
    ca_verify_path = None
    cert_dir = None

    # Strategy 1: update-ca-certificates (matches production install_certificates())
    try:
        cert_dir = Path("/usr/local/share/ca-certificates/custom")
        cert_dir.mkdir(parents=True, exist_ok=True)
        ca_path = cert_dir / "test_ss_tools_ca.crt"
        ca_path.write_text(ca_pem)
        subprocess.run(
            ["update-ca-certificates", "--fresh"],
            check=True,
            timeout=30,
            capture_output=True,
        )
        ca_verify_path = "/etc/ssl/certs"
        used_update_ca = True
    except (subprocess.CalledProcessError, PermissionError, OSError):
        # Strategy 2: fallback — SSL_CERT_DIR
        cert_dir = Path(tempfile.mkdtemp(prefix="test_ca_"))
        ca_path = cert_dir / "test_ss_tools_ca.pem"
        ca_path.write_text(ca_pem)

        # Create hash symlink for capath
        result = subprocess.run(
            ["openssl", "x509", "-hash", "-noout"],
            input=ca_pem,
            capture_output=True,
            text=True,
            timeout=10,
        )
        cert_hash = result.stdout.strip()
        symlink_path = cert_dir / f"{cert_hash}.0"
        if symlink_path.exists():
            symlink_path.unlink()
        symlink_path.symlink_to(ca_path.name)

        os.environ["SSL_CERT_DIR"] = str(cert_dir)
        ca_verify_path = str(cert_dir)
        used_update_ca = False

    yield {
        "ca_verify_path": ca_verify_path,
        "used_update_ca": used_update_ca,
        "ca_pem": ca_pem,
    }

    # Cleanup
    if used_update_ca:
        ca_file = Path("/usr/local/share/ca-certificates/custom/test_ss_tools_ca.crt")
        if ca_file.exists():
            ca_file.unlink()
        try:
            subprocess.run(
                ["update-ca-certificates", "--fresh"],
                timeout=30,
                capture_output=True,
            )
        except Exception:
            pass
    else:
        if "SSL_CERT_DIR" in os.environ:
            del os.environ["SSL_CERT_DIR"]
        if cert_dir and cert_dir.exists():
            import shutil

            shutil.rmtree(str(cert_dir), ignore_errors=True)


# #endregion install_custom_ca


# #endregion IntegrationTestConftest