480 lines
16 KiB
Python
480 lines
16 KiB
Python
"""Tests for AutoCora job submission and result polling."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from unittest.mock import MagicMock
|
|
|
|
import pytest
|
|
|
|
from cheddahbot.config import AutoCoraConfig, ClickUpConfig, Config
|
|
from cheddahbot.tools.autocora import (
|
|
_group_by_keyword,
|
|
_make_job_id,
|
|
_parse_result,
|
|
_slugify,
|
|
poll_autocora_results,
|
|
submit_autocora_jobs,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fixtures
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@dataclass
|
|
class FakeTask:
|
|
"""Minimal stand-in for ClickUpTask."""
|
|
|
|
id: str
|
|
name: str
|
|
status: str = "to do"
|
|
task_type: str = "Content Creation"
|
|
due_date: str = ""
|
|
custom_fields: dict[str, Any] = field(default_factory=dict)
|
|
|
|
|
|
@pytest.fixture()
|
|
def autocora_config():
|
|
return AutoCoraConfig(enabled=True)
|
|
|
|
|
|
@pytest.fixture()
|
|
def cfg(tmp_path, autocora_config):
|
|
"""Config with AutoCora pointing at tmp dirs."""
|
|
jobs_dir = tmp_path / "jobs"
|
|
results_dir = tmp_path / "results"
|
|
jobs_dir.mkdir()
|
|
results_dir.mkdir()
|
|
autocora_config.jobs_dir = str(jobs_dir)
|
|
autocora_config.results_dir = str(results_dir)
|
|
return Config(
|
|
autocora=autocora_config,
|
|
clickup=ClickUpConfig(
|
|
api_token="test-token",
|
|
workspace_id="ws1",
|
|
space_id="sp1",
|
|
task_type_field_name="Work Category",
|
|
),
|
|
)
|
|
|
|
|
|
@pytest.fixture()
|
|
def ctx(cfg, tmp_db):
|
|
return {"config": cfg, "db": tmp_db}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helper tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestSlugify:
|
|
def test_basic(self):
|
|
assert _slugify("Hello World") == "hello-world"
|
|
|
|
def test_special_chars(self):
|
|
assert _slugify("CNC machining & milling!") == "cnc-machining-milling"
|
|
|
|
def test_multiple_spaces(self):
|
|
assert _slugify(" too many spaces ") == "too-many-spaces"
|
|
|
|
def test_truncation(self):
|
|
result = _slugify("a" * 200)
|
|
assert len(result) <= 80
|
|
|
|
|
|
class TestMakeJobId:
|
|
def test_format(self):
|
|
jid = _make_job_id("precision machining")
|
|
assert jid.startswith("job-")
|
|
assert "precision-machining" in jid
|
|
|
|
def test_uniqueness(self):
|
|
a = _make_job_id("test")
|
|
b = _make_job_id("test")
|
|
# Millisecond timestamp — may be equal if very fast, but format is correct
|
|
assert a.startswith("job-")
|
|
assert b.startswith("job-")
|
|
|
|
|
|
class TestParseResult:
|
|
def test_json_success(self):
|
|
raw = json.dumps({"status": "SUCCESS", "task_ids": ["abc"]})
|
|
result = _parse_result(raw)
|
|
assert result["status"] == "SUCCESS"
|
|
assert result["task_ids"] == ["abc"]
|
|
|
|
def test_json_failure(self):
|
|
raw = json.dumps({"status": "FAILURE", "reason": "Cora not running", "task_ids": ["x"]})
|
|
result = _parse_result(raw)
|
|
assert result["status"] == "FAILURE"
|
|
assert result["reason"] == "Cora not running"
|
|
|
|
def test_legacy_success(self):
|
|
result = _parse_result("SUCCESS")
|
|
assert result["status"] == "SUCCESS"
|
|
|
|
def test_legacy_failure(self):
|
|
result = _parse_result("FAILURE: timeout exceeded")
|
|
assert result["status"] == "FAILURE"
|
|
assert result["reason"] == "timeout exceeded"
|
|
|
|
def test_unknown(self):
|
|
result = _parse_result("some garbage")
|
|
assert result["status"] == "UNKNOWN"
|
|
|
|
|
|
class TestGroupByKeyword:
|
|
def test_basic_grouping(self):
|
|
tasks = [
|
|
FakeTask(id="t1", name="Task 1", custom_fields={"Keyword": "CNC", "IMSURL": "http://a.com"}),
|
|
FakeTask(id="t2", name="Task 2", custom_fields={"Keyword": "cnc", "IMSURL": "http://a.com"}),
|
|
]
|
|
groups, alerts = _group_by_keyword(tasks, tasks)
|
|
assert len(groups) == 1
|
|
assert "cnc" in groups
|
|
assert set(groups["cnc"]["task_ids"]) == {"t1", "t2"}
|
|
assert alerts == []
|
|
|
|
def test_missing_keyword(self):
|
|
tasks = [FakeTask(id="t1", name="No KW", custom_fields={"IMSURL": "http://a.com"})]
|
|
groups, alerts = _group_by_keyword(tasks, tasks)
|
|
assert len(groups) == 0
|
|
assert any("missing Keyword" in a for a in alerts)
|
|
|
|
def test_missing_imsurl(self):
|
|
tasks = [FakeTask(id="t1", name="No URL", custom_fields={"Keyword": "test"})]
|
|
groups, alerts = _group_by_keyword(tasks, tasks)
|
|
assert len(groups) == 0
|
|
assert any("missing IMSURL" in a for a in alerts)
|
|
|
|
def test_sibling_tasks(self):
|
|
"""Tasks sharing a keyword from all_tasks should be included."""
|
|
due_tasks = [
|
|
FakeTask(id="t1", name="Due Task", custom_fields={"Keyword": "CNC", "IMSURL": "http://a.com"}),
|
|
]
|
|
all_tasks = [
|
|
FakeTask(id="t1", name="Due Task", custom_fields={"Keyword": "CNC", "IMSURL": "http://a.com"}),
|
|
FakeTask(id="t2", name="Sibling", custom_fields={"Keyword": "cnc", "IMSURL": "http://a.com"}),
|
|
FakeTask(id="t3", name="Other KW", custom_fields={"Keyword": "welding", "IMSURL": "http://b.com"}),
|
|
]
|
|
groups, alerts = _group_by_keyword(due_tasks, all_tasks)
|
|
assert set(groups["cnc"]["task_ids"]) == {"t1", "t2"}
|
|
assert "welding" not in groups
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Submit tool tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestSubmitAutocoraJobs:
|
|
def test_disabled(self, ctx):
|
|
ctx["config"].autocora.enabled = False
|
|
result = submit_autocora_jobs(ctx=ctx)
|
|
assert "disabled" in result.lower()
|
|
|
|
def test_no_context(self):
|
|
result = submit_autocora_jobs()
|
|
assert "Error" in result
|
|
|
|
def test_no_qualifying_tasks(self, ctx, monkeypatch):
|
|
monkeypatch.setattr(
|
|
"cheddahbot.tools.autocora._find_qualifying_tasks", lambda *a, **kw: []
|
|
)
|
|
result = submit_autocora_jobs(target_date="2025-01-01", ctx=ctx)
|
|
assert "No qualifying tasks" in result
|
|
|
|
def test_submit_writes_job_file(self, ctx, monkeypatch, tmp_path):
|
|
"""Valid tasks produce a job JSON file on disk."""
|
|
task = FakeTask(
|
|
id="t1",
|
|
name="CNC Page",
|
|
due_date="1700000000000",
|
|
custom_fields={"Keyword": "CNC Machining", "IMSURL": "http://example.com"},
|
|
)
|
|
monkeypatch.setattr(
|
|
"cheddahbot.tools.autocora._find_qualifying_tasks", lambda *a, **kw: [task]
|
|
)
|
|
monkeypatch.setattr(
|
|
"cheddahbot.tools.autocora._find_all_todo_tasks", lambda *a, **kw: [task]
|
|
)
|
|
|
|
result = submit_autocora_jobs(target_date="2025-01-01", ctx=ctx)
|
|
assert "Submitted 1 job" in result
|
|
|
|
# Check job file exists
|
|
jobs_dir = Path(ctx["config"].autocora.jobs_dir)
|
|
job_files = list(jobs_dir.glob("job-*.json"))
|
|
assert len(job_files) == 1
|
|
|
|
# Verify contents
|
|
job_data = json.loads(job_files[0].read_text())
|
|
assert job_data["keyword"] == "CNC Machining"
|
|
assert job_data["url"] == "http://example.com"
|
|
assert job_data["task_ids"] == ["t1"]
|
|
|
|
def test_submit_tracks_kv(self, ctx, monkeypatch):
|
|
"""KV store tracks submitted jobs."""
|
|
task = FakeTask(
|
|
id="t1",
|
|
name="Test",
|
|
due_date="1700000000000",
|
|
custom_fields={"Keyword": "test keyword", "IMSURL": "http://example.com"},
|
|
)
|
|
monkeypatch.setattr(
|
|
"cheddahbot.tools.autocora._find_qualifying_tasks", lambda *a, **kw: [task]
|
|
)
|
|
monkeypatch.setattr(
|
|
"cheddahbot.tools.autocora._find_all_todo_tasks", lambda *a, **kw: [task]
|
|
)
|
|
|
|
submit_autocora_jobs(target_date="2025-01-01", ctx=ctx)
|
|
|
|
raw = ctx["db"].kv_get("autocora:job:test keyword")
|
|
assert raw is not None
|
|
state = json.loads(raw)
|
|
assert state["status"] == "submitted"
|
|
assert "t1" in state["task_ids"]
|
|
|
|
def test_duplicate_prevention(self, ctx, monkeypatch):
|
|
"""Already-submitted keywords are skipped."""
|
|
task = FakeTask(
|
|
id="t1",
|
|
name="Test",
|
|
due_date="1700000000000",
|
|
custom_fields={"Keyword": "test", "IMSURL": "http://example.com"},
|
|
)
|
|
monkeypatch.setattr(
|
|
"cheddahbot.tools.autocora._find_qualifying_tasks", lambda *a, **kw: [task]
|
|
)
|
|
monkeypatch.setattr(
|
|
"cheddahbot.tools.autocora._find_all_todo_tasks", lambda *a, **kw: [task]
|
|
)
|
|
|
|
# First submit
|
|
submit_autocora_jobs(target_date="2025-01-01", ctx=ctx)
|
|
|
|
# Second submit — should skip
|
|
result = submit_autocora_jobs(target_date="2025-01-01", ctx=ctx)
|
|
assert "Skipped 1" in result
|
|
|
|
def test_missing_keyword_alert(self, ctx, monkeypatch):
|
|
"""Tasks without Keyword field produce alerts."""
|
|
task = FakeTask(
|
|
id="t1",
|
|
name="No KW Task",
|
|
due_date="1700000000000",
|
|
custom_fields={"IMSURL": "http://example.com"},
|
|
)
|
|
monkeypatch.setattr(
|
|
"cheddahbot.tools.autocora._find_qualifying_tasks", lambda *a, **kw: [task]
|
|
)
|
|
monkeypatch.setattr(
|
|
"cheddahbot.tools.autocora._find_all_todo_tasks", lambda *a, **kw: [task]
|
|
)
|
|
|
|
result = submit_autocora_jobs(target_date="2025-01-01", ctx=ctx)
|
|
assert "missing Keyword" in result
|
|
|
|
def test_missing_imsurl_alert(self, ctx, monkeypatch):
|
|
"""Tasks without IMSURL field produce alerts."""
|
|
task = FakeTask(
|
|
id="t1",
|
|
name="No URL Task",
|
|
due_date="1700000000000",
|
|
custom_fields={"Keyword": "test"},
|
|
)
|
|
monkeypatch.setattr(
|
|
"cheddahbot.tools.autocora._find_qualifying_tasks", lambda *a, **kw: [task]
|
|
)
|
|
monkeypatch.setattr(
|
|
"cheddahbot.tools.autocora._find_all_todo_tasks", lambda *a, **kw: [task]
|
|
)
|
|
|
|
result = submit_autocora_jobs(target_date="2025-01-01", ctx=ctx)
|
|
assert "missing IMSURL" in result
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Poll tool tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestPollAutocoraResults:
|
|
def test_disabled(self, ctx):
|
|
ctx["config"].autocora.enabled = False
|
|
result = poll_autocora_results(ctx=ctx)
|
|
assert "disabled" in result.lower()
|
|
|
|
def test_no_pending(self, ctx):
|
|
result = poll_autocora_results(ctx=ctx)
|
|
assert "No pending" in result
|
|
|
|
def test_success_json(self, ctx, monkeypatch):
|
|
"""JSON SUCCESS result updates KV and ClickUp."""
|
|
db = ctx["db"]
|
|
results_dir = Path(ctx["config"].autocora.results_dir)
|
|
|
|
# Set up submitted job in KV
|
|
job_id = "job-123-test"
|
|
kv_key = "autocora:job:test keyword"
|
|
db.kv_set(
|
|
kv_key,
|
|
json.dumps({
|
|
"status": "submitted",
|
|
"job_id": job_id,
|
|
"keyword": "test keyword",
|
|
"task_ids": ["t1", "t2"],
|
|
}),
|
|
)
|
|
|
|
# Write result file
|
|
result_data = {"status": "SUCCESS", "task_ids": ["t1", "t2"]}
|
|
(results_dir / f"{job_id}.result").write_text(json.dumps(result_data))
|
|
|
|
# Mock ClickUp client
|
|
mock_client = MagicMock()
|
|
monkeypatch.setattr(
|
|
"cheddahbot.tools.autocora._get_clickup_client", lambda ctx: mock_client
|
|
)
|
|
|
|
result = poll_autocora_results(ctx=ctx)
|
|
assert "SUCCESS: test keyword" in result
|
|
|
|
# Verify KV updated
|
|
state = json.loads(db.kv_get(kv_key))
|
|
assert state["status"] == "completed"
|
|
|
|
# Verify ClickUp calls
|
|
assert mock_client.update_task_status.call_count == 2
|
|
mock_client.update_task_status.assert_any_call("t1", "running cora")
|
|
mock_client.update_task_status.assert_any_call("t2", "running cora")
|
|
assert mock_client.add_comment.call_count == 2
|
|
|
|
def test_failure_json(self, ctx, monkeypatch):
|
|
"""JSON FAILURE result updates KV and ClickUp with error."""
|
|
db = ctx["db"]
|
|
results_dir = Path(ctx["config"].autocora.results_dir)
|
|
|
|
job_id = "job-456-fail"
|
|
kv_key = "autocora:job:fail keyword"
|
|
db.kv_set(
|
|
kv_key,
|
|
json.dumps({
|
|
"status": "submitted",
|
|
"job_id": job_id,
|
|
"keyword": "fail keyword",
|
|
"task_ids": ["t3"],
|
|
}),
|
|
)
|
|
|
|
result_data = {
|
|
"status": "FAILURE",
|
|
"reason": "Cora not running",
|
|
"task_ids": ["t3"],
|
|
}
|
|
(results_dir / f"{job_id}.result").write_text(json.dumps(result_data))
|
|
|
|
mock_client = MagicMock()
|
|
monkeypatch.setattr(
|
|
"cheddahbot.tools.autocora._get_clickup_client", lambda ctx: mock_client
|
|
)
|
|
|
|
result = poll_autocora_results(ctx=ctx)
|
|
assert "FAILURE: fail keyword" in result
|
|
assert "Cora not running" in result
|
|
|
|
state = json.loads(db.kv_get(kv_key))
|
|
assert state["status"] == "failed"
|
|
assert state["error"] == "Cora not running"
|
|
|
|
mock_client.update_task_status.assert_called_once_with("t3", "error")
|
|
|
|
def test_legacy_plain_text(self, ctx, monkeypatch):
|
|
"""Legacy plain-text SUCCESS result still works."""
|
|
db = ctx["db"]
|
|
results_dir = Path(ctx["config"].autocora.results_dir)
|
|
|
|
job_id = "job-789-legacy"
|
|
kv_key = "autocora:job:legacy kw"
|
|
db.kv_set(
|
|
kv_key,
|
|
json.dumps({
|
|
"status": "submitted",
|
|
"job_id": job_id,
|
|
"keyword": "legacy kw",
|
|
"task_ids": ["t5"],
|
|
}),
|
|
)
|
|
|
|
# Legacy format — plain text, no JSON
|
|
(results_dir / f"{job_id}.result").write_text("SUCCESS")
|
|
|
|
mock_client = MagicMock()
|
|
monkeypatch.setattr(
|
|
"cheddahbot.tools.autocora._get_clickup_client", lambda ctx: mock_client
|
|
)
|
|
|
|
result = poll_autocora_results(ctx=ctx)
|
|
assert "SUCCESS: legacy kw" in result
|
|
|
|
# task_ids come from KV fallback
|
|
mock_client.update_task_status.assert_called_once_with("t5", "running cora")
|
|
|
|
def test_task_ids_from_result_preferred(self, ctx, monkeypatch):
|
|
"""task_ids from result file take precedence over KV."""
|
|
db = ctx["db"]
|
|
results_dir = Path(ctx["config"].autocora.results_dir)
|
|
|
|
job_id = "job-100-pref"
|
|
kv_key = "autocora:job:pref kw"
|
|
db.kv_set(
|
|
kv_key,
|
|
json.dumps({
|
|
"status": "submitted",
|
|
"job_id": job_id,
|
|
"keyword": "pref kw",
|
|
"task_ids": ["old_t1"], # KV has old IDs
|
|
}),
|
|
)
|
|
|
|
# Result has updated task_ids
|
|
result_data = {"status": "SUCCESS", "task_ids": ["new_t1", "new_t2"]}
|
|
(results_dir / f"{job_id}.result").write_text(json.dumps(result_data))
|
|
|
|
mock_client = MagicMock()
|
|
monkeypatch.setattr(
|
|
"cheddahbot.tools.autocora._get_clickup_client", lambda ctx: mock_client
|
|
)
|
|
|
|
poll_autocora_results(ctx=ctx)
|
|
|
|
# Should use result file task_ids, not KV
|
|
calls = [c.args for c in mock_client.update_task_status.call_args_list]
|
|
assert ("new_t1", "running cora") in calls
|
|
assert ("new_t2", "running cora") in calls
|
|
assert ("old_t1", "running cora") not in calls
|
|
|
|
def test_still_pending(self, ctx):
|
|
"""Jobs without result files show as still pending."""
|
|
db = ctx["db"]
|
|
db.kv_set(
|
|
"autocora:job:waiting",
|
|
json.dumps({
|
|
"status": "submitted",
|
|
"job_id": "job-999-wait",
|
|
"keyword": "waiting",
|
|
"task_ids": ["t99"],
|
|
}),
|
|
)
|
|
|
|
result = poll_autocora_results(ctx=ctx)
|
|
assert "Still pending" in result
|
|
assert "waiting" in result
|