"""AURORA-1848: тяжёлые GET отдают готовый JSON из Postgres.

Эталон — старый путь (методы репозиториев и сервисов, которыми пользуется выгрузка Excel)
в той же сессии теста. Спека: openspec/changes/aurora-1848-sheet-json/specs/sheet-read-api/spec.md
"""
import json
import uuid
from types import SimpleNamespace

import pytest
from sqlalchemy import text

from src.main import app
from src.repository.project_repository import ProjectRepository
from src.repository.sheet_repository import SheetRepository
from src.repository.summary_repository import SummaryRepository
from src.services.project_service import ProjectService
from src.services.summary_service import SummaryService

ADMIN = SimpleNamespace(id=1, role_id=1)  # fixture.sql: admin
ISP = SimpleNamespace(id=2, role_id=2)    # fixture.sql: isp1, EXECUTOR_DFIP
YEAR = 2026

REPORT_NOT_FOUND = (400, {"code": 400, "field": "", "message": "Отчёт не найден"})
RF_ROLLUP_SQL_ERROR = (400, {"code": 400, "field": "", "message": "<class 'asyncpg.exceptions.UndefinedColumnError'>: column r.year does not exist"})
RF_FOREIGN = (403, {"code": 403, "field": "", "message": "Доступ запрещен"})

HEAVY_GET_PATHS = [
    "/api/v1/form/{form_id}/sheet/{sheet}",
    "/api/v1/svod/1/{sheet}/{year}",
    "/api/v1/svod/2/{sheet}/{year}",
    "/api/v1/projects/{project_id}/report/{year}/{report_type}",
    "/api/v1/rf-rollup/{branch_id}/{year}",
    "/api/v1/projects/{project_id}/summary",
    "/api/v1/projects/plan-summary",
    "/api/v1/projects/fact-summary",
    "/api/v1/projects/detail-summary",
]
OPENAPI_REF = "#/components/schemas/BaseListResponse_SheetResponse_"


def _payload(rows) -> list[dict]:
    """Строки старого пути в том виде, в каком их отдавал BaseListResponse[SheetResponse]."""
    return [{"row_type": r[0], "depth": r[1], "sort_order": r[2], "data": r[3]} for r in rows]


def old_rows(client, coro_factory) -> list[dict]:
    """Вызвать старый метод (coro_factory() -> awaitable list[tuple]) в цикле приложения."""
    async def _go():
        return _payload(await coro_factory())

    return client.portal.call(_go)


def expected_sheet(rows: list[dict]) -> list[dict]:
    """Ожидаемый ответ листа и сводов по эталону старого пути. Фаза B (Task 13) меняет тело на strip_nulls."""
    return rows


def get_ok(client, url, headers, params=None) -> list[dict]:
    r = client.get(url, headers=headers, params=params)
    assert r.status_code == 200, r.text
    assert r.headers["content-type"] == "application/json"
    assert float(r.headers["x-db-time-ms"]) >= 0
    body = r.json()
    assert list(body) == ["success", "message", "result", "count"]
    assert body["success"] is True and body["message"] is None
    assert body["count"] == len(body["result"])
    for row in body["result"]:
        assert row.keys() == {"row_type", "depth", "sort_order", "data"}
    return body["result"]


def test_openapi_heavy_get_schema_unchanged():
    spec = app.openapi()
    for path in HEAVY_GET_PATHS:
        schema = spec["paths"][path]["get"]["responses"]["200"]["content"]["application/json"]["schema"]
        assert schema == {"$ref": OPENAPI_REF}, path


SHEET_CASES = [
    (1, "AHR", "Support", None),        # FORM_1, лист с direction
    (4, "AHR", None, None),             # FORM_2, без direction
    (4, "AHR", None, ["plan", "q1"]),   # FORM_2, с sections
]


def _sheet_params(direction, sections):
    params = {}
    if direction:
        params["direction"] = direction
    if sections:
        params["sections"] = ",".join(sections)
    return params


