diff options
| author | Craig Jennings <c@cjennings.net> | 2026-05-06 21:59:52 -0500 |
|---|---|---|
| committer | Craig Jennings <c@cjennings.net> | 2026-05-06 21:59:52 -0500 |
| commit | d81b23ad6b6e437dfe3c338a00a4be39bc555146 (patch) | |
| tree | 2d4b0d7890fd1fc70d81282b81fed2808c28a106 /.ai/scripts/tests/test_cross_agent_watch.py | |
| parent | 201377f57430ef28d02e703a2191434bbee55c75 (diff) | |
| download | rulesets-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.py | 155 |
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) |
