167 lines
5.8 KiB
Python
167 lines
5.8 KiB
Python
"""Tests for the Database kv_scan, notifications, and conversations methods."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
|
|
from cheddahbot.db import Database
|
|
|
|
|
|
class TestConversationsAgentName:
|
|
"""Conversations are tagged by agent_name for per-agent history filtering."""
|
|
|
|
def test_create_with_default_agent_name(self, tmp_db):
|
|
tmp_db.create_conversation("conv1")
|
|
convs = tmp_db.list_conversations()
|
|
assert len(convs) == 1
|
|
assert convs[0]["agent_name"] == "default"
|
|
|
|
def test_create_with_custom_agent_name(self, tmp_db):
|
|
tmp_db.create_conversation("conv1", agent_name="writer")
|
|
convs = tmp_db.list_conversations()
|
|
assert convs[0]["agent_name"] == "writer"
|
|
|
|
def test_list_filters_by_agent_name(self, tmp_db):
|
|
tmp_db.create_conversation("c1", agent_name="default")
|
|
tmp_db.create_conversation("c2", agent_name="writer")
|
|
tmp_db.create_conversation("c3", agent_name="default")
|
|
|
|
default_convs = tmp_db.list_conversations(agent_name="default")
|
|
writer_convs = tmp_db.list_conversations(agent_name="writer")
|
|
all_convs = tmp_db.list_conversations()
|
|
|
|
assert len(default_convs) == 2
|
|
assert len(writer_convs) == 1
|
|
assert len(all_convs) == 3
|
|
|
|
def test_list_without_filter_returns_all(self, tmp_db):
|
|
tmp_db.create_conversation("c1", agent_name="a")
|
|
tmp_db.create_conversation("c2", agent_name="b")
|
|
|
|
convs = tmp_db.list_conversations()
|
|
assert len(convs) == 2
|
|
|
|
def test_list_returns_agent_name_in_results(self, tmp_db):
|
|
tmp_db.create_conversation("c1", agent_name="researcher")
|
|
convs = tmp_db.list_conversations()
|
|
assert "agent_name" in convs[0]
|
|
assert convs[0]["agent_name"] == "researcher"
|
|
|
|
def test_migration_idempotency(self, tmp_path):
|
|
"""Running _init_schema twice doesn't error (ALTER TABLE is skipped)."""
|
|
db_path = tmp_path / "test_migrate.db"
|
|
db1 = Database(db_path)
|
|
db1.create_conversation("c1", agent_name="ops")
|
|
# Re-init on same DB file triggers migration again
|
|
db2 = Database(db_path)
|
|
convs = db2.list_conversations()
|
|
assert len(convs) == 1
|
|
assert convs[0]["agent_name"] == "ops"
|
|
|
|
|
|
class TestKvScan:
|
|
"""kv_scan is used to find all tracked ClickUp task states efficiently.
|
|
If it breaks, the scheduler can't find tasks to execute or recover."""
|
|
|
|
def test_returns_matching_pairs(self, tmp_db):
|
|
tmp_db.kv_set("clickup:task:abc:state", '{"state": "discovered"}')
|
|
tmp_db.kv_set("clickup:task:def:state", '{"state": "approved"}')
|
|
tmp_db.kv_set("other:key", "unrelated")
|
|
|
|
results = tmp_db.kv_scan("clickup:task:")
|
|
|
|
assert len(results) == 2
|
|
keys = {k for k, _ in results}
|
|
assert keys == {"clickup:task:abc:state", "clickup:task:def:state"}
|
|
|
|
def test_returns_empty_on_no_match(self, tmp_db):
|
|
tmp_db.kv_set("other:key", "value")
|
|
|
|
results = tmp_db.kv_scan("clickup:")
|
|
|
|
assert results == []
|
|
|
|
def test_prefix_is_exact_not_substring(self, tmp_db):
|
|
"""'click' should not match 'clickup:' prefix."""
|
|
tmp_db.kv_set("clickup:task:1:state", "data")
|
|
tmp_db.kv_set("clicked:something", "other")
|
|
|
|
results = tmp_db.kv_scan("clickup:")
|
|
|
|
assert len(results) == 1
|
|
assert results[0][0] == "clickup:task:1:state"
|
|
|
|
def test_values_are_returned_correctly(self, tmp_db):
|
|
state = json.dumps({"state": "completed", "task_name": "Test"})
|
|
tmp_db.kv_set("clickup:task:x:state", state)
|
|
|
|
results = tmp_db.kv_scan("clickup:task:x:")
|
|
|
|
assert len(results) == 1
|
|
parsed = json.loads(results[0][1])
|
|
assert parsed["state"] == "completed"
|
|
assert parsed["task_name"] == "Test"
|
|
|
|
|
|
class TestNotifications:
|
|
"""Notifications back the NotificationBus. If these break, no UI
|
|
gets informed about ClickUp task discoveries, completions, or failures."""
|
|
|
|
def test_add_and_retrieve(self, tmp_db):
|
|
nid = tmp_db.add_notification("Task discovered", "clickup")
|
|
|
|
assert nid >= 1
|
|
notifs = tmp_db.get_notifications_after(0)
|
|
assert len(notifs) == 1
|
|
assert notifs[0]["message"] == "Task discovered"
|
|
assert notifs[0]["category"] == "clickup"
|
|
|
|
def test_after_id_filters_correctly(self, tmp_db):
|
|
id1 = tmp_db.add_notification("First", "clickup")
|
|
_id2 = tmp_db.add_notification("Second", "clickup")
|
|
_id3 = tmp_db.add_notification("Third", "clickup")
|
|
|
|
# Should only get notifications after id1
|
|
notifs = tmp_db.get_notifications_after(id1)
|
|
|
|
assert len(notifs) == 2
|
|
assert notifs[0]["message"] == "Second"
|
|
assert notifs[1]["message"] == "Third"
|
|
|
|
def test_after_latest_returns_empty(self, tmp_db):
|
|
id1 = tmp_db.add_notification("Only one", "clickup")
|
|
|
|
notifs = tmp_db.get_notifications_after(id1)
|
|
|
|
assert notifs == []
|
|
|
|
def test_limit_is_respected(self, tmp_db):
|
|
for i in range(10):
|
|
tmp_db.add_notification(f"Msg {i}", "clickup")
|
|
|
|
notifs = tmp_db.get_notifications_after(0, limit=3)
|
|
|
|
assert len(notifs) == 3
|
|
|
|
def test_default_category(self, tmp_db):
|
|
tmp_db.add_notification("No category specified")
|
|
|
|
notifs = tmp_db.get_notifications_after(0)
|
|
|
|
assert notifs[0]["category"] == "clickup"
|
|
|
|
def test_created_at_is_populated(self, tmp_db):
|
|
tmp_db.add_notification("Timestamped")
|
|
|
|
notifs = tmp_db.get_notifications_after(0)
|
|
|
|
assert notifs[0]["created_at"] is not None
|
|
assert len(notifs[0]["created_at"]) > 10 # ISO format
|
|
|
|
def test_ids_are_monotonically_increasing(self, tmp_db):
|
|
id1 = tmp_db.add_notification("A")
|
|
id2 = tmp_db.add_notification("B")
|
|
id3 = tmp_db.add_notification("C")
|
|
|
|
assert id1 < id2 < id3
|