391 lines
15 KiB
Python
391 lines
15 KiB
Python
"""Tests for the MAF roundtable V2 Conversation API (Phase 1 / R1).
|
|
|
|
Covers:
|
|
- Create, list, detail, message pagination, user isolation.
|
|
- Public API responses never leak internal_session_id, active_run_id, run_id,
|
|
owner_user_id, thread ids, or secret-class fields.
|
|
- First message creates a real internal session; run FK is valid.
|
|
- Transaction rollback: any flush failure rolls back message + run + event + status.
|
|
- Concurrent send: only one succeeds, the other gets 409; no duplicate runs.
|
|
- Mixed tool-events and messages: last_stream_seq is contiguous and usable as SSE cursor.
|
|
- Empty / overlong / forged-internal-field content → 422.
|
|
- Non-owner cannot read/write; admin cannot write another user's conversation.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
from sqlalchemy import inspect, select
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
|
|
|
from app.gateway.routers.maf_conversations import _user_ctx
|
|
from app.maf_roundtable.conversation_schemas import (
|
|
CreateConversationRequest,
|
|
SendMessageRequest,
|
|
)
|
|
from app.maf_roundtable.conversation_service import (
|
|
ConversationConflictError,
|
|
ConversationDisabledError,
|
|
ConversationNotFoundError,
|
|
MafConversationService,
|
|
)
|
|
from deerflow.persistence.base import Base
|
|
from deerflow.persistence.maf_roundtable import make_maf_roundtable_store
|
|
from deerflow.persistence.maf_roundtable.model import (
|
|
MafRoundtableConversationRow,
|
|
MafRoundtableMessageRow,
|
|
MafRoundtableRunRow,
|
|
MafRoundtableSessionRow,
|
|
)
|
|
|
|
|
|
class FakeMafConfig:
|
|
enabled = True
|
|
default_mode = "structured"
|
|
default_model_name = "test-model"
|
|
default_reasoning_effort = None
|
|
default_manager_agent_id = "mgr-agent"
|
|
default_max_rounds = 8
|
|
default_write_policy = "confirm_high_risk"
|
|
admin_diagnostic_access = True
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def store(tmp_path):
|
|
engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'conv.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()
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def svc(store):
|
|
return MafConversationService(store, maf_config=FakeMafConfig())
|
|
|
|
|
|
FORBIDDEN_FIELDS = {
|
|
"owner_user_id", "internal_session_id", "active_run_id",
|
|
"thread_id", "capability_snapshot", "idempotency_key",
|
|
"target_key", "internal_canary",
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_and_get_conversation(svc):
|
|
conv = await svc.create_conversation(user_id="u1", title="Test", task_context={"business_code": "3Q"})
|
|
assert conv["id"]
|
|
assert conv["title"] == "Test"
|
|
assert conv["safe_status"] == "idle"
|
|
assert conv["last_stream_seq"] == 0
|
|
assert conv["task_context"] == {"business_code": "3Q"}
|
|
for field in FORBIDDEN_FIELDS:
|
|
assert field not in conv, f"Leaked field: {field}"
|
|
|
|
fetched = await svc.get_conversation(conversation_id=conv["id"], user_id="u1")
|
|
assert fetched["id"] == conv["id"]
|
|
for field in FORBIDDEN_FIELDS:
|
|
assert field not in fetched
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_conversations_owner_scoped(svc):
|
|
await svc.create_conversation(user_id="u1", title="A")
|
|
await svc.create_conversation(user_id="u1", title="B")
|
|
await svc.create_conversation(user_id="u2", title="C")
|
|
|
|
list_u1, total_u1 = await svc.list_conversations(user_id="u1")
|
|
assert total_u1 == 2
|
|
assert all(c["id"] for c in list_u1)
|
|
|
|
list_u2, total_u2 = await svc.list_conversations(user_id="u2")
|
|
assert total_u2 == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_conversation_wrong_owner_returns_404(svc):
|
|
conv = await svc.create_conversation(user_id="u1", title="Private")
|
|
with pytest.raises(ConversationNotFoundError):
|
|
await svc.get_conversation(conversation_id=conv["id"], user_id="u2")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_admin_can_read_but_not_write(svc):
|
|
conv = await svc.create_conversation(user_id="u1", title="AdminRead")
|
|
fetched = await svc.get_conversation(conversation_id=conv["id"], user_id="admin", is_admin=True)
|
|
assert fetched["id"] == conv["id"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_message_creates_real_session_and_run(svc, store):
|
|
conv = await svc.create_conversation(user_id="u1", title="SendTest")
|
|
result = await svc.send_message(conversation_id=conv["id"], user_id="u1", content="Hello world")
|
|
|
|
assert result["message_id"]
|
|
assert result["conversation_id"] == conv["id"]
|
|
assert result["seq"] == 1
|
|
|
|
async with store.session_factory() as session:
|
|
conv_row = (await session.execute(
|
|
select(MafRoundtableConversationRow).where(MafRoundtableConversationRow.id == conv["id"])
|
|
)).scalars().first()
|
|
assert conv_row is not None
|
|
assert conv_row.internal_session_id is not None
|
|
assert conv_row.safe_status == "running"
|
|
assert conv_row.active_run_id is not None
|
|
# send_message writes only user.message.created; the manager.dispatch
|
|
# projection comes later from the real workflow via bridge_workflow_event.
|
|
assert conv_row.last_stream_seq == 1
|
|
|
|
session_row = (await session.execute(
|
|
select(MafRoundtableSessionRow).where(MafRoundtableSessionRow.id == conv_row.internal_session_id)
|
|
)).scalars().first()
|
|
assert session_row is not None
|
|
assert session_row.user_id == "u1"
|
|
|
|
run_row = (await session.execute(
|
|
select(MafRoundtableRunRow).where(MafRoundtableRunRow.id == conv_row.active_run_id)
|
|
)).scalars().first()
|
|
assert run_row is not None
|
|
assert run_row.session_id == conv_row.internal_session_id
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_message_wrong_owner_404(svc):
|
|
conv = await svc.create_conversation(user_id="u1", title="X")
|
|
with pytest.raises(ConversationNotFoundError):
|
|
await svc.send_message(conversation_id=conv["id"], user_id="u2", content="hack")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_concurrent_send_only_one_succeeds(svc, store):
|
|
conv = await svc.create_conversation(user_id="u1", title="Race")
|
|
|
|
async def try_send(content: str):
|
|
try:
|
|
r = await svc.send_message(conversation_id=conv["id"], user_id="u1", content=content)
|
|
return ("ok", r)
|
|
except ConversationConflictError:
|
|
return ("conflict", None)
|
|
|
|
results = await asyncio.gather(try_send("msg1"), try_send("msg2"))
|
|
statuses = [r[0] for r in results]
|
|
assert statuses.count("ok") == 1
|
|
assert statuses.count("conflict") == 1
|
|
|
|
async with store.session_factory() as session:
|
|
runs = (await session.execute(
|
|
select(MafRoundtableRunRow).join(
|
|
MafRoundtableSessionRow,
|
|
MafRoundtableRunRow.session_id == MafRoundtableSessionRow.id,
|
|
).where(MafRoundtableSessionRow.user_id == "u1")
|
|
)).scalars().all()
|
|
assert len(runs) == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_message_public_dto_no_run_id(svc, store):
|
|
conv = await svc.create_conversation(user_id="u1", title="PubDTO")
|
|
await svc.send_message(conversation_id=conv["id"], user_id="u1", content="test content")
|
|
|
|
messages, last_seq = await svc.list_messages(conversation_id=conv["id"], user_id="u1")
|
|
assert len(messages) == 1
|
|
msg = messages[0]
|
|
assert "run_id" not in msg
|
|
assert msg["content"] == "test content"
|
|
assert msg["message_kind"] == "user"
|
|
assert last_seq == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_last_stream_seq_contiguous(svc, store):
|
|
conv = await svc.create_conversation(user_id="u1", title="SeqTest")
|
|
await svc.send_message(conversation_id=conv["id"], user_id="u1", content="first")
|
|
|
|
async with store.session_factory() as session:
|
|
conv_row = (await session.execute(
|
|
select(MafRoundtableConversationRow).where(MafRoundtableConversationRow.id == conv["id"])
|
|
)).scalars().first()
|
|
conv_row.safe_status = "idle"
|
|
conv_row.active_run_id = None
|
|
await session.commit()
|
|
|
|
await svc.send_message(conversation_id=conv["id"], user_id="u1", content="second")
|
|
|
|
async with store.session_factory() as session:
|
|
conv_row = (await session.execute(
|
|
select(MafRoundtableConversationRow).where(MafRoundtableConversationRow.id == conv["id"])
|
|
)).scalars().first()
|
|
assert conv_row.last_stream_seq == 2
|
|
|
|
msgs = (await session.execute(
|
|
select(MafRoundtableMessageRow).where(
|
|
MafRoundtableMessageRow.conversation_id == conv["id"]
|
|
).order_by(MafRoundtableMessageRow.seq)
|
|
)).scalars().all()
|
|
seqs = [m.seq for m in msgs]
|
|
assert seqs == [1, 2]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_schema_rejects_forbidden_fields():
|
|
with pytest.raises(Exception):
|
|
CreateConversationRequest.model_validate({"title": "x", "owner_user_id": "hacker"})
|
|
|
|
with pytest.raises(Exception):
|
|
CreateConversationRequest.model_validate({"title": "x", "internal_session_id": "s1"})
|
|
|
|
with pytest.raises(Exception):
|
|
SendMessageRequest.model_validate({"content": "hi", "run_id": "r1"})
|
|
|
|
with pytest.raises(Exception):
|
|
SendMessageRequest.model_validate({"content": "hi", "thread_id": "t1"})
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_schema_rejects_empty_and_overlong_content():
|
|
with pytest.raises(Exception):
|
|
SendMessageRequest.model_validate({"content": ""})
|
|
|
|
with pytest.raises(Exception):
|
|
SendMessageRequest.model_validate({"content": " "})
|
|
|
|
with pytest.raises(Exception):
|
|
SendMessageRequest.model_validate({"content": "x" * 20001})
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_task_context_whitelist(svc):
|
|
conv = await svc.create_conversation(
|
|
user_id="u1", title="Ctx",
|
|
task_context={"business_code": "6BF", "secret_key": "LEAKED", "task_name": "T1"},
|
|
)
|
|
assert conv["task_context"] == {"business_code": "6BF", "task_name": "T1"}
|
|
assert "secret_key" not in (conv["task_context"] or {})
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_task_context_rejects_unknown_or_nested_values():
|
|
with pytest.raises(Exception):
|
|
CreateConversationRequest.model_validate({"task_context": {"secret_key": "nope"}})
|
|
with pytest.raises(Exception):
|
|
CreateConversationRequest.model_validate({"task_context": {"task_id": {"secret": "nope"}}})
|
|
with pytest.raises(Exception):
|
|
CreateConversationRequest.model_validate({"title": "x", "session_id": "forged"})
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_disabled_feature_rejects_conversation_writes(store):
|
|
disabled = MafConversationService(store, maf_config=SimpleNamespace(enabled=False))
|
|
with pytest.raises(ConversationDisabledError):
|
|
await disabled.create_conversation(user_id="u1", title="disabled")
|
|
|
|
|
|
def test_router_uses_authenticated_user_object_for_owner_scope():
|
|
user = SimpleNamespace(id="real-user", system_role="admin")
|
|
request = SimpleNamespace(state=SimpleNamespace(user=user))
|
|
assert _user_ctx(request) == ("real-user", True)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_admin_cross_user_read_respects_diagnostic_config(store):
|
|
owner_service = MafConversationService(store, maf_config=FakeMafConfig())
|
|
conv = await owner_service.create_conversation(user_id="u1", title="Private")
|
|
no_diagnostics = MafConversationService(
|
|
store,
|
|
maf_config=SimpleNamespace(enabled=True, admin_diagnostic_access=False),
|
|
)
|
|
with pytest.raises(ConversationNotFoundError):
|
|
await no_diagnostics.get_conversation(
|
|
conversation_id=conv["id"], user_id="admin", is_admin=True
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_conversation_internal_links_have_real_foreign_keys(store):
|
|
async with store.session_factory() as session:
|
|
def _fks(sync_session):
|
|
inspector = inspect(sync_session.connection())
|
|
return {
|
|
(tuple(fk["constrained_columns"]), fk["referred_table"])
|
|
for fk in inspector.get_foreign_keys("maf_roundtable_conversations")
|
|
}
|
|
|
|
foreign_keys = await session.run_sync(_fks)
|
|
assert (("active_run_id",), "maf_roundtable_runs") in foreign_keys
|
|
assert (("internal_session_id",), "maf_roundtable_sessions") in foreign_keys
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_event_only_item_advances_the_same_sse_cursor(svc, store):
|
|
conv = await svc.create_conversation(user_id="u1", title="Unified cursor")
|
|
await svc.send_message(conversation_id=conv["id"], user_id="u1", content="first")
|
|
|
|
async with store.session_factory() as session:
|
|
async with session.begin():
|
|
row = (
|
|
await session.execute(
|
|
select(MafRoundtableConversationRow)
|
|
.where(MafRoundtableConversationRow.id == conv["id"])
|
|
.with_for_update()
|
|
)
|
|
).scalars().one()
|
|
item = await store.append_conversation_item(
|
|
session,
|
|
conversation=row,
|
|
event_type="manager.intent",
|
|
event_data={"status": "understood"},
|
|
)
|
|
assert item["seq"] == 2
|
|
|
|
messages, last_stream_seq = await svc.list_messages(conversation_id=conv["id"], user_id="u1")
|
|
assert [message["seq"] for message in messages] == [1]
|
|
assert last_stream_seq == 2
|
|
events = await store.conversation_events.list_events(conv["id"])
|
|
assert [event["seq"] for event in events] == [1, 2]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_standalone_repositories_cannot_create_a_second_cursor(svc, store):
|
|
conv = await svc.create_conversation(user_id="u1", title="No split cursor")
|
|
with pytest.raises(RuntimeError, match="append_conversation_item"):
|
|
await store.messages.append_message(
|
|
conversation_id=conv["id"], message_kind="manager", role="assistant", content="unsafe"
|
|
)
|
|
with pytest.raises(RuntimeError, match="append_conversation_item"):
|
|
await store.conversation_events.append_event(
|
|
conversation_id=conv["id"], event_type="manager.intent"
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_atomic_send_rolls_back_on_item_write_failure(svc, store, monkeypatch):
|
|
conv = await svc.create_conversation(user_id="u1", title="Rollback")
|
|
|
|
async def _fail_append(*args, **kwargs):
|
|
raise RuntimeError("simulated item write failure")
|
|
|
|
monkeypatch.setattr(store, "append_conversation_item", _fail_append)
|
|
with pytest.raises(RuntimeError, match="simulated item write failure"):
|
|
await svc.send_message(conversation_id=conv["id"], user_id="u1", content="must roll back")
|
|
|
|
async with store.session_factory() as session:
|
|
refreshed = (
|
|
await session.execute(
|
|
select(MafRoundtableConversationRow).where(MafRoundtableConversationRow.id == conv["id"])
|
|
)
|
|
).scalars().one()
|
|
assert refreshed.safe_status == "idle"
|
|
assert refreshed.internal_session_id is None
|
|
assert refreshed.active_run_id is None
|
|
assert refreshed.last_stream_seq == 0
|
|
assert (await session.execute(select(MafRoundtableMessageRow))).scalars().all() == []
|
|
assert (await session.execute(select(MafRoundtableRunRow))).scalars().all() == []
|