11598aebe3
WHAT: regression test mounting the kanban plugin router exactly as `_mount_plugin_api_routes` does (behind `_plugin_route_secret_scope`) and driving Specify under `set_multiplex_active(True)` for launch profile A -> `?profile=workerb` -> A, asserting the aux call's `get_secret` sees each profile's own key. WHY: `_run_aux` / `_run_estimate` only pinned the board, never a profile scope, so the aux client's provider-key read failed closed under multi-profile hosting. The scope is now bound once at the plugin mount (previous commit) rather than per aux call inside the plugin: one mechanism for every plugin router, and the `?profile=` request convention works for the kanban routes too. Fixes #123372 (cherry picked from commit 6f01ddc9e455b3eee7046d5b1ac969df3499cbeb)
1174 lines
46 KiB
Python
1174 lines
46 KiB
Python
"""Tests for the Kanban dashboard plugin backend (plugins/kanban/dashboard/plugin_api.py).
|
|
|
|
The plugin mounts as /api/plugins/kanban/ inside the dashboard's FastAPI app,
|
|
but here we attach its router to a bare FastAPI instance so we can test the
|
|
REST surface without spinning up the whole dashboard.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import shutil
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
|
|
from hermes_cli import kanban_db as kb
|
|
from hermes_cli import kanban_db_connect as kbc
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fixtures
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _load_plugin_router():
|
|
"""Dynamically load plugins/kanban/dashboard/plugin_api.py and return its router."""
|
|
repo_root = Path(__file__).resolve().parents[2]
|
|
plugin_file = repo_root / "plugins" / "kanban" / "dashboard" / "plugin_api.py"
|
|
assert plugin_file.exists(), f"plugin file missing: {plugin_file}"
|
|
|
|
spec = importlib.util.spec_from_file_location(
|
|
"hermes_dashboard_plugin_kanban_test", plugin_file,
|
|
)
|
|
assert spec is not None and spec.loader is not None
|
|
mod = importlib.util.module_from_spec(spec)
|
|
sys.modules[spec.name] = mod
|
|
spec.loader.exec_module(mod)
|
|
return mod.router
|
|
|
|
@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
|
|
|
|
@pytest.fixture
|
|
def client(kanban_home):
|
|
app = FastAPI()
|
|
app.include_router(_load_plugin_router(), prefix="/api/plugins/kanban")
|
|
return TestClient(app)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# GET /board on an empty DB
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_board_empty(client):
|
|
r = client.get("/api/plugins/kanban/board")
|
|
assert r.status_code == 200
|
|
data = r.json()
|
|
# All canonical columns present (triage + the rest), each empty.
|
|
names = [c["name"] for c in data["columns"]]
|
|
assert set(names) == kb.VALID_STATUSES - {"archived"}
|
|
for expected in ("triage", "todo", "scheduled", "ready", "running", "blocked", "done"):
|
|
assert expected in names, f"missing column {expected}: {names}"
|
|
assert all(len(c["tasks"]) == 0 for c in data["columns"])
|
|
assert data["tenants"] == []
|
|
assert data["assignees"] == []
|
|
assert data["latest_event_id"] == 0
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# POST /tasks then GET /board sees it
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_create_task_appears_on_board(client):
|
|
r = client.post(
|
|
"/api/plugins/kanban/tasks",
|
|
json={
|
|
"title": "Research LLM caching",
|
|
"assignee": "researcher",
|
|
"priority": 3,
|
|
"tenant": "acme",
|
|
},
|
|
)
|
|
assert r.status_code == 200, r.text
|
|
task = r.json()["task"]
|
|
assert task["title"] == "Research LLM caching"
|
|
assert task["assignee"] == "researcher"
|
|
assert task["status"] == "ready" # no parents -> immediately ready
|
|
assert task["priority"] == 3
|
|
assert task["tenant"] == "acme"
|
|
task_id = task["id"]
|
|
|
|
# Board now lists it under 'ready'.
|
|
r = client.get("/api/plugins/kanban/board")
|
|
assert r.status_code == 200
|
|
data = r.json()
|
|
ready = next(c for c in data["columns"] if c["name"] == "ready")
|
|
assert len(ready["tasks"]) == 1
|
|
assert ready["tasks"][0]["id"] == task_id
|
|
assert "acme" in data["tenants"]
|
|
assert "researcher" in data["assignees"]
|
|
|
|
def test_patch_board_sets_project_directory(client, tmp_path):
|
|
"""Board-level default_workdir must be editable after creation."""
|
|
kb.create_board("late-config")
|
|
project_dir = tmp_path / "late-project"
|
|
project_dir.mkdir()
|
|
|
|
response = client.patch(
|
|
"/api/plugins/kanban/boards/late-config",
|
|
json={"default_workdir": str(project_dir)},
|
|
)
|
|
|
|
assert response.status_code == 200, response.text
|
|
board = response.json()["board"]
|
|
assert board["default_workdir"] == str(project_dir.resolve())
|
|
# The recommendation flips from scratch to a persistent kind so the
|
|
# create-task dialog's workspace default follows the board setting.
|
|
assert board["default_workspace_kind"] == "dir"
|
|
assert kb.read_board_metadata("late-config")["default_workdir"] == str(
|
|
project_dir.resolve()
|
|
)
|
|
|
|
def test_scheduled_tasks_have_their_own_column_not_todo(client):
|
|
"""Scheduled/time-delay tasks must not be silently bucketed into todo."""
|
|
|
|
task = client.post(
|
|
"/api/plugins/kanban/tasks",
|
|
json={"title": "wait for indexed data", "assignee": "ops"},
|
|
).json()["task"]
|
|
|
|
conn = kbc.connect()
|
|
try:
|
|
with kb.write_txn(conn):
|
|
conn.execute(
|
|
"UPDATE tasks SET status = 'scheduled' WHERE id = ?",
|
|
(task["id"],),
|
|
)
|
|
finally:
|
|
conn.close()
|
|
|
|
r = client.get("/api/plugins/kanban/board")
|
|
assert r.status_code == 200
|
|
columns = {c["name"]: c["tasks"] for c in r.json()["columns"]}
|
|
assert any(t["id"] == task["id"] for t in columns["scheduled"])
|
|
assert not any(t["id"] == task["id"] for t in columns["todo"])
|
|
|
|
def test_tenant_filter(client):
|
|
client.post("/api/plugins/kanban/tasks", json={"title": "A", "tenant": "t1"})
|
|
client.post("/api/plugins/kanban/tasks", json={"title": "B", "tenant": "t2"})
|
|
|
|
r = client.get("/api/plugins/kanban/board?tenant=t1")
|
|
counts = {c["name"]: len(c["tasks"]) for c in r.json()["columns"]}
|
|
total = sum(counts.values())
|
|
assert total == 1
|
|
|
|
r = client.get("/api/plugins/kanban/board?tenant=t2")
|
|
total = sum(len(c["tasks"]) for c in r.json()["columns"])
|
|
assert total == 1
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# GET /tasks/:id returns body + comments + events + links
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_task_detail_includes_links_and_events(client):
|
|
parent = client.post(
|
|
"/api/plugins/kanban/tasks", json={"title": "parent"},
|
|
).json()["task"]
|
|
child = client.post(
|
|
"/api/plugins/kanban/tasks",
|
|
json={"title": "child", "parents": [parent["id"]]},
|
|
).json()["task"]
|
|
assert child["status"] == "todo" # parent not done yet
|
|
|
|
# Detail for the child shows the parent link.
|
|
r = client.get(f"/api/plugins/kanban/tasks/{child['id']}")
|
|
assert r.status_code == 200
|
|
data = r.json()
|
|
assert data["task"]["id"] == child["id"]
|
|
assert parent["id"] in data["links"]["parents"]
|
|
|
|
# Detail for the parent shows the child.
|
|
r = client.get(f"/api/plugins/kanban/tasks/{parent['id']}")
|
|
assert child["id"] in r.json()["links"]["children"]
|
|
|
|
# Events exist from creation.
|
|
assert len(data["events"]) >= 1
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# PATCH /tasks/:id — status transitions
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_patch_review_lifecycle_preserves_handoff_and_reopens(client):
|
|
secret = "ghp_" + "D" * 40
|
|
task = client.post(
|
|
"/api/plugins/kanban/tasks", json={"title": "review me", "assignee": "builder"},
|
|
).json()["task"]
|
|
|
|
response = client.patch(
|
|
f"/api/plugins/kanban/tasks/{task['id']}",
|
|
json={
|
|
"status": "review",
|
|
"assignee": "reviewer",
|
|
"summary": f"Implementation ready. {secret}",
|
|
"metadata": {"tests_run": 4, "token": secret},
|
|
},
|
|
)
|
|
assert response.status_code == 200, response.text
|
|
assert response.json()["task"]["status"] == "review"
|
|
with kbc.connect() as conn:
|
|
run = kb.latest_run(conn, task["id"])
|
|
assert run is not None
|
|
assert run.outcome == "review_requested"
|
|
assert run.metadata is not None
|
|
assert run.metadata["tests_run"] == 4
|
|
assert secret not in str(run.summary)
|
|
assert secret not in json.dumps(run.metadata)
|
|
review_event = [
|
|
event for event in kb.list_events(conn, task["id"])
|
|
if event.kind == "review_requested"
|
|
][-1]
|
|
assert secret not in json.dumps(review_event.payload)
|
|
assert review_event.payload is not None
|
|
assert review_event.payload["implementer"] == "builder"
|
|
assert review_event.payload["reviewer"] == "reviewer"
|
|
|
|
response = client.patch(
|
|
f"/api/plugins/kanban/tasks/{task['id']}",
|
|
json={"status": "ready"},
|
|
)
|
|
assert response.status_code == 200, response.text
|
|
assert response.json()["task"]["status"] == "ready"
|
|
assert response.json()["task"]["assignee"] == "builder"
|
|
with kbc.connect() as conn:
|
|
assert any(
|
|
event.kind == "review_reopened"
|
|
for event in kb.list_events(conn, task["id"])
|
|
)
|
|
|
|
def test_reopening_parent_demotes_ready_child(client):
|
|
"""Reopening a completed parent must invalidate ready children immediately.
|
|
|
|
The dispatcher re-checks parent completion on claim, but the dashboard
|
|
should not keep showing a stale child as ready after an operator drags
|
|
its parent back out of done for more work.
|
|
"""
|
|
parent = client.post("/api/plugins/kanban/tasks", json={"title": "p"}).json()["task"]
|
|
child = client.post(
|
|
"/api/plugins/kanban/tasks",
|
|
json={"title": "c", "parents": [parent["id"]]},
|
|
).json()["task"]
|
|
assert child["status"] == "todo"
|
|
|
|
r = client.patch(
|
|
f"/api/plugins/kanban/tasks/{parent['id']}",
|
|
json={"status": "done", "result": "done", "summary": "done"},
|
|
)
|
|
assert r.status_code == 200
|
|
|
|
child_after_done = client.get(
|
|
f"/api/plugins/kanban/tasks/{child['id']}"
|
|
).json()["task"]
|
|
assert child_after_done["status"] == "ready"
|
|
|
|
r = client.patch(
|
|
f"/api/plugins/kanban/tasks/{parent['id']}",
|
|
json={"status": "todo"},
|
|
)
|
|
assert r.status_code == 200
|
|
|
|
child_after_reopen = client.get(
|
|
f"/api/plugins/kanban/tasks/{child['id']}"
|
|
).json()["task"]
|
|
assert child_after_reopen["status"] == "todo"
|
|
|
|
def test_reopening_parent_retracts_review_and_blocks_approval(client):
|
|
with kbc.connect() as conn:
|
|
parent_id = kb.create_task(conn, title="parent", assignee="planner")
|
|
assert kb.complete_task(conn, parent_id, result="done")
|
|
child_id = kb.create_task(
|
|
conn,
|
|
title="child in review",
|
|
assignee="reviewer",
|
|
parents=[parent_id],
|
|
)
|
|
grandchild_id = kb.create_task(
|
|
conn,
|
|
title="downstream",
|
|
assignee="writer",
|
|
parents=[child_id],
|
|
)
|
|
implementation = kb.claim_task(conn, child_id)
|
|
assert implementation is not None
|
|
assert kb.request_review(
|
|
conn,
|
|
child_id,
|
|
summary="ready",
|
|
expected_run_id=implementation.current_run_id,
|
|
)
|
|
active_review = kb.claim_review_task(conn, child_id)
|
|
assert active_review is not None
|
|
|
|
response = client.patch(
|
|
f"/api/plugins/kanban/tasks/{parent_id}",
|
|
json={"status": "ready"},
|
|
)
|
|
assert response.status_code == 200, response.text
|
|
|
|
with kbc.connect() as conn:
|
|
child = kb.get_task(conn, child_id)
|
|
assert child is not None
|
|
assert child.status == "todo"
|
|
reclaimed = kb.latest_run(conn, child_id)
|
|
assert reclaimed is not None
|
|
assert reclaimed.outcome == "reclaimed"
|
|
assert kb.claim_review_task(conn, child_id) is None
|
|
assert not kb.complete_task(conn, child_id, summary="must not approve")
|
|
grandchild = kb.get_task(conn, grandchild_id)
|
|
assert grandchild is not None
|
|
assert grandchild.status == "todo"
|
|
|
|
response = client.patch(
|
|
f"/api/plugins/kanban/tasks/{parent_id}",
|
|
json={"status": "done", "result": "done", "summary": "done"},
|
|
)
|
|
assert response.status_code == 200, response.text
|
|
|
|
with kbc.connect() as conn:
|
|
child = kb.get_task(conn, child_id)
|
|
assert child is not None
|
|
assert child.status == "review"
|
|
review = kb.claim_review_task(conn, child_id)
|
|
assert review is not None
|
|
assert kb.complete_task(
|
|
conn,
|
|
child_id,
|
|
summary="approved after parent stabilized",
|
|
expected_run_id=review.current_run_id,
|
|
)
|
|
grandchild = kb.get_task(conn, grandchild_id)
|
|
assert grandchild is not None
|
|
assert grandchild.status == "ready"
|
|
|
|
def test_reopening_parent_recursively_retracts_done_and_running_descendants(client):
|
|
with kbc.connect() as conn:
|
|
parent_id = kb.create_task(conn, title="root", assignee="planner")
|
|
assert kb.complete_task(conn, parent_id, result="done")
|
|
child_id = kb.create_task(
|
|
conn,
|
|
title="accepted child",
|
|
assignee="builder",
|
|
parents=[parent_id],
|
|
)
|
|
assert kb.complete_task(conn, child_id, result="done")
|
|
grandchild_id = kb.create_task(
|
|
conn,
|
|
title="running grandchild",
|
|
assignee="writer",
|
|
parents=[child_id],
|
|
)
|
|
grandchild_run = kb.claim_task(conn, grandchild_id)
|
|
assert grandchild_run is not None
|
|
|
|
response = client.patch(
|
|
f"/api/plugins/kanban/tasks/{parent_id}",
|
|
json={"status": "ready"},
|
|
)
|
|
assert response.status_code == 200, response.text
|
|
|
|
with kbc.connect() as conn:
|
|
child = kb.get_task(conn, child_id)
|
|
grandchild = kb.get_task(conn, grandchild_id)
|
|
assert child is not None and child.status == "todo"
|
|
assert grandchild is not None and grandchild.status == "todo"
|
|
assert grandchild.current_run_id is None
|
|
assert kb.claim_task(conn, grandchild_id) is None
|
|
reclaimed = kb.latest_run(conn, grandchild_id)
|
|
assert reclaimed is not None
|
|
assert reclaimed.outcome == "reclaimed"
|
|
|
|
response = client.patch(
|
|
f"/api/plugins/kanban/tasks/{parent_id}",
|
|
json={"status": "done", "result": "done", "summary": "done"},
|
|
)
|
|
assert response.status_code == 200, response.text
|
|
with kbc.connect() as conn:
|
|
child = kb.get_task(conn, child_id)
|
|
grandchild = kb.get_task(conn, grandchild_id)
|
|
assert child is not None and child.status == "ready"
|
|
assert grandchild is not None and grandchild.status == "todo"
|
|
|
|
def test_dashboard_reclaim_of_active_review_preserves_review_phase(client):
|
|
with kbc.connect() as conn:
|
|
task_id = kb.create_task(conn, title="active review", assignee="reviewer")
|
|
implementation = kb.claim_task(conn, task_id)
|
|
assert implementation is not None
|
|
assert kb.request_review(
|
|
conn,
|
|
task_id,
|
|
summary="ready",
|
|
expected_run_id=implementation.current_run_id,
|
|
)
|
|
review = kb.claim_review_task(conn, task_id)
|
|
assert review is not None
|
|
|
|
response = client.patch(
|
|
f"/api/plugins/kanban/tasks/{task_id}",
|
|
json={"status": "ready"},
|
|
)
|
|
assert response.status_code == 200, response.text
|
|
assert response.json()["task"]["status"] == "review"
|
|
assert response.json()["task"]["assignee"] == "reviewer"
|
|
with kbc.connect() as conn:
|
|
run = kb.latest_run(conn, task_id)
|
|
assert run is not None
|
|
assert run.outcome == "reclaimed"
|
|
next_review = kb.claim_review_task(conn, task_id)
|
|
assert next_review is not None
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# DELETE /tasks/:id
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_delete_task(client):
|
|
t = client.post("/api/plugins/kanban/tasks", json={"title": "to-delete"}).json()["task"]
|
|
r = client.delete(f"/api/plugins/kanban/tasks/{t['id']}")
|
|
assert r.status_code == 200
|
|
assert r.json()["deleted"] is True
|
|
assert r.json()["task_id"] == t["id"]
|
|
|
|
# Gone from board
|
|
board = client.get("/api/plugins/kanban/board").json()
|
|
all_ids = [tt["id"] for col in board["columns"] for tt in col["tasks"]]
|
|
assert t["id"] not in all_ids
|
|
|
|
# Gone from detail
|
|
r = client.get(f"/api/plugins/kanban/tasks/{t['id']}")
|
|
assert r.status_code == 404
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Comments + Links
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_add_comment(client):
|
|
t = client.post("/api/plugins/kanban/tasks", json={"title": "x"}).json()["task"]
|
|
r = client.post(
|
|
f"/api/plugins/kanban/tasks/{t['id']}/comments",
|
|
json={"body": "how's progress?", "author": "teknium"},
|
|
)
|
|
assert r.status_code == 200
|
|
|
|
r = client.get(f"/api/plugins/kanban/tasks/{t['id']}")
|
|
comments = r.json()["comments"]
|
|
assert len(comments) == 1
|
|
assert comments[0]["body"] == "how's progress?"
|
|
assert comments[0]["author"] == "teknium"
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Dispatch nudge
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_dispatch_dry_run(client):
|
|
client.post(
|
|
"/api/plugins/kanban/tasks",
|
|
json={"title": "work", "assignee": "researcher"},
|
|
)
|
|
r = client.post("/api/plugins/kanban/dispatch?dry_run=true&max=4")
|
|
assert r.status_code == 200
|
|
body = r.json()
|
|
# DispatchResult is serialized as a dataclass dict.
|
|
assert isinstance(body, dict)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Triage column (new v1 status)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Progress rollup (done children / total children)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Auto-init on first board read
|
|
# ---------------------------------------------------------------------------
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# WebSocket auth (query-param token)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_ws_events_rejects_when_token_required(tmp_path, monkeypatch):
|
|
"""Loopback mode: a missing or wrong ?token= must be rejected with
|
|
policy-violation; the correct token is accepted. The kanban WS now
|
|
delegates to web_server_chat._ws_auth_ok, so we stub that with the real
|
|
loopback-token semantics (auth_required False → constant-time token
|
|
compare)."""
|
|
home = tmp_path / ".hermes"
|
|
home.mkdir()
|
|
monkeypatch.setenv("HERMES_HOME", str(home))
|
|
monkeypatch.setattr(Path, "home", lambda: tmp_path)
|
|
kb.init_db()
|
|
|
|
# Stub web_server_chat with a loopback-mode _ws_auth_ok (auth_required False →
|
|
# accept only the correct ?token=). Mirrors the real gate's loopback path.
|
|
import hermes_cli
|
|
import types
|
|
|
|
def _fake_ws_auth_ok(ws):
|
|
return ws.query_params.get("token", "") == "secret-xyz"
|
|
|
|
stub = types.SimpleNamespace(
|
|
_SESSION_TOKEN="secret-xyz",
|
|
_ws_auth_ok=_fake_ws_auth_ok,
|
|
)
|
|
monkeypatch.setitem(sys.modules, "hermes_cli.web_server_chat", stub)
|
|
monkeypatch.setattr(hermes_cli, "web_server_chat", stub, raising=False)
|
|
|
|
app = FastAPI()
|
|
app.include_router(_load_plugin_router(), prefix="/api/plugins/kanban")
|
|
c = TestClient(app)
|
|
|
|
# No token → policy violation close.
|
|
from starlette.websockets import WebSocketDisconnect
|
|
with pytest.raises(WebSocketDisconnect) as exc:
|
|
with c.websocket_connect("/api/plugins/kanban/events"):
|
|
pass
|
|
assert exc.value.code == 1008
|
|
|
|
# Wrong token → policy violation close.
|
|
with pytest.raises(WebSocketDisconnect) as exc:
|
|
with c.websocket_connect("/api/plugins/kanban/events?token=nope"):
|
|
pass
|
|
assert exc.value.code == 1008
|
|
|
|
# Correct token → accepted (connect then close cleanly from our side).
|
|
with c.websocket_connect(
|
|
"/api/plugins/kanban/events?token=secret-xyz"
|
|
) as ws:
|
|
assert ws is not None # handshake succeeded
|
|
|
|
# The bug symptom was a traceback; we don't assert on stderr because
|
|
# capturing asyncio's internal "exception was never retrieved" logging
|
|
# is flaky. The assertion that matters is: no CancelledError escaped.
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Bulk actions
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_bulk_status_ready(client):
|
|
a = client.post("/api/plugins/kanban/tasks", json={"title": "a"}).json()["task"]
|
|
b = client.post("/api/plugins/kanban/tasks", json={"title": "b"}).json()["task"]
|
|
c2 = client.post("/api/plugins/kanban/tasks", json={"title": "c"}).json()["task"]
|
|
# Parent-less tasks land in "ready" already; push them to blocked first.
|
|
for tid in (a["id"], b["id"], c2["id"]):
|
|
client.patch(
|
|
f"/api/plugins/kanban/tasks/{tid}",
|
|
json={"status": "blocked", "block_reason": "wait"},
|
|
)
|
|
|
|
response = client.post(
|
|
"/api/plugins/kanban/tasks/bulk",
|
|
json={"ids": [a["id"], b["id"], c2["id"]], "status": "ready"},
|
|
)
|
|
assert response.status_code == 200
|
|
results = response.json()["results"]
|
|
assert all(item["ok"] for item in results)
|
|
# All three are now ready.
|
|
board = client.get("/api/plugins/kanban/board").json()
|
|
ready = next(col for col in board["columns"] if col["name"] == "ready")
|
|
ids = {task["id"] for task in ready["tasks"]}
|
|
assert {a["id"], b["id"], c2["id"]}.issubset(ids)
|
|
|
|
def test_bulk_review_assignment_preserves_implementer_provenance(client):
|
|
tasks = [
|
|
client.post(
|
|
"/api/plugins/kanban/tasks",
|
|
json={"title": title, "assignee": "builder"},
|
|
).json()["task"]
|
|
for title in ("review a", "review b")
|
|
]
|
|
response = client.post(
|
|
"/api/plugins/kanban/tasks/bulk",
|
|
json={
|
|
"ids": [task["id"] for task in tasks],
|
|
"status": "review",
|
|
"assignee": "reviewer",
|
|
"summary": "ready",
|
|
},
|
|
)
|
|
assert response.status_code == 200, response.text
|
|
assert all(item["ok"] for item in response.json()["results"])
|
|
with kbc.connect() as conn:
|
|
for task in tasks:
|
|
current = kb.get_task(conn, task["id"])
|
|
assert current is not None
|
|
assert current.status == "review"
|
|
assert current.assignee == "reviewer"
|
|
event = [
|
|
item for item in kb.list_events(conn, task["id"])
|
|
if item.kind == "review_requested"
|
|
][-1]
|
|
assert event.payload is not None
|
|
assert event.payload["implementer"] == "builder"
|
|
assert event.payload["reviewer"] == "reviewer"
|
|
|
|
def test_bulk_status_done_forwards_completion_summary(client):
|
|
a = client.post("/api/plugins/kanban/tasks", json={"title": "a"}).json()["task"]
|
|
b = client.post("/api/plugins/kanban/tasks", json={"title": "b"}).json()["task"]
|
|
|
|
r = client.post(
|
|
"/api/plugins/kanban/tasks/bulk",
|
|
json={
|
|
"ids": [a["id"], b["id"]],
|
|
"status": "done",
|
|
"result": "DECIDED: ship it",
|
|
"summary": "DECIDED: ship it",
|
|
"metadata": {"source": "dashboard"},
|
|
},
|
|
)
|
|
|
|
assert r.status_code == 200
|
|
assert all(r["ok"] for r in r.json()["results"])
|
|
conn = kbc.connect()
|
|
try:
|
|
for tid in (a["id"], b["id"]):
|
|
task = kb.get_task(conn, tid)
|
|
run = kb.latest_run(conn, tid)
|
|
assert task.status == "done"
|
|
assert task.result == "DECIDED: ship it"
|
|
assert run.summary == "DECIDED: ship it"
|
|
assert run.metadata == {"source": "dashboard"}
|
|
finally:
|
|
conn.close()
|
|
|
|
def _gated_child(client):
|
|
parent = client.post("/api/plugins/kanban/tasks", json={"title": "parent"}).json()["task"]
|
|
child = client.post(
|
|
"/api/plugins/kanban/tasks", json={"title": "child", "parents": [parent["id"]]},
|
|
).json()["task"]
|
|
return parent["id"], child["id"]
|
|
|
|
def test_patch_done_or_review_refused_by_open_parent_names_it(client):
|
|
"""A completion refused by the dependency gate must say which parent is open,
|
|
not the generic 'not valid from current state'."""
|
|
parent_id, child_id = _gated_child(client)
|
|
for status in ("done", "review"):
|
|
r = client.patch(f"/api/plugins/kanban/tasks/{child_id}", json={"status": status})
|
|
assert r.status_code == 409, r.text
|
|
detail = r.json()["detail"]
|
|
assert f"{parent_id} (ready)" in detail, detail
|
|
assert "unsatisfied parent" in detail, detail
|
|
|
|
def test_bulk_done_refused_by_open_parent_names_it(client):
|
|
parent_id, child_id = _gated_child(client)
|
|
r = client.post("/api/plugins/kanban/tasks/bulk", json={"ids": [child_id], "status": "done"})
|
|
assert r.status_code == 200
|
|
entry = r.json()["results"][0]
|
|
assert entry["ok"] is False
|
|
assert f"{parent_id} (ready)" in entry["error"], entry
|
|
assert "unsatisfied parent" in entry["error"], entry
|
|
|
|
def test_bulk_status_running_rejected(client):
|
|
"""Bulk updates must match single-task PATCH: direct 'running' is invalid."""
|
|
t = client.post("/api/plugins/kanban/tasks", json={"title": "x"}).json()["task"]
|
|
|
|
r = client.post(
|
|
"/api/plugins/kanban/tasks/bulk",
|
|
json={"ids": [t["id"]], "status": "running"},
|
|
)
|
|
|
|
assert r.status_code == 200
|
|
results = r.json()["results"]
|
|
assert len(results) == 1
|
|
assert results[0]["id"] == t["id"]
|
|
assert results[0]["ok"] is False
|
|
assert "running" in results[0]["error"]
|
|
|
|
board = client.get("/api/plugins/kanban/board").json()
|
|
statuses = {
|
|
tt["id"]: col["name"]
|
|
for col in board["columns"]
|
|
for tt in col["tasks"]
|
|
}
|
|
assert statuses.get(t["id"]) != "running"
|
|
|
|
def test_dashboard_confirm_dispatches_expected_patch_body(client):
|
|
"""Behavioral: the PATCH body shape the bundle produces on confirm
|
|
(status + result + summary) must be accepted by the backend without
|
|
rejection. The backend stores ``result`` as the human-readable
|
|
completion summary (the bundle comments confirm ``summary`` is sent
|
|
duplicatively so the backend can store the value under its preferred
|
|
key while the wire format remains explicit).
|
|
This is the contract the bundle's performMoveTask relies on.
|
|
"""
|
|
t = client.post("/api/plugins/kanban/tasks",
|
|
json={"title": "x"}).json()["task"]
|
|
# Bundle's performMoveTask on confirm with a summary produces:
|
|
# { status, result: summary, summary: summary }
|
|
r = client.patch(
|
|
f"/api/plugins/kanban/tasks/{t['id']}",
|
|
json={"status": "done", "result": "shipped", "summary": "shipped"},
|
|
)
|
|
assert r.status_code == 200, r.text
|
|
body = r.json()["task"]
|
|
assert body["status"] == "done"
|
|
assert body.get("result") == "shipped"
|
|
|
|
def test_bulk_archive(client):
|
|
a = client.post("/api/plugins/kanban/tasks", json={"title": "a"}).json()["task"]
|
|
b = client.post("/api/plugins/kanban/tasks", json={"title": "b"}).json()["task"]
|
|
r = client.post("/api/plugins/kanban/tasks/bulk",
|
|
json={"ids": [a["id"], b["id"]], "archive": True})
|
|
assert r.status_code == 200
|
|
assert all(r["ok"] for r in r.json()["results"])
|
|
# Default board (archived hidden) — both gone.
|
|
board = client.get("/api/plugins/kanban/board").json()
|
|
ids = {t["id"] for col in board["columns"] for t in col["tasks"]}
|
|
assert a["id"] not in ids
|
|
assert b["id"] not in ids
|
|
|
|
def test_bulk_reassign(client):
|
|
a = client.post("/api/plugins/kanban/tasks",
|
|
json={"title": "a", "assignee": "old"}).json()["task"]
|
|
b = client.post("/api/plugins/kanban/tasks",
|
|
json={"title": "b", "assignee": "old"}).json()["task"]
|
|
r = client.post("/api/plugins/kanban/tasks/bulk",
|
|
json={"ids": [a["id"], b["id"]], "assignee": "new"})
|
|
assert r.status_code == 200
|
|
for tid in (a["id"], b["id"]):
|
|
t = client.get(f"/api/plugins/kanban/tasks/{tid}").json()["task"]
|
|
assert t["assignee"] == "new"
|
|
|
|
def test_bulk_unassign_via_empty_string(client):
|
|
a = client.post("/api/plugins/kanban/tasks",
|
|
json={"title": "a", "assignee": "x"}).json()["task"]
|
|
r = client.post("/api/plugins/kanban/tasks/bulk",
|
|
json={"ids": [a["id"]], "assignee": ""})
|
|
assert r.status_code == 200
|
|
t = client.get(f"/api/plugins/kanban/tasks/{a['id']}").json()["task"]
|
|
assert t["assignee"] is None
|
|
|
|
def test_bulk_partial_failure_doesnt_abort_siblings(client):
|
|
"""One bad id in the middle of a batch must not prevent others from
|
|
applying."""
|
|
a = client.post("/api/plugins/kanban/tasks", json={"title": "a"}).json()["task"]
|
|
c2 = client.post("/api/plugins/kanban/tasks", json={"title": "c"}).json()["task"]
|
|
r = client.post("/api/plugins/kanban/tasks/bulk",
|
|
json={"ids": [a["id"], "bogus-id", c2["id"]], "priority": 7})
|
|
assert r.status_code == 200
|
|
results = r.json()["results"]
|
|
assert len(results) == 3
|
|
ok_ids = {r["id"] for r in results if r["ok"]}
|
|
assert a["id"] in ok_ids
|
|
assert c2["id"] in ok_ids
|
|
assert any(not r["ok"] and r["id"] == "bogus-id" for r in results)
|
|
# Good siblings actually got the priority bump.
|
|
for tid in (a["id"], c2["id"]):
|
|
t = client.get(f"/api/plugins/kanban/tasks/{tid}").json()["task"]
|
|
assert t["priority"] == 7
|
|
|
|
def test_bulk_empty_ids_400(client):
|
|
r = client.post("/api/plugins/kanban/tasks/bulk", json={"ids": []})
|
|
assert r.status_code == 400
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# /config endpoint
|
|
# ---------------------------------------------------------------------------
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# /config endpoint
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_config_reads_dashboard_kanban_section(tmp_path, monkeypatch, client):
|
|
home = Path(os.environ["HERMES_HOME"])
|
|
(home / "config.yaml").write_text(
|
|
"dashboard:\n"
|
|
" kanban:\n"
|
|
" default_tenant: acme\n"
|
|
" lane_by_profile: false\n"
|
|
" include_archived_by_default: true\n"
|
|
" render_markdown: false\n"
|
|
)
|
|
r = client.get("/api/plugins/kanban/config")
|
|
assert r.status_code == 200
|
|
data = r.json()
|
|
assert data["default_tenant"] == "acme"
|
|
assert data["lane_by_profile"] is False
|
|
assert data["include_archived_by_default"] is True
|
|
assert data["render_markdown"] is False
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Runs surfacing (vulcan-artivus RFC feedback)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_event_dict_includes_run_id(client):
|
|
"""GET /tasks/:id returns events with run_id populated."""
|
|
r = client.post("/api/plugins/kanban/tasks", json={"title": "e", "assignee": "worker"})
|
|
tid = r.json()["task"]["id"]
|
|
from hermes_cli import kanban_db as kb
|
|
from hermes_cli import kanban_db_connect as kbc
|
|
conn = kbc.connect()
|
|
try:
|
|
kb.claim_task(conn, tid)
|
|
run_id = kb.latest_run(conn, tid).id
|
|
kb.complete_task(conn, tid, summary="wss")
|
|
finally:
|
|
conn.close()
|
|
|
|
r = client.get(f"/api/plugins/kanban/tasks/{tid}")
|
|
assert r.status_code == 200
|
|
events = r.json()["events"]
|
|
# Every event in the response must have a run_id key (None or int).
|
|
for e in events:
|
|
assert "run_id" in e, f"missing run_id in event: {e}"
|
|
# completed event must have the actual run_id.
|
|
comp = [e for e in events if e["kind"] == "completed"]
|
|
assert comp[0]["run_id"] == run_id
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Per-task force-loaded skills via REST
|
|
# ---------------------------------------------------------------------------
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Dispatcher-presence warning in POST /tasks response
|
|
# ---------------------------------------------------------------------------
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _task_dict — outer try/except fallback when task_age raises
|
|
#
|
|
# Background: kanban_db.task_age was hardened in 061a1830 to return None for
|
|
# corrupt timestamp values via _safe_int. The companion fix added a belt-and-
|
|
# suspenders try/except in plugin_api._task_dict so that *any future* exception
|
|
# from task_age (not just ValueError on '%s') still yields a usable dict
|
|
# instead of 500'ing GET /board for the entire org.
|
|
#
|
|
# kanban_db._safe_int / task_age corruption paths are covered in
|
|
# tests/hermes_cli/test_kanban_db.py. The OUTER fallback here is not, which
|
|
# means a refactor that drops the try/except would not be caught by CI. The
|
|
# tests below pin that contract.
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_FALLBACK_AGE = {
|
|
"created_age_seconds": None,
|
|
"started_age_seconds": None,
|
|
"time_to_complete_seconds": None,
|
|
}
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Home-channel subscription endpoints (#19534 follow-up: GUI opt-in)
|
|
# ---------------------------------------------------------------------------
|
|
#
|
|
# Dashboard surface for per-task, per-platform notification toggles. The
|
|
# backend endpoints read the live GatewayConfig, so tests set env vars
|
|
# (BOT_TOKEN + HOME_CHANNEL) to simulate a user who has run /sethome on
|
|
# telegram and discord.
|
|
|
|
@pytest.fixture
|
|
def with_home_channels(monkeypatch):
|
|
"""Simulate a user with home channels set on telegram and discord."""
|
|
monkeypatch.setenv("TELEGRAM_BOT_TOKEN", "abc:fake")
|
|
monkeypatch.setenv("TELEGRAM_HOME_CHANNEL", "1234567")
|
|
monkeypatch.setenv("TELEGRAM_HOME_CHANNEL_THREAD_ID", "42")
|
|
monkeypatch.setenv("TELEGRAM_HOME_CHANNEL_NAME", "Main TG")
|
|
monkeypatch.setenv("DISCORD_BOT_TOKEN", "disc_fake")
|
|
monkeypatch.setenv("DISCORD_HOME_CHANNEL", "9999999")
|
|
monkeypatch.setenv("DISCORD_HOME_CHANNEL_NAME", "Main Discord")
|
|
# Slack has a token but NO home — should be excluded from the list.
|
|
monkeypatch.setenv("SLACK_BOT_TOKEN", "slack_fake")
|
|
|
|
def test_home_channels_lists_only_platforms_with_home(client, with_home_channels):
|
|
"""GET /home-channels returns entries only for platforms where the
|
|
user has set a home; untoggled-subscribed bool is false by default."""
|
|
r = client.get("/api/plugins/kanban/home-channels")
|
|
assert r.status_code == 200
|
|
platforms = {h["platform"] for h in r.json()["home_channels"]}
|
|
assert platforms == {"telegram", "discord"}, (
|
|
f"slack has a token but no home — must not appear. got {platforms}"
|
|
)
|
|
for h in r.json()["home_channels"]:
|
|
assert h["subscribed"] is False
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Recovery endpoints (reclaim + reassign) and warnings field
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_reclaim_endpoint_releases_running_claim(client):
|
|
"""POST /tasks/<id>/reclaim drops the claim, returns ok, and emits
|
|
a manual reclaimed event."""
|
|
import secrets
|
|
conn = kbc.connect()
|
|
try:
|
|
t = kb.create_task(conn, title="running", assignee="x")
|
|
lock = secrets.token_hex(8)
|
|
future = int(time.time()) + 3600
|
|
conn.execute(
|
|
"UPDATE tasks SET status='running', claim_lock=?, claim_expires=?, "
|
|
"worker_pid=? WHERE id=?",
|
|
(lock, future, 99999, t),
|
|
)
|
|
conn.execute(
|
|
"INSERT INTO task_runs (task_id, status, claim_lock, claim_expires, "
|
|
"worker_pid, started_at) VALUES (?, 'running', ?, ?, ?, ?)",
|
|
(t, lock, future, 99999, int(time.time())),
|
|
)
|
|
run_id = conn.execute("SELECT last_insert_rowid()").fetchone()[0]
|
|
conn.execute("UPDATE tasks SET current_run_id=? WHERE id=?", (run_id, t))
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|
|
|
|
r = client.post(
|
|
f"/api/plugins/kanban/tasks/{t}/reclaim",
|
|
json={"reason": "browser recovery"},
|
|
)
|
|
assert r.status_code == 200, r.text
|
|
body = r.json()
|
|
assert body["ok"] is True
|
|
assert body["task_id"] == t
|
|
|
|
# Confirm the task is back to ready.
|
|
conn2 = kbc.connect()
|
|
try:
|
|
row = conn2.execute(
|
|
"SELECT status, claim_lock FROM tasks WHERE id=?", (t,),
|
|
).fetchone()
|
|
assert row["status"] == "ready"
|
|
assert row["claim_lock"] is None
|
|
finally:
|
|
conn2.close()
|
|
|
|
def test_reassign_endpoint_switches_profile(client):
|
|
"""POST /tasks/<id>/reassign changes the assignee field."""
|
|
conn = kbc.connect()
|
|
try:
|
|
t = kb.create_task(conn, title="task", assignee="orig")
|
|
finally:
|
|
conn.close()
|
|
|
|
r = client.post(
|
|
f"/api/plugins/kanban/tasks/{t}/reassign",
|
|
json={"profile": "newbie", "reclaim_first": False},
|
|
)
|
|
assert r.status_code == 200, r.text
|
|
assert r.json()["assignee"] == "newbie"
|
|
|
|
conn2 = kbc.connect()
|
|
try:
|
|
row = conn2.execute(
|
|
"SELECT assignee FROM tasks WHERE id=?", (t,),
|
|
).fetchone()
|
|
assert row["assignee"] == "newbie"
|
|
finally:
|
|
conn2.close()
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Diagnostics endpoint (/api/plugins/kanban/diagnostics)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_diagnostics_endpoint_surfaces_blocked_hallucination(client):
|
|
conn = kbc.connect()
|
|
try:
|
|
parent = kb.create_task(conn, title="parent", assignee="alice")
|
|
real = kb.create_task(conn, title="real", assignee="x", created_by="alice")
|
|
import pytest as _pytest
|
|
with _pytest.raises(kb.HallucinatedCardsError):
|
|
kb.complete_task(
|
|
conn, parent, summary="phantom",
|
|
created_cards=[real, "t_ffff00001234"],
|
|
)
|
|
finally:
|
|
conn.close()
|
|
|
|
r = client.get("/api/plugins/kanban/diagnostics")
|
|
assert r.status_code == 200
|
|
data = r.json()
|
|
assert data["count"] == 1
|
|
row = data["diagnostics"][0]
|
|
assert row["task_id"] == parent
|
|
assert row["diagnostics"][0]["kind"] == "hallucinated_cards"
|
|
assert row["diagnostics"][0]["severity"] == "error"
|
|
assert "t_ffff00001234" in row["diagnostics"][0]["data"]["phantom_ids"]
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# POST /tasks/:id/specify — triage specifier endpoint
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _patch_specifier_response(monkeypatch, *, content, model="test-model"):
|
|
"""Helper: install a fake auxiliary client so the specifier endpoint
|
|
can run without hitting any real provider."""
|
|
from unittest.mock import MagicMock
|
|
|
|
resp = MagicMock()
|
|
resp.choices = [MagicMock()]
|
|
resp.choices[0].message.content = content
|
|
# specify_task routes through call_llm now (#35566) — mock it directly.
|
|
fake_call = MagicMock(return_value=resp)
|
|
monkeypatch.setattr("agent.auxiliary_client.call_llm", fake_call)
|
|
return fake_call
|
|
|
|
def test_specify_happy_path(client, monkeypatch):
|
|
import json as jsonlib
|
|
|
|
# Create a triage task.
|
|
t = client.post(
|
|
"/api/plugins/kanban/tasks",
|
|
json={"title": "one-liner", "triage": True},
|
|
).json()["task"]
|
|
assert t["status"] == "triage"
|
|
|
|
_patch_specifier_response(
|
|
monkeypatch,
|
|
content=jsonlib.dumps(
|
|
{"title": "Polished", "body": "**Goal**\nDo the thing."}
|
|
),
|
|
)
|
|
|
|
r = client.post(
|
|
f"/api/plugins/kanban/tasks/{t['id']}/specify",
|
|
json={"author": "ui-tester"},
|
|
)
|
|
assert r.status_code == 200
|
|
body = r.json()
|
|
assert body["ok"] is True
|
|
assert body["task_id"] == t["id"]
|
|
assert body["new_title"] == "Polished"
|
|
|
|
# Task should have moved off the triage column.
|
|
detail = client.get(f"/api/plugins/kanban/tasks/{t['id']}").json()["task"]
|
|
assert detail["status"] in {"todo", "ready"}
|
|
assert detail["title"] == "Polished"
|
|
assert "**Goal**" in (detail["body"] or "")
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Aux-LLM endpoints under multiplexed hosting — profile secret scope (#123372)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_specify_resolves_each_profiles_key_under_multiplex(kanban_home, tmp_path, monkeypatch):
|
|
"""Specify / Decompose / Estimate reach the aux client with no agent turn, so under
|
|
multi-profile hosting an unscoped provider-key read fails closed (``LLM error:
|
|
UnscopedSecretError``). The plugin router is mounted the way ``_mount_plugin_api_routes``
|
|
mounts every plugin router — behind ``_plugin_route_secret_scope`` — so the launch profile
|
|
(A) and a ``?profile=`` request (B) each resolve their OWN key, and B never leaks into A."""
|
|
import agent.secret_scope as ss
|
|
from fastapi import Depends
|
|
from hermes_cli import profiles
|
|
from hermes_cli.web_server_dashboard import _plugin_route_secret_scope
|
|
from tui_gateway import launch_profile_policy
|
|
from unittest.mock import MagicMock
|
|
|
|
(kanban_home / ".env").write_text("KANBAN_AUX_SCOPE_TEST_KEY=key-of-launch-a\n")
|
|
profiles_root = tmp_path / "profiles"
|
|
(profiles_root / "workerb").mkdir(parents=True)
|
|
(profiles_root / "workerb" / ".env").write_text("KANBAN_AUX_SCOPE_TEST_KEY=key-of-worker-b\n")
|
|
monkeypatch.setattr(profiles, "_get_default_hermes_home", lambda: kanban_home)
|
|
monkeypatch.setattr(profiles, "_get_profiles_root", lambda: profiles_root)
|
|
|
|
seen: list = []
|
|
|
|
def fake_call_llm(**kwargs):
|
|
seen.append(ss.get_secret("KANBAN_AUX_SCOPE_TEST_KEY"))
|
|
resp = MagicMock()
|
|
resp.choices = [MagicMock()]
|
|
resp.choices[0].message.content = json.dumps({"title": "Polished", "body": "**Goal**\nDo it."})
|
|
return resp
|
|
|
|
monkeypatch.setattr("agent.auxiliary_client.call_llm", fake_call_llm)
|
|
app = FastAPI()
|
|
app.include_router(_load_plugin_router(), prefix="/api/plugins/kanban",
|
|
dependencies=[Depends(_plugin_route_secret_scope)])
|
|
client = TestClient(app)
|
|
|
|
def _specify(profile=None):
|
|
params = {"profile": profile} if profile else None
|
|
task = client.post("/api/plugins/kanban/tasks", params=params,
|
|
json={"title": "one-liner", "triage": True}).json()["task"]
|
|
return client.post(f"/api/plugins/kanban/tasks/{task['id']}/specify", params=params,
|
|
json={"author": "ui-tester"}).json()
|
|
|
|
was_active, snapshot = ss.is_multiplex_active(), launch_profile_policy._snapshot
|
|
ss.set_multiplex_active(True)
|
|
try:
|
|
for profile in (None, "workerb", None):
|
|
body = _specify(profile)
|
|
assert body["ok"] is True, body
|
|
finally:
|
|
ss.set_multiplex_active(was_active)
|
|
launch_profile_policy._snapshot = snapshot
|
|
assert seen == ["key-of-launch-a", "key-of-worker-b", "key-of-launch-a"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Final result visibility for Done cards
|
|
# ---------------------------------------------------------------------------
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Touch drag-vs-tap threshold (#115568)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_touch_card_tap_opens_instead_of_dragging():
|
|
"""attachTouchDrag() must not claim a stationary tap: without a movement threshold,
|
|
every touch pointerdown called preventDefault() immediately, which suppresses the
|
|
synthesized click TaskCard.handleClick relies on to call props.onOpen() (#115568).
|
|
The bundle has no build step, so this runs the real function (extracted verbatim, not
|
|
regex-matched) through a real pointerdown/move/up sequence with a minimal DOM stub —
|
|
behavioral, not a source-text pin.
|
|
"""
|
|
node = shutil.which("node")
|
|
if not node:
|
|
pytest.skip("node not available")
|
|
bundle = Path(__file__).resolve().parents[2] / "plugins" / "kanban" / "dashboard" / "dist" / "index.js"
|
|
probe = Path(__file__).parent / "fixtures" / "kanban_touch_drag_probe.js"
|
|
result = subprocess.run(
|
|
[node, str(probe), str(bundle)],
|
|
capture_output=True, text=True, timeout=30,
|
|
)
|
|
assert result.returncode == 0, f"stdout={result.stdout!r} stderr={result.stderr!r}"
|
|
assert "PASS" in result.stdout
|
|
|
|
|
|
# Run clock: current run start, not first-ever start
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_board_card_exposes_current_run_start(client):
|
|
"""#99819: after a review timeout + retry, the card must expose the fresh
|
|
run's start (not the task's first-ever start) so the run clock ticks from
|
|
the current attempt."""
|
|
now = int(time.time())
|
|
first_start = now - 7200 # task first started 2h ago
|
|
retry_start = now - 90 # retry run started 90s ago
|
|
conn = kbc.connect()
|
|
try:
|
|
t = kb.create_task(conn, title="retried", assignee="x")
|
|
lock = "lock-runclock"
|
|
future = now + 3600
|
|
conn.execute(
|
|
"UPDATE tasks SET status='running', started_at=?, claim_lock=?, "
|
|
"claim_expires=?, worker_pid=? WHERE id=?",
|
|
(first_start, lock, future, 99999, t),
|
|
)
|
|
conn.execute(
|
|
"INSERT INTO task_runs (task_id, status, claim_lock, claim_expires, "
|
|
"worker_pid, started_at) VALUES (?, 'running', ?, ?, ?, ?)",
|
|
(t, lock, future, 99999, retry_start),
|
|
)
|
|
run_id = conn.execute("SELECT last_insert_rowid()").fetchone()[0]
|
|
conn.execute("UPDATE tasks SET current_run_id=? WHERE id=?", (run_id, t))
|
|
# A sibling task with no run at all: key present, null.
|
|
u = kb.create_task(conn, title="unclaimed", assignee="x")
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|
|
|
|
r = client.get("/api/plugins/kanban/board")
|
|
assert r.status_code == 200, r.text
|
|
columns = {c["name"]: c for c in r.json()["columns"]}
|
|
card = next(c for c in columns["running"]["tasks"] if c["id"] == t)
|
|
assert card["started_at"] == first_start
|
|
# Red on base: this key did not exist at all.
|
|
assert card["current_run_started_at"] == retry_start
|
|
todo = next(c for c in columns["ready"]["tasks"] if c["id"] == u)
|
|
assert todo["current_run_started_at"] is None
|
|
|
|
# The detail endpoint carries the same contract.
|
|
detail = client.get(f"/api/plugins/kanban/tasks/{t}").json()["task"]
|
|
assert detail["current_run_started_at"] == retry_start
|