287 lines
12 KiB
Python
287 lines
12 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
"""test_muse_resume_pool.py — focused tests for the per-repo, profile-aware
|
||
|
|
Muse Code resume pool and guard. Fixture-only: never touches the real
|
||
|
|
~/.config/muse or ~/.local/share/muse."""
|
||
|
|
|
||
|
|
import json
|
||
|
|
import os
|
||
|
|
import sqlite3
|
||
|
|
import sys
|
||
|
|
import tempfile
|
||
|
|
import unittest
|
||
|
|
|
||
|
|
REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
|
|
sys.path.insert(0, os.path.join(REPO_ROOT, "bin"))
|
||
|
|
|
||
|
|
import muse_resume_pool as mrp
|
||
|
|
|
||
|
|
|
||
|
|
WS_NETVM = "/home/super/Projects/NetVM"
|
||
|
|
WS_HOME = "/home/super"
|
||
|
|
|
||
|
|
|
||
|
|
def make_row(session_id, workspace, name=None, created_us=1791000000000000,
|
||
|
|
updated_us=1791300000000000, status="valid", title="t"):
|
||
|
|
return {
|
||
|
|
"session_id": session_id,
|
||
|
|
"session_name": name,
|
||
|
|
"workspace_root": workspace,
|
||
|
|
"workspace_key": workspace,
|
||
|
|
"provider_id": "meta",
|
||
|
|
"model_id": "m",
|
||
|
|
"git_branch": None,
|
||
|
|
"title": title,
|
||
|
|
"first_user_prompt": title,
|
||
|
|
"created_at_us": created_us,
|
||
|
|
"updated_at_us": updated_us,
|
||
|
|
"prompt_count": 3,
|
||
|
|
"status": status,
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def make_auth(active="muse", mapping=None, history=None, readable=True):
|
||
|
|
return {
|
||
|
|
"readable": readable,
|
||
|
|
"active": active,
|
||
|
|
"session_profiles": dict(mapping or {}),
|
||
|
|
"switch_history": list(history or []),
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
class TestProfileResolution(unittest.TestCase):
|
||
|
|
def test_cached_mapping_wins_over_history(self):
|
||
|
|
auth = make_auth(
|
||
|
|
mapping={"s1": "def"},
|
||
|
|
history=[{"profile": "muse", "epoch": 1790000000.0}],
|
||
|
|
)
|
||
|
|
profile, source = mrp.resolve_profile("s1", 1791000000000000, auth)
|
||
|
|
self.assertEqual((profile, source), ("def", "cached"))
|
||
|
|
|
||
|
|
def test_history_picks_latest_switch_before_start(self):
|
||
|
|
auth = make_auth(history=[
|
||
|
|
{"profile": "muse", "epoch": 1790000000.0},
|
||
|
|
{"profile": "def", "epoch": 1791100000.0},
|
||
|
|
{"profile": "opm", "epoch": 1791400000.0},
|
||
|
|
])
|
||
|
|
profile, source = mrp.resolve_profile("sx", 1791200000000000, auth)
|
||
|
|
self.assertEqual((profile, source), ("def", "history"))
|
||
|
|
|
||
|
|
def test_history_predating_all_switches_uses_earliest(self):
|
||
|
|
auth = make_auth(history=[
|
||
|
|
{"profile": "def", "epoch": 1791100000.0},
|
||
|
|
{"profile": "opm", "epoch": 1791400000.0},
|
||
|
|
])
|
||
|
|
profile, source = mrp.resolve_profile("sx", 1790000000000000, auth)
|
||
|
|
self.assertEqual((profile, source), ("def", "history"))
|
||
|
|
|
||
|
|
def test_no_history_falls_back_to_active(self):
|
||
|
|
auth = make_auth(active="pip", history=[])
|
||
|
|
profile, source = mrp.resolve_profile("sx", 1791200000000000, auth)
|
||
|
|
self.assertEqual((profile, source), ("pip", "fallback"))
|
||
|
|
|
||
|
|
def test_unreadable_auth_dir_is_unknown(self):
|
||
|
|
auth = make_auth(readable=False)
|
||
|
|
profile, source = mrp.resolve_profile("s1", 1791200000000000, auth)
|
||
|
|
self.assertEqual((profile, source), (None, "unknown"))
|
||
|
|
|
||
|
|
|
||
|
|
class TestWorkspacePool(unittest.TestCase):
|
||
|
|
def test_exact_workspace_match_only(self):
|
||
|
|
rows = [
|
||
|
|
make_row("a" * 32, WS_NETVM, name="n1"),
|
||
|
|
make_row("b" * 32, WS_HOME, name="n2"),
|
||
|
|
make_row("c" * 32, WS_NETVM, name="n3"),
|
||
|
|
]
|
||
|
|
pool = mrp.pool_for_workspace(rows, WS_NETVM)
|
||
|
|
self.assertEqual([r["session_id"] for r in pool], ["a" * 32, "c" * 32])
|
||
|
|
|
||
|
|
def test_subdirectory_does_not_match(self):
|
||
|
|
rows = [make_row("a" * 32, WS_NETVM + "/subdir")]
|
||
|
|
self.assertEqual(mrp.pool_for_workspace(rows, WS_NETVM), [])
|
||
|
|
|
||
|
|
def test_split_hides_only_proven_mismatches(self):
|
||
|
|
rows = [
|
||
|
|
make_row("a" * 32, WS_NETVM),
|
||
|
|
make_row("b" * 32, WS_NETVM),
|
||
|
|
make_row("c" * 32, WS_NETVM),
|
||
|
|
]
|
||
|
|
rows[0]["auth_profile"] = "muse"
|
||
|
|
rows[1]["auth_profile"] = "def"
|
||
|
|
rows[2]["auth_profile"] = None # unknown stays visible, flagged
|
||
|
|
resumable, blocked = mrp.split_resumable(rows, "muse")
|
||
|
|
self.assertEqual([r["session_id"] for r in resumable], ["a" * 32, "c" * 32])
|
||
|
|
self.assertEqual([r["session_id"] for r in blocked], ["b" * 32])
|
||
|
|
|
||
|
|
|
||
|
|
class TestRefResolution(unittest.TestCase):
|
||
|
|
def setUp(self):
|
||
|
|
self.rows = [
|
||
|
|
make_row("01a113f1" + "0" * 24, WS_NETVM, name="alder-halo"),
|
||
|
|
make_row("01a113d8" + "1" * 24, WS_NETVM, name="blue-merak"),
|
||
|
|
make_row("01a113f1" + "2" * 24, WS_NETVM, name="third-wheel"),
|
||
|
|
]
|
||
|
|
|
||
|
|
def test_full_uuid(self):
|
||
|
|
matches, kind = mrp.resolve_ref(self.rows, "01a113f1" + "0" * 24)
|
||
|
|
self.assertEqual((len(matches), kind), (1, "uuid"))
|
||
|
|
|
||
|
|
def test_name(self):
|
||
|
|
matches, kind = mrp.resolve_ref(self.rows, "blue-merak")
|
||
|
|
self.assertEqual((len(matches), kind), (1, "name"))
|
||
|
|
|
||
|
|
def test_prefix(self):
|
||
|
|
matches, kind = mrp.resolve_ref(self.rows, "01a113f100")
|
||
|
|
self.assertEqual((len(matches), kind), (1, "prefix"))
|
||
|
|
|
||
|
|
def test_short_prefix_rejected(self):
|
||
|
|
matches, kind = mrp.resolve_ref(self.rows, "01a1")
|
||
|
|
self.assertEqual((matches, kind), ([], "none"))
|
||
|
|
|
||
|
|
def test_unknown(self):
|
||
|
|
matches, kind = mrp.resolve_ref(self.rows, "nope-missing")
|
||
|
|
self.assertEqual((matches, kind), ([], "none"))
|
||
|
|
|
||
|
|
def test_ambiguous_prefix(self):
|
||
|
|
matches, kind = mrp.resolve_ref(self.rows, "01a113f1")
|
||
|
|
self.assertEqual(len(matches), 2)
|
||
|
|
self.assertEqual(kind, "prefix")
|
||
|
|
|
||
|
|
|
||
|
|
class TestGuard(unittest.TestCase):
|
||
|
|
def setUp(self):
|
||
|
|
self.rows = mrp.annotate_rows([
|
||
|
|
make_row("01a113f1" + "0" * 24, WS_NETVM, name="alder-halo"),
|
||
|
|
make_row("f" * 32, WS_HOME, name="far-away"),
|
||
|
|
], make_auth(active="muse", mapping={
|
||
|
|
"01a113f1" + "0" * 24: "def",
|
||
|
|
"f" * 32: "muse",
|
||
|
|
}))
|
||
|
|
|
||
|
|
def test_wrong_profile_blocked_with_fix(self):
|
||
|
|
d = mrp.check_resume(self.rows, "alder-halo", WS_NETVM,
|
||
|
|
make_auth(active="muse", mapping={
|
||
|
|
"01a113f1" + "0" * 24: "def"}))
|
||
|
|
self.assertFalse(d["ok"])
|
||
|
|
self.assertEqual(d["reason"], "wrong-profile")
|
||
|
|
self.assertIn("muse-auth use def", d["fix"])
|
||
|
|
|
||
|
|
def test_matching_profile_ok(self):
|
||
|
|
auth = make_auth(active="muse", mapping={"01a113f1" + "0" * 24: "muse"})
|
||
|
|
rows = mrp.annotate_rows(
|
||
|
|
[make_row("01a113f1" + "0" * 24, WS_NETVM, name="alder-halo")], auth)
|
||
|
|
d = mrp.check_resume(rows, "alder-halo", WS_NETVM, auth)
|
||
|
|
self.assertTrue(d["ok"])
|
||
|
|
self.assertEqual(d["reason"], "ok")
|
||
|
|
|
||
|
|
def test_wrong_workspace_blocked(self):
|
||
|
|
auth = make_auth(active="muse", mapping={"f" * 32: "muse"})
|
||
|
|
rows = mrp.annotate_rows([make_row("f" * 32, WS_HOME, name="far-away")], auth)
|
||
|
|
d = mrp.check_resume(rows, "far-away", WS_NETVM, auth)
|
||
|
|
self.assertFalse(d["ok"])
|
||
|
|
self.assertEqual(d["reason"], "wrong-workspace")
|
||
|
|
self.assertIn("cd", d["fix"])
|
||
|
|
|
||
|
|
def test_unknown_session(self):
|
||
|
|
d = mrp.check_resume(self.rows, "ghost-nope", WS_NETVM, make_auth())
|
||
|
|
self.assertFalse(d["ok"])
|
||
|
|
self.assertEqual(d["reason"], "unknown-session")
|
||
|
|
|
||
|
|
def test_bad_status_blocked(self):
|
||
|
|
auth = make_auth(active="muse")
|
||
|
|
rows = mrp.annotate_rows(
|
||
|
|
[make_row("d" * 32, WS_NETVM, name="stale-one", status="deleted")], auth)
|
||
|
|
d = mrp.check_resume(rows, "stale-one", WS_NETVM, auth)
|
||
|
|
self.assertFalse(d["ok"])
|
||
|
|
self.assertEqual(d["reason"], "bad-status")
|
||
|
|
|
||
|
|
def test_unknown_profile_warns_not_blocks(self):
|
||
|
|
auth = make_auth(readable=False)
|
||
|
|
rows = mrp.annotate_rows(
|
||
|
|
[make_row("e" * 32, WS_NETVM, name="sandboxed-one")], auth)
|
||
|
|
d = mrp.check_resume(rows, "sandboxed-one", WS_NETVM, auth)
|
||
|
|
self.assertTrue(d["ok"])
|
||
|
|
self.assertIn("unverified", d["detail"].lower())
|
||
|
|
|
||
|
|
|
||
|
|
class TestIndexAndLogSources(unittest.TestCase):
|
||
|
|
def test_load_index_rows_read_only(self):
|
||
|
|
with tempfile.TemporaryDirectory() as tmp:
|
||
|
|
db_path = os.path.join(tmp, "session-index.db")
|
||
|
|
conn = sqlite3.connect(db_path)
|
||
|
|
conn.execute(
|
||
|
|
"CREATE TABLE sessions (session_id TEXT PRIMARY KEY,"
|
||
|
|
" session_name TEXT, workspace_root TEXT, workspace_key TEXT,"
|
||
|
|
" provider_id TEXT, model_id TEXT, git_branch TEXT, title TEXT,"
|
||
|
|
" first_user_prompt TEXT, created_at_us INTEGER,"
|
||
|
|
" updated_at_us INTEGER, prompt_count INTEGER, status TEXT)"
|
||
|
|
)
|
||
|
|
older, newer = make_row("a" * 32, WS_NETVM, name="old",
|
||
|
|
updated_us=1791000000000000), \
|
||
|
|
make_row("b" * 32, WS_NETVM, name="new",
|
||
|
|
updated_us=1791300000000000)
|
||
|
|
for r in (older, newer):
|
||
|
|
conn.execute(
|
||
|
|
"INSERT INTO sessions VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)",
|
||
|
|
tuple(r[c] for c in mrp.INDEX_COLUMNS),
|
||
|
|
)
|
||
|
|
conn.commit()
|
||
|
|
conn.close()
|
||
|
|
rows = mrp.load_index_rows(db_path)
|
||
|
|
self.assertEqual([r["session_name"] for r in rows], ["new", "old"])
|
||
|
|
|
||
|
|
def test_missing_index_returns_none(self):
|
||
|
|
self.assertIsNone(mrp.load_index_rows("/nonexistent/x.db"))
|
||
|
|
|
||
|
|
def test_log_scan_fallback_extracts_workspace_and_name(self):
|
||
|
|
with tempfile.TemporaryDirectory() as tmp:
|
||
|
|
session_id = "9" * 32
|
||
|
|
sdir = os.path.join(tmp, "2026", "10", "07", session_id)
|
||
|
|
os.makedirs(sdir)
|
||
|
|
records = [
|
||
|
|
{"payload_type": "runtime.session.metadata",
|
||
|
|
"recorded_at": 1791000000000000,
|
||
|
|
"payload": {"record": {"workspace_root": WS_NETVM}}},
|
||
|
|
{"payload_type": "session.name.changed",
|
||
|
|
"recorded_at": 1791000001000000,
|
||
|
|
"payload": {"new_name": "scanned-one"}},
|
||
|
|
{"payload_type": "runtime.session",
|
||
|
|
"recorded_at": 1791000002000000,
|
||
|
|
"payload": {"event": {"kind": "started", "prompt": "hello world"}}},
|
||
|
|
]
|
||
|
|
with open(os.path.join(sdir, "session.jsonl"), "w") as fh:
|
||
|
|
for rec in records:
|
||
|
|
fh.write(json.dumps(rec) + "\n")
|
||
|
|
rows = mrp.scan_session_logs(tmp)
|
||
|
|
self.assertEqual(len(rows), 1)
|
||
|
|
row = rows[0]
|
||
|
|
self.assertEqual(row["session_id"], session_id)
|
||
|
|
self.assertEqual(row["workspace_key"], WS_NETVM)
|
||
|
|
self.assertEqual(row["session_name"], "scanned-one")
|
||
|
|
self.assertEqual(row["prompt_count"], 1)
|
||
|
|
|
||
|
|
def test_auth_state_never_reads_auth_json(self):
|
||
|
|
with tempfile.TemporaryDirectory() as tmp:
|
||
|
|
with open(os.path.join(tmp, "active_profile"), "w") as fh:
|
||
|
|
fh.write("muse\n")
|
||
|
|
with open(os.path.join(tmp, "session_profiles.json"), "w") as fh:
|
||
|
|
json.dump({"s1": "muse"}, fh)
|
||
|
|
with open(os.path.join(tmp, "switch_history.jsonl"), "w") as fh:
|
||
|
|
fh.write(json.dumps({"profile": "muse", "epoch": 1.0}) + "\n")
|
||
|
|
# A decoy auth.json with secret bytes must not be opened: make it
|
||
|
|
# unreadable and assert load still succeeds without error.
|
||
|
|
decoy = os.path.join(tmp, "auth.json")
|
||
|
|
with open(decoy, "w") as fh:
|
||
|
|
fh.write('{"secret": true}')
|
||
|
|
os.chmod(decoy, 0)
|
||
|
|
try:
|
||
|
|
state = mrp.load_auth_state(tmp)
|
||
|
|
finally:
|
||
|
|
os.chmod(decoy, 0o600)
|
||
|
|
self.assertTrue(state["readable"])
|
||
|
|
self.assertEqual(state["active"], "muse")
|
||
|
|
self.assertEqual(state["session_profiles"], {"s1": "muse"})
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
unittest.main()
|