aboutsummaryrefslogtreecommitdiff
path: root/.ai/scripts/tests/test_cross_agent_watch.py
diff options
context:
space:
mode:
authorCraig Jennings <c@cjennings.net>2026-05-06 21:59:52 -0500
committerCraig Jennings <c@cjennings.net>2026-05-06 21:59:52 -0500
commitd81b23ad6b6e437dfe3c338a00a4be39bc555146 (patch)
tree2d4b0d7890fd1fc70d81282b81fed2808c28a106 /.ai/scripts/tests/test_cross_agent_watch.py
parent201377f57430ef28d02e703a2191434bbee55c75 (diff)
downloadrulesets-d81b23ad6b6e437dfe3c338a00a4be39bc555146.tar.gz
rulesets-d81b23ad6b6e437dfe3c338a00a4be39bc555146.zip
chore(ai): initialize project notes and Claude tooling surfaces
Replace the seed notes.org with project-specific context (layout, install modes, task tracker location, recent inflection point). Bring in the synced template surfaces (protocols, workflows, scripts, references, retrospectives, someday-maybe) as tracked content for this content/documentation project.
Diffstat (limited to '.ai/scripts/tests/test_cross_agent_watch.py')
-rw-r--r--.ai/scripts/tests/test_cross_agent_watch.py155
1 files changed, 155 insertions, 0 deletions
diff --git a/.ai/scripts/tests/test_cross_agent_watch.py b/.ai/scripts/tests/test_cross_agent_watch.py
new file mode 100644
index 0000000..417cc19
--- /dev/null
+++ b/.ai/scripts/tests/test_cross_agent_watch.py
@@ -0,0 +1,155 @@
+"""Tests for cross-agent-watch.
+
+Black-box: spawn the script, drop files into a watched dir, read the log.
+Tests use --no-notify to avoid firing real desktop notifications.
+"""
+
+from __future__ import annotations
+
+import os
+import subprocess
+import time
+from pathlib import Path
+
+import pytest
+
+SCRIPT = Path(__file__).resolve().parent.parent / "cross-agent-comms" / "cross-agent-watch"
+
+
+def _spawn(watched_dir: Path, log_path: Path, env: dict) -> subprocess.Popen:
+ return subprocess.Popen(
+ [
+ str(SCRIPT),
+ "--projects-glob", str(watched_dir) + "/",
+ "--log", str(log_path),
+ "--no-notify",
+ "--quiet",
+ ],
+ stdout=subprocess.DEVNULL,
+ stderr=subprocess.PIPE,
+ env=env,
+ )
+
+
+def _wait_for_log_lines(log_path: Path, expected: int, timeout: float = 5.0) -> list[str]:
+ deadline = time.time() + timeout
+ while time.time() < deadline:
+ if log_path.exists():
+ lines = [ln for ln in log_path.read_text().splitlines() if ln]
+ if len(lines) >= expected:
+ return lines
+ time.sleep(0.1)
+ if log_path.exists():
+ return [ln for ln in log_path.read_text().splitlines() if ln]
+ return []
+
+
+@pytest.fixture
+def isolated_env(tmp_path, monkeypatch):
+ fake_home = tmp_path / "home"
+ fake_home.mkdir()
+ monkeypatch.setenv("HOME", str(fake_home))
+ return fake_home
+
+
+def test_watch_help(isolated_env):
+ result = subprocess.run(
+ [str(SCRIPT), "--help"],
+ capture_output=True, text=True,
+ env={**os.environ, "HOME": str(isolated_env)},
+ )
+ assert result.returncode == 0
+ assert "Usage:" in result.stdout
+
+
+def test_watch_empty_glob_exits_nonzero(isolated_env):
+ """Glob resolving to zero dirs should exit non-zero with a clear message."""
+ result = subprocess.run(
+ [str(SCRIPT), "--projects-glob", "/nonexistent/path/*/foo/", "--no-notify", "--quiet"],
+ capture_output=True, text=True,
+ env={**os.environ, "HOME": str(isolated_env)},
+ timeout=3,
+ )
+ assert result.returncode != 0
+ assert "0 directories" in result.stderr
+
+
+def test_watch_logs_org_file_create(isolated_env, tmp_path):
+ watched = tmp_path / "watched"
+ watched.mkdir()
+ log = tmp_path / "watch.log"
+ proc = _spawn(watched, log, {**os.environ, "HOME": str(isolated_env)})
+ try:
+ # Give inotifywait a moment to attach.
+ time.sleep(0.3)
+ (watched / "test-msg.org").write_text("hello")
+ lines = _wait_for_log_lines(log, expected=1, timeout=3.0)
+ assert len(lines) >= 1
+ assert "test-msg.org" in lines[-1]
+ finally:
+ proc.terminate()
+ proc.wait(timeout=2)
+
+
+def test_watch_filters_tmp_files(isolated_env, tmp_path):
+ """Files starting with .tmp. must NOT trigger log entries."""
+ watched = tmp_path / "watched"
+ watched.mkdir()
+ log = tmp_path / "watch.log"
+ proc = _spawn(watched, log, {**os.environ, "HOME": str(isolated_env)})
+ try:
+ time.sleep(0.3)
+ (watched / ".tmp.staging-file.org").write_text("hello")
+ # Wait briefly to confirm nothing logs.
+ time.sleep(0.5)
+ if log.exists():
+ content = log.read_text()
+ assert ".tmp.staging-file" not in content
+ # Then drop a real file to confirm watcher is alive.
+ (watched / "real.org").write_text("real")
+ lines = _wait_for_log_lines(log, expected=1, timeout=3.0)
+ assert any("real.org" in ln for ln in lines)
+ finally:
+ proc.terminate()
+ proc.wait(timeout=2)
+
+
+def test_watch_filters_asc_sidecars(isolated_env, tmp_path):
+ """Only .org events fire; .asc sidecars are silent."""
+ watched = tmp_path / "watched"
+ watched.mkdir()
+ log = tmp_path / "watch.log"
+ proc = _spawn(watched, log, {**os.environ, "HOME": str(isolated_env)})
+ try:
+ time.sleep(0.3)
+ (watched / "msg.org.asc").write_text("sig")
+ time.sleep(0.5)
+ if log.exists():
+ assert "msg.org.asc" not in log.read_text()
+ # .org event still works.
+ (watched / "msg.org").write_text("body")
+ lines = _wait_for_log_lines(log, expected=1, timeout=3.0)
+ assert any(ln.endswith("msg.org") for ln in lines)
+ finally:
+ proc.terminate()
+ proc.wait(timeout=2)
+
+
+def test_watch_halt_suppresses_but_logs(isolated_env, tmp_path):
+ """When HALT is set, watcher logs the event with (suppressed by HALT) marker."""
+ halt = isolated_env / ".config" / "cross-agent-comms" / "HALT"
+ halt.parent.mkdir(parents=True)
+ halt.write_text("halted")
+ watched = tmp_path / "watched"
+ watched.mkdir()
+ log = tmp_path / "watch.log"
+ proc = _spawn(watched, log, {**os.environ, "HOME": str(isolated_env)})
+ try:
+ time.sleep(0.3)
+ (watched / "halted-event.org").write_text("body")
+ lines = _wait_for_log_lines(log, expected=1, timeout=3.0)
+ assert len(lines) >= 1
+ assert "suppressed by HALT" in lines[-1]
+ finally:
+ proc.terminate()
+ proc.wait(timeout=2)