82 lines
2.6 KiB
Python
82 lines
2.6 KiB
Python
"""Tests for clickup_runner.state."""
|
|
|
|
import json
|
|
|
|
import pytest
|
|
|
|
from clickup_runner.state import StateDB
|
|
|
|
|
|
@pytest.fixture
|
|
def db(tmp_path):
|
|
return StateDB(tmp_path / "test.db")
|
|
|
|
|
|
class TestKVStore:
|
|
def test_set_and_get(self, db):
|
|
db.kv_set("key1", "value1")
|
|
assert db.kv_get("key1") == "value1"
|
|
|
|
def test_get_missing_key(self, db):
|
|
assert db.kv_get("nope") is None
|
|
|
|
def test_set_overwrites(self, db):
|
|
db.kv_set("key1", "v1")
|
|
db.kv_set("key1", "v2")
|
|
assert db.kv_get("key1") == "v2"
|
|
|
|
def test_delete(self, db):
|
|
db.kv_set("key1", "v1")
|
|
db.kv_delete("key1")
|
|
assert db.kv_get("key1") is None
|
|
|
|
def test_scan(self, db):
|
|
db.kv_set("autocora:job:kw1", "submitted")
|
|
db.kv_set("autocora:job:kw2", "submitted")
|
|
db.kv_set("other:key", "val")
|
|
results = db.kv_scan("autocora:job:")
|
|
assert len(results) == 2
|
|
keys = {k for k, _ in results}
|
|
assert keys == {"autocora:job:kw1", "autocora:job:kw2"}
|
|
|
|
def test_json_round_trip(self, db):
|
|
data = {"status": "submitted", "job_id": "job-001", "task_ids": ["t1", "t2"]}
|
|
db.kv_set_json("autocora:job:test", data)
|
|
result = db.kv_get_json("autocora:job:test")
|
|
assert result == data
|
|
|
|
def test_json_get_missing(self, db):
|
|
assert db.kv_get_json("nope") is None
|
|
|
|
|
|
class TestRunLog:
|
|
def test_log_start_and_finish(self, db):
|
|
run_id = db.log_run_start("t1", "Test Task", "Content Creation", "outline")
|
|
assert run_id > 0
|
|
|
|
db.log_run_finish(run_id, "completed", result="outline.md created")
|
|
runs = db.get_recent_runs(limit=1)
|
|
assert len(runs) == 1
|
|
assert runs[0]["task_id"] == "t1"
|
|
assert runs[0]["status"] == "completed"
|
|
assert runs[0]["result"] == "outline.md created"
|
|
assert runs[0]["error"] is None
|
|
|
|
def test_log_error(self, db):
|
|
run_id = db.log_run_start("t2", "Failing Task", "Press Release", "draft")
|
|
db.log_run_finish(run_id, "error", error="Claude Code exit code 1")
|
|
runs = db.get_recent_runs(limit=1)
|
|
assert runs[0]["status"] == "error"
|
|
assert "exit code 1" in runs[0]["error"]
|
|
|
|
def test_recent_runs_ordered(self, db):
|
|
r1 = db.log_run_start("t1", "First", "PR", "draft")
|
|
db.log_run_finish(r1, "completed")
|
|
r2 = db.log_run_start("t2", "Second", "CC", "outline")
|
|
db.log_run_finish(r2, "completed")
|
|
|
|
runs = db.get_recent_runs(limit=10)
|
|
# Most recent first
|
|
assert runs[0]["task_name"] == "Second"
|
|
assert runs[1]["task_name"] == "First"
|