150 lines
6.2 KiB
Python
150 lines
6.2 KiB
Python
"""Project recall boundaries use real state/project stores, never transcript reads."""
|
|
|
|
import importlib
|
|
import json
|
|
import sqlite3
|
|
import subprocess
|
|
from contextlib import contextmanager
|
|
|
|
import pytest
|
|
|
|
from hermes_cli import projects_db
|
|
from hermes_state import SessionDB
|
|
|
|
|
|
@pytest.fixture
|
|
def stores(tmp_path, monkeypatch):
|
|
home = tmp_path / "profile"
|
|
home.mkdir()
|
|
monkeypatch.setenv("HERMES_HOME", str(home))
|
|
db = SessionDB(db_path=home / "state.db")
|
|
with projects_db.connect_closing() as projects:
|
|
yield db, projects, tmp_path
|
|
db.close()
|
|
|
|
|
|
def _resolve(db, sid):
|
|
# Import inside the test so RED is an assertion about the missing API.
|
|
import importlib.util
|
|
assert importlib.util.find_spec("hermes_project_scope"), "core scope API is missing"
|
|
return importlib.import_module("hermes_project_scope").resolve_project_scope(db, sid)
|
|
|
|
|
|
def _session(db, sid, cwd, source="cli", **kwargs):
|
|
db.create_session(sid, source, cwd=str(cwd) if cwd else None, **kwargs)
|
|
|
|
|
|
def test_nested_ownership_enumerates_all_metadata_without_reading_text(stores, monkeypatch):
|
|
db, projects, root = stores
|
|
outer, inner = root / "A", root / "A" / "sub"
|
|
inner.mkdir(parents=True)
|
|
p = projects_db.create_project(projects, name="P", folders=[str(outer)])
|
|
q = projects_db.create_project(projects, name="Q", folders=[str(inner)])
|
|
# Global active pointer must not override the actual session's ownership.
|
|
projects_db.set_active(projects, q)
|
|
_session(db, "current", outer)
|
|
_session(db, "nested", inner)
|
|
_session(db, "prefix", root / "AB")
|
|
with db._lock:
|
|
conn = db._conn
|
|
conn.executemany(
|
|
"INSERT INTO sessions(id, source, started_at, cwd) VALUES (?, 'cli', 0, ?)",
|
|
[(f"old-{i:05d}", str(outer)) for i in range(5001)],
|
|
)
|
|
conn.commit()
|
|
reads = []
|
|
|
|
def authorize(action, table, column, database, trigger):
|
|
if action == sqlite3.SQLITE_READ:
|
|
reads.append((table, column))
|
|
if table != "sessions" or column in {"system_prompt", "preview", "title"}:
|
|
return sqlite3.SQLITE_DENY
|
|
return sqlite3.SQLITE_OK
|
|
|
|
read_ctx = db._read_ctx
|
|
|
|
@contextmanager
|
|
def audited_read():
|
|
with read_ctx() as conn:
|
|
conn.set_authorizer(authorize)
|
|
try:
|
|
yield conn
|
|
finally:
|
|
conn.set_authorizer(None)
|
|
|
|
with monkeypatch.context() as patch:
|
|
patch.setattr(db, "_read_ctx", audited_read)
|
|
scope = _resolve(db, "current")
|
|
assert scope["status"] == "ready"
|
|
assert scope["project_key"] == p
|
|
assert scope["label"] == "P"
|
|
assert set(scope["allowed_session_ids"]) == {"current", *(f"old-{i:05d}" for i in range(5001))}
|
|
assert reads and all(table == "sessions" for table, _ in reads)
|
|
assert json.loads(json.dumps(scope)) == scope
|
|
nested = _resolve(db, "nested")
|
|
assert nested["project_key"] == q
|
|
assert nested["allowed_session_ids"] == ["nested"]
|
|
assert scope["revision"] == _resolve(db, "current")["revision"]
|
|
_session(db, "new", outer)
|
|
assert scope["revision"] != _resolve(db, "current")["revision"]
|
|
_session(db, "known-user", outer, user_id="alice")
|
|
_session(db, "alice-peer", outer, user_id="alice")
|
|
_session(db, "bob", outer, user_id="bob")
|
|
for source in ("kanban", "subagent", "tool", "cron"):
|
|
_session(db, source, outer, source=source, user_id="alice")
|
|
_session(db, "hidden", outer, user_id="alice")
|
|
_session(db, "archived", outer, user_id="alice")
|
|
with db._lock:
|
|
conn = db._conn
|
|
conn.execute("UPDATE sessions SET hidden = 1 WHERE id = 'hidden'")
|
|
conn.execute("UPDATE sessions SET archived = 1 WHERE id = 'archived'")
|
|
conn.commit()
|
|
assert _resolve(db, "known-user")["allowed_session_ids"] == ["alice-peer", "archived", "known-user"]
|
|
assert "known-user" not in _resolve(db, "current")["allowed_session_ids"]
|
|
assert _resolve(db, "current")["coverage"]["principal_policy"] == "unknown_local_only"
|
|
|
|
|
|
def test_auto_home_and_real_linked_worktree_membership(stores):
|
|
db, projects, root = stores
|
|
notes = root / "notes"
|
|
notes.mkdir()
|
|
_session(db, "notes", notes)
|
|
_session(db, "notes-peer", notes)
|
|
assert _resolve(db, "notes")["allowed_session_ids"] == ["notes", "notes-peer"]
|
|
assert _resolve(db, "notes")["label"] == "notes"
|
|
_session(db, "home", None)
|
|
assert _resolve(db, "home")["status"] == "scope_unresolved"
|
|
assert _resolve(db, "home")["allowed_session_ids"] == []
|
|
assert _resolve(db, "absent")["status"] == "scope_unresolved"
|
|
repo, worktree = root / "repo", root / "linked"
|
|
subprocess.run(["git", "init", str(repo)], check=True, capture_output=True)
|
|
subprocess.run(["git", "-C", str(repo), "-c", "user.name=Test", "-c",
|
|
"user.email=test@example.invalid", "commit", "--allow-empty", "-m", "init"],
|
|
check=True, capture_output=True)
|
|
subprocess.run(["git", "-C", str(repo), "worktree", "add", "-b", "linked", str(worktree)],
|
|
check=True, capture_output=True)
|
|
_session(db, "repo", repo)
|
|
_session(db, "linked", worktree)
|
|
linked = _resolve(db, "linked")
|
|
assert linked["allowed_session_ids"] == ["linked", "repo"]
|
|
assert linked["project_key"] == _resolve(db, "repo")["project_key"]
|
|
|
|
|
|
def test_equal_depth_owners_deny_regardless_of_order_and_revision_tracks_projects(stores):
|
|
db, projects, root = stores
|
|
folder = root / "shared"
|
|
folder.mkdir()
|
|
_session(db, "current", folder)
|
|
p = projects_db.create_project(projects, name="P", folders=[str(folder)])
|
|
before = _resolve(db, "current")
|
|
q = projects_db.create_project(projects, name="Q", folders=[str(folder)], allow_duplicate_path=True)
|
|
ambiguous = _resolve(db, "current")
|
|
assert ambiguous["status"] == "scope_ambiguous"
|
|
assert ambiguous["project_key"] is None
|
|
assert ambiguous["allowed_session_ids"] == []
|
|
assert ambiguous["revision"] != before["revision"]
|
|
projects.execute("UPDATE projects SET created_at = CASE id WHEN ? THEN 100 ELSE 0 END", (p,))
|
|
projects.commit()
|
|
assert _resolve(db, "current")["status"] == "scope_ambiguous"
|
|
projects_db.archive_project(projects, q)
|
|
assert _resolve(db, "current")["project_key"] == p |