#!/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()