aboutsummaryrefslogtreecommitdiff
path: root/.ai/scripts/tests/test_cross_agent_watch.py
diff options
context:
space:
mode:
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, 0 insertions, 155 deletions
diff --git a/.ai/scripts/tests/test_cross_agent_watch.py b/.ai/scripts/tests/test_cross_agent_watch.py
deleted file mode 100644
index 417cc19..0000000
--- a/.ai/scripts/tests/test_cross_agent_watch.py
+++ /dev/null
@@ -1,155 +0,0 @@
-"""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)