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()


@pytest.fixture
def isp2_tokens(client, isp_password: str):
    # Исполнитель в org_unit формы id=1 (isp1 в нём намеренно не состоит — на этом
    # держатся тесты «исполнитель из чужой организации»). Нужен вместе с isp3_tokens,
    # чтобы проверять, что исполнитель не может удалить чужую строку.
    response = client.post(
        "/api/v1/auth/login",
        json={"username": "isp2", "password": isp_password},
    )
    assert response.status_code == 200
    return response.json()


@pytest.fixture
def isp3_tokens(client, isp_password: str):
    # Второй исполнитель в том же org_unit, что и isp2 (форма id=1).
    response = client.post(
        "/api/v1/auth/login",
        json={"username": "isp3", "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, org_ids=None) -> list[dict]:
    """Ответ свода 2 без параметров нового режима - то, что получает экран (оракульная база).

    После AURORA-1848 data в нём без null-полей: оракул экрана «до» работает именно с такими строками
    (решение пользователя 08.11, R-30: поиск совпадает с тем, что видно на экране).
    """
    url = f"/api/v1/svod/2/{sheet}/{SVOD2_YEAR}"
    if org_ids is not None:
        url += f"?org_ids={org_ids}"
    response = client.get(url, headers=headers)
    assert response.status_code == 200, response.text
    return response.json()["result"]


@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)