788 lines
33 KiB
Python
788 lines
33 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,
|
|
CURRENT_CONFIG_VERSION,
|
|
ENV_VAR,
|
|
PRESENT_NOTE,
|
|
TRACKER_ABSENT_NOTE,
|
|
Check,
|
|
CheckResult,
|
|
Ctx,
|
|
adr_system_present,
|
|
config_version_current,
|
|
project_graph_present,
|
|
run_all_json,
|
|
subagent_model_env_override,
|
|
tracker_configured,
|
|
vault_hub_note_present,
|
|
)
|
|
from state import ( # noqa: E402
|
|
StateDir,
|
|
clear_snooze,
|
|
find_project_root,
|
|
read_config,
|
|
write_config_value,
|
|
)
|
|
|
|
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 AuditDueCheckTest(unittest.TestCase):
|
|
def ledger_env(self, tmp, rows):
|
|
path = Path(tmp) / "metrics.tsv"
|
|
header = "date\tsince\tsessions\n"
|
|
path.write_text(header + "".join(f"{r}\t2026-01-01\t5\n" for r in rows))
|
|
return {checks.AUDIT_LEDGER_ENV: str(path)}
|
|
|
|
def test_missing_ledger_is_silent_machine_not_opted_in(self):
|
|
ctx = make_ctx(environ={checks.AUDIT_LEDGER_ENV: "/nonexistent/metrics.tsv"})
|
|
self.assertEqual("ok", checks.orchestration_audit_due(ctx).status)
|
|
|
|
def test_recent_run_is_ok(self):
|
|
with TemporaryDirectory() as tmp:
|
|
env = self.ledger_env(tmp, [date.today().isoformat()])
|
|
self.assertEqual("ok", checks.orchestration_audit_due(make_ctx(environ=env)).status)
|
|
|
|
def test_stale_run_warns_with_last_date(self):
|
|
with TemporaryDirectory() as tmp:
|
|
env = self.ledger_env(tmp, ["2026-01-01", "2026-02-01"])
|
|
result = checks.orchestration_audit_due(make_ctx(environ=env))
|
|
self.assertEqual("warn", result.status)
|
|
self.assertIn("2026-02-01", result.message)
|
|
self.assertIn("/os-context:audit-sessions", result.message)
|
|
|
|
def test_corrupt_ledger_reads_as_missing(self):
|
|
with TemporaryDirectory() as tmp:
|
|
path = Path(tmp) / "metrics.tsv"
|
|
path.write_text("header only\nnot-a-date\tx\n")
|
|
env = {checks.AUDIT_LEDGER_ENV: str(path)}
|
|
self.assertEqual("ok", checks.orchestration_audit_due(make_ctx(environ=env)).status)
|
|
|
|
|
|
class TrackerCheckTest(unittest.TestCase):
|
|
def test_absent_warns_pointing_to_routing_skill(self):
|
|
with TemporaryDirectory() as tmp:
|
|
root = git_project(tmp)
|
|
result = tracker_configured(make_ctx(project_root=root))
|
|
self.assertEqual(CheckResult("warn", TRACKER_ABSENT_NOTE), result)
|
|
self.assertIn("/os-backlog:route", result.message)
|
|
|
|
def test_planka_value_is_ok(self):
|
|
with TemporaryDirectory() as tmp:
|
|
root = git_project(tmp)
|
|
ctx = make_ctx(project_root=root, config={"tracker": "planka:my-board"})
|
|
self.assertEqual("ok", tracker_configured(ctx).status)
|
|
|
|
def test_forgejo_value_is_ok(self):
|
|
with TemporaryDirectory() as tmp:
|
|
root = git_project(tmp)
|
|
ctx = make_ctx(project_root=root, config={"tracker": "forgejo:jared/cc-os"})
|
|
self.assertEqual("ok", tracker_configured(ctx).status)
|
|
|
|
def test_github_value_is_ok(self):
|
|
with TemporaryDirectory() as tmp:
|
|
root = git_project(tmp)
|
|
ctx = make_ctx(project_root=root, config={"tracker": "github:jared/cc-os"})
|
|
self.assertEqual("ok", tracker_configured(ctx).status)
|
|
|
|
def test_repo_value_is_ok(self):
|
|
with TemporaryDirectory() as tmp:
|
|
root = git_project(tmp)
|
|
ctx = make_ctx(project_root=root, config={"tracker": "repo:issues/"})
|
|
self.assertEqual("ok", tracker_configured(ctx).status)
|
|
|
|
def test_unknown_scheme_warns_without_crashing(self):
|
|
with TemporaryDirectory() as tmp:
|
|
root = git_project(tmp)
|
|
ctx = make_ctx(project_root=root, config={"tracker": "trello:board123"})
|
|
result = tracker_configured(ctx)
|
|
self.assertEqual("warn", result.status)
|
|
self.assertIn("trello:board123", result.message)
|
|
|
|
def test_missing_separator_warns_without_crashing(self):
|
|
with TemporaryDirectory() as tmp:
|
|
root = git_project(tmp)
|
|
ctx = make_ctx(project_root=root, config={"tracker": "planka-my-board"})
|
|
result = tracker_configured(ctx)
|
|
self.assertEqual("warn", result.status)
|
|
|
|
def test_forgejo_missing_repo_part_warns(self):
|
|
with TemporaryDirectory() as tmp:
|
|
root = git_project(tmp)
|
|
ctx = make_ctx(project_root=root, config={"tracker": "forgejo:jared"})
|
|
result = tracker_configured(ctx)
|
|
self.assertEqual("warn", result.status)
|
|
|
|
def test_is_project_scoped_and_daily_snoozed_via_registry(self):
|
|
entry = next(c for c in checks.REGISTRY if c.name == "tracker-configured")
|
|
self.assertTrue(entry.project_scoped)
|
|
|
|
|
|
class ConfigVersionCheckTest(unittest.TestCase):
|
|
def test_missing_version_warns(self):
|
|
with TemporaryDirectory() as tmp:
|
|
root = git_project(tmp)
|
|
result = config_version_current(make_ctx(project_root=root))
|
|
self.assertEqual("warn", result.status)
|
|
self.assertIn("/os-status:fix", result.message)
|
|
|
|
def test_current_version_is_ok(self):
|
|
with TemporaryDirectory() as tmp:
|
|
root = git_project(tmp)
|
|
ctx = make_ctx(project_root=root, config={"version": str(CURRENT_CONFIG_VERSION)})
|
|
self.assertEqual("ok", config_version_current(ctx).status)
|
|
|
|
def test_old_version_warns(self):
|
|
with TemporaryDirectory() as tmp:
|
|
root = git_project(tmp)
|
|
ctx = make_ctx(project_root=root, config={"version": "0"})
|
|
self.assertEqual("warn", config_version_current(ctx).status)
|
|
|
|
def test_non_numeric_version_warns_without_crashing(self):
|
|
with TemporaryDirectory() as tmp:
|
|
root = git_project(tmp)
|
|
ctx = make_ctx(project_root=root, config={"version": "banana"})
|
|
result = config_version_current(ctx)
|
|
self.assertEqual("warn", result.status)
|
|
|
|
|
|
class ProjectGraphCheckTest(unittest.TestCase):
|
|
def test_missing_graph_warns_naming_onboard(self):
|
|
with TemporaryDirectory() as tmp:
|
|
root = git_project(tmp)
|
|
result = project_graph_present(make_ctx(project_root=root))
|
|
self.assertEqual("warn", result.status)
|
|
self.assertIn("/os-vault:onboard-project", result.message)
|
|
|
|
def test_present_graph_is_ok(self):
|
|
with TemporaryDirectory() as tmp:
|
|
root = git_project(tmp)
|
|
(root / "graphify-out").mkdir()
|
|
(root / "graphify-out" / "graph.json").write_text("{}")
|
|
result = project_graph_present(make_ctx(project_root=root))
|
|
self.assertEqual("ok", result.status)
|
|
|
|
|
|
class RemediationTest(unittest.TestCase):
|
|
def test_every_registry_entry_has_a_remediation(self):
|
|
for entry in checks.REGISTRY:
|
|
self.assertTrue(entry.remediation, f"{entry.name} has no remediation")
|
|
|
|
def test_known_remediations(self):
|
|
by_name = {c.name: c.remediation for c in checks.REGISTRY}
|
|
self.assertEqual("/os-adr:init", by_name["adr-system-present"])
|
|
self.assertEqual("/os-backlog:route", by_name["tracker-configured"])
|
|
self.assertEqual("/os-vault:onboard-project", by_name["project-graph-present"])
|
|
self.assertEqual("/os-status:fix", by_name["config-version-current"])
|
|
|
|
|
|
class JsonRunnerTest(unittest.TestCase):
|
|
def test_shape_has_name_status_message_remediation(self):
|
|
with TemporaryDirectory() as tmp:
|
|
root = git_project(tmp)
|
|
ctx = make_ctx(project_root=root)
|
|
results = run_all_json(ctx)
|
|
self.assertTrue(results)
|
|
for entry in results:
|
|
self.assertEqual({"name", "status", "message", "remediation"}, set(entry.keys()))
|
|
|
|
def test_project_scoped_checks_skipped_when_no_project_root(self):
|
|
results = run_all_json(make_ctx(project_root=None))
|
|
names = {r["name"] for r in results}
|
|
self.assertNotIn("adr-system-present", names)
|
|
self.assertNotIn("tracker-configured", names)
|
|
|
|
def test_run_all_json_self_clears_stale_snooze_on_ok(self):
|
|
with TemporaryDirectory() as tmp:
|
|
root = git_project(tmp)
|
|
(root / ".cc-os" / "status").mkdir(parents=True)
|
|
(root / ".cc-os" / "status" / "snooze-adr-system-present").write_text(
|
|
date.today().isoformat()
|
|
)
|
|
(root / "docs" / "adr").mkdir(parents=True)
|
|
(root / "docs" / "adr" / "README.md").write_text("index")
|
|
ctx = make_ctx(project_root=root)
|
|
run_all_json(ctx)
|
|
self.assertFalse(
|
|
(root / ".cc-os" / "status" / "snooze-adr-system-present").exists()
|
|
)
|
|
|
|
def test_subprocess_json_flag_prints_valid_json_array(self):
|
|
with TemporaryDirectory() as tmp:
|
|
root = git_project(tmp)
|
|
proc = subprocess.run(
|
|
[sys.executable, str(PLUGIN_ROOT / "hooks" / "checks.py"), "--json"],
|
|
cwd=str(root),
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
self.assertEqual(0, proc.returncode)
|
|
payload = json.loads(proc.stdout)
|
|
self.assertIsInstance(payload, list)
|
|
names = {e["name"] for e in payload}
|
|
self.assertIn("subagent-model-env-override", names)
|
|
|
|
|
|
class WriteConfigValueTest(unittest.TestCase):
|
|
def test_stamps_new_key(self):
|
|
with TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
write_config_value(root, "version", "1")
|
|
self.assertEqual({"version": "1"}, read_config(root))
|
|
|
|
def test_preserves_existing_keys(self):
|
|
with TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
(root / ".cc-os").mkdir()
|
|
(root / ".cc-os" / "config").write_text("hub = my-hub\ntracker = planka:board\n")
|
|
write_config_value(root, "version", "1")
|
|
self.assertEqual(
|
|
{"hub": "my-hub", "tracker": "planka:board", "version": "1"},
|
|
read_config(root),
|
|
)
|
|
|
|
def test_updates_existing_key_in_place(self):
|
|
with TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
(root / ".cc-os").mkdir()
|
|
(root / ".cc-os" / "config").write_text("version = 0\nhub = my-hub\n")
|
|
write_config_value(root, "version", "1")
|
|
config = read_config(root)
|
|
self.assertEqual("1", config["version"])
|
|
self.assertEqual("my-hub", config["hub"])
|
|
|
|
def test_new_key_written_without_spaces_around_equals(self):
|
|
with TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
write_config_value(root, "version", "1")
|
|
raw = (root / ".cc-os" / "config").read_text()
|
|
self.assertEqual("version=1\n", raw)
|
|
|
|
def test_normalizes_existing_spaced_lines_on_rewrite(self):
|
|
with TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
(root / ".cc-os").mkdir()
|
|
(root / ".cc-os" / "config").write_text("version = 0\nhub = my-hub\n")
|
|
write_config_value(root, "version", "1")
|
|
raw = (root / ".cc-os" / "config").read_text()
|
|
for line in raw.splitlines():
|
|
self.assertNotIn(" = ", line)
|
|
self.assertNotIn(" =", line)
|
|
self.assertNotIn("= ", line)
|
|
self.assertEqual({"version": "1", "hub": "my-hub"}, read_config(root))
|
|
|
|
def test_comments_and_blank_lines_untouched(self):
|
|
with TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
(root / ".cc-os").mkdir()
|
|
(root / ".cc-os" / "config").write_text("# a comment\n\nhub = my-hub\n")
|
|
write_config_value(root, "version", "1")
|
|
raw = (root / ".cc-os" / "config").read_text()
|
|
self.assertIn("# a comment", raw)
|
|
|
|
|
|
class BannerWordingTest(unittest.TestCase):
|
|
def test_banner_header_names_fix_skill(self):
|
|
self.assertIn("/os-status:fix", session_start.BANNER_HEADER)
|
|
|
|
|
|
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" / "status" / "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)
|
|
|
|
def test_snooze_self_clears_when_check_flips_to_ok(self):
|
|
with TemporaryDirectory() as tmp:
|
|
root = git_project(tmp)
|
|
state = StateDir(root)
|
|
warn_registry = [self.check("a", CheckResult("warn", "boom"))]
|
|
ctx = make_ctx(project_root=root)
|
|
self.run_with(warn_registry, ctx, state)
|
|
self.assertTrue((root / ".cc-os" / "status" / "snooze-a").exists())
|
|
|
|
ok_registry = [self.check("a", CheckResult("ok"))]
|
|
self.run_with(ok_registry, ctx, state)
|
|
self.assertFalse((root / ".cc-os" / "status" / "snooze-a").exists())
|
|
|
|
def test_snooze_self_clears_when_check_flips_to_note(self):
|
|
with TemporaryDirectory() as tmp:
|
|
root = git_project(tmp)
|
|
state = StateDir(root)
|
|
warn_registry = [self.check("a", CheckResult("warn", "boom"))]
|
|
ctx = make_ctx(project_root=root)
|
|
self.run_with(warn_registry, ctx, state)
|
|
self.assertTrue((root / ".cc-os" / "status" / "snooze-a").exists())
|
|
|
|
note_registry = [self.check("a", CheckResult("note", "fine now"))]
|
|
self.run_with(note_registry, ctx, state)
|
|
self.assertFalse((root / ".cc-os" / "status" / "snooze-a").exists())
|
|
|
|
|
|
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)))
|
|
|
|
def test_stamp_writes_to_status_subdir(self):
|
|
with TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
StateDir(root).stamp("a", TODAY)
|
|
self.assertTrue((root / ".cc-os" / "status" / "snooze-a").exists())
|
|
self.assertFalse((root / ".cc-os" / "snooze-a").exists())
|
|
|
|
def test_snoozed_falls_back_to_legacy_top_level_path(self):
|
|
with TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
(root / ".cc-os").mkdir()
|
|
(root / ".cc-os" / "snooze-a").write_text(TODAY.isoformat())
|
|
self.assertTrue(StateDir(root).snoozed("a", TODAY))
|
|
|
|
def test_suppressed_falls_back_to_legacy_top_level_path(self):
|
|
with TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
(root / ".cc-os").mkdir()
|
|
(root / ".cc-os" / "suppress-a").write_text("")
|
|
self.assertTrue(StateDir(root).suppressed("a"))
|
|
|
|
def test_current_path_wins_over_legacy_when_both_exist(self):
|
|
with TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
(root / ".cc-os" / "status").mkdir(parents=True)
|
|
(root / ".cc-os" / "status" / "snooze-a").write_text(TODAY.isoformat())
|
|
(root / ".cc-os" / "snooze-a").write_text(YESTERDAY.isoformat())
|
|
self.assertTrue(StateDir(root).snoozed("a", TODAY))
|
|
|
|
def test_write_migrates_legacy_files_and_removes_them(self):
|
|
with TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
(root / ".cc-os").mkdir()
|
|
(root / ".cc-os" / "snooze-b").write_text(YESTERDAY.isoformat())
|
|
(root / ".cc-os" / "suppress-c").write_text("")
|
|
StateDir(root).stamp("a", TODAY)
|
|
# Legacy files migrated up and removed from the old location.
|
|
self.assertTrue((root / ".cc-os" / "status" / "snooze-b").exists())
|
|
self.assertTrue((root / ".cc-os" / "status" / "suppress-c").exists())
|
|
self.assertFalse((root / ".cc-os" / "snooze-b").exists())
|
|
self.assertFalse((root / ".cc-os" / "suppress-c").exists())
|
|
# config, if present, is left alone by migration.
|
|
self.assertFalse((root / ".cc-os" / "status" / "config").exists())
|
|
|
|
def test_migration_does_not_touch_config(self):
|
|
with TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
(root / ".cc-os").mkdir()
|
|
(root / ".cc-os" / "config").write_text("tracker=repo:x\n")
|
|
(root / ".cc-os" / "snooze-a").write_text(YESTERDAY.isoformat())
|
|
StateDir(root).stamp("a", TODAY)
|
|
self.assertTrue((root / ".cc-os" / "config").exists())
|
|
self.assertEqual("tracker=repo:x\n", (root / ".cc-os" / "config").read_text())
|
|
|
|
|
|
class ClearSnoozeTest(unittest.TestCase):
|
|
def test_clears_current_path_snooze(self):
|
|
with TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
(root / ".cc-os" / "status").mkdir(parents=True)
|
|
(root / ".cc-os" / "status" / "snooze-a").write_text(TODAY.isoformat())
|
|
clear_snooze(root, "a")
|
|
self.assertFalse((root / ".cc-os" / "status" / "snooze-a").exists())
|
|
|
|
def test_clears_legacy_path_snooze(self):
|
|
with TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
(root / ".cc-os").mkdir()
|
|
(root / ".cc-os" / "snooze-a").write_text(TODAY.isoformat())
|
|
clear_snooze(root, "a")
|
|
self.assertFalse((root / ".cc-os" / "snooze-a").exists())
|
|
|
|
def test_missing_snooze_is_a_silent_noop(self):
|
|
with TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
clear_snooze(root, "a") # must not raise
|
|
|
|
def test_does_not_clear_other_checks_snooze(self):
|
|
with TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
(root / ".cc-os" / "status").mkdir(parents=True)
|
|
(root / ".cc-os" / "status" / "snooze-a").write_text(TODAY.isoformat())
|
|
(root / ".cc-os" / "status" / "snooze-b").write_text(TODAY.isoformat())
|
|
clear_snooze(root, "a")
|
|
self.assertFalse((root / ".cc-os" / "status" / "snooze-a").exists())
|
|
self.assertTrue((root / ".cc-os" / "status" / "snooze-b").exists())
|
|
|
|
|
|
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()
|