2026-05-18 07:53:30 -07:00
|
|
|
"""Slash-command parser + dispatcher tests."""
|
|
|
|
|
|
|
|
|
|
from __future__ import annotations
|
|
|
|
|
|
|
|
|
|
import uuid as _uuid
|
|
|
|
|
|
|
|
|
|
import pytest
|
|
|
|
|
from pydantic import ValidationError
|
|
|
|
|
|
2026-05-20 19:54:05 -07:00
|
|
|
from claire import events as ev
|
|
|
|
|
from claire.domain import ChatScope, TaskStatus
|
|
|
|
|
from claire.web.chat import commands as cmd
|
2026-05-18 07:53:30 -07:00
|
|
|
|
|
|
|
|
|
|
|
|
|
ORCH = cmd.ScopeCtx(scope=ChatScope.ORCHESTRATOR, scope_ref=None)
|
|
|
|
|
ALPHA = cmd.ScopeCtx(scope=ChatScope.PROJECT, scope_ref="alpha")
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
# is_slash / parse error paths
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_is_slash_leading_whitespace() -> None:
|
|
|
|
|
assert cmd.is_slash(" /help")
|
|
|
|
|
assert not cmd.is_slash("hello")
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_parse_empty_command_rejected() -> None:
|
|
|
|
|
with pytest.raises(cmd.ParseError):
|
|
|
|
|
cmd.parse("/", ORCH)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_parse_unknown_command_rejected() -> None:
|
|
|
|
|
with pytest.raises(cmd.ParseError):
|
|
|
|
|
cmd.parse("/nope", ORCH)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_parse_unterminated_quote_rejected() -> None:
|
|
|
|
|
with pytest.raises(cmd.ParseError):
|
|
|
|
|
cmd.parse('/project new "alpha', ORCH)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
# /help, /pull, /status
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_parse_help() -> None:
|
|
|
|
|
assert isinstance(cmd.parse("/help", ORCH), cmd.HelpCmd)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_parse_pull() -> None:
|
|
|
|
|
assert isinstance(cmd.parse("/pull", ORCH), cmd.PullCmd)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_parse_status() -> None:
|
|
|
|
|
assert isinstance(cmd.parse("/status", ORCH), cmd.StatusCmd)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
# /project new
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_parse_project_new_with_goal() -> None:
|
|
|
|
|
c = cmd.parse('/project new alpha --goal "build it"', ORCH)
|
|
|
|
|
assert isinstance(c, cmd.ProjectNewCmd)
|
|
|
|
|
assert c.name == "alpha"
|
|
|
|
|
assert c.goal == "build it"
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_parse_project_new_equals_form() -> None:
|
|
|
|
|
c = cmd.parse('/project new alpha --goal=focused', ORCH)
|
|
|
|
|
assert isinstance(c, cmd.ProjectNewCmd)
|
|
|
|
|
assert c.goal == "focused"
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_parse_project_new_requires_name() -> None:
|
|
|
|
|
with pytest.raises(cmd.ParseError):
|
|
|
|
|
cmd.parse("/project new", ORCH)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
# /task new — scope-aware defaults
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_parse_task_new_defaults_project_from_scope() -> None:
|
|
|
|
|
c = cmd.parse('/task new "first task" --priority 1', ALPHA)
|
|
|
|
|
assert isinstance(c, cmd.TaskNewCmd)
|
|
|
|
|
assert c.project == "alpha"
|
|
|
|
|
assert c.title == "first task"
|
|
|
|
|
assert c.priority == 1
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_parse_task_new_requires_project_outside_project_chat() -> None:
|
|
|
|
|
with pytest.raises(cmd.ParseError):
|
|
|
|
|
cmd.parse('/task new "x"', ORCH)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_parse_task_new_explicit_project_wins() -> None:
|
|
|
|
|
c = cmd.parse('/task new "x" --project beta', ALPHA)
|
|
|
|
|
assert isinstance(c, cmd.TaskNewCmd)
|
|
|
|
|
assert c.project == "beta"
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_parse_task_new_priority_out_of_range_rejected() -> None:
|
|
|
|
|
with pytest.raises(ValidationError):
|
|
|
|
|
cmd.parse('/task new "x" --priority 99', ALPHA)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_parse_task_new_priority_non_integer_rejected() -> None:
|
|
|
|
|
with pytest.raises(cmd.ParseError):
|
|
|
|
|
cmd.parse('/task new "x" --priority foo', ALPHA)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
# /task list
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_parse_task_list_status_filter() -> None:
|
|
|
|
|
c = cmd.parse("/task list --status todo", ALPHA)
|
|
|
|
|
assert isinstance(c, cmd.TaskListCmd)
|
|
|
|
|
assert c.status == TaskStatus.TODO
|
|
|
|
|
assert c.project == "alpha"
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_parse_task_list_unknown_status_rejected() -> None:
|
|
|
|
|
with pytest.raises(cmd.ParseError):
|
|
|
|
|
cmd.parse("/task list --status garbage", ALPHA)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
# /assign
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_parse_assign_two_positionals() -> None:
|
|
|
|
|
c = cmd.parse("/assign abcd1234 efgh5678", ORCH)
|
|
|
|
|
assert isinstance(c, cmd.AssignCmd)
|
|
|
|
|
assert c.task_ref == "abcd1234"
|
|
|
|
|
assert c.session_ref == "efgh5678"
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_parse_assign_requires_two_args() -> None:
|
|
|
|
|
with pytest.raises(cmd.ParseError):
|
|
|
|
|
cmd.parse("/assign abcd1234", ORCH)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
# /broadcast — scope sensitivity
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_parse_broadcast_uses_project_scope_as_target() -> None:
|
|
|
|
|
c = cmd.parse("/broadcast start now", ALPHA)
|
|
|
|
|
assert isinstance(c, cmd.BroadcastCmd)
|
|
|
|
|
assert c.target == "alpha"
|
|
|
|
|
assert c.text == "start now"
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_parse_broadcast_orchestrator_rejected() -> None:
|
|
|
|
|
with pytest.raises(cmd.ParseError):
|
|
|
|
|
cmd.parse("/broadcast hi", ORCH)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_parse_broadcast_session_scope_uses_short_prefix() -> None:
|
|
|
|
|
sid = "abcd1234-5678-90ab-cdef-1234567890ab"
|
|
|
|
|
ctx = cmd.ScopeCtx(scope=ChatScope.SESSION, scope_ref=sid)
|
|
|
|
|
c = cmd.parse("/broadcast hello there", ctx)
|
|
|
|
|
assert isinstance(c, cmd.BroadcastCmd)
|
|
|
|
|
assert c.target == "@abcd1234"
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
# Dispatcher — runs against an in-memory DB
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_dispatch_help(conn, gen) -> None:
|
|
|
|
|
result = cmd.dispatch(conn, gen, cmd.HelpCmd(), ORCH)
|
|
|
|
|
assert "Slash commands" in result.body
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_dispatch_project_new_then_task_new(conn, gen) -> None:
|
|
|
|
|
cmd.dispatch(conn, gen, cmd.ProjectNewCmd(name="alpha"), ORCH)
|
|
|
|
|
result = cmd.dispatch(
|
|
|
|
|
conn, gen,
|
|
|
|
|
cmd.TaskNewCmd(project="alpha", title="first", priority=1),
|
|
|
|
|
ALPHA,
|
|
|
|
|
)
|
|
|
|
|
assert "first" in result.body
|
|
|
|
|
assert result.meta["kind"] == "task"
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_dispatch_task_list_returns_cards(conn, gen) -> None:
|
|
|
|
|
cmd.dispatch(conn, gen, cmd.ProjectNewCmd(name="alpha"), ORCH)
|
|
|
|
|
cmd.dispatch(conn, gen, cmd.TaskNewCmd(project="alpha", title="t1"), ALPHA)
|
|
|
|
|
cmd.dispatch(conn, gen, cmd.TaskNewCmd(project="alpha", title="t2"), ALPHA)
|
|
|
|
|
result = cmd.dispatch(
|
|
|
|
|
conn, gen, cmd.TaskListCmd(project="alpha", status=None), ALPHA,
|
|
|
|
|
)
|
|
|
|
|
assert len(result.meta["tasks"]) == 2
|
|
|
|
|
titles = {t["title"] for t in result.meta["tasks"]}
|
|
|
|
|
assert titles == {"t1", "t2"}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_dispatch_assign_resolves_prefixes(conn, gen) -> None:
|
|
|
|
|
pid = _uuid.uuid4()
|
|
|
|
|
tid = _uuid.uuid4()
|
|
|
|
|
sid = _uuid.uuid4()
|
|
|
|
|
ev.append(conn, gen, ev.ProjectCreated(project_id=pid, name="alpha"))
|
|
|
|
|
ev.append(conn, gen, ev.TaskAdded(task_id=tid, project_id=pid, title="t"))
|
|
|
|
|
ev.append(conn, gen, ev.SessionObserved(session_uuid=sid, host="local"))
|
|
|
|
|
|
|
|
|
|
result = cmd.dispatch(
|
|
|
|
|
conn, gen,
|
|
|
|
|
cmd.AssignCmd(task_ref=str(tid)[:8], session_ref=str(sid)[:8]),
|
|
|
|
|
ALPHA,
|
|
|
|
|
)
|
|
|
|
|
assert "Assigned" in result.body
|
|
|
|
|
assert result.meta["task_id"] == str(tid)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_dispatch_status_summary(conn, gen) -> None:
|
|
|
|
|
cmd.dispatch(conn, gen, cmd.ProjectNewCmd(name="alpha"), ORCH)
|
|
|
|
|
cmd.dispatch(conn, gen, cmd.TaskNewCmd(project="alpha", title="t"), ALPHA)
|
|
|
|
|
result = cmd.dispatch(conn, gen, cmd.StatusCmd(), ORCH)
|
|
|
|
|
assert result.meta["projects"] == 1
|
|
|
|
|
assert result.meta["open_tasks"] == 1
|