diff options
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, 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) |
