"""Exercise the kvb-org-rasci register: fields, roles, priority order, claims, artifacts, tools.""" from __future__ import annotations import datetime import json import os import subprocess import sys import tempfile import time import unittest from pathlib import Path REPO_ROOT = Path(__file__).resolve().parents[1] RASCI = REPO_ROOT / "skills" / "kvb-org-rasci" / "scripts" / "rasci.py" COUNCIL = REPO_ROOT / "skills" / "kvb-org-organize" / "scripts" / "council.py" WIDE_HEADER = "| Prio | WSJF | Task | OKR | Status | Worker | Artifacts | Start | End | R | A | S | C | I |" COMPACT_HEADER = "| Prio | WSJF | Task | Status | Worker | A |" def local(iso: str) -> str: return datetime.datetime.fromisoformat(iso).astimezone().strftime("%m-%d %H:%M") class TestRasci(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.addCleanup(self.tmp.cleanup) self.home = Path(self.tmp.name) / "home" self.root = Path(self.tmp.name) / "council" self.sessions = self.home / ".claude" / "sessions" self.sessions.mkdir(parents=True) for sid, name in (("s1", "One"), ("s2", "Two")): (self.sessions / f"{sid}.json").write_text( json.dumps({"sessionId": sid, "name": name}), encoding="utf-8" ) def _run(self, *args: str, script: Path = RASCI, sid: str = "s1", timeout: float = 120) -> subprocess.CompletedProcess: env = { "PATH": "/usr/bin:/bin", "HOME": str(self.home), "AGENTS_COUNCIL_ROOT": str(self.root), "CLAUDE_CODE_SESSION_ID": sid, } return subprocess.run( [sys.executable, str(script), *args], capture_output=True, text=True, env=env, cwd=self.tmp.name, check=False, timeout=timeout, ) def _out(self, *args: str) -> list[str]: result = self._run(*args) self.assertEqual(result.returncode, 0, result.stderr) return result.stdout.splitlines() def _claim(self, office: str, session: str, name: str, work: str) -> None: path = self.root / office path.mkdir(parents=True) claim = { "office": office, "session": session, "runtime": "claude-code", "name": name, "since": "2026-11-01T00:00:00+00:00", "beat": time.time(), "work": work, } (path / "claim.json").write_text(json.dumps(claim), encoding="utf-8") def _two_tasks(self) -> str: start = "2026-11-01T10:00:00+00:00" self._run("set", "work/a", "--title", "Alpha", "--objective", "Ship A", "--prio", "2", "--status", "running", "--worker", "Skills", "--start", start) self._run("role", "work/a", "Skills", "ar") self._run("role", "work/a", "Operator", "ic") self._run("role", "work/a", "Pkgmgr", "C") self._run("set", "work/b", "--title", "Beta", "--prio", "1", "--worker", "subagent Explore (haiku)") self._run("role", "work/b", "Pkgmgr", "R") return start def test_wide_matrix_orders_by_priority_and_lists_roles(self): start = self._two_tasks() lines = self._out("show", "--wide") self.assertEqual(lines[0], WIDE_HEADER) self.assertEqual( lines[2], "| 1 | - | work/b: Beta | - | queued | subagent Explore (haiku) | - | - | - | Pkgmgr | | | | |" ) self.assertEqual( lines[3], f"| 2 | - | work/a: Alpha | Ship A | running | Skills | - | {local(start)} | - " "| Skills | Skills | | Operator, Pkgmgr | Operator |", ) self.assertEqual( lines[4], "gaps: work/b has 0 A; work/b has no key result; work/a has no key result" ) def test_show_prints_the_compact_table_by_default(self): self._two_tasks() self.assertEqual( self._out("show"), [ COMPACT_HEADER, "|---|---|---|---|---|---|", "| 1 | - | work/b: Beta | queued | subagent Explore (haiku) | |", "| 2 | - | work/a: Alpha | running | Skills | Skills |", "gaps: work/b has 0 A; work/b has no key result; work/a has no key result", ], ) def test_compact_cells_keep_the_slug_and_truncate_title_worker_and_a(self): slug = "work/" + "s" * 45 self._run("set", slug, "--title", "Hidden", "--prio", "1") self._run("set", "work/a", "--title", "A title far longer than the forty character cell", "--prio", "2", "--worker", "subagent general-purpose (default)") self._run("role", "work/a", "Accountable Party With A Long Name", "A") lines = self._out("show") self.assertEqual(lines[2], f"| 1 | - | {slug}… | queued | - | |") self.assertEqual( lines[3], "| 2 | - | work/a: A title far longer than the for… | queued | subagent general-purpos… " "| Accountable Party With … |", ) def test_key_results_gate_done_and_fill_the_okr_column(self): self._run("set", "work/a", "--objective", "Ship A", "--status", "running") self._run("role", "work/a", "Skills", "RA") self._run("kr", "work/a", "add", "tests green") self._run("kr", "work/a", "add", "released") self._run("kr", "work/a", "met", "1") lines = self._out("show", "--wide") self.assertIn("| Ship A (1/2 KR) | running |", lines[2]) self.assertEqual(len(lines), 3) self.assertNotIn("Ship A", self._out("show")[2]) self.assertEqual(self._run("set", "work/a", "--status", "done").returncode, 2) self.assertIn("key result 2 [ ]: released", self._out("get", "work/a")) self._run("kr", "work/a", "met", "2") self.assertEqual(self._run("set", "work/a", "--status", "done").returncode, 0) self.assertEqual(self._out("show"), ["no tasks"]) def test_status_sets_start_and_end(self): self._run("set", "work/a", "--status", "running") self._run("set", "work/a", "--status", "done", "--description", "long text") lines = self._out("get", "work/a") self.assertIn("status: done", lines) self.assertIn("description: long text", lines) self.assertTrue(any(line.startswith("start: 20") for line in lines)) self.assertTrue(any(line.startswith("end: 20") for line in lines)) self._run("set", "work/a", "--status", "running") self.assertFalse(any(line.startswith("end:") for line in self._out("get", "work/a"))) def test_base_names_the_upstream_branch_after_the_branch(self): self._run("set", "work/side", "--branch", "sidetree", "--base", "line") lines = self._out("get", "work/side") self.assertIn("base: line", lines) self.assertEqual(lines.index("base: line"), lines.index("branch: sidetree") + 1) self.assertEqual(json.loads((self.root / "rasci.json").read_text(encoding="utf-8"))["work/side"]["base"], "line") self._run("set", "work/side", "--base", "") self.assertFalse(any(line.startswith("base:") for line in self._out("get", "work/side"))) def _after(self) -> dict[str, list[str]]: rows = json.loads((self.root / "rasci.json").read_text(encoding="utf-8")) return {task: row["after"] for task, row in rows.items() if "after" in row} def test_unpin_drops_the_pin_a_change_stamps_only_its_row_and_the_rank_follows_after(self): rows = lambda: json.loads((self.root / "rasci.json").read_text(encoding="utf-8")) self._out("set", "work/a", "--prio", "1", "--worker", "One") self._out("set", "work/b", "--prio", "2", "--worker", "One") self.assertEqual(self._out("agents")[0].split("\t")[3], "work/a, work/b") self._out("set", "work/a", "--after", "work/b") self.assertEqual(self._out("agents")[0].split("\t")[3], "work/b, work/a") stamped = rows()["work/a"]["updated"] self._out("set", "work/b", "--unpin") self.assertNotIn("prio", rows()["work/b"]) self.assertEqual(rows()["work/a"]["updated"], stamped) self.assertIn("updated", rows()["work/b"]) refused = self._run("set", "work/b", "--unpin", "--prio", "3") self.assertEqual(refused.returncode, 2) self.assertIn("--unpin", refused.stderr) def test_after_stores_and_drops_a_dependency_and_refuses_unknown_tasks_and_cycles(self): for task in ("work/a", "work/b", "work/c"): self._out("set", task) self._out("set", "work/b", "--after", "work/a") self._out("set", "work/c", "--after", "work/b", "--after", "work/a", "--after", "work/b") self.assertEqual(self._after(), {"work/b": ["work/a"], "work/c": ["work/b", "work/a"]}) self.assertIn("after: work/b", self._out("get", "work/c")) before = (self.root / "rasci.json").read_bytes() for args, reason in ( (("set", "work/a", "--after", "work/c"), "closes a cycle"), (("set", "work/a", "--after", "work/b", "--title", "kept out"), "closes a cycle"), (("set", "work/a", "--after", "work/a"), "cannot follow itself"), (("set", "work/a", "--after", "work/nope"), "no task of the register"), (("set", "work/new", "--after", "work/nope"), "no task of the register"), (("set", "work/a", "--after", "Bad Task"), "invalid task"), ): with self.subTest(args=args): result = self._run(*args) self.assertEqual(result.returncode, 2, result.stdout) self.assertIn(reason, result.stderr) self.assertEqual((self.root / "rasci.json").read_bytes(), before) self._out("set", "work/c", "--drop-after", "work/b") self.assertEqual(self._after(), {"work/b": ["work/a"], "work/c": ["work/a"]}) self._out("set", "work/b", "--drop-after", "work/a") self._out("set", "work/a", "--after", "work/b") self.assertEqual(self._after(), {"work/a": ["work/b"], "work/c": ["work/a"]}) def test_the_operator_business_value_replaces_the_median_and_may_stand_alone(self): self._run("estimate", "work/a", "One", "--value", "8", "--complexity", "2") self._run("estimate", "work/a", "Two", "--value", "13", "--complexity", "3") self._out("estimate", "work/a", "Operator", "--value", "34") row = json.loads((self.root / "rasci.json").read_text(encoding="utf-8"))["work/a"] self.assertNotIn("complexity", row["estimates"]["Operator"]) self.assertEqual(self._out("wsjf")[2], "| 1 | work/a | - | 34 | 2.5 | 13.6 | One, Operator, Two | queued |") self.assertIn("estimate Operator: business value 34, complexity -", self._out("get", "work/a")) self._out("estimate", "work/a", "Operator", "--value", "21", "--complexity", "8") self._out("estimate", "work/a", "Operator", "--value", "5") row = json.loads((self.root / "rasci.json").read_text(encoding="utf-8"))["work/a"] self.assertEqual((row["estimates"]["Operator"]["value"], row["estimates"]["Operator"]["complexity"]), (5, 8)) self.assertEqual(self._out("wsjf")[2], "| 1 | work/a | - | 5 | 3 | 1.67 | One, Operator, Two | queued |") self._out("estimate", "work/b", "Operator", "--value", "8") self.assertEqual(self._out("wsjf")[3:], ["| 2 | work/b | - | - | - | - | Operator | queued |", "unestimated: work/b"]) def test_done_tasks_show_only_with_all(self): self._run("set", "work/a", "--status", "done", "--end", "2026-11-01T12:00:00+00:00") self._run("role", "work/a", "Skills", "RA") self.assertEqual(self._out("show"), ["no tasks"]) self.assertEqual(self._out("show", "--wide"), ["no tasks"]) self.assertEqual(self._out("show", "--all")[2], "| 1 | - | work/a | done | - | Skills |") self.assertIn( f"| done | - | - | - | {local('2026-11-01T12:00:00+00:00')} |", self._out("show", "--all", "--wide")[2] ) def test_artifacts_and_tools(self): self._run("set", "work/a", "--workdir", "/w/a", "--branch", "feat/a", "--status", "done", "--artifact", "image:core:dev", "--artifact", "volume:db", "--tool", "council.py", "--tool", "make test") self._run("set", "work/b", "--worker", "Two", "--tool", "make test", "--artifact", "dir:/tmp/b") self._run("set", "work/a", "--drop-artifact", "volume:db") self._run("set", "work/b", "--branch", "") rows = [line.split("\t") for line in self._out("artifacts")] self.assertEqual( [row[:4] for row in rows], [ ["done", "work/a", "-", "workdir:/w/a"], ["done", "work/a", "-", "branch:feat/a"], ["done", "work/a", "-", "image:core:dev"], ["queued", "work/b", "Two", "dir:/tmp/b"], ], ) self.assertEqual(rows[0][4], "-") self.assertRegex(rows[2][4], r"^\d\d-\d\d \d\d:\d\d$") self.assertEqual([r.split("\t")[1] for r in self._out("artifacts", "--ended")], ["work/a"] * 3) self.assertEqual(len(self._out("artifacts", "--task", "work/b")), 1) self.assertIn("| queued | Two | 1 |", self._out("show", "--wide")[2]) self.assertTrue(any(line.startswith("artifact: image:core:dev (since ") for line in self._out("get", "work/a"))) self.assertEqual( self._out("tools"), ["2\tmake test\twork/a work/b", "1\tcouncil.py\twork/a"] ) self.assertIn("tool: council.py", self._out("get", "work/a")) def test_claims_join_the_matrix_with_holder_as_ra(self): self._claim("work/live", "s1", "One", "Gamma") self._claim("cluster/dead", "s9", "Gone", "Delta") self._claim("chair", "s1", "One", "allocator") self._run("set", "work/live", "--prio", "1", "--status", "running") self._run("role", "work/live", "Two", "S") lines = self._out("show", "--wide") self.assertTrue(lines[2].startswith("| 1 | - | work/live | - | running | One |")) self.assertTrue(lines[2].endswith("| One | One | Two | | |")) self.assertTrue(lines[3].startswith("| 2 | - | cluster/dead | - | stale | Gone |")) self.assertFalse(any("chair" in line for line in lines)) lines = self._out("show") self.assertEqual(lines[2], "| 1 | - | work/live | running | One | One |") self.assertEqual(lines[3], "| 2 | - | cluster/dead | stale | Gone | Gone |") self.assertFalse(any("chair" in line for line in lines)) def test_council_claim_takeover_and_release_write_through(self): self.assertEqual(self._run("claim", "work/x", "--work", "Do X", script=COUNCIL).returncode, 0) lines = self._out("get", "work/x") self.assertIn("status: running", lines) self.assertIn("worker: One", lines) self.assertIn("description: Do X", lines) self.assertIn(f"workdir: {self.tmp.name}", lines) self.assertIn("roles: One=RA", lines) (self.sessions / "s1.json").unlink() self.assertEqual(self._run("takeover", "work/x", "--work", "Do X", script=COUNCIL, sid="s2").returncode, 0) lines = self._out("get", "work/x") self.assertIn("worker: Two", lines) self.assertIn("roles: Two=RA", lines) self._run("kr", "work/x", "add", "X shipped") self.assertEqual(self._run("release", "work/x", script=COUNCIL, sid="s2").returncode, 0) lines = self._out("get", "work/x") self.assertIn("status: stopped", lines) self.assertTrue(any(line.startswith("end: 20") for line in lines)) def test_rm_and_clear_a_party(self): self._run("set", "work/a", "--title", "Alpha") self._run("role", "work/a", "Skills", "RA") self._run("role", "work/a", "Pkgmgr", "C") self._run("role", "work/a", "Pkgmgr", "-") self.assertTrue(self._out("show", "--wide")[2].endswith("| Skills | Skills | | | |")) self.assertEqual(self._out("rm", "work/a"), ["removed work/a"]) self.assertEqual(self._out("show"), ["no tasks"]) def test_invalid_input_exits_2(self): for args in ( ("role", "work/a", "Skills", "RR"), ("role", "work/a", "Skills", "Q"), ("set", "Bad Task"), ("set", "work/a", "--prio", "0"), ("set", "work/a", "--status", "finished"), ("set", "work/a", "--start", "yesterday"), ("get", "work/none"), ("kr", "work/a", "met", "1"), ("set", "work/a", "--artifact", "no-kind"), ("estimate", "work/a", "One", "--value", "4", "--complexity", "2"), ("estimate", "work/a", "One", "--value", "8"), ("estimate", "work/a", "Operator", "--complexity", "8"), ("estimate", "work/a", "Operator", "--value", "4"), ): with self.subTest(args=args): self.assertEqual(self._run(*args).returncode, 2) self.assertFalse((self.root / "rasci.json").exists()) def test_pipe_in_a_title_does_not_break_the_table(self): self._run("set", "work/a", "--title", "a | b") self._run("role", "work/a", "Skills", "RA") self.assertTrue(self._out("show", "--wide")[2].startswith("| 1 | - | work/a: a \\| b | - | queued |")) self.assertEqual(self._out("show")[2], "| 1 | - | work/a: a \\| b | queued | - | Skills |") def test_wsjf_ranks_by_median_estimates(self): self._run("set", "work/a", "--objective", "Ship A") self._run("kr", "work/a", "add", "released") self._run("estimate", "work/a", "One", "--value", "8", "--complexity", "2") self._run("estimate", "work/a", "Two", "--value", "13", "--complexity", "3") self._run("estimate", "work/b", "One", "--value", "5", "--complexity", "5") self._run("set", "work/c", "--title", "Gamma") lines = self._out("wsjf") self.assertEqual( lines[0], "| Rank | Task | OKR | Business value | Complexity | WSJF | Estimators | Status |" ) self.assertEqual(lines[2], "| 1 | work/a | Ship A (0/1 KR) | 10.5 | 2.5 | 4.2 | One, Two | queued |") self.assertEqual(lines[3], "| 2 | work/b | - | 5 | 5 | 1 | One | queued |") self.assertEqual(lines[4], "| 3 | work/c | - | - | - | - | - | queued |") self.assertEqual(lines[5], "unestimated: work/c") self.assertTrue(self._out("show")[2].startswith("| 1 | 4.2 | work/a |")) self.assertIn("wsjf: 4.2", self._out("get", "work/a")) self._run("set", "work/c", "--prio", "1") self.assertTrue(self._out("show")[2].startswith("| 1 | - | work/c: Gamma |")) self._run("estimate", "work/b", "One", "--drop") self.assertFalse(any(line.startswith("estimate ") for line in self._out("get", "work/b"))) def test_hotfix_and_operator_rows_are_no_work_tasks_for_gaps_and_wsjf(self): self._run("set", "work/a") self._run("role", "work/a", "Skills", "RA") self._run("kr", "work/a", "add", "released") self._run("set", "hotfix/db", "--status", "running") self._run("set", "operator/push", "--status", "running") shown = self._out("show") self.assertEqual([line.split(" | ")[2] for line in shown[2:]], ["hotfix/db", "operator/push", "work/a"]) self.assertFalse(any(line.startswith("gaps:") for line in shown)) self.assertEqual( self._out("wsjf")[2:], ["| 1 | work/a | - (0/1 KR) | - | - | - | - | queued |", "unestimated: work/a"] ) self._run("set", "work/b", "--title", "Beta") self.assertEqual(self._out("show")[-1], "gaps: work/b has 0 A; work/b has no R; work/b has no key result") self._run("set", "hotfix/db", "--status", "done", "--branch", "fix/db", "--workdir", "/w/db") self._run("set", "operator/push", "--branch", "main", "--artifact", "dir:/tmp/push") self._run("set", "work/a", "--branch", "feat/a") for args in (("artifacts",), ("artifacts", "--ended"), ("artifacts", "--task", "hotfix/db")): with self.subTest(args=args): self.assertEqual([line.split("\t")[1] for line in self._out(*args)], ["work/a"] if args == ("artifacts",) else []) def _ranked(self) -> list[str]: return [line.split(" | ")[2].split(":")[0] for line in self._out("show")[2:] if line.startswith("| ")] def test_agents_order_breaks_ties_only_and_never_touches_the_task_pins(self): self._run("set", "work/a1", "--worker", "Alpha", "--prio", "1") self._run("set", "work/b1", "--worker", "Beta", "--prio", "1") self._run("set", "work/b2", "--worker", "Beta", "--prio", "2") self._run("set", "work/c1", "--worker", "Gamma") self._run("set", "work/d1", "--worker", "Delta") self._claim("security/gui", "s9", "Epsilon", "line") self.assertEqual( self._out("agents"), ["1\tAlpha\t1\twork/a1", "2\tBeta\t2\twork/b1, work/b2", "3\tGamma\t1\twork/c1", "4\tDelta\t1\twork/d1", "5\tEpsilon\t0\t-"], ) self.assertEqual(self._ranked(), ["work/a1", "work/b1", "work/b2", "work/c1", "work/d1"]) register = (self.root / "rasci.json").read_text(encoding="utf-8") self.assertEqual( self._out("agents", "order", "Gamma, Beta,Epsilon"), ["agent order: Gamma, Beta, Epsilon; the others follow by register rank; task --prio pins unchanged"], ) self.assertEqual((self.root / "rasci.json").read_text(encoding="utf-8"), register) stored = json.loads((self.root / "agent-order.json").read_text(encoding="utf-8")) self.assertEqual(stored["agents"], ["Gamma", "Beta", "Epsilon"]) self.assertEqual( self._out("agents"), ["1\tGamma\t1\twork/c1", "2\tBeta\t2\twork/b1, work/b2", "3\tEpsilon\t0\t-", "4\tAlpha\t1\twork/a1", "5\tDelta\t1\twork/d1"], ) self.assertEqual(self._ranked(), ["work/b1", "work/a1", "work/b2", "work/c1", "work/d1"]) self.assertIn("prio: 1", self._out("get", "work/a1")) for args in (("agents", "order", "Gamma,Nobody"), ("agents", "order", "Beta,Beta"), ("agents", "order", " , "), ("agents", "sort")): with self.subTest(args=args): self.assertEqual(self._run(*args).returncode, 2) self.assertEqual(json.loads((self.root / "agent-order.json").read_text(encoding="utf-8")), stored) def test_the_old_order_command_stores_the_agent_order_and_says_so(self): self._run("set", "work/a1", "--worker", "Alpha", "--prio", "3") self._run("set", "work/b1", "--worker", "Beta") done = self._run("order", "Beta,Alpha") self.assertEqual(done.returncode, 0, done.stderr) self.assertIn("rasci: `order` is now `agents order`", done.stderr) self.assertIn("no longer writes --prio pins", done.stderr) self.assertEqual(json.loads((self.root / "agent-order.json").read_text(encoding="utf-8"))["agents"], ["Beta", "Alpha"]) self.assertIn("prio: 3", self._out("get", "work/a1")) self.assertNotIn("prio: 1", self._out("get", "work/b1")) def test_a_broken_agent_order_file_ranks_like_none(self): self._run("set", "work/a1", "--worker", "Alpha") self._run("set", "work/b1", "--worker", "Beta") for text in ("{not json", json.dumps({"agents": "Beta"}), json.dumps({"agents": [1, "Beta"]}), json.dumps([1])): with self.subTest(text=text): (self.root / "agent-order.json").write_text(text, encoding="utf-8") self.assertEqual(self._ranked(), ["work/a1", "work/b1"]) def test_the_agent_order_never_reads_or_writes_through_a_link_fifo_or_deep_nesting(self): self._run("set", "work/a1", "--worker", "Alpha") self._run("set", "work/b1", "--worker", "Beta") outside = Path(self.tmp.name) / "outside.json" outside.write_text(json.dumps({"agents": ["Beta"]}), encoding="utf-8") order = self.root / "agent-order.json" order.symlink_to(outside) self.assertEqual(self._ranked(), ["work/a1", "work/b1"]) order.unlink() order.write_text("[" * 100000, encoding="utf-8") self.assertEqual(self._ranked(), ["work/a1", "work/b1"]) order.unlink() os.mkfifo(order, 0o600) self.assertEqual(self._run("agents", timeout=20).stdout.splitlines(), ["1\tAlpha\t1\twork/a1", "2\tBeta\t1\twork/b1"]) order.unlink() lock = self.root / "rasci.lock" lock.unlink(missing_ok=True) planted = Path(self.tmp.name) / "planted.lock" lock.symlink_to(planted) refused = self._run("agents", "order", "Beta,Alpha") self.assertEqual(refused.returncode, 2, refused.stderr) self.assertIn("agents order", refused.stderr) self.assertFalse(planted.exists() or order.exists()) def test_a_symlinked_register_is_refused_and_never_copied_into_the_council(self): outside = Path(self.tmp.name) / "outside.json" outside.write_text(json.dumps({"outside": {"title": "marker-outside"}}), encoding="utf-8") self.root.mkdir() (self.root / "rasci.json").symlink_to(outside) for args in (("show", "--all"), ("set", "work/a", "--title", "A")): with self.subTest(args=args): result = self._run(*args, timeout=60) self.assertEqual(result.returncode, 2, result.stderr) self.assertNotIn("outside", result.stdout) self.assertTrue((self.root / "rasci.json").is_symlink()) copies = [p.name for p in self.root.iterdir() if p.is_file() and not p.is_symlink() and "marker-outside" in p.read_text(encoding="utf-8")] self.assertEqual(copies, []) def test_a_hard_linked_register_is_refused(self): outside = Path(self.tmp.name) / "outside.json" outside.write_text(json.dumps({"outside": {"title": "marker-outside"}}), encoding="utf-8") self.root.mkdir() os.link(outside, self.root / "rasci.json") result = self._run("show", "--all", timeout=60) self.assertEqual(result.returncode, 2, result.stderr) self.assertNotIn("outside", result.stdout) def test_a_symlinked_lock_is_refused_and_creates_nothing_at_its_target(self): aimed = Path(self.tmp.name) / "elsewhere" / "aimed" aimed.parent.mkdir() self.root.mkdir() (self.root / "rasci.lock").symlink_to(aimed) result = self._run("set", "work/a", "--title", "A", timeout=60) self.assertEqual(result.returncode, 2, result.stderr) self.assertFalse(aimed.exists()) self.assertFalse((self.root / "rasci.json").exists()) def test_a_fifo_register_fails_instead_of_hanging(self): self.root.mkdir() os.mkfifo(self.root / "rasci.json") for args in (("show",), ("set", "work/a")): with self.subTest(args=args): self.assertEqual(self._run(*args, timeout=20).returncode, 2) def test_a_register_that_is_no_object_of_tasks_exits_2(self): self.root.mkdir() for text in ("[1, 2]", json.dumps({"work/a": 3}), "[" * 100000): with self.subTest(text=text[:12]): (self.root / "rasci.json").write_text(text, encoding="utf-8") result = self._run("show", "--all", timeout=60) self.assertEqual(result.returncode, 2, result.stderr) self.assertIn("rasci:", result.stderr) class TestOrgSkillsUseTheRegister(unittest.TestCase): def test_every_org_skill_names_the_register(self): for skill in sorted((REPO_ROOT / "skills").glob("kvb-org-*/SKILL.md")): if skill.parent.name == "kvb-org-rasci": continue with self.subTest(skill=skill.parent.name): self.assertIn("`kvb-org-rasci`", skill.read_text(encoding="utf-8")) if __name__ == "__main__": unittest.main()