forked from Keelrleek/Harbor
KeelrLeek
dbc0c45ef2
Kimi's agent-core reads a top-level {"decision": "block"} as allow, so
Harbor's keep-going never took effect for Kimi; it honours exit code 2
with the reason on stderr (its hooks run through cmd.exe), once per turn.
Tool-call denials keep the JSON form, which Kimi does honour.
Also: attention skips stopped agents and closed items; file scopes of
function files carry their address.
390 lines
20 KiB
Python
390 lines
20 KiB
Python
"""Regression tests for the correctness review of rounds five and six."""
|
|
import json
|
|
import os
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import threading
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
os.environ["HARBOR_NO_TOASTS"] = "1"
|
|
from helpers import ROOT, HarborCase, make_home, run_module, temp_dir
|
|
from harbor import notify, projects
|
|
from harbor.core import Harbor, HarborError
|
|
|
|
|
|
class StopState(HarborCase):
|
|
def test_02_trial_for_one_agent_while_everyone_is_stopped(self):
|
|
self.h.set_flag("stop:all", "on", "User", "web")
|
|
self.clock.t += 60
|
|
self.h.set_flag("stop:Codex", "off", "User", "web", until="30m")
|
|
self.assertEqual((self.h.stopped("Codex"), bool(self.h.trial("Codex"))), ([], True))
|
|
self.assertTrue(self.h.stopped("Kimi"))
|
|
self.clock.t += 60
|
|
self.h.set_flag("stop:all", "on", "Claude", "cli") # an agent's re-stop changes nothing
|
|
self.assertFalse(self.h.stopped("Codex"))
|
|
self.h.set_flag("stop:all", "on", "User", "web") # the human pressing it again is a
|
|
self.assertTrue(self.h.stopped("Codex")) # new decision: it covers everyone
|
|
|
|
def test_03_coordinator_stop_and_lift_keep_the_trial_end(self):
|
|
self.h.set_flag("stop:Codex", "on", "User", "web")
|
|
self.h.set_flag("stop:Codex", "off", "User", "web", until="30m")
|
|
self.h.set_flag("stop:Codex", "on", "Claude", "cli")
|
|
self.h.set_flag("stop:Codex", "off", "Claude", "cli")
|
|
self.assertIsNotNone(self.h.trial("Codex"))
|
|
self.clock.t += 31 * 60
|
|
self.assertTrue(self.h.stopped("Codex")) # the trial still ends
|
|
|
|
def test_04_ended_trial_is_a_stop_everywhere(self):
|
|
(self.dir / "x").mkdir()
|
|
self.h.set_flag("stop:Codex", "on", "User", "web")
|
|
self.h.set_flag("stop:Codex", "off", "User", "web", until="1m")
|
|
self.clock.t += 120
|
|
self.assertTrue(self.h.stopped("Codex"))
|
|
self.assertIn("trial run ended", self.h.stopped("Codex")[0]["reason"])
|
|
launch = {a["agent"]: a for a in self.h.launch_status()}
|
|
self.assertEqual(launch["Codex"]["state"], "stopped")
|
|
|
|
def test_07_launch_shows_team_trials(self):
|
|
self.h.set_flag("stop:all", "on", "User", "web")
|
|
self.clock.t += 60
|
|
self.h.set_flag("stop:all", "off", "User", "web", until="15m")
|
|
self.assertEqual({a["agent"]: a for a in self.h.launch_status()}["Kimi"]["state"], "trial")
|
|
|
|
|
|
class RollCalls(HarborCase):
|
|
def test_06_second_roll_call_keeps_the_first(self):
|
|
self.h.rollcall("User", "web")
|
|
codex_q = next(q for q in self.h.questions(to="Codex"))
|
|
self.h.answer("Codex", codex_q["id"], "Here")
|
|
self.h.rollcall("User", "web", agents=["Kimi"])
|
|
rows = {r["agent"]: r for r in self.h.rollcall_status()["agents"]}
|
|
self.assertEqual(rows["Codex"]["answer"], "Here")
|
|
self.assertEqual(len(rows), 5)
|
|
self.assertEqual(len([q for q in self.h.questions(to="Kimi")]), 1) # the older one is closed
|
|
|
|
|
|
class Notifications(HarborCase):
|
|
def test_05_backlog_over_fifty(self):
|
|
notify.set_enabled(self.h, True)
|
|
for i in range(60):
|
|
self.h.send("Codex", "All", f"broadcast {i}")
|
|
self.h.send("Codex", ["User"], "the one that matters")
|
|
pending, top = notify.pending_for_human(self.h)
|
|
self.assertEqual([m["body"] for m in pending], ["the one that matters"])
|
|
|
|
|
|
class FlowFixes(unittest.TestCase):
|
|
def setUp(self):
|
|
self.dir = temp_dir()
|
|
|
|
def tearDown(self):
|
|
shutil.rmtree(self.dir, ignore_errors=True)
|
|
|
|
def project(self, **extra):
|
|
data = {"name": "f", "coordinators": ["Lead"], "database": str(self.dir / "f.db"),
|
|
"agents": [{"name": "Lead"}, {"name": "Ann"}, {"name": "Bob"}], **extra}
|
|
return projects.project_from_dict(data)
|
|
|
|
def test_08_09_bulk_and_templates_by_a_worker(self):
|
|
h = Harbor(self.project(templates={"t": {"steps": [{"name": "a", "title": "Fix {name} in {scope}",
|
|
"assignee": "Bob"}]}}))
|
|
h.claim("Lead", "range:0x440000-0x47FFFF", kind="band", on_behalf_of="Bob")
|
|
r = h.bulk_tasks("Ann", "0x445310 One\n0x445320 Two")
|
|
self.assertEqual([t["assignee"] for t in r["created"]], [None, None]) # no routing to someone else
|
|
with self.assertRaises(HarborError):
|
|
h.run_template("Ann", "t", scope="fn:0x445400")
|
|
self.assertEqual(len(h.tasks()), 2) # nothing half-created
|
|
task = h.run_template("Lead", "t", scope="fn:0x445400")[0]
|
|
self.assertEqual(task["title"], "Fix {name} in fn:0x445400")
|
|
h.close()
|
|
|
|
def test_11_policy_checked_at_load(self):
|
|
p = self.project(policy={"land_requires": "Ann"})
|
|
self.assertEqual(p.policy["land_requires"], ["Ann"])
|
|
for bad in ({"land_requires": ["Ghost"]}, {"all_per_hour": "lots"}):
|
|
with self.assertRaises(projects.ProjectError):
|
|
self.project(policy=bad)
|
|
h = Harbor(self.project(policy={"land_requires": ["Ann"]}))
|
|
item = h.submit("Ann", "x", commit="abc")
|
|
h.transition("Lead", item["id"], "approved")
|
|
h.transition("Lead", item["id"], "landed", commit="b4") # the author is not required
|
|
h.close()
|
|
|
|
def test_10_rebase_does_not_look_new(self):
|
|
env = {**os.environ, "GIT_AUTHOR_NAME": "t", "GIT_AUTHOR_EMAIL": "t@x", "GIT_COMMITTER_NAME": "t",
|
|
"GIT_COMMITTER_EMAIL": "t@x", "GIT_AUTHOR_DATE": "2026-01-01T10:00:00", "GIT_COMMITTER_DATE": "2026-01-01T10:00:00"}
|
|
repo = self.dir / "r"
|
|
repo.mkdir()
|
|
g = lambda *a, e=env: subprocess.run(["git", "-C", str(repo), *a], check=True, capture_output=True, env=e)
|
|
g("init", "-q", "-b", "main")
|
|
(repo / "a").write_text("a")
|
|
g("add", "a")
|
|
g("commit", "-q", "-m", "base")
|
|
g("checkout", "-q", "-b", "work")
|
|
(repo / "b").write_text("b")
|
|
g("add", "b")
|
|
g("commit", "-q", "-m", "draft")
|
|
h = Harbor(projects.project_from_dict({"name": "g", "database": str(self.dir / "g.db"),
|
|
"agents": [{"name": "Ann", "workdir": str(repo)}]}))
|
|
h.clock = lambda: 1767268800.0 + 3600 * 5 # 2026-01-01 15:00 UTC
|
|
sha = h.git_status("Ann", max_age=0)["last_commit"]["sha"]
|
|
h.submit("Ann", "draft", commit=sha, reviewers=["User"])
|
|
h.clock = lambda: 1767268800.0 + 3600 * 6
|
|
g("checkout", "-q", "main")
|
|
(repo / "c").write_text("c")
|
|
g("add", "c")
|
|
g("commit", "-q", "-m", "main moves", e={**env, "GIT_AUTHOR_DATE": "2026-01-01T16:00:00",
|
|
"GIT_COMMITTER_DATE": "2026-01-01T16:00:00"})
|
|
g("checkout", "-q", "work")
|
|
g("rebase", "-q", "main", e={**env, "GIT_COMMITTER_DATE": "2026-01-01T16:30:00"})
|
|
self.assertEqual(h.git_status("Ann", max_age=0)["unsubmitted"], [])
|
|
h.close()
|
|
|
|
|
|
class Pages(unittest.TestCase):
|
|
def test_01_no_code_hidden_behind_a_line_comment(self):
|
|
for page in ("app.html",):
|
|
scripts = re.findall(r"<script>([\s\S]*?)</script>", (ROOT / "harbor" / "web" / page).read_text(encoding="utf-8"))
|
|
self.assertTrue(scripts, page)
|
|
for line in "\n".join(scripts).splitlines():
|
|
code = re.sub(r"'(?:\\.|[^'\\])*'|\"(?:\\.|[^\"\\])*\"|`(?:\\.|[^`\\])*`|/(?:\\.|[^/\\\n])+/[gimsuy]*", "", line)
|
|
if "//" in code:
|
|
after = code.split("//", 1)[1]
|
|
self.assertFalse(re.search(r"\b(const|let|var|function|if|return)\b", after), (page, line[:120]))
|
|
|
|
|
|
class Race(HarborCase):
|
|
def test_answer_race_for_real(self):
|
|
q = self.h.ask("Codex", "Ship?", ["Yes", "No"], fallback="Claude")
|
|
self.h.set_presence("User", "web", "away")
|
|
results, gate = [], threading.Barrier(2)
|
|
|
|
def answer(who, value, origin):
|
|
h = Harbor(self.project, clock=self.clock)
|
|
try:
|
|
gate.wait()
|
|
results.append(h.answer(who, q["id"], value, origin=origin)["answer"])
|
|
except HarborError as error:
|
|
results.append(f"refused: {error}")
|
|
finally:
|
|
h.close()
|
|
threads = [threading.Thread(target=answer, args=a) for a in (("Claude", "Yes", "cli"), ("User", "No", "web"))]
|
|
[t.start() for t in threads]
|
|
[t.join() for t in threads]
|
|
self.assertEqual(sum(1 for r in results if not r.startswith("refused")), 1, results)
|
|
self.assertEqual(len([m for m in self.h.inbox("Codex", unread_only=False) if m["kind"] == "answer"]), 1)
|
|
|
|
|
|
class LaunchFindings(unittest.TestCase):
|
|
"""What the first launch with the real agent apps showed."""
|
|
|
|
def gate(self, command=None, tool="shell"):
|
|
from harbor import hooks
|
|
payload = {"tool_name": tool}
|
|
if command is not None:
|
|
payload["tool_input"] = {"command": command}
|
|
return hooks._tool_is_harbor(payload)
|
|
|
|
def test_messages_may_hold_shell_characters_inside_double_quotes(self):
|
|
run = "py -3.12 -B R:/Workspace/Tools/Harbor/run.py --as Kimi "
|
|
for c in [run + 'send Claude --thread launch "Kimi is online (through the command; MCP was blocked) & ready"',
|
|
'harbor ack 1708 --note "ok, I\'m stopped <as asked> {no work} @Claude ^"',
|
|
'harbor send Claude "it\u2019s fine"', "harbor send Claude 'plain words'"]:
|
|
self.assertTrue(self.gate(c), c)
|
|
for c in ['harbor send All "$(id)"', 'harbor send All "a" ; rm x', "harbor send All '\"'; rm x; '\"'",
|
|
'harbor send All "x\\" ; rm y ; \\"z"', 'harbor send All "100%"', 'harbor send All "line\nRemove-Item x"',
|
|
'harbor send All "a\u201d ; rm x ; \u201cb"', "harbor send All 'it (x)'", 'harbor send All "unterminated',
|
|
'harbor send All "`whoami`"', 'harbor send All \\"x\\"']:
|
|
self.assertFalse(self.gate(c), c)
|
|
|
|
def test_every_hosts_harbor_tool_names_and_tool_loaders(self):
|
|
for name in ("mcp__plugin-harbor-open-seas_harbor__send", "mcp__harbor-open-seas__brief", "harbor-open-seas.inbox",
|
|
"mcp__harbor__wait_for_messages", "harbor_ack", "select_tools", "ToolSearch"):
|
|
self.assertTrue(self.gate(tool=name), name)
|
|
for name in ("mcp__plugin-harbor-open-seas_harbor__land", "mcp__plugin-harbor-open-seas_harbor__next_task",
|
|
"mcp__other__send", "my_harbor_tool", "Harbor", "mcp__plugin-harbor-open-seas_harbor__resume"):
|
|
self.assertFalse(self.gate(tool=name), name)
|
|
|
|
def test_the_stop_message_says_sending_is_allowed(self):
|
|
from harbor import hooks
|
|
|
|
class FakeHarbor:
|
|
human = "User"
|
|
text = hooks._stop_text(FakeHarbor(), "Kimi", [{"key": "stop:Kimi", "set_by": "User", "set_at": "now", "reason": None}])
|
|
self.assertIn("send", text)
|
|
self.assertIn("double quotes", text)
|
|
|
|
|
|
class AllocationWhileStopped(HarborCase):
|
|
def test_coordinator_records_bands_for_stopped_agents_but_no_work(self):
|
|
self.h.set_flag("stop:Codex", "on", "User", "web")
|
|
self.h.claim("Claude", "range:0x401000-0x43FFFF", kind="band", on_behalf_of="Codex")
|
|
self.h.claim("Claude", "fn:0x401200", kind="naming", on_behalf_of="Codex")
|
|
with self.assertRaises(HarborError):
|
|
self.h.claim("Claude", "fn:0x401300", kind="draft", on_behalf_of="Codex")
|
|
with self.assertRaises(HarborError):
|
|
self.h.claim("Codex", "fn:0x401400")
|
|
|
|
|
|
class BulkNotice(HarborCase):
|
|
def test_bulk_sends_one_notice_per_assignee(self):
|
|
self.h.claim("Claude", "range:0x401000-0x43FFFF", kind="band", on_behalf_of="Codex")
|
|
before = len(self.h.inbox("Codex", unread_only=False, limit=500, mark_read=False))
|
|
r = self.h.bulk_tasks("Claude", "0x401100 One\n0x401200 Two\n0x401300 Three")
|
|
self.assertEqual([t["assignee"] for t in r["created"]], ["Codex"] * 3)
|
|
new = self.h.inbox("Codex", unread_only=False, limit=500, mark_read=False)[before:]
|
|
self.assertEqual(len(new), 1)
|
|
self.assertIn("3 new tasks", new[0]["body"])
|
|
|
|
|
|
class Autopilot(HarborCase):
|
|
def setUp(self):
|
|
super().setUp()
|
|
self.h.project.autopilot = True
|
|
|
|
def test_running_agents_keep_going_and_stopped_or_stuck_ones_rest(self):
|
|
self.h.claim("Claude", "range:0x401000-0x43FFFF", kind="band", on_behalf_of="Codex")
|
|
t = self.h.add_task("Claude", "Make 0x401100 readable", scope="fn:0x401100", assignee="Codex")
|
|
self.assertIn(f"#{t['id']}", self.h.continuation("Codex")) # offers the next task
|
|
self.h.take_task("Codex", t["id"])
|
|
self.assertIn(f"Continue task #{t['id']}", self.h.continuation("Codex"))
|
|
self.assertIsNotNone(self.h.continuation("Codex"))
|
|
self.assertIsNone(self.h.continuation("Codex")) # three idle turn ends: rest
|
|
self.assertEqual([a["agent"] for a in self.h.attention() if a["kind"] == "resting"], ["Codex"])
|
|
self.h.activity("Codex", "Edit", "codex", target="src/x.c") # real work resets the guard
|
|
self.assertIsNotNone(self.h.continuation("Codex"))
|
|
self.assertEqual([a for a in self.h.attention() if a["kind"] == "resting"], [])
|
|
self.h.set_flag("stop:Codex", "on", "User", "web")
|
|
self.assertIsNone(self.h.continuation("Codex")) # the human's stop rests it
|
|
|
|
def test_band_fallback_and_reviewer_work(self):
|
|
self.h.claim("Claude", "range:0x401000-0x43FFFF", kind="band", on_behalf_of="Codex")
|
|
self.assertIn("own band", self.h.continuation("Codex"))
|
|
self.assertIn("wait_for_messages", self.h.continuation("Kimi"))
|
|
item = self.h.submit("Codex", "Readable 0x401100", commit="abc123")
|
|
self.assertIn(f"#{item['id']}", self.h.continuation("Kimi"))
|
|
|
|
|
|
class LandMany(HarborCase):
|
|
def test_batch_landing_sends_one_message_per_author_and_one_win(self):
|
|
ids = [self.h.submit("Codex", f"Readable 0x40{i}000", commit=f"c{i}")["id"] for i in range(3)]
|
|
before = self.h.conn.execute("SELECT COUNT(*) FROM messages").fetchone()[0]
|
|
landed = self.h.land_many("Claude", ids, "abc1234", note="readable check passed")
|
|
self.assertEqual([i["state"] for i in landed], ["landed"] * 3)
|
|
self.assertEqual(self.h.conn.execute("SELECT COUNT(*) FROM messages").fetchone()[0] - before, 2)
|
|
with self.assertRaises(HarborError):
|
|
self.h.land_many("Codex", ids, "abc1234")
|
|
|
|
|
|
class AuthorRefreshesCommit(HarborCase):
|
|
def test_author_hands_in_a_newer_commit_while_submitted_or_approved(self):
|
|
item = self.h.submit("Codex", "Readable 0x401000", commit="c1")
|
|
self.h.transition("Codex", item["id"], "submitted", note="newer", commit="c2")
|
|
self.assertEqual(self.h.item(item["id"])[0]["commit_hash"], "c2")
|
|
self.h.transition("Kimi", item["id"], "approved")
|
|
self.h.transition("Codex", item["id"], "submitted", note="one more fix", commit="c3")
|
|
self.assertEqual(self.h.item(item["id"])[0]["state"], "submitted") # back for review with c3
|
|
with self.assertRaises(HarborError):
|
|
self.h.transition("Codex", item["id"], "submitted") # no commit: nothing to refresh
|
|
with self.assertRaises(HarborError):
|
|
self.h.transition("Kimi", item["id"], "submitted", commit="c4") # only the author
|
|
|
|
|
|
class ProjectRefresh(unittest.TestCase):
|
|
def test_edits_to_the_project_file_apply_without_a_restart(self):
|
|
d = temp_dir()
|
|
try:
|
|
path = d / "p.toml"
|
|
body = 'name = "p"\ndatabase = \'{db}\'\nreviewers = ["Ann"]\n[[agents]]\nname = "Ann"\n[[agents]]\nname = "Bob"\n'
|
|
path.write_text(body.format(db=d / "p.db"), encoding="utf-8")
|
|
h = Harbor(projects.load_file(path))
|
|
self.assertFalse(h.refresh_project())
|
|
path.write_text(body.format(db=d / "p.db").replace('reviewers = ["Ann"]', 'reviewers = ["Bob"]\nautopilot = true'),
|
|
encoding="utf-8")
|
|
os.utime(path, (os.path.getmtime(path) + 5,) * 2)
|
|
self.assertTrue(h.refresh_project())
|
|
self.assertEqual((h.project.reviewers, h.project.autopilot), (("Bob",), True))
|
|
path.write_text("this is not toml [", encoding="utf-8")
|
|
os.utime(path, (os.path.getmtime(path) + 10,) * 2)
|
|
self.assertFalse(h.refresh_project()) # keeps the last good one
|
|
self.assertEqual(h.project.reviewers, ("Bob",))
|
|
h.close()
|
|
finally:
|
|
shutil.rmtree(d, ignore_errors=True)
|
|
|
|
|
|
class SharedFiles(HarborCase):
|
|
def test_shared_annotation_files_are_not_collisions(self):
|
|
self.h.project.shared_files = ("annotations/*.csv", "annotations/prototypes.h")
|
|
self.h.record_touches("Codex", ["annotations/names.csv", "annotations/prototypes.h", "src/a.c"])
|
|
cols = self.h.record_touches("Kimi", ["annotations/names.csv", "annotations/prototypes.h", "src/a.c"])
|
|
self.assertEqual([c["path"] for c in cols], ["src/a.c"])
|
|
self.assertEqual([c["path"] for c in self.h.recent_collisions()], ["src/a.c"])
|
|
|
|
|
|
class WorkingWithoutHooks(HarborCase):
|
|
def test_harbor_calls_and_file_scans_count_as_work_but_not_as_hooks(self):
|
|
self.h.activity("DeepSeek", "harbor submit", "deepseek", channel="mcp")
|
|
row = next(a for a in self.h.agents() if a["name"] == "DeepSeek")
|
|
self.assertTrue(self.h.working(row))
|
|
self.assertIsNone({r["agent"]: r for r in self.h.launch_status()}["DeepSeek"]["hooks"])
|
|
self.h.activity("DeepSeek", "edit", "", target="src/x.c", channel="files")
|
|
self.assertIn("files", {c["channel"] for c in self.h.connections()["DeepSeek"]})
|
|
|
|
|
|
class UnreadQuestions(HarborCase):
|
|
def test_unread_questions_to_an_agent_need_attention_until_read(self):
|
|
q = self.h.send("DeepSeek", ["Claude"], "May I take the 0x4C slice?", kind="question")
|
|
self.h.send("DeepSeek", ["All"], "status for everyone", kind="request")
|
|
kinds = lambda: [a for a in self.h.attention() if a["kind"] == "unread_questions"]
|
|
self.assertEqual(kinds(), []) # not yet 20 minutes old
|
|
self.clock.t += 25 * 60
|
|
found = kinds()
|
|
self.assertEqual([(a["agent"], f"#{q['id']}" in a["text"]) for a in found], [("Claude", True)])
|
|
self.h.inbox("Claude") # reading it clears the item
|
|
self.assertEqual(kinds(), [])
|
|
|
|
|
|
class AttentionWithoutFalseAlarms(HarborCase):
|
|
def test_stopped_agents_owe_no_acks_and_file_scopes_carry_their_address(self):
|
|
from harbor.util import parse_scope
|
|
self.assertEqual(parse_scope("file:src/game/_unassigned/4C6540_UnkPediaTextHelper.c")[1:], (0x4C6540, 0x4C6540))
|
|
self.assertEqual(parse_scope("file:annotations/names.csv")[1:], (None, None))
|
|
m = self.h.send("Codex", ["Kimi"], "please confirm", ack_required=True)
|
|
self.clock.t += 120 * 60
|
|
self.assertTrue(any(a["kind"] == "ack_overdue" for a in self.h.attention()))
|
|
self.h.set_flag("stop:Kimi", "on", "Claude", "cli")
|
|
self.assertFalse(any(a["kind"] == "ack_overdue" and f"#{m['id']}" in a["text"] for a in self.h.attention()))
|
|
|
|
|
|
class KimiStopKeepGoing(unittest.TestCase):
|
|
def test_kimi_gets_exit_2_and_stderr_others_get_json(self):
|
|
import contextlib, io
|
|
from harbor import hooks
|
|
out, err = io.StringIO(), io.StringIO()
|
|
with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err):
|
|
code = hooks._keep_going("kimi", "keep going")
|
|
self.assertEqual((code, out.getvalue(), err.getvalue().strip()), (2, "", "keep going"))
|
|
out = io.StringIO()
|
|
with contextlib.redirect_stdout(out):
|
|
code = hooks._keep_going("codex", "keep going")
|
|
self.assertEqual(code, 0)
|
|
self.assertEqual(json.loads(out.getvalue())["decision"], "block")
|
|
|
|
|
|
class HooksVisible(HarborCase):
|
|
def test_mcp_without_hooks_needs_attention(self):
|
|
self.h.connect("Codex", "mcp", "codex")
|
|
no_hooks = lambda: [a["agent"] for a in self.h.attention() if a["kind"] == "no_hooks"]
|
|
launch = lambda: {r["agent"]: r for r in self.h.launch_status()}["Codex"]["hooks"]
|
|
self.assertEqual((no_hooks(), launch()), (["Codex"], None))
|
|
self.h.connect("Codex", "hook", "codex")
|
|
self.assertEqual(no_hooks(), [])
|
|
self.assertTrue(launch())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|