feat(formulas): formula и formula_meta в cell_updated по WebSocket (AURORA-1896)

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

api/src/api/v1/websocket.py

@@ -411,6 +411,20 @@ class ConnectionCache:
         self._forms[form_id] = self._clock()
 
 
+def _apply_saved_formula(event_data: dict, formula: dict | None) -> None:
+    """Перезаписывает formula/formula_meta в данных рассылки тем, что сохранил сервер."""
+    if formula:
+        updated_at = formula["updated_at"]
+        event_data["formula"] = formula["expression"]
+        event_data["formula_meta"] = {
+            "updated_by": formula["updated_by"],
+            "updated_at": updated_at.isoformat() if hasattr(updated_at, "isoformat") else updated_at,
+        }
+    else:
+        event_data["formula"] = None
+        event_data["formula_meta"] = None
+
+
 class FormEventProcess:
     def __init__(self, db: AsyncSession):
         self.sheet_service: SheetService = SheetService(db)
@@ -527,7 +541,7 @@ class FormEventProcess:
         if not await self._check_form_access(form_id=form_id, user=user):
             return None
 
-        return await self.sheet_service.update_cell(
+        rows, formula = await self.sheet_service.update_cell_with_formula(
             form_id=form_id,
             sheet=sheet,
             direction=direction,
@@ -535,8 +549,11 @@ class FormEventProcess:
             line_id=event_data["line_id"],
             column=event_data["column"],
             value=event_data["value"],
+            formula=event_data.get("formula"),
             user=user,
         )
+        _apply_saved_formula(event_data, formula)
+        return rows
 
     async def __add_row(
             self,
@@ -761,15 +778,18 @@ class ProjectEventProcess:
         if not project:
             return None
 
-        return await self.project_service.upd_form3_cell(
+        rows, formula = await self.project_service.upd_form3_cell_with_formula(
             project_id=project_id,
             year=year,
             report_type=report_type,
             line_id=event_data["line_id"],
             column=event_data["column"],
             value=event_data["value"],
+            formula=event_data.get("formula"),
             user=user,
         )
+        _apply_saved_formula(event_data, formula)
+        return rows
 
     async def __add_row(
             self,

api/tests/integration/test_cell_formula_ws_integration.py

@@ -0,0 +1,118 @@
+"""AURORA-1896: formula и formula_meta в cell_updated по WebSocket (лист формы)."""
+import asyncio
+
+from sqlalchemy import text
+from sqlalchemy.ext.asyncio import create_async_engine
+from sqlalchemy.pool import NullPool
+
+from src.core.config import settings
+
+WS_URL = "/api/v1/ws/form/1/sheet/AHR?direction=Support"
+CELL = {"line_id": 1, "column": "plan.q1"}
+
+
+def _roundtrip(ws, event: str, data: dict) -> dict:
+    ws.send_json({"event": event, "data": data})
+    return ws.receive_json()
+
+
+def _db_scalar(sql: str, commit: bool = False):
+    """Отдельная сессия на собственном движке (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))
+                row = res.first() if res.returns_rows else None
+                if commit:
+                    await conn.commit()
+                return row[0] if row else None
+        finally:
+            await engine.dispose()
+
+    return asyncio.run(run())
+
+
+def _cleanup_formula():
+    _db_scalar(
+        "DELETE FROM v3.cell_formula WHERE budget_form_id = 1 "
+        "AND line_id = 1 AND column_key = 'plan.q1'",
+        commit=True,
+    )
+
+
+def _edit(ws, data: dict) -> dict:
+    assert not _roundtrip(ws, "cell_edit_start", CELL).get("error")
+    updated = _roundtrip(ws, "cell_updated", {**CELL, **data})
+    _roundtrip(ws, "cell_edit_end", CELL)
+    return updated
+
+
+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"]}})
+        try:
+            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 isinstance(meta["updated_by"]["id"], int)
+            assert "full_name" in meta["updated_by"]
+            assert isinstance(meta["updated_at"], str)
+
+            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:
+            _roundtrip(ws, "cell_edit_start", CELL)
+            _roundtrip(ws, "cell_updated", {**CELL, "value": None})
+            _roundtrip(ws, "cell_edit_end", CELL)
+            _cleanup_formula()
+
+
+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"]}})
+        try:
+            updated = _edit(ws, {"value": 150, "formula": " =100+50 "})
+            assert not updated.get("error"), updated
+            assert updated["data"]["formula"] == "100+50"
+            stored = _db_scalar(
+                "SELECT expression FROM v3.cell_formula WHERE budget_form_id = 1 "
+                "AND line_id = 1 AND column_key = 'plan.q1'"
+            )
+            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()
+
+
+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"]}})
+        try:
+            before = _db_scalar(
+                "SELECT count(*) FROM v3.cell_formula WHERE budget_form_id = 1 "
+                "AND line_id = 1 AND column_key = 'plan.q1'"
+            )
+            updated = _edit(ws, {"value": 999, "formula": "abc+1"})
+            assert updated.get("error"), updated
+            assert updated["error"]["code"] == 422
+            assert updated["error"]["message"].startswith("invalid_formula")
+            assert "result" not in updated
+            after = _db_scalar(
+                "SELECT count(*) FROM v3.cell_formula WHERE budget_form_id = 1 "
+                "AND line_id = 1 AND column_key = 'plan.q1'"
+            )
+            assert after == before
+            # значение не закоммичено: следующий успешный ответ не видит 999
+            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()

api/tests/unit/test_websocket_connection_cache.py

@@ -11,7 +11,7 @@ def _processor(user=None, form=object()):
     proc = FormEventProcess(MagicMock())
     proc.user_service = MagicMock(get=AsyncMock(return_value=user or SimpleNamespace(id=7, role_id=2)))
     proc.bf_service = MagicMock(get=AsyncMock(return_value=form))
-    proc.sheet_service = MagicMock(update_cell=AsyncMock(return_value=[]))
+    proc.sheet_service = MagicMock(update_cell_with_formula=AsyncMock(return_value=([], None)))
     return proc
 
 
@@ -30,7 +30,7 @@ async def test_cache_reads_user_and_form_once_per_connection():
     await _send(proc, cache, 5)
     assert proc.user_service.get.await_count == 1
     assert proc.bf_service.get.await_count == 1
-    assert proc.sheet_service.update_cell.await_count == 5
+    assert proc.sheet_service.update_cell_with_formula.await_count == 5
 
 
 @pytest.mark.asyncio
@@ -61,7 +61,7 @@ async def test_missing_form_is_not_cached():
         res = await proc.process(event_data=EVENT, form_id=5, user_id=7, sheet="S", cache=cache)
         assert res is None
     assert proc.bf_service.get.await_count == 3
-    proc.sheet_service.update_cell.assert_not_awaited()
+    proc.sheet_service.update_cell_with_formula.assert_not_awaited()
 
 
 @pytest.mark.asyncio