287 lines
11 KiB
Python
287 lines
11 KiB
Python
"""Tests for the MAF roundtable persistence + service layer (?????).
|
|
|
|
Covers the Phase 1 exit criteria:
|
|
- Session CRUD with participants + capability snapshot.
|
|
- Run lifecycle (create / get / list / cancel).
|
|
- Monotonic per-run event seq under concurrent appends (MAF-01-06).
|
|
- Event ???? (list_events after_seq=N).
|
|
- Action idempotency-key uniqueness.
|
|
- Approval decision transition + "already decided" guard.
|
|
- User isolation (MAF-01-07) ??non-owner gets None / 404, not 403.
|
|
- Admin diagnostic access bypass.
|
|
- Feature flag (MAF-01-04) ??writes raise MafDisabledError when off; reads open.
|
|
- Run gating ??running session rejects config edits.
|
|
- capability_snapshot frozen at creation.
|
|
- MAF tables have no FK to legacy roundtable_* tables.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
from maf_test_helpers import make_maf_service
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
|
|
|
from app.maf_roundtable.service import (
|
|
MafConflictError,
|
|
MafNotFoundError,
|
|
)
|
|
from deerflow.persistence.base import Base
|
|
from deerflow.persistence.maf_roundtable import make_maf_roundtable_store
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def store(tmp_path):
|
|
engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'maf.db'}")
|
|
async with engine.begin() as conn:
|
|
import deerflow.persistence.maf_roundtable.model # noqa: F401
|
|
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
sf = async_sessionmaker(engine, expire_on_commit=False)
|
|
yield make_maf_roundtable_store(sf)
|
|
await engine.dispose()
|
|
|
|
|
|
def _svc(store, *, enabled=True, admin_diagnostic_access=True):
|
|
return make_maf_service(store, enabled=enabled, admin_diagnostic_access=admin_diagnostic_access)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_and_get_session_round_trips(store):
|
|
svc = _svc(store)
|
|
created = await svc.create_session(
|
|
user_id="u1",
|
|
body={
|
|
"title": "????",
|
|
"task_text": "????????",
|
|
"mode": "structured",
|
|
"model_name": "glm-5",
|
|
"max_rounds": 5,
|
|
"write_policy": "confirm_high_risk",
|
|
"participants": [
|
|
{"agent_id": "mgr", "display_name": "host", "role": "manager", "sort_order": 0},
|
|
{"agent_id": "seat-a", "display_name": "?A", "role": "seat", "sort_order": 1},
|
|
],
|
|
},
|
|
)
|
|
assert created["user_id"] == "u1"
|
|
assert created["status"] == "draft"
|
|
assert created["mode"] == "structured"
|
|
snap = created["capability_snapshot"]
|
|
assert snap is not None
|
|
assert snap.get("version") == 1
|
|
assert isinstance(snap.get("agents"), list)
|
|
assert len(snap["agents"]) == 2
|
|
assert len(created["participants"]) == 2
|
|
assert created["participants"][0]["agent_id"] == "mgr"
|
|
|
|
fetched = await svc.get_session(session_id=created["id"], user_id="u1")
|
|
assert fetched["id"] == created["id"]
|
|
assert len(fetched["participants"]) == 2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_sessions_owner_scoped(store):
|
|
svc = _svc(store)
|
|
await svc.create_session(user_id="u1", body={"task_text": "t1"})
|
|
await svc.create_session(user_id="u2", body={"task_text": "t2"})
|
|
rows_u1, total_u1 = await svc.list_sessions(user_id="u1")
|
|
assert total_u1 == 1
|
|
assert rows_u1[0]["user_id"] == "u1"
|
|
rows_u2, total_u2 = await svc.list_sessions(user_id="u2")
|
|
assert total_u2 == 1
|
|
assert rows_u2[0]["user_id"] == "u2"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_session_partial_and_participants_replace(store):
|
|
svc = _svc(store)
|
|
s = await svc.create_session(
|
|
user_id="u1",
|
|
body={"task_text": "orig", "participants": [{"agent_id": "a1", "role": "seat"}]},
|
|
)
|
|
updated = await svc.update_session(
|
|
session_id=s["id"],
|
|
user_id="u1",
|
|
fields={"title": "new-title", "participants": [{"agent_id": "a2", "role": "seat"}]},
|
|
)
|
|
assert updated["title"] == "new-title"
|
|
assert updated["task_text"] == "orig"
|
|
assert {p["agent_id"] for p in updated["participants"]} == {"a2"}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_session_freezes_capability_snapshot(store):
|
|
svc = _svc(store)
|
|
s = await svc.create_session(user_id="u1", body={"task_text": "t"})
|
|
original = s["capability_snapshot"]
|
|
updated = await svc.update_session(
|
|
session_id=s["id"], user_id="u1", fields={"capability_snapshot": {"v": 999}}
|
|
)
|
|
assert updated["capability_snapshot"] == original
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_session_owner_scoped(store):
|
|
svc = _svc(store)
|
|
s = await svc.create_session(user_id="u1", body={"task_text": "t"})
|
|
with pytest.raises(MafNotFoundError):
|
|
await svc.delete_session(session_id=s["id"], user_id="u2")
|
|
await svc.delete_session(session_id=s["id"], user_id="u1")
|
|
with pytest.raises(MafNotFoundError):
|
|
await svc.get_session(session_id=s["id"], user_id="u1")
|
|
|
|
|
|
# ?? run lifecycle ??????????????????????????????????????????????????????????
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_run_sets_session_running(store):
|
|
svc = _svc(store)
|
|
s = await svc.create_session(user_id="u1", body={"task_text": "t"})
|
|
run = await svc.create_run(session_id=s["id"], user_id="u1")
|
|
# No real MAF executor exists yet (Phase 2+): the run is parked in
|
|
# ``pending`` (a clearly reserved, not-yet-executing state) rather than
|
|
# pretending to be ``running``; the session is occupied (``running``).
|
|
assert run["status"] == "pending"
|
|
refreshed = await svc.get_session(session_id=s["id"], user_id="u1")
|
|
assert refreshed["status"] == "running"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_run_rejects_second_active_run(store):
|
|
svc = _svc(store)
|
|
s = await svc.create_session(user_id="u1", body={"task_text": "t"})
|
|
await svc.create_run(session_id=s["id"], user_id="u1")
|
|
with pytest.raises(MafConflictError):
|
|
await svc.create_run(session_id=s["id"], user_id="u1")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_session_rejected_while_running(store):
|
|
svc = _svc(store)
|
|
s = await svc.create_session(user_id="u1", body={"task_text": "t"})
|
|
await svc.create_run(session_id=s["id"], user_id="u1")
|
|
with pytest.raises(MafConflictError):
|
|
await svc.update_session(session_id=s["id"], user_id="u1", fields={"title": "x"})
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cancel_run_marks_session_cancelled(store):
|
|
svc = _svc(store)
|
|
s = await svc.create_session(user_id="u1", body={"task_text": "t"})
|
|
run = await svc.create_run(session_id=s["id"], user_id="u1")
|
|
cancelled = await svc.cancel_run(run_id=run["id"], user_id="u1")
|
|
assert cancelled["status"] == "cancelled"
|
|
refreshed = await svc.get_session(session_id=s["id"], user_id="u1")
|
|
assert refreshed["status"] == "cancelled"
|
|
|
|
|
|
# ?? events: monotonic seq + concurrent appends (MAF-01-06) ????????????????
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_append_event_monotonic_seq(store):
|
|
svc = _svc(store)
|
|
s = await svc.create_session(user_id="u1", body={"task_text": "t"})
|
|
run = await svc.create_run(session_id=s["id"], user_id="u1")
|
|
e1 = await svc.append_event(run_id=run["id"], user_id="u1", event_type="run.started")
|
|
e2 = await svc.append_event(run_id=run["id"], user_id="u1", event_type="agent.message.completed")
|
|
e3 = await svc.append_event(run_id=run["id"], user_id="u1", event_type="run.completed")
|
|
assert [e1["seq"], e2["seq"], e3["seq"]] == [1, 2, 3]
|
|
refreshed_run = await svc.get_run(run_id=run["id"], user_id="u1")
|
|
assert refreshed_run["last_event_seq"] == 3
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_concurrent_appends_produce_dense_seq_set(store):
|
|
"""Many parallel appends must produce a dense 1..N seq set ??no gaps, no
|
|
collisions ??exercising the MAX+1 retry on the (run_id, seq) unique
|
|
constraint (MAF-01-06)."""
|
|
svc = _svc(store)
|
|
s = await svc.create_session(user_id="u1", body={"task_text": "t"})
|
|
run = await svc.create_run(session_id=s["id"], user_id="u1")
|
|
|
|
n = 40
|
|
results = await asyncio.gather(
|
|
*[
|
|
svc.append_event(
|
|
run_id=run["id"],
|
|
user_id="u1",
|
|
event_type="agent.message.delta",
|
|
payload={"i": i},
|
|
)
|
|
for i in range(n)
|
|
]
|
|
)
|
|
seqs = sorted(r["seq"] for r in results)
|
|
assert seqs == list(range(1, n + 1))
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_events_after_seq_pull(store):
|
|
"""????: events with seq > after_seq in ascending order."""
|
|
svc = _svc(store)
|
|
s = await svc.create_session(user_id="u1", body={"task_text": "t"})
|
|
run = await svc.create_run(session_id=s["id"], user_id="u1")
|
|
for i in range(5):
|
|
await svc.append_event(run_id=run["id"], user_id="u1", event_type="tool.progress", payload={"i": i})
|
|
events, latest = await svc.list_events(run_id=run["id"], user_id="u1", after_seq=2)
|
|
assert [e["seq"] for e in events] == [3, 4, 5]
|
|
assert latest == 5
|
|
|
|
|
|
# ?? actions: idempotency key ??????????????????????????????????????????????
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_action_idempotency_key_unique(store):
|
|
svc = _svc(store)
|
|
s = await svc.create_session(user_id="u1", body={"task_text": "t"})
|
|
run = await svc.create_run(session_id=s["id"], user_id="u1")
|
|
from sqlalchemy.exc import IntegrityError
|
|
|
|
await store.actions.create_action(
|
|
{
|
|
"session_id": s["id"],
|
|
"run_id": run["id"],
|
|
"tool_name": "update_task",
|
|
"idempotency_key": "sess1+run1+seat-a+update_task+act1",
|
|
"risk_level": "R2",
|
|
"status": "pending",
|
|
}
|
|
)
|
|
with pytest.raises(IntegrityError):
|
|
await store.actions.create_action(
|
|
{
|
|
"session_id": s["id"],
|
|
"run_id": run["id"],
|
|
"tool_name": "update_task",
|
|
"idempotency_key": "sess1+run1+seat-a+update_task+act1",
|
|
"status": "pending",
|
|
}
|
|
)
|
|
found = await store.actions.get_by_idempotency_key("sess1+run1+seat-a+update_task+act1")
|
|
assert found is not None
|
|
assert found["tool_name"] == "update_task"
|
|
|
|
|
|
# ?? approvals: decision transition ????????????????????????????????????????
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_approval_decision_and_already_decided_guard(store):
|
|
svc = _svc(store)
|
|
s = await svc.create_session(user_id="u1", body={"task_text": "t"})
|
|
run = await svc.create_run(session_id=s["id"], user_id="u1")
|
|
appr = await store.approvals.create_approval(
|
|
{"session_id": s["id"], "run_id": run["id"], "summary": "approve write", "status": "pending"}
|
|
)
|
|
decided = await svc.decide_approval(approval_id=appr["id"], decision="approved", user_id="u1")
|
|
assert decided["status"] == "approved"
|
|
assert decided["decided_by"] == "u1"
|
|
with pytest.raises(MafConflictError):
|
|
await svc.decide_approval(approval_id=appr["id"], decision="rejected", user_id="u1")
|