test(formulas): правки minor-находок T1.1–T4.2 (AURORA-1896)

T1.1 каскад hard_delete проверяется по AHR_LIMIT; T1.2 тесты триггеров
rent/utility/security/fixed_asset_report; T1.3 NaN в set_cell_formulas
отклоняется (0059 правится на месте); T3.1 unit-кейсы asyncpg-обёртки;
T3.2 предусловие SAVEPOINT-теста и комментарий к 403; T4.1 проверка
незаписи значения при 422 отчёта проекта; T4.2 убрана мёртвая логика.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

api/alembic/versions/sql/0059_cell_formula.sql

@@ -106,7 +106,7 @@ BEGIN
         EXCEPTION WHEN others THEN
             RAISE EXCEPTION 'invalid_formula: value';
         END;
-        IF v_value IS NULL THEN
+        IF v_value IS NULL OR v_value = 'NaN'::numeric THEN
             RAISE EXCEPTION 'invalid_formula: value';
         END IF;
 

api/tests/integration/test_cell_formula_integration.py

@@ -97,24 +97,51 @@ def test_invalid_formula_does_not_write_value(client, headers, line_col, value,
     from src.services.sheet_service import SheetService
 
     line_id = _add_line(client, headers)
-    before = _cell_value(client, headers, line_id, line_col)
+    # исходное значение отличается от пробного, иначе «значение не изменилось» ничего не доказывает
+    def _read():
+        if line_col == "header.justification":  # текстовой колонки нет в JSON листа: читаем из БД
+            return _scalar(client, "SELECT justification FROM v3.budget_line WHERE id = :l", {"l": line_id})
+        return _cell_value(client, headers, line_id, line_col)
 
-    async def _run():
+    seed = "исходный" if line_col.startswith("header.") else 5
+    seed_r = _patch(client, headers, {"line_id": line_id, "column": line_col, "value": seed})
+    assert seed_r.status_code == 200, seed_r.text
+    before = _read()
+    assert before == seed and before != value
+
+    from src.db.models.app_user import AppUser
+    from src.services.sheet_service import SheetService
+
+    async def _update(formula):
         db = client.db_session
         user = await db.get(AppUser, 1)
+        await SheetService(db).update_cell(
+            form_id=1, sheet="AHR", direction="Support", sections=["plan", "contract_summary"],
+            line_id=line_id, column=line_col, value=value, formula=formula, user=user,
+        )
+
+    async def _precondition():
+        # тот же вызов без формулы пишет значение: после отката SAVEPOINT его снова нет
+        nested = await client.db_session.begin_nested()
+        await _update(None)
+        return nested
+
+    nested = client.portal.call(_precondition)
+    assert _read() != before, "без формулы значение должно записаться"
+    client.portal.call(nested.rollback)
+    assert _read() == before
+
+    async def _run():
         try:
-            async with db.begin_nested():
+            async with client.db_session.begin_nested():
-                await SheetService(db).update_cell(
+                await _update(formula)
-                    form_id=1, sheet="AHR", direction="Support", sections=["plan", "contract_summary"],
-                    line_id=line_id, column=line_col, value=value, formula=formula, user=user,
-                )
         except DBAPIError as exc:
             return str(exc.orig)
         return None
 
     err = client.portal.call(_run)
     assert err is not None and "invalid_formula" in err
-    assert _cell_value(client, headers, line_id, line_col) == before
+    assert _read() == before
     assert _line_formulas(client, headers, line_id) == []
 
 
@@ -137,6 +164,8 @@ def test_patch_formula_null_value_422(client, headers):
 def test_patch_formula_forbidden_403_no_formula(client, headers, isp2_tokens, auth_headers):
     # isp2 в оргединице формы 1, но колонку plan.q2 ему править нельзя (role_not_allowed).
     # isp1 форму 1 не видит вовсе (404), поэтому для 403 берём isp2.
+    # Опирается на порядок записи (проверка прав до записи формулы) и откат сессии фикстурой:
+    # если бы формула успела записаться, GET ниже увидел бы её в той же сессии.
     r = client.patch(
         f"{FORM}/sheet/AHR/cell{QS}",
         json={"line_id": 1, "column": "plan.q2", "value": 5, "formula": "2+3"},
@@ -295,6 +324,52 @@ def test_project_report_invalid_formula_422(client, headers):
     year = datetime.now().year
     pid = _create_project(client, headers, year)
     line_id = _add_project_line(client, headers, pid, year)
+    qsql = "SELECT actual_m1 FROM v3.rf_project_report_quarter WHERE rf_project_report_line_id = :l AND quarter = 1"
+    before = _scalar(client, qsql, {"l": line_id})
+    # HTTP-ошибка откатывает всю сессию теста вместе с созданным проектом, поэтому «значение не записано»
+    # проверяем до неё: тот же вызов сервиса внутри SAVEPOINT (с предусловием: без формулы значение пишется).
+    from sqlalchemy.exc import DBAPIError
+
+    from src.db.models.app_user import AppUser
+    from src.services.project_service import ProjectService
+
+    async def _call(formula):
+        db = client.db_session
+        user = await db.get(AppUser, 1)
+        await ProjectService(db).upd_form3_cell(
+            project_id=pid, year=year, report_type="LIMIT", line_id=line_id,
+            column="q1.m1", value=5, user=user, formula=formula,
+        )
+
+    async def _precondition():
+        nested = await client.db_session.begin_nested()
+        await _call(None)
+        return nested
+
+    nested = client.portal.call(_precondition)
+    assert _scalar(client, qsql, {"l": line_id}) == 5, "без формулы значение должно записаться"
+    client.portal.call(nested.rollback)
+    assert _scalar(client, qsql, {"l": line_id}) == before
+
+    async def _run():
+        try:
+            async with client.db_session.begin_nested():
+                await _call("drop table")
+        except DBAPIError as exc:
+            return str(exc.orig)
+        return None
+
+    err = client.portal.call(_run)
+    assert err is not None and "invalid_formula" in err
+    assert _scalar(client, qsql, {"l": line_id}) == before
+    assert _scalar(
+        client,
+        "SELECT count(*) FROM v3.cell_formula WHERE rf_project_report_id IN "
+        "(SELECT id FROM v3.rf_project_report WHERE project_id = :pid)",
+        {"pid": pid},
+    ) == 0
+
+    # а контракт REST: 422 (после него сессия откатывается, проверять БД дальше нельзя)
     r = client.patch(
         f"{_report_url(pid, year)}/cell",
         json={"line_id": line_id, "column": "q1.m1", "value": 5, "formula": "drop table"},
@@ -368,8 +443,7 @@ def test_rf_project_report_line_trigger_direct_delete(client, headers):
     pid = _create_project(client, headers, year)
     l1 = _add_project_line(client, headers, pid, year)
     l2 = _add_project_line(client, headers, pid, year, nth=1)
-    other = [l2]
+    for line in (l1, l2):
-    for line in [l1] + other[:1]:
         assert client.patch(
             f"{_report_url(pid, year)}/cell",
             json={"line_id": line, "column": "q1.m1", "value": 3, "formula": "1+2"},
@@ -384,5 +458,4 @@ def test_rf_project_report_line_trigger_direct_delete(client, headers):
     client.portal.call(_delete_direct)
     left = {f["line_id"] for f in _project_formulas(client, headers, pid, year)}
     assert l1 not in left
-    if other:
+    assert l2 in left
-        assert other[0] in left

api/tests/integration/test_cell_formula_sql.py

@@ -87,7 +87,7 @@ def test_empty_after_normalization_is_no_formula(svod2_sql):
 
 @pytest.mark.parametrize(
     "formula,value",
-    [("drop table", 1), ("1" * 501, 1), ("1+1", None), ("1+1", "abc")],
+    [("drop table", 1), ("1" * 501, 1), ("1+1", None), ("1+1", "abc"), ("1+1", "NaN")],
 )
 def test_invalid_formula_raises(svod2_sql, formula, value):
     admin, line = _admin(svod2_sql), _line(svod2_sql)
@@ -127,12 +127,66 @@ def test_trigger_budget_line_delete(svod2_sql):
 
 
 def test_cascade_hard_delete_form(svod2_sql):
+    """Формула AHR_LIMIT (триггеры её не трогают) уходит только каскадом FK при hard_delete_budget_form;
+    формула AHR на budget_line может быть снята и триггером, поэтому каскад проверяем по AHR_LIMIT."""
     admin, line = _admin(svod2_sql), _line(svod2_sql)
+    tpl = svod2_sql("SELECT id FROM v3.limit_template WHERE row_type = 'LEAF' LIMIT 1")
+    if not tpl:
+        pytest.skip("нет limit_template LEAF")
     _set(svod2_sql, 1, "AHR", [_item(line, "100+50")], admin)
+    _set(svod2_sql, 1, "AHR_LIMIT", [_item(tpl[0][0], "100+50")], admin)
+    assert len(_list(svod2_sql, 1, "AHR_LIMIT")) == 1
     svod2_sql("SELECT v3.hard_delete_budget_form(1)")
+    assert svod2_sql("SELECT count(*) FROM v3.cell_formula WHERE budget_form_id = 1 AND sheet = 'AHR_LIMIT'")[0][0] == 0
     assert svod2_sql("SELECT count(*) FROM v3.cell_formula WHERE budget_form_id = 1")[0][0] == 0
 
 
+def _vsp(sql):
+    rows = sql("SELECT id FROM v3.vsp ORDER BY id LIMIT 1")
+    if not rows:
+        pytest.skip("нет vsp")
+    return rows[0][0]
+
+
+@pytest.mark.parametrize(
+    "table,sheet",
+    [("rent_detail", "AHR_RENT"), ("utility_detail", "AHR_UTILITY"), ("security_detail", "AHR_SECURITY")],
+)
+def test_trigger_detail_delete(svod2_sql, table, sheet):
+    """Глобальные id (D4): DELETE строки detail-таблицы чистит формулу только этой строки."""
+    admin, line, vsp = _admin(svod2_sql), _line(svod2_sql), _vsp(svod2_sql)
+    ids = [
+        svod2_sql(
+            f"INSERT INTO v3.{table} (line_id, vsp_id) VALUES (:l, :v) RETURNING id",
+            {"l": line, "v": vsp},
+        )[0][0]
+        for _ in range(2)
+    ]
+    for i in ids:
+        _set(svod2_sql, 1, sheet, [_item(i, "1+1", value=2)], admin)
+    svod2_sql(f"DELETE FROM v3.{table} WHERE id = :i", {"i": ids[0]})
+    left = {x.line_id for x in _list(svod2_sql, 1, sheet)}
+    assert ids[0] not in left
+    assert ids[1] in left
+
+
+def test_trigger_fixed_asset_report_delete(svod2_sql):
+    admin = _admin(svod2_sql)
+    ids = [
+        svod2_sql(
+            "INSERT INTO v3.fixed_asset_report (budget_form_id, month) VALUES (1, :m) RETURNING id",
+            {"m": m},
+        )[0][0]
+        for m in (1, 2)
+    ]
+    for i in ids:
+        _set(svod2_sql, 1, "OTCH9F", [_item(i, "1+1", value=2)], admin)
+    svod2_sql("DELETE FROM v3.fixed_asset_report WHERE id = :i", {"i": ids[0]})
+    left = {x.line_id for x in _list(svod2_sql, 1, "OTCH9F")}
+    assert ids[0] not in left
+    assert ids[1] in left
+
+
 def test_owner_check(svod2_sql):
     admin = _admin(svod2_sql)
     with pytest.raises(Exception, match="invalid_formula: owner"):

api/tests/unit/test_exception_handlers.py

@@ -50,6 +50,28 @@ def test_map_sqlalchemy_error_validation_prefix():
     assert field == "q5.plan"
 
 
+def test_map_sqlalchemy_error_validation_prefix_asyncpg_wrapped():
+    exc = SQLAlchemyError()
+    exc.orig = Exception("<class 'asyncpg.exceptions.RaiseError'>: invalid_formula: chars")
+
+    status_code, code, message, field = map_sqlalchemy_error(exc)
+
+    assert status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
+    assert code == status.HTTP_422_UNPROCESSABLE_ENTITY
+    assert message == "invalid_formula: chars"
+    assert "asyncpg" not in message
+    assert field == "chars"
+
+
+def test_map_sqlalchemy_error_asyncpg_wrapped_non_validation_keeps_message():
+    exc = SQLAlchemyError()
+    exc.orig = Exception("<class 'asyncpg.exceptions.RaiseError'>: something else")
+
+    status_code, _, message, _ = map_sqlalchemy_error(exc)
+
+    assert status_code != status.HTTP_422_UNPROCESSABLE_ENTITY
+
+
 def test_map_sqlalchemy_error_not_found():
     exc = SQLAlchemyError()
     exc.orig = Exception("budget_form #99 не существует")