"""Tests for `muse auth` CLI commands — whoami, logout. The identity store is redirected to a temporary directory per test so these tests never touch ~/.muse/identity.toml. Network calls are not made — auth commands read/write the local identity store only. """ from __future__ import annotations import json import pathlib import pytest from tests.cli_test_helper import CliRunner from muse._version import __version__ cli = None # argparse migration — CliRunner ignores this arg from muse.cli.config import get_hub_url, set_hub_url from muse.core.identity import ( IdentityEntry, get_identity_path, list_all_identities, load_identity, save_identity, ) runner = CliRunner() # --------------------------------------------------------------------------- # Fixture: minimal repo + isolated identity store # --------------------------------------------------------------------------- @pytest.fixture() def repo(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path: """Minimal .muse/ repo with a pre-configured hub URL. The identity store is redirected to *tmp_path* so tests never touch the real ``~/.muse/identity.toml``. """ muse_dir = tmp_path / ".muse" (muse_dir / "refs" / "heads").mkdir(parents=True) (muse_dir / "objects").mkdir() (muse_dir / "commits").mkdir() (muse_dir / "snapshots").mkdir() (muse_dir / "repo.json").write_text( json.dumps({"repo_id": "test-repo", "schema_version": __version__, "domain": "midi"}) ) (muse_dir / "HEAD").write_text("ref: refs/heads/main\n") (muse_dir / "config.toml").write_text( '[hub]\nurl = "https://musehub.ai"\n' ) monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path)) monkeypatch.chdir(tmp_path) # Isolate the identity store. fake_dir = tmp_path / "home" / ".muse" fake_dir.mkdir(parents=True) fake_file = fake_dir / "identity.toml" monkeypatch.setattr("muse.core.identity._IDENTITY_DIR", fake_dir) monkeypatch.setattr("muse.core.identity._IDENTITY_FILE", fake_file) return tmp_path # --------------------------------------------------------------------------- # muse auth whoami # --------------------------------------------------------------------------- class TestAuthWhoami: def _store_entry(self, hub: str = "https://musehub.ai") -> None: entry: IdentityEntry = { "type": "human", "handle": "Alice", "key_path": "/fake/key.pem", "algorithm": "ed25519", "fingerprint": "abc123fingerprint", } save_identity(hub, entry) def test_whoami_shows_hub(self, repo: pathlib.Path) -> None: self._store_entry() result = runner.invoke(cli, ["auth", "whoami"]) assert result.exit_code == 0 assert "musehub.ai" in result.output def test_whoami_shows_type(self, repo: pathlib.Path) -> None: self._store_entry() result = runner.invoke(cli, ["auth", "whoami"]) assert "human" in result.output def test_whoami_shows_handle(self, repo: pathlib.Path) -> None: self._store_entry() result = runner.invoke(cli, ["auth", "whoami"]) assert "Alice" in result.output def test_whoami_json_output(self, repo: pathlib.Path) -> None: self._store_entry() result = runner.invoke(cli, ["auth", "whoami", "--json"]) assert result.exit_code == 0 data = json.loads(result.output) assert data["type"] == "human" assert data["handle"] == "Alice" assert isinstance(data.get("key_set"), bool) def test_whoami_no_identity_exits_nonzero(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["auth", "whoami"]) assert result.exit_code != 0 def test_whoami_hub_option_selects_specific_hub(self, repo: pathlib.Path) -> None: save_identity("https://staging.musehub.ai", {"type": "agent", "handle": "bot", "key_path": "/k"}) result = runner.invoke(cli, ["auth", "whoami", "--hub", "https://staging.musehub.ai"]) assert result.exit_code == 0 assert "staging.musehub.ai" in result.output def test_whoami_all_lists_all_hubs(self, repo: pathlib.Path) -> None: self._store_entry("https://hub1.example.com") self._store_entry("https://hub2.example.com") result = runner.invoke(cli, ["auth", "whoami", "--all"]) assert result.exit_code == 0 assert "hub1.example.com" in result.output assert "hub2.example.com" in result.output def test_whoami_all_no_identities_exits_nonzero(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["auth", "whoami", "--all"]) assert result.exit_code != 0 def test_whoami_capabilities_shown(self, repo: pathlib.Path) -> None: entry: IdentityEntry = { "type": "agent", "handle": "worker", "key_path": "/k", "capabilities": ["read:*", "write:midi"], } save_identity("https://musehub.ai", entry) result = runner.invoke(cli, ["auth", "whoami"]) assert "read:*" in result.output or "write:midi" in result.output def test_whoami_key_set_is_bool(self, repo: pathlib.Path) -> None: self._store_entry() result = runner.invoke(cli, ["auth", "whoami", "--json"]) assert result.exit_code == 0 raw = result.output assert '"key_set": true' in raw or '"key_set":true' in raw assert '"key_set": "true"' not in raw def test_whoami_all_json_is_single_array(self, repo: pathlib.Path) -> None: """--all --json must emit one JSON array, not multiple separate objects.""" save_identity("https://hub-a.example.com", {"type": "human", "handle": "a", "key_path": "/k1"}) save_identity("https://hub-b.example.com", {"type": "agent", "handle": "b", "key_path": "/k2"}) result = runner.invoke(cli, ["auth", "whoami", "--all", "--json"]) assert result.exit_code == 0 parsed = json.loads(result.output) # would raise if multiple top-level values assert isinstance(parsed, list) assert len(parsed) == 2 def test_whoami_ansi_in_fingerprint_stripped(self, repo: pathlib.Path) -> None: """ANSI escape sequences in stored fingerprint must not appear in text output.""" import unittest.mock evil_entry: IdentityEntry = { "type": "human", "handle": "alice", "fingerprint": "\x1b[31mevil-fp\x1b[0m", } with unittest.mock.patch("muse.core.identity._load_all", return_value={"musehub.ai": evil_entry}): result = runner.invoke(cli, ["auth", "whoami"]) assert result.exit_code == 0 assert "\x1b" not in result.output def test_whoami_short_flags_accepted(self, repo: pathlib.Path) -> None: """-j and -a short flags work.""" self._store_entry() result = runner.invoke(cli, ["auth", "whoami", "-j"]) assert result.exit_code == 0 json.loads(result.output) save_identity("https://hub-x.example.com", {"type": "agent", "handle": "bot", "key_path": "/k"}) result2 = runner.invoke(cli, ["auth", "whoami", "-a"]) assert result2.exit_code == 0 assert "musehub.ai" in result2.output or "hub-x.example.com" in result2.output # --------------------------------------------------------------------------- # muse auth logout # --------------------------------------------------------------------------- class TestAuthLogout: def _store(self, hub: str = "https://musehub.ai") -> None: entry: IdentityEntry = {"type": "human", "handle": "alice", "key_path": "/k"} save_identity(hub, entry) def test_logout_removes_identity(self, repo: pathlib.Path) -> None: self._store() result = runner.invoke(cli, ["auth", "logout"]) assert result.exit_code == 0 assert load_identity("https://musehub.ai") is None def test_logout_shows_success_message(self, repo: pathlib.Path) -> None: self._store() result = runner.invoke(cli, ["auth", "logout"]) assert "musehub.ai" in result.output or "Logged out" in result.output def test_logout_nothing_to_do_does_not_fail(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["auth", "logout"]) assert result.exit_code == 0 assert "nothing" in result.output.lower() or "nothing to do" in result.output.lower() def test_logout_hub_option_removes_specific_hub(self, repo: pathlib.Path) -> None: self._store("https://hub1.example.com") self._store("https://hub2.example.com") result = runner.invoke(cli, ["auth", "logout", "--hub", "https://hub1.example.com"]) assert result.exit_code == 0 assert load_identity("https://hub1.example.com") is None assert load_identity("https://hub2.example.com") is not None def test_logout_all_removes_all_identities(self, repo: pathlib.Path) -> None: self._store("https://hub1.example.com") self._store("https://hub2.example.com") result = runner.invoke(cli, ["auth", "logout", "--all"]) assert result.exit_code == 0 assert not list_all_identities() def test_logout_all_reports_count(self, repo: pathlib.Path) -> None: self._store("https://hub1.example.com") self._store("https://hub2.example.com") result = runner.invoke(cli, ["auth", "logout", "--all"]) assert "2" in result.output def test_logout_all_no_identities_succeeds(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["auth", "logout", "--all"]) assert result.exit_code == 0 def test_logout_fails_without_hub_source( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: """With no hub in config and no --hub flag, logout should fail.""" muse_dir = tmp_path / ".muse" muse_dir.mkdir() (muse_dir / "config.toml").write_text("") (muse_dir / "repo.json").write_text( json.dumps({"repo_id": "r", "schema_version": __version__, "domain": "midi"}) ) (muse_dir / "HEAD").write_text("ref: refs/heads/main\n") monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path)) monkeypatch.chdir(tmp_path) fake_dir = tmp_path / "home" / ".muse" fake_dir.mkdir(parents=True) monkeypatch.setattr("muse.core.identity._IDENTITY_DIR", fake_dir) monkeypatch.setattr("muse.core.identity._IDENTITY_FILE", fake_dir / "identity.toml") result = runner.invoke(cli, ["auth", "logout"]) assert result.exit_code != 0