api/tests/test_models_part_07_09.py

@@ -226,16 +226,34 @@ async def _check_active_lock_is_unique(session: AsyncSession) -> None:
 
     first = lock()
     session.add(first)
-    await session.flush()
+    await session.commit()  # первая активная блокировка зафиксирована: rollback ниже её не заденет
     first_id = first.id
+
     session.add(lock())
-    with pytest.raises(IntegrityError):
+    with pytest.raises(IntegrityError) as exc_info:
         await session.flush()  # второй активный lock на ту же версию
+    assert "uq_consolidation_lock_version_active" in str(exc_info.value)
     await session.rollback()
 
-    await session.execute(
+    released = await session.execute(
         text("UPDATE dkb.consolidation_lock SET released_at = :now WHERE id = :id"), {"now": now, "id": first_id}
     )
-    session.add(lock())  # после снятия первой блокировки — можно
-    await session.flush()
+    assert released.rowcount == 1  # type: ignore[attr-defined]  # первая блокировка существовала и снята
+    session.add(lock())  # после снятия первой блокировки слот свободен — именно из-за `WHERE released_at IS NULL`
+    await session.commit()
+
+    # снятая и активная блокировки одной версии сосуществуют; активной по-прежнему ровно одна
+    session.add(lock())
+    with pytest.raises(IntegrityError):
+        await session.flush()
     await session.rollback()
+    counts = (
+        await session.execute(
+            text(
+                "SELECT count(*) FILTER (WHERE released_at IS NULL), count(*) FILTER (WHERE released_at IS NOT NULL) "
+                "FROM dkb.consolidation_lock WHERE consolidation_version_id = :v"
+            ),
+            {"v": version_id},
+        )
+    ).one()
+    assert tuple(counts) == (1, 1)