api/tests/integration/test_cell_formula_ws_integration.py

@@ -16,13 +16,13 @@ def _roundtrip(ws, event: str, data: dict) -> dict:
     return ws.receive_json()
 
 
-def _db_scalar(sql: str, commit: bool = False):
+def _db_scalar(sql: str, commit: bool = False, **params):
     """Отдельная сессия на собственном движке (WS коммитит в БД)."""
     async def run():
         engine = create_async_engine(settings.DATABASE_URL, poolclass=NullPool)
         try:
             async with engine.connect() as conn:
-                res = await conn.execute(text(sql))
+                res = await conn.execute(text(sql), params)
                 row = res.first() if res.returns_rows else None
                 if commit:
                     await conn.commit()
@@ -33,6 +33,14 @@ def _db_scalar(sql: str, commit: bool = False):
     return asyncio.run(run())
 
 
+PLAN_Q1_SQL = "SELECT plan_q1 FROM v3.plan WHERE line_id = 1"
+
+
+def _restore_cell(prior):
+    """Вернуть plan.q1 строки 1 к значению до теста (NULL -> None) прямо в БД."""
+    _db_scalar("UPDATE v3.plan SET plan_q1 = :v WHERE line_id = 1", commit=True, v=prior)
+
+
 def _cleanup_formula():
     _db_scalar(
         "DELETE FROM v3.cell_formula WHERE budget_form_id = 1 "
@@ -51,6 +59,7 @@ def _edit(ws, data: dict) -> dict:
 def test_ws_cell_updated_carries_saved_formula(client, admin_tokens):
     with client.websocket_connect(WS_URL) as ws:
         ws.send_json({"event": "user_login", "data": {"token": admin_tokens["access_token"]}})
+        prior = _db_scalar(PLAN_Q1_SQL)
         try:
             updated = _edit(ws, {"value": 150, "formula": "100+50"})
             assert not updated.get("error"), updated
@@ -65,15 +74,14 @@ def test_ws_cell_updated_carries_saved_formula(client, admin_tokens):
             assert plain["data"]["formula"] is None
             assert plain["data"]["formula_meta"] is None
         finally:
-            _roundtrip(ws, "cell_edit_start", CELL)
-            _roundtrip(ws, "cell_updated", {**CELL, "value": None})
-            _roundtrip(ws, "cell_edit_end", CELL)
             _cleanup_formula()
+            _restore_cell(prior)
 
 
 def test_ws_broadcasts_normalized_formula_not_client_string(client, admin_tokens):
     with client.websocket_connect(WS_URL) as ws:
         ws.send_json({"event": "user_login", "data": {"token": admin_tokens["access_token"]}})
+        prior = _db_scalar(PLAN_Q1_SQL)
         try:
             updated = _edit(ws, {"value": 150, "formula": " =100+50 "})
             assert not updated.get("error"), updated
@@ -84,16 +92,16 @@ def test_ws_broadcasts_normalized_formula_not_client_string(client, admin_tokens
             )
             assert stored == "100+50"
         finally:
-            _roundtrip(ws, "cell_edit_start", CELL)
-            _roundtrip(ws, "cell_updated", {**CELL, "value": None})
-            _roundtrip(ws, "cell_edit_end", CELL)
             _cleanup_formula()
+            _restore_cell(prior)
 
 
 def test_ws_invalid_formula_returns_error_and_does_not_commit(client, admin_tokens):
     with client.websocket_connect(WS_URL) as ws:
         ws.send_json({"event": "user_login", "data": {"token": admin_tokens["access_token"]}})
+        prior = _db_scalar(PLAN_Q1_SQL)
         try:
+            value_before = _db_scalar(PLAN_Q1_SQL)
             before = _db_scalar(
                 "SELECT count(*) FROM v3.cell_formula WHERE budget_form_id = 1 "
                 "AND line_id = 1 AND column_key = 'plan.q1'"
@@ -108,11 +116,75 @@ def test_ws_invalid_formula_returns_error_and_does_not_commit(client, admin_toke
                 "AND line_id = 1 AND column_key = 'plan.q1'"
             )
             assert after == before
-            # значение не закоммичено: следующий успешный ответ не видит 999
+            # значение не закоммичено: ячейка в БД не изменилась
+            assert _db_scalar(PLAN_Q1_SQL) == value_before
             ok = _edit(ws, {"value": 5})
             assert not ok.get("error"), ok
         finally:
-            _roundtrip(ws, "cell_edit_start", CELL)
-            _roundtrip(ws, "cell_updated", {**CELL, "value": None})
-            _roundtrip(ws, "cell_edit_end", CELL)
             _cleanup_formula()
+            _restore_cell(prior)
+
+
+# ---- Отчёт проекта (ProjectEventProcess) ----
+
+PROJECT_ID = 2
+YEAR = 2026
+PROJECT_WS_URL = f"/api/v1/ws/projects/{PROJECT_ID}/report/{YEAR}/LIMIT"
+LIMIT_REPORT_SQL = (
+    "SELECT r.id FROM v3.rf_project_report r "
+    "JOIN v3.rf_project_year py ON py.id = r.rf_project_year_id "
+    "WHERE r.project_id = :pid AND r.report_type = 'LIMIT' AND py.year = :y"
+)
+
+
+def test_ws_project_report_cell_updated_carries_saved_formula(client, admin_tokens):
+    """ProjectEventProcess.__update_cell: formula и formula_meta в рассылке."""
+    report_id = _db_scalar(LIMIT_REPORT_SQL, pid=PROJECT_ID, y=YEAR)
+    assert report_id, "в фикстурной БД нет LIMIT-отчёта проекта 2"
+    line_id = None
+    try:
+        # строк отчёта в фикстуре нет: создаём свою (удаляется в finally)
+        line_id = _db_scalar(
+            "INSERT INTO v3.rf_project_report_line (rf_project_report_id, expense_item_id) "
+            "SELECT :rid, id FROM v3.expense_item ORDER BY id LIMIT 1 RETURNING id",
+            commit=True,
+            rid=report_id,
+        )
+        cell = {"line_id": line_id, "column": "q1.m1"}
+        qsql = (
+            "SELECT actual_m1 FROM v3.rf_project_report_quarter "
+            f"WHERE rf_project_report_line_id = {line_id} AND quarter = 1"
+        )
+
+        def edit(ws, data):
+            assert not _roundtrip(ws, "cell_edit_start", cell).get("error")
+            res = _roundtrip(ws, "cell_updated", {**cell, **data})
+            _roundtrip(ws, "cell_edit_end", cell)
+            return res
+
+        with client.websocket_connect(PROJECT_WS_URL) as ws:
+            ws.send_json({"event": "user_login", "data": {"token": admin_tokens["access_token"]}})
+            updated = edit(ws, {"value": 150, "formula": "100+50"})
+            assert not updated.get("error"), updated
+            assert updated["data"]["formula"] == "100+50"
+            meta = updated["data"]["formula_meta"]
+            assert meta["updated_by"]["id"] == _db_scalar(
+                "SELECT id FROM v3.app_user WHERE username = 'admin'"
+            )
+            assert isinstance(meta["updated_at"], str)
+            assert _db_scalar(qsql) == 150
+
+            plain = edit(ws, {"value": 7})
+            assert not plain.get("error"), plain
+            assert plain["data"]["formula"] is None
+            assert plain["data"]["formula_meta"] is None
+    finally:
+        if line_id is not None:
+            # prior-значения нет: строка создана тестом; удаляем её вместе с формулами
+            _db_scalar(
+                "DELETE FROM v3.cell_formula WHERE rf_project_report_id = :rid "
+                "AND line_id = :l", commit=True, rid=report_id, l=line_id)
+            _db_scalar(
+                "DELETE FROM v3.rf_project_report_quarter WHERE rf_project_report_line_id = :l",
+                commit=True, l=line_id)
+            _db_scalar("DELETE FROM v3.rf_project_report_line WHERE id = :l", commit=True, l=line_id)