118 lines
4.2 KiB
Python
118 lines
4.2 KiB
Python
"""Tests for note_compile.py — the thin drain launcher (pre-check + lock).
|
|
|
|
The nanobot import is deferred inside run_compile, so importing the module and testing
|
|
pending_sources / the lock needs no nanobot-ai install.
|
|
"""
|
|
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "scripts"))
|
|
import note_compile # noqa: E402
|
|
|
|
|
|
@pytest.fixture
|
|
def workspace(tmp_path, monkeypatch):
|
|
notes = tmp_path / "notes"
|
|
inbox = notes / "inbox"
|
|
inbox.mkdir(parents=True)
|
|
monkeypatch.setattr(note_compile, "NOTES", notes)
|
|
monkeypatch.setattr(note_compile, "INBOX", inbox)
|
|
monkeypatch.setattr(note_compile, "LOCK", notes / ".compile.lock")
|
|
monkeypatch.setattr(note_compile, "LOG", tmp_path / "log" / "note_compile_cron.log")
|
|
monkeypatch.setattr(note_compile, "WORKSPACE", tmp_path)
|
|
return tmp_path
|
|
|
|
|
|
def _git(workspace, *args) -> str:
|
|
result = subprocess.run(
|
|
["git", "-C", str(workspace), *args], check=True, capture_output=True, text=True
|
|
)
|
|
return result.stdout
|
|
|
|
|
|
def _init_repo(workspace) -> None:
|
|
_git(workspace, "init", "-q")
|
|
_git(workspace, "config", "user.email", "test@example.com")
|
|
_git(workspace, "config", "user.name", "test")
|
|
|
|
|
|
def test_pending_sources_empty(workspace):
|
|
assert note_compile.pending_sources() == []
|
|
|
|
|
|
def test_pending_sources_ignores_hidden(workspace):
|
|
inbox = workspace / "notes" / "inbox"
|
|
(inbox / "2026-07-01_10_00_00_000000-a.md").write_text("x")
|
|
(inbox / ".hidden.tmp").write_text("x")
|
|
names = [p.name for p in note_compile.pending_sources()]
|
|
assert names == ["2026-07-01_10_00_00_000000-a.md"]
|
|
|
|
|
|
def test_pending_sources_sorted(workspace):
|
|
inbox = workspace / "notes" / "inbox"
|
|
for stem in ("2026-07-01_10_00_02_000000-c", "2026-07-01_10_00_01_000000-b"):
|
|
(inbox / f"{stem}.md").write_text("x")
|
|
names = [p.name for p in note_compile.pending_sources()]
|
|
assert names == sorted(names)
|
|
|
|
|
|
def test_lock_acquire_then_blocked(workspace):
|
|
assert note_compile.acquire_lock() is True
|
|
assert note_compile.acquire_lock() is False # live lock held by this pid
|
|
|
|
|
|
def test_lock_reclaims_dead_pid(workspace):
|
|
lock = workspace / "notes" / ".compile.lock"
|
|
lock.write_text('{"pid": 999999, "started": "2999-01-01T00:00:00+00:00"}')
|
|
assert note_compile.acquire_lock() is True # dead pid -> reclaimed
|
|
|
|
|
|
def test_lock_reclaims_unparseable(workspace):
|
|
(workspace / "notes" / ".compile.lock").write_text("garbage")
|
|
assert note_compile.acquire_lock() is True
|
|
|
|
|
|
def test_main_empty_inbox_returns_zero_without_lock(workspace):
|
|
# No pending work -> exit before acquire_lock / nanobot import.
|
|
assert note_compile.main() == 0
|
|
assert not (workspace / "notes" / ".compile.lock").exists()
|
|
|
|
|
|
def test_main_dry_run_releases_lock(workspace, monkeypatch):
|
|
(workspace / "notes" / "inbox" / "2026-07-01_10_00_00_000000-a.md").write_text("x")
|
|
monkeypatch.setattr(sys, "argv", ["note_compile.py", "--dry-run"])
|
|
assert note_compile.main() == 0
|
|
assert not (workspace / "notes" / ".compile.lock").exists()
|
|
|
|
|
|
def test_commit_notes_creates_commit(workspace):
|
|
_init_repo(workspace)
|
|
(workspace / "notes" / "notes.md").write_text("# Notes\n\n## X\n\n- a fact\n")
|
|
note_compile.commit_notes(2)
|
|
assert "note: cron drain (2 captures)" in _git(workspace, "log", "--oneline")
|
|
assert "notes/notes.md" in _git(workspace, "ls-files", "notes/")
|
|
|
|
|
|
def test_commit_notes_noop_when_clean(workspace):
|
|
_init_repo(workspace)
|
|
(workspace / "notes" / "notes.md").write_text("# Notes\n")
|
|
note_compile.commit_notes(1)
|
|
before = _git(workspace, "rev-list", "--count", "HEAD").strip()
|
|
note_compile.commit_notes(1) # nothing changed since last commit
|
|
assert _git(workspace, "rev-list", "--count", "HEAD").strip() == before
|
|
|
|
|
|
def test_commit_notes_skips_gitignored_lock(workspace):
|
|
_init_repo(workspace)
|
|
(workspace / ".gitignore").write_text("notes/.compile.lock\n")
|
|
(workspace / "notes" / ".compile.lock").write_text('{"pid": 1}')
|
|
(workspace / "notes" / "notes.md").write_text("# Notes\n")
|
|
note_compile.commit_notes(1)
|
|
tracked = _git(workspace, "ls-files", "notes/")
|
|
assert "notes/notes.md" in tracked
|
|
assert ".compile.lock" not in tracked
|