mirror of
https://github.com/headroomlabs-ai/headroom.git
synced 2026-08-27 14:17:10 -04:00
efd2ac1ca4
`.gitattributes` declares `*.py text eol=lf` and `*.sh text eol=lf`, but 74 files (73 .py, 1 .sh) are stored in the index with CRLF line endings, violating that contract. Every macOS/Linux clone reports these files as "modified" on fresh checkout because git's diff engine sees the stored bytes don't match the attribute contract, even though the working tree and index match byte-for-byte. Running `git add --renormalize .` rewrites each affected blob so the stored form matches the attribute declaration. No semantic changes — every affected file's diff is "N insertions, N deletions" with inserts and deletes being the same lines modulo line endings. Follow-up commit adds `.git-blame-ignore-revs` so `git blame` / GitHub blame skip this mechanical commit. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
352 lines
12 KiB
Python
352 lines
12 KiB
Python
from __future__ import annotations
|
|
|
|
import io
|
|
import json
|
|
import subprocess
|
|
import tarfile
|
|
from pathlib import Path
|
|
from types import ModuleType, SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from headroom.graph import installer, watcher
|
|
|
|
|
|
def _build_archive(member_name: str = installer.CBM_BIN_NAME) -> bytes:
|
|
payload = io.BytesIO()
|
|
with tarfile.open(fileobj=payload, mode="w:gz") as tar:
|
|
data = b"#!/bin/sh\necho version\n"
|
|
info = tarfile.TarInfo(name=member_name)
|
|
info.size = len(data)
|
|
tar.addfile(info, io.BytesIO(data))
|
|
return payload.getvalue()
|
|
|
|
|
|
class FakeResponse:
|
|
def __init__(self, data: bytes) -> None:
|
|
self._data = data
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc, tb) -> None:
|
|
return None
|
|
|
|
def read(self) -> bytes:
|
|
return self._data
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("system", "machine", "expected"),
|
|
[
|
|
("Darwin", "arm64", "darwin-arm64"),
|
|
("Darwin", "x86_64", "darwin-amd64"),
|
|
("Linux", "aarch64", "linux-arm64"),
|
|
("Linux", "arm64", "linux-arm64"),
|
|
("Linux", "x86_64", "linux-amd64"),
|
|
("Windows", "AMD64", "windows-amd64"),
|
|
],
|
|
)
|
|
def test_detect_platform_variants(monkeypatch, system: str, machine: str, expected: str) -> None:
|
|
monkeypatch.setattr(installer.platform, "system", lambda: system)
|
|
monkeypatch.setattr(installer.platform, "machine", lambda: machine)
|
|
assert installer._detect_platform() == expected
|
|
|
|
|
|
def test_detect_platform_rejects_unknown_system(monkeypatch) -> None:
|
|
monkeypatch.setattr(installer.platform, "system", lambda: "Solaris")
|
|
monkeypatch.setattr(installer.platform, "machine", lambda: "sparc")
|
|
with pytest.raises(RuntimeError, match="Unsupported platform"):
|
|
installer._detect_platform()
|
|
|
|
|
|
def test_get_cbm_path_prefers_path_then_install_dir(monkeypatch, tmp_path: Path) -> None:
|
|
on_path = tmp_path / "on-path"
|
|
installed = tmp_path / installer.CBM_BIN_NAME
|
|
installed.write_text("bin")
|
|
monkeypatch.setattr(installer, "CBM_BIN_DIR", tmp_path)
|
|
monkeypatch.setattr(installer.shutil, "which", lambda name: str(on_path))
|
|
assert installer.get_cbm_path() == on_path
|
|
|
|
monkeypatch.setattr(installer.shutil, "which", lambda name: None)
|
|
assert installer.get_cbm_path() == installed
|
|
|
|
installed.unlink()
|
|
assert installer.get_cbm_path() is None
|
|
|
|
|
|
def test_download_cbm_success_and_verification_paths(monkeypatch, tmp_path: Path) -> None:
|
|
monkeypatch.setattr(installer, "CBM_BIN_DIR", tmp_path)
|
|
monkeypatch.setattr(installer, "_detect_platform", lambda: "linux-amd64")
|
|
monkeypatch.setattr(
|
|
installer, "urlopen", lambda url, timeout=60: FakeResponse(_build_archive())
|
|
)
|
|
|
|
run_calls: list[list[str]] = []
|
|
|
|
def fake_run(command, **kwargs):
|
|
run_calls.append(command)
|
|
return SimpleNamespace(returncode=1, stdout="")
|
|
|
|
monkeypatch.setattr("subprocess.run", fake_run)
|
|
path = installer.download_cbm(version="v1.2.3")
|
|
assert path == tmp_path / installer.CBM_BIN_NAME
|
|
assert path.exists()
|
|
assert run_calls == [[str(path), "--version"]]
|
|
|
|
monkeypatch.setattr(
|
|
"subprocess.run", lambda *args, **kwargs: (_ for _ in ()).throw(RuntimeError("boom"))
|
|
)
|
|
assert installer.download_cbm(version="v1.2.3") == path
|
|
|
|
monkeypatch.setattr(
|
|
"subprocess.run",
|
|
lambda *args, **kwargs: SimpleNamespace(returncode=0, stdout="cbm v1.2.3\n"),
|
|
)
|
|
assert installer.download_cbm(version="v1.2.3") == path
|
|
|
|
|
|
def test_download_cbm_invalid_url_download_failure_and_extract_errors(
|
|
monkeypatch, tmp_path: Path
|
|
) -> None:
|
|
monkeypatch.setattr(installer, "CBM_BIN_DIR", tmp_path)
|
|
monkeypatch.setattr(installer, "_detect_platform", lambda: "linux-amd64")
|
|
|
|
monkeypatch.setattr(installer, "GITHUB_RELEASE_URL", "ftp://example.test/releases")
|
|
with pytest.raises(RuntimeError, match="Invalid URL"):
|
|
installer.download_cbm()
|
|
|
|
monkeypatch.setattr(installer, "GITHUB_RELEASE_URL", "https://example.test/releases")
|
|
monkeypatch.setattr(
|
|
installer,
|
|
"urlopen",
|
|
lambda url, timeout=60: (_ for _ in ()).throw(OSError("network down")),
|
|
)
|
|
with pytest.raises(RuntimeError, match="Failed to download codebase-memory-mcp"):
|
|
installer.download_cbm()
|
|
|
|
monkeypatch.setattr(
|
|
installer,
|
|
"urlopen",
|
|
lambda url, timeout=60: FakeResponse(_build_archive("some/other-binary")),
|
|
)
|
|
with pytest.raises(RuntimeError, match="binary not found in archive"):
|
|
installer.download_cbm()
|
|
|
|
monkeypatch.setattr(installer, "urlopen", lambda url, timeout=60: FakeResponse(b"not a tar"))
|
|
with pytest.raises(RuntimeError, match="Failed to extract archive"):
|
|
installer.download_cbm()
|
|
|
|
|
|
def test_ensure_cbm_uses_existing_or_returns_none_on_failure(monkeypatch, tmp_path: Path) -> None:
|
|
existing = tmp_path / installer.CBM_BIN_NAME
|
|
monkeypatch.setattr(installer, "get_cbm_path", lambda: existing)
|
|
assert installer.ensure_cbm() == existing
|
|
|
|
monkeypatch.setattr(installer, "get_cbm_path", lambda: None)
|
|
monkeypatch.setattr(
|
|
installer, "download_cbm", lambda: (_ for _ in ()).throw(RuntimeError("nope"))
|
|
)
|
|
assert installer.ensure_cbm() is None
|
|
|
|
|
|
def test_code_graph_watcher_init_start_stop_and_event_filtering(
|
|
monkeypatch, tmp_path: Path
|
|
) -> None:
|
|
monkeypatch.setattr("headroom.graph.installer.get_cbm_path", lambda: tmp_path / "cbm")
|
|
graph_watcher = watcher.CodeGraphWatcher(tmp_path)
|
|
assert graph_watcher.cbm_binary == str(tmp_path / "cbm")
|
|
|
|
explicit = watcher.CodeGraphWatcher(tmp_path, cbm_binary="explicit-cbm")
|
|
assert explicit.cbm_binary == "explicit-cbm"
|
|
|
|
missing = watcher.CodeGraphWatcher(tmp_path, cbm_binary=None)
|
|
missing.cbm_binary = None
|
|
assert missing.start() is False
|
|
|
|
watchdog_mod = ModuleType("watchdog")
|
|
events_mod = ModuleType("watchdog.events")
|
|
observers_mod = ModuleType("watchdog.observers")
|
|
|
|
class FileSystemEventHandler:
|
|
pass
|
|
|
|
class FakeObserver:
|
|
def __init__(self) -> None:
|
|
self.scheduled = None
|
|
self.daemon = False
|
|
self.started = False
|
|
self.stopped = False
|
|
self.join_timeout = None
|
|
|
|
def schedule(self, handler, project_dir, recursive=True) -> None:
|
|
self.scheduled = (handler, project_dir, recursive)
|
|
|
|
def start(self) -> None:
|
|
self.started = True
|
|
|
|
def stop(self) -> None:
|
|
self.stopped = True
|
|
|
|
def join(self, timeout=None) -> None:
|
|
self.join_timeout = timeout
|
|
|
|
events_mod.FileSystemEventHandler = FileSystemEventHandler
|
|
observers_mod.Observer = FakeObserver
|
|
monkeypatch.setitem(__import__("sys").modules, "watchdog", watchdog_mod)
|
|
monkeypatch.setitem(__import__("sys").modules, "watchdog.events", events_mod)
|
|
monkeypatch.setitem(__import__("sys").modules, "watchdog.observers", observers_mod)
|
|
|
|
scheduled: list[str] = []
|
|
monkeypatch.setattr(graph_watcher, "_schedule_reindex", lambda: scheduled.append("reindex"))
|
|
|
|
assert graph_watcher.start() is True
|
|
handler, project_dir, recursive = graph_watcher._observer.scheduled
|
|
assert project_dir == str(tmp_path)
|
|
assert recursive is True
|
|
|
|
handler.on_any_event(SimpleNamespace(src_path=""))
|
|
handler.on_any_event(SimpleNamespace(src_path=str(tmp_path / ".git" / "config")))
|
|
handler.on_any_event(SimpleNamespace(src_path=str(tmp_path / "notes.txt")))
|
|
handler.on_any_event(SimpleNamespace(src_path=str(tmp_path / ".temp.py")))
|
|
handler.on_any_event(SimpleNamespace(src_path=str(tmp_path / "main.py~")))
|
|
handler.on_any_event(SimpleNamespace(src_path=str(tmp_path / "main.py")))
|
|
assert scheduled == ["reindex"]
|
|
|
|
class FakeTimer:
|
|
def __init__(self) -> None:
|
|
self.cancelled = False
|
|
|
|
def cancel(self) -> None:
|
|
self.cancelled = True
|
|
|
|
timer = FakeTimer()
|
|
graph_watcher._debounce_timer = timer
|
|
graph_watcher._reindex_count = 1
|
|
graph_watcher.stop()
|
|
assert timer.cancelled is True
|
|
assert graph_watcher._observer is None
|
|
|
|
|
|
def test_code_graph_watcher_start_returns_false_without_watchdog(
|
|
monkeypatch, tmp_path: Path
|
|
) -> None:
|
|
graph_watcher = watcher.CodeGraphWatcher(tmp_path, cbm_binary="cbm")
|
|
|
|
import builtins
|
|
|
|
real_import = builtins.__import__
|
|
|
|
def fake_import(name, globals=None, locals=None, fromlist=(), level=0):
|
|
if name.startswith("watchdog"):
|
|
raise ImportError("missing watchdog")
|
|
return real_import(name, globals, locals, fromlist, level)
|
|
|
|
monkeypatch.setattr(builtins, "__import__", fake_import)
|
|
assert graph_watcher.start() is False
|
|
|
|
|
|
def test_code_graph_watcher_stop_handles_missing_timer_and_observer_methods(tmp_path: Path) -> None:
|
|
graph_watcher = watcher.CodeGraphWatcher(tmp_path, cbm_binary="cbm")
|
|
graph_watcher._observer = object()
|
|
graph_watcher.stop()
|
|
assert graph_watcher._observer is None
|
|
|
|
graph_watcher.stop()
|
|
|
|
|
|
def test_schedule_reindex_replaces_existing_timer(monkeypatch, tmp_path: Path) -> None:
|
|
graph_watcher = watcher.CodeGraphWatcher(tmp_path, debounce_seconds=3.5, cbm_binary="cbm")
|
|
timers: list[FakeTimer] = []
|
|
|
|
class FakeTimer:
|
|
def __init__(self, interval, callback) -> None:
|
|
self.interval = interval
|
|
self.callback = callback
|
|
self.daemon = False
|
|
self.started = False
|
|
self.cancelled = False
|
|
timers.append(self)
|
|
|
|
def start(self) -> None:
|
|
self.started = True
|
|
|
|
def cancel(self) -> None:
|
|
self.cancelled = True
|
|
|
|
monkeypatch.setattr(watcher.threading, "Timer", FakeTimer)
|
|
graph_watcher._schedule_reindex()
|
|
graph_watcher._schedule_reindex()
|
|
|
|
assert len(timers) == 2
|
|
assert timers[0].cancelled is True
|
|
assert timers[1].started is True
|
|
assert timers[1].daemon is True
|
|
assert timers[1].interval == 3.5
|
|
|
|
|
|
def test_do_reindex_success_failure_timeout_and_stats(monkeypatch, tmp_path: Path) -> None:
|
|
graph_watcher = watcher.CodeGraphWatcher(tmp_path, cbm_binary="cbm")
|
|
graph_watcher._running = True
|
|
|
|
monotonic_values = iter([10.0, 10.4, 20.0, 20.5, 30.0, 30.5, 40.0, 40.5])
|
|
monkeypatch.setattr(watcher.time, "monotonic", lambda: next(monotonic_values))
|
|
monkeypatch.setattr(watcher.time, "time", lambda: 1234.0)
|
|
|
|
run_calls: list[list[str]] = []
|
|
|
|
def success_run(command, **kwargs):
|
|
run_calls.append(command)
|
|
return SimpleNamespace(returncode=0, stderr="indexed\nchanged=7 files\n")
|
|
|
|
monkeypatch.setattr(watcher.subprocess, "run", success_run)
|
|
graph_watcher._do_reindex()
|
|
assert graph_watcher.stats == {
|
|
"running": True,
|
|
"project_dir": str(tmp_path),
|
|
"reindex_count": 1,
|
|
"last_reindex": 1234.0,
|
|
"debounce_seconds": 2.0,
|
|
}
|
|
assert run_calls == [
|
|
["cbm", "cli", "index_repository", json.dumps({"repo_path": str(tmp_path), "mode": "fast"})]
|
|
]
|
|
|
|
monkeypatch.setattr(
|
|
watcher.subprocess,
|
|
"run",
|
|
lambda command, **kwargs: SimpleNamespace(returncode=1, stderr="failed"),
|
|
)
|
|
graph_watcher._do_reindex()
|
|
assert graph_watcher._reindex_count == 2
|
|
|
|
monkeypatch.setattr(
|
|
watcher.subprocess,
|
|
"run",
|
|
lambda command, **kwargs: SimpleNamespace(
|
|
returncode=0, stderr="indexed\nchanged=oops\nstill running\n"
|
|
),
|
|
)
|
|
graph_watcher._do_reindex()
|
|
assert graph_watcher._reindex_count == 3
|
|
|
|
monkeypatch.setattr(
|
|
watcher.subprocess,
|
|
"run",
|
|
lambda command, **kwargs: (_ for _ in ()).throw(subprocess.TimeoutExpired(command, 30)),
|
|
)
|
|
graph_watcher._do_reindex()
|
|
|
|
monkeypatch.setattr(
|
|
watcher.subprocess,
|
|
"run",
|
|
lambda command, **kwargs: (_ for _ in ()).throw(RuntimeError("boom")),
|
|
)
|
|
graph_watcher._do_reindex()
|
|
|
|
graph_watcher._running = False
|
|
graph_watcher._do_reindex()
|
|
|
|
graph_watcher._running = True
|
|
graph_watcher.cbm_binary = None
|
|
graph_watcher._do_reindex()
|