106 lines
3.4 KiB
Python
106 lines
3.4 KiB
Python
"""Chat service-layer tests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid as _uuid
|
|
|
|
import pytest
|
|
|
|
from claire.domain import ChatRole, ChatScope
|
|
from claire.web import service
|
|
|
|
|
|
def test_post_orchestrator_message_returns_row(conn, gen) -> None:
|
|
msg = service.post_chat_message(
|
|
conn, gen,
|
|
scope=ChatScope.ORCHESTRATOR, scope_ref=None,
|
|
role=ChatRole.USER, body="hi",
|
|
)
|
|
assert msg.body == "hi"
|
|
assert msg.role == ChatRole.USER
|
|
assert msg.rowid > 0
|
|
|
|
|
|
def test_post_project_message_requires_existing_project(conn, gen) -> None:
|
|
with pytest.raises(service.NotFound):
|
|
service.post_chat_message(
|
|
conn, gen,
|
|
scope=ChatScope.PROJECT, scope_ref="missing",
|
|
role=ChatRole.USER, body="hi",
|
|
)
|
|
|
|
|
|
def test_post_session_rejects_non_uuid_ref(conn, gen) -> None:
|
|
with pytest.raises(service.InvalidInput):
|
|
service.post_chat_message(
|
|
conn, gen,
|
|
scope=ChatScope.SESSION, scope_ref="not-a-uuid",
|
|
role=ChatRole.USER, body="hi",
|
|
)
|
|
|
|
|
|
def test_orchestrator_rejects_scope_ref(conn, gen) -> None:
|
|
with pytest.raises(service.InvalidInput):
|
|
service.post_chat_message(
|
|
conn, gen,
|
|
scope=ChatScope.ORCHESTRATOR, scope_ref="something",
|
|
role=ChatRole.USER, body="hi",
|
|
)
|
|
|
|
|
|
def test_empty_body_rejected(conn, gen) -> None:
|
|
with pytest.raises(service.InvalidInput):
|
|
service.post_chat_message(
|
|
conn, gen,
|
|
scope=ChatScope.ORCHESTRATOR, scope_ref=None,
|
|
role=ChatRole.USER, body=" ",
|
|
)
|
|
|
|
|
|
def test_list_chat_messages_cursor(conn, gen) -> None:
|
|
service.post_chat_message(
|
|
conn, gen, scope=ChatScope.ORCHESTRATOR, scope_ref=None,
|
|
role=ChatRole.USER, body="first",
|
|
)
|
|
m2 = service.post_chat_message(
|
|
conn, gen, scope=ChatScope.ORCHESTRATOR, scope_ref=None,
|
|
role=ChatRole.USER, body="second",
|
|
)
|
|
after = service.list_chat_messages(
|
|
conn, scope=ChatScope.ORCHESTRATOR, scope_ref=None,
|
|
after_rowid=m2.rowid - 1,
|
|
)
|
|
assert len(after) == 1
|
|
assert after[0].body == "second"
|
|
|
|
|
|
def test_autocomplete_project_prefix(conn, gen) -> None:
|
|
from claire import events as ev
|
|
pid1, pid2 = _uuid.uuid4(), _uuid.uuid4()
|
|
ev.append(conn, gen, ev.ProjectCreated(project_id=pid1, name="alpha"))
|
|
ev.append(conn, gen, ev.ProjectCreated(project_id=pid2, name="beta"))
|
|
hits = service.chat_autocomplete(conn, kind="project", query="al")
|
|
assert [h.value for h in hits] == ["alpha"]
|
|
|
|
|
|
def test_autocomplete_task_by_id_prefix(conn, gen) -> None:
|
|
from claire import events as ev
|
|
pid, tid = _uuid.uuid4(), _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="t1"))
|
|
short = str(tid)[:6]
|
|
hits = service.chat_autocomplete(conn, kind="task", query=short)
|
|
assert any(h.value == str(tid) for h in hits)
|
|
|
|
|
|
def test_autocomplete_rejects_unknown_kind(conn) -> None:
|
|
with pytest.raises(service.InvalidInput):
|
|
service.chat_autocomplete(conn, kind="garbage", query="x")
|
|
|
|
|
|
def test_resolve_session_id_prefix(conn, gen) -> None:
|
|
from claire import events as ev
|
|
sid = _uuid.uuid4()
|
|
ev.append(conn, gen, ev.SessionObserved(session_uuid=sid, host="local"))
|
|
resolved = service.resolve_session_id(conn, str(sid)[:8])
|
|
assert resolved == sid
|