From 84dbaaf361f565f642cfeaca54e5795e25d738ee Mon Sep 17 00:00:00 2001 From: Andrey Epifancev Date: Wed, 11 Nov 2026 19:36:09 +0400 Subject: [PATCH] =?UTF-8?q?test(formulas):=20=D0=BF=D1=80=D0=B0=D0=B2?= =?UTF-8?q?=D0=BA=D0=B8=20minor-=D0=BD=D0=B0=D1=85=D0=BE=D0=B4=D0=BE=D0=BA?= =?UTF-8?q?=20T1.1=E2=80=93T4.2=20(AURORA-1896)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- .../versions/sql/0059_cell_formula.sql | 2 +- .../test_cell_formula_integration.py | 97 ++++++++++++++++--- .../integration/test_cell_formula_sql.py | 56 ++++++++++- api/tests/unit/test_exception_handlers.py | 22 +++++ 4 files changed, 163 insertions(+), 14 deletions(-) diff --git a/api/alembic/versions/sql/0059_cell_formula.sql b/api/alembic/versions/sql/0059_cell_formula.sql index 3f0c30d..6cac185 100644 --- a/api/alembic/versions/sql/0059_cell_formula.sql +++ b/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; diff --git a/api/tests/integration/test_cell_formula_integration.py b/api/tests/integration/test_cell_formula_integration.py index baee100..4dfe13e 100644 --- a/api/tests/integration/test_cell_formula_integration.py +++ b/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(): - 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 with client.db_session.begin_nested(): + await _update(formula) 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] + other[:1]: + for line in (l1, l2): 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 other[0] in left + assert l2 in left diff --git a/api/tests/integration/test_cell_formula_sql.py b/api/tests/integration/test_cell_formula_sql.py index 7b6a2c6..fd24c7e 100644 --- a/api/tests/integration/test_cell_formula_sql.py +++ b/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"): diff --git a/api/tests/unit/test_exception_handlers.py b/api/tests/unit/test_exception_handlers.py index 82c3c4a..703e717 100644 --- a/api/tests/unit/test_exception_handlers.py +++ b/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(": 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(": 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 не существует")