api/alembic/versions/sql/0059_form_phase_direction_fix.sql

@@ -3,7 +3,7 @@
 -- 2. direction в идентификаторе этапа в аудите (add/upd/del), direction в payload upd.
 -- 3. del_form_phase удаляет только этап с указанным phase_code.
 
-DO $$
+DO $do$
 DECLARE
     v_deleted INT;
 BEGIN
@@ -20,7 +20,7 @@ BEGIN
     GET DIAGNOSTICS v_deleted = ROW_COUNT;
     RAISE NOTICE 'form_phase: удалено дублей этапов без направления: %', v_deleted;
 END
-$$;
+$do$;
 
 CREATE UNIQUE INDEX IF NOT EXISTS form_phase_nodir_uidx
     ON v3.form_phase (budget_form_id, sheet, phase_code)

api/tests/integration/test_form_phases_api_smoke.py

@@ -493,8 +493,13 @@ def test_form_phases_delete_removes_only_named_phase(client, admin_tokens):
     response = client.delete("/api/v1/stages/form/1/AHR/T_same?direction=Support", headers=headers)
     assert response.status_code == 200
 
-    result = client.get("/api/v1/stages/form/1?sheet=AHR", headers=headers).json()["result"]
-    left = {(p["phase_code"], p["direction"]) for p in result if p["phase_code"] in ("T_same", "T_other")}
+    # ?sheet= без direction фильтрует direction IS NULL — берём весь список и отбираем AHR здесь.
+    result = client.get("/api/v1/stages/form/1", headers=headers).json()["result"]
+    left = {
+        (p["phase_code"], p["direction"])
+        for p in result
+        if p["sheet"] == "AHR" and p["phase_code"] in ("T_same", "T_other")
+    }
     assert left == {("T_other", "Support"), ("T_same", "Development")}
 
 
@@ -517,9 +522,23 @@ def test_form_phases_duplicate_without_direction_rejected(client, admin_tokens):
     second = client.post("/api/v1/stages/form/1", json=body, headers=headers)
     assert 400 <= second.status_code < 500
     assert "phase_code" in second.text
+    # После ошибки клиентская сессия откатывается целиком (tests/integration/conftest.py),
+    # поэтому число строк здесь не проверяем — защиту БД проверяет следующий тест.
 
-    result = client.get("/api/v1/stages/form/1?sheet=OPER&phase_code=T_dup", headers=headers).json()["result"]
-    assert len(result) == 1
+
+def test_form_phase_nodir_unique_index_blocks_duplicate(client, svod2_sql):
+    rows = svod2_sql(
+        "SELECT budget_form_id, sheet, phase_code FROM v3.form_phase WHERE direction IS NULL LIMIT 1"
+    )
+    if not rows:
+        pytest.skip("Нет этапа без направления в текущей БД")
+    form_id, sheet, phase_code = rows[0]
+    with pytest.raises(Exception, match="form_phase_nodir_uidx"):
+        svod2_sql(
+            "INSERT INTO v3.form_phase SELECT * FROM v3.form_phase "
+            "WHERE budget_form_id = :f AND sheet = :s AND phase_code = :p AND direction IS NULL",
+            {"f": form_id, "s": sheet, "p": phase_code},
+        )
 
 
 def test_form_phases_patch_audit_has_direction(client, admin_tokens, svod2_sql):