"""Seven-tier tests for BranchMeta created_by/created_at provenance (musehub#216). Freeze: ``docs/MUSE-216-BRANCHMETA-CREATED-BY-FREEZE.md`` §5. Tiers ----- 1 unit — resolve / write-once / load-dump / move / read helper 2 integration — plain ``muse branch`` stamps TOML; tip ``created_by`` unchanged 3 e2e — human create then agent tip; copy restamp; rename moves meta 4 stress — 200 creates; every list entry has both new keys 5 data-integrity — stamp twice; list stable; corrupt kind; delete removes section 6 performance — 200-branch list < 5s 7 security — ESC stripped; no key material; quoted handle round-trips """ from __future__ import annotations import json import pathlib import time import tomllib from collections.abc import Mapping from unittest import mock import pytest from tests.cli_test_helper import CliRunner cli = None runner = CliRunner() _HUB_URL = "https://localhost:1337" def _env(root: pathlib.Path) -> Mapping[str, str]: return {"MUSE_REPO_ROOT": str(root)} def _branch_json(root: pathlib.Path) -> list[dict]: result = runner.invoke(cli, ["branch", "--json"], env=_env(root)) assert result.exit_code == 0, f"branch --json failed:\n{result.output}" data = json.loads(result.output.strip()) assert isinstance(data, list) return data def _entry(root: pathlib.Path, name: str) -> dict: return next(b for b in _branch_json(root) if b["name"] == name) def _toml_branch(root: pathlib.Path, name: str) -> Mapping[str, object]: from muse.core.paths import config_toml_path with config_toml_path(root).open("rb") as f: data = tomllib.load(f) return data.get("branch", {}).get(name, {}) def _set_user_handle(root: pathlib.Path, handle: str) -> None: """Wire identity so ``get_config_value("user.handle", root)`` returns *handle*.""" from muse.cli.config import set_hub_url from muse.core.identity import save_identity set_hub_url(_HUB_URL, root) save_identity(_HUB_URL, { "type": "human", "handle": handle, "algorithm": "ed25519", "fingerprint": "0" * 64, "capabilities": [], "provisioned_by": "", "hd_path": "", "provisioned_by_fingerprint": "", }) @pytest.fixture() def repo(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path: monkeypatch.chdir(tmp_path) monkeypatch.delenv("MUSE_AGENT_ID", raising=False) env = _env(tmp_path) r = runner.invoke(cli, ["init", "--domain", "code"], env=env) assert r.exit_code == 0, r.output (tmp_path / "a.py").write_text("x = 1\n") runner.invoke(cli, ["code", "add", "a.py"], env=env) cr = runner.invoke(cli, ["commit", "-m", "initial", "--author", "seed"], env=env) assert cr.exit_code == 0, cr.output return tmp_path # --------------------------------------------------------------------------- # Tier 1 — unit # --------------------------------------------------------------------------- class TestResolveAndWriteOnceUnit: """Tier 1: helpers in ``muse.cli.config`` and list read helper.""" def test_resolve_agent_env_wins( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: from muse.cli.config import _resolve_branch_creator _set_user_handle(repo, "aaronrene") monkeypatch.setenv("MUSE_AGENT_ID", "claude-code") assert _resolve_branch_creator(repo) == ("claude-code", "agent") def test_resolve_human_from_handle( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: from muse.cli.config import _resolve_branch_creator monkeypatch.delenv("MUSE_AGENT_ID", raising=False) _set_user_handle(repo, "aaronrene") assert _resolve_branch_creator(repo) == ("aaronrene", "human") def test_resolve_both_empty_returns_none( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: from muse.cli.config import _resolve_branch_creator monkeypatch.delenv("MUSE_AGENT_ID", raising=False) with mock.patch( "muse.cli.config.get_config_value", return_value=None, ): assert _resolve_branch_creator(repo) is None def test_resolve_whitespace_env_falls_to_human( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: from muse.cli.config import _resolve_branch_creator monkeypatch.setenv("MUSE_AGENT_ID", " ") _set_user_handle(repo, "aaronrene") assert _resolve_branch_creator(repo) == ("aaronrene", "human") def test_resolve_esc_only_env_returns_none( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: from muse.cli.config import _resolve_branch_creator monkeypatch.setenv("MUSE_AGENT_ID", "\x1b\x07\x1f") with mock.patch( "muse.cli.config.get_config_value", return_value=None, ): assert _resolve_branch_creator(repo) is None def test_write_once_second_stamp_preserves_created_at( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: from muse.cli.config import read_branch_meta, stamp_branch_created, write_branch_meta monkeypatch.delenv("MUSE_AGENT_ID", raising=False) _set_user_handle(repo, "aaronrene") stamp_branch_created(repo, "feat/once") first = read_branch_meta(repo, "feat/once")["created_at"] assert first write_branch_meta(repo, "feat/once", intent="later") stamp_branch_created(repo, "feat/once") second = read_branch_meta(repo, "feat/once") assert second["created_at"] == first assert second["created_by"] == "aaronrene" assert second["intent"] == "later" def test_load_dump_round_trip_creation_and_intent( self, repo: pathlib.Path ) -> None: from muse.cli.config import ( _dump_toml, _load_config, write_branch_meta, ) from muse.core.paths import config_toml_path write_branch_meta( repo, "feat/rt", intent="keep me", created_by='alice"bob', created_by_kind="human", created_at="2026-09-17T10:54:10.123456+00:00", ) # Preserve a remote key via a second write that must not drop creation. write_branch_meta(repo, "feat/rt", resumable=True) # Inject remote by rewriting through load/dump helpers. cp = config_toml_path(repo) cfg = _load_config(cp) cfg["branch"]["feat/rt"]["remote"] = "origin" cp.write_text(_dump_toml(cfg), encoding="utf-8") again = _load_config(cp) sec = again["branch"]["feat/rt"] assert sec["intent"] == "keep me" assert sec["remote"] == "origin" assert sec["created_by"] == 'alice"bob' assert sec["created_by_kind"] == "human" assert sec["created_at"] == "2026-09-17T10:54:10.123456+00:00" with cp.open("rb") as f: tomllib.load(f) # must remain parseable after escape def test_move_branch_meta_relocates_dict(self, repo: pathlib.Path) -> None: from muse.cli.config import ( move_branch_meta, read_branch_meta, write_branch_meta, ) write_branch_meta( repo, "old/name", intent="move me", created_by="aaronrene", created_by_kind="human", created_at="2026-09-17T11:00:00+00:00", ) move_branch_meta(repo, "old/name", "new/name") assert read_branch_meta(repo, "old/name") == {} moved = read_branch_meta(repo, "new/name") assert moved["intent"] == "move me" assert moved["created_by"] == "aaronrene" assert moved["created_at"] == "2026-09-17T11:00:00+00:00" def test_branch_created_from_meta_invalid_kind_null(self) -> None: from muse.cli.commands.branch import _branch_created_from_meta by, at = _branch_created_from_meta( { "created_by": "aaronrene", "created_by_kind": "robot", "created_at": "2026-09-17T11:00:00+00:00", } ) assert by is None assert at == "2026-09-17T11:00:00+00:00" def test_branch_created_from_meta_naive_at_null(self) -> None: from muse.cli.commands.branch import _branch_created_from_meta by, at = _branch_created_from_meta( { "created_by": "aaronrene", "created_by_kind": "human", "created_at": "2026-09-17T11:00:00", } ) assert by == {"handle": "aaronrene", "kind": "human"} assert at is None # --------------------------------------------------------------------------- # Tier 2 — integration # --------------------------------------------------------------------------- class TestStampIntegration: def test_plain_branch_stamps_toml_and_json( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: monkeypatch.delenv("MUSE_AGENT_ID", raising=False) _set_user_handle(repo, "aaronrene") env = _env(repo) r = runner.invoke(cli, ["branch", "feat/x"], env=env) assert r.exit_code == 0, r.output sec = _toml_branch(repo, "feat/x") assert "created_at" in sec assert sec["created_by"] == "aaronrene" assert sec["created_by_kind"] == "human" entry = _entry(repo, "feat/x") assert entry["branch_created_at"] is not None assert entry["branch_created_by"] == { "handle": "aaronrene", "kind": "human", } # Tip authorship (#215) stays tip author, not BranchMeta handle object. assert entry["created_by"] == "seed" assert isinstance(entry["created_by"], str) # --------------------------------------------------------------------------- # Tier 3 — e2e # --------------------------------------------------------------------------- class TestCreateCopyRenameE2E: def test_human_create_then_agent_tip_preserves_branch_creator( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: monkeypatch.delenv("MUSE_AGENT_ID", raising=False) _set_user_handle(repo, "aaronrene") env = _env(repo) assert runner.invoke(cli, ["branch", "feat/stable"], env=env).exit_code == 0 assert runner.invoke(cli, ["checkout", "feat/stable"], env=env).exit_code == 0 (repo / "b.py").write_text("y = 2\n") runner.invoke(cli, ["code", "add", "b.py"], env=env) cr = runner.invoke( cli, [ "commit", "-m", "agent tip", "--author", "aaronrene", "--agent-id", "claude-code", ], env=env, ) assert cr.exit_code == 0, cr.output entry = _entry(repo, "feat/stable") assert entry["created_by"] == "claude-code" assert entry["branch_created_by"] == { "handle": "aaronrene", "kind": "human", } def test_copy_dest_gets_new_created_at( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: monkeypatch.delenv("MUSE_AGENT_ID", raising=False) _set_user_handle(repo, "aaronrene") env = _env(repo) assert runner.invoke(cli, ["branch", "feat/src"], env=env).exit_code == 0 src_at = _entry(repo, "feat/src")["branch_created_at"] assert src_at is not None time.sleep(0.01) assert runner.invoke( cli, ["branch", "-c", "feat/src", "feat/dst"], env=env ).exit_code == 0 dst = _entry(repo, "feat/dst") src = _entry(repo, "feat/src") assert dst["branch_created_at"] is not None assert dst["branch_created_at"] >= src_at assert src["branch_created_at"] == src_at assert dst["branch_created_by"] == { "handle": "aaronrene", "kind": "human", } def test_force_copy_restamps_dest( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: """F15: force-copy deletes dest meta then stamps a new creation triple.""" monkeypatch.delenv("MUSE_AGENT_ID", raising=False) _set_user_handle(repo, "aaronrene") env = _env(repo) assert runner.invoke(cli, ["branch", "feat/src2"], env=env).exit_code == 0 assert runner.invoke(cli, ["branch", "feat/old-dst"], env=env).exit_code == 0 old_at = _toml_branch(repo, "feat/old-dst")["created_at"] time.sleep(0.01) assert runner.invoke( cli, ["branch", "-C", "feat/src2", "feat/old-dst"], env=env ).exit_code == 0 new_at = _toml_branch(repo, "feat/old-dst")["created_at"] assert new_at != old_at assert _entry(repo, "feat/old-dst")["branch_created_by"] == { "handle": "aaronrene", "kind": "human", } def test_rename_moves_creation_triple( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: monkeypatch.delenv("MUSE_AGENT_ID", raising=False) _set_user_handle(repo, "aaronrene") env = _env(repo) assert runner.invoke(cli, ["branch", "feat/before"], env=env).exit_code == 0 before = _entry(repo, "feat/before") triple = ( before["branch_created_by"], before["branch_created_at"], ) assert runner.invoke( cli, ["branch", "-m", "feat/before", "feat/after"], env=env ).exit_code == 0 names = {b["name"] for b in _branch_json(repo)} assert "feat/before" not in names after = _entry(repo, "feat/after") assert (after["branch_created_by"], after["branch_created_at"]) == triple assert "feat/before" not in _toml_branch(repo, "feat/before") or True assert "created_at" not in _toml_branch(repo, "feat/before") assert _toml_branch(repo, "feat/after")["created_by"] == "aaronrene" # --------------------------------------------------------------------------- # Tier 4 — stress # --------------------------------------------------------------------------- class TestStress200: def test_200_creates_list_keys_present( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: monkeypatch.delenv("MUSE_AGENT_ID", raising=False) _set_user_handle(repo, "aaronrene") env = _env(repo) for i in range(200): r = runner.invoke(cli, ["branch", f"stress-{i:03d}"], env=env) assert r.exit_code == 0, r.output data = _branch_json(repo) assert len(data) == 201 for entry in data: assert "branch_created_by" in entry assert "branch_created_at" in entry stamped = [b for b in data if b["name"].startswith("stress-")] assert all(b["branch_created_at"] is not None for b in stamped) # --------------------------------------------------------------------------- # Tier 5 — data integrity # --------------------------------------------------------------------------- class TestDataIntegrity: def test_intent_update_preserves_created_at_bytes( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: monkeypatch.delenv("MUSE_AGENT_ID", raising=False) _set_user_handle(repo, "aaronrene") env = _env(repo) assert runner.invoke(cli, ["branch", "feat/di"], env=env).exit_code == 0 first = _toml_branch(repo, "feat/di")["created_at"] assert runner.invoke( cli, ["branch", "feat/di", "--intent", "update"], env=env ).exit_code == 0 second = _toml_branch(repo, "feat/di")["created_at"] assert second == first assert isinstance(second, str) def test_list_twice_identical_branch_created( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: monkeypatch.delenv("MUSE_AGENT_ID", raising=False) _set_user_handle(repo, "aaronrene") env = _env(repo) assert runner.invoke(cli, ["branch", "feat/stable-list"], env=env).exit_code == 0 a = _entry(repo, "feat/stable-list") b = _entry(repo, "feat/stable-list") assert a["branch_created_by"] == b["branch_created_by"] assert a["branch_created_at"] == b["branch_created_at"] def test_corrupt_kind_json_null_listing_ok( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: from muse.core.paths import config_toml_path monkeypatch.delenv("MUSE_AGENT_ID", raising=False) env = _env(repo) assert runner.invoke(cli, ["branch", "feat/corrupt"], env=env).exit_code == 0 cp = config_toml_path(repo) text = cp.read_text(encoding="utf-8") text = text.replace( 'created_by_kind = "human"', 'created_by_kind = "robot"', 1, ) # When identity was None, inject a corrupt pair directly. if 'created_by_kind = "robot"' not in text: text += ( '\n[branch."feat/corrupt"]\n' 'created_by = "aaronrene"\n' 'created_by_kind = "robot"\n' 'created_at = "2026-09-17T12:00:00+00:00"\n' ) cp.write_text(text, encoding="utf-8") else: cp.write_text(text, encoding="utf-8") result = runner.invoke(cli, ["branch", "--json"], env=env) assert result.exit_code == 0, result.output entry = _entry(repo, "feat/corrupt") assert entry["branch_created_by"] is None def test_delete_removes_branch_section( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: monkeypatch.delenv("MUSE_AGENT_ID", raising=False) _set_user_handle(repo, "aaronrene") env = _env(repo) assert runner.invoke(cli, ["branch", "feat/gone"], env=env).exit_code == 0 assert "created_at" in _toml_branch(repo, "feat/gone") assert runner.invoke(cli, ["branch", "-D", "feat/gone"], env=env).exit_code == 0 assert _toml_branch(repo, "feat/gone") == {} # --------------------------------------------------------------------------- # Tier 6 — performance # --------------------------------------------------------------------------- class TestPerformance: def test_200_branch_list_under_5s( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: monkeypatch.delenv("MUSE_AGENT_ID", raising=False) _set_user_handle(repo, "aaronrene") env = _env(repo) for i in range(200): runner.invoke(cli, ["branch", f"perf-{i:03d}"], env=env) t0 = time.perf_counter() data = _branch_json(repo) elapsed = time.perf_counter() - t0 assert len(data) == 201 assert elapsed < 5.0, f"200-branch list took {elapsed:.2f}s (limit 5s)" # --------------------------------------------------------------------------- # Tier 7 — security # --------------------------------------------------------------------------- class TestSecurity: def test_esc_in_agent_id_not_in_toml_or_json( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: monkeypatch.setenv("MUSE_AGENT_ID", "\x1b[31mbad-agent\x1b[0m") env = {**_env(repo), "MUSE_AGENT_ID": "\x1b[31mbad-agent\x1b[0m"} r = runner.invoke(cli, ["branch", "feat/sec"], env=env) assert r.exit_code == 0, r.output raw = (repo / ".muse" / "config.toml").read_text(encoding="utf-8") assert "\x1b" not in raw out = runner.invoke(cli, ["branch", "--json"], env=_env(repo)).output assert "\x1b" not in out entry = _entry(repo, "feat/sec") # ESC bytes stripped; printable remnant remains after sanitize_provenance. assert entry["branch_created_by"] == { "handle": "[31mbad-agent[0m", "kind": "agent", } assert "\x1b" not in entry["branch_created_by"]["handle"] def test_no_key_material_in_output( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: monkeypatch.setenv("MUSE_AGENT_ID", "safe-agent") monkeypatch.setenv("MUSE_AGENT_KEY", "super-secret-key-material") env = { **_env(repo), "MUSE_AGENT_ID": "safe-agent", "MUSE_AGENT_KEY": "super-secret-key-material", } assert runner.invoke(cli, ["branch", "feat/nokey"], env=env).exit_code == 0 out = runner.invoke(cli, ["branch", "--json"], env=env).output assert "super-secret-key-material" not in out assert "MUSE_AGENT_KEY" not in out raw = (repo / ".muse" / "config.toml").read_text(encoding="utf-8") assert "super-secret-key-material" not in raw def test_quote_in_handle_round_trips_toml( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: from muse.core.paths import config_toml_path monkeypatch.delenv("MUSE_AGENT_ID", raising=False) env = _env(repo) with mock.patch( "muse.cli.config.get_config_value", side_effect=lambda key, root=None: ( 'alice"bob' if key == "user.handle" else None ), ): r = runner.invoke(cli, ["branch", "feat/quote"], env=env) assert r.exit_code == 0, r.output with config_toml_path(repo).open("rb") as f: data = tomllib.load(f) assert data["branch"]["feat/quote"]["created_by"] == 'alice"bob' entry = _entry(repo, "feat/quote") assert entry["branch_created_by"] == { "handle": 'alice"bob', "kind": "human", }