@pytest.mark.parametrize("form_id,sheet,direction,sections", SHEET_CASES)
def test_sheet_matches_old_path(client, db_session, admin_tokens, auth_headers, form_id, sheet, direction, sections):
    result = get_ok(client, f"/api/v1/form/{form_id}/sheet/{sheet}", auth_headers(admin_tokens), _sheet_params(direction, sections))
    old = old_rows(client, lambda: SheetRepository(db_session).get(form_id=form_id, sheet=sheet, direction=direction, sections=sections))
    assert result
    assert result[0]["row_type"] == "META" and result[0]["depth"] == -1
    assert result == expected_sheet(old)


TRICKY = 'Кавычка " и \\ слэш\nперенос\tтаб'


def test_sheet_text_escaping_matches_old_path(client, db_session, admin_tokens, auth_headers):
    # В фикстуре у FORM_2 нет budget_line: вставляем строку с TRICKY в name (header.name листа).
    # justification в ответе листа не выводится; вставка откатывается вместе с сессией теста.
    async def _set():
        await db_session.execute(
            text(
                "INSERT INTO v3.budget_line (budget_form_id, expense_item_id, name, direction) "
                "SELECT 4, t.id, :t, 'Support' FROM v3.mv_expense_item_tree t "
                "JOIN v3.expense_item_form_type eft ON eft.expense_item_id = t.id "
                "WHERE eft.form_type_code = 'FORM_2' AND t.sheet = 'AHR' ORDER BY t.id LIMIT 1"
            ),
            {"t": TRICKY},
        )
    client.portal.call(_set)
    result = get_ok(client, "/api/v1/form/4/sheet/AHR", auth_headers(admin_tokens))
    old = old_rows(client, lambda: SheetRepository(db_session).get(form_id=4, sheet="AHR", direction=None))
    assert result == expected_sheet(old)
    assert any((r["data"].get("header") or {}).get("name") == TRICKY for r in result)


def test_heavy_get_gzip_compressed(client, admin_tokens, auth_headers):
    """Спека, сценарий «Сжатие»: ответ >= 1000 байт (GZipMiddleware minimum_size=1000) уходит с Content-Encoding: gzip."""
    headers = {**auth_headers(admin_tokens), "Accept-Encoding": "gzip"}
    r = client.get("/api/v1/form/4/sheet/AHR", headers=headers)
    assert r.status_code == 200, r.text
    assert r.headers.get("content-encoding") == "gzip"
    assert r.json()["success"] is True  # клиент распаковал тело


def test_sheet_get_does_not_use_old_path(client, admin_tokens, auth_headers, monkeypatch):
    async def boom(*args, **kwargs):
        raise AssertionError("GET листа не должен вызывать SheetRepository.get")
    monkeypatch.setattr(SheetRepository, "get", boom)
    get_ok(client, "/api/v1/form/4/sheet/AHR", auth_headers(admin_tokens))


# Свод 2 берёт строки по версиям v3.v_svod2_go / v3.v_svod2_rf (миграция 0049, SQL-файл 0049_summary2_related_records.sql):
#   GO - budget_form FORM_1/FORM_4 за год, RF (с 0041) - только FORM_2; статья с sheet IN (AHR, CAP, OPER);
#   строка попадает в свод, если у неё есть хотя бы одна "присутствующая" секция (например, plan с ненулевым планом).
# Хардкод (предположение о fixture.sql, T4.2): орг. единицы 1 (Test_SSP) и 2 (Test РФ) - единственные в фикстуре;
# в фикстуре за YEAR есть только FORM_2 (budget_form id=4, org 2) без budget_line; статьи 88 (AHR, R061161003),
# 523 (CAP, 3.00.0.), 729 (OPER) существуют в справочнике и имеют разные section_code - это нужно, чтобы
# sort_by=section_code и offset/limit давали различимый порядок. Если fixture.sql или справочник изменятся,
# test_svod2_fixture_not_empty упадёт первым.
SVOD2_ORG_IDS = (1, 2)
SVOD2_EXPENSE_ITEM_IDS = (88, 523, 729)  # AHR, CAP, OPER
SVOD2_FORMS = [("FORM_1", 1), ("FORM_1", 2), ("FORM_2", 1), ("FORM_2", 2)]  # GO: FORM_1, RF: FORM_2
SVOD2_MIN_ROWS = 6  # 2 орг. единицы x 3 статьи на каждый из листов GO и RF


