DFiP_Budget_planing/api/tests/integration/conftest.py

import asyncio
from contextlib import contextmanager
import os

import pytest
from fastapi.testclient import TestClient
from sqlalchemy import text


from src.db.base import SessionLocal
from src.db.session import get_db
from src.main import app


            
@pytest.fixture
def db_session():
    """Сессия БД теста: её же использует client, всё откатывается в конце теста."""
    db_ref = SessionLocal()
    yield db_ref


@pytest.fixture
def client(db_session):
    db_ref = db_session
    loop_ref = None

    async def _get_db():
        nonlocal loop_ref
        loop_ref = asyncio.get_running_loop()
        try:
            yield db_ref
            await db_ref.flush()
        except:
            await db_ref.rollback()
            raise

    app.dependency_overrides[get_db] = _get_db
    with TestClient(app) as test_client:
        try:
            test_client.db_session = db_ref
            yield test_client
        finally:
            if db_ref is not None and loop_ref:
                async def cleanup():
                    await db_ref.rollback()
                    await db_ref.close()

                future = asyncio.run_coroutine_threadsafe(cleanup(), loop_ref)
                future.result()
            app.dependency_overrides.clear()


@pytest.fixture
def admin_password() -> str:
    # Пароль админа берём из окружения, чтобы не хардкодить локальные отличия.
    return os.getenv("OPENBAO__TEST_ADMIN_PASSWORD", "admin123")


@pytest.fixture
def isp_password() -> str:
    # Пароль админа берём из окружения, чтобы не хардкодить локальные отличия.
    return os.getenv("OPENBAO__TEST_EXECUTOR_PASSWORD", "admin123")


@pytest.fixture
def admin_tokens(client, admin_password: str):
    response = client.post(
        "/api/v1/auth/login",
        json={"username": "admin", "password": admin_password},
    )
    assert response.status_code == 200
    return response.json()


@pytest.fixture
def auth_headers():
    def _build(tokens: dict) -> dict:
        return {"Authorization": f"Bearer {tokens['access_token']}"}

    return _build

@pytest.fixture
def isp_tokens(client, isp_password: str):
    response = client.post(
        "/api/v1/auth/login",
        json={"username": "isp1", "password": isp_password},
    )
    assert response.status_code == 200
    return response.json()


SVOD2_YEAR = 2031
SVOD2_SEED_PATH = os.path.join(os.path.dirname(__file__), "..", "sqls", "svod2_seed.sql")


def legacy_full(client, sheet: str, headers: dict = None, org_ids=None) -> list[dict]:
    """Прежний полный ответ свода 2 без параметров нового режима (оракульная база).

    Строится прямым вызовом функции представления, а не через GET без параметров: после AURORA-1848
    этот GET отдаёт data без null-полей (jsonb_strip_nulls), а оракул экрана «до» работает с полным
    data - в нём пустая секция contract_detail затирает поля contract_summary значением null.
    Запрос идёт от имени администратора, headers оставлен для совместимости вызовов.
    """
    function_name = {"GO": "v3.v_svod2_go", "RF": "v3.v_svod2_rf"}[sheet]

    async def _fetch():
        result = await client.db_session.execute(
            text(
                f"SELECT sort_order, data FROM {function_name}("
                "CAST(:year AS INT), CAST(:org_ids AS INT[]), NULL, NULL, 'org_name', 'asc')"
            ),
            {"year": SVOD2_YEAR, "org_ids": org_ids},
        )
        return [
            {"row_type": "INPUT", "depth": 3, "sort_order": sort_order, "data": data}
            for sort_order, data in result.fetchall()
        ]

    return client.portal.call(_fetch)


@pytest.fixture
def svod2_sql(client):
    """Выполняет text(sql) на сессии клиента (тот же цикл событий, что у запросов)."""

    def _run(sql: str, params: dict | None = None):
        async def _exec():
            result = await client.db_session.execute(text(sql), params or {})
            return list(result.fetchall()) if result.returns_rows else []

        return client.portal.call(_exec)

    return _run


@pytest.fixture
def svod2_data(client, svod2_sql) -> dict:
    """Засевает данные свода 2 (откатываются вместе с сессией клиента)."""
    with open(SVOD2_SEED_PATH, "r", encoding="utf-8") as f:
        script = f.read()
    for statement in script.split(";"):
        statement = statement.strip()
        if statement:
            svod2_sql(statement)
    # id строки contract_detail: contract_detail.line_id = 910002, технический id,
    # который попадает в шапку строки, — expense_item_id этой строки
    cd_id = svod2_sql("SELECT expense_item_id FROM v3.budget_line WHERE id = 910002")[0][0]
    yield {
        "year": SVOD2_YEAR,
        "org_a": 910001,
        "org_b": 910002,
        "org_b_twin": 910003,
        "org_b_case": 910004,
        "cd_id": cd_id,
        "ckk_only_value": "777.25",
    }
    # откат, пока portal открыт: иначе при отсутствии запросов к API транзакция засева
    # остаётся висеть и блокирует следующий засев
    client.portal.call(client.db_session.rollback)