750 lines
22 KiB
Python
750 lines
22 KiB
Python
"""Behavior tests for deterministic same-gateway Discussion policy."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from gateway import hosted_room_discussion as discussion
|
|
from gateway import hosted_room_driver as driver
|
|
from gateway import hosted_rooms
|
|
|
|
|
|
ROOM_ID = "room-1"
|
|
GATEWAY_ID = "gateway-a"
|
|
LOCAL_PROFILES = ("research", "build", "review", "ops", "qa", "docs")
|
|
MEMBERS = [
|
|
{
|
|
"member_id": f"member-{profile}",
|
|
"profile": profile,
|
|
"handle": profile,
|
|
"display_name": profile.title(),
|
|
}
|
|
for profile in LOCAL_PROFILES[:3]
|
|
]
|
|
|
|
|
|
@pytest.fixture
|
|
def room_db(tmp_path: Path) -> tuple[Path, dict]:
|
|
db = tmp_path / "state.db"
|
|
room = hosted_rooms.create_room(
|
|
db,
|
|
room_id=ROOM_ID,
|
|
name="Release",
|
|
members=MEMBERS,
|
|
authority_gateway_id=GATEWAY_ID,
|
|
now=1,
|
|
)
|
|
return db, room
|
|
|
|
|
|
def _events(db: Path) -> list[dict]:
|
|
return hosted_rooms.read_events(
|
|
db,
|
|
room_id=ROOM_ID,
|
|
since_seq=0,
|
|
limit=hosted_rooms.MAX_LOG_LIMIT,
|
|
)["events"]
|
|
|
|
|
|
def _append_user(
|
|
db: Path,
|
|
*,
|
|
event_id: str,
|
|
text: str,
|
|
thread_id: str = "thread-1",
|
|
) -> dict:
|
|
return hosted_rooms.append_event(
|
|
db,
|
|
room_id=ROOM_ID,
|
|
event_id=event_id,
|
|
kind="message.user",
|
|
actor={"kind": "user", "id": "local-user"},
|
|
authority_gateway_id=GATEWAY_ID,
|
|
authority_epoch=1,
|
|
payload={"text": text, "thread_id": thread_id},
|
|
now=time.time(),
|
|
)
|
|
|
|
|
|
def _append_publication(
|
|
db: Path,
|
|
plan: discussion.PublicationPlan,
|
|
) -> list[dict]:
|
|
return [
|
|
hosted_rooms.append_event(
|
|
db,
|
|
**event.append_kwargs(ROOM_ID),
|
|
now=time.time(),
|
|
)
|
|
for event in plan.events
|
|
]
|
|
|
|
|
|
def _append_activity(
|
|
db: Path,
|
|
*,
|
|
event_id: str,
|
|
discussion_event_id: str,
|
|
thread_id: str,
|
|
) -> dict:
|
|
return hosted_rooms.append_event(
|
|
db,
|
|
room_id=ROOM_ID,
|
|
event_id=event_id,
|
|
kind="room.activity",
|
|
actor={"kind": "gateway", "id": GATEWAY_ID},
|
|
payload={
|
|
"status": "settled",
|
|
"reason_code": "silent_round",
|
|
"thread_id": thread_id,
|
|
"discussion_event_id": discussion_event_id,
|
|
},
|
|
authority_gateway_id=GATEWAY_ID,
|
|
authority_epoch=1,
|
|
)
|
|
|
|
|
|
def _next_task(room: dict, db: Path) -> discussion.DiscussionTaskPlan:
|
|
decision = discussion.plan_next_task(
|
|
room,
|
|
_events(db),
|
|
local_profiles=LOCAL_PROFILES,
|
|
)
|
|
assert decision.status == "task", decision
|
|
assert decision.task is not None
|
|
return decision.task
|
|
|
|
|
|
def _settle_next(
|
|
room: dict,
|
|
db: Path,
|
|
*,
|
|
text: str,
|
|
) -> discussion.DiscussionTaskPlan:
|
|
task = _next_task(room, db)
|
|
publication = discussion.plan_publication(
|
|
room,
|
|
_events(db),
|
|
task,
|
|
status="settled",
|
|
result={"text": text},
|
|
local_profiles=LOCAL_PROFILES,
|
|
)
|
|
_append_publication(db, publication)
|
|
return task
|
|
|
|
|
|
def test_deferred_member_allows_next_mentioned_member_and_later_terminal_result(
|
|
room_db,
|
|
):
|
|
db, room = room_db
|
|
_append_user(db, event_id="user-1", text="Report.")
|
|
first = _next_task(room, db)
|
|
deferred = discussion.plan_publication(
|
|
room,
|
|
_events(db),
|
|
first,
|
|
status="deferred",
|
|
result={"reason": "member_unavailable"},
|
|
execution_generation=1,
|
|
local_profiles=LOCAL_PROFILES,
|
|
)
|
|
_append_publication(db, deferred)
|
|
|
|
second = _next_task(room, db)
|
|
assert second.member.member_id != first.member.member_id
|
|
|
|
settled = discussion.plan_publication(
|
|
room,
|
|
_events(db),
|
|
first,
|
|
status="settled",
|
|
result={"text": "Recovered on explicit retry."},
|
|
local_profiles=LOCAL_PROFILES,
|
|
)
|
|
_append_publication(db, settled)
|
|
decision = discussion.plan_next_task(
|
|
room,
|
|
_events(db),
|
|
local_profiles=LOCAL_PROFILES,
|
|
)
|
|
assert decision.status == "task"
|
|
assert decision.task is not None
|
|
assert decision.task.member.member_id == second.member.member_id
|
|
|
|
|
|
def test_distinct_threads_are_planned_fifo_without_skipping(room_db):
|
|
db, room = room_db
|
|
_append_user(db, event_id="user-1", text="First", thread_id="thread-1")
|
|
_append_user(db, event_id="user-2", text="Second", thread_id="thread-2")
|
|
|
|
first = _next_task(room, db)
|
|
assert first.discussion_event_id == "user-1"
|
|
_append_activity(
|
|
db,
|
|
event_id="activity-1",
|
|
discussion_event_id="user-1",
|
|
thread_id="thread-1",
|
|
)
|
|
second = _next_task(room, db)
|
|
assert second.discussion_event_id == "user-2"
|
|
|
|
|
|
def test_room_stop_fences_old_work_but_allows_a_later_message(room_db):
|
|
db, room = room_db
|
|
_append_user(db, event_id="user-1", text="First", thread_id="thread-1")
|
|
stop = hosted_rooms.request_room_stop(
|
|
db,
|
|
room_id=ROOM_ID,
|
|
cancel_id="user-stop-1",
|
|
expected_gateway_id=str(room["authority_gateway_id"]),
|
|
expected_epoch=int(room["authority_epoch"]),
|
|
)
|
|
decision = discussion.plan_next_task(
|
|
room,
|
|
_events(db),
|
|
local_profiles=LOCAL_PROFILES,
|
|
)
|
|
assert decision.status == "idle"
|
|
assert stop["kind"] == "room.stop_requested"
|
|
|
|
_append_user(db, event_id="user-2", text="Continue", thread_id="thread-2")
|
|
resumed = _next_task(room, db)
|
|
assert resumed.discussion_event_id == "user-2"
|
|
|
|
|
|
def test_deterministic_task_fits_existing_driver_and_reconstructs_after_restart(
|
|
room_db: tuple[Path, dict],
|
|
):
|
|
db, room = room_db
|
|
user = _append_user(db, event_id="user-1", text="Check the release.")
|
|
|
|
first = _next_task(room, db)
|
|
repeated = _next_task(room, db)
|
|
assert first == repeated
|
|
assert first.identity.thread_id == "thread-1"
|
|
assert first.payload == {
|
|
"target_member_id": "member-research",
|
|
"target_profile": "research",
|
|
"prompt": first.payload["prompt"],
|
|
"source_event_seq": user["seq"],
|
|
}
|
|
assert set(first.payload) == {
|
|
"target_member_id",
|
|
"target_profile",
|
|
"prompt",
|
|
"source_event_seq",
|
|
}
|
|
|
|
admitted = driver.admit_task(
|
|
db,
|
|
first.identity,
|
|
payload=first.payload,
|
|
clock=time.time,
|
|
)
|
|
stored = driver.get_task(db, first.identity)
|
|
reconstructed = discussion.reconstruct_task_plan(
|
|
room,
|
|
_events(db),
|
|
stored,
|
|
local_profiles=LOCAL_PROFILES,
|
|
)
|
|
assert admitted["status"] == "queued"
|
|
assert reconstructed == first
|
|
|
|
reopened_events = _events(db)
|
|
assert (
|
|
discussion.reconstruct_task_plan(
|
|
room,
|
|
reopened_events,
|
|
driver.get_task(db, first.identity),
|
|
local_profiles=LOCAL_PROFILES,
|
|
)
|
|
== first
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("text", "expected_profile"),
|
|
[
|
|
("@build please inspect this", "build"),
|
|
("@all inspect this", "research"),
|
|
("@everyone inspect this", "research"),
|
|
("inspect this", "research"),
|
|
("@unknown inspect this", "research"),
|
|
],
|
|
)
|
|
def test_mentions_select_handles_or_everyone(
|
|
room_db: tuple[Path, dict],
|
|
text: str,
|
|
expected_profile: str,
|
|
):
|
|
db, room = room_db
|
|
_append_user(db, event_id="user-1", text=text)
|
|
|
|
assert _next_task(room, db).member.profile == expected_profile
|
|
|
|
|
|
def test_member_mention_joins_the_next_round_not_the_current_round(
|
|
room_db: tuple[Path, dict],
|
|
):
|
|
db, room = room_db
|
|
_append_user(db, event_id="user-1", text="@research lead this")
|
|
|
|
first = _settle_next(room, db, text="@build can add the implementation detail.")
|
|
second = _next_task(room, db)
|
|
|
|
assert first.member.profile == "research"
|
|
assert first.round_index == 0
|
|
assert second.member.profile == "build"
|
|
assert second.round_index == 1
|
|
assert "@research lead this" in second.payload["prompt"]
|
|
assert "@build can add the implementation detail." in second.payload["prompt"]
|
|
|
|
|
|
def test_plain_member_reply_does_not_wake_another_bot_round(
|
|
room_db: tuple[Path, dict],
|
|
):
|
|
db, room = room_db
|
|
_append_user(db, event_id="user-1", text="@research answer the user")
|
|
_settle_next(room, db, text="The answer is ready for the user.")
|
|
|
|
decision = discussion.plan_next_task(
|
|
room,
|
|
_events(db),
|
|
local_profiles=LOCAL_PROFILES,
|
|
)
|
|
|
|
assert decision.status == "settled"
|
|
assert decision.reason == "silent_round"
|
|
|
|
|
|
@pytest.mark.parametrize("value", ["", "pass", "pass.", "(pass)", " ( PASS ). "])
|
|
def test_pass_detection(value: str):
|
|
assert discussion.is_pass_text(value)
|
|
|
|
|
|
def test_real_text_is_not_a_pass():
|
|
assert not discussion.is_pass_text("I found the issue.")
|
|
|
|
|
|
def test_full_pass_round_settles_without_member_messages(
|
|
room_db: tuple[Path, dict],
|
|
):
|
|
db, room = room_db
|
|
_append_user(db, event_id="user-1", text="Any concerns?")
|
|
|
|
for _member in MEMBERS:
|
|
_settle_next(room, db, text="(pass)")
|
|
|
|
decision = discussion.plan_next_task(
|
|
room,
|
|
_events(db),
|
|
local_profiles=LOCAL_PROFILES,
|
|
)
|
|
assert decision.status == "settled"
|
|
assert decision.reason == "silent_round"
|
|
assert [event["kind"] for event in _events(db)].count("message.member") == 0
|
|
|
|
|
|
def test_failed_members_advance_the_round_as_silence(
|
|
room_db: tuple[Path, dict],
|
|
):
|
|
db, room = room_db
|
|
_append_user(db, event_id="user-1", text="Any concerns?")
|
|
|
|
for expected in ("research", "build", "review"):
|
|
task = _next_task(room, db)
|
|
assert task.member.profile == expected
|
|
publication = discussion.plan_publication(
|
|
room,
|
|
_events(db),
|
|
task,
|
|
status="failed",
|
|
result={"error": f"{expected} unavailable"},
|
|
local_profiles=LOCAL_PROFILES,
|
|
)
|
|
assert publication.terminal_kind == "turn.failed"
|
|
assert len(publication.events) == 1
|
|
assert publication.events[0].payload["reason_code"] == "unknown"
|
|
_append_publication(db, publication)
|
|
|
|
decision = discussion.plan_next_task(
|
|
room,
|
|
_events(db),
|
|
local_profiles=LOCAL_PROFILES,
|
|
)
|
|
assert decision.status == "settled"
|
|
assert decision.reason == "silent_round"
|
|
|
|
|
|
def test_failed_publication_preserves_a_typed_actionable_reason(
|
|
room_db: tuple[Path, dict],
|
|
):
|
|
db, room = room_db
|
|
_append_user(db, event_id="user-1", text="Please continue.")
|
|
task = _next_task(room, db)
|
|
publication = discussion.plan_publication(
|
|
room,
|
|
_events(db),
|
|
task,
|
|
status="failed",
|
|
result={"error": "HTTP 401 authentication failed"},
|
|
local_profiles=LOCAL_PROFILES,
|
|
)
|
|
assert publication.events[0].payload["reason_code"] == "provider_auth_or_access"
|
|
|
|
|
|
def test_failed_publication_rejects_an_untrusted_reason_code(
|
|
room_db: tuple[Path, dict],
|
|
):
|
|
db, room = room_db
|
|
_append_user(db, event_id="user-1", text="Please continue.")
|
|
task = _next_task(room, db)
|
|
publication = discussion.plan_publication(
|
|
room,
|
|
_events(db),
|
|
task,
|
|
status="failed",
|
|
result={"error": "failed", "reason_code": "invented"},
|
|
local_profiles=LOCAL_PROFILES,
|
|
)
|
|
assert publication.events[0].payload["reason_code"] == "unknown"
|
|
|
|
|
|
def test_publication_is_idempotent_and_changed_result_conflicts(
|
|
room_db: tuple[Path, dict],
|
|
):
|
|
db, room = room_db
|
|
_append_user(db, event_id="user-1", text="Report.")
|
|
task = _next_task(room, db)
|
|
publication = discussion.plan_publication(
|
|
room,
|
|
_events(db),
|
|
task,
|
|
status="settled",
|
|
result={"text": "Ready."},
|
|
local_profiles=LOCAL_PROFILES,
|
|
)
|
|
|
|
first = _append_publication(db, publication)
|
|
repeated = _append_publication(db, publication)
|
|
assert [event["seq"] for event in first] == [event["seq"] for event in repeated]
|
|
assert all(event["idempotent"] for event in repeated)
|
|
|
|
changed = discussion.plan_publication(
|
|
room,
|
|
_events(db),
|
|
task,
|
|
status="settled",
|
|
result={"text": "Different."},
|
|
local_profiles=LOCAL_PROFILES,
|
|
)
|
|
with pytest.raises(hosted_rooms.EventConflictError):
|
|
_append_publication(db, changed)
|
|
|
|
|
|
def test_partial_publication_replays_same_effects_before_policy_advances(
|
|
room_db: tuple[Path, dict],
|
|
):
|
|
db, room = room_db
|
|
_append_user(db, event_id="user-1", text="Report.")
|
|
task = _next_task(room, db)
|
|
publication = discussion.plan_publication(
|
|
room,
|
|
_events(db),
|
|
task,
|
|
status="settled",
|
|
result={"text": "Ready."},
|
|
local_profiles=LOCAL_PROFILES,
|
|
)
|
|
|
|
message_effect = publication.events[0]
|
|
hosted_rooms.append_event(
|
|
db,
|
|
**message_effect.append_kwargs(ROOM_ID),
|
|
now=time.time(),
|
|
)
|
|
assert _next_task(room, db).identity == task.identity
|
|
|
|
replayed = discussion.plan_publication(
|
|
room,
|
|
_events(db),
|
|
task,
|
|
status="settled",
|
|
result={"text": "Ready."},
|
|
local_profiles=LOCAL_PROFILES,
|
|
)
|
|
_append_publication(db, replayed)
|
|
assert _next_task(room, db).member.profile == "build"
|
|
|
|
|
|
def test_watermark_excludes_a_members_old_input_and_own_reply(
|
|
room_db: tuple[Path, dict],
|
|
):
|
|
db, room = room_db
|
|
_append_user(db, event_id="user-1", text="Old request.")
|
|
first = _settle_next(room, db, text="Old answer.")
|
|
watermark = discussion.derive_member_watermarks(
|
|
room,
|
|
_events(db),
|
|
local_profiles=LOCAL_PROFILES,
|
|
)[("thread-1", first.member.member_id)]
|
|
assert watermark == max(
|
|
event["seq"]
|
|
for event in _events(db)
|
|
if event["kind"] == "message.member"
|
|
and event["payload"]["task_id"] == first.identity.task_id
|
|
)
|
|
|
|
latest = _append_user(db, event_id="user-2", text="New request.")
|
|
next_task = _next_task(room, db)
|
|
assert next_task.member.profile == "research"
|
|
assert next_task.payload["source_event_seq"] == latest["seq"]
|
|
assert "New request." in next_task.payload["prompt"]
|
|
assert "Old request." not in next_task.payload["prompt"]
|
|
assert "Old answer." not in next_task.payload["prompt"]
|
|
|
|
|
|
def test_newer_same_thread_user_event_cancels_a_late_result(
|
|
room_db: tuple[Path, dict],
|
|
):
|
|
db, room = room_db
|
|
_append_user(db, event_id="user-1", text="First request.")
|
|
stale = _next_task(room, db)
|
|
latest = _append_user(db, event_id="user-2", text="Second request.")
|
|
|
|
publication = discussion.plan_publication(
|
|
room,
|
|
_events(db),
|
|
stale,
|
|
status="settled",
|
|
result={"text": "Late stale answer."},
|
|
local_profiles=LOCAL_PROFILES,
|
|
)
|
|
assert publication.terminal_kind == "turn.cancelled"
|
|
assert [event.kind for event in publication.events] == ["turn.cancelled"]
|
|
assert publication.events[0].payload["reason"] == "superseded_by_newer_user_event"
|
|
_append_publication(db, publication)
|
|
|
|
current = _next_task(room, db)
|
|
assert current.payload["source_event_seq"] == latest["seq"]
|
|
assert "Second request." in current.payload["prompt"]
|
|
|
|
|
|
def test_cross_thread_newer_user_does_not_discard_completed_old_reply(
|
|
room_db: tuple[Path, dict],
|
|
):
|
|
db, room = room_db
|
|
_append_user(db, event_id="user-1", text="First request.", thread_id="thread-1")
|
|
old = _next_task(room, db)
|
|
_append_user(db, event_id="user-2", text="Other topic.", thread_id="thread-2")
|
|
|
|
publication = discussion.plan_publication(
|
|
room,
|
|
_events(db),
|
|
old,
|
|
status="settled",
|
|
result={"text": "Completed first topic."},
|
|
local_profiles=LOCAL_PROFILES,
|
|
)
|
|
assert [event.kind for event in publication.events] == [
|
|
"message.member",
|
|
"turn.settled",
|
|
]
|
|
|
|
|
|
def test_oversized_member_reply_is_truncated_and_next_turn_stays_serviceable(
|
|
room_db: tuple[Path, dict],
|
|
):
|
|
db, room = room_db
|
|
_append_user(
|
|
db,
|
|
event_id="user-large",
|
|
text="u" * discussion.MAX_USER_TEXT_BYTES,
|
|
)
|
|
first = _next_task(room, db)
|
|
publication = discussion.plan_publication(
|
|
room,
|
|
_events(db),
|
|
first,
|
|
status="settled",
|
|
result={"text": "é" * (discussion.MAX_MEMBER_TEXT_BYTES + 100)},
|
|
local_profiles=LOCAL_PROFILES,
|
|
)
|
|
|
|
member_event = next(event for event in publication.events if event.kind == "message.member")
|
|
member_text = member_event.payload["text"]
|
|
assert len(member_text.encode("utf-8")) <= discussion.MAX_MEMBER_TEXT_BYTES
|
|
assert member_text.endswith("share the full result as a file.]")
|
|
_append_publication(db, publication)
|
|
|
|
followup = _next_task(room, db)
|
|
assert len(followup.payload["prompt"].encode("utf-8")) <= driver.MAX_PROMPT_BYTES
|
|
assert "Earlier content omitted" in followup.payload["prompt"]
|
|
|
|
|
|
def test_three_round_bound(room_db: tuple[Path, dict]):
|
|
db, room = room_db
|
|
room["members"] = MEMBERS[:2]
|
|
_append_user(db, event_id="user-1", text="Discuss.")
|
|
|
|
for index in range(6):
|
|
task = _next_task(room, db)
|
|
peer = "build" if task.member.profile == "research" else "research"
|
|
publication = discussion.plan_publication(
|
|
room,
|
|
_events(db),
|
|
task,
|
|
status="settled",
|
|
result={"text": f"Reply {index}. @{peer}"},
|
|
local_profiles=LOCAL_PROFILES,
|
|
)
|
|
_append_publication(db, publication)
|
|
|
|
decision = discussion.plan_next_task(
|
|
room,
|
|
_events(db),
|
|
local_profiles=LOCAL_PROFILES,
|
|
)
|
|
assert decision.status == "bounded"
|
|
assert decision.reason == "max_rounds"
|
|
|
|
|
|
def test_ten_message_bound(tmp_path: Path):
|
|
db = tmp_path / "state.db"
|
|
members = [
|
|
{
|
|
"member_id": f"member-{profile}",
|
|
"profile": profile,
|
|
"handle": profile,
|
|
}
|
|
for profile in LOCAL_PROFILES
|
|
]
|
|
room = hosted_rooms.create_room(
|
|
db,
|
|
room_id=ROOM_ID,
|
|
name="Large",
|
|
members=members,
|
|
authority_gateway_id=GATEWAY_ID,
|
|
now=1,
|
|
)
|
|
_append_user(db, event_id="user-1", text="Discuss.")
|
|
|
|
for index in range(discussion.MAX_DISCUSSION_MESSAGES):
|
|
_settle_next(room, db, text=f"Reply {index}. @everyone")
|
|
|
|
decision = discussion.plan_next_task(
|
|
room,
|
|
_events(db),
|
|
local_profiles=LOCAL_PROFILES,
|
|
)
|
|
assert decision.status == "bounded"
|
|
assert decision.reason == "max_messages"
|
|
|
|
|
|
def test_prompt_delta_is_bounded_to_24_message_lines(
|
|
room_db: tuple[Path, dict],
|
|
):
|
|
db, room = room_db
|
|
for index in range(30):
|
|
_append_user(
|
|
db,
|
|
event_id=f"user-{index}",
|
|
text=f"Message {index}.",
|
|
)
|
|
|
|
task = _next_task(room, db)
|
|
assert task.payload["prompt"].count("User (user):") == 24
|
|
assert "Message 5." not in task.payload["prompt"]
|
|
assert "Message 6." in task.payload["prompt"]
|
|
assert "Message 29." in task.payload["prompt"]
|
|
|
|
|
|
def test_attachment_payload_is_rejected_by_local_text_only_boundary():
|
|
with pytest.raises(discussion.DiscussionValidationError, match="unknown fields"):
|
|
discussion.validate_user_payload({
|
|
"text": "Review.",
|
|
"thread_id": "thread-1",
|
|
"attachments": [{"name": "notes.txt"}],
|
|
})
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("members", "match"),
|
|
[
|
|
(MEMBERS[:1], "between 2 and 6"),
|
|
(MEMBERS + MEMBERS + MEMBERS[:1], "between 2 and 6"),
|
|
(
|
|
[MEMBERS[0], {**MEMBERS[1], "profile": "research"}],
|
|
"profiles must be unique",
|
|
),
|
|
([MEMBERS[0], {**MEMBERS[1], "handle": "RESEARCH"}], "handles must be unique"),
|
|
(
|
|
[MEMBERS[0], {**MEMBERS[1], "member_id": "MEMBER-RESEARCH"}],
|
|
"ids must be unique",
|
|
),
|
|
([MEMBERS[0], {**MEMBERS[1], "route": {"mode": "ssh"}}], "cross-gateway"),
|
|
([MEMBERS[0], {**MEMBERS[1], "connectionId": "remote"}], "cross-gateway"),
|
|
([MEMBERS[0], {**MEMBERS[1], "profile": "missing"}], "not local"),
|
|
],
|
|
)
|
|
def test_malformed_or_remote_roster_is_rejected(members: list[dict], match: str):
|
|
with pytest.raises(discussion.DiscussionValidationError, match=match):
|
|
discussion.validate_roster(members, local_profiles=LOCAL_PROFILES)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"payload",
|
|
[
|
|
{"text": "hello"},
|
|
{"text": "hello", "thread_id": "thread-1", "images": []},
|
|
{"text": "", "thread_id": "thread-1"},
|
|
{"text": "hello", "thread_id": "../escape"},
|
|
{"text": ["hello"], "thread_id": "thread-1"},
|
|
],
|
|
)
|
|
def test_user_payload_is_exact_and_text_only(payload: dict):
|
|
with pytest.raises(discussion.DiscussionValidationError):
|
|
discussion.validate_user_payload(payload)
|
|
|
|
|
|
def test_malformed_log_and_task_reconstruction_fail_closed(
|
|
room_db: tuple[Path, dict],
|
|
):
|
|
db, room = room_db
|
|
_append_user(db, event_id="user-1", text="Report.")
|
|
_append_user(db, event_id="user-2", text="Report again.")
|
|
task = _next_task(room, db)
|
|
events = _events(db)
|
|
|
|
with pytest.raises(discussion.DiscussionValidationError, match="sequence order"):
|
|
discussion.plan_next_task(
|
|
room,
|
|
list(reversed(events)),
|
|
local_profiles=LOCAL_PROFILES,
|
|
)
|
|
|
|
malformed = {
|
|
"identity": driver.TaskIdentity(
|
|
room_id=task.identity.room_id,
|
|
task_id="dtask:wrong",
|
|
thread_id=task.identity.thread_id,
|
|
turn_id=task.identity.turn_id,
|
|
),
|
|
"payload": dict(task.payload),
|
|
}
|
|
with pytest.raises(
|
|
discussion.DiscussionReconstructionError,
|
|
match="deterministic reconstruction",
|
|
):
|
|
discussion.reconstruct_task_plan(
|
|
room,
|
|
events,
|
|
malformed,
|
|
local_profiles=LOCAL_PROFILES,
|
|
)
|