@pytest.fixture
def svod2_plan_line(client, db_session):
    """Фикстура не даёт свода 2 строк за YEAR: достраиваем формы FORM_1 и FORM_2 по двум орг. единицам
    и строки трёх статей (AHR, CAP, OPER) с ненулевым планом. Откатывается вместе с сессией теста."""
    async def _set():
        for form_type, org_id in SVOD2_FORMS:
            await db_session.execute(
                text(
                    "INSERT INTO v3.budget_form (form_type_code, year, org_unit_id) "
                    "VALUES (:ft, :year, :org) ON CONFLICT (form_type_code, org_unit_id, year) DO NOTHING"
                ),
                {"ft": form_type, "year": YEAR, "org": org_id},
            )
        await db_session.execute(
            text(
                "INSERT INTO v3.budget_line (budget_form_id, expense_item_id) "
                "SELECT bf.id, e FROM v3.budget_form bf CROSS JOIN unnest(CAST(:items AS int[])) AS e "
                "WHERE bf.year = :year AND bf.form_type_code IN ('FORM_1', 'FORM_2') "
                "AND NOT EXISTS (SELECT 1 FROM v3.budget_line x "
                "WHERE x.budget_form_id = bf.id AND x.expense_item_id = e)"
            ),
            {"year": YEAR, "items": list(SVOD2_EXPENSE_ITEM_IDS)},
        )
        await db_session.execute(
            text(
                "INSERT INTO v3.plan (line_id, plan_q1) "
                "SELECT bl.id, 1000 FROM v3.budget_line bl JOIN v3.budget_form bf ON bf.id = bl.budget_form_id "
                "WHERE bf.year = :year AND bf.form_type_code IN ('FORM_1', 'FORM_2') "
                "AND bl.expense_item_id = ANY(CAST(:items AS int[])) "
                "ON CONFLICT (line_id) DO UPDATE SET plan_q1 = 1000"
            ),
            {"year": YEAR, "items": list(SVOD2_EXPENSE_ITEM_IDS)},
        )
    client.portal.call(_set)


@pytest.mark.parametrize("sheet", ["FORM_1", "FORM_2", "FORM_3", "FORM_4", "MAIN"])
def test_svod1_matches_old_path(client, db_session, admin_tokens, auth_headers, sheet):
    result = get_ok(client, f"/api/v1/svod/1/{sheet}/{YEAR}", auth_headers(admin_tokens))
    old = old_rows(client, lambda: SummaryRepository(db_session).get_svod1_rows(year=YEAR, sheet=sheet, org_ids=None))
    assert result == expected_sheet(old)


def test_svod1_org_ids_matches_old_path(client, db_session, admin_tokens, auth_headers):
    result = get_ok(client, f"/api/v1/svod/1/FORM_1/{YEAR}", auth_headers(admin_tokens), {"org_ids": "1,2,3"})
    old = old_rows(client, lambda: SummaryRepository(db_session).get_svod1_rows(year=YEAR, sheet="FORM_1", org_ids=[1, 2, 3]))
    assert result
    assert result == expected_sheet(old)


SVOD2_CASES = [
    ("GO", {}, {}),
    ("RF", {}, {}),
    ("GO", {"org_ids": "1,2"}, {"org_ids": [1, 2]}),
    ("GO", {"offset": 1, "limit": 2}, {"offset": 1, "limit": 2}),
    ("RF", {"sort_by": "section_code", "sort_direction": "desc"}, {"sort_by": "section_code", "sort_direction": "desc"}),
]


def _svod2_min_rows(query: dict) -> int:
    if "limit" in query:
        return 1
    if "sort_by" in query:
        return 2
    return 1


@pytest.mark.usefixtures("svod2_plan_line")
@pytest.mark.parametrize("sheet,query,repo_kwargs", SVOD2_CASES)
def test_svod2_matches_old_path(client, db_session, admin_tokens, auth_headers, sheet, query, repo_kwargs):
    result = get_ok(client, f"/api/v1/svod/2/{sheet}/{YEAR}", auth_headers(admin_tokens), query)
    old = old_rows(client, lambda: SummaryRepository(db_session).get_svod2_rows(year=YEAR, sheet=sheet, **repo_kwargs))
    assert len(result) >= _svod2_min_rows(query), "сравнение на пустом/слишком коротком списке ничего не доказывает"
    assert result == expected_sheet(old)


