This reverts commit 15937a6b46.
This commit is contained in:
Teknium
@@ -1,210 +0,0 @@
|
||||
"""Tests for the kanban CLI surface (hermes_cli.kanban)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from hermes_cli import kanban as kc
|
||||
from hermes_cli import kanban_db as kb
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def kanban_home(tmp_path, monkeypatch):
|
||||
home = tmp_path / ".hermes"
|
||||
home.mkdir()
|
||||
monkeypatch.setenv("HERMES_HOME", str(home))
|
||||
monkeypatch.setattr(Path, "home", lambda: tmp_path)
|
||||
kb.init_db()
|
||||
return home
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Workspace flag parsing
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"value,expected",
|
||||
[
|
||||
("scratch", ("scratch", None)),
|
||||
("worktree", ("worktree", None)),
|
||||
("dir:/tmp/work", ("dir", "/tmp/work")),
|
||||
],
|
||||
)
|
||||
def test_parse_workspace_flag_valid(value, expected):
|
||||
assert kc._parse_workspace_flag(value) == expected
|
||||
|
||||
|
||||
def test_parse_workspace_flag_expands_user():
|
||||
kind, path = kc._parse_workspace_flag("dir:~/vault")
|
||||
assert kind == "dir"
|
||||
assert path.endswith("/vault")
|
||||
assert not path.startswith("~")
|
||||
|
||||
|
||||
@pytest.mark.parametrize("bad", ["cloud", "dir:", "", "worktree:/x"])
|
||||
def test_parse_workspace_flag_rejects(bad):
|
||||
if not bad:
|
||||
# Empty -> defaults; not an error.
|
||||
assert kc._parse_workspace_flag(bad) == ("scratch", None)
|
||||
return
|
||||
with pytest.raises(argparse.ArgumentTypeError):
|
||||
kc._parse_workspace_flag(bad)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# run_slash smoke tests (end-to-end via the same entry both CLI and gateway use)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_run_slash_no_args_shows_usage(kanban_home):
|
||||
out = kc.run_slash("")
|
||||
assert "kanban" in out.lower()
|
||||
assert "create" in out.lower() or "subcommand" in out.lower() or "action" in out.lower()
|
||||
|
||||
|
||||
def test_run_slash_create_and_list(kanban_home):
|
||||
out = kc.run_slash("create 'ship feature' --assignee alice")
|
||||
assert "Created" in out
|
||||
out = kc.run_slash("list")
|
||||
assert "ship feature" in out
|
||||
assert "alice" in out
|
||||
|
||||
|
||||
def test_run_slash_create_with_parent_and_cascade(kanban_home):
|
||||
# Parent then child via --parent
|
||||
out1 = kc.run_slash("create 'parent' --assignee alice")
|
||||
# Extract the "t_xxxx" id from "Created t_xxxx (ready, ...)"
|
||||
import re
|
||||
m = re.search(r"(t_[a-f0-9]+)", out1)
|
||||
assert m
|
||||
p = m.group(1)
|
||||
out2 = kc.run_slash(f"create 'child' --assignee bob --parent {p}")
|
||||
assert "todo" in out2 # child starts as todo
|
||||
|
||||
# Complete parent; list should promote child to ready
|
||||
kc.run_slash(f"complete {p}")
|
||||
# Explicit filter: child should now be ready (was todo before complete).
|
||||
ready_list = kc.run_slash("list --status ready")
|
||||
assert "child" in ready_list
|
||||
|
||||
|
||||
def test_run_slash_show_includes_comments(kanban_home):
|
||||
out = kc.run_slash("create 'x'")
|
||||
import re
|
||||
tid = re.search(r"(t_[a-f0-9]+)", out).group(1)
|
||||
kc.run_slash(f"comment {tid} 'source is paywalled'")
|
||||
show = kc.run_slash(f"show {tid}")
|
||||
assert "source is paywalled" in show
|
||||
|
||||
|
||||
def test_run_slash_block_unblock_cycle(kanban_home):
|
||||
out = kc.run_slash("create 'x' --assignee alice")
|
||||
import re
|
||||
tid = re.search(r"(t_[a-f0-9]+)", out).group(1)
|
||||
# Claim first so block() finds it running
|
||||
kc.run_slash(f"claim {tid}")
|
||||
assert "Blocked" in kc.run_slash(f"block {tid} 'need decision'")
|
||||
assert "Unblocked" in kc.run_slash(f"unblock {tid}")
|
||||
|
||||
|
||||
def test_run_slash_json_output(kanban_home):
|
||||
out = kc.run_slash("create 'jsontask' --assignee alice --json")
|
||||
payload = json.loads(out)
|
||||
assert payload["title"] == "jsontask"
|
||||
assert payload["assignee"] == "alice"
|
||||
assert payload["status"] == "ready"
|
||||
|
||||
|
||||
def test_run_slash_dispatch_dry_run_counts(kanban_home):
|
||||
kc.run_slash("create 'a' --assignee alice")
|
||||
kc.run_slash("create 'b' --assignee bob")
|
||||
out = kc.run_slash("dispatch --dry-run")
|
||||
assert "Spawned:" in out
|
||||
|
||||
|
||||
def test_run_slash_context_output_format(kanban_home):
|
||||
out = kc.run_slash("create 'tech spec' --assignee alice --body 'write an RFC'")
|
||||
import re
|
||||
tid = re.search(r"(t_[a-f0-9]+)", out).group(1)
|
||||
kc.run_slash(f"comment {tid} 'remember to include performance section'")
|
||||
ctx = kc.run_slash(f"context {tid}")
|
||||
assert "tech spec" in ctx
|
||||
assert "write an RFC" in ctx
|
||||
assert "performance section" in ctx
|
||||
|
||||
|
||||
def test_run_slash_tenant_filter(kanban_home):
|
||||
kc.run_slash("create 'biz-a task' --tenant biz-a --assignee alice")
|
||||
kc.run_slash("create 'biz-b task' --tenant biz-b --assignee alice")
|
||||
a = kc.run_slash("list --tenant biz-a")
|
||||
b = kc.run_slash("list --tenant biz-b")
|
||||
assert "biz-a task" in a and "biz-b task" not in a
|
||||
assert "biz-b task" in b and "biz-a task" not in b
|
||||
|
||||
|
||||
def test_run_slash_usage_error_returns_message(kanban_home):
|
||||
# Missing required argument for create
|
||||
out = kc.run_slash("create")
|
||||
assert "usage" in out.lower() or "error" in out.lower()
|
||||
|
||||
|
||||
def test_run_slash_assign_reassigns(kanban_home):
|
||||
out = kc.run_slash("create 'x' --assignee alice")
|
||||
import re
|
||||
tid = re.search(r"(t_[a-f0-9]+)", out).group(1)
|
||||
assert "Assigned" in kc.run_slash(f"assign {tid} bob")
|
||||
show = kc.run_slash(f"show {tid}")
|
||||
assert "bob" in show
|
||||
|
||||
|
||||
def test_run_slash_link_unlink(kanban_home):
|
||||
a = kc.run_slash("create 'a'")
|
||||
b = kc.run_slash("create 'b'")
|
||||
import re
|
||||
ta = re.search(r"(t_[a-f0-9]+)", a).group(1)
|
||||
tb = re.search(r"(t_[a-f0-9]+)", b).group(1)
|
||||
assert "Linked" in kc.run_slash(f"link {ta} {tb}")
|
||||
# After link, b is todo
|
||||
show = kc.run_slash(f"show {tb}")
|
||||
assert "todo" in show
|
||||
assert "Unlinked" in kc.run_slash(f"unlink {ta} {tb}")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Integration with the COMMAND_REGISTRY
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_kanban_is_resolvable():
|
||||
from hermes_cli.commands import resolve_command
|
||||
|
||||
cmd = resolve_command("kanban")
|
||||
assert cmd is not None
|
||||
assert cmd.name == "kanban"
|
||||
|
||||
|
||||
def test_kanban_bypasses_active_session_guard():
|
||||
from hermes_cli.commands import should_bypass_active_session
|
||||
|
||||
assert should_bypass_active_session("kanban")
|
||||
|
||||
|
||||
def test_kanban_in_autocomplete_table():
|
||||
from hermes_cli.commands import COMMANDS, SUBCOMMANDS
|
||||
|
||||
assert "/kanban" in COMMANDS
|
||||
subs = SUBCOMMANDS.get("/kanban") or []
|
||||
assert "create" in subs
|
||||
assert "dispatch" in subs
|
||||
|
||||
|
||||
def test_kanban_not_gateway_only():
|
||||
# kanban is available in BOTH CLI and gateway surfaces.
|
||||
from hermes_cli.commands import COMMAND_REGISTRY
|
||||
|
||||
cmd = next(c for c in COMMAND_REGISTRY if c.name == "kanban")
|
||||
assert not cmd.cli_only
|
||||
assert not cmd.gateway_only
|
||||
@@ -1,438 +0,0 @@
|
||||
"""Tests for the Kanban DB layer (hermes_cli.kanban_db)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import concurrent.futures
|
||||
import os
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from hermes_cli import kanban_db as kb
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def kanban_home(tmp_path, monkeypatch):
|
||||
"""Isolated HERMES_HOME with an empty kanban DB."""
|
||||
home = tmp_path / ".hermes"
|
||||
home.mkdir()
|
||||
monkeypatch.setenv("HERMES_HOME", str(home))
|
||||
monkeypatch.setattr(Path, "home", lambda: tmp_path)
|
||||
kb.init_db()
|
||||
return home
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Schema / init
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_init_db_is_idempotent(kanban_home):
|
||||
# Second call should not error or drop data.
|
||||
with kb.connect() as conn:
|
||||
kb.create_task(conn, title="persisted")
|
||||
kb.init_db()
|
||||
with kb.connect() as conn:
|
||||
tasks = kb.list_tasks(conn)
|
||||
assert len(tasks) == 1
|
||||
assert tasks[0].title == "persisted"
|
||||
|
||||
|
||||
def test_init_creates_expected_tables(kanban_home):
|
||||
with kb.connect() as conn:
|
||||
rows = conn.execute(
|
||||
"SELECT name FROM sqlite_master WHERE type='table' ORDER BY name"
|
||||
).fetchall()
|
||||
names = {r["name"] for r in rows}
|
||||
assert {"tasks", "task_links", "task_comments", "task_events"} <= names
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Task creation + status inference
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_create_task_no_parents_is_ready(kanban_home):
|
||||
with kb.connect() as conn:
|
||||
tid = kb.create_task(conn, title="ship it", assignee="alice")
|
||||
t = kb.get_task(conn, tid)
|
||||
assert t is not None
|
||||
assert t.status == "ready"
|
||||
assert t.assignee == "alice"
|
||||
assert t.workspace_kind == "scratch"
|
||||
|
||||
|
||||
def test_create_task_with_parent_is_todo_until_parent_done(kanban_home):
|
||||
with kb.connect() as conn:
|
||||
p = kb.create_task(conn, title="parent")
|
||||
c = kb.create_task(conn, title="child", parents=[p])
|
||||
assert kb.get_task(conn, c).status == "todo"
|
||||
kb.complete_task(conn, p, result="ok")
|
||||
assert kb.get_task(conn, c).status == "ready"
|
||||
|
||||
|
||||
def test_create_task_unknown_parent_errors(kanban_home):
|
||||
with kb.connect() as conn, pytest.raises(ValueError, match="unknown parent"):
|
||||
kb.create_task(conn, title="orphan", parents=["t_ghost"])
|
||||
|
||||
|
||||
def test_workspace_kind_validation(kanban_home):
|
||||
with kb.connect() as conn, pytest.raises(ValueError, match="workspace_kind"):
|
||||
kb.create_task(conn, title="bad ws", workspace_kind="cloud")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Links + dependency resolution
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_link_demotes_ready_child_to_todo_when_parent_not_done(kanban_home):
|
||||
with kb.connect() as conn:
|
||||
a = kb.create_task(conn, title="a")
|
||||
b = kb.create_task(conn, title="b")
|
||||
assert kb.get_task(conn, b).status == "ready"
|
||||
kb.link_tasks(conn, a, b)
|
||||
assert kb.get_task(conn, b).status == "todo"
|
||||
|
||||
|
||||
def test_link_keeps_ready_child_when_parent_already_done(kanban_home):
|
||||
with kb.connect() as conn:
|
||||
a = kb.create_task(conn, title="a")
|
||||
kb.complete_task(conn, a)
|
||||
b = kb.create_task(conn, title="b")
|
||||
assert kb.get_task(conn, b).status == "ready"
|
||||
kb.link_tasks(conn, a, b)
|
||||
assert kb.get_task(conn, b).status == "ready"
|
||||
|
||||
|
||||
def test_link_rejects_self_loop(kanban_home):
|
||||
with kb.connect() as conn:
|
||||
a = kb.create_task(conn, title="a")
|
||||
with pytest.raises(ValueError, match="itself"):
|
||||
kb.link_tasks(conn, a, a)
|
||||
|
||||
|
||||
def test_link_detects_cycle(kanban_home):
|
||||
with kb.connect() as conn:
|
||||
a = kb.create_task(conn, title="a")
|
||||
b = kb.create_task(conn, title="b", parents=[a])
|
||||
c = kb.create_task(conn, title="c", parents=[b])
|
||||
with pytest.raises(ValueError, match="cycle"):
|
||||
kb.link_tasks(conn, c, a)
|
||||
with pytest.raises(ValueError, match="cycle"):
|
||||
kb.link_tasks(conn, b, a)
|
||||
|
||||
|
||||
def test_recompute_ready_cascades_through_chain(kanban_home):
|
||||
with kb.connect() as conn:
|
||||
a = kb.create_task(conn, title="a")
|
||||
b = kb.create_task(conn, title="b", parents=[a])
|
||||
c = kb.create_task(conn, title="c", parents=[b])
|
||||
assert [kb.get_task(conn, x).status for x in (a, b, c)] == \
|
||||
["ready", "todo", "todo"]
|
||||
kb.complete_task(conn, a)
|
||||
assert kb.get_task(conn, b).status == "ready"
|
||||
kb.complete_task(conn, b)
|
||||
assert kb.get_task(conn, c).status == "ready"
|
||||
|
||||
|
||||
def test_recompute_ready_fan_in_waits_for_all_parents(kanban_home):
|
||||
with kb.connect() as conn:
|
||||
a = kb.create_task(conn, title="a")
|
||||
b = kb.create_task(conn, title="b")
|
||||
c = kb.create_task(conn, title="c", parents=[a, b])
|
||||
kb.complete_task(conn, a)
|
||||
assert kb.get_task(conn, c).status == "todo"
|
||||
kb.complete_task(conn, b)
|
||||
assert kb.get_task(conn, c).status == "ready"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Atomic claim (CAS)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_claim_once_wins_second_loses(kanban_home):
|
||||
with kb.connect() as conn:
|
||||
t = kb.create_task(conn, title="x", assignee="a")
|
||||
first = kb.claim_task(conn, t, claimer="host:1")
|
||||
assert first is not None and first.status == "running"
|
||||
second = kb.claim_task(conn, t, claimer="host:2")
|
||||
assert second is None
|
||||
|
||||
|
||||
def test_claim_fails_on_non_ready(kanban_home):
|
||||
with kb.connect() as conn:
|
||||
t = kb.create_task(conn, title="x")
|
||||
# Move to todo by introducing an unsatisfied parent.
|
||||
p = kb.create_task(conn, title="p")
|
||||
kb.link_tasks(conn, p, t)
|
||||
assert kb.get_task(conn, t).status == "todo"
|
||||
assert kb.claim_task(conn, t) is None
|
||||
|
||||
|
||||
def test_stale_claim_reclaimed(kanban_home):
|
||||
with kb.connect() as conn:
|
||||
t = kb.create_task(conn, title="x", assignee="a")
|
||||
kb.claim_task(conn, t)
|
||||
# Rewind claim_expires so it looks stale.
|
||||
conn.execute(
|
||||
"UPDATE tasks SET claim_expires = ? WHERE id = ?",
|
||||
(int(time.time()) - 3600, t),
|
||||
)
|
||||
reclaimed = kb.release_stale_claims(conn)
|
||||
assert reclaimed == 1
|
||||
assert kb.get_task(conn, t).status == "ready"
|
||||
|
||||
|
||||
def test_heartbeat_extends_claim(kanban_home):
|
||||
with kb.connect() as conn:
|
||||
t = kb.create_task(conn, title="x", assignee="a")
|
||||
claimer = "host:hb"
|
||||
kb.claim_task(conn, t, claimer=claimer, ttl_seconds=60)
|
||||
original = kb.get_task(conn, t).claim_expires
|
||||
# Rewind then heartbeat.
|
||||
conn.execute("UPDATE tasks SET claim_expires = ? WHERE id = ?", (0, t))
|
||||
ok = kb.heartbeat_claim(conn, t, claimer=claimer, ttl_seconds=3600)
|
||||
assert ok
|
||||
new = kb.get_task(conn, t).claim_expires
|
||||
assert new > int(time.time()) + 3000
|
||||
|
||||
|
||||
def test_concurrent_claims_only_one_wins(kanban_home):
|
||||
"""Fire N threads claiming the same task; exactly one must win."""
|
||||
with kb.connect() as conn:
|
||||
t = kb.create_task(conn, title="race", assignee="a")
|
||||
|
||||
def attempt(i):
|
||||
with kb.connect() as c:
|
||||
return kb.claim_task(c, t, claimer=f"host:{i}")
|
||||
|
||||
n_workers = 8
|
||||
with concurrent.futures.ThreadPoolExecutor(max_workers=n_workers) as ex:
|
||||
results = list(ex.map(attempt, range(n_workers)))
|
||||
winners = [r for r in results if r is not None]
|
||||
assert len(winners) == 1
|
||||
assert winners[0].status == "running"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Complete / block / unblock / archive / assign
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_complete_records_result(kanban_home):
|
||||
with kb.connect() as conn:
|
||||
t = kb.create_task(conn, title="x")
|
||||
assert kb.complete_task(conn, t, result="done and dusted")
|
||||
task = kb.get_task(conn, t)
|
||||
assert task.status == "done"
|
||||
assert task.result == "done and dusted"
|
||||
assert task.completed_at is not None
|
||||
|
||||
|
||||
def test_block_then_unblock(kanban_home):
|
||||
with kb.connect() as conn:
|
||||
t = kb.create_task(conn, title="x", assignee="a")
|
||||
kb.claim_task(conn, t)
|
||||
assert kb.block_task(conn, t, reason="need input")
|
||||
assert kb.get_task(conn, t).status == "blocked"
|
||||
assert kb.unblock_task(conn, t)
|
||||
assert kb.get_task(conn, t).status == "ready"
|
||||
|
||||
|
||||
def test_assign_refuses_while_running(kanban_home):
|
||||
with kb.connect() as conn:
|
||||
t = kb.create_task(conn, title="x", assignee="a")
|
||||
kb.claim_task(conn, t)
|
||||
with pytest.raises(RuntimeError, match="currently running"):
|
||||
kb.assign_task(conn, t, "b")
|
||||
|
||||
|
||||
def test_assign_reassigns_when_not_running(kanban_home):
|
||||
with kb.connect() as conn:
|
||||
t = kb.create_task(conn, title="x", assignee="a")
|
||||
assert kb.assign_task(conn, t, "b")
|
||||
assert kb.get_task(conn, t).assignee == "b"
|
||||
|
||||
|
||||
def test_archive_hides_from_default_list(kanban_home):
|
||||
with kb.connect() as conn:
|
||||
t = kb.create_task(conn, title="x")
|
||||
kb.complete_task(conn, t)
|
||||
assert kb.archive_task(conn, t)
|
||||
assert len(kb.list_tasks(conn)) == 0
|
||||
assert len(kb.list_tasks(conn, include_archived=True)) == 1
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Comments / events / worker context
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_comments_recorded_in_order(kanban_home):
|
||||
with kb.connect() as conn:
|
||||
t = kb.create_task(conn, title="x")
|
||||
kb.add_comment(conn, t, "user", "first")
|
||||
kb.add_comment(conn, t, "researcher", "second")
|
||||
comments = kb.list_comments(conn, t)
|
||||
assert [c.body for c in comments] == ["first", "second"]
|
||||
assert [c.author for c in comments] == ["user", "researcher"]
|
||||
|
||||
|
||||
def test_empty_comment_rejected(kanban_home):
|
||||
with kb.connect() as conn:
|
||||
t = kb.create_task(conn, title="x")
|
||||
with pytest.raises(ValueError, match="body is required"):
|
||||
kb.add_comment(conn, t, "user", "")
|
||||
|
||||
|
||||
def test_events_capture_lifecycle(kanban_home):
|
||||
with kb.connect() as conn:
|
||||
t = kb.create_task(conn, title="x", assignee="a")
|
||||
kb.claim_task(conn, t)
|
||||
kb.complete_task(conn, t, result="ok")
|
||||
events = kb.list_events(conn, t)
|
||||
kinds = [e.kind for e in events]
|
||||
assert "created" in kinds
|
||||
assert "claimed" in kinds
|
||||
assert "completed" in kinds
|
||||
|
||||
|
||||
def test_worker_context_includes_parent_results_and_comments(kanban_home):
|
||||
with kb.connect() as conn:
|
||||
p = kb.create_task(conn, title="p")
|
||||
kb.complete_task(conn, p, result="PARENT_RESULT_MARKER")
|
||||
c = kb.create_task(conn, title="child", parents=[p])
|
||||
kb.add_comment(conn, c, "user", "CLARIFICATION_MARKER")
|
||||
ctx = kb.build_worker_context(conn, c)
|
||||
assert "PARENT_RESULT_MARKER" in ctx
|
||||
assert "CLARIFICATION_MARKER" in ctx
|
||||
assert c in ctx
|
||||
assert "child" in ctx
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Dispatcher
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_dispatch_dry_run_does_not_claim(kanban_home):
|
||||
with kb.connect() as conn:
|
||||
t1 = kb.create_task(conn, title="a", assignee="alice")
|
||||
t2 = kb.create_task(conn, title="b", assignee="bob")
|
||||
res = kb.dispatch_once(conn, dry_run=True)
|
||||
assert {s[0] for s in res.spawned} == {t1, t2}
|
||||
with kb.connect() as conn:
|
||||
# Dry run must NOT mutate status.
|
||||
assert kb.get_task(conn, t1).status == "ready"
|
||||
assert kb.get_task(conn, t2).status == "ready"
|
||||
|
||||
|
||||
def test_dispatch_skips_unassigned(kanban_home):
|
||||
with kb.connect() as conn:
|
||||
t = kb.create_task(conn, title="floater")
|
||||
res = kb.dispatch_once(conn, dry_run=True)
|
||||
assert t in res.skipped_unassigned
|
||||
assert not res.spawned
|
||||
|
||||
|
||||
def test_dispatch_promotes_ready_and_spawns(kanban_home):
|
||||
spawns = []
|
||||
|
||||
def fake_spawn(task, workspace):
|
||||
spawns.append((task.id, task.assignee, workspace))
|
||||
|
||||
with kb.connect() as conn:
|
||||
p = kb.create_task(conn, title="p", assignee="alice")
|
||||
c = kb.create_task(conn, title="c", assignee="bob", parents=[p])
|
||||
# Finish parent outside dispatch; promotion happens inside.
|
||||
kb.complete_task(conn, p)
|
||||
res = kb.dispatch_once(conn, spawn_fn=fake_spawn)
|
||||
# Spawned c (a was already done when dispatch was called).
|
||||
assert len(spawns) == 1
|
||||
assert spawns[0][0] == c
|
||||
assert spawns[0][1] == "bob"
|
||||
# c is now running
|
||||
with kb.connect() as conn:
|
||||
assert kb.get_task(conn, c).status == "running"
|
||||
|
||||
|
||||
def test_dispatch_spawn_failure_releases_claim(kanban_home):
|
||||
def boom(task, workspace):
|
||||
raise RuntimeError("spawn failed")
|
||||
|
||||
with kb.connect() as conn:
|
||||
t = kb.create_task(conn, title="boom", assignee="alice")
|
||||
kb.dispatch_once(conn, spawn_fn=boom)
|
||||
# Must return to ready so the next tick can retry.
|
||||
assert kb.get_task(conn, t).status == "ready"
|
||||
assert kb.get_task(conn, t).claim_lock is None
|
||||
|
||||
|
||||
def test_dispatch_reclaims_stale_before_spawning(kanban_home):
|
||||
with kb.connect() as conn:
|
||||
t = kb.create_task(conn, title="x", assignee="alice")
|
||||
kb.claim_task(conn, t)
|
||||
conn.execute(
|
||||
"UPDATE tasks SET claim_expires = ? WHERE id = ?",
|
||||
(int(time.time()) - 1, t),
|
||||
)
|
||||
res = kb.dispatch_once(conn, dry_run=True)
|
||||
assert res.reclaimed == 1
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Workspace resolution
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_scratch_workspace_created_under_hermes_home(kanban_home):
|
||||
with kb.connect() as conn:
|
||||
t = kb.create_task(conn, title="x")
|
||||
task = kb.get_task(conn, t)
|
||||
ws = kb.resolve_workspace(task)
|
||||
assert ws.exists()
|
||||
assert ws.is_dir()
|
||||
assert "kanban" in str(ws)
|
||||
|
||||
|
||||
def test_dir_workspace_honors_given_path(kanban_home, tmp_path):
|
||||
target = tmp_path / "my-vault"
|
||||
with kb.connect() as conn:
|
||||
t = kb.create_task(
|
||||
conn, title="biz", workspace_kind="dir", workspace_path=str(target)
|
||||
)
|
||||
task = kb.get_task(conn, t)
|
||||
ws = kb.resolve_workspace(task)
|
||||
assert ws == target
|
||||
assert ws.exists()
|
||||
|
||||
|
||||
def test_worktree_workspace_returns_intended_path(kanban_home, tmp_path):
|
||||
target = str(tmp_path / ".worktrees" / "my-task")
|
||||
with kb.connect() as conn:
|
||||
t = kb.create_task(
|
||||
conn, title="ship", workspace_kind="worktree", workspace_path=target
|
||||
)
|
||||
task = kb.get_task(conn, t)
|
||||
ws = kb.resolve_workspace(task)
|
||||
# We do NOT auto-create worktrees; the worker's skill handles that.
|
||||
assert str(ws) == target
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tenancy
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_tenant_column_filters_listings(kanban_home):
|
||||
with kb.connect() as conn:
|
||||
kb.create_task(conn, title="a1", tenant="biz-a")
|
||||
kb.create_task(conn, title="b1", tenant="biz-b")
|
||||
kb.create_task(conn, title="shared") # no tenant
|
||||
biz_a = kb.list_tasks(conn, tenant="biz-a")
|
||||
biz_b = kb.list_tasks(conn, tenant="biz-b")
|
||||
assert [t.title for t in biz_a] == ["a1"]
|
||||
assert [t.title for t in biz_b] == ["b1"]
|
||||
|
||||
|
||||
def test_tenant_propagates_to_events(kanban_home):
|
||||
with kb.connect() as conn:
|
||||
t = kb.create_task(conn, title="tenant-task", tenant="biz-a")
|
||||
events = kb.list_events(conn, t)
|
||||
# The "created" event should have tenant in its payload.
|
||||
created = [e for e in events if e.kind == "created"]
|
||||
assert created and created[0].payload.get("tenant") == "biz-a"
|
||||
Reference in New Issue
Block a user