smart-org/apps/api/tests/conftest.py

"""Shared test fixtures.

Tests run against an in-memory SQLite database with the real agy-shaped adapter
wiring and a fake model provider, so the suite needs no Postgres, no Redis, no
Google and no accounts. agy is the only model provider the platform talks to
(see ``app/infrastructure/adapters/registry.py``); the fake keeps completions
deterministic and offline.
"""

from __future__ import annotations

import pytest
import pytest_asyncio
from httpx import ASGITransport, AsyncClient
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.pool import StaticPool

# Importing the ORM package registers every table on ``Base.metadata``. Bound to
# a private name on purpose: ``app`` is also the FastAPI instance imported below,
# and binding the package to the same name would shadow it.
import app.infrastructure.db.models as _orm_models  # noqa: F401
from app.core.config import settings
from app.core.database import Base, enable_sqlite_foreign_keys, get_session
from app.core.security import create_access_token, hash_password
from app.domain.base import new_id
from app.domain.identity.models import PlatformRole, User
from app.infrastructure.repositories.identity import UserRepository

# The internal "full" app holds every HTTP handler; the public app (app.main.app)
# now serves only /health + /api/ws, so REST tests must target the full app.
from app.interfaces.realtime.rpc_app import get_rpc_app

app = get_rpc_app()


@pytest.fixture(scope="session", autouse=True)
def _production_shaped_jwt_secret():
    """Sign test tokens with a 32+ byte key.

    The development default is ``change-me`` (9 bytes), so PyJWT emits an
    ``InsecureKeyLengthWarning`` for every token signed or verified — hundreds of
    lines per run, which buries genuine warnings. Production refuses a short
    secret outright; that guard is covered by ``tests/test_config_guard.py``.
    """
    settings.jwt_secret = "test-only-jwt-secret-with-more-than-32-bytes"


@pytest.fixture(scope="session", autouse=True)
def _hermetic_realtime_transport():
    """Pin the realtime bus to in-process delivery.

    ``auto`` would make every publish attempt a Redis connection to
    ``localhost:6379`` — the suite has no broker, so each event would add a
    connection timeout and log noise. The fan-out behaviour under test does not
    depend on the transport.
    """
    settings.realtime_transport = "local"


@pytest_asyncio.fixture
async def engine():
    # StaticPool keeps one connection alive so the in-memory schema survives
    # across the sessions a single request opens.
    test_engine = create_async_engine(
        "sqlite+aiosqlite:///:memory:",
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,
    )
    # Same pragma the app engine gets, so ON DELETE CASCADE behaves here the
    # way it does on Postgres.
    enable_sqlite_foreign_keys(test_engine)
    async with test_engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield test_engine
    await test_engine.dispose()


@pytest_asyncio.fixture
async def session_factory(engine):
    return async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)


@pytest_asyncio.fixture
async def redirect_error_log(engine, monkeypatch):
    """Point the error recorder and the workers at the in-memory database.

    Neither goes through the ``get_session`` dependency override that ``client``
    installs — the recorder deliberately opens its own session, and the workers
    are not HTTP requests at all. Without this redirect, any test that triggers a
    failure path would write into the developer's real ``agy_web.db``.

    Not autouse: only the modules that exercise failure paths need it, and the
    ``engine`` it depends on builds the whole schema.
    """
    from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

    import app.infrastructure.error_log as error_log
    from app.workers import base as worker_base

    factory = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
    monkeypatch.setattr(error_log, "SessionFactory", factory)
    monkeypatch.setattr(worker_base, "SessionFactory", factory)
    return factory


@pytest_asyncio.fixture
async def client(engine, session_factory):
    async def _override_session():
        async with session_factory() as session:
            try:
                yield session
                await session.commit()
            except Exception:
                await session.rollback()
                raise

    # Publish the factory for the module-level ``signup`` helper below. The
    # in-memory engine each test builds is private to its fixtures; a helper
    # that needs the same database has to reach it through this door.
    _session_factory_holder["factory"] = session_factory
    app.dependency_overrides[get_session] = _override_session
    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as test_client:
        yield test_client
    app.dependency_overrides.clear()


#: Set by the ``client`` fixture; read by :func:`signup`.
_session_factory_holder: dict[str, async_sessionmaker] = {}


# ---------------------------------------------------------------------------
# Fake model provider — the only sanctioned stand-in for agy in tests
# ---------------------------------------------------------------------------


