conftest.py

"""Root-level pytest conftest.

Applies the openGauss SQLAlchemy compatibility monkey-patch globally so any
test in the repo that connects to the real openGauss instance works correctly.

Note: Service-specific integration tests (lightrag, pykt, ralph-lrs, ai-tutor)
are designed to run from their own service directory:

    cd services/lightrag && python -m pytest tests/test_integration_lightrag_tutor.py
    cd services/pykt && python -m pytest tests/test_integration_pykt_lmscore.py
    cd services/ralph-lrs && python -m pytest tests/test_integration_xapi_etl.py
    cd services/ai-tutor && python -m pytest tests/test_integration_tutor_flow.py

The top-level tests/ directory (test_c_package_integration.py) uses subprocess
isolation and can be run from the edu-platform root:

    python -m pytest tests/test_c_package_integration.py
"""
from __future__ import annotations

import asyncio
import os
import re
import sys
import uuid
from pathlib import Path

import bcrypt


_ROOT = Path(__file__).resolve().parent
_SERVICE_DIRS = {
    (_ROOT / "services" / "ai-tutor").resolve(),
    (_ROOT / "services" / "lightrag").resolve(),
    (_ROOT / "services" / "lms-core").resolve(),
    (_ROOT / "services" / "pykt").resolve(),
    (_ROOT / "services" / "ralph-lrs").resolve(),
}

_OG_HOST = "localhost"
_OG_PORT = 15432
_OG_USER = "edu_app"
_OG_PASS = "EduApp#2026"
_OG_DB = "edu_platform_test"


async def _insert_privileged_user(email: str, password: str, role: str) -> str:
    import asyncpg

    conn = await asyncpg.connect(
        host=_OG_HOST,
        port=_OG_PORT,
        user=_OG_USER,
        password=_OG_PASS,
        database=_OG_DB,
    )
    try:
        existing = await conn.fetchrow("SELECT id FROM student WHERE email=$1", email)
        if existing:
            return str(existing["id"])
        uid = str(uuid.uuid4())
        pw_hash = bcrypt.hashpw(password.encode(), bcrypt.gensalt()).decode()
        trust = 3 if role in ("teacher", "admin") else 1
        await conn.execute(
            """INSERT INTO student (id, email, password_hash, role, trust_level, reputation_score, profile_json)
               VALUES ($1, $2, $3, $4, $5, 0, '{}')""",
            uid,
            email,
            pw_hash,
            role,
            trust,
        )
        return uid
    finally:
        await conn.close()


def create_privileged_user(email: str, password: str, role: str = "teacher") -> str:
    """Create a teacher/admin test user in the real openGauss database."""
    return asyncio.run(_insert_privileged_user(email, password, role))


async def _db_exec_async(sql: str, params: tuple = ()) -> None:
    import asyncpg
    conn = await asyncpg.connect(host=_OG_HOST, port=_OG_PORT, user=_OG_USER, password=_OG_PASS, database=_OG_DB)
    try:
        pg_sql = sql
        if params:
            for i in range(len(params)):
                pg_sql = pg_sql.replace("%s", f"${i + 1}", 1)
        await conn.execute(pg_sql, *params)
    finally:
        await conn.close()


async def _db_fetch_async(sql: str, params: tuple = ()) -> list:
    import asyncpg
    conn = await asyncpg.connect(host=_OG_HOST, port=_OG_PORT, user=_OG_USER, password=_OG_PASS, database=_OG_DB)
    try:
        pg_sql = sql
        if params:
            for i in range(len(params)):
                pg_sql = pg_sql.replace("%s", f"${i + 1}", 1)
        return await conn.fetch(pg_sql, *params)
    finally:
        await conn.close()


def db_exec(sql: str, params: tuple = ()) -> None:
    """Execute a DML statement against the openGauss test database."""
    asyncio.run(_db_exec_async(sql, params))


def db_fetch(sql: str, params: tuple = ()) -> list[dict]:
    """Fetch rows from the openGauss test database."""
    return asyncio.run(_db_fetch_async(sql, params))


def _set_service_import_root(path: Path) -> None:
    resolved = path.resolve()
    service_root = next(
        (root for root in _SERVICE_DIRS if root in resolved.parents),
        None,
    )
    if service_root is None:
        return

    root_s = os.fspath(_ROOT)
    service_root_s = os.fspath(service_root)
    other_roots = {os.fspath(root) for root in _SERVICE_DIRS if root != service_root}
    sys.path[:] = [
        entry
        for entry in sys.path
        if os.path.abspath(entry or ".") not in other_roots
    ]
    if root_s not in sys.path:
        sys.path.insert(0, root_s)
    if service_root_s in sys.path:
        sys.path.remove(service_root_s)
    sys.path.insert(0, service_root_s)

    loaded_app = sys.modules.get("app")
    loaded_file = getattr(loaded_app, "__file__", "") if loaded_app else ""
    loaded_paths = getattr(loaded_app, "__path__", []) if loaded_app else []
    file_matches = bool(loaded_file) and Path(loaded_file).resolve().is_relative_to(service_root)
    path_matches = any(Path(p).resolve().is_relative_to(service_root) for p in loaded_paths)
    if not (file_matches or path_matches):
        for name in list(sys.modules):
            if name == "app" or name.startswith("app."):
                sys.modules.pop(name, None)
        for name in list(sys.modules):
            if name == "app" or name.startswith("app."):
                sys.modules.pop(name, None)


def pytest_collect_file(file_path, parent):
    """Keep same-named service packages isolated during root-level collection."""
    _set_service_import_root(Path(file_path))
    return None


def pytest_runtest_setup(item):
    """Switch top-level service imports before each test executes."""
    _set_service_import_root(Path(item.path))


def pytest_configure(config):
    """Load sprint0 env and apply openGauss SQLAlchemy compatibility patch at session start."""
    _sprint0_env = _ROOT.parent / "edu-platform-sprint0" / ".env"
    if _sprint0_env.exists():
        for line in _sprint0_env.read_text().splitlines():
            line = line.strip()
            if not line or line.startswith("#") or "=" not in line:
                continue
            k, _, v = line.partition("=")
            k = k.strip()
            v = v.strip().strip('"').strip("'")
            os.environ.setdefault(k, v)
    # Cross-name aliases: sprint0 uses LLM_API_KEY, services expect ALIYUN_API_KEY
    os.environ.setdefault("ALIYUN_API_KEY", os.environ.get("LLM_API_KEY", ""))
    os.environ.setdefault("REDIS_URL", "redis://localhost:16379/0")
    os.environ.setdefault("CELERY_BROKER_URL", "redis://localhost:16379/0")
    os.environ.setdefault("CELERY_RESULT_BACKEND", "redis://localhost:16379/1")
    try:
        from sqlalchemy.dialects.postgresql.base import PGDialect

        _orig = PGDialect._get_server_version_info

        def _patched(self, connection):
            try:
                return _orig(self, connection)
            except AssertionError:
                v = connection.exec_driver_sql("SELECT version()").scalar()
                m = re.search(r"openGauss\s+(\d+)\.(\d+)", v or "")
                if m:
                    return (int(m.group(1)), int(m.group(2)))
                return (14, 0)

        PGDialect._get_server_version_info = _patched
    except ImportError:
        pass