@pytest.mark.usefixtures("svod2_plan_line")
def test_svod2_fixture_not_empty(client, admin_tokens, auth_headers):
    """Иначе сравнения свода 2 выше проходят на пустых списках."""
    go = get_ok(client, f"/api/v1/svod/2/GO/{YEAR}", auth_headers(admin_tokens))
    rf = get_ok(client, f"/api/v1/svod/2/RF/{YEAR}", auth_headers(admin_tokens))
    assert len(go) >= SVOD2_MIN_ROWS, len(go)
    assert len(rf) >= SVOD2_MIN_ROWS, len(rf)
    for rows in (go, rf):
        assert {r["data"]["header"]["source_org_unit_id"] for r in rows} == set(SVOD2_ORG_IDS)
        assert len({r["data"]["header"]["section_code"] for r in rows}) >= 2


@pytest.mark.usefixtures("svod2_plan_line")
def test_svod2_isp_matches_service_old_path(client, db_session, isp_tokens, auth_headers):
    result = get_ok(client, f"/api/v1/svod/2/GO/{YEAR}", auth_headers(isp_tokens))
    old = old_rows(client, lambda: SummaryService(db_session).get_svod2_rows(year=YEAR, sheet="GO", user=ISP))
    assert result == expected_sheet(old)


def test_svod2_foreign_org_ids_empty_body(client, isp_tokens, auth_headers):
    r = client.get(f"/api/v1/svod/2/GO/{YEAR}", headers=auth_headers(isp_tokens), params={"org_ids": "999999"})
    assert r.status_code == 200
    assert r.json() == {"success": True, "message": None, "result": [], "count": 0}
    assert float(r.headers["x-db-time-ms"]) >= 0


@pytest.mark.parametrize("url,method", [
    (f"/api/v1/svod/1/MAIN/{YEAR}", "get_svod1_rows"),
    (f"/api/v1/svod/2/GO/{YEAR}", "get_svod2_rows"),
])
def test_svod_get_does_not_use_old_path(client, admin_tokens, auth_headers, monkeypatch, url, method):
    async def boom(*args, **kwargs):
        raise AssertionError(f"GET свода не должен вызывать SummaryRepository.{method}")
    monkeypatch.setattr(SummaryRepository, method, boom)
    get_ok(client, url, auth_headers(admin_tokens))


def _create_project(client, headers, year=YEAR, branch_id=1) -> int:
    r = client.post(
        "/api/v1/projects",
        json={
            "name": f"PT_{uuid.uuid4().hex[:10]}",
            "year": year,
            "branch_id": branch_id,
            "project_type": "Открытие ВСП",
            "vsp_format": "Типовой",
            "placement_type": "Собственность",
            "object_address": "Тестовая улица 1",
        },
        headers=headers,
    )
    assert r.status_code == 200, r.text
    return r.json()["project_id"]


@pytest.fixture
def project_id(client, admin_tokens, auth_headers) -> int:
    """Новый проект сразу получает отчёт LIMIT со всеми строками статей (POST строки на нём даёт 400/409
    и ломает транзакцию теста), так что отдельная строка не нужна."""
    return _create_project(client, auth_headers(admin_tokens))


@pytest.mark.parametrize("sections", [None, "q1,year"])
def test_project_report_matches_old_path(client, db_session, admin_tokens, auth_headers, project_id, sections):
    pid = project_id
    result = get_ok(client, f"/api/v1/projects/{pid}/report/{YEAR}/LIMIT", auth_headers(admin_tokens),
                    {"sections": sections} if sections else None)
    old = old_rows(client, lambda: ProjectService(db_session).get_report_rows(
        project_id=pid, year=YEAR, report_type="LIMIT",
        sections=sections.split(",") if sections else None, user=ADMIN))
    assert result == old
    assert any(row["row_type"] == "INPUT" for row in result)