class _FakeModelProvider:
    """Offline stand-in for the agy adapter.

    The canned text must read like model output, not like a debug tag: the
    provider/model identity is platform-internal and must never surface in
    anything a customer can read (see the e2e smoke test).
    """

    async def complete(
        self,
        *,
        model,
        messages,
        system="",
        temperature=0.2,
        max_tokens=2048,
        credential=None,
        credentials=None,
    ):
        import asyncio
        import time

        from app.application.ports import ModelResponse

        started = time.perf_counter()
        last_user = next((m.content for m in reversed(messages) if m.role == "user"), "")
        prompt_tokens = sum(len(m.content.split()) for m in messages)
        # A tiny sleep so latency bookkeeping runs a real code path.
        await asyncio.sleep(0)
        content = (
            "این پاسخ آزمایشی است. "
            f"درخواست دریافتی: {last_user[:280] or '—'}"
        )
        return ModelResponse(
            content=content,
            model_internal_name=model.internal_name,
            input_tokens=prompt_tokens + len(system.split()),
            output_tokens=len(content.split()),
            latency_ms=int((time.perf_counter() - started) * 1000),
            raw={},
        )

    async def health(self):
        from app.application.ports import ModelHealth

        return ModelHealth(ok=True, detail="fake agy pool")


@pytest.fixture(autouse=True)
def _fake_model_provider(monkeypatch):
    """Swap the real model adapter out of every request path.

    ``container`` and the health router bind ``get_model_provider`` at import
    time, so both import sites are patched; without this a completion test
    would open the developer's real database and reach the real Google API.
    """
    from app.application import container as di_container
    from app.interfaces.http.routers import health as health_router

    provider = _FakeModelProvider()
    monkeypatch.setattr(di_container, "get_model_provider", lambda: provider)
    monkeypatch.setattr(health_router, "get_model_provider", lambda: provider)
    return provider


async def signup(
    client: AsyncClient,
    email: str,
    password: str = "password123",
    name: str = "",
    platform_role: PlatformRole = PlatformRole.MANAGER,
) -> dict:
    """Create an account directly in the database and mint a token for it.

    There is no public signup endpoint anymore — accounts are created by the
    platform owner or by redeeming an invite — so this helper writes the row
    those flows would and returns the same shape ``POST /auth/signup`` used to.
    The ``client`` argument is kept for the historical call signature.

    The default platform role is ``manager`` because most tests exercise the
    tenant surface, and creating an organization requires that tier. Tests
    that model a plain invited user pass ``platform_role=PlatformRole.USER``.
    """
    factory = _session_factory_holder["factory"]
    async with factory() as session:
        user = await UserRepository(session).create(
            User(
                email=email,
                password_hash=hash_password(password),
                name=name,
                platform_role=platform_role,
            )
        )
        await session.commit()
    return {
        "access_token": create_access_token(user.id),
        "user": {
            "id": user.id,
            "email": user.email,
            "name": user.display_name,
            "platform_role": user.platform_role.value,
            "is_manager": user.is_manager,
            "is_platform_owner": user.is_platform_owner,
        },
    }


def auth_header(token: str) -> dict[str, str]:
    return {"Authorization": f"Bearer {token}"}


@pytest_asyncio.fixture
async def platform_owner_token(client: AsyncClient, session_factory) -> str:
    """A token for the one and only platform owner.

    Owner seats are not grantable through the API by design (a manager must
    not be able to mint themselves an owner), so the fixture creates the
    account with the owner role directly — the same seat the seed installs.
    """
    email = "platform-owner@example.com"
    body = await signup(client, email, name="مالک سامانه", platform_role=PlatformRole.OWNER)
    return body["access_token"]


@pytest_asyncio.fixture
async def manager_token(client: AsyncClient) -> str:
    """A token for a plain manager account (the org-creating tier)."""
    body = await signup(client, "manager@example.com", name="مدیر")
    return body["access_token"]


@pytest_asyncio.fixture
async def plain_user_token(client: AsyncClient) -> str:
    """A token for a plain user — the tier that may not create organizations."""
    body = await signup(
        client, "plain-user@example.com", name="کاربر عادی", platform_role=PlatformRole.USER
    )
    return body["access_token"]


#: The system agent templates ``app/seed.py`` installs. Mirrored here so the test
#: suite exercises the same employee-creation path a seeded install uses.
SYSTEM_TEMPLATES: list[tuple[str, str, str, list[str]]] = [
    ("مدیر سازمان", "manager", "کارها را تقسیم می‌کند.", ["knowledge.search"]),
    (
        "برنامه‌نویس",
        "developer",
        "کد می‌نویسد و تست می‌کند.",
        ["sandbox.run_command", "sandbox.write_file"],
    ),
]

#: The routing policies ``app/seed.py`` installs.
SYSTEM_ROUTING_POLICIES: list[tuple[str, str, bool, str]] = [
    ("سریع و ارزان", "cheap", True, "برای کارهای روزمره."),
    ("استدلال عمیق", "reasoning", False, "برای تحلیل پیچیده."),
]

