148 lines
4.5 KiB
Python
148 lines
4.5 KiB
Python
"""Tests for the ClickUp chat tools."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
|
|
from cheddahbot.tools.clickup_tool import (
|
|
clickup_list_tasks,
|
|
clickup_reset_all,
|
|
clickup_reset_task,
|
|
clickup_task_status,
|
|
)
|
|
|
|
|
|
def _make_ctx(db):
|
|
return {"db": db}
|
|
|
|
|
|
def _seed_task(db, task_id, state, **overrides):
|
|
"""Insert a task state into kv_store."""
|
|
data = {
|
|
"state": state,
|
|
"clickup_task_id": task_id,
|
|
"clickup_task_name": f"Task {task_id}",
|
|
"task_type": "Press Release",
|
|
"skill_name": "write_press_releases",
|
|
"discovered_at": "2026-01-01T00:00:00",
|
|
"started_at": None,
|
|
"completed_at": None,
|
|
"error": None,
|
|
"deliverable_paths": [],
|
|
"custom_fields": {},
|
|
}
|
|
data.update(overrides)
|
|
db.kv_set(f"clickup:task:{task_id}:state", json.dumps(data))
|
|
|
|
|
|
class TestClickupListTasks:
|
|
def test_empty_when_no_tasks(self, tmp_db):
|
|
result = clickup_list_tasks(ctx=_make_ctx(tmp_db))
|
|
assert "No ClickUp tasks" in result
|
|
|
|
def test_lists_all_tracked_tasks(self, tmp_db):
|
|
_seed_task(tmp_db, "a1", "discovered")
|
|
_seed_task(tmp_db, "a2", "approved")
|
|
|
|
result = clickup_list_tasks(ctx=_make_ctx(tmp_db))
|
|
|
|
assert "a1" in result
|
|
assert "a2" in result
|
|
assert "2" in result # count
|
|
|
|
def test_filter_by_status(self, tmp_db):
|
|
_seed_task(tmp_db, "a1", "discovered")
|
|
_seed_task(tmp_db, "a2", "approved")
|
|
_seed_task(tmp_db, "a3", "completed")
|
|
|
|
result = clickup_list_tasks(status="approved", ctx=_make_ctx(tmp_db))
|
|
|
|
assert "a2" in result
|
|
assert "a1" not in result
|
|
assert "a3" not in result
|
|
|
|
def test_filter_returns_empty_message(self, tmp_db):
|
|
_seed_task(tmp_db, "a1", "discovered")
|
|
|
|
result = clickup_list_tasks(status="completed", ctx=_make_ctx(tmp_db))
|
|
|
|
assert "No ClickUp tasks with state" in result
|
|
|
|
|
|
class TestClickupTaskStatus:
|
|
def test_shows_details(self, tmp_db):
|
|
_seed_task(tmp_db, "a1", "executing", started_at="2026-01-01T12:00:00")
|
|
|
|
result = clickup_task_status(task_id="a1", ctx=_make_ctx(tmp_db))
|
|
|
|
assert "Task a1" in result
|
|
assert "executing" in result
|
|
assert "Press Release" in result
|
|
assert "2026-01-01T12:00:00" in result
|
|
|
|
def test_unknown_task(self, tmp_db):
|
|
result = clickup_task_status(task_id="nonexistent", ctx=_make_ctx(tmp_db))
|
|
|
|
assert "No tracked state" in result
|
|
|
|
def test_shows_error_when_failed(self, tmp_db):
|
|
_seed_task(tmp_db, "f1", "failed", error="API timeout")
|
|
|
|
result = clickup_task_status(task_id="f1", ctx=_make_ctx(tmp_db))
|
|
|
|
assert "API timeout" in result
|
|
|
|
def test_shows_deliverables(self, tmp_db):
|
|
_seed_task(tmp_db, "c1", "completed", deliverable_paths=["/data/pr1.txt", "/data/pr2.txt"])
|
|
|
|
result = clickup_task_status(task_id="c1", ctx=_make_ctx(tmp_db))
|
|
|
|
assert "/data/pr1.txt" in result
|
|
|
|
|
|
class TestClickupResetTask:
|
|
def test_resets_failed_task(self, tmp_db):
|
|
_seed_task(tmp_db, "f1", "failed")
|
|
|
|
result = clickup_reset_task(task_id="f1", ctx=_make_ctx(tmp_db))
|
|
|
|
assert "cleared" in result.lower()
|
|
assert tmp_db.kv_get("clickup:task:f1:state") is None
|
|
|
|
def test_resets_completed_task(self, tmp_db):
|
|
_seed_task(tmp_db, "c1", "completed")
|
|
|
|
result = clickup_reset_task(task_id="c1", ctx=_make_ctx(tmp_db))
|
|
|
|
assert "cleared" in result.lower()
|
|
assert tmp_db.kv_get("clickup:task:c1:state") is None
|
|
|
|
def test_unknown_task(self, tmp_db):
|
|
result = clickup_reset_task(task_id="nope", ctx=_make_ctx(tmp_db))
|
|
assert "Nothing to reset" in result
|
|
|
|
|
|
class TestClickupResetAll:
|
|
def test_clears_all_states(self, tmp_db):
|
|
_seed_task(tmp_db, "a1", "completed")
|
|
_seed_task(tmp_db, "a2", "failed")
|
|
_seed_task(tmp_db, "a3", "executing")
|
|
|
|
result = clickup_reset_all(ctx=_make_ctx(tmp_db))
|
|
|
|
assert "3" in result
|
|
assert tmp_db.kv_get("clickup:task:a1:state") is None
|
|
assert tmp_db.kv_get("clickup:task:a2:state") is None
|
|
assert tmp_db.kv_get("clickup:task:a3:state") is None
|
|
|
|
def test_clears_legacy_active_ids(self, tmp_db):
|
|
tmp_db.kv_set("clickup:active_task_ids", json.dumps(["a1", "a2"]))
|
|
|
|
clickup_reset_all(ctx=_make_ctx(tmp_db))
|
|
|
|
assert tmp_db.kv_get("clickup:active_task_ids") is None
|
|
|
|
def test_empty_returns_zero(self, tmp_db):
|
|
result = clickup_reset_all(ctx=_make_ctx(tmp_db))
|
|
assert "0" in result
|