def test_rf_rollup_pinned_preexisting_sql_error(client, admin_tokens, auth_headers):
    """Фиксирует СУЩЕСТВУЮЩИЙ дефект вне AURORA-1848: v3.v_form3_rf_rollup_jsonb в миграции 0002_functions.sql
    ссылается на несуществующий r.year, поэтому на любой БД ответ - 400 с текстом SQL-ошибки (так же отвечал
    и старый путь). Когда дефект будет исправлен, этот тест упадёт и его нужно превратить в тест
    эквивалентности с эталоном ProjectService.get_rf_rollup_rows (как остальные get_ok-сравнения)."""
    r = client.get(f"/api/v1/rf-rollup/1/{YEAR}", headers=auth_headers(admin_tokens))
    assert (r.status_code, r.json()) == RF_ROLLUP_SQL_ERROR


def test_project_summary_matches_old_path(client, db_session, admin_tokens, auth_headers, project_id):
    pid = project_id
    result = get_ok(client, f"/api/v1/projects/{pid}/summary", auth_headers(admin_tokens))
    old = old_rows(client, lambda: ProjectService(db_session).get_project_summary_rows(project_id=pid, user=ADMIN))
    assert result == old


@pytest.mark.parametrize("kind", ["plan", "fact", "detail"])
def test_projects_summaries_match_old_path(client, db_session, admin_tokens, auth_headers, project_id, kind):
    result = get_ok(client, f"/api/v1/projects/{kind}-summary", auth_headers(admin_tokens), {"year": YEAR})
    method = getattr(ProjectService(db_session), f"get_{kind}_summary_rows")
    old = old_rows(client, lambda: method(year=YEAR, branch_ids=None, user=ADMIN))
    assert result == old


def test_projects_summaries_not_all_empty(client, admin_tokens, auth_headers, project_id):
    pid = project_id
    got = [get_ok(client, f"/api/v1/projects/{k}-summary", auth_headers(admin_tokens), {"year": YEAR})
           for k in ("plan", "fact", "detail")]
    got.append(get_ok(client, f"/api/v1/projects/{pid}/summary", auth_headers(admin_tokens)))
    assert all(got), [bool(g) for g in got]


GET_METHODS = [
    ("/api/v1/projects/{pid}/report/%d/LIMIT" % YEAR, "get_report_rows"),
    ("/api/v1/rf-rollup/1/%d" % YEAR, "get_rf_rollup_rows"),
    ("/api/v1/projects/{pid}/summary", "get_project_summary_rows"),
    ("/api/v1/projects/plan-summary?year=%d" % YEAR, "get_plan_summary_rows"),
    ("/api/v1/projects/fact-summary?year=%d" % YEAR, "get_fact_summary_rows"),
    ("/api/v1/projects/detail-summary?year=%d" % YEAR, "get_detail_summary_rows"),
]


@pytest.mark.parametrize("url,method", GET_METHODS)
def test_projects_get_does_not_use_old_path(client, admin_tokens, auth_headers, project_id, monkeypatch, url, method):
    called = []

    async def boom(*args, **kwargs):
        called.append(method)
        raise AssertionError(f"GET не должен вызывать ProjectRepository.{method}")
    monkeypatch.setattr(ProjectRepository, method, boom)
    r = client.get(url.format(pid=project_id), headers=auth_headers(admin_tokens))
    assert not called
    if method != "get_rf_rollup_rows":  # rf-rollup: см. test_rf_rollup_pinned_preexisting_sql_error
        assert r.status_code == 200, r.text


# Характеризующие тесты ошибок: литералы сняты со старого кода (до правки роутеров).
def test_project_report_not_found_unchanged(client, admin_tokens, auth_headers, project_id):
    pid = project_id
    r = client.get(f"/api/v1/projects/{pid}/report/2099/LIMIT", headers=auth_headers(admin_tokens))
    assert (r.status_code, r.json()) == REPORT_NOT_FOUND


def test_rf_rollup_foreign_branch_unchanged(client, isp_tokens, auth_headers):
    r = client.get(f"/api/v1/rf-rollup/1/{YEAR}", headers=auth_headers(isp_tokens))
    assert (r.status_code, r.json()) == RF_FOREIGN