404 lines
16 KiB
Python
404 lines
16 KiB
Python
|
"""Tests for the os-status hooks. Run: python3 tests/hook_test.py"""
|
||
|
|
|
||
|
|
import importlib.util
|
||
|
|
import io
|
||
|
|
import json
|
||
|
|
import os
|
||
|
|
import subprocess
|
||
|
|
import sys
|
||
|
|
import unittest
|
||
|
|
from datetime import date
|
||
|
|
from pathlib import Path
|
||
|
|
from tempfile import TemporaryDirectory
|
||
|
|
from unittest import mock
|
||
|
|
|
||
|
|
PLUGIN_ROOT = Path(__file__).resolve().parent.parent
|
||
|
|
REPO_PLUGINS = PLUGIN_ROOT.parent
|
||
|
|
sys.path.insert(0, str(PLUGIN_ROOT / "hooks"))
|
||
|
|
|
||
|
|
import checks # noqa: E402
|
||
|
|
import session_start # noqa: E402
|
||
|
|
from checks import ( # noqa: E402
|
||
|
|
ABSENT_NOTE,
|
||
|
|
ENV_VAR,
|
||
|
|
PRESENT_NOTE,
|
||
|
|
Check,
|
||
|
|
CheckResult,
|
||
|
|
Ctx,
|
||
|
|
adr_system_present,
|
||
|
|
subagent_model_env_override,
|
||
|
|
vault_hub_note_present,
|
||
|
|
)
|
||
|
|
from state import StateDir, find_project_root, read_config # noqa: E402
|
||
|
|
|
||
|
|
TODAY = date(2026, 7, 6)
|
||
|
|
YESTERDAY = date(2026, 7, 5)
|
||
|
|
|
||
|
|
|
||
|
|
def make_ctx(
|
||
|
|
project_root=None,
|
||
|
|
settings_path=Path("/nonexistent/settings.json"),
|
||
|
|
vault_path=Path("/nonexistent/vault"),
|
||
|
|
config=None,
|
||
|
|
environ=None,
|
||
|
|
):
|
||
|
|
return Ctx(
|
||
|
|
project_root=project_root,
|
||
|
|
settings_path=settings_path,
|
||
|
|
vault_path=vault_path,
|
||
|
|
config=config or {},
|
||
|
|
environ=environ or {},
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def git_project(tmp):
|
||
|
|
root = Path(tmp)
|
||
|
|
(root / ".git").mkdir(parents=True)
|
||
|
|
return root
|
||
|
|
|
||
|
|
|
||
|
|
class EnvOverrideCheckTest(unittest.TestCase):
|
||
|
|
def test_ok_when_unset_everywhere(self):
|
||
|
|
self.assertEqual("ok", subagent_model_env_override(make_ctx()).status)
|
||
|
|
|
||
|
|
def test_warns_on_process_environment(self):
|
||
|
|
result = subagent_model_env_override(make_ctx(environ={ENV_VAR: "haiku"}))
|
||
|
|
self.assertEqual("warn", result.status)
|
||
|
|
self.assertIn(ENV_VAR, result.message)
|
||
|
|
self.assertIn("haiku", result.message)
|
||
|
|
|
||
|
|
def test_warns_on_settings_env_block_naming_the_file(self):
|
||
|
|
with TemporaryDirectory() as tmp:
|
||
|
|
settings = Path(tmp) / "settings.json"
|
||
|
|
settings.write_text(json.dumps({"env": {ENV_VAR: "haiku"}}))
|
||
|
|
result = subagent_model_env_override(make_ctx(settings_path=settings))
|
||
|
|
self.assertEqual("warn", result.status)
|
||
|
|
self.assertIn(str(settings), result.message)
|
||
|
|
self.assertIn("haiku", result.message)
|
||
|
|
|
||
|
|
def test_names_both_sources_when_both_set(self):
|
||
|
|
with TemporaryDirectory() as tmp:
|
||
|
|
settings = Path(tmp) / "settings.json"
|
||
|
|
settings.write_text(json.dumps({"env": {ENV_VAR: "haiku"}}))
|
||
|
|
result = subagent_model_env_override(
|
||
|
|
make_ctx(settings_path=settings, environ={ENV_VAR: "sonnet"})
|
||
|
|
)
|
||
|
|
self.assertIn("environment", result.message)
|
||
|
|
self.assertIn(str(settings), result.message)
|
||
|
|
|
||
|
|
def test_malformed_settings_is_ignored(self):
|
||
|
|
with TemporaryDirectory() as tmp:
|
||
|
|
settings = Path(tmp) / "settings.json"
|
||
|
|
settings.write_text("{not json")
|
||
|
|
self.assertEqual(
|
||
|
|
"ok", subagent_model_env_override(make_ctx(settings_path=settings)).status
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
class AdrCheckTest(unittest.TestCase):
|
||
|
|
def test_present_notes_verbatim_wording(self):
|
||
|
|
with TemporaryDirectory() as tmp:
|
||
|
|
root = git_project(tmp)
|
||
|
|
(root / "docs" / "adr").mkdir(parents=True)
|
||
|
|
(root / "docs" / "adr" / "README.md").write_text("index")
|
||
|
|
result = adr_system_present(make_ctx(project_root=root))
|
||
|
|
self.assertEqual(CheckResult("note", PRESENT_NOTE), result)
|
||
|
|
|
||
|
|
def test_dir_without_index_counts_as_absent(self):
|
||
|
|
with TemporaryDirectory() as tmp:
|
||
|
|
root = git_project(tmp)
|
||
|
|
(root / "docs" / "adr").mkdir(parents=True)
|
||
|
|
result = adr_system_present(make_ctx(project_root=root))
|
||
|
|
self.assertEqual(CheckResult("warn", ABSENT_NOTE), result)
|
||
|
|
|
||
|
|
def test_absent_warns_with_init_and_migrate(self):
|
||
|
|
with TemporaryDirectory() as tmp:
|
||
|
|
root = git_project(tmp)
|
||
|
|
result = adr_system_present(make_ctx(project_root=root))
|
||
|
|
self.assertEqual("warn", result.status)
|
||
|
|
self.assertIn("/os-adr:init", result.message)
|
||
|
|
self.assertIn("/os-adr:migrate", result.message)
|
||
|
|
|
||
|
|
def test_legacy_os_adr_suppress_is_honored(self):
|
||
|
|
with TemporaryDirectory() as tmp:
|
||
|
|
root = git_project(tmp)
|
||
|
|
(root / ".os-adr").mkdir()
|
||
|
|
(root / ".os-adr" / "suppress").write_text("")
|
||
|
|
self.assertEqual("ok", adr_system_present(make_ctx(project_root=root)).status)
|
||
|
|
|
||
|
|
|
||
|
|
class HubNoteCheckTest(unittest.TestCase):
|
||
|
|
def vault_with(self, tmp, name, body):
|
||
|
|
vault = Path(tmp) / "vault"
|
||
|
|
vault.mkdir(exist_ok=True)
|
||
|
|
(vault / name).write_text(body)
|
||
|
|
return vault
|
||
|
|
|
||
|
|
def hub_note(self, project):
|
||
|
|
return (
|
||
|
|
f"---\ntype: hub\ntags:\n - type/hub\n - project/{project}\n---\n# hub\n"
|
||
|
|
)
|
||
|
|
|
||
|
|
def test_missing_vault_is_silent(self):
|
||
|
|
with TemporaryDirectory() as tmp:
|
||
|
|
root = git_project(tmp)
|
||
|
|
ctx = make_ctx(project_root=root, vault_path=Path(tmp) / "no-vault")
|
||
|
|
self.assertEqual("ok", vault_hub_note_present(ctx).status)
|
||
|
|
|
||
|
|
def test_facet_tag_scan_finds_hub(self):
|
||
|
|
with TemporaryDirectory() as tmp:
|
||
|
|
root = git_project(Path(tmp) / "myproj")
|
||
|
|
vault = self.vault_with(tmp, "myproj-hub.md", self.hub_note("myproj"))
|
||
|
|
ctx = make_ctx(project_root=root, vault_path=vault)
|
||
|
|
self.assertEqual("ok", vault_hub_note_present(ctx).status)
|
||
|
|
|
||
|
|
def test_no_hub_warns_naming_corrective_skill(self):
|
||
|
|
with TemporaryDirectory() as tmp:
|
||
|
|
root = git_project(Path(tmp) / "myproj")
|
||
|
|
vault = self.vault_with(tmp, "other.md", self.hub_note("otherproj"))
|
||
|
|
result = vault_hub_note_present(make_ctx(project_root=root, vault_path=vault))
|
||
|
|
self.assertEqual("warn", result.status)
|
||
|
|
self.assertIn("/os-vault:write", result.message)
|
||
|
|
self.assertIn("myproj", result.message)
|
||
|
|
|
||
|
|
def test_project_name_is_not_prefix_matched(self):
|
||
|
|
with TemporaryDirectory() as tmp:
|
||
|
|
root = git_project(Path(tmp) / "myproj")
|
||
|
|
vault = self.vault_with(tmp, "other.md", self.hub_note("myproj2"))
|
||
|
|
result = vault_hub_note_present(make_ctx(project_root=root, vault_path=vault))
|
||
|
|
self.assertEqual("warn", result.status)
|
||
|
|
|
||
|
|
def test_config_slug_wins_over_inference(self):
|
||
|
|
with TemporaryDirectory() as tmp:
|
||
|
|
root = git_project(Path(tmp) / "myproj")
|
||
|
|
vault = self.vault_with(tmp, "custom-hub.md", "any content")
|
||
|
|
ctx = make_ctx(
|
||
|
|
project_root=root, vault_path=vault, config={"hub": "custom-hub"}
|
||
|
|
)
|
||
|
|
self.assertEqual("ok", vault_hub_note_present(ctx).status)
|
||
|
|
|
||
|
|
def test_config_slug_pointing_nowhere_warns(self):
|
||
|
|
with TemporaryDirectory() as tmp:
|
||
|
|
root = git_project(Path(tmp) / "myproj")
|
||
|
|
vault = self.vault_with(tmp, "other.md", "content")
|
||
|
|
ctx = make_ctx(project_root=root, vault_path=vault, config={"hub": "ghost"})
|
||
|
|
result = vault_hub_note_present(ctx)
|
||
|
|
self.assertEqual("warn", result.status)
|
||
|
|
self.assertIn("ghost", result.message)
|
||
|
|
|
||
|
|
def test_templates_and_graphify_out_are_skipped(self):
|
||
|
|
with TemporaryDirectory() as tmp:
|
||
|
|
root = git_project(Path(tmp) / "myproj")
|
||
|
|
vault = Path(tmp) / "vault"
|
||
|
|
(vault / "_templates").mkdir(parents=True)
|
||
|
|
(vault / "_templates" / "t.md").write_text(self.hub_note("myproj"))
|
||
|
|
(vault / "graphify-out").mkdir()
|
||
|
|
(vault / "graphify-out" / "g.md").write_text(self.hub_note("myproj"))
|
||
|
|
result = vault_hub_note_present(make_ctx(project_root=root, vault_path=vault))
|
||
|
|
self.assertEqual("warn", result.status)
|
||
|
|
|
||
|
|
|
||
|
|
class RunnerTest(unittest.TestCase):
|
||
|
|
"""run() contract: routing, aggregation, snooze/suppress, isolation."""
|
||
|
|
|
||
|
|
def run_with(self, registry, ctx, state=None, today=TODAY):
|
||
|
|
out = io.StringIO()
|
||
|
|
with mock.patch.object(session_start, "REGISTRY", registry):
|
||
|
|
session_start.run(ctx, state, today, out)
|
||
|
|
raw = out.getvalue()
|
||
|
|
return json.loads(raw) if raw else {}
|
||
|
|
|
||
|
|
def check(self, name, result, project_scoped=False):
|
||
|
|
return Check(name, lambda ctx: result, project_scoped)
|
||
|
|
|
||
|
|
def test_all_ok_emits_nothing(self):
|
||
|
|
registry = [self.check("a", CheckResult("ok")), self.check("b", CheckResult("ok"))]
|
||
|
|
self.assertEqual({}, self.run_with(registry, make_ctx()))
|
||
|
|
|
||
|
|
def test_note_passes_through_as_additional_context_without_banner(self):
|
||
|
|
registry = [self.check("a", CheckResult("note", "hello"))]
|
||
|
|
payload = self.run_with(registry, make_ctx())
|
||
|
|
self.assertEqual("hello", payload["hookSpecificOutput"]["additionalContext"])
|
||
|
|
self.assertNotIn("systemMessage", payload)
|
||
|
|
|
||
|
|
def test_multiple_warns_aggregate_into_one_banner(self):
|
||
|
|
registry = [
|
||
|
|
self.check("a", CheckResult("warn", "first problem")),
|
||
|
|
self.check("b", CheckResult("warn", "second problem")),
|
||
|
|
]
|
||
|
|
payload = self.run_with(registry, make_ctx())
|
||
|
|
banner = payload["systemMessage"]
|
||
|
|
self.assertIn("first problem", banner)
|
||
|
|
self.assertIn("second problem", banner)
|
||
|
|
self.assertEqual(1, banner.count(session_start.BANNER_HEADER))
|
||
|
|
|
||
|
|
def test_raising_check_becomes_generic_warn_and_others_still_run(self):
|
||
|
|
def boom(ctx):
|
||
|
|
raise RuntimeError("kaboom")
|
||
|
|
|
||
|
|
registry = [
|
||
|
|
Check("broken", boom, False),
|
||
|
|
self.check("healthy", CheckResult("note", "still here")),
|
||
|
|
]
|
||
|
|
payload = self.run_with(registry, make_ctx())
|
||
|
|
self.assertIn("broken", payload["systemMessage"])
|
||
|
|
self.assertNotIn("kaboom", payload["systemMessage"])
|
||
|
|
self.assertIn("still here", payload["hookSpecificOutput"]["additionalContext"])
|
||
|
|
|
||
|
|
def test_non_checkresult_return_becomes_generic_warn(self):
|
||
|
|
registry = [Check("weird", lambda ctx: "nope", False)]
|
||
|
|
payload = self.run_with(registry, make_ctx())
|
||
|
|
self.assertIn("weird", payload["systemMessage"])
|
||
|
|
|
||
|
|
def test_project_scoped_checks_skipped_outside_git_project(self):
|
||
|
|
registry = [self.check("proj", CheckResult("warn", "boom"), project_scoped=True)]
|
||
|
|
self.assertEqual({}, self.run_with(registry, make_ctx(project_root=None)))
|
||
|
|
|
||
|
|
def test_warn_stamps_snooze_and_second_run_same_day_is_silent(self):
|
||
|
|
with TemporaryDirectory() as tmp:
|
||
|
|
root = git_project(tmp)
|
||
|
|
state = StateDir(root)
|
||
|
|
registry = [self.check("a", CheckResult("warn", "boom"))]
|
||
|
|
ctx = make_ctx(project_root=root)
|
||
|
|
first = self.run_with(registry, ctx, state)
|
||
|
|
self.assertIn("boom", first["systemMessage"])
|
||
|
|
self.assertTrue((root / ".cc-os" / "snooze-a").exists())
|
||
|
|
self.assertEqual({}, self.run_with(registry, ctx, state))
|
||
|
|
|
||
|
|
def test_next_day_warns_again(self):
|
||
|
|
with TemporaryDirectory() as tmp:
|
||
|
|
root = git_project(tmp)
|
||
|
|
state = StateDir(root)
|
||
|
|
registry = [self.check("a", CheckResult("warn", "boom"))]
|
||
|
|
ctx = make_ctx(project_root=root)
|
||
|
|
self.run_with(registry, ctx, state, today=YESTERDAY)
|
||
|
|
self.assertIn("boom", self.run_with(registry, ctx, state)["systemMessage"])
|
||
|
|
|
||
|
|
def test_suppress_marker_silences_permanently(self):
|
||
|
|
with TemporaryDirectory() as tmp:
|
||
|
|
root = git_project(tmp)
|
||
|
|
(root / ".cc-os").mkdir()
|
||
|
|
(root / ".cc-os" / "suppress-a").write_text("")
|
||
|
|
state = StateDir(root)
|
||
|
|
registry = [self.check("a", CheckResult("warn", "boom"))]
|
||
|
|
self.assertEqual({}, self.run_with(registry, make_ctx(project_root=root), state))
|
||
|
|
|
||
|
|
def test_notes_are_never_snoozed_or_suppressed(self):
|
||
|
|
with TemporaryDirectory() as tmp:
|
||
|
|
root = git_project(tmp)
|
||
|
|
(root / ".cc-os").mkdir()
|
||
|
|
(root / ".cc-os" / "suppress-a").write_text("")
|
||
|
|
(root / ".cc-os" / "snooze-a").write_text(TODAY.isoformat())
|
||
|
|
state = StateDir(root)
|
||
|
|
registry = [self.check("a", CheckResult("note", "always"))]
|
||
|
|
payload = self.run_with(registry, make_ctx(project_root=root), state)
|
||
|
|
self.assertEqual("always", payload["hookSpecificOutput"]["additionalContext"])
|
||
|
|
|
||
|
|
def test_snoozed_warn_does_not_suppress_other_warns(self):
|
||
|
|
with TemporaryDirectory() as tmp:
|
||
|
|
root = git_project(tmp)
|
||
|
|
(root / ".cc-os").mkdir()
|
||
|
|
(root / ".cc-os" / "snooze-a").write_text(TODAY.isoformat())
|
||
|
|
state = StateDir(root)
|
||
|
|
registry = [
|
||
|
|
self.check("a", CheckResult("warn", "old news")),
|
||
|
|
self.check("b", CheckResult("warn", "new problem")),
|
||
|
|
]
|
||
|
|
banner = self.run_with(registry, make_ctx(project_root=root), state)["systemMessage"]
|
||
|
|
self.assertNotIn("old news", banner)
|
||
|
|
self.assertIn("new problem", banner)
|
||
|
|
|
||
|
|
|
||
|
|
class StateTest(unittest.TestCase):
|
||
|
|
def test_read_config_parses_key_value_and_skips_comments(self):
|
||
|
|
with TemporaryDirectory() as tmp:
|
||
|
|
root = Path(tmp)
|
||
|
|
(root / ".cc-os").mkdir()
|
||
|
|
(root / ".cc-os" / "config").write_text(
|
||
|
|
"# comment\nhub = my-hub\nvault_path=/somewhere\nbadline\n"
|
||
|
|
)
|
||
|
|
self.assertEqual(
|
||
|
|
{"hub": "my-hub", "vault_path": "/somewhere"}, read_config(root)
|
||
|
|
)
|
||
|
|
|
||
|
|
def test_read_config_missing_is_empty(self):
|
||
|
|
with TemporaryDirectory() as tmp:
|
||
|
|
self.assertEqual({}, read_config(Path(tmp)))
|
||
|
|
|
||
|
|
def test_corrupt_snooze_stamp_reads_as_unset(self):
|
||
|
|
with TemporaryDirectory() as tmp:
|
||
|
|
root = Path(tmp)
|
||
|
|
(root / ".cc-os").mkdir()
|
||
|
|
(root / ".cc-os" / "snooze-a").write_text("not-a-date")
|
||
|
|
self.assertFalse(StateDir(root).snoozed("a", TODAY))
|
||
|
|
|
||
|
|
def test_find_project_root_walks_up(self):
|
||
|
|
with TemporaryDirectory() as tmp:
|
||
|
|
root = Path(tmp)
|
||
|
|
(root / ".git").mkdir()
|
||
|
|
nested = root / "a" / "b"
|
||
|
|
nested.mkdir(parents=True)
|
||
|
|
self.assertEqual(root, find_project_root(nested))
|
||
|
|
|
||
|
|
def test_find_project_root_none_outside_git(self):
|
||
|
|
with TemporaryDirectory() as tmp:
|
||
|
|
self.assertIsNone(find_project_root(Path(tmp)))
|
||
|
|
|
||
|
|
|
||
|
|
class WordingInvariantTest(unittest.TestCase):
|
||
|
|
"""PRESENT_NOTE / ABSENT_NOTE must stay byte-identical to os-adr's
|
||
|
|
hooks/session_start.py, the wording source of record (invariants.md)."""
|
||
|
|
|
||
|
|
def os_adr_module(self):
|
||
|
|
source = REPO_PLUGINS / "os-adr" / "hooks" / "session_start.py"
|
||
|
|
spec = importlib.util.spec_from_file_location("os_adr_session_start", source)
|
||
|
|
module = importlib.util.module_from_spec(spec)
|
||
|
|
spec.loader.exec_module(module)
|
||
|
|
return module
|
||
|
|
|
||
|
|
def test_notes_byte_identical_to_os_adr(self):
|
||
|
|
os_adr = self.os_adr_module()
|
||
|
|
self.assertEqual(os_adr.PRESENT_NOTE, PRESENT_NOTE)
|
||
|
|
self.assertEqual(os_adr.ABSENT_NOTE, ABSENT_NOTE)
|
||
|
|
|
||
|
|
|
||
|
|
class ExitZeroEnvelopeTest(unittest.TestCase):
|
||
|
|
def test_main_returns_zero_even_when_run_raises(self):
|
||
|
|
with mock.patch.object(session_start, "run", side_effect=RuntimeError):
|
||
|
|
self.assertEqual(0, session_start.main())
|
||
|
|
|
||
|
|
def test_subprocess_in_non_git_cwd_exits_zero_silently(self):
|
||
|
|
with TemporaryDirectory() as tmp:
|
||
|
|
env = {k: v for k, v in os.environ.items() if k != ENV_VAR}
|
||
|
|
env["HOME"] = tmp # no settings.json, no vault
|
||
|
|
proc = subprocess.run(
|
||
|
|
[sys.executable, str(PLUGIN_ROOT / "hooks" / "session_start.py")],
|
||
|
|
cwd=tmp,
|
||
|
|
env=env,
|
||
|
|
capture_output=True,
|
||
|
|
text=True,
|
||
|
|
)
|
||
|
|
self.assertEqual(0, proc.returncode)
|
||
|
|
self.assertEqual("", proc.stdout)
|
||
|
|
self.assertFalse((Path(tmp) / ".cc-os").exists())
|
||
|
|
|
||
|
|
def test_subprocess_with_env_override_warns_but_exits_zero(self):
|
||
|
|
with TemporaryDirectory() as tmp:
|
||
|
|
env = dict(os.environ)
|
||
|
|
env["HOME"] = tmp
|
||
|
|
env[ENV_VAR] = "haiku"
|
||
|
|
proc = subprocess.run(
|
||
|
|
[sys.executable, str(PLUGIN_ROOT / "hooks" / "session_start.py")],
|
||
|
|
cwd=tmp,
|
||
|
|
env=env,
|
||
|
|
capture_output=True,
|
||
|
|
text=True,
|
||
|
|
)
|
||
|
|
self.assertEqual(0, proc.returncode)
|
||
|
|
payload = json.loads(proc.stdout)
|
||
|
|
self.assertIn(ENV_VAR, payload["systemMessage"])
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
unittest.main()
|