#: The agy models ``app/seed.py`` installs: (internal_name, display,
#: capabilities, reasoning_effort). agy is the only provider, so this is the
#: whole catalogue.
SYSTEM_MODELS: list[tuple[str, str, list[str], str]] = [
    ("gemini-3-flash", "Gemini 3 Flash", ["text", "code", "long_context"], ""),
    (
        "gemini-3-pro-high",
        "Gemini 3 Pro",
        ["text", "reasoning", "code", "vision", "long_context"],
        "high",
    ),
    ("claude-sonnet-4-5", "Claude Sonnet 4.5", ["text", "code", "tools"], ""),
    (
        "claude-sonnet-4-5-thinking",
        "Claude Sonnet 4.5 (Thinking)",
        ["text", "code", "reasoning"],
        "high",
    ),
    (
        "claude-opus-4-5-thinking",
        "Claude Opus 4.5 (Thinking)",
        ["text", "reasoning", "code", "long_context"],
        "high",
    ),
    ("gpt-oss-120b-medium", "GPT-OSS 120B Medium", ["text", "code"], ""),
]

#: The rate cards ``app/seed.py`` installs: (unit, customer_price, internal_cost).
SYSTEM_RATE_CARDS: list[tuple[str, float, float]] = [
    ("model_input_tokens", 0.0000004, 0.0000002),
    ("model_output_tokens", 0.0000016, 0.0000008),
    ("tool_call", 0.0005, 0.0001),
    ("premium_employee_run", 0.02, 0.008),
]

#: The sandbox templates ``app/seed.py`` installs: (name, image, description).
SYSTEM_SANDBOX_TEMPLATES: list[tuple[str, str, str]] = [
    ("python-3.13", "python:3.13-slim", "محیط پایتون برای اجرای تست و اسکریپت."),
    ("node-22", "node:22-slim", "محیط Node برای build و اجرای پروژه‌های فرانت‌اند."),
]


@pytest_asyncio.fixture
async def seeded(session_factory):
    """Install the platform-level seed rows a fresh install would have.

    Agent templates, routing policies, the agy provider with its model catalog,
    the rate cards and the sandbox templates are all *platform* data (no
    organization owns them), so without this an employee cannot be created, a
    run cannot resolve a model to route to, usage cannot be priced and no
    sandbox can be started — none of which the real seeded install lacks.
    """
    from app.infrastructure.db.models import (
        AgentTemplateRow,
        AiModelRow,
        AiProviderRow,
        RateCardRow,
        RoutingPolicyRow,
        SandboxTemplateRow,
    )

    async with session_factory() as session:
        for name, role_key, description, tools in SYSTEM_TEMPLATES:
            session.add(
                AgentTemplateRow(
                    name=name,
                    role_key=role_key,
                    description=description,
                    default_prompt=f"تو {name} هستی. وظیفه‌ات: {description}",
                    default_tools_json=tools,
                    is_system=True,
                )
            )
        for name, mode, is_default, description in SYSTEM_ROUTING_POLICIES:
            session.add(
                RoutingPolicyRow(
                    name=name, description=description, mode=mode,
                    is_default=is_default, is_active=True,
                )
            )

        # A run cannot reach a model without an active provider + model, and
        # ``pick_for_mode`` returns ``None`` when the catalogue is empty — which
        # surfaces as a ``ProviderUnavailableError`` and a *failed* run.
        provider_id = new_id()
        session.add(
            AiProviderRow(
                id=provider_id, name="agy", type="agy", base_url="", is_active=True
            )
        )
        # Flush the provider before its models: with foreign keys enforced the
        # batched model INSERT must not run before the row it references
        # exists.
        await session.flush()
        for internal, display, capabilities, effort in SYSTEM_MODELS:
            session.add(
                AiModelRow(
                    provider_id=provider_id,
                    internal_name=internal,
                    display_name_for_admin=display,
                    capabilities=capabilities,
                    input_price_per_1m=0.3,
                    output_price_per_1m=1.2,
                    reasoning_effort=effort,
                    is_active=True,
                )
            )
        for unit, customer_price, internal_cost in SYSTEM_RATE_CARDS:
            session.add(
                RateCardRow(
                    name=unit.replace("_", " "), unit=unit,
                    customer_price=customer_price,
                    internal_cost_estimate=internal_cost, is_active=True,
                )
            )
        for name, image, description in SYSTEM_SANDBOX_TEMPLATES:
            session.add(
                SandboxTemplateRow(
                    name=name, provider="mock", image=image, description=description,
                    default_cpu=1.0, default_memory_mb=1024, default_disk_mb=2048,
                    is_system=True,
                )
            )
        await session.commit()
    return True