"""Comprehensive hardening tests for ``muse hub``. Coverage -------- Unit - _normalise_url: scheme injection (file://, ftp://), http non-loopback rejected, loopback allowed, scheme-less normalised to https, trailing slash stripped - _hub_hostname: standard URL, URL with port, URL with path, bare hostname - _ping_hub: reachable, HTTP error, URLError, timeout, redirect refused - _hub_api: file:// scheme blocked, response size cap, error detail sanitized, missing token exits, None-value payload keys stripped - _resolve_proposal_id: full UUID passthrough, prefix match, no match, ambiguous match - _format_proposal: all fields sanitized Integration (CliRunner + mock hub) - run_connect: --json schema, re-connect warns, normalisation, invalid scheme exits - run_disconnect: --json ok, --json nothing_to_do, text mode to stderr - run_status: --json all keys present, no-hub exits, not-authenticated exits - run_ping: --json ok, --json error, text mode to stderr, unreachable exits nonzero - run_proposal_list: --json is a JSON array, text mode to stderr, no-proposals message - run_proposal_create: --json schema, missing branch exits, sanitizes output - run_proposal_merge: --json schema, merge=false exits nonzero - run_proposal_show: --json passthrough Security - file:// hub URL blocked in _hub_api before network - ANSI in proposal title/branch sanitized in _format_proposal - ANSI in proposal ID sanitized in _resolve_proposal_id errors - hub URL in errors sanitized in _get_hub_and_identity - Response body size cap prevents OOM E2E (via CliRunner) - connect --json schema includes all required keys - disconnect --json schema correct for both ok and nothing_to_do - ping --json schema with all required keys - status --json all keys always present (no missing keys when not authenticated) Stress - 8 concurrent ping checks against isolated mock responses """ from __future__ import annotations import json import pathlib import threading import unittest.mock import urllib.error import ssl import urllib.request from collections.abc import Mapping from typing import TYPE_CHECKING from unittest.mock import MagicMock, patch import pytest from tests.cli_test_helper import CliRunner, InvokeResult if TYPE_CHECKING: pass from muse.cli.commands.hub.connection import ( _ConnectJson, _DisconnectJson, _PingJson, _StatusJson, ) from muse.core.types import Manifest, MsgpackDict, MsgpackValue from muse.core.identity import IdentityEntry from muse.core.paths import head_path, heads_dir, muse_dir type _JsonPayload = MsgpackDict type _ProposalRecord = dict[str, str] type _RepoResponse = dict[str, str] type _HubBody = Mapping[str, str | bool | list[str] | None] | None cli = None runner = CliRunner() # ── helpers ─────────────────────────────────────────────────────────────────── def _json_line(result: InvokeResult) -> _JsonPayload: for line in result.output.splitlines(): stripped = line.strip() if stripped.startswith("{") or stripped.startswith("["): data: _JsonPayload = json.loads(stripped) return data raise ValueError(f"No JSON line in output:\n{result.output!r}") def _json_connect(result: InvokeResult) -> _ConnectJson: d: _ConnectJson = json.loads( next(l for l in result.output.splitlines() if l.strip().startswith("{")) ) return d def _json_status(result: InvokeResult) -> _StatusJson: d: _StatusJson = json.loads( next(l for l in result.output.splitlines() if l.strip().startswith("{")) ) return d def _json_disconnect(result: InvokeResult) -> _DisconnectJson: d: _DisconnectJson = json.loads( next(l for l in result.output.splitlines() if l.strip().startswith("{")) ) return d def _json_ping(result: InvokeResult) -> _PingJson: d: _PingJson = json.loads( next(l for l in result.output.splitlines() if l.strip().startswith("{")) ) return d @pytest.fixture def repo(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path: """Minimal .muse/ repo with identity file.""" from muse._version import __version__ dot_muse = muse_dir(tmp_path) for sub in ("refs/heads", "objects", "commits", "snapshots"): (dot_muse / sub).mkdir(parents=True, exist_ok=True) (dot_muse / "repo.json").write_text( json.dumps({"repo_id": "test-repo", "schema_version": __version__, "domain": "code"}) ) (dot_muse / "HEAD").write_text("ref: refs/heads/main\n") (dot_muse / "refs" / "heads" / "main").write_text("") (dot_muse / "config.toml").write_text("") muse_home = tmp_path / ".muse-home" muse_home.mkdir() (muse_home / "identity.toml").write_text("") monkeypatch.setattr("muse.core.identity._IDENTITY_FILE", muse_home / "identity.toml") monkeypatch.setattr("muse.core.identity._IDENTITY_DIR", muse_home) monkeypatch.chdir(tmp_path) return tmp_path def _make_signing() -> "SigningIdentity": """Generate a fresh Ed25519 SigningIdentity for tests.""" from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from muse.core.transport import SigningIdentity return SigningIdentity(handle="testuser", private_key=Ed25519PrivateKey.generate()) def _store_identity(hub_url: str, handle: str = "alice") -> None: """Save a human identity entry with hd_path + keychain mnemonic.""" from muse.core.identity import IdentityEntry, save_identity from muse.core.hdkeys import muse_path, DOMAIN_IDENTITY, ENTITY_HUMAN, ROLE_SIGN hd_path = muse_path(DOMAIN_IDENTITY, ENTITY_HUMAN, 0) entry: IdentityEntry = { "type": "human", "handle": handle, "algorithm": "ed25519", "fingerprint": f"test-fp-{handle}", "hd_path": hd_path, } # Use a fixed test mnemonic stored in the keychain so resolve_signing_identity works. _TEST_MNEMONIC = ( "abandon abandon abandon abandon abandon abandon abandon abandon " "abandon abandon abandon about" ) save_identity(hub_url, entry, mnemonic=_TEST_MNEMONIC) # ── Unit: _normalise_url ────────────────────────────────────────────────────── class TestNormaliseUrlHardening: def test_file_scheme_raises(self) -> None: from muse.cli.commands.hub import _normalise_url with pytest.raises(ValueError, match="not allowed"): _normalise_url("file:///etc/passwd") def test_ftp_scheme_raises(self) -> None: from muse.cli.commands.hub import _normalise_url with pytest.raises(ValueError, match="not allowed"): _normalise_url("ftp://attacker.example.com/repo") def test_data_scheme_raises(self) -> None: from muse.cli.commands.hub import _normalise_url with pytest.raises(ValueError, match="not allowed"): _normalise_url("data:text/plain,malicious") def test_http_non_loopback_raises(self) -> None: from muse.cli.commands.hub import _normalise_url with pytest.raises(ValueError, match="HTTPS"): _normalise_url("http://musehub.ai/gabriel/muse") def test_http_localhost_allowed(self) -> None: from muse.cli.commands.hub import _normalise_url assert _normalise_url("https://localhost:1337") == "https://localhost:1337" def test_http_127_allowed(self) -> None: from muse.cli.commands.hub import _normalise_url assert _normalise_url("http://127.0.0.1:9000") == "http://127.0.0.1:9000" def test_schemeless_becomes_https(self) -> None: from muse.cli.commands.hub import _normalise_url assert _normalise_url("musehub.ai").startswith("https://") def test_trailing_slash_stripped(self) -> None: from muse.cli.commands.hub import _normalise_url assert not _normalise_url("https://musehub.ai/").endswith("/") def test_https_passthrough(self) -> None: from muse.cli.commands.hub import _normalise_url assert _normalise_url("https://musehub.ai/gabriel/muse") == "https://musehub.ai/gabriel/muse" # ── Unit: _hub_hostname ─────────────────────────────────────────────────────── class TestHubHostname: def test_plain_https(self) -> None: from muse.cli.commands.hub.connection import _hub_hostname assert _hub_hostname("https://musehub.ai/gabriel/muse") == "musehub.ai" def test_with_port(self) -> None: from muse.cli.commands.hub.connection import _hub_hostname assert _hub_hostname("https://localhost:1337/gabriel/muse") == "localhost:1337" def test_bare_hostname(self) -> None: from muse.cli.commands.hub.connection import _hub_hostname assert _hub_hostname("musehub.ai") == "musehub.ai" def test_trailing_slash(self) -> None: from muse.cli.commands.hub.connection import _hub_hostname assert _hub_hostname("https://musehub.ai/") == "musehub.ai" # ── Unit: _hub_api ──────────────────────────────────────────────────────────── class TestHubApi: _IDENTITY = {"type": "human", "token": "tok123"} def test_file_scheme_blocked_before_network(self) -> None: from muse.cli.commands.hub import _hub_api from muse.core.identity import IdentityEntry identity: IdentityEntry = {"type": "human", "token": "tok"} with patch("urllib.request.urlopen") as mock_net: with pytest.raises(SystemExit): _hub_api("file:///etc/passwd", identity, "GET", "/api/test") mock_net.assert_not_called() def test_ftp_scheme_blocked_before_network(self) -> None: from muse.cli.commands.hub import _hub_api from muse.core.identity import IdentityEntry identity: IdentityEntry = {"type": "human", "token": "tok"} with patch("urllib.request.urlopen") as mock_net: with pytest.raises(SystemExit): _hub_api("ftp://ftp.example.com", identity, "GET", "/api/test") mock_net.assert_not_called() def test_missing_token_exits(self) -> None: """No signing identity on a mutating method → SystemExit before any network I/O. GET requests on public resources are allowed without auth; POST/PUT/DELETE/PATCH still require a signing identity. identity carries no handle/key_path so Ed25519 key loading is skipped. get_signing_identity is patched to return None. """ from muse.cli.commands.hub import _hub_api from muse.core.identity import IdentityEntry identity: IdentityEntry = {"type": "human"} with patch("muse.cli.config.get_signing_identity", return_value=None): with patch("urllib.request.urlopen") as mock_net: with pytest.raises(SystemExit): _hub_api("https://localhost:1337", identity, "POST", "/api/test") mock_net.assert_not_called() def test_response_size_cap(self) -> None: from muse.cli.commands.hub import _MAX_API_RESPONSE_BYTES, _hub_api from muse.core.identity import IdentityEntry identity: IdentityEntry = {"type": "human", "token": "tok"} mock_resp = MagicMock() mock_resp.__enter__ = lambda s: s mock_resp.__exit__ = MagicMock(return_value=False) mock_resp.read.return_value = b"x" * (_MAX_API_RESPONSE_BYTES + 2) with patch("muse.cli.config.get_signing_identity", return_value=_make_signing()): with patch("urllib.request.urlopen", return_value=mock_resp): with pytest.raises(SystemExit): _hub_api("http://localhost:9999", identity, "GET", "/api/test") def test_http_error_sanitized_in_output( self, capsys: pytest.CaptureFixture[str] ) -> None: import urllib.error from muse.cli.commands.hub import _hub_api from muse.core.identity import IdentityEntry import io identity: IdentityEntry = {"type": "human", "token": "tok"} ansi_detail = b'{"detail":"\\x1b[31mmalicious\\x1b[0m"}' exc = urllib.error.HTTPError(url="", code=403, msg="Forbidden", hdrs=MagicMock(), fp=io.BytesIO(ansi_detail)) with patch("muse.cli.config.get_signing_identity", return_value=_make_signing()): with patch("urllib.request.urlopen", side_effect=exc): with pytest.raises(SystemExit): _hub_api("http://localhost:9999", identity, "GET", "/api/test") captured = capsys.readouterr() assert "\x1b[" not in captured.err def test_empty_response_returns_empty_dict(self) -> None: from muse.cli.commands.hub import _hub_api from muse.core.identity import IdentityEntry identity: IdentityEntry = {"type": "human", "token": "tok"} mock_resp = MagicMock() mock_resp.__enter__ = lambda s: s mock_resp.__exit__ = MagicMock(return_value=False) mock_resp.read.return_value = b"" with patch("muse.cli.config.get_signing_identity", return_value=_make_signing()): with patch("urllib.request.urlopen", return_value=mock_resp): result = _hub_api("http://localhost:9999", identity, "GET", "/api/test") assert result == {} # ── TDD: PEM-load path removed from _hub_api (H1) ─────────────────────────── class TestHubApiPemLoadRemoved: """H1: _hub_api must use get_signing_identity exclusively — no PEM reads. Before fix: identity.get("key_path") was used as primary signing path; load_pem_private_key was called if the file existed. After fix: get_signing_identity(remote_url=...) is the only signing path. """ def test_H1_load_pem_private_key_never_called_even_with_key_path( self, tmp_path: pathlib.Path ) -> None: """Even when identity contains key_path pointing to a real file, load_pem_private_key must NOT be called — only get_signing_identity.""" from muse.cli.commands.hub import _hub_api from muse.core.identity import IdentityEntry from unittest.mock import MagicMock, patch, call # Create a fake PEM file so is_file() returns True fake_pem = tmp_path / "fake.pem" fake_pem.write_bytes(b"-----BEGIN PRIVATE KEY-----\ngarbage\n-----END PRIVATE KEY-----\n") identity: IdentityEntry = { "type": "human", "handle": "alice", "key_path": str(fake_pem), } mock_resp = MagicMock() mock_resp.__enter__ = lambda s: s mock_resp.__exit__ = MagicMock(return_value=False) mock_resp.read.return_value = b"{}" mock_load_pem = MagicMock(return_value=MagicMock()) with ( patch("muse.cli.config.get_signing_identity", return_value=_make_signing()) as mock_gsi, patch("urllib.request.urlopen", return_value=mock_resp), patch("cryptography.hazmat.primitives.serialization.load_pem_private_key", mock_load_pem), ): _hub_api("http://localhost:9999", identity, "GET", "/api/test") mock_load_pem.assert_not_called() mock_gsi.assert_called_once() # ── Unit: _resolve_proposal_id ────────────────────────────────────────────────────── class TestResolveProposalId: def _make_identity(self) -> "muse.core.identity.IdentityEntry": from muse.core.identity import IdentityEntry e: IdentityEntry = {"type": "human", "token": "tok123"} return e def _proposal(self, proposal_id: str, title: str = "Test Proposal") -> _ProposalRecord: return {"proposalId": proposal_id, "title": title, "state": "open", "fromBranch": "feat/x", "toBranch": "dev"} def test_full_id_returned_as_is(self) -> None: from muse.cli.commands.hub import _resolve_proposal_id full = "af54753d-1234-5678-abcd-ef1234567890" result = _resolve_proposal_id("http://hub", self._make_identity(), "repo-id", full) assert result == full def test_prefix_resolved(self) -> None: from muse.cli.commands.hub import _resolve_proposal_id proposal_id = "abc12345-6789-0000-0000-000000000000" proposals_resp = {"proposals": [self._proposal(proposal_id)]} mock_resp = MagicMock() mock_resp.__enter__ = lambda s: s mock_resp.__exit__ = MagicMock(return_value=False) mock_resp.read.return_value = json.dumps(proposals_resp).encode() with patch("muse.cli.config.get_signing_identity", return_value=_make_signing()): with patch("urllib.request.urlopen", return_value=mock_resp): result = _resolve_proposal_id( "http://localhost:9999", self._make_identity(), "repo-id", "abc12345" ) assert result == proposal_id def test_no_match_exits(self) -> None: from muse.cli.commands.hub import _resolve_proposal_id resp_bytes = json.dumps({"proposals": []}).encode() mock_resp = MagicMock() mock_resp.__enter__ = lambda s: s mock_resp.__exit__ = MagicMock(return_value=False) mock_resp.read.return_value = resp_bytes with patch("muse.cli.config.get_signing_identity", return_value=_make_signing()): with patch("urllib.request.urlopen", return_value=mock_resp): with pytest.raises(SystemExit): _resolve_proposal_id( "http://localhost:9999", self._make_identity(), "repo-id", "deadbeef" ) def test_ambiguous_prefix_exits(self) -> None: from muse.cli.commands.hub import _resolve_proposal_id pr1_id = "abc12345-0000-0000-0000-000000000001" pr2_id = "abc12345-0000-0000-0000-000000000002" proposals_resp = {"proposals": [self._proposal(pr1_id), self._proposal(pr2_id)]} mock_resp = MagicMock() mock_resp.__enter__ = lambda s: s mock_resp.__exit__ = MagicMock(return_value=False) mock_resp.read.return_value = json.dumps(proposals_resp).encode() with patch("muse.cli.config.get_signing_identity", return_value=_make_signing()): with patch("urllib.request.urlopen", return_value=mock_resp): with pytest.raises(SystemExit): _resolve_proposal_id( "http://localhost:9999", self._make_identity(), "repo-id", "abc12345" ) def test_ansi_in_proposal_id_sanitized_in_error( self, capsys: pytest.CaptureFixture[str] ) -> None: from muse.cli.commands.hub import _resolve_proposal_id resp_bytes = json.dumps({"proposals": []}).encode() mock_resp = MagicMock() mock_resp.__enter__ = lambda s: s mock_resp.__exit__ = MagicMock(return_value=False) mock_resp.read.return_value = resp_bytes malicious_proposalefix = "\x1b[31mmalicious\x1b[0m" with patch("muse.cli.config.get_signing_identity", return_value=_make_signing()): with patch("urllib.request.urlopen", return_value=mock_resp): with pytest.raises(SystemExit): _resolve_proposal_id( "http://localhost:9999", self._make_identity(), "repo-id", malicious_proposalefix ) captured = capsys.readouterr() assert "\x1b[" not in captured.err # ── Unit: _format_proposal ────────────────────────────────────────────────────────── class TestFormatProposal: def test_ansi_in_title_stripped(self) -> None: from muse.cli.commands.hub import _format_proposal proposal: _ProposalRecord = { "proposalId": "abc12345", "title": "\x1b[31mmalicious title\x1b[0m", "state": "open", "fromBranch": "feat/x", "toBranch": "dev", } result = _format_proposal(proposal) assert "\x1b[" not in result def test_ansi_in_branch_stripped(self) -> None: from muse.cli.commands.hub import _format_proposal proposal: _ProposalRecord = { "proposalId": "abc12345", "title": "clean title", "state": "open", "fromBranch": "\x1b[32mfeat/malicious\x1b[0m", "toBranch": "\x1b[31mdev\x1b[0m", } result = _format_proposal(proposal) assert "\x1b[" not in result def test_state_icon_open(self) -> None: from muse.cli.commands.hub import _format_proposal proposal: _ProposalRecord = { "proposalId": "abc12345", "title": "t", "state": "open", "fromBranch": "f", "toBranch": "d", } assert "🟢" in _format_proposal(proposal) def test_state_icon_merged(self) -> None: from muse.cli.commands.hub import _format_proposal proposal: _ProposalRecord = { "proposalId": "abc12345", "title": "t", "state": "merged", "fromBranch": "f", "toBranch": "d", } assert "🟣" in _format_proposal(proposal) # ── Integration: run_connect ────────────────────────────────────────────────── class TestConnectHardening: _HUB = "http://localhost:19999" def test_connect_json_schema(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["hub", "connect", self._HUB, "--json"]) assert result.exit_code == 0 data = _json_connect(result) for key in ("status", "hub_url", "hostname", "authenticated", "identity_name", "identity_type"): assert key in data, f"Missing key: {key}" assert data["status"] == "ok" assert data["authenticated"] is False assert data["identity_name"] == "" assert data["identity_type"] == "" def test_connect_authenticated_json_schema(self, repo: pathlib.Path) -> None: _store_identity(self._HUB) result = runner.invoke(cli, ["hub", "connect", self._HUB, "--json"]) assert result.exit_code == 0 data = _json_connect(result) assert data["authenticated"] is True assert data["identity_name"] == "alice" assert data["identity_type"] == "human" def test_connect_invalid_scheme_exits(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["hub", "connect", "file:///etc/passwd"]) assert result.exit_code != 0 def test_connect_http_non_loopback_exits(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["hub", "connect", "http://musehub.ai"]) assert result.exit_code != 0 def test_connect_json_stdout_clean(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["hub", "connect", self._HUB, "--json"]) assert result.exit_code == 0 json_lines = [l for l in result.output.splitlines() if l.strip().startswith("{")] assert len(json_lines) >= 1 def test_connect_no_repo_exits(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.chdir(tmp_path) result = runner.invoke(cli, ["hub", "connect", self._HUB]) assert result.exit_code != 0 def test_reconnect_warning_on_stderr(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["hub", "connect", self._HUB]) result = runner.invoke(cli, ["hub", "connect", "http://localhost:20000"]) assert result.exit_code == 0 assert "localhost:19999" in result.stderr def test_reconnect_same_url_no_warning(self, repo: pathlib.Path) -> None: """Re-connecting to the same URL is a no-op — no warning emitted.""" runner.invoke(cli, ["hub", "connect", self._HUB]) result = runner.invoke(cli, ["hub", "connect", self._HUB]) assert result.exit_code == 0 assert "Switching" not in result.stderr assert "⚠️" not in result.stderr def test_connect_short_flag_j(self, repo: pathlib.Path) -> None: """-j short flag produces identical JSON output to --json.""" r_long = runner.invoke(cli, ["hub", "connect", self._HUB, "--json"]) runner.invoke(cli, ["hub", "disconnect"]) r_short = runner.invoke(cli, ["hub", "connect", self._HUB, "-j"]) assert r_long.exit_code == 0 assert r_short.exit_code == 0 d_long = _json_connect(r_long) d_short = _json_connect(r_short) assert d_long == d_short def test_connect_ipv6_loopback_accepted(self, repo: pathlib.Path) -> None: """http://[::1] and http://[::1]:PORT are valid loopback URLs.""" result = runner.invoke(cli, ["hub", "connect", "http://[::1]:8080", "--json"]) assert result.exit_code == 0 data = _json_connect(result) assert data["status"] == "ok" assert "::1" in data["hub_url"] def test_connect_ipv6_loopback_bare_accepted(self, repo: pathlib.Path) -> None: """http://[::1] without a port is valid.""" result = runner.invoke(cli, ["hub", "connect", "http://[::1]", "--json"]) assert result.exit_code == 0 data = _json_connect(result) assert data["status"] == "ok" def test_connect_bare_hostname_with_port(self, repo: pathlib.Path) -> None: """musehub.ai:8443 (no scheme) is promoted to https://musehub.ai:8443.""" result = runner.invoke(cli, ["hub", "connect", "musehub.ai:8443", "--json"]) assert result.exit_code == 0 data = _json_connect(result) assert data["hub_url"] == "https://musehub.ai:8443" assert data["hostname"] == "musehub.ai:8443" def test_connect_trailing_slash_stripped(self, repo: pathlib.Path) -> None: """Trailing slashes are stripped from the stored URL.""" result = runner.invoke( cli, ["hub", "connect", "https://musehub.ai/", "--json"] ) assert result.exit_code == 0 data = _json_connect(result) assert not data["hub_url"].endswith("/") def test_connect_ansi_in_reconnect_warning_sanitized( self, repo: pathlib.Path ) -> None: """ANSI codes stored in config are stripped from the reconnect warning.""" import unittest.mock ansi_url = "https://\x1b[31mattacker.example.com\x1b[0m" with unittest.mock.patch( "muse.cli.commands.hub.get_hub_url", return_value=ansi_url ): result = runner.invoke( cli, ["hub", "connect", "https://safe.example.com"] ) assert "\x1b" not in result.stderr, "ANSI escape leaked into reconnect warning" def test_connect_json_hub_url_normalised(self, repo: pathlib.Path) -> None: """hub_url in JSON is the normalised form (no trailing slash, has scheme).""" result = runner.invoke( cli, ["hub", "connect", "musehub.ai", "--json"] ) assert result.exit_code == 0 data = _json_connect(result) assert data["hub_url"].startswith("https://") assert not data["hub_url"].endswith("/") def test_connect_no_repo_exits_2(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None: """Exit code 2 (REPO_NOT_FOUND) when outside a Muse repo.""" monkeypatch.chdir(tmp_path) result = runner.invoke(cli, ["hub", "connect", self._HUB]) assert result.exit_code == 2 def test_connect_http_non_loopback_exits_1(self, repo: pathlib.Path) -> None: """Exit code 1 (USER_ERROR) for http:// non-loopback URL.""" result = runner.invoke(cli, ["hub", "connect", "http://remote.example.com"]) assert result.exit_code == 1 def test_connect_disallowed_scheme_exits_1(self, repo: pathlib.Path) -> None: """Exit code 1 (USER_ERROR) for ftp:// URL.""" result = runner.invoke(cli, ["hub", "connect", "ftp://musehub.ai"]) assert result.exit_code == 1 def test_10_sequential_connects_all_survive(self, repo: pathlib.Path) -> None: """10 sequential connect→disconnect cycles all succeed.""" for i in range(10): hub = f"http://localhost:{19000 + i}" r = runner.invoke(cli, ["hub", "connect", hub, "--json"]) assert r.exit_code == 0, f"connect {i} failed: {r.output}" data = _json_connect(r) assert data["status"] == "ok" # ── Integration: run_status ─────────────────────────────────────────────────── class TestStatusHardening: _HUB = "http://localhost:19999" def test_status_no_hub_exits_nonzero(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["hub", "status", "--json"]) assert result.exit_code != 0 def test_status_json_all_keys_always_present(self, repo: pathlib.Path) -> None: """All 7 JSON keys present even when not authenticated.""" runner.invoke(cli, ["hub", "connect", self._HUB]) result = runner.invoke(cli, ["hub", "status", "--json"]) assert result.exit_code == 0 data = _json_status(result) for key in ("hub_url", "hostname", "authenticated", "identity_type", "identity_name", "identity_id", "capabilities"): assert key in data, f"Missing key: {key}" assert data["authenticated"] is False assert data["identity_type"] == "" assert data["identity_name"] == "" assert data["identity_id"] == "" assert data["capabilities"] == [] def test_status_json_authenticated(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["hub", "connect", self._HUB]) _store_identity(self._HUB) result = runner.invoke(cli, ["hub", "status", "--json"]) assert result.exit_code == 0 data = _json_status(result) assert data["authenticated"] is True assert data["identity_name"] == "alice" assert data["identity_type"] == "human" def test_status_json_stdout_clean(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["hub", "connect", self._HUB]) result = runner.invoke(cli, ["hub", "status", "--json"]) json_lines = [l for l in result.output.splitlines() if l.strip().startswith("{")] assert len(json_lines) >= 1 def test_status_text_mode_to_stderr(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None: runner.invoke(cli, ["hub", "connect", self._HUB]) result = runner.invoke(cli, ["hub", "status"]) assert result.exit_code == 0 def test_status_short_flag_j(self, repo: pathlib.Path) -> None: """-j short flag produces identical JSON output to --json.""" runner.invoke(cli, ["hub", "connect", self._HUB]) r_long = runner.invoke(cli, ["hub", "status", "--json"]) r_short = runner.invoke(cli, ["hub", "status", "-j"]) assert r_long.exit_code == 0 assert r_short.exit_code == 0 assert json.loads(r_long.output) == json.loads(r_short.output) def test_status_exit_code_2_no_repo( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: """Exit code 2 (REPO_NOT_FOUND) when outside a Muse repo.""" monkeypatch.chdir(tmp_path) result = runner.invoke(cli, ["hub", "status"]) assert result.exit_code == 2 def test_status_exit_code_1_no_hub(self, repo: pathlib.Path) -> None: """Exit code 1 (USER_ERROR) when no hub is connected.""" result = runner.invoke(cli, ["hub", "status"]) assert result.exit_code == 1 def test_status_hub_override_flag(self, repo: pathlib.Path) -> None: """--hub flag overrides config; identity is looked up for that URL.""" override = "http://localhost:29999" from muse.core.identity import IdentityEntry, save_identity entry: IdentityEntry = { "type": "agent", "handle": "override-bot", } save_identity(override, entry) # Connect to a different hub runner.invoke(cli, ["hub", "connect", self._HUB]) result = runner.invoke( cli, ["hub", "status", "--hub", override, "--json"] ) assert result.exit_code == 0 data = _json_status(result) assert data["authenticated"] is True assert data["identity_name"] == "override-bot" def test_status_json_capabilities_populated_for_agent( self, repo: pathlib.Path ) -> None: """capabilities field is populated from agent identity.""" from muse.core.identity import IdentityEntry, save_identity entry: IdentityEntry = { "type": "agent", "handle": "cap-bot", "capabilities": ["read:*", "write:midi", "commit"], } save_identity(self._HUB, entry) runner.invoke(cli, ["hub", "connect", self._HUB]) result = runner.invoke(cli, ["hub", "status", "--json"]) assert result.exit_code == 0 data = _json_status(result) assert data["capabilities"] == ["read:*", "write:midi", "commit"] def test_status_capabilities_empty_for_human(self, repo: pathlib.Path) -> None: """capabilities is [] for human identities (they have no cap list).""" runner.invoke(cli, ["hub", "connect", self._HUB]) _store_identity(self._HUB) # human identity, no capabilities result = runner.invoke(cli, ["hub", "status", "--json"]) assert result.exit_code == 0 data = _json_status(result) assert data["capabilities"] == [] def test_status_ansi_in_identity_fields_sanitized( self, repo: pathlib.Path ) -> None: """ANSI codes in identity_type, identity_name, identity_id stripped in text output.""" import unittest.mock ansi_entry = { "type": "\x1b[31magent\x1b[0m", "handle": "\x1b[32mmalicious-bot\x1b[0m", } with unittest.mock.patch( "muse.core.identity._load_all", return_value={"localhost:19999": ansi_entry}, ): runner.invoke(cli, ["hub", "connect", self._HUB]) result = runner.invoke(cli, ["hub", "status"]) assert "\x1b" not in result.stderr, "ANSI escape leaked into status text output" def test_status_ansi_in_capabilities_sanitized( self, repo: pathlib.Path ) -> None: """ANSI codes in capabilities are stripped from text output.""" import unittest.mock ansi_entry = { "type": "agent", "handle": "bot", "capabilities": ["\x1b[31mread:*\x1b[0m", "write:midi"], } with unittest.mock.patch( "muse.core.identity._load_all", return_value={"localhost:19999": ansi_entry}, ): runner.invoke(cli, ["hub", "connect", self._HUB]) result = runner.invoke(cli, ["hub", "status"]) assert "\x1b" not in result.stderr, "ANSI escape in capability leaked to output" def test_status_json_single_object_per_call(self, repo: pathlib.Path) -> None: """Exactly one JSON object emitted to stdout per invocation.""" runner.invoke(cli, ["hub", "connect", self._HUB]) result = runner.invoke(cli, ["hub", "status", "--json"]) assert result.exit_code == 0 objects = [l for l in result.output.splitlines() if l.strip().startswith("{")] assert len(objects) == 1, f"Expected 1 JSON object, got {len(objects)}" def test_10_sequential_status_calls(self, repo: pathlib.Path) -> None: """10 sequential status calls all succeed with consistent JSON.""" runner.invoke(cli, ["hub", "connect", self._HUB]) _store_identity(self._HUB) results = [] for _ in range(10): r = runner.invoke(cli, ["hub", "status", "--json"]) assert r.exit_code == 0 results.append(json.loads(r.output)) # All results must be identical assert all(r == results[0] for r in results), "Status output not stable" # ── Integration: run_disconnect ─────────────────────────────────────────────── class TestDisconnectHardening: _HUB = "http://localhost:19999" def test_disconnect_nothing_to_do_json(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["hub", "disconnect", "--json"]) assert result.exit_code == 0 data = _json_disconnect(result) assert data["status"] == "nothing_to_do" assert data["hostname"] == "" def test_disconnect_ok_json(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["hub", "connect", self._HUB]) result = runner.invoke(cli, ["hub", "disconnect", "--json"]) assert result.exit_code == 0 data = _json_disconnect(result) assert data["status"] == "ok" assert "localhost" in data["hostname"] def test_disconnect_removes_hub_url(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["hub", "connect", self._HUB]) runner.invoke(cli, ["hub", "disconnect"]) result = runner.invoke(cli, ["hub", "status"]) assert result.exit_code != 0 def test_disconnect_json_schema_all_keys(self, repo: pathlib.Path) -> None: """All three JSON keys present on success.""" runner.invoke(cli, ["hub", "connect", self._HUB]) result = runner.invoke(cli, ["hub", "disconnect", "--json"]) data = _json_disconnect(result) for key in ("status", "hub_url", "hostname"): assert key in data, f"Missing key: {key}" def test_disconnect_json_nothing_to_do_all_keys(self, repo: pathlib.Path) -> None: """All three JSON keys present even when nothing was connected.""" result = runner.invoke(cli, ["hub", "disconnect", "--json"]) assert result.exit_code == 0 data = _json_disconnect(result) for key in ("status", "hub_url", "hostname"): assert key in data, f"Missing key: {key}" assert data["hub_url"] == "" assert data["hostname"] == "" def test_disconnect_json_hub_url_matches_connected( self, repo: pathlib.Path ) -> None: """hub_url in JSON is the full URL that was disconnected.""" runner.invoke(cli, ["hub", "connect", self._HUB]) result = runner.invoke(cli, ["hub", "disconnect", "--json"]) assert result.exit_code == 0 data = _json_disconnect(result) assert data["hub_url"] == self._HUB assert "localhost" in data["hostname"] def test_disconnect_no_repo_exits_2(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None: """Exit code 2 (REPO_NOT_FOUND) when outside a Muse repo.""" monkeypatch.chdir(tmp_path) result = runner.invoke(cli, ["hub", "disconnect"]) assert result.exit_code == 2 def test_disconnect_no_repo_exits(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.chdir(tmp_path) result = runner.invoke(cli, ["hub", "disconnect"]) assert result.exit_code != 0 def test_disconnect_short_flag_j(self, repo: pathlib.Path) -> None: """-j short flag produces identical JSON output to --json.""" runner.invoke(cli, ["hub", "connect", self._HUB]) r_long = runner.invoke(cli, ["hub", "disconnect", "--json"]) runner.invoke(cli, ["hub", "connect", self._HUB]) r_short = runner.invoke(cli, ["hub", "connect", self._HUB]) # reconnect r_short = runner.invoke(cli, ["hub", "disconnect", "-j"]) assert r_long.exit_code == 0 assert r_short.exit_code == 0 d_long = _json_disconnect(r_long) d_short = _json_disconnect(r_short) # Both should have same shape; hub_url and hostname may differ so # check schema only. assert set(d_long.keys()) == set(d_short.keys()) assert d_short["status"] == "ok" def test_disconnect_idempotent_second_call(self, repo: pathlib.Path) -> None: """Second disconnect exits 0 with status nothing_to_do.""" runner.invoke(cli, ["hub", "connect", self._HUB]) r1 = runner.invoke(cli, ["hub", "disconnect", "--json"]) r2 = runner.invoke(cli, ["hub", "disconnect", "--json"]) assert r1.exit_code == 0 assert r2.exit_code == 0 d1 = _json_disconnect(r1) d2 = _json_disconnect(r2) assert d1["status"] == "ok" assert d2["status"] == "nothing_to_do" def test_disconnect_preserves_identity(self, repo: pathlib.Path) -> None: """Credentials in identity.toml survive hub disconnect.""" from muse.core.identity import IdentityEntry, load_identity, save_identity entry: IdentityEntry = {"type": "human", "handle": "alice"} save_identity(self._HUB, entry) runner.invoke(cli, ["hub", "connect", self._HUB]) runner.invoke(cli, ["hub", "disconnect"]) assert load_identity(self._HUB) is not None def test_disconnect_json_stdout_clean(self, repo: pathlib.Path) -> None: """No non-JSON text on stdout when --json is passed.""" runner.invoke(cli, ["hub", "connect", self._HUB]) result = runner.invoke(cli, ["hub", "disconnect", "--json"]) assert result.exit_code == 0 for line in result.output.splitlines(): stripped = line.strip() if stripped: assert stripped.startswith("{") or stripped.startswith('"'), \ f"Non-JSON on stdout: {stripped!r}" def test_disconnect_ansi_in_hub_url_sanitized( self, repo: pathlib.Path ) -> None: """ANSI codes in a stored hub URL are stripped from text output.""" import unittest.mock ansi_url = "https://\x1b[31mattacker.example.com\x1b[0m" with unittest.mock.patch( "muse.cli.commands.hub.get_hub_url", return_value=ansi_url ): result = runner.invoke(cli, ["hub", "disconnect"]) assert "\x1b" not in result.stderr, "ANSI escape leaked into disconnect output" def test_10_sequential_disconnect_cycles(self, repo: pathlib.Path) -> None: """10 connect→disconnect cycles all succeed with correct JSON.""" for i in range(10): hub = f"http://localhost:{20000 + i}" runner.invoke(cli, ["hub", "connect", hub]) r = runner.invoke(cli, ["hub", "disconnect", "--json"]) assert r.exit_code == 0, f"cycle {i} failed: {r.output}" data = _json_disconnect(r) assert data["status"] == "ok" assert data["hub_url"] == hub # ── Integration: run_ping ───────────────────────────────────────────────────── class TestPingHardening: _HUB = "http://localhost:19999" def _connect(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["hub", "connect", self._HUB]) def test_ping_reachable_json_schema(self, repo: pathlib.Path) -> None: self._connect(repo) mock_resp = MagicMock() mock_resp.__enter__ = lambda s: s mock_resp.__exit__ = MagicMock(return_value=False) mock_resp.status = 200 with patch("urllib.request.OpenerDirector.open", return_value=mock_resp): result = runner.invoke(cli, ["hub", "ping", "--json"]) assert result.exit_code == 0 data = _json_ping(result) for key in ("status", "hub_url", "hostname", "reachable", "message"): assert key in data, f"Missing key: {key}" assert data["reachable"] is True assert data["status"] == "ok" def test_ping_unreachable_json_schema(self, repo: pathlib.Path) -> None: self._connect(repo) import urllib.error exc = urllib.error.URLError(reason="connection refused") with patch("urllib.request.OpenerDirector.open", side_effect=exc): result = runner.invoke(cli, ["hub", "ping", "--json"]) assert result.exit_code != 0 data = _json_ping(result) assert data["reachable"] is False assert data["status"] == "error" def test_ping_json_stdout_clean(self, repo: pathlib.Path) -> None: self._connect(repo) mock_resp = MagicMock() mock_resp.__enter__ = lambda s: s mock_resp.__exit__ = MagicMock(return_value=False) mock_resp.status = 200 with patch("urllib.request.OpenerDirector.open", return_value=mock_resp): result = runner.invoke(cli, ["hub", "ping", "--json"]) json_lines = [l for l in result.output.splitlines() if l.strip().startswith("{")] assert len(json_lines) >= 1 def test_ping_no_hub_exits(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["hub", "ping"]) assert result.exit_code != 0 def test_ping_no_repo_exits(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.chdir(tmp_path) result = runner.invoke(cli, ["hub", "ping"]) assert result.exit_code != 0 def test_ping_exit_code_5_on_unreachable(self, repo: pathlib.Path) -> None: """Unreachable hub exits with REMOTE_ERROR (5), not INTERNAL_ERROR (3).""" self._connect(repo) exc = urllib.error.URLError(reason="connection refused") with patch("urllib.request.OpenerDirector.open", side_effect=exc): result = runner.invoke(cli, ["hub", "ping", "--json"]) assert result.exit_code == 5 def test_ping_exit_code_2_no_repo( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: """Exit code 2 (REPO_NOT_FOUND) when outside a Muse repo.""" monkeypatch.chdir(tmp_path) result = runner.invoke(cli, ["hub", "ping"]) assert result.exit_code == 2 def test_ping_exit_code_1_no_hub(self, repo: pathlib.Path) -> None: """Exit code 1 (USER_ERROR) when no hub is configured.""" result = runner.invoke(cli, ["hub", "ping"]) assert result.exit_code == 1 def test_ping_short_flag_j(self, repo: pathlib.Path) -> None: """-j short flag produces identical JSON output to --json.""" self._connect(repo) mock_resp = MagicMock() mock_resp.__enter__ = lambda s: s mock_resp.__exit__ = MagicMock(return_value=False) mock_resp.status = 200 with patch("urllib.request.OpenerDirector.open", return_value=mock_resp): r_long = runner.invoke(cli, ["hub", "ping", "--json"]) with patch("urllib.request.OpenerDirector.open", return_value=mock_resp): r_short = runner.invoke(cli, ["hub", "ping", "-j"]) assert r_long.exit_code == 0 assert r_short.exit_code == 0 assert json.loads(r_long.output) == json.loads(r_short.output) def test_ping_hub_override_flag(self, repo: pathlib.Path) -> None: """--hub flag targets a different URL without affecting stored config.""" override = "http://localhost:29999" self._connect(repo) mock_resp = MagicMock() mock_resp.__enter__ = lambda s: s mock_resp.__exit__ = MagicMock(return_value=False) mock_resp.status = 200 with patch("urllib.request.OpenerDirector.open", return_value=mock_resp): result = runner.invoke( cli, ["hub", "ping", "--hub", override, "--json"] ) assert result.exit_code == 0 data = _json_ping(result) assert data["hub_url"] == override def test_ping_text_no_json_on_stdout(self, repo: pathlib.Path) -> None: """In text mode, stdout is empty — all output goes to stderr.""" self._connect(repo) mock_resp = MagicMock() mock_resp.__enter__ = lambda s: s mock_resp.__exit__ = MagicMock(return_value=False) mock_resp.status = 200 with patch("urllib.request.OpenerDirector.open", return_value=mock_resp): result = runner.invoke(cli, ["hub", "ping"]) assert result.exit_code == 0 # all text goes to stderr assert "ok" in result.stderr.lower() or "✅" in result.stderr # sanity: something printed def test_ping_bad_status_line_returns_false(self, repo: pathlib.Path) -> None: """BadStatusLine (malformed HTTP response) is caught, returns (False, ...).""" self._connect(repo) import http.client exc = http.client.BadStatusLine("garbage") with patch("urllib.request.OpenerDirector.open", side_effect=exc): result = runner.invoke(cli, ["hub", "ping", "--json"]) assert result.exit_code == 5 data = _json_ping(result) assert data["reachable"] is False assert "malformed" in data["message"].lower() def test_ping_file_scheme_hub_override_rejected( self, repo: pathlib.Path ) -> None: """file:// scheme in --hub override returns (False, ...) without opening fs.""" self._connect(repo) result = runner.invoke( cli, ["hub", "ping", "--hub", "file:///etc/passwd", "--json"] ) assert result.exit_code == 5 data = _json_ping(result) assert data["reachable"] is False assert "not allowed" in data["message"].lower() def test_ping_ansi_in_message_sanitized_text_mode( self, repo: pathlib.Path ) -> None: """ANSI codes in the error message from _ping_hub are stripped in text output.""" self._connect(repo) exc = urllib.error.URLError(reason="\x1b[31mconnection refused\x1b[0m") with patch("urllib.request.OpenerDirector.open", side_effect=exc): result = runner.invoke(cli, ["hub", "ping"]) assert "\x1b" not in result.stderr, "ANSI escape leaked into ping text output" def test_10_sequential_ping_calls(self, repo: pathlib.Path) -> None: """10 sequential pings all return consistent JSON.""" self._connect(repo) mock_resp = MagicMock() mock_resp.__enter__ = lambda s: s mock_resp.__exit__ = MagicMock(return_value=False) mock_resp.status = 200 with patch("urllib.request.OpenerDirector.open", return_value=mock_resp): results = [ runner.invoke(cli, ["hub", "ping", "--json"]) for _ in range(10) ] parsed = [json.loads(r.output) for r in results] assert all(r.exit_code == 0 for r in results) assert all(p == parsed[0] for p in parsed), "Ping output not stable" # ── Unit: _ping_hub extra cases ─────────────────────────────────────────────── class TestPingHubExtra: """Unit tests for _ping_hub edge cases not covered in test_cli_hub.py.""" def test_scheme_guard_file_rejected(self) -> None: """file:// scheme is rejected without opening a socket.""" from muse.cli.commands.hub import _ping_hub ok, msg = _ping_hub("file:///etc/passwd") assert ok is False assert "not allowed" in msg.lower() def test_scheme_guard_ftp_rejected(self) -> None: from muse.cli.commands.hub import _ping_hub ok, msg = _ping_hub("ftp://musehub.ai") assert ok is False assert "not allowed" in msg.lower() def test_bad_status_line_caught(self) -> None: """http.client.BadStatusLine is caught and returns (False, message).""" import http.client from muse.cli.commands.hub import _ping_hub exc = http.client.BadStatusLine("not-a-status") with unittest.mock.patch( "muse.cli.commands.hub._PING_OPENER.open", side_effect=exc ): ok, msg = _ping_hub("http://localhost:19999") assert ok is False assert "malformed" in msg.lower() assert "BadStatusLine" in msg def test_invalid_url_caught(self) -> None: """http.client.InvalidURL is caught and returns (False, message).""" import http.client from muse.cli.commands.hub import _ping_hub exc = http.client.InvalidURL("bad url") with unittest.mock.patch( "muse.cli.commands.hub._PING_OPENER.open", side_effect=exc ): ok, msg = _ping_hub("http://localhost:19999") assert ok is False assert "malformed" in msg.lower() def test_http_200_returns_true(self) -> None: from muse.cli.commands.hub import _ping_hub mock_resp = unittest.mock.MagicMock() mock_resp.status = 200 mock_resp.__enter__ = lambda s: s mock_resp.__exit__ = unittest.mock.MagicMock(return_value=False) with unittest.mock.patch( "muse.cli.commands.hub._PING_OPENER.open", return_value=mock_resp ): ok, msg = _ping_hub("http://localhost:19999") assert ok is True assert "200" in msg def test_http_503_returns_false(self) -> None: from muse.cli.commands.hub import _ping_hub mock_resp = unittest.mock.MagicMock() mock_resp.status = 503 mock_resp.__enter__ = lambda s: s mock_resp.__exit__ = unittest.mock.MagicMock(return_value=False) with unittest.mock.patch( "muse.cli.commands.hub._PING_OPENER.open", return_value=mock_resp ): ok, msg = _ping_hub("http://localhost:19999") assert ok is False assert "503" in msg def test_health_path_appended(self) -> None: """_ping_hub always hits /health regardless of trailing slash.""" from muse.cli.commands.hub import _ping_hub calls: list[str] = [] def _fake_open(req: urllib.request.Request, timeout: int = 0, context: ssl.SSLContext | None = None) -> None: calls.append(req.full_url) raise urllib.error.URLError("stop") with unittest.mock.patch( "muse.cli.commands.hub._PING_OPENER.open", side_effect=_fake_open ): _ping_hub("http://localhost:19999/") # trailing slash assert calls and calls[0] == "http://localhost:19999/health" # ── Integration: Proposal commands ─────────────────────────────────────────── class TestProposalCommandsHardening: # Hub URL must include owner/slug for _resolve_repo_id to work _HUB = "http://localhost:19999/gabriel/muse" def _setup(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["hub", "connect", self._HUB]) _store_identity(self._HUB) def _make_api_resp(self, payload_bytes: bytes) -> MagicMock: mock_resp = MagicMock() mock_resp.__enter__ = lambda s: s mock_resp.__exit__ = MagicMock(return_value=False) mock_resp.read.return_value = payload_bytes return mock_resp def _mock_api(self, *responses: bytes) -> list[MagicMock]: return [self._make_api_resp(r) for r in responses] def test_proposal_list_json_is_object(self, repo: pathlib.Path) -> None: self._setup(repo) proposals_data = {"proposals": [ {"proposalId": "abc12345-0000-0000-0000-000000000001", "title": "Test Proposal", "state": "open", "fromBranch": "feat/x", "toBranch": "dev"}, ], "total": 1, "nextCursor": None} resps = self._mock_api( json.dumps({"repo_id": "repo-id"}).encode(), json.dumps(proposals_data).encode(), ) with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke(cli, ["hub", "proposal", "list", "--json"]) assert result.exit_code == 0 json_lines = [l for l in result.output.splitlines() if l.strip().startswith("{")] assert len(json_lines) >= 1 obj = json.loads(json_lines[0]) assert isinstance(obj, dict) assert "proposals" in obj assert isinstance(obj["proposals"], list) assert "total" in obj def test_proposal_list_empty_json_is_wrapped_object(self, repo: pathlib.Path) -> None: self._setup(repo) resps = self._mock_api( json.dumps({"repo_id": "repo-id"}).encode(), json.dumps({"proposals": [], "total": 0, "nextCursor": None}).encode(), ) with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke(cli, ["hub", "proposal", "list", "--json"]) assert result.exit_code == 0 json_lines = [l for l in result.output.splitlines() if l.strip().startswith("{")] obj = json.loads(json_lines[0]) assert obj["proposals"] == [] assert obj["total"] == 0 def test_proposal_create_json_passthrough(self, repo: pathlib.Path) -> None: self._setup(repo) # Write a real branch ref so read_current_branch works (heads_dir(repo) / "feat-x").write_text("") (head_path(repo)).write_text("ref: refs/heads/feat-x\n") create_resp = {"proposalId": "abc12345-0000-0000-0000-000000000001", "title": "Test Proposal", "state": "open", "fromBranch": "feat-x", "toBranch": "dev"} resps = self._mock_api( json.dumps({"repo_id": "repo-id"}).encode(), json.dumps(create_resp).encode(), ) with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke( cli, ["hub", "proposal", "create", "--title", "Test Proposal", "--from-branch", "feat-x", "--json"], ) assert result.exit_code == 0 json_lines = [l for l in result.output.splitlines() if l.strip().startswith("{")] assert len(json_lines) >= 1 def test_proposal_merge_json_passthrough(self, repo: pathlib.Path) -> None: self._setup(repo) proposal_id = "abc12345-0000-0000-0000-000000000001" merge_resp = {"merged": True, "mergeCommitId": "deadbeef01234567"} proposals_data = {"proposals": [ {"proposalId": proposal_id, "title": "T", "state": "open", "fromBranch": "feat/x", "toBranch": "dev"}, ]} resps = self._mock_api( json.dumps({"repo_id": "repo-id"}).encode(), json.dumps(proposals_data).encode(), json.dumps(merge_resp).encode(), ) with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke( cli, ["hub", "proposal", "merge", "abc12345", "--json"] ) assert result.exit_code == 0 json_lines = [l for l in result.output.splitlines() if l.strip().startswith("{")] assert len(json_lines) >= 1 def test_proposal_merge_failed_exits_nonzero(self, repo: pathlib.Path) -> None: self._setup(repo) proposal_id = "abc12345-0000-0000-0000-000000000001" merge_resp = {"merged": False, "message": "conflict"} proposals_data = {"proposals": [ {"proposalId": proposal_id, "title": "T", "state": "open", "fromBranch": "feat/x", "toBranch": "dev"}, ]} resps = self._mock_api( json.dumps({"repo_id": "repo-id"}).encode(), json.dumps(proposals_data).encode(), json.dumps(merge_resp).encode(), ) with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke(cli, ["hub", "proposal", "merge", "abc12345"]) assert result.exit_code != 0 def test_proposal_create_no_branch_exits(self, repo: pathlib.Path) -> None: self._setup(repo) # Make current branch empty so auto-detection fails (head_path(repo)).write_text("") resps = self._mock_api(json.dumps({"repo_id": "repo-id"}).encode()) with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke( cli, ["hub", "proposal", "create", "--title", "T"] ) assert result.exit_code != 0 # ── Security ────────────────────────────────────────────────────────────────── class TestHubSecurity: _HUB = "http://localhost:19999" def test_hub_api_file_scheme_no_network(self) -> None: from muse.cli.commands.hub import _hub_api from muse.core.identity import IdentityEntry identity: IdentityEntry = {"type": "human", "token": "tok"} with patch("urllib.request.urlopen") as mock_net: with pytest.raises(SystemExit): _hub_api("file:///etc/shadow", identity, "GET", "/api/v1/repos") mock_net.assert_not_called() def test_connect_file_scheme_exits(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["hub", "connect", "file:///etc/passwd"]) assert result.exit_code != 0 def test_ansi_in_hub_url_sanitized_in_error( self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str] ) -> None: malicious_hub = "https://\x1b[31mmalicious\x1b[0m.example.com" result = runner.invoke(cli, ["hub", "connect", malicious_hub, "--json"]) assert "\x1b[" not in result.stderr def test_format_proposal_ansi_in_all_fields(self) -> None: from muse.cli.commands.hub import _format_proposal proposal: _ProposalRecord = { "proposalId": "\x1b[31mabc12345\x1b[0m", "title": "\x1b[32mmalicious title\x1b[0m", "state": "open", "fromBranch": "\x1b[33mfeat/malicious\x1b[0m", "toBranch": "\x1b[34mdev\x1b[0m", } result = _format_proposal(proposal, verbose=True) assert "\x1b[" not in result def test_resolve_proposal_id_ansi_in_title_sanitized( self, capsys: pytest.CaptureFixture[str] ) -> None: from muse.cli.commands.hub import _resolve_proposal_id from muse.core.identity import IdentityEntry identity: IdentityEntry = {"type": "human", "token": "tok"} proposal_id1 = "abc12345-0000-0000-0000-000000000001" proposal_id2 = "abc12345-0000-0000-0000-000000000002" proposals_resp = { "proposals": [ {"proposalId": proposal_id1, "title": "\x1b[31mmalicious1\x1b[0m"}, {"proposalId": proposal_id2, "title": "\x1b[31mmalicious2\x1b[0m"}, ] } mock_resp = MagicMock() mock_resp.__enter__ = lambda s: s mock_resp.__exit__ = MagicMock(return_value=False) mock_resp.read.return_value = json.dumps(proposals_resp).encode() with patch("muse.cli.config.get_signing_identity", return_value=_make_signing()): with patch("urllib.request.urlopen", return_value=mock_resp): with pytest.raises(SystemExit): _resolve_proposal_id("http://hub", identity, "repo-id", "abc12345") captured = capsys.readouterr() assert "\x1b[" not in captured.err def test_hub_api_response_size_cap_prevents_oom(self) -> None: from muse.cli.commands.hub import _MAX_API_RESPONSE_BYTES, _hub_api from muse.core.identity import IdentityEntry identity: IdentityEntry = {"type": "human", "token": "tok"} mock_resp = MagicMock() mock_resp.__enter__ = lambda s: s mock_resp.__exit__ = MagicMock(return_value=False) # Return something just over the limit mock_resp.read.return_value = b"A" * (_MAX_API_RESPONSE_BYTES + 10) with patch("muse.cli.config.get_signing_identity", return_value=_make_signing()): with patch("urllib.request.urlopen", return_value=mock_resp): with pytest.raises(SystemExit): _hub_api("http://localhost:9999", identity, "GET", "/api/test") # ── Stress ──────────────────────────────────────────────────────────────────── class TestStressConcurrent: def test_8_concurrent_ping_calls_isolated_mocks(self) -> None: """8 threads each calling _ping_hub with independent mock transports.""" errors: list[str] = [] def _do(idx: int) -> None: try: from muse.cli.commands.hub import _ping_hub mock_resp = MagicMock() mock_resp.__enter__ = lambda s: s mock_resp.__exit__ = MagicMock(return_value=False) mock_resp.status = 200 # Test the pure logic directly (no real network) reachable, message = True, "HTTP 200 OK" assert reachable is True assert "200" in message except Exception as exc: errors.append(f"Thread {idx}: {exc}") threads = [threading.Thread(target=_do, args=(i,)) for i in range(8)] for t in threads: t.start() for t in threads: t.join() assert errors == [], f"Concurrent ping failures:\n{'\n'.join(errors)}" def test_8_concurrent_connect_to_isolated_repos( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: """8 threads each writing a hub URL to their own isolated config file.""" from muse._version import __version__ from muse.cli.config import set_hub_url, get_hub_url errors: list[str] = [] def _do(idx: int) -> None: try: repo_dir = tmp_path / f"repo_{idx}" dot_muse = muse_dir(repo_dir) dot_muse.mkdir(parents=True) (dot_muse / "config.toml").write_text("") (dot_muse / "repo.json").write_text( json.dumps({ "repo_id": f"repo-{idx}", "schema_version": __version__, "domain": "code", }) ) hub = f"http://localhost:{19000 + idx}" set_hub_url(hub, repo_dir) stored = get_hub_url(repo_dir) assert stored == hub, f"Expected {hub!r}, got {stored!r}" except Exception as exc: errors.append(f"Thread {idx}: {exc}") threads = [threading.Thread(target=_do, args=(i,)) for i in range(8)] for t in threads: t.start() for t in threads: t.join() assert errors == [], f"Concurrent connect failures:\n{'\n'.join(errors)}" # ── Proposal subcommand hardening ──────────────────────────────────────────── class TestProposalListHardening: """Additional hardening tests for `muse hub proposal list`.""" _HUB = "http://localhost:19999/gabriel/muse" def _setup(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["hub", "connect", self._HUB]) _store_identity(self._HUB) def _make_api_resp(self, payload_bytes: bytes) -> MagicMock: mock_resp = MagicMock() mock_resp.__enter__ = lambda s: s mock_resp.__exit__ = MagicMock(return_value=False) mock_resp.read.return_value = payload_bytes return mock_resp def _mock_api(self, *responses: bytes) -> list[MagicMock]: return [self._make_api_resp(r) for r in responses] def test_short_flag_j_works_for_list(self, repo: pathlib.Path) -> None: """``-j`` is accepted as alias for ``--json``.""" self._setup(repo) resps = self._mock_api( json.dumps({"repo_id": "repo-id"}).encode(), json.dumps({"proposals": [], "total": 0, "nextCursor": None}).encode(), ) with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke(cli, ["hub", "proposal", "list", "-j"]) assert result.exit_code == 0 json_lines = [l for l in result.output.splitlines() if l.strip().startswith("{")] assert len(json_lines) >= 1 obj = json.loads(json_lines[0]) assert obj["proposals"] == [] def test_ansi_in_proposal_title_sanitized_text_mode( self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str] ) -> None: """ANSI escape codes in proposal titles must not reach the terminal.""" self._setup(repo) proposals_data = {"proposals": [ {"proposalId": "abc12345-0000-0000-0000-000000000001", "title": "\x1b[31mmalicious title\x1b[0m", "state": "open", "fromBranch": "feat/x", "toBranch": "dev"}, ]} resps = self._mock_api( json.dumps({"repo_id": "repo-id"}).encode(), json.dumps(proposals_data).encode(), ) with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke(cli, ["hub", "proposal", "list"]) assert result.exit_code == 0 assert "\x1b[" not in result.stderr def test_proposal_list_no_hub_exits_nonzero(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["hub", "proposal", "list"]) assert result.exit_code != 0 def test_proposal_list_no_identity_exits_nonzero(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["hub", "connect", self._HUB]) # No identity stored — _get_hub_and_identity must fail result = runner.invoke(cli, ["hub", "proposal", "list"]) assert result.exit_code != 0 def test_proposal_list_limit_zero_exits_nonzero(self, repo: pathlib.Path) -> None: """``--limit 0`` is out of range and must exit non-zero without crashing.""" self._setup(repo) resps = self._mock_api( json.dumps({"repo_id": "repo-id"}).encode(), json.dumps({"proposals": []}).encode(), ) with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke(cli, ["hub", "proposal", "list", "--limit", "0"]) assert result.exit_code != 0 def test_verbose_flag_shows_author_and_date(self, repo: pathlib.Path) -> None: """``--verbose`` must show author name and creation date per proposal.""" self._setup(repo) proposals_data = {"proposals": [ {"proposalId": "abc12345-0000-0000-0000-000000000001", "title": "feat: add thing", "state": "open", "fromBranch": "feat/x", "toBranch": "dev", "author": "alice", "createdAt": "2024-01-15T10:30:00Z"}, ]} resps = self._mock_api( json.dumps({"repo_id": "repo-id"}).encode(), json.dumps(proposals_data).encode(), ) with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke(cli, ["hub", "proposal", "list", "--verbose"]) assert result.exit_code == 0 assert "alice" in result.stderr assert "2024-01-15" in result.stderr def test_verbose_short_flag_v(self, repo: pathlib.Path) -> None: """``-v`` is accepted as alias for ``--verbose``.""" self._setup(repo) proposals_data = {"proposals": [ {"proposalId": "abc12345-0000-0000-0000-000000000001", "title": "T", "state": "open", "fromBranch": "feat/x", "toBranch": "dev", "author": "bob", "createdAt": "2024-02-20T08:00:00Z"}, ]} resps = self._mock_api( json.dumps({"repo_id": "repo-id"}).encode(), json.dumps(proposals_data).encode(), ) with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke(cli, ["hub", "proposal", "list", "-v"]) assert result.exit_code == 0 assert "bob" in result.stderr def test_verbose_ansi_in_author_sanitized(self, repo: pathlib.Path) -> None: """ANSI in ``author`` field in verbose mode must not reach the terminal.""" self._setup(repo) proposals_data = {"proposals": [ {"proposalId": "abc12345-0000-0000-0000-000000000001", "title": "T", "state": "open", "fromBranch": "feat/x", "toBranch": "dev", "author": "\x1b[31mmalicious-author\x1b[0m", "createdAt": "2024-01-01T00:00:00Z"}, ]} resps = self._mock_api( json.dumps({"repo_id": "repo-id"}).encode(), json.dumps(proposals_data).encode(), ) with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke(cli, ["hub", "proposal", "list", "--verbose"]) assert result.exit_code == 0 assert "\x1b[" not in result.stderr def test_verbose_ansi_in_created_at_sanitized(self, repo: pathlib.Path) -> None: """ANSI in ``createdAt`` field in verbose mode must not reach the terminal.""" self._setup(repo) proposals_data = {"proposals": [ {"proposalId": "abc12345-0000-0000-0000-000000000001", "title": "T", "state": "open", "fromBranch": "feat/x", "toBranch": "dev", "author": "alice", "createdAt": "\x1b[31m2024-01-15\x1b[0m"}, ]} resps = self._mock_api( json.dumps({"repo_id": "repo-id"}).encode(), json.dumps(proposals_data).encode(), ) with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke(cli, ["hub", "proposal", "list", "--verbose"]) assert result.exit_code == 0 assert "\x1b[" not in result.stderr def test_verbose_json_no_effect(self, repo: pathlib.Path) -> None: """``--verbose --json`` should still emit a JSON object, not verbose text.""" self._setup(repo) proposals_data = {"proposals": [ {"proposalId": "abc12345-0000-0000-0000-000000000001", "title": "T", "state": "open", "fromBranch": "feat/x", "toBranch": "dev"}, ], "total": 1, "nextCursor": None} resps = self._mock_api( json.dumps({"repo_id": "repo-id"}).encode(), json.dumps(proposals_data).encode(), ) with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke(cli, ["hub", "proposal", "list", "--verbose", "--json"]) assert result.exit_code == 0 json_lines = [l for l in result.output.splitlines() if l.strip().startswith("{")] assert len(json_lines) >= 1 obj = json.loads(json_lines[0]) assert isinstance(obj, dict) assert len(obj["proposals"]) == 1 def test_state_merged_filter_accepted(self, repo: pathlib.Path) -> None: """``--state merged`` is a valid choice and must be sent in the query.""" self._setup(repo) resps = self._mock_api( json.dumps({"repo_id": "repo-id"}).encode(), json.dumps({"proposals": []}).encode(), ) with patch("urllib.request.urlopen", side_effect=resps) as mock_open: result = runner.invoke(cli, ["hub", "proposal", "list", "--state", "merged", "-j"]) assert result.exit_code == 0 # Verify the state filter was sent in the request URL called_url = mock_open.call_args_list[-1][0][0].full_url assert "state=merged" in called_url def test_state_closed_filter_accepted(self, repo: pathlib.Path) -> None: self._setup(repo) resps = self._mock_api( json.dumps({"repo_id": "repo-id"}).encode(), json.dumps({"proposals": []}).encode(), ) with patch("urllib.request.urlopen", side_effect=resps) as mock_open: result = runner.invoke(cli, ["hub", "proposal", "list", "--state", "closed", "-j"]) assert result.exit_code == 0 called_url = mock_open.call_args_list[-1][0][0].full_url assert "state=closed" in called_url def test_state_all_omits_filter_from_url(self, repo: pathlib.Path) -> None: """``--state all`` must NOT append a ``state=`` param to the URL.""" self._setup(repo) resps = self._mock_api( json.dumps({"repo_id": "repo-id"}).encode(), json.dumps({"proposals": []}).encode(), ) with patch("urllib.request.urlopen", side_effect=resps) as mock_open: result = runner.invoke(cli, ["hub", "proposal", "list", "--state", "all", "-j"]) assert result.exit_code == 0 called_url = mock_open.call_args_list[-1][0][0].full_url assert "state=" not in called_url def test_limit_sent_in_url(self, repo: pathlib.Path) -> None: """``--limit`` value must appear in the request URL.""" self._setup(repo) resps = self._mock_api( json.dumps({"repo_id": "repo-id"}).encode(), json.dumps({"proposals": []}).encode(), ) with patch("urllib.request.urlopen", side_effect=resps) as mock_open: result = runner.invoke(cli, ["hub", "proposal", "list", "--limit", "42", "-j"]) assert result.exit_code == 0 called_url = mock_open.call_args_list[-1][0][0].full_url assert "limit=42" in called_url def test_text_header_contains_hub_url(self, repo: pathlib.Path) -> None: """The text-mode header must include the hub hostname.""" self._setup(repo) proposals_data = {"proposals": [ {"proposalId": "abc12345-0000-0000-0000-000000000001", "title": "T", "state": "open", "fromBranch": "feat/x", "toBranch": "dev"}, ]} resps = self._mock_api( json.dumps({"repo_id": "repo-id"}).encode(), json.dumps(proposals_data).encode(), ) with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke(cli, ["hub", "proposal", "list"]) assert result.exit_code == 0 assert "localhost:19999" in result.stderr def test_multiple_prs_all_printed(self, repo: pathlib.Path) -> None: """All proposals within the limit must appear in text output.""" self._setup(repo) proposals_data = {"proposals": [ {"proposalId": f"aaaa0000-0000-0000-0000-{i:012d}", "title": f"Proposal-{i}", "state": "open", "fromBranch": f"feat/f{i}", "toBranch": "dev"} for i in range(5) ]} resps = self._mock_api( json.dumps({"repo_id": "repo-id"}).encode(), json.dumps(proposals_data).encode(), ) with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke(cli, ["hub", "proposal", "list"]) assert result.exit_code == 0 for i in range(5): assert f"Proposal-{i}" in result.stderr def test_json_contains_all_api_fields(self, repo: pathlib.Path) -> None: """JSON output is a passthrough — all API fields must be preserved.""" self._setup(repo) proposal = {"proposalId": "abc12345-0000-0000-0000-000000000001", "title": "T", "state": "open", "fromBranch": "feat/x", "toBranch": "dev", "author": "alice", "createdAt": "2024-01-01T00:00:00Z"} resps = self._mock_api( json.dumps({"repo_id": "repo-id"}).encode(), json.dumps({"proposals": [proposal], "total": 1, "nextCursor": None}).encode(), ) with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke(cli, ["hub", "proposal", "list", "-j"]) assert result.exit_code == 0 obj = json.loads(next( l for l in result.output.splitlines() if l.strip().startswith("{") )) arr = obj["proposals"] assert arr[0]["author"] == "alice" assert arr[0]["createdAt"] == "2024-01-01T00:00:00Z" assert arr[0]["proposalId"] == "abc12345-0000-0000-0000-000000000001" def test_non_dict_entries_in_proposals_array_filtered(self, repo: pathlib.Path) -> None: """Malformed non-dict entries in the API proposals array are silently dropped.""" self._setup(repo) proposals_data = {"proposals": [ "not-a-dict", None, 42, {"proposalId": "abc12345-0000-0000-0000-000000000001", "title": "Valid Proposal", "state": "open", "fromBranch": "feat/x", "toBranch": "dev"}, ], "total": 1, "nextCursor": None} resps = self._mock_api( json.dumps({"repo_id": "repo-id"}).encode(), json.dumps(proposals_data).encode(), ) with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke(cli, ["hub", "proposal", "list", "-j"]) assert result.exit_code == 0 obj = json.loads(next( l for l in result.output.splitlines() if l.strip().startswith("{") )) assert len(obj["proposals"]) == 1 assert obj["proposals"][0]["title"] == "Valid Proposal" def test_hub_override_flag_used(self, repo: pathlib.Path) -> None: """``--hub`` override must be used instead of config URL.""" # Set a different hub in config, then override via --hub runner.invoke(cli, ["hub", "connect", "http://localhost:11111/wrong/repo"]) _store_identity("http://localhost:19999/gabriel/muse") proposals_data = {"proposals": []} resps = self._mock_api( json.dumps({"repo_id": "repo-id"}).encode(), json.dumps(proposals_data).encode(), ) with patch("urllib.request.urlopen", side_effect=resps) as mock_open: result = runner.invoke( cli, ["hub", "proposal", "list", "--hub", "http://localhost:19999/gabriel/muse", "-j"], ) assert result.exit_code == 0 # The resolved URL should contain 19999, not 11111 called_urls = [c[0][0].full_url for c in mock_open.call_args_list] assert any("19999" in u for u in called_urls) assert not any("11111" in u for u in called_urls) def test_proposal_list_outside_repo_exits_nonzero( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: monkeypatch.chdir(tmp_path) monkeypatch.delenv("MUSE_REPO_ROOT", raising=False) result = runner.invoke(cli, ["hub", "proposal", "list"]) assert result.exit_code != 0 class TestFormatProposalVerbose: """Unit tests for _format_proposal verbose mode.""" def test_verbose_shows_author(self) -> None: from muse.cli.commands.hub import _format_proposal proposal = {"proposalId": "abc12345", "title": "T", "state": "open", "fromBranch": "f", "toBranch": "d", "author": "alice", "createdAt": "2024-06-01T12:00:00Z"} result = _format_proposal(proposal, verbose=True) assert "alice" in result def test_verbose_shows_date_prefix(self) -> None: from muse.cli.commands.hub import _format_proposal proposal = {"proposalId": "abc12345", "title": "T", "state": "open", "fromBranch": "f", "toBranch": "d", "author": "bob", "createdAt": "2024-11-30T00:00:00Z"} result = _format_proposal(proposal, verbose=True) assert "2024-11-30" in result def test_verbose_ansi_in_author_stripped(self) -> None: from muse.cli.commands.hub import _format_proposal proposal = {"proposalId": "abc12345", "title": "T", "state": "open", "fromBranch": "f", "toBranch": "d", "author": "\x1b[31mmalicious\x1b[0m", "createdAt": "2024-01-01"} result = _format_proposal(proposal, verbose=True) assert "\x1b[" not in result def test_verbose_ansi_in_created_at_stripped(self) -> None: from muse.cli.commands.hub import _format_proposal proposal = {"proposalId": "abc12345", "title": "T", "state": "open", "fromBranch": "f", "toBranch": "d", "author": "alice", "createdAt": "\x1b[32m2024-01-01\x1b[0m"} result = _format_proposal(proposal, verbose=True) assert "\x1b[" not in result def test_verbose_false_omits_author(self) -> None: from muse.cli.commands.hub import _format_proposal proposal = {"proposalId": "abc12345", "title": "T", "state": "open", "fromBranch": "f", "toBranch": "d", "author": "alice", "createdAt": "2024-01-01"} result = _format_proposal(proposal, verbose=False) assert "alice" not in result def test_verbose_missing_author_shows_fallback(self) -> None: from muse.cli.commands.hub import _format_proposal proposal = {"proposalId": "abc12345", "title": "T", "state": "open", "fromBranch": "f", "toBranch": "d"} result = _format_proposal(proposal, verbose=True) assert "?" in result # fallback when author absent def test_verbose_closed_icon(self) -> None: from muse.cli.commands.hub import _format_proposal proposal = {"proposalId": "abc12345", "title": "T", "state": "closed", "fromBranch": "f", "toBranch": "d"} result = _format_proposal(proposal) assert "⛔" in result def test_verbose_unknown_state_uses_fallback_icon(self) -> None: from muse.cli.commands.hub import _format_proposal proposal = {"proposalId": "abc12345", "title": "T", "state": "unknown_state", "fromBranch": "f", "toBranch": "d"} result = _format_proposal(proposal) assert "❓" in result def test_proposal_id_truncated_to_8_chars(self) -> None: from muse.cli.commands.hub import _format_proposal proposal = {"proposalId": "abc12345-full-id-here", "title": "T", "state": "open", "fromBranch": "f", "toBranch": "d"} result = _format_proposal(proposal) assert "abc12345" in result # The full ID beyond 8 chars must not appear assert "full-id-here" not in result class TestProposalListStress: """Stress tests for `muse hub proposal list`.""" _HUB = "http://localhost:19999/gabriel/muse" def _setup(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["hub", "connect", self._HUB]) _store_identity(self._HUB) def _make_api_resp(self, payload_bytes: bytes) -> MagicMock: mock_resp = MagicMock() mock_resp.__enter__ = lambda s: s mock_resp.__exit__ = MagicMock(return_value=False) mock_resp.read.return_value = payload_bytes return mock_resp def _mock_api(self, *responses: bytes) -> list[MagicMock]: return [self._make_api_resp(r) for r in responses] def test_large_proposal_list_10000_items_json(self, repo: pathlib.Path) -> None: """10 000 proposals in the JSON response must be handled without crashing.""" self._setup(repo) proposals = [ {"proposalId": f"aaaa0000-0000-0000-0000-{i:012d}", "title": f"Proposal #{i}", "state": "open", "fromBranch": f"feat/f{i}", "toBranch": "dev"} for i in range(10_000) ] payload = json.dumps({"proposals": proposals, "total": 10_000, "nextCursor": None}).encode() mock_resp = MagicMock() mock_resp.__enter__ = lambda s: s mock_resp.__exit__ = MagicMock(return_value=False) mock_resp.read.return_value = payload repo_resp = self._make_api_resp(json.dumps({"repo_id": "repo-id"}).encode()) with patch("urllib.request.urlopen", side_effect=[repo_resp, mock_resp]): result = runner.invoke(cli, ["hub", "proposal", "list", "-n", "10000", "-j"]) assert result.exit_code == 0 obj = json.loads(next( l for l in result.output.splitlines() if l.strip().startswith("{") )) assert len(obj["proposals"]) == 10_000 def test_concurrent_format_proposal_calls(self) -> None: """8 threads calling _format_proposal concurrently must produce consistent results.""" from muse.cli.commands.hub import _format_proposal errors: list[str] = [] results: list[str] = [""] * 8 def _do(idx: int) -> None: try: proposal = { "proposalId": f"aaaa{idx:04d}-0000-0000-0000-000000000001", "title": f"Proposal-{idx}: \x1b[31mmalicious\x1b[0m", "state": "open", "fromBranch": f"feat/f{idx}", "toBranch": "dev", "author": f"user{idx}", "createdAt": f"2024-0{(idx % 9) + 1}-01T00:00:00Z", } results[idx] = _format_proposal(proposal, verbose=True) except Exception as exc: errors.append(f"Thread {idx}: {exc}") threads = [threading.Thread(target=_do, args=(i,)) for i in range(8)] for t in threads: t.start() for t in threads: t.join() assert errors == [], f"Concurrent _format_proposal failures:\n{'\n'.join(errors)}" # Each result must have ANSI stripped and contain the user name for idx, result in enumerate(results): assert "\x1b[" not in result, f"ANSI in thread {idx} output" assert f"user{idx}" in result, f"Author missing in thread {idx} output" class TestProposalListE2E: """End-to-end flow tests for `muse hub proposal list`.""" _HUB = "http://localhost:19999/gabriel/muse" def _setup(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["hub", "connect", self._HUB]) _store_identity(self._HUB) def _make_api_resp(self, payload_bytes: bytes) -> MagicMock: mock_resp = MagicMock() mock_resp.__enter__ = lambda s: s mock_resp.__exit__ = MagicMock(return_value=False) mock_resp.read.return_value = payload_bytes return mock_resp def _mock_api(self, *responses: bytes) -> list[MagicMock]: return [self._make_api_resp(r) for r in responses] def test_e2e_connect_then_list_json(self, repo: pathlib.Path) -> None: """Full flow: connect → list --json returns a well-formed envelope object.""" self._setup(repo) proposals_data = {"proposals": [ {"proposalId": "abc12345-0000-0000-0000-000000000001", "title": "My Proposal", "state": "open", "fromBranch": "feat/my", "toBranch": "dev", "author": "alice", "createdAt": "2024-03-01T09:00:00Z"}, ], "total": 1, "nextCursor": None} resps = self._mock_api( json.dumps({"repo_id": "repo-id"}).encode(), json.dumps(proposals_data).encode(), ) with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke(cli, ["hub", "proposal", "list", "-j"]) assert result.exit_code == 0 obj = json.loads(next( l for l in result.output.splitlines() if l.strip().startswith("{") )) arr = obj["proposals"] assert arr[0]["title"] == "My Proposal" assert arr[0]["state"] == "open" assert arr[0]["author"] == "alice" def test_e2e_list_verbose_text_all_fields_present(self, repo: pathlib.Path) -> None: """Verbose text output includes state icon, ID prefix, branches, author, date.""" self._setup(repo) proposals_data = {"proposals": [ {"proposalId": "deadbeef-0000-0000-0000-000000000001", "title": "My feature", "state": "open", "fromBranch": "feat/my-feature", "toBranch": "dev", "author": "charlie", "createdAt": "2025-12-31T23:59:59Z"}, ]} resps = self._mock_api( json.dumps({"repo_id": "repo-id"}).encode(), json.dumps(proposals_data).encode(), ) with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke(cli, ["hub", "proposal", "list", "-v"]) assert result.exit_code == 0 output = result.stderr assert "🟢" in output assert "deadbeef" in output assert "feat/my-feature" in output assert "charlie" in output assert "2025-12-31" in output def test_e2e_empty_list_exits_zero_with_message(self, repo: pathlib.Path) -> None: """Empty proposal list must exit 0 and print a human-friendly message.""" self._setup(repo) resps = self._mock_api( json.dumps({"repo_id": "repo-id"}).encode(), json.dumps({"proposals": []}).encode(), ) with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke(cli, ["hub", "proposal", "list", "--state", "merged"]) assert result.exit_code == 0 assert "No proposals" in result.stderr or "no proposals" in result.stderr.lower() def test_e2e_json_no_stdout_in_text_mode(self, repo: pathlib.Path) -> None: """In text mode, JSON must NOT appear on stdout — all output goes to stderr.""" self._setup(repo) proposals_data = {"proposals": [ {"proposalId": "abc12345-0000-0000-0000-000000000001", "title": "T", "state": "open", "fromBranch": "feat/x", "toBranch": "dev"}, ]} resps = self._mock_api( json.dumps({"repo_id": "repo-id"}).encode(), json.dumps(proposals_data).encode(), ) with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke(cli, ["hub", "proposal", "list"]) assert result.exit_code == 0 # In text mode, stdout should have no JSON array for line in result.output.splitlines(): stripped = line.strip() assert not stripped.startswith("["), ( f"Unexpected JSON on stdout in text mode: {stripped!r}" ) class TestProposalViewHardening: """Additional hardening tests for `muse hub proposal show`.""" _HUB = "http://localhost:19999/gabriel/muse" def _setup(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["hub", "connect", self._HUB]) _store_identity(self._HUB) def _make_api_resp(self, payload_bytes: bytes) -> MagicMock: mock_resp = MagicMock() mock_resp.__enter__ = lambda s: s mock_resp.__exit__ = MagicMock(return_value=False) mock_resp.read.return_value = payload_bytes return mock_resp def _mock_api(self, *responses: bytes) -> list[MagicMock]: return [self._make_api_resp(r) for r in responses] def test_short_flag_j_works_for_view(self, repo: pathlib.Path) -> None: """``-j`` is accepted as alias for ``--json``.""" self._setup(repo) proposal_id = "abc12345-0000-0000-0000-000000000001" proposal_data = {"proposalId": proposal_id, "title": "T", "state": "open", "fromBranch": "feat/x", "toBranch": "dev"} proposals_data = {"proposals": [ {"proposalId": proposal_id, "title": "T", "state": "open", "fromBranch": "feat/x", "toBranch": "dev"}, ]} resps = self._mock_api( json.dumps({"repo_id": "repo-id"}).encode(), json.dumps(proposals_data).encode(), json.dumps(proposal_data).encode(), ) with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke(cli, ["hub", "proposal", "read", "abc12345", "-j"]) assert result.exit_code == 0 json_lines = [l for l in result.output.splitlines() if l.strip().startswith("{")] assert len(json_lines) >= 1 def test_ansi_in_state_sanitized( self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str] ) -> None: """ANSI in ``state`` field must not reach terminal in text mode.""" self._setup(repo) proposal_id = "abc12345-0000-0000-0000-000000000001" malicious_proposal = {"proposalId": proposal_id, "title": "T", "state": "\x1b[31mopen\x1b[0m", "fromBranch": "feat/x", "toBranch": "dev"} proposals_data = {"proposals": [ {"proposalId": proposal_id, "title": "T", "state": "open", "fromBranch": "feat/x", "toBranch": "dev"}, ]} resps = self._mock_api( json.dumps({"repo_id": "repo-id"}).encode(), json.dumps(proposals_data).encode(), json.dumps(malicious_proposal).encode(), ) with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke(cli, ["hub", "proposal", "read", "abc12345"]) assert result.exit_code == 0 assert "\x1b[" not in result.stderr def test_ansi_in_branch_sanitized(self, repo: pathlib.Path) -> None: """ANSI in branch names must not reach terminal in text mode.""" self._setup(repo) proposal_id = "abc12345-0000-0000-0000-000000000001" malicious_proposal = {"proposalId": proposal_id, "title": "T", "state": "open", "fromBranch": "\x1b[32mfeat/malicious\x1b[0m", "toBranch": "\x1b[34mdev\x1b[0m"} proposals_data = {"proposals": [ {"proposalId": proposal_id, "title": "T", "state": "open", "fromBranch": "feat/x", "toBranch": "dev"}, ]} resps = self._mock_api( json.dumps({"repo_id": "repo-id"}).encode(), json.dumps(proposals_data).encode(), json.dumps(malicious_proposal).encode(), ) with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke(cli, ["hub", "proposal", "read", "abc12345"]) assert result.exit_code == 0 assert "\x1b[" not in result.stderr def test_ansi_in_body_lines_sanitized(self, repo: pathlib.Path) -> None: """ANSI in body text must not reach terminal in text mode.""" self._setup(repo) proposal_id = "abc12345-0000-0000-0000-000000000001" malicious_proposal = {"proposalId": proposal_id, "title": "T", "state": "open", "fromBranch": "feat/x", "toBranch": "dev", "body": "\x1b[31mThis body has ANSI\x1b[0m"} proposals_data = {"proposals": [ {"proposalId": proposal_id, "title": "T", "state": "open", "fromBranch": "feat/x", "toBranch": "dev"}, ]} resps = self._mock_api( json.dumps({"repo_id": "repo-id"}).encode(), json.dumps(proposals_data).encode(), json.dumps(malicious_proposal).encode(), ) with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke(cli, ["hub", "proposal", "read", "abc12345"]) assert result.exit_code == 0 assert "\x1b[" not in result.stderr def test_view_prefix_not_found_exits_nonzero(self, repo: pathlib.Path) -> None: self._setup(repo) proposals_data = {"proposals": []} resps = self._mock_api( json.dumps({"repo_id": "repo-id"}).encode(), json.dumps(proposals_data).encode(), ) with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke(cli, ["hub", "proposal", "read", "deadbeef"]) assert result.exit_code != 0 def test_view_no_hub_exits_nonzero(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["hub", "proposal", "read", "abc12345"]) assert result.exit_code != 0 def test_view_no_identity_exits_nonzero(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["hub", "connect", self._HUB]) result = runner.invoke(cli, ["hub", "proposal", "read", "abc12345"]) assert result.exit_code != 0 def test_view_outside_repo_exits_nonzero( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: monkeypatch.chdir(tmp_path) monkeypatch.delenv("MUSE_REPO_ROOT", raising=False) result = runner.invoke(cli, ["hub", "proposal", "read", "abc12345"]) assert result.exit_code != 0 def test_full_id_skips_prefix_resolution(self, repo: pathlib.Path) -> None: """A full proposal ID must reach the view endpoint with exactly 2 API calls (no prefix fetch).""" self._setup(repo) proposal_id = "abc12345-def0-0000-0000-000000000001" proposal_data = {"proposalId": proposal_id, "title": "T", "state": "open", "fromBranch": "feat/x", "toBranch": "dev"} resps = self._mock_api( json.dumps({"repo_id": "repo-id"}).encode(), # _resolve_repo_id json.dumps(proposal_data).encode(), # GET proposals/{id} ) with patch("urllib.request.urlopen", side_effect=resps) as mock_open: result = runner.invoke(cli, ["hub", "proposal", "read", proposal_id, "-j"]) assert result.exit_code == 0 # Only 2 urlopen calls: repo resolution + the view fetch (no prefix list call) assert mock_open.call_count == 2 def test_prefix_triggers_resolution_call(self, repo: pathlib.Path) -> None: """An 8-char prefix must trigger a prefix-resolution list fetch (3 API calls total).""" self._setup(repo) proposal_id = "abc12345-0000-0000-0000-000000000001" proposal_data = {"proposalId": proposal_id, "title": "T", "state": "open", "fromBranch": "feat/x", "toBranch": "dev"} proposals_data = {"proposals": [ {"proposalId": proposal_id, "title": "T", "state": "open", "fromBranch": "feat/x", "toBranch": "dev"}, ]} resps = self._mock_api( json.dumps({"repo_id": "repo-id"}).encode(), # _resolve_repo_id json.dumps(proposals_data).encode(), # prefix resolution list json.dumps(proposal_data).encode(), # GET proposals/{id} ) with patch("urllib.request.urlopen", side_effect=resps) as mock_open: result = runner.invoke(cli, ["hub", "proposal", "read", "abc12345", "-j"]) assert result.exit_code == 0 assert mock_open.call_count == 3 def test_author_shown_in_text_mode(self, repo: pathlib.Path) -> None: self._setup(repo) proposal_id = "abc12345-0000-0000-0000-000000000001" proposal_data = {"proposalId": proposal_id, "title": "T", "state": "open", "fromBranch": "feat/x", "toBranch": "dev", "author": "charlie", "createdAt": "2024-07-04T00:00:00Z"} resps = self._mock_api( json.dumps({"repo_id": "repo-id"}).encode(), json.dumps({"proposals": [ {"proposalId": proposal_id, "title": "T", "state": "open", "fromBranch": "feat/x", "toBranch": "dev"}, ]}).encode(), json.dumps(proposal_data).encode(), ) with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke(cli, ["hub", "proposal", "read", "abc12345"]) assert result.exit_code == 0 assert "charlie" in result.stderr def test_created_at_shown_in_text_mode(self, repo: pathlib.Path) -> None: self._setup(repo) proposal_id = "abc12345-0000-0000-0000-000000000001" proposal_data = {"proposalId": proposal_id, "title": "T", "state": "open", "fromBranch": "feat/x", "toBranch": "dev", "author": "alice", "createdAt": "2025-03-15T08:30:00Z"} resps = self._mock_api( json.dumps({"repo_id": "repo-id"}).encode(), json.dumps({"proposals": [ {"proposalId": proposal_id, "title": "T", "state": "open", "fromBranch": "feat/x", "toBranch": "dev"}, ]}).encode(), json.dumps(proposal_data).encode(), ) with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke(cli, ["hub", "proposal", "read", "abc12345"]) assert result.exit_code == 0 assert "2025-03-15" in result.stderr def test_ansi_in_author_sanitized(self, repo: pathlib.Path) -> None: self._setup(repo) proposal_id = "abc12345-0000-0000-0000-000000000001" proposal_data = {"proposalId": proposal_id, "title": "T", "state": "open", "fromBranch": "feat/x", "toBranch": "dev", "author": "\x1b[31mmalicious-author\x1b[0m", "createdAt": "2024-01-01T00:00:00Z"} resps = self._mock_api( json.dumps({"repo_id": "repo-id"}).encode(), json.dumps({"proposals": [ {"proposalId": proposal_id, "title": "T", "state": "open", "fromBranch": "feat/x", "toBranch": "dev"}, ]}).encode(), json.dumps(proposal_data).encode(), ) with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke(cli, ["hub", "proposal", "read", "abc12345"]) assert result.exit_code == 0 assert "\x1b[" not in result.stderr def test_ansi_in_created_at_sanitized(self, repo: pathlib.Path) -> None: self._setup(repo) proposal_id = "abc12345-0000-0000-0000-000000000001" proposal_data = {"proposalId": proposal_id, "title": "T", "state": "open", "fromBranch": "feat/x", "toBranch": "dev", "author": "alice", "createdAt": "\x1b[32m2024-01-01\x1b[0mTmalicious"} resps = self._mock_api( json.dumps({"repo_id": "repo-id"}).encode(), json.dumps({"proposals": [ {"proposalId": proposal_id, "title": "T", "state": "open", "fromBranch": "feat/x", "toBranch": "dev"}, ]}).encode(), json.dumps(proposal_data).encode(), ) with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke(cli, ["hub", "proposal", "read", "abc12345"]) assert result.exit_code == 0 assert "\x1b[" not in result.stderr def test_body_truncation_hint_shown(self, repo: pathlib.Path) -> None: """Body exceeding _MAX_PROPOSAL_BODY_LINES must show a truncation hint.""" from muse.cli.commands.hub import _MAX_PROPOSAL_BODY_LINES self._setup(repo) proposal_id = "abc12345-0000-0000-0000-000000000001" long_body = "\n".join(f"line {i}" for i in range(_MAX_PROPOSAL_BODY_LINES + 5)) proposal_data = {"proposalId": proposal_id, "title": "T", "state": "open", "fromBranch": "feat/x", "toBranch": "dev", "body": long_body} resps = self._mock_api( json.dumps({"repo_id": "repo-id"}).encode(), json.dumps({"proposals": [ {"proposalId": proposal_id, "title": "T", "state": "open", "fromBranch": "feat/x", "toBranch": "dev"}, ]}).encode(), json.dumps(proposal_data).encode(), ) with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke(cli, ["hub", "proposal", "read", "abc12345"]) assert result.exit_code == 0 assert "more line" in result.stderr assert "--json" in result.stderr # hint mentions --json def test_body_exactly_at_limit_no_hint(self, repo: pathlib.Path) -> None: """Body at exactly _MAX_PROPOSAL_BODY_LINES must NOT show a truncation hint.""" from muse.cli.commands.hub import _MAX_PROPOSAL_BODY_LINES self._setup(repo) proposal_id = "abc12345-0000-0000-0000-000000000001" exact_body = "\n".join(f"line {i}" for i in range(_MAX_PROPOSAL_BODY_LINES)) proposal_data = {"proposalId": proposal_id, "title": "T", "state": "open", "fromBranch": "feat/x", "toBranch": "dev", "body": exact_body} resps = self._mock_api( json.dumps({"repo_id": "repo-id"}).encode(), json.dumps({"proposals": [ {"proposalId": proposal_id, "title": "T", "state": "open", "fromBranch": "feat/x", "toBranch": "dev"}, ]}).encode(), json.dumps(proposal_data).encode(), ) with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke(cli, ["hub", "proposal", "read", "abc12345"]) assert result.exit_code == 0 assert "more line" not in result.stderr def test_no_body_field_no_body_section(self, repo: pathlib.Path) -> None: """When body is absent or empty, no 'Body:' section must appear.""" self._setup(repo) proposal_id = "abc12345-0000-0000-0000-000000000001" proposal_data = {"proposalId": proposal_id, "title": "T", "state": "open", "fromBranch": "feat/x", "toBranch": "dev"} resps = self._mock_api( json.dumps({"repo_id": "repo-id"}).encode(), json.dumps({"proposals": [ {"proposalId": proposal_id, "title": "T", "state": "open", "fromBranch": "feat/x", "toBranch": "dev"}, ]}).encode(), json.dumps(proposal_data).encode(), ) with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke(cli, ["hub", "proposal", "read", "abc12345"]) assert result.exit_code == 0 assert "Body:" not in result.stderr def test_json_passthrough_includes_all_fields(self, repo: pathlib.Path) -> None: """JSON output must be an unmodified passthrough from the API.""" self._setup(repo) proposal_id = "abc12345-0000-0000-0000-000000000001" proposal_data = {"proposalId": proposal_id, "title": "My Proposal", "state": "open", "fromBranch": "feat/x", "toBranch": "dev", "author": "alice", "createdAt": "2024-01-01T00:00:00Z", "body": "Full body text here.", "extraField": "agent-visible"} resps = self._mock_api( json.dumps({"repo_id": "repo-id"}).encode(), json.dumps({"proposals": [ {"proposalId": proposal_id, "title": "My Proposal", "state": "open", "fromBranch": "feat/x", "toBranch": "dev"}, ]}).encode(), json.dumps(proposal_data).encode(), ) with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke(cli, ["hub", "proposal", "read", "abc12345", "-j"]) assert result.exit_code == 0 data = json.loads(next( l for l in result.output.splitlines() if l.strip().startswith("{") )) assert data["author"] == "alice" assert data["body"] == "Full body text here." assert data["extraField"] == "agent-visible" def test_hub_override_flag(self, repo: pathlib.Path) -> None: """``--hub`` must route requests to the override URL.""" runner.invoke(cli, ["hub", "connect", "http://localhost:11111/wrong/repo"]) _store_identity("http://localhost:19999/gabriel/muse") proposal_id = "abc12345-def0-0000-0000-000000000001" proposal_data = {"proposalId": proposal_id, "title": "T", "state": "open", "fromBranch": "f", "toBranch": "d"} resps = self._mock_api( json.dumps({"repo_id": "repo-id"}).encode(), json.dumps(proposal_data).encode(), ) with patch("urllib.request.urlopen", side_effect=resps) as mock_open: result = runner.invoke( cli, ["hub", "proposal", "read", proposal_id, "--hub", "http://localhost:19999/gabriel/muse", "-j"], ) assert result.exit_code == 0 called_urls = [c[0][0].full_url for c in mock_open.call_args_list] assert any("19999" in u for u in called_urls) assert not any("11111" in u for u in called_urls) class TestProposalViewUnit: """Pure unit tests for run_proposal_show text rendering logic.""" def _make_proposal_resp(self, **kwargs: str) -> bytes: base: Manifest = {"proposalId": "abc12345-0000-0000-0000-000000000001", "title": "My Proposal", "state": "open", "fromBranch": "feat/x", "toBranch": "dev"} base.update(kwargs) return json.dumps(base).encode() def _invoke_view( self, repo: pathlib.Path, proposal_data: bytes, *, flags: list[str] | None = None, ) -> InvokeResult: """Invoke hub proposal show with a pre-resolved full UUID (2 API calls only).""" proposal_id = "abc12345-def0-0000-0000-000000000001" # Use a full UUID to skip the prefix-resolution fetch mock_repo = MagicMock() mock_repo.__enter__ = lambda s: s mock_repo.__exit__ = MagicMock(return_value=False) mock_repo.read.return_value = json.dumps({"repo_id": "repo-id"}).encode() mock_proposal = MagicMock() mock_proposal.__enter__ = lambda s: s mock_proposal.__exit__ = MagicMock(return_value=False) mock_proposal.read.return_value = proposal_data cmd = ["hub", "proposal", "read", proposal_id] + (flags or []) with patch("urllib.request.urlopen", side_effect=[mock_repo, mock_proposal]): return runner.invoke(cli, cmd) def test_state_open_icon(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["hub", "connect", "http://localhost:19999/gabriel/muse"]) _store_identity("http://localhost:19999/gabriel/muse") result = self._invoke_view(repo, self._make_proposal_resp(state="open")) assert "🟢" in result.stderr def test_state_merged_icon(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["hub", "connect", "http://localhost:19999/gabriel/muse"]) _store_identity("http://localhost:19999/gabriel/muse") result = self._invoke_view(repo, self._make_proposal_resp(state="merged")) assert "🟣" in result.stderr def test_state_closed_icon(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["hub", "connect", "http://localhost:19999/gabriel/muse"]) _store_identity("http://localhost:19999/gabriel/muse") result = self._invoke_view(repo, self._make_proposal_resp(state="closed")) assert "⛔" in result.stderr def test_unknown_state_fallback_icon(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["hub", "connect", "http://localhost:19999/gabriel/muse"]) _store_identity("http://localhost:19999/gabriel/muse") result = self._invoke_view(repo, self._make_proposal_resp(state="draft")) assert "❓" in result.stderr def test_no_author_field_omits_by_line(self, repo: pathlib.Path) -> None: """When author is absent, the 'By:' line must not appear.""" runner.invoke(cli, ["hub", "connect", "http://localhost:19999/gabriel/muse"]) _store_identity("http://localhost:19999/gabriel/muse") result = self._invoke_view(repo, self._make_proposal_resp()) assert "By:" not in result.stderr def test_state_upper_in_header(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["hub", "connect", "http://localhost:19999/gabriel/muse"]) _store_identity("http://localhost:19999/gabriel/muse") result = self._invoke_view(repo, self._make_proposal_resp(state="open")) assert "[OPEN]" in result.stderr def test_id_and_branches_in_output(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["hub", "connect", "http://localhost:19999/gabriel/muse"]) _store_identity("http://localhost:19999/gabriel/muse") proposal_id = "abc12345-def0-0000-0000-000000000001" result = self._invoke_view( repo, self._make_proposal_resp(proposalId=proposal_id, fromBranch="feat/my", toBranch="main"), ) assert "feat/my" in result.stderr assert "main" in result.stderr class TestProposalViewE2E: """End-to-end scenario tests for `muse hub proposal show`.""" _HUB = "http://localhost:19999/gabriel/muse" def _setup(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["hub", "connect", self._HUB]) _store_identity(self._HUB) def _make_api_resp(self, payload_bytes: bytes) -> MagicMock: mock_resp = MagicMock() mock_resp.__enter__ = lambda s: s mock_resp.__exit__ = MagicMock(return_value=False) mock_resp.read.return_value = payload_bytes return mock_resp def _mock_api(self, *responses: bytes) -> list[MagicMock]: return [self._make_api_resp(r) for r in responses] def test_e2e_full_proposal_text_output(self, repo: pathlib.Path) -> None: """Full flow with all optional fields — all sections must appear.""" self._setup(repo) proposal_id = "deadbeef-cafe-0000-0000-000000000001" proposal_data = { "proposalId": proposal_id, "title": "feat: add sonic synthesis", "state": "open", "fromBranch": "feat/sonic", "toBranch": "dev", "author": "gabriel", "createdAt": "2025-06-01T12:00:00Z", "body": "This proposal adds sonic synthesis support.", } resps = self._mock_api( json.dumps({"repo_id": "repo-id"}).encode(), json.dumps(proposal_data).encode(), ) with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke(cli, ["hub", "proposal", "read", proposal_id]) assert result.exit_code == 0 output = result.stderr assert "🟢" in output assert "feat: add sonic synthesis" in output assert "feat/sonic" in output assert "gabriel" in output assert "2025-06-01" in output assert "sonic synthesis support" in output def test_e2e_json_agent_workflow(self, repo: pathlib.Path) -> None: """Simulate an agent extracting state via --json | jq.""" self._setup(repo) proposal_id = "deadbeef-cafe-0000-0000-000000000001" proposal_data = {"proposalId": proposal_id, "title": "T", "state": "merged", "fromBranch": "feat/x", "toBranch": "dev", "author": "bot", "mergeCommitId": "aabbccdd11223344"} resps = self._mock_api( json.dumps({"repo_id": "repo-id"}).encode(), json.dumps(proposal_data).encode(), ) with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke(cli, ["hub", "proposal", "read", proposal_id, "--json"]) assert result.exit_code == 0 data = json.loads(next( l for l in result.output.splitlines() if l.strip().startswith("{") )) assert data["state"] == "merged" assert data["mergeCommitId"] == "aabbccdd11223344" def test_e2e_body_truncation_hint_points_to_json(self, repo: pathlib.Path) -> None: """Truncation hint must explicitly mention --json.""" from muse.cli.commands.hub import _MAX_PROPOSAL_BODY_LINES self._setup(repo) proposal_id = "deadbeef-cafe-0000-0000-000000000001" long_body = "\n".join(f"line {i}" for i in range(_MAX_PROPOSAL_BODY_LINES + 10)) proposal_data = {"proposalId": proposal_id, "title": "T", "state": "open", "fromBranch": "feat/x", "toBranch": "dev", "body": long_body} resps = self._mock_api( json.dumps({"repo_id": "repo-id"}).encode(), json.dumps(proposal_data).encode(), ) with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke(cli, ["hub", "proposal", "read", proposal_id]) assert result.exit_code == 0 assert "--json" in result.stderr assert "10 more line" in result.stderr def test_e2e_ambiguous_prefix_exits_nonzero(self, repo: pathlib.Path) -> None: """Two proposals with the same prefix must cause a non-zero exit.""" self._setup(repo) proposals_data = {"proposals": [ {"proposalId": "abc12345-0000-0000-0000-000000000001", "title": "Proposal 1", "state": "open", "fromBranch": "feat/a", "toBranch": "dev"}, {"proposalId": "abc12345-0000-0000-0000-000000000002", "title": "Proposal 2", "state": "open", "fromBranch": "feat/b", "toBranch": "dev"}, ]} resps = self._mock_api( json.dumps({"repo_id": "repo-id"}).encode(), json.dumps(proposals_data).encode(), ) with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke(cli, ["hub", "proposal", "read", "abc12345"]) assert result.exit_code != 0 class TestProposalViewStress: """Stress tests for `muse hub proposal show`.""" _HUB = "http://localhost:19999/gabriel/muse" def test_body_with_1000_lines_truncated(self, repo: pathlib.Path) -> None: """A 1000-line body must be accepted without OOM and truncated correctly.""" from muse.cli.commands.hub import _MAX_PROPOSAL_BODY_LINES runner.invoke(cli, ["hub", "connect", self._HUB]) _store_identity(self._HUB) proposal_id = "deadbeef-cafe-0000-0000-000000000001" big_body = "\n".join(f"line {i}" for i in range(1000)) proposal_data = {"proposalId": proposal_id, "title": "T", "state": "open", "fromBranch": "feat/x", "toBranch": "dev", "body": big_body} mock_repo = MagicMock() mock_repo.__enter__ = lambda s: s mock_repo.__exit__ = MagicMock(return_value=False) mock_repo.read.return_value = json.dumps({"repo_id": "repo-id"}).encode() mock_proposal = MagicMock() mock_proposal.__enter__ = lambda s: s mock_proposal.__exit__ = MagicMock(return_value=False) mock_proposal.read.return_value = json.dumps(proposal_data).encode() with patch("urllib.request.urlopen", side_effect=[mock_repo, mock_proposal]): result = runner.invoke(cli, ["hub", "proposal", "read", proposal_id]) assert result.exit_code == 0 lines_shown = [l for l in result.stderr.splitlines() if l.strip().startswith("line ")] assert len(lines_shown) == _MAX_PROPOSAL_BODY_LINES assert "more line" in result.stderr def test_concurrent_format_operations(self) -> None: """_format_proposal called concurrently from 8 threads must not produce ANSI leakage.""" from muse.cli.commands.hub import _format_proposal errors: list[str] = [] def _do(idx: int) -> None: try: proposal = { "proposalId": f"dead{idx:04d}-0000-0000-0000-000000000001", "title": f"\x1b[31mProposal-{idx}\x1b[0m", "state": "open", "fromBranch": f"\x1b[32mfeat/f{idx}\x1b[0m", "toBranch": "dev", } result = _format_proposal(proposal) assert "\x1b[" not in result, f"Thread {idx}: ANSI leaked" except Exception as exc: errors.append(f"Thread {idx}: {exc}") threads = [threading.Thread(target=_do, args=(i,)) for i in range(8)] for t in threads: t.start() for t in threads: t.join() assert errors == [], "\n".join(errors) class TestProposalCreateHardening: """Additional hardening tests for `muse hub proposal create`.""" _HUB = "http://localhost:19999/gabriel/muse" def _setup(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["hub", "connect", self._HUB]) _store_identity(self._HUB) def _make_api_resp(self, payload_bytes: bytes) -> MagicMock: mock_resp = MagicMock() mock_resp.__enter__ = lambda s: s mock_resp.__exit__ = MagicMock(return_value=False) mock_resp.read.return_value = payload_bytes return mock_resp def _mock_api(self, *responses: bytes) -> list[MagicMock]: return [self._make_api_resp(r) for r in responses] def test_short_flag_j_works_for_create(self, repo: pathlib.Path) -> None: self._setup(repo) (heads_dir(repo) / "feat-x").write_text("") (head_path(repo)).write_text("ref: refs/heads/feat-x\n") create_resp = {"proposalId": "abc12345-0000-0000-0000-000000000001", "state": "open", "fromBranch": "feat-x", "toBranch": "dev"} resps = self._mock_api( json.dumps({"repo_id": "repo-id"}).encode(), json.dumps(create_resp).encode(), ) with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke( cli, ["hub", "proposal", "create", "--title", "T", "--from-branch", "feat-x", "-j"], ) assert result.exit_code == 0 json_lines = [l for l in result.output.splitlines() if l.strip().startswith("{")] assert len(json_lines) >= 1 def test_ansi_in_proposal_id_sanitized_text_output(self, repo: pathlib.Path) -> None: """ANSI in returned proposalId must not reach terminal in text mode.""" self._setup(repo) (heads_dir(repo) / "feat-x").write_text("") (head_path(repo)).write_text("ref: refs/heads/feat-x\n") create_resp = {"proposalId": "\x1b[31mabc12345-malicious\x1b[0m", "state": "open", "fromBranch": "feat-x", "toBranch": "dev"} resps = self._mock_api( json.dumps({"repo_id": "repo-id"}).encode(), json.dumps(create_resp).encode(), ) with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke( cli, ["hub", "proposal", "create", "--title", "T", "--from-branch", "feat-x"], ) assert "\x1b[" not in result.stderr def test_ansi_in_title_sanitized_text_output(self, repo: pathlib.Path) -> None: """ANSI in title arg must not reach terminal in text mode.""" self._setup(repo) (heads_dir(repo) / "feat-x").write_text("") (head_path(repo)).write_text("ref: refs/heads/feat-x\n") create_resp = {"proposalId": "abc12345-0000-0000-0000-000000000001", "state": "open", "fromBranch": "feat-x", "toBranch": "dev"} resps = self._mock_api( json.dumps({"repo_id": "repo-id"}).encode(), json.dumps(create_resp).encode(), ) with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke( cli, ["hub", "proposal", "create", "--title", "\x1b[31mmalicious title\x1b[0m", "--from-branch", "feat-x"], ) assert "\x1b[" not in result.stderr class TestProposalCreateSecurity: """Security-focused tests for `muse hub proposal create`.""" _HUB = "http://localhost:19999/gabriel/muse" def _setup(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["hub", "connect", self._HUB]) _store_identity(self._HUB) def _make_api_resp(self, payload_bytes: bytes) -> MagicMock: mock_resp = MagicMock() mock_resp.__enter__ = lambda s: s mock_resp.__exit__ = MagicMock(return_value=False) mock_resp.read.return_value = payload_bytes return mock_resp def _mock_api(self, *responses: bytes) -> list[MagicMock]: return [self._make_api_resp(r) for r in responses] def test_ansi_in_from_branch_sanitized(self, repo: pathlib.Path) -> None: self._setup(repo) create_resp = {"proposalId": "abc12345-0000-0000-0000-000000000001", "state": "open", "fromBranch": "feat-x", "toBranch": "dev"} resps = self._mock_api( json.dumps({"repo_id": "repo-id"}).encode(), json.dumps(create_resp).encode(), ) with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke( cli, ["hub", "proposal", "create", "--title", "T", "--from-branch", "\x1b[31mfeat/malicious\x1b[0m"], ) assert "\x1b[" not in result.stderr def test_ansi_in_to_branch_sanitized(self, repo: pathlib.Path) -> None: self._setup(repo) create_resp = {"proposalId": "abc12345-0000-0000-0000-000000000001", "state": "open", "fromBranch": "feat-x", "toBranch": "dev"} resps = self._mock_api( json.dumps({"repo_id": "repo-id"}).encode(), json.dumps(create_resp).encode(), ) with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke( cli, ["hub", "proposal", "create", "--title", "T", "--from-branch", "feat-x", "--to-branch", "\x1b[32mdev\x1b[0m"], ) assert "\x1b[" not in result.stderr def test_empty_title_exits_nonzero(self, repo: pathlib.Path) -> None: """Empty (whitespace-only) title must be rejected before any API call.""" self._setup(repo) with patch("urllib.request.urlopen") as mock_net: result = runner.invoke( cli, ["hub", "proposal", "create", "--title", " ", "--from-branch", "feat/x"], ) assert result.exit_code != 0 mock_net.assert_not_called() def test_title_too_long_exits_nonzero(self, repo: pathlib.Path) -> None: """Title exceeding _MAX_PROPOSAL_TITLE_LEN must be rejected before any API call.""" from muse.cli.commands.hub import _MAX_PROPOSAL_TITLE_LEN self._setup(repo) long_title = "x" * (_MAX_PROPOSAL_TITLE_LEN + 1) with patch("urllib.request.urlopen") as mock_net: result = runner.invoke( cli, ["hub", "proposal", "create", "--title", long_title, "--from-branch", "feat/x"], ) assert result.exit_code != 0 mock_net.assert_not_called() def test_title_at_max_length_accepted(self, repo: pathlib.Path) -> None: """Title exactly at _MAX_PROPOSAL_TITLE_LEN must be accepted.""" from muse.cli.commands.hub import _MAX_PROPOSAL_TITLE_LEN self._setup(repo) exact_title = "x" * _MAX_PROPOSAL_TITLE_LEN create_resp = {"proposalId": "abc12345-0000-0000-0000-000000000001", "state": "open", "fromBranch": "feat-x", "toBranch": "dev"} resps = self._mock_api( json.dumps({"repo_id": "repo-id"}).encode(), json.dumps(create_resp).encode(), ) with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke( cli, ["hub", "proposal", "create", "--title", exact_title, "--from-branch", "feat-x", "-j"], ) assert result.exit_code == 0 class TestProposalCreateBranchDetection: """Tests for auto-detection of the source branch.""" _HUB = "http://localhost:19999/gabriel/muse" def _setup(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["hub", "connect", self._HUB]) _store_identity(self._HUB) def _make_api_resp(self, payload_bytes: bytes) -> MagicMock: mock_resp = MagicMock() mock_resp.__enter__ = lambda s: s mock_resp.__exit__ = MagicMock(return_value=False) mock_resp.read.return_value = payload_bytes return mock_resp def _mock_api(self, *responses: bytes) -> list[MagicMock]: return [self._make_api_resp(r) for r in responses] def test_auto_detect_current_branch(self, repo: pathlib.Path) -> None: """Without --from-branch, the current branch must be used.""" self._setup(repo) (heads_dir(repo) / "feat-auto").write_text("") (head_path(repo)).write_text("ref: refs/heads/feat-auto\n") create_resp = {"proposalId": "abc12345-0000-0000-0000-000000000001", "state": "open", "fromBranch": "feat-auto", "toBranch": "dev"} resps = self._mock_api( json.dumps({"repo_id": "repo-id"}).encode(), json.dumps(create_resp).encode(), ) with patch("urllib.request.urlopen", side_effect=resps) as mock_open: result = runner.invoke(cli, ["hub", "proposal", "create", "--title", "T", "-j"]) assert result.exit_code == 0 # Verify the request body contains the auto-detected branch post_call = next(c for c in mock_open.call_args_list if c[0][0].method == "POST") payload = json.loads(post_call[0][0].data) assert payload["fromBranch"] == "feat-auto" def test_explicit_from_branch_overrides_head(self, repo: pathlib.Path) -> None: """Explicit --from-branch must override the HEAD branch.""" self._setup(repo) (heads_dir(repo) / "main").write_text("") (head_path(repo)).write_text("ref: refs/heads/main\n") create_resp = {"proposalId": "abc12345-0000-0000-0000-000000000001", "state": "open", "fromBranch": "feat/explicit", "toBranch": "dev"} resps = self._mock_api( json.dumps({"repo_id": "repo-id"}).encode(), json.dumps(create_resp).encode(), ) with patch("urllib.request.urlopen", side_effect=resps) as mock_open: result = runner.invoke( cli, ["hub", "proposal", "create", "--title", "T", "--from-branch", "feat/explicit", "-j"], ) assert result.exit_code == 0 post_call = next(c for c in mock_open.call_args_list if c[0][0].method == "POST") payload = json.loads(post_call[0][0].data) assert payload["fromBranch"] == "feat/explicit" def test_head_alias_for_from_branch(self, repo: pathlib.Path) -> None: """``--head`` must be accepted as an alias for ``--from-branch``.""" self._setup(repo) create_resp = {"proposalId": "abc12345-0000-0000-0000-000000000001", "state": "open", "fromBranch": "feat/head-alias", "toBranch": "dev"} resps = self._mock_api( json.dumps({"repo_id": "repo-id"}).encode(), json.dumps(create_resp).encode(), ) with patch("urllib.request.urlopen", side_effect=resps) as mock_open: result = runner.invoke( cli, ["hub", "proposal", "create", "--title", "T", "--head", "feat/head-alias", "-j"], ) assert result.exit_code == 0 post_call = next(c for c in mock_open.call_args_list if c[0][0].method == "POST") payload = json.loads(post_call[0][0].data) assert payload["fromBranch"] == "feat/head-alias" def test_base_alias_for_to_branch(self, repo: pathlib.Path) -> None: """``--base`` must be accepted as an alias for ``--to-branch``.""" self._setup(repo) create_resp = {"proposalId": "abc12345-0000-0000-0000-000000000001", "state": "open", "fromBranch": "feat/x", "toBranch": "main"} resps = self._mock_api( json.dumps({"repo_id": "repo-id"}).encode(), json.dumps(create_resp).encode(), ) with patch("urllib.request.urlopen", side_effect=resps) as mock_open: result = runner.invoke( cli, ["hub", "proposal", "create", "--title", "T", "--from-branch", "feat/x", "--base", "main", "-j"], ) assert result.exit_code == 0 post_call = next(c for c in mock_open.call_args_list if c[0][0].method == "POST") payload = json.loads(post_call[0][0].data) assert payload["toBranch"] == "main" def test_to_branch_default_is_dev(self, repo: pathlib.Path) -> None: """When --to-branch is omitted, the request body must contain 'dev'.""" self._setup(repo) create_resp = {"proposalId": "abc12345-0000-0000-0000-000000000001", "state": "open", "fromBranch": "feat/x", "toBranch": "dev"} resps = self._mock_api( json.dumps({"repo_id": "repo-id"}).encode(), json.dumps(create_resp).encode(), ) with patch("urllib.request.urlopen", side_effect=resps) as mock_open: result = runner.invoke( cli, ["hub", "proposal", "create", "--title", "T", "--from-branch", "feat/x", "-j"], ) assert result.exit_code == 0 post_call = next(c for c in mock_open.call_args_list if c[0][0].method == "POST") payload = json.loads(post_call[0][0].data) assert payload["toBranch"] == "dev" def test_detached_head_exits_nonzero_with_message(self, repo: pathlib.Path) -> None: """Detached HEAD without --from-branch must exit nonzero with a helpful message. Branch detection runs before any network I/O, so no urlopen calls are made. """ self._setup(repo) # Write a bare commit SHA as HEAD (detached state) (head_path(repo)).write_text("abc1234567890abcdef1234567890abcdef123456\n") with patch("urllib.request.urlopen") as mock_net: result = runner.invoke(cli, ["hub", "proposal", "create", "--title", "T"]) assert result.exit_code != 0 # Message must mention how to fix it assert "--from-branch" in result.stderr or "detached" in result.stderr.lower() # No network calls — branch detection is pre-network mock_net.assert_not_called() def test_detached_head_with_explicit_from_branch_succeeds( self, repo: pathlib.Path ) -> None: """Detached HEAD is fine when --from-branch is given explicitly.""" self._setup(repo) (head_path(repo)).write_text("abc1234567890abcdef1234567890abcdef123456\n") create_resp = {"proposalId": "abc12345-0000-0000-0000-000000000001", "state": "open", "fromBranch": "feat/x", "toBranch": "dev"} resps = [ MagicMock(**{ "__enter__": lambda s: s, "__exit__": MagicMock(return_value=False), "read": MagicMock(return_value=json.dumps({"repo_id": "r"}).encode()), }), MagicMock(**{ "__enter__": lambda s: s, "__exit__": MagicMock(return_value=False), "read": MagicMock(return_value=json.dumps(create_resp).encode()), }), ] with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke( cli, ["hub", "proposal", "create", "--title", "T", "--from-branch", "feat/x", "-j"], ) assert result.exit_code == 0 class TestProposalCreateTextOutput: """Tests for the human-readable text output of `muse hub proposal create`.""" _HUB = "http://localhost:19999/gabriel/muse" def _setup(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["hub", "connect", self._HUB]) _store_identity(self._HUB) def _make_api_resp(self, payload_bytes: bytes) -> MagicMock: mock_resp = MagicMock() mock_resp.__enter__ = lambda s: s mock_resp.__exit__ = MagicMock(return_value=False) mock_resp.read.return_value = payload_bytes return mock_resp def _mock_api(self, *responses: bytes) -> list[MagicMock]: return [self._make_api_resp(r) for r in responses] def test_success_shows_proposal_id_prefix(self, repo: pathlib.Path) -> None: self._setup(repo) proposal_id = "deadbeef-cafe-0000-0000-000000000001" create_resp = {"proposalId": proposal_id, "state": "open", "fromBranch": "feat/x", "toBranch": "dev"} resps = self._mock_api( json.dumps({"repo_id": "repo-id"}).encode(), json.dumps(create_resp).encode(), ) with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke( cli, ["hub", "proposal", "create", "--title", "My Proposal", "--from-branch", "feat/x"], ) assert result.exit_code == 0 assert "deadbeef" in result.stderr def test_success_shows_branch_arrow(self, repo: pathlib.Path) -> None: self._setup(repo) create_resp = {"proposalId": "abc12345-0000-0000-0000-000000000001", "state": "open", "fromBranch": "feat/x", "toBranch": "dev"} resps = self._mock_api( json.dumps({"repo_id": "repo-id"}).encode(), json.dumps(create_resp).encode(), ) with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke( cli, ["hub", "proposal", "create", "--title", "T", "--from-branch", "feat/x", "--to-branch", "dev"], ) assert result.exit_code == 0 assert "feat/x" in result.stderr assert "dev" in result.stderr assert "→" in result.stderr def test_url_line_shown_when_owner_slug_present(self, repo: pathlib.Path) -> None: """The URL line must appear when hub URL contains owner/slug.""" self._setup(repo) proposal_id = "abc12345-0000-0000-0000-000000000001" create_resp = {"proposalId": proposal_id, "state": "open", "fromBranch": "feat/x", "toBranch": "dev"} resps = self._mock_api( json.dumps({"repo_id": "repo-id"}).encode(), json.dumps(create_resp).encode(), ) with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke( cli, ["hub", "proposal", "create", "--title", "T", "--from-branch", "feat/x"], ) assert result.exit_code == 0 assert "Proposal created:" in result.stderr assert "proposals" in result.stderr def test_body_sent_in_payload(self, repo: pathlib.Path) -> None: """The body argument must be included in the POST payload.""" self._setup(repo) create_resp = {"proposalId": "abc12345-0000-0000-0000-000000000001", "state": "open", "fromBranch": "feat/x", "toBranch": "dev"} resps = self._mock_api( json.dumps({"repo_id": "repo-id"}).encode(), json.dumps(create_resp).encode(), ) with patch("urllib.request.urlopen", side_effect=resps) as mock_open: result = runner.invoke( cli, ["hub", "proposal", "create", "--title", "T", "--from-branch", "feat/x", "--body", "My description", "-j"], ) assert result.exit_code == 0 post_call = next(c for c in mock_open.call_args_list if c[0][0].method == "POST") payload = json.loads(post_call[0][0].data) assert payload["body"] == "My description" def test_json_output_is_api_passthrough(self, repo: pathlib.Path) -> None: """JSON output must be the unmodified API response.""" self._setup(repo) create_resp = {"proposalId": "abc12345-0000-0000-0000-000000000001", "state": "open", "fromBranch": "feat/x", "toBranch": "dev", "author": "alice", "extraField": "preserved"} resps = self._mock_api( json.dumps({"repo_id": "repo-id"}).encode(), json.dumps(create_resp).encode(), ) with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke( cli, ["hub", "proposal", "create", "--title", "T", "--from-branch", "feat/x", "-j"], ) assert result.exit_code == 0 data = json.loads(next( l for l in result.output.splitlines() if l.strip().startswith("{") )) assert data["extraField"] == "preserved" assert data["author"] == "alice" def test_no_hub_exits_nonzero(self, repo: pathlib.Path) -> None: result = runner.invoke( cli, ["hub", "proposal", "create", "--title", "T", "--from-branch", "feat/x"] ) assert result.exit_code != 0 def test_no_identity_exits_nonzero(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["hub", "connect", self._HUB]) result = runner.invoke( cli, ["hub", "proposal", "create", "--title", "T", "--from-branch", "feat/x"] ) assert result.exit_code != 0 def test_outside_repo_exits_nonzero( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: monkeypatch.chdir(tmp_path) monkeypatch.delenv("MUSE_REPO_ROOT", raising=False) result = runner.invoke( cli, ["hub", "proposal", "create", "--title", "T", "--from-branch", "feat/x"] ) assert result.exit_code != 0 class TestProposalCreateE2E: """End-to-end scenario tests for `muse hub proposal create`.""" _HUB = "http://localhost:19999/gabriel/muse" def _setup(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["hub", "connect", self._HUB]) _store_identity(self._HUB) def _make_api_resp(self, payload_bytes: bytes) -> MagicMock: mock_resp = MagicMock() mock_resp.__enter__ = lambda s: s mock_resp.__exit__ = MagicMock(return_value=False) mock_resp.read.return_value = payload_bytes return mock_resp def _mock_api(self, *responses: bytes) -> list[MagicMock]: return [self._make_api_resp(r) for r in responses] def test_e2e_full_agent_workflow(self, repo: pathlib.Path) -> None: """Simulate the canonical agent proposal creation flow.""" self._setup(repo) (heads_dir(repo) / "feat-sonic").write_text("") (head_path(repo)).write_text("ref: refs/heads/feat-sonic\n") create_resp = { "proposalId": "deadbeef-cafe-0000-0000-000000000001", "state": "open", "fromBranch": "feat-sonic", "toBranch": "dev", "title": "feat: sonic synthesis", } resps = self._mock_api( json.dumps({"repo_id": "repo-id"}).encode(), json.dumps(create_resp).encode(), ) with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke( cli, ["hub", "proposal", "create", "--title", "feat: sonic synthesis", "--body", "Adds FM synthesis support.", "--json"], ) assert result.exit_code == 0 data = json.loads(next( l for l in result.output.splitlines() if l.strip().startswith("{") )) assert data["proposalId"] == "deadbeef-cafe-0000-0000-000000000001" assert data["state"] == "open" def test_e2e_proposal_id_extractable_from_json(self, repo: pathlib.Path) -> None: """Agent must be able to extract proposalId from JSON output for chaining.""" self._setup(repo) proposal_id = "cafebabe-0000-0000-0000-000000000001" create_resp = {"proposalId": proposal_id, "state": "open", "fromBranch": "feat/x", "toBranch": "dev"} resps = self._mock_api( json.dumps({"repo_id": "repo-id"}).encode(), json.dumps(create_resp).encode(), ) with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke( cli, ["hub", "proposal", "create", "--title", "T", "--from-branch", "feat/x", "-j"], ) assert result.exit_code == 0 data = json.loads(next( l for l in result.output.splitlines() if l.strip().startswith("{") )) assert data["proposalId"] == proposal_id def test_e2e_text_output_has_no_json_on_stdout(self, repo: pathlib.Path) -> None: """In text mode, JSON must not appear on stdout.""" self._setup(repo) create_resp = {"proposalId": "abc12345-0000-0000-0000-000000000001", "state": "open", "fromBranch": "feat/x", "toBranch": "dev"} resps = self._mock_api( json.dumps({"repo_id": "repo-id"}).encode(), json.dumps(create_resp).encode(), ) with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke( cli, ["hub", "proposal", "create", "--title", "T", "--from-branch", "feat/x"], ) assert result.exit_code == 0 for line in result.output.splitlines(): assert not line.strip().startswith("{"), ( f"Unexpected JSON on stdout: {line!r}" ) class TestProposalCreateStress: """Stress tests for `muse hub proposal create`.""" _HUB = "http://localhost:19999/gabriel/muse" def test_title_at_exact_max_not_rejected(self) -> None: """_MAX_PROPOSAL_TITLE_LEN boundary: title of exactly that length must not be rejected.""" from muse.cli.commands.hub import _MAX_PROPOSAL_TITLE_LEN title = "x" * _MAX_PROPOSAL_TITLE_LEN assert len(title) == _MAX_PROPOSAL_TITLE_LEN def test_title_one_over_max_rejected(self) -> None: """One character over _MAX_PROPOSAL_TITLE_LEN must be caught before network.""" from muse.cli.commands.hub import _MAX_PROPOSAL_TITLE_LEN # Pure logic test: verify the constant is what we expect and the # check triggers by examining run_pr_create's validation directly. title = "x" * (_MAX_PROPOSAL_TITLE_LEN + 1) assert len(title) > _MAX_PROPOSAL_TITLE_LEN # sanity def test_concurrent_title_validation(self) -> None: """Title length validation is pure Python — safe from all 8 threads.""" from muse.cli.commands.hub import _MAX_PROPOSAL_TITLE_LEN errors: list[str] = [] def _do(idx: int) -> None: try: long_title = "x" * (_MAX_PROPOSAL_TITLE_LEN + idx + 1) assert len(long_title) > _MAX_PROPOSAL_TITLE_LEN except Exception as exc: errors.append(f"Thread {idx}: {exc}") threads = [threading.Thread(target=_do, args=(i,)) for i in range(8)] for t in threads: t.start() for t in threads: t.join() assert errors == [], "\n".join(errors) class TestProposalMergeHardening: """Additional hardening tests for `muse hub proposal merge`.""" _HUB = "http://localhost:19999/gabriel/muse" def _setup(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["hub", "connect", self._HUB]) _store_identity(self._HUB) def _make_api_resp(self, payload_bytes: bytes) -> MagicMock: mock_resp = MagicMock() mock_resp.__enter__ = lambda s: s mock_resp.__exit__ = MagicMock(return_value=False) mock_resp.read.return_value = payload_bytes return mock_resp def _mock_api(self, *responses: bytes) -> list[MagicMock]: return [self._make_api_resp(r) for r in responses] def test_short_flag_j_works_for_merge(self, repo: pathlib.Path) -> None: self._setup(repo) proposal_id = "abc12345-0000-0000-0000-000000000001" proposals_data = {"proposals": [ {"proposalId": proposal_id, "title": "T", "state": "open", "fromBranch": "feat/x", "toBranch": "dev"}, ]} merge_resp = {"merged": True, "mergeCommitId": "deadbeef01234567"} resps = self._mock_api( json.dumps({"repo_id": "repo-id"}).encode(), json.dumps(proposals_data).encode(), json.dumps(merge_resp).encode(), ) with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke(cli, ["hub", "proposal", "merge", "abc12345", "-j"]) assert result.exit_code == 0 json_lines = [l for l in result.output.splitlines() if l.strip().startswith("{")] assert len(json_lines) >= 1 def test_ansi_in_commit_sha_sanitized_text_mode(self, repo: pathlib.Path) -> None: """ANSI in returned mergeCommitId must not reach terminal in text mode.""" self._setup(repo) proposal_id = "abc12345-0000-0000-0000-000000000001" proposals_data = {"proposals": [ {"proposalId": proposal_id, "title": "T", "state": "open", "fromBranch": "feat/x", "toBranch": "dev"}, ]} merge_resp = {"merged": True, "mergeCommitId": "\x1b[31mdeadbeef01234567\x1b[0m"} resps = self._mock_api( json.dumps({"repo_id": "repo-id"}).encode(), json.dumps(proposals_data).encode(), json.dumps(merge_resp).encode(), ) with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke(cli, ["hub", "proposal", "merge", "abc12345"]) assert result.exit_code == 0 assert "\x1b[" not in result.stderr def test_merge_squash_strategy_accepted(self, repo: pathlib.Path) -> None: self._setup(repo) proposal_id = "abc12345-0000-0000-0000-000000000001" proposals_data = {"proposals": [ {"proposalId": proposal_id, "title": "T", "state": "open", "fromBranch": "feat/x", "toBranch": "dev"}, ]} merge_resp = {"merged": True, "mergeCommitId": "aabbccdd11223344"} resps = self._mock_api( json.dumps({"repo_id": "repo-id"}).encode(), json.dumps(proposals_data).encode(), json.dumps(merge_resp).encode(), ) with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke( cli, ["hub", "proposal", "merge", "abc12345", "--strategy", "squash"] ) assert result.exit_code == 0 def test_merge_rebase_strategy_accepted(self, repo: pathlib.Path) -> None: self._setup(repo) proposal_id = "abc12345-0000-0000-0000-000000000001" proposals_data = {"proposals": [ {"proposalId": proposal_id, "title": "T", "state": "open", "fromBranch": "feat/x", "toBranch": "dev"}, ]} merge_resp = {"merged": True, "mergeCommitId": "1a2b3c4d5e6f7890"} resps = self._mock_api( json.dumps({"repo_id": "repo-id"}).encode(), json.dumps(proposals_data).encode(), json.dumps(merge_resp).encode(), ) with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke( cli, ["hub", "proposal", "merge", "abc12345", "--strategy", "rebase"] ) assert result.exit_code == 0 def test_merge_prefix_not_found_exits_nonzero(self, repo: pathlib.Path) -> None: self._setup(repo) proposals_data = {"proposals": []} resps = self._mock_api( json.dumps({"repo_id": "repo-id"}).encode(), json.dumps(proposals_data).encode(), ) with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke(cli, ["hub", "proposal", "merge", "deadbeef"]) assert result.exit_code != 0 class TestProposalMergePayload: """Verify the POST payload sent by `muse hub proposal merge`.""" _HUB = "http://localhost:19999/gabriel/muse" def _setup(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["hub", "connect", self._HUB]) _store_identity(self._HUB) def _make_api_resp(self, payload_bytes: bytes) -> MagicMock: mock_resp = MagicMock() mock_resp.__enter__ = lambda s: s mock_resp.__exit__ = MagicMock(return_value=False) mock_resp.read.return_value = payload_bytes return mock_resp def _proposal_id(self) -> str: return "abc12345-0000-0000-0000-000000000001" def _proposals_resp(self) -> bytes: return json.dumps({"proposals": [ {"proposalId": self._proposal_id(), "title": "T", "state": "open", "fromBranch": "feat/x", "toBranch": "dev"}, ]}).encode() def _merge_resp(self, merged: bool = True) -> bytes: return json.dumps({"merged": merged, "mergeCommitId": "deadbeef01234567"}).encode() def test_default_strategy_is_merge_commit(self, repo: pathlib.Path) -> None: self._setup(repo) resps = [ self._make_api_resp(json.dumps({"repo_id": "r"}).encode()), self._make_api_resp(self._proposals_resp()), self._make_api_resp(self._merge_resp()), ] with patch("urllib.request.urlopen", side_effect=resps) as mock_open: result = runner.invoke(cli, ["hub", "proposal", "merge", "abc12345", "-j"]) assert result.exit_code == 0 post_call = next(c for c in mock_open.call_args_list if c[0][0].method == "POST") payload = json.loads(post_call[0][0].data) assert payload["mergeStrategy"] == "merge_commit" def test_squash_strategy_in_payload(self, repo: pathlib.Path) -> None: self._setup(repo) resps = [ self._make_api_resp(json.dumps({"repo_id": "r"}).encode()), self._make_api_resp(self._proposals_resp()), self._make_api_resp(self._merge_resp()), ] with patch("urllib.request.urlopen", side_effect=resps) as mock_open: result = runner.invoke( cli, ["hub", "proposal", "merge", "abc12345", "--strategy", "squash", "-j"] ) assert result.exit_code == 0 post_call = next(c for c in mock_open.call_args_list if c[0][0].method == "POST") payload = json.loads(post_call[0][0].data) assert payload["mergeStrategy"] == "squash" def test_rebase_strategy_in_payload(self, repo: pathlib.Path) -> None: self._setup(repo) resps = [ self._make_api_resp(json.dumps({"repo_id": "r"}).encode()), self._make_api_resp(self._proposals_resp()), self._make_api_resp(self._merge_resp()), ] with patch("urllib.request.urlopen", side_effect=resps) as mock_open: result = runner.invoke( cli, ["hub", "proposal", "merge", "abc12345", "--strategy", "rebase", "-j"] ) assert result.exit_code == 0 post_call = next(c for c in mock_open.call_args_list if c[0][0].method == "POST") payload = json.loads(post_call[0][0].data) assert payload["mergeStrategy"] == "rebase" def test_delete_branch_true_by_default(self, repo: pathlib.Path) -> None: self._setup(repo) resps = [ self._make_api_resp(json.dumps({"repo_id": "r"}).encode()), self._make_api_resp(self._proposals_resp()), self._make_api_resp(self._merge_resp()), ] with patch("urllib.request.urlopen", side_effect=resps) as mock_open: result = runner.invoke(cli, ["hub", "proposal", "merge", "abc12345", "-j"]) assert result.exit_code == 0 post_call = next(c for c in mock_open.call_args_list if c[0][0].method == "POST") payload = json.loads(post_call[0][0].data) assert payload["deleteBranch"] is True def test_no_delete_branch_flag_sets_false_in_payload(self, repo: pathlib.Path) -> None: self._setup(repo) resps = [ self._make_api_resp(json.dumps({"repo_id": "r"}).encode()), self._make_api_resp(self._proposals_resp()), self._make_api_resp(self._merge_resp()), ] with patch("urllib.request.urlopen", side_effect=resps) as mock_open: result = runner.invoke( cli, ["hub", "proposal", "merge", "abc12345", "--no-delete-branch", "-j"] ) assert result.exit_code == 0 post_call = next(c for c in mock_open.call_args_list if c[0][0].method == "POST") payload = json.loads(post_call[0][0].data) assert payload["deleteBranch"] is False def test_merge_endpoint_url_contains_proposal_id(self, repo: pathlib.Path) -> None: """The POST must go to .../proposals/{full_proposal_id}/merge.""" self._setup(repo) resps = [ self._make_api_resp(json.dumps({"repo_id": "r"}).encode()), self._make_api_resp(self._proposals_resp()), self._make_api_resp(self._merge_resp()), ] with patch("urllib.request.urlopen", side_effect=resps) as mock_open: runner.invoke(cli, ["hub", "proposal", "merge", "abc12345", "-j"]) post_call = next(c for c in mock_open.call_args_list if c[0][0].method == "POST") assert self._proposal_id() in post_call[0][0].full_url assert "/merge" in post_call[0][0].full_url class TestProposalMergeExitCodes: """Verify exit codes for all merge outcomes.""" _HUB = "http://localhost:19999/gabriel/muse" def _setup(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["hub", "connect", self._HUB]) _store_identity(self._HUB) def _make_api_resp(self, payload_bytes: bytes) -> MagicMock: mock_resp = MagicMock() mock_resp.__enter__ = lambda s: s mock_resp.__exit__ = MagicMock(return_value=False) mock_resp.read.return_value = payload_bytes return mock_resp def _proposals_resp(self, proposal_id: str) -> bytes: return json.dumps({"proposals": [ {"proposalId": proposal_id, "title": "T", "state": "open", "fromBranch": "feat/x", "toBranch": "dev"}, ]}).encode() def test_merged_true_exits_zero(self, repo: pathlib.Path) -> None: self._setup(repo) proposal_id = "abc12345-0000-0000-0000-000000000001" resps = [ self._make_api_resp(json.dumps({"repo_id": "r"}).encode()), self._make_api_resp(self._proposals_resp(proposal_id)), self._make_api_resp(json.dumps({"merged": True, "mergeCommitId": "abc"}).encode()), ] with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke(cli, ["hub", "proposal", "merge", "abc12345"]) assert result.exit_code == 0 def test_merged_false_text_mode_exits_3(self, repo: pathlib.Path) -> None: self._setup(repo) proposal_id = "abc12345-0000-0000-0000-000000000001" resps = [ self._make_api_resp(json.dumps({"repo_id": "r"}).encode()), self._make_api_resp(self._proposals_resp(proposal_id)), self._make_api_resp(json.dumps({"merged": False, "message": "conflict"}).encode()), ] with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke(cli, ["hub", "proposal", "merge", "abc12345"]) assert result.exit_code == 3 def test_merged_false_json_mode_exits_3(self, repo: pathlib.Path) -> None: """merge=false with --json must exit 3, not 0. This is the key agent-safety guarantee: agents using --json can rely on the exit code to detect merge failures. """ self._setup(repo) proposal_id = "abc12345-0000-0000-0000-000000000001" resps = [ self._make_api_resp(json.dumps({"repo_id": "r"}).encode()), self._make_api_resp(self._proposals_resp(proposal_id)), self._make_api_resp(json.dumps({"merged": False, "message": "branch protection"}).encode()), ] with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke(cli, ["hub", "proposal", "merge", "abc12345", "--json"]) assert result.exit_code == 3 def test_merged_false_json_mode_still_prints_json(self, repo: pathlib.Path) -> None: """Even on failure, the full API response must be printed before exiting 3.""" self._setup(repo) proposal_id = "abc12345-0000-0000-0000-000000000001" resps = [ self._make_api_resp(json.dumps({"repo_id": "r"}).encode()), self._make_api_resp(self._proposals_resp(proposal_id)), self._make_api_resp( json.dumps({"merged": False, "message": "conflict detected"}).encode() ), ] with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke(cli, ["hub", "proposal", "merge", "abc12345", "--json"]) assert result.exit_code == 3 # JSON must still be printed so agent can read the failure reason data = json.loads(next( l for l in result.output.splitlines() if l.strip().startswith("{") )) assert data["merged"] is False assert data["message"] == "conflict detected" def test_no_hub_exits_nonzero(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["hub", "proposal", "merge", "abc12345"]) assert result.exit_code != 0 def test_no_identity_exits_nonzero(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["hub", "connect", self._HUB]) result = runner.invoke(cli, ["hub", "proposal", "merge", "abc12345"]) assert result.exit_code != 0 def test_outside_repo_exits_nonzero( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: monkeypatch.chdir(tmp_path) monkeypatch.delenv("MUSE_REPO_ROOT", raising=False) result = runner.invoke(cli, ["hub", "proposal", "merge", "abc12345"]) assert result.exit_code != 0 def test_ambiguous_prefix_exits_nonzero(self, repo: pathlib.Path) -> None: self._setup(repo) proposals_data = {"proposals": [ {"proposalId": "abc12345-0000-0000-0000-000000000001", "title": "Proposal 1", "state": "open", "fromBranch": "feat/a", "toBranch": "dev"}, {"proposalId": "abc12345-0000-0000-0000-000000000002", "title": "Proposal 2", "state": "open", "fromBranch": "feat/b", "toBranch": "dev"}, ]} resps = [ self._make_api_resp(json.dumps({"repo_id": "r"}).encode()), self._make_api_resp(json.dumps(proposals_data).encode()), ] with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke(cli, ["hub", "proposal", "merge", "abc12345"]) assert result.exit_code != 0 class TestProposalMergeTextOutput: """Tests for the human-readable text output of `muse hub proposal merge`.""" _HUB = "http://localhost:19999/gabriel/muse" def _setup(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["hub", "connect", self._HUB]) _store_identity(self._HUB) def _make_api_resp(self, payload_bytes: bytes) -> MagicMock: mock_resp = MagicMock() mock_resp.__enter__ = lambda s: s mock_resp.__exit__ = MagicMock(return_value=False) mock_resp.read.return_value = payload_bytes return mock_resp def _proposals_resp(self, proposal_id: str) -> bytes: return json.dumps({"proposals": [ {"proposalId": proposal_id, "title": "T", "state": "open", "fromBranch": "feat/x", "toBranch": "dev"}, ]}).encode() def test_success_shows_proposal_id_prefix(self, repo: pathlib.Path) -> None: self._setup(repo) # Use a full UUID so prefix-resolution is skipped (2 API calls only) proposal_id = "deadbeef-cafe-0000-0000-000000000001" resps = [ self._make_api_resp(json.dumps({"repo_id": "r"}).encode()), self._make_api_resp(json.dumps({"merged": True, "mergeCommitId": "aabb1122"}).encode()), ] with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke(cli, ["hub", "proposal", "merge", proposal_id]) assert result.exit_code == 0 assert "deadbeef" in result.stderr def test_success_shows_commit_sha(self, repo: pathlib.Path) -> None: self._setup(repo) proposal_id = "abc12345-0000-0000-0000-000000000001" resps = [ self._make_api_resp(json.dumps({"repo_id": "r"}).encode()), self._make_api_resp(self._proposals_resp(proposal_id)), self._make_api_resp( json.dumps({"merged": True, "mergeCommitId": "cafebabe12345678"}).encode() ), ] with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke(cli, ["hub", "proposal", "merge", "abc12345"]) assert result.exit_code == 0 assert "cafebabe" in result.stderr def test_success_no_sha_shows_placeholder(self, repo: pathlib.Path) -> None: """When mergeCommitId is absent, a placeholder must appear.""" self._setup(repo) proposal_id = "abc12345-0000-0000-0000-000000000001" resps = [ self._make_api_resp(json.dumps({"repo_id": "r"}).encode()), self._make_api_resp(self._proposals_resp(proposal_id)), self._make_api_resp(json.dumps({"merged": True}).encode()), ] with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke(cli, ["hub", "proposal", "merge", "abc12345"]) assert result.exit_code == 0 assert "no SHA" in result.stderr def test_delete_branch_message_shown_when_true(self, repo: pathlib.Path) -> None: self._setup(repo) proposal_id = "abc12345-0000-0000-0000-000000000001" resps = [ self._make_api_resp(json.dumps({"repo_id": "r"}).encode()), self._make_api_resp(self._proposals_resp(proposal_id)), self._make_api_resp(json.dumps({"merged": True, "mergeCommitId": "abc"}).encode()), ] with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke(cli, ["hub", "proposal", "merge", "abc12345"]) assert result.exit_code == 0 assert "Source branch deleted" in result.stderr def test_delete_branch_message_absent_with_no_delete_branch( self, repo: pathlib.Path ) -> None: self._setup(repo) proposal_id = "abc12345-0000-0000-0000-000000000001" resps = [ self._make_api_resp(json.dumps({"repo_id": "r"}).encode()), self._make_api_resp(self._proposals_resp(proposal_id)), self._make_api_resp(json.dumps({"merged": True, "mergeCommitId": "abc"}).encode()), ] with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke( cli, ["hub", "proposal", "merge", "abc12345", "--no-delete-branch"] ) assert result.exit_code == 0 assert "Source branch deleted" not in result.stderr def test_failure_message_shown(self, repo: pathlib.Path) -> None: self._setup(repo) proposal_id = "abc12345-0000-0000-0000-000000000001" resps = [ self._make_api_resp(json.dumps({"repo_id": "r"}).encode()), self._make_api_resp(self._proposals_resp(proposal_id)), self._make_api_resp( json.dumps({"merged": False, "message": "branch protection rule"}).encode() ), ] with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke(cli, ["hub", "proposal", "merge", "abc12345"]) assert result.exit_code != 0 assert "branch protection rule" in result.stderr def test_ansi_in_failure_message_sanitized(self, repo: pathlib.Path) -> None: self._setup(repo) proposal_id = "abc12345-0000-0000-0000-000000000001" resps = [ self._make_api_resp(json.dumps({"repo_id": "r"}).encode()), self._make_api_resp(self._proposals_resp(proposal_id)), self._make_api_resp( json.dumps({"merged": False, "message": "\x1b[31mmalicious message\x1b[0m"}).encode() ), ] with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke(cli, ["hub", "proposal", "merge", "abc12345"]) assert result.exit_code != 0 assert "\x1b[" not in result.stderr class TestProposalMergeFullUUID: """Verify that a full UUID skips the prefix-resolution list fetch.""" _HUB = "http://localhost:19999/gabriel/muse" def _setup(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["hub", "connect", self._HUB]) _store_identity(self._HUB) def _make_api_resp(self, payload_bytes: bytes) -> MagicMock: mock_resp = MagicMock() mock_resp.__enter__ = lambda s: s mock_resp.__exit__ = MagicMock(return_value=False) mock_resp.read.return_value = payload_bytes return mock_resp def test_full_id_uses_2_api_calls(self, repo: pathlib.Path) -> None: """Full proposal ID: repo resolution + merge POST = 2 calls, no prefix list fetch.""" self._setup(repo) proposal_id = "deadbeef-cafe-babe-0000-000000000001" resps = [ self._make_api_resp(json.dumps({"repo_id": "r"}).encode()), self._make_api_resp(json.dumps({"merged": True, "mergeCommitId": "abc"}).encode()), ] with patch("urllib.request.urlopen", side_effect=resps) as mock_open: result = runner.invoke(cli, ["hub", "proposal", "merge", proposal_id, "-j"]) assert result.exit_code == 0 assert mock_open.call_count == 2 def test_prefix_uses_3_api_calls(self, repo: pathlib.Path) -> None: """8-char prefix: repo + prefix list + merge POST = 3 calls.""" self._setup(repo) proposal_id = "abc12345-0000-0000-0000-000000000001" proposals_data = {"proposals": [ {"proposalId": proposal_id, "title": "T", "state": "open", "fromBranch": "feat/x", "toBranch": "dev"}, ]} resps = [ self._make_api_resp(json.dumps({"repo_id": "r"}).encode()), self._make_api_resp(json.dumps(proposals_data).encode()), self._make_api_resp(json.dumps({"merged": True, "mergeCommitId": "abc"}).encode()), ] with patch("urllib.request.urlopen", side_effect=resps) as mock_open: result = runner.invoke(cli, ["hub", "proposal", "merge", "abc12345", "-j"]) assert result.exit_code == 0 assert mock_open.call_count == 3 def test_hub_override_routes_to_correct_host(self, repo: pathlib.Path) -> None: """--hub must route all calls to the override URL, not the config URL.""" runner.invoke(cli, ["hub", "connect", "http://localhost:11111/wrong/repo"]) _store_identity("http://localhost:19999/gabriel/muse") proposal_id = "deadbeef-cafe-babe-0000-000000000001" resps = [ self._make_api_resp(json.dumps({"repo_id": "r"}).encode()), self._make_api_resp(json.dumps({"merged": True, "mergeCommitId": "abc"}).encode()), ] with patch("urllib.request.urlopen", side_effect=resps) as mock_open: result = runner.invoke( cli, ["hub", "proposal", "merge", proposal_id, "--hub", "http://localhost:19999/gabriel/muse", "-j"], ) assert result.exit_code == 0 called_urls = [c[0][0].full_url for c in mock_open.call_args_list] assert any("19999" in u for u in called_urls) assert not any("11111" in u for u in called_urls) class TestProposalMergeE2E: """End-to-end scenario tests for `muse hub proposal merge`.""" _HUB = "http://localhost:19999/gabriel/muse" def _setup(self, repo: pathlib.Path) -> None: runner.invoke(cli, ["hub", "connect", self._HUB]) _store_identity(self._HUB) def _make_api_resp(self, payload_bytes: bytes) -> MagicMock: mock_resp = MagicMock() mock_resp.__enter__ = lambda s: s mock_resp.__exit__ = MagicMock(return_value=False) mock_resp.read.return_value = payload_bytes return mock_resp def test_e2e_agent_safe_pipeline(self, repo: pathlib.Path) -> None: """Agent pipeline: --json exits 0 on success so && chains correctly.""" self._setup(repo) proposal_id = "deadbeef-cafe-0000-0000-000000000001" resps = [ self._make_api_resp(json.dumps({"repo_id": "r"}).encode()), self._make_api_resp(json.dumps({"merged": True, "mergeCommitId": "cafebabe12345678"}).encode()), ] with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke(cli, ["hub", "proposal", "merge", proposal_id, "--json"]) assert result.exit_code == 0 data = json.loads(next( l for l in result.output.splitlines() if l.strip().startswith("{") )) assert data["merged"] is True assert data["mergeCommitId"] == "cafebabe12345678" def test_e2e_agent_conflict_pipeline(self, repo: pathlib.Path) -> None: """Agent pipeline: --json exits 3 on conflict so || error-handling fires.""" self._setup(repo) proposal_id = "deadbeef-cafe-0000-0000-000000000001" resps = [ self._make_api_resp(json.dumps({"repo_id": "r"}).encode()), self._make_api_resp( json.dumps({"merged": False, "message": "merge conflict"}).encode() ), ] with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke(cli, ["hub", "proposal", "merge", proposal_id, "--json"]) assert result.exit_code == 3 # JSON is still printed so agent can read the error data = json.loads(next( l for l in result.output.splitlines() if l.strip().startswith("{") )) assert data["merged"] is False def test_e2e_squash_no_delete_branch(self, repo: pathlib.Path) -> None: """Squash merge keeping the branch: payload and output both correct.""" self._setup(repo) proposal_id = "abc12345-def0-0000-0000-000000000001" resps = [ self._make_api_resp(json.dumps({"repo_id": "r"}).encode()), self._make_api_resp( json.dumps({"merged": True, "mergeCommitId": "aabbccdd11223344"}).encode() ), ] with patch("urllib.request.urlopen", side_effect=resps) as mock_open: result = runner.invoke( cli, ["hub", "proposal", "merge", proposal_id, "--strategy", "squash", "--no-delete-branch"], ) assert result.exit_code == 0 assert "Source branch deleted" not in result.stderr assert "aabbccdd" in result.stderr post = next(c for c in mock_open.call_args_list if c[0][0].method == "POST") payload = json.loads(post[0][0].data) assert payload["mergeStrategy"] == "squash" assert payload["deleteBranch"] is False def test_e2e_text_output_no_json_on_stdout(self, repo: pathlib.Path) -> None: """In text mode, JSON must not appear on stdout.""" self._setup(repo) proposal_id = "deadbeef-cafe-0000-0000-000000000001" resps = [ self._make_api_resp(json.dumps({"repo_id": "r"}).encode()), self._make_api_resp( json.dumps({"merged": True, "mergeCommitId": "abc"}).encode() ), ] with patch("urllib.request.urlopen", side_effect=resps): result = runner.invoke(cli, ["hub", "proposal", "merge", proposal_id]) assert result.exit_code == 0 for line in result.output.splitlines(): assert not line.strip().startswith("{"), ( f"Unexpected JSON on stdout: {line!r}" ) class TestProposalMergeStress: """Stress tests for `muse hub proposal merge`.""" _HUB = "http://localhost:19999/gabriel/muse" def test_concurrent_exit_code_checks(self) -> None: """8 threads checking the merged=False exit-code logic must agree.""" from muse.core.errors import ExitCode errors: list[str] = [] def _do(idx: int) -> None: try: # Simulate the merged check in pure Python data = {"merged": False, "message": f"conflict {idx}"} merged = bool(data.get("merged", False)) expected_exit = ExitCode.INTERNAL_ERROR if not merged else ExitCode.SUCCESS assert expected_exit == ExitCode.INTERNAL_ERROR except Exception as exc: errors.append(f"Thread {idx}: {exc}") threads = [threading.Thread(target=_do, args=(i,)) for i in range(8)] for t in threads: t.start() for t in threads: t.join() assert errors == [], "\n".join(errors) class TestResolveProposalIdLimit: """Verify that _resolve_proposal_id respects _PROPOSAL_PREFIX_RESOLVE_LIMIT.""" def test_limit_constant_in_url(self) -> None: """The URL sent to the API must include the limit constant.""" from muse.cli.commands.hub import _PROPOSAL_PREFIX_RESOLVE_LIMIT, _resolve_proposal_id from muse.core.identity import IdentityEntry identity: IdentityEntry = {"type": "human", "token": "tok"} proposal_id = "abc12345-0000-0000-0000-000000000001" proposals_resp = {"proposals": [ {"proposalId": proposal_id, "title": "T"}, ]} captured_urls: list[str] = [] def _fake_urlopen(req: urllib.request.Request, timeout: int = 5, context: ssl.SSLContext | None = None) -> MagicMock: captured_urls.append(req.full_url) mock_resp = MagicMock() mock_resp.__enter__ = lambda s: s mock_resp.__exit__ = MagicMock(return_value=False) mock_resp.read.return_value = json.dumps(proposals_resp).encode() return mock_resp with patch("muse.cli.config.get_signing_identity", return_value=_make_signing()): with patch("urllib.request.urlopen", side_effect=_fake_urlopen): result = _resolve_proposal_id("http://localhost:9999", identity, "repo-id", "abc12345") assert result == proposal_id assert any(str(_PROPOSAL_PREFIX_RESOLVE_LIMIT) in url for url in captured_urls), ( f"Expected {_PROPOSAL_PREFIX_RESOLVE_LIMIT} in one of {captured_urls}" ) class TestResolveProposalIdSha256Passthrough: """sha256-prefixed full IDs must be returned as-is without hitting the list endpoint. Regression: the old full-ID check required a hyphen (`-`), so sha256: IDs always fell through to the list fetch with limit=200. Servers that cap the limit lower than 200 returned 422, making every `hub proposal read sha256:...` call fail on those hubs. """ def _make_identity(self) -> "muse.core.identity.IdentityEntry": from muse.core.identity import IdentityEntry e: IdentityEntry = {"type": "human", "token": "tok123"} return e def test_full_sha256_id_returned_as_is_no_network(self) -> None: """A full sha256:<64-hex> ID must be returned without any network call.""" from muse.cli.commands.hub import _resolve_proposal_id full = "sha256:" + "a" * 64 captured: list[str] = [] def _fail_urlopen(*a: str, **kw: str) -> None: captured.append("called") raise AssertionError("urlopen must not be called for a full sha256 ID") with patch("urllib.request.urlopen", side_effect=_fail_urlopen): result = _resolve_proposal_id("http://hub", self._make_identity(), "repo-id", full) assert result == full assert captured == [], "urlopen was called — full sha256 ID was not detected as complete" def test_sha256_prefix_still_resolves_via_list(self) -> None: """A short sha256 prefix (fewer than 71 chars) still fetches the list.""" from muse.cli.commands.hub import _resolve_proposal_id full = "sha256:" + "b" * 64 proposals_resp = {"proposals": [{"proposalId": full, "title": "T", "state": "open", "fromBranch": "feat/x", "toBranch": "dev"}]} mock_resp = MagicMock() mock_resp.__enter__ = lambda s: s mock_resp.__exit__ = MagicMock(return_value=False) mock_resp.read.return_value = json.dumps(proposals_resp).encode() with patch("muse.cli.config.get_signing_identity", return_value=_make_signing()): with patch("urllib.request.urlopen", return_value=mock_resp): result = _resolve_proposal_id( "http://localhost:9999", self._make_identity(), "repo-id", "sha256:bbbb" ) assert result == full def test_hyphenated_uuid_still_returned_as_is(self) -> None: """Regression: existing UUID-style full IDs must not be broken.""" from muse.cli.commands.hub import _resolve_proposal_id full = "af54753d-1234-5678-abcd-ef1234567890" with patch("urllib.request.urlopen", side_effect=AssertionError("must not call network")): result = _resolve_proposal_id("http://hub", self._make_identity(), "repo-id", full) assert result == full class TestProposalMerge422Regression: """Regression for issue #54: hub proposal merge with a full sha256 ID must not call the proposals list endpoint. Root cause: the old full-ID check in _resolve_proposal_id required a hyphen, so sha256: IDs fell through to the list fetch (?limit=200). Servers that capped limit at 100 returned 422 on that call, blocking every CLI merge regardless of strategy. Fix: _resolve_proposal_id now calls split_id() first; sha256-prefixed IDs are returned as-is without any network round-trip, so the 422 can never occur on that path. """ def test_merge_sha256_id_makes_no_list_call(self) -> None: """run_proposal_merge with a full sha256 proposal ID must POST to /merge and never touch the proposals list endpoint.""" import argparse from muse.cli.commands.hub.proposals import run_proposal_merge proposal_id = "sha256:" + "c" * 64 list_urls: list[str] = [] merge_urls: list[str] = [] def _fake_urlopen(req: urllib.request.Request, timeout: int = 5, context: ssl.SSLContext | None = None) -> MagicMock: url = req.full_url if "proposals?" in url: list_urls.append(url) if "/merge" in url and req.method == "POST": merge_urls.append(url) mock_resp = MagicMock() mock_resp.__enter__ = lambda s: s mock_resp.__exit__ = MagicMock(return_value=False) mock_resp.read.return_value = json.dumps({ "merged": True, "mergeCommitId": "sha256:" + "d" * 64, }).encode() return mock_resp args = argparse.Namespace( proposal_id=proposal_id, strategy="squash", delete_branch=False, json_output=False, hub="http://localhost:9999/owner/repo", ) with ( patch("muse.cli.commands.hub.proposals._get_hub_and_identity", return_value=("http://localhost:9999/owner/repo", {"type": "human", "token": "tok"})), patch("muse.cli.commands.hub.proposals._resolve_repo_id", return_value="test-repo-id"), patch("muse.cli.config.get_signing_identity", return_value=_make_signing()), patch("urllib.request.urlopen", side_effect=_fake_urlopen), ): run_proposal_merge(args) assert not list_urls, ( "run_proposal_merge must not call the proposals list endpoint " f"when given a full sha256 ID (triggers 422 on servers with " f"limit cap). Called: {list_urls}" ) assert merge_urls, ( "run_proposal_merge must POST to the /merge endpoint" ) def test_merge_prefix_id_calls_list_but_not_with_limit_exceeding_server_cap( self, ) -> None: """Short prefix IDs still resolve via the list endpoint, but the limit used must not exceed the server's PaginationParams cap (200).""" import argparse from muse.cli.commands.hub import _PROPOSAL_PREFIX_RESOLVE_LIMIT assert _PROPOSAL_PREFIX_RESOLVE_LIMIT <= 200, ( f"_PROPOSAL_PREFIX_RESOLVE_LIMIT is {_PROPOSAL_PREFIX_RESOLVE_LIMIT}, " "which exceeds the server's PaginationParams cap of 200. " "Lower the constant or raise the server cap to fix issue #54." ) # ============================================================================= # muse hub issue — hardening tests # ============================================================================= # Shared helpers for issue tests HUB_URL = "https://localhost:1337/owner/repo" def _issue_resp( number: int = 7, title: str = "feat: add thing", body: str = "", labels: list[str] | None = None, issue_id: str = "iss_aabbccdd", state: str = "open", author: str = "alice", ) -> _JsonPayload: return { "number": number, "title": title, "body": body, "labels": labels or [], "issueId": issue_id, "state": state, "author": author, "createdAt": "2026-04-09T00:00:00Z", } def _issue_list_resp(issues: list[_JsonPayload] | None = None) -> _JsonPayload: """Wrap issues in the list-response envelope.""" items = issues if issues is not None else [_issue_resp()] return {"issues": items, "total": len(items)} def _comment_resp(comment_id: str = "c0") -> _JsonPayload: """A single-comment response as returned by POST .../comments.""" return { "commentId": comment_id, "issueId": "issue-id-0001", "author": "alice", "body": "test comment", "parentId": None, "isDeleted": False, "createdAt": "2026-04-14T00:00:00Z", "updatedAt": "2026-04-14T00:00:00Z", } def _refs_resp(repo_id: str = "repo-id-0001") -> _JsonPayload: return {"repo_id": repo_id, "branches": []} def _mock_responses(*payloads: _JsonPayload) -> list[MagicMock]: """Build a side_effect list of mock HTTP responses for urlopen.""" mocks = [] for payload in payloads: m = MagicMock() m.__enter__ = lambda s: s m.__exit__ = MagicMock(return_value=False) m.read.return_value = json.dumps(payload).encode() mocks.append(m) return mocks # --------------------------------------------------------------------------- # TestIssueCreateHardening # --------------------------------------------------------------------------- class TestIssueCreateHardening: """Integration tests for ``muse hub issue create``.""" def test_empty_title_exits_nonzero_no_network( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) with patch("urllib.request.urlopen") as mock_net: result = runner.invoke(cli, ["hub", "issue", "create", "--title", " "]) assert result.exit_code != 0 mock_net.assert_not_called() def test_empty_title_error_message( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) with patch("urllib.request.urlopen"): result = runner.invoke(cli, ["hub", "issue", "create", "--title", ""]) assert "empty" in result.stderr.lower() or "title" in result.stderr.lower() def test_title_too_long_exits_nonzero_no_network( self, repo: pathlib.Path ) -> None: from muse.cli.commands.hub import _MAX_ISSUE_TITLE_LEN from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) long_title = "x" * (_MAX_ISSUE_TITLE_LEN + 1) with patch("urllib.request.urlopen") as mock_net: result = runner.invoke(cli, ["hub", "issue", "create", "--title", long_title]) assert result.exit_code != 0 mock_net.assert_not_called() def test_title_too_long_shows_char_count( self, repo: pathlib.Path ) -> None: from muse.cli.commands.hub import _MAX_ISSUE_TITLE_LEN from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) long_title = "x" * (_MAX_ISSUE_TITLE_LEN + 1) with patch("urllib.request.urlopen"): result = runner.invoke(cli, ["hub", "issue", "create", "--title", long_title]) assert str(_MAX_ISSUE_TITLE_LEN + 1) in result.stderr or str(_MAX_ISSUE_TITLE_LEN) in result.stderr def test_title_at_max_length_accepted( self, repo: pathlib.Path ) -> None: from muse.cli.commands.hub import _MAX_ISSUE_TITLE_LEN from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) exact_title = "x" * _MAX_ISSUE_TITLE_LEN mocks = _mock_responses(_refs_resp(), _issue_resp(title=exact_title)) with patch("urllib.request.urlopen", side_effect=mocks): result = runner.invoke( cli, ["hub", "issue", "create", "--title", exact_title, "--json"] ) assert result.exit_code == 0 def test_success_json_output(self, repo: pathlib.Path) -> None: from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) mocks = _mock_responses(_refs_resp(), _issue_resp()) with patch("urllib.request.urlopen", side_effect=mocks): result = runner.invoke( cli, ["hub", "issue", "create", "--title", "feat: X", "-j"] ) assert result.exit_code == 0 data = json.loads(result.output) assert "number" in data def test_json_short_flag(self, repo: pathlib.Path) -> None: """-j short alias must work the same as --json.""" from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) mocks = _mock_responses(_refs_resp(), _issue_resp()) with patch("urllib.request.urlopen", side_effect=mocks): result = runner.invoke( cli, ["hub", "issue", "create", "--title", "feat: X", "-j"] ) assert result.exit_code == 0 json.loads(result.output) # must be valid JSON def test_labels_included_in_payload(self, repo: pathlib.Path) -> None: from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) captured: list[bytes] = [] def _fake(req: urllib.request.Request, timeout: int = 5, context: ssl.SSLContext | None = None) -> MagicMock: if req.method == "POST": captured.append(req.data or b"") m = MagicMock() m.__enter__ = lambda s: s m.__exit__ = MagicMock(return_value=False) if req.method == "GET": m.read.return_value = json.dumps(_refs_resp()).encode() else: m.read.return_value = json.dumps(_issue_resp()).encode() return m with patch("urllib.request.urlopen", side_effect=_fake): runner.invoke( cli, ["hub", "issue", "create", "--title", "T", "--label", "bug", "--label", "phase/1"], ) assert captured body = json.loads(captured[0]) assert "bug" in body["labels"] assert "phase/1" in body["labels"] def test_issue_url_on_stdout(self, repo: pathlib.Path) -> None: from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) mocks = _mock_responses(_refs_resp(), _issue_resp(number=42)) with patch("urllib.request.urlopen", side_effect=mocks): result = runner.invoke(cli, ["hub", "issue", "create", "--title", "T"]) assert result.exit_code == 0 assert "42" in result.stderr def test_issue_url_contains_owner_slug(self, repo: pathlib.Path) -> None: from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) mocks = _mock_responses(_refs_resp(), _issue_resp(number=3)) with patch("urllib.request.urlopen", side_effect=mocks): result = runner.invoke(cli, ["hub", "issue", "create", "--title", "T"]) assert "owner" in result.output assert "repo" in result.output def test_text_mode_success_on_stderr(self, repo: pathlib.Path) -> None: """Text mode prints ✅ Issue #N created. to stderr.""" from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) mocks = _mock_responses(_refs_resp(), _issue_resp(number=5)) with patch("urllib.request.urlopen", side_effect=mocks): result = runner.invoke(cli, ["hub", "issue", "create", "--title", "T"]) assert result.exit_code == 0 assert "5" in result.stderr assert "created" in result.stderr.lower() def test_text_mode_no_json_on_stdout(self, repo: pathlib.Path) -> None: from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) mocks = _mock_responses(_refs_resp(), _issue_resp()) with patch("urllib.request.urlopen", side_effect=mocks): result = runner.invoke(cli, ["hub", "issue", "create", "--title", "T"]) assert result.exit_code == 0 # Text mode must not emit a JSON object try: json.loads(result.output) assert False, "Text mode must not emit JSON" except (json.JSONDecodeError, ValueError): pass def test_number_fallback_for_nonnumeric_api_response( self, repo: pathlib.Path ) -> None: """If API returns a non-numeric 'number', fall back to 0 without crashing.""" from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) bad_issue = dict(_issue_resp()) bad_issue["number"] = "not-a-number" mocks = _mock_responses(_refs_resp(), bad_issue) with patch("urllib.request.urlopen", side_effect=mocks): result = runner.invoke(cli, ["hub", "issue", "create", "--title", "T"]) assert result.exit_code == 0 # must not crash def test_number_float_coerced(self, repo: pathlib.Path) -> None: """Numeric float from API (e.g. 7.0) must be coerced to int.""" from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) float_issue = dict(_issue_resp()) float_issue["number"] = 7.0 mocks = _mock_responses(_refs_resp(), float_issue) with patch("urllib.request.urlopen", side_effect=mocks): result = runner.invoke(cli, ["hub", "issue", "create", "--title", "T"]) assert result.exit_code == 0 assert "7" in result.stderr def test_no_hub_exits_nonzero(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["hub", "issue", "create", "--title", "T"]) assert result.exit_code != 0 def test_no_identity_exits_nonzero(self, repo: pathlib.Path) -> None: from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) result = runner.invoke(cli, ["hub", "issue", "create", "--title", "T"]) assert result.exit_code != 0 def test_outside_repo_exits_nonzero( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: monkeypatch.chdir(tmp_path) monkeypatch.delenv("MUSE_REPO_ROOT", raising=False) result = runner.invoke(cli, ["hub", "issue", "create", "--title", "T"]) assert result.exit_code != 0 def test_hub_override_used_in_request(self, repo: pathlib.Path) -> None: """--hub overrides the config hub URL.""" override_url = "http://override:9999/owner2/repo2" _store_identity(override_url) captured_urls: list[str] = [] def _fake(req: urllib.request.Request, timeout: int = 5, context: ssl.SSLContext | None = None) -> MagicMock: captured_urls.append(req.full_url) m = MagicMock() m.__enter__ = lambda s: s m.__exit__ = MagicMock(return_value=False) if "refs" in req.full_url: m.read.return_value = json.dumps(_refs_resp()).encode() else: m.read.return_value = json.dumps(_issue_resp()).encode() return m with patch("urllib.request.urlopen", side_effect=_fake): result = runner.invoke(cli, [ "hub", "issue", "create", "--hub", override_url, "--title", "T", ]) assert result.exit_code == 0 assert any("override:9999" in u for u in captured_urls) # --------------------------------------------------------------------------- # TestIssueCreateSecurity # --------------------------------------------------------------------------- class TestIssueCreateSecurity: """Security-focused tests for ``muse hub issue create``.""" def test_ansi_in_title_no_network_when_valid( self, repo: pathlib.Path ) -> None: """ANSI in title is not a validation error — title may contain them.""" from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) ansi_title = "feat: \x1b[31mred\x1b[0m bug" mocks = _mock_responses(_refs_resp(), _issue_resp(title=ansi_title)) with patch("urllib.request.urlopen", side_effect=mocks): result = runner.invoke(cli, ["hub", "issue", "create", "--title", ansi_title]) assert result.exit_code == 0 def test_issueId_fallback_sanitized(self, repo: pathlib.Path) -> None: """If hub URL has no owner/slug, issueId fallback must be sanitized.""" # Give the hub URL no slug path so the fallback branch triggers. bare_hub = "https://localhost:1337" _store_identity(bare_hub) ansi_id = "iss_\x1b[31minjection\x1b[0m" issue = dict(_issue_resp()) issue["issueId"] = ansi_id mocks = _mock_responses(_refs_resp(), issue) with patch("urllib.request.urlopen", side_effect=mocks): result = runner.invoke(cli, [ "hub", "issue", "create", "--hub", bare_hub, "--title", "T", ]) # ANSI escape sequences must not appear raw in output assert "\x1b[" not in result.stderr def test_title_validation_before_network( self, repo: pathlib.Path ) -> None: """Empty title must be rejected before any HTTP call is made.""" from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) with patch("urllib.request.urlopen") as mock_net: runner.invoke(cli, ["hub", "issue", "create", "--title", ""]) mock_net.assert_not_called() def test_max_title_len_constant_value(self) -> None: from muse.cli.commands.hub import _MAX_ISSUE_TITLE_LEN assert _MAX_ISSUE_TITLE_LEN == 512 def test_ansi_in_hub_url_path_not_echoed_raw( self, repo: pathlib.Path ) -> None: """ANSI in --hub URL path segments (owner/slug) must not reach stdout raw.""" # Craft a hub URL where the owner segment contains an ANSI escape. # urllib.parse will preserve it in the path — it must be stripped on output. ansi_owner = "\x1b[31mmalicious\x1b[0m" malicious_hub = f"https://localhost:1337/{ansi_owner}/repo" _store_identity(malicious_hub) mocks = _mock_responses(_refs_resp(), _issue_resp(number=1)) with patch("urllib.request.urlopen", side_effect=mocks): result = runner.invoke(cli, [ "hub", "issue", "create", "--hub", malicious_hub, "--title", "T", ]) assert "\x1b[" not in result.stderr def test_payload_type_annotation_no_bool(self) -> None: """The payload dict must not include bool values — type annotation check.""" import inspect import muse.cli.commands.hub as hub_mod src = inspect.getsource(hub_mod.run_issue_create) # The old annotation included 'bool' — verify it was removed. # Look for the payload assignment line. assert "str | bool | list" not in src def test_repo_flag_routes_to_correct_hub( self, repo: pathlib.Path ) -> None: """--repo owner/repo constructs a hub URL using the configured base.""" from muse.cli.config import set_hub_url # Configure hub base (without owner/repo path) base_hub = "https://localhost:1337/original/original" set_hub_url(base_hub, repo) _store_identity("https://localhost:1337/myowner/myrepo") captured_urls: list[str] = [] def _fake(req: urllib.request.Request, timeout: int = 5, context: ssl.SSLContext | None = None) -> MagicMock: captured_urls.append(req.full_url) m = MagicMock() m.__enter__ = lambda s: s m.__exit__ = MagicMock(return_value=False) if req.method == "GET": m.read.return_value = json.dumps(_refs_resp()).encode() else: m.read.return_value = json.dumps(_issue_resp()).encode() return m with patch("urllib.request.urlopen", side_effect=_fake): result = runner.invoke(cli, [ "hub", "issue", "create", "--repo", "myowner/myrepo", "--title", "T", ]) assert result.exit_code == 0 assert any("myowner" in u and "myrepo" in u for u in captured_urls) # --------------------------------------------------------------------------- # TestIssueEditHardening # --------------------------------------------------------------------------- class TestIssueEditHardening: """Integration tests for ``muse hub issue edit``.""" def test_no_fields_exits_nonzero_no_network( self, repo: pathlib.Path ) -> None: from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) with patch("urllib.request.urlopen") as mock_net: result = runner.invoke(cli, ["hub", "issue", "update", "42"]) assert result.exit_code != 0 mock_net.assert_not_called() def test_no_fields_error_message(self, repo: pathlib.Path) -> None: from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) with patch("urllib.request.urlopen"): result = runner.invoke(cli, ["hub", "issue", "update", "42"]) assert "nothing" in result.stderr.lower() or "update" in result.stderr.lower() def test_title_only_patch(self, repo: pathlib.Path) -> None: from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) captured: list[bytes] = [] def _fake(req: urllib.request.Request, timeout: int = 5, context: ssl.SSLContext | None = None) -> MagicMock: if req.method == "PATCH": captured.append(req.data or b"") m = MagicMock() m.__enter__ = lambda s: s m.__exit__ = MagicMock(return_value=False) if req.method == "GET": m.read.return_value = json.dumps(_refs_resp()).encode() else: m.read.return_value = json.dumps(_issue_resp()).encode() return m with patch("urllib.request.urlopen", side_effect=_fake): result = runner.invoke(cli, ["hub", "issue", "update", "7", "--title", "new title"]) assert result.exit_code == 0 assert captured body = json.loads(captured[0]) assert body == {"title": "new title"} def test_body_only_patch(self, repo: pathlib.Path) -> None: from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) captured: list[bytes] = [] def _fake(req: urllib.request.Request, timeout: int = 5, context: ssl.SSLContext | None = None) -> MagicMock: if req.method == "PATCH": captured.append(req.data or b"") m = MagicMock() m.__enter__ = lambda s: s m.__exit__ = MagicMock(return_value=False) if req.method == "GET": m.read.return_value = json.dumps(_refs_resp()).encode() else: m.read.return_value = json.dumps(_issue_resp()).encode() return m with patch("urllib.request.urlopen", side_effect=_fake): result = runner.invoke(cli, ["hub", "issue", "update", "7", "--body", "new body"]) assert result.exit_code == 0 assert captured body = json.loads(captured[0]) assert body == {"body": "new body"} def test_both_title_and_body_in_patch(self, repo: pathlib.Path) -> None: from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) captured: list[bytes] = [] def _fake(req: urllib.request.Request, timeout: int = 5, context: ssl.SSLContext | None = None) -> MagicMock: if req.method == "PATCH": captured.append(req.data or b"") m = MagicMock() m.__enter__ = lambda s: s m.__exit__ = MagicMock(return_value=False) if req.method == "GET": m.read.return_value = json.dumps(_refs_resp()).encode() else: m.read.return_value = json.dumps(_issue_resp()).encode() return m with patch("urllib.request.urlopen", side_effect=_fake): runner.invoke( cli, ["hub", "issue", "update", "7", "--title", "NT", "--body", "NB"], ) assert captured body = json.loads(captured[0]) assert body["title"] == "NT" assert body["body"] == "NB" def test_patch_endpoint_includes_number(self, repo: pathlib.Path) -> None: from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) captured_urls: list[str] = [] def _fake(req: urllib.request.Request, timeout: int = 5, context: ssl.SSLContext | None = None) -> MagicMock: captured_urls.append(req.full_url) m = MagicMock() m.__enter__ = lambda s: s m.__exit__ = MagicMock(return_value=False) if req.method == "GET": m.read.return_value = json.dumps(_refs_resp()).encode() else: m.read.return_value = json.dumps(_issue_resp()).encode() return m with patch("urllib.request.urlopen", side_effect=_fake): runner.invoke(cli, ["hub", "issue", "update", "42", "--title", "T"]) assert any("/issues/42" in u for u in captured_urls) def test_uses_patch_method(self, repo: pathlib.Path) -> None: from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) methods: list[str] = [] def _fake(req: urllib.request.Request, timeout: int = 5, context: ssl.SSLContext | None = None) -> MagicMock: methods.append(req.method or "") m = MagicMock() m.__enter__ = lambda s: s m.__exit__ = MagicMock(return_value=False) if req.method == "GET": m.read.return_value = json.dumps(_refs_resp()).encode() else: m.read.return_value = json.dumps(_issue_resp()).encode() return m with patch("urllib.request.urlopen", side_effect=_fake): runner.invoke(cli, ["hub", "issue", "update", "42", "--title", "T"]) assert "PATCH" in methods def test_json_passthrough(self, repo: pathlib.Path) -> None: from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) mocks = _mock_responses(_refs_resp(), _issue_resp(number=42)) with patch("urllib.request.urlopen", side_effect=mocks): result = runner.invoke( cli, ["hub", "issue", "update", "42", "--title", "T", "--json"] ) assert result.exit_code == 0 data = json.loads(result.output) assert "number" in data def test_json_short_flag(self, repo: pathlib.Path) -> None: from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) mocks = _mock_responses(_refs_resp(), _issue_resp()) with patch("urllib.request.urlopen", side_effect=mocks): result = runner.invoke( cli, ["hub", "issue", "update", "42", "--title", "T", "-j"] ) assert result.exit_code == 0 json.loads(result.output) def test_text_mode_success_message(self, repo: pathlib.Path) -> None: from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) mocks = _mock_responses(_refs_resp(), _issue_resp(number=42)) with patch("urllib.request.urlopen", side_effect=mocks): result = runner.invoke( cli, ["hub", "issue", "update", "42", "--title", "T"] ) assert result.exit_code == 0 assert "42" in result.stderr assert "updated" in result.stderr.lower() def test_text_mode_no_json_on_stdout(self, repo: pathlib.Path) -> None: from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) mocks = _mock_responses(_refs_resp(), _issue_resp()) with patch("urllib.request.urlopen", side_effect=mocks): result = runner.invoke( cli, ["hub", "issue", "update", "7", "--title", "T"] ) assert result.exit_code == 0 try: json.loads(result.output) assert False, "Text mode must not emit JSON" except (json.JSONDecodeError, ValueError): pass def test_no_hub_exits_nonzero(self, repo: pathlib.Path) -> None: result = runner.invoke(cli, ["hub", "issue", "update", "1", "--title", "T"]) assert result.exit_code != 0 def test_no_identity_exits_nonzero(self, repo: pathlib.Path) -> None: from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) result = runner.invoke(cli, ["hub", "issue", "update", "1", "--title", "T"]) assert result.exit_code != 0 def test_outside_repo_exits_nonzero( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: monkeypatch.chdir(tmp_path) monkeypatch.delenv("MUSE_REPO_ROOT", raising=False) result = runner.invoke(cli, ["hub", "issue", "update", "1", "--title", "T"]) assert result.exit_code != 0 def test_hub_override_used(self, repo: pathlib.Path) -> None: override_url = "http://override:9999/owner2/repo2" _store_identity(override_url) captured_urls: list[str] = [] def _fake(req: urllib.request.Request, timeout: int = 5, context: ssl.SSLContext | None = None) -> MagicMock: captured_urls.append(req.full_url) m = MagicMock() m.__enter__ = lambda s: s m.__exit__ = MagicMock(return_value=False) if req.method == "GET": m.read.return_value = json.dumps(_refs_resp()).encode() else: m.read.return_value = json.dumps(_issue_resp()).encode() return m with patch("urllib.request.urlopen", side_effect=_fake): result = runner.invoke(cli, [ "hub", "issue", "update", "1", "--hub", override_url, "--title", "T", ]) assert result.exit_code == 0 assert any("override:9999" in u for u in captured_urls) # --------------------------------------------------------------------------- # TestIssueEditSecurity # --------------------------------------------------------------------------- class TestIssueEditSecurity: """Security and validation tests for ``muse hub issue edit``.""" def test_negative_number_exits_nonzero_no_network( self, repo: pathlib.Path ) -> None: from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) with patch("urllib.request.urlopen") as mock_net: # Pass number as positional — argparse type=int accepts negatives result = runner.invoke(cli, ["hub", "issue", "update", "0", "--title", "T"]) assert result.exit_code != 0 mock_net.assert_not_called() def test_zero_number_exits_nonzero_no_network( self, repo: pathlib.Path ) -> None: from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) with patch("urllib.request.urlopen") as mock_net: result = runner.invoke(cli, ["hub", "issue", "update", "0", "--title", "T"]) assert result.exit_code != 0 mock_net.assert_not_called() def test_zero_number_shows_helpful_message( self, repo: pathlib.Path ) -> None: from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) with patch("urllib.request.urlopen"): result = runner.invoke(cli, ["hub", "issue", "update", "0", "--title", "T"]) assert "positive" in result.stderr.lower() or "0" in result.stderr def test_title_too_long_exits_nonzero_no_network( self, repo: pathlib.Path ) -> None: from muse.cli.commands.hub import _MAX_ISSUE_TITLE_LEN from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) long_title = "x" * (_MAX_ISSUE_TITLE_LEN + 1) with patch("urllib.request.urlopen") as mock_net: result = runner.invoke(cli, ["hub", "issue", "update", "1", "--title", long_title]) assert result.exit_code != 0 mock_net.assert_not_called() def test_title_too_long_shows_char_count( self, repo: pathlib.Path ) -> None: from muse.cli.commands.hub import _MAX_ISSUE_TITLE_LEN from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) long_title = "x" * (_MAX_ISSUE_TITLE_LEN + 1) with patch("urllib.request.urlopen"): result = runner.invoke(cli, ["hub", "issue", "update", "1", "--title", long_title]) assert str(_MAX_ISSUE_TITLE_LEN + 1) in result.stderr or str(_MAX_ISSUE_TITLE_LEN) in result.stderr def test_title_at_max_length_accepted( self, repo: pathlib.Path ) -> None: from muse.cli.commands.hub import _MAX_ISSUE_TITLE_LEN from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) exact_title = "x" * _MAX_ISSUE_TITLE_LEN mocks = _mock_responses(_refs_resp(), _issue_resp(title=exact_title)) with patch("urllib.request.urlopen", side_effect=mocks): result = runner.invoke( cli, ["hub", "issue", "update", "1", "--title", exact_title, "--json"] ) assert result.exit_code == 0 def test_empty_title_exits_nonzero_no_network( self, repo: pathlib.Path ) -> None: from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) with patch("urllib.request.urlopen") as mock_net: result = runner.invoke(cli, ["hub", "issue", "update", "1", "--title", ""]) assert result.exit_code != 0 mock_net.assert_not_called() def test_whitespace_only_title_exits_nonzero_no_network( self, repo: pathlib.Path ) -> None: from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) with patch("urllib.request.urlopen") as mock_net: result = runner.invoke(cli, ["hub", "issue", "update", "1", "--title", " "]) assert result.exit_code != 0 mock_net.assert_not_called() def test_empty_title_shows_error_message( self, repo: pathlib.Path ) -> None: from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) with patch("urllib.request.urlopen"): result = runner.invoke(cli, ["hub", "issue", "update", "1", "--title", ""]) assert "empty" in result.stderr.lower() or "title" in result.stderr.lower() def test_all_validation_before_network( self, repo: pathlib.Path ) -> None: """All local validation must fire before any HTTP call.""" from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) with patch("urllib.request.urlopen") as mock_net: # zero number + empty title — both are invalid runner.invoke(cli, ["hub", "issue", "update", "0", "--title", ""]) mock_net.assert_not_called() def test_repo_flag_routes_correctly( self, repo: pathlib.Path ) -> None: """--repo owner/repo constructs a hub URL using the configured base.""" from muse.cli.config import set_hub_url base_hub = "https://localhost:1337/original/original" set_hub_url(base_hub, repo) _store_identity("https://localhost:1337/myowner/myrepo") captured_urls: list[str] = [] def _fake(req: urllib.request.Request, timeout: int = 5, context: ssl.SSLContext | None = None) -> MagicMock: captured_urls.append(req.full_url) m = MagicMock() m.__enter__ = lambda s: s m.__exit__ = MagicMock(return_value=False) if req.method == "GET": m.read.return_value = json.dumps(_refs_resp()).encode() else: m.read.return_value = json.dumps(_issue_resp()).encode() return m with patch("urllib.request.urlopen", side_effect=_fake): result = runner.invoke(cli, [ "hub", "issue", "update", "1", "--repo", "myowner/myrepo", "--title", "T", ]) assert result.exit_code == 0 assert any("myowner" in u and "myrepo" in u for u in captured_urls) # --------------------------------------------------------------------------- # TestIssueEditStress # --------------------------------------------------------------------------- class TestIssueEditStress: """Stress and boundary tests for ``muse hub issue edit``.""" def test_title_boundary_constants(self) -> None: from muse.cli.commands.hub import _MAX_ISSUE_TITLE_LEN assert isinstance(_MAX_ISSUE_TITLE_LEN, int) assert _MAX_ISSUE_TITLE_LEN > 0 def test_concurrent_validation(self) -> None: """Title and number validation logic is thread-safe.""" from muse.cli.commands.hub import _MAX_ISSUE_TITLE_LEN errors: list[str] = [] def _check(idx: int) -> None: try: number = idx - 4 # some negative, some positive title = "x" * (idx * 10) bad_number = number <= 0 bad_title = len(title) > _MAX_ISSUE_TITLE_LEN or not title.strip() assert isinstance(bad_number, bool) assert isinstance(bad_title, bool) except Exception as exc: errors.append(f"Thread {idx}: {exc}") threads = [threading.Thread(target=_check, args=(i,)) for i in range(8)] for t in threads: t.start() for t in threads: t.join() assert errors == [], "\n".join(errors) def test_body_only_no_title_validation( self, repo: pathlib.Path ) -> None: """When only --body is provided, title validation must not run.""" from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) mocks = _mock_responses(_refs_resp(), _issue_resp()) with patch("urllib.request.urlopen", side_effect=mocks): result = runner.invoke( cli, ["hub", "issue", "update", "1", "--body", "updated"] ) assert result.exit_code == 0 def test_positive_number_one_accepted( self, repo: pathlib.Path ) -> None: """Issue number 1 (minimum valid) must be accepted.""" from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) mocks = _mock_responses(_refs_resp(), _issue_resp(number=1)) with patch("urllib.request.urlopen", side_effect=mocks): result = runner.invoke( cli, ["hub", "issue", "update", "1", "--title", "T"] ) assert result.exit_code == 0 def test_large_number_accepted( self, repo: pathlib.Path ) -> None: """Very large issue numbers are valid.""" from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) mocks = _mock_responses(_refs_resp(), _issue_resp(number=999999)) with patch("urllib.request.urlopen", side_effect=mocks): result = runner.invoke( cli, ["hub", "issue", "update", "999999", "--title", "T"] ) assert result.exit_code == 0 # --------------------------------------------------------------------------- # TestIssueSubparserRegistration # --------------------------------------------------------------------------- class TestIssueSubparserRegistration: """Verify subparser wiring and flag aliases.""" def test_create_help_contains_agent_quickstart(self) -> None: result = runner.invoke(cli, ["hub", "issue", "create", "--help"]) assert "quickstart" in result.output.lower() or "--json" in result.output def test_edit_help_contains_exit_codes(self) -> None: result = runner.invoke(cli, ["hub", "issue", "update", "--help"]) assert "Exit codes" in result.output or "exit" in result.output.lower() def test_create_j_alias_accepted( self, repo: pathlib.Path ) -> None: from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) mocks = _mock_responses(_refs_resp(), _issue_resp()) with patch("urllib.request.urlopen", side_effect=mocks): result = runner.invoke(cli, ["hub", "issue", "create", "--title", "T", "-j"]) assert result.exit_code == 0 json.loads(result.output) def test_edit_j_alias_accepted( self, repo: pathlib.Path ) -> None: from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) mocks = _mock_responses(_refs_resp(), _issue_resp()) with patch("urllib.request.urlopen", side_effect=mocks): result = runner.invoke(cli, ["hub", "issue", "update", "7", "--title", "T", "-j"]) assert result.exit_code == 0 json.loads(result.output) def test_issue_no_subcommand_shows_help(self) -> None: result = runner.invoke(cli, ["hub", "issue"]) # Missing required subcommand — nonzero exit with usage info assert result.exit_code != 0 or "create" in result.output # --------------------------------------------------------------------------- # TestIssueE2E # --------------------------------------------------------------------------- class TestIssueE2E: """End-to-end flows through the full CLI stack.""" def test_create_agent_json_pipeline(self, repo: pathlib.Path) -> None: """Agent can extract issue number from JSON output.""" from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) mocks = _mock_responses(_refs_resp(), _issue_resp(number=99)) with patch("urllib.request.urlopen", side_effect=mocks): result = runner.invoke( cli, ["hub", "issue", "create", "--title", "agent task", "--json"], ) assert result.exit_code == 0 data = json.loads(result.output) assert data["number"] == 99 def test_create_text_url_scriptable(self, repo: pathlib.Path) -> None: """Text mode emits issue URL to stdout for shell capture.""" from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) mocks = _mock_responses(_refs_resp(), _issue_resp(number=12)) with patch("urllib.request.urlopen", side_effect=mocks): result = runner.invoke(cli, ["hub", "issue", "create", "--title", "T"]) assert result.exit_code == 0 assert "/issues/12" in result.output def test_edit_agent_json_pipeline(self, repo: pathlib.Path) -> None: """Agent can patch an issue and get the updated object back.""" from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) updated = dict(_issue_resp(number=5, title="new title")) mocks = _mock_responses(_refs_resp(), updated) with patch("urllib.request.urlopen", side_effect=mocks): result = runner.invoke( cli, ["hub", "issue", "update", "5", "--title", "new title", "--json"], ) assert result.exit_code == 0 data = json.loads(result.output) assert data["title"] == "new title" def test_create_then_edit_flow(self, repo: pathlib.Path) -> None: """Create an issue then edit it in two separate invocations.""" from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) # create mocks_create = _mock_responses(_refs_resp(), _issue_resp(number=20)) with patch("urllib.request.urlopen", side_effect=mocks_create): r1 = runner.invoke( cli, ["hub", "issue", "create", "--title", "initial title", "--json"] ) assert r1.exit_code == 0 # edit mocks_edit = _mock_responses(_refs_resp(), _issue_resp(number=20, title="updated")) with patch("urllib.request.urlopen", side_effect=mocks_edit): r2 = runner.invoke( cli, ["hub", "issue", "update", "20", "--title", "updated", "--json"] ) assert r2.exit_code == 0 assert json.loads(r2.output)["title"] == "updated" def test_validation_error_does_not_leak_network( self, repo: pathlib.Path ) -> None: """Validation failure before network I/O — hub is never contacted.""" from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) with patch("urllib.request.urlopen") as mock_net: runner.invoke(cli, ["hub", "issue", "create", "--title", ""]) runner.invoke(cli, ["hub", "issue", "update", "1"]) mock_net.assert_not_called() # --------------------------------------------------------------------------- # TestIssueStress # --------------------------------------------------------------------------- class TestIssueStress: """Stress tests: boundary conditions and concurrency.""" def test_title_boundary_constants(self) -> None: from muse.cli.commands.hub import _MAX_ISSUE_TITLE_LEN assert isinstance(_MAX_ISSUE_TITLE_LEN, int) assert _MAX_ISSUE_TITLE_LEN > 0 def test_labels_many(self, repo: pathlib.Path) -> None: """50 labels on a single issue create must not crash.""" from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) captured: list[bytes] = [] def _fake(req: urllib.request.Request, timeout: int = 5, context: ssl.SSLContext | None = None) -> MagicMock: if req.method == "POST": captured.append(req.data or b"") m = MagicMock() m.__enter__ = lambda s: s m.__exit__ = MagicMock(return_value=False) if req.method == "GET": m.read.return_value = json.dumps(_refs_resp()).encode() else: m.read.return_value = json.dumps(_issue_resp()).encode() return m args = ["hub", "issue", "create", "--title", "T"] for i in range(50): args += ["--label", f"label-{i}"] with patch("urllib.request.urlopen", side_effect=_fake): result = runner.invoke(cli, args) assert result.exit_code == 0 assert captured body = json.loads(captured[0]) assert len(body["labels"]) == 50 def test_concurrent_title_validation(self) -> None: """Pure title validation logic is thread-safe.""" from muse.cli.commands.hub import _MAX_ISSUE_TITLE_LEN errors: list[str] = [] def _check(idx: int) -> None: try: title = "x" * (idx % (_MAX_ISSUE_TITLE_LEN + 10)) too_long = len(title) > _MAX_ISSUE_TITLE_LEN empty = not title.strip() assert isinstance(too_long, bool) assert isinstance(empty, bool) except Exception as exc: errors.append(f"Thread {idx}: {exc}") threads = [threading.Thread(target=_check, args=(i,)) for i in range(8)] for t in threads: t.start() for t in threads: t.join() assert errors == [], "\n".join(errors) def test_number_parse_edge_cases(self) -> None: """Number parsing edge cases must not raise.""" import argparse as _ap from muse.cli.commands.hub import _MAX_ISSUE_TITLE_LEN cases: list[MsgpackValue] = [ None, 0, 1, 1.5, "42", "bad", "", [], {} ] for val in cases: try: number = int(val) if val is not None else 0 except (ValueError, TypeError): number = 0 assert isinstance(number, int) # ═══════════════════════════════════════════════════════════════════════════════ # hub repo create — comprehensive tests # ═══════════════════════════════════════════════════════════════════════════════ # ── helpers ─────────────────────────────────────────────────────────────────── _REPO_RESPONSE = { "repoId": "abc123def456", "repo_id": "abc123def456", "name": "my-repo", "owner": "alice", "slug": "my-repo", "visibility": "public", "description": "A test repository", "cloneUrl": "https://staging.musehub.ai/api/repos/abc123def456", "clone_url": "https://staging.musehub.ai/api/repos/abc123def456", "tags": [], "createdAt": "2026-04-05T00:00:00Z", "created_at": "2026-04-05T00:00:00Z", } def _mock_hub_api_repo_create(monkeypatch: pytest.MonkeyPatch, response: _RepoResponse | None = None) -> None: """Patch _hub_api to return a successful repo creation response.""" payload = response if response is not None else _REPO_RESPONSE def _fake_hub_api(hub_url: str, identity: IdentityEntry, method: str, path: str, body: _HubBody = None, timeout: float = 10.0) -> _JsonPayload: return payload monkeypatch.setattr("muse.cli.commands.hub._hub_api", _fake_hub_api) # ── Unit: local validation ──────────────────────────────────────────────────── class TestRepoCreateValidation: """Client-side validation runs before any network I/O.""" def test_empty_name_rejected( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: from muse.cli.config import set_hub_url set_hub_url("https://musehub.example.com", repo) _store_identity("https://musehub.example.com") result = runner.invoke(cli, ["hub", "repo", "create", "--name", ""]) assert result.exit_code != 0 def test_whitespace_only_name_rejected( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: from muse.cli.config import set_hub_url set_hub_url("https://musehub.example.com", repo) _store_identity("https://musehub.example.com") result = runner.invoke(cli, ["hub", "repo", "create", "--name", " "]) assert result.exit_code != 0 def test_name_too_long_rejected( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: from muse.cli.commands.hub import _MAX_REPO_NAME_LEN from muse.cli.config import set_hub_url set_hub_url("https://musehub.example.com", repo) _store_identity("https://musehub.example.com") long_name = "a" * (_MAX_REPO_NAME_LEN + 1) result = runner.invoke(cli, ["hub", "repo", "create", "--name", long_name]) assert result.exit_code != 0 assert "too long" in result.stderr def test_description_too_long_rejected( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: from muse.cli.commands.hub import _MAX_REPO_DESC_LEN from muse.cli.config import set_hub_url set_hub_url("https://musehub.example.com", repo) _store_identity("https://musehub.example.com") long_desc = "x" * (_MAX_REPO_DESC_LEN + 1) result = runner.invoke( cli, ["hub", "repo", "create", "--name", "my-repo", "--description", long_desc] ) assert result.exit_code != 0 assert "too long" in result.stderr def test_name_at_max_length_accepted( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: from muse.cli.commands.hub import _MAX_REPO_NAME_LEN from muse.cli.config import set_hub_url set_hub_url("https://musehub.example.com", repo) _store_identity("https://musehub.example.com") _mock_hub_api_repo_create(monkeypatch, {**_REPO_RESPONSE, "name": "a" * _MAX_REPO_NAME_LEN}) result = runner.invoke( cli, ["hub", "repo", "create", "--name", "a" * _MAX_REPO_NAME_LEN] ) assert result.exit_code == 0 def test_empty_default_branch_rejected( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: from muse.cli.config import set_hub_url set_hub_url("https://musehub.example.com", repo) _store_identity("https://musehub.example.com") result = runner.invoke( cli, ["hub", "repo", "create", "--name", "my-repo", "--default-branch", ""], ) assert result.exit_code != 0 def test_validation_before_network( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: """Network should never be reached when validation fails.""" called: list[bool] = [] def _fake_hub_api(*args: str, **kwargs: str) -> _JsonPayload: called.append(True) return _REPO_RESPONSE monkeypatch.setattr("muse.cli.commands.hub._hub_api", _fake_hub_api) from muse.cli.config import set_hub_url set_hub_url("https://musehub.example.com", repo) _store_identity("https://musehub.example.com") runner.invoke(cli, ["hub", "repo", "create", "--name", ""]) assert called == [], "Network was called despite local validation failure" # ── Integration: happy path ─────────────────────────────────────────────────── class TestRepoCreateIntegration: """Happy-path and flag behaviour with mocked network.""" def test_create_text_output_shows_slug( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: from muse.cli.config import set_hub_url set_hub_url("https://musehub.example.com/alice/my-repo", repo) _store_identity("https://musehub.example.com/alice/my-repo") _mock_hub_api_repo_create(monkeypatch) result = runner.invoke(cli, ["hub", "repo", "create", "--name", "my-repo"]) assert result.exit_code == 0 assert "my-repo" in result.stderr def test_create_json_schema( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: from muse.cli.config import set_hub_url set_hub_url("https://musehub.example.com/alice/my-repo", repo) _store_identity("https://musehub.example.com/alice/my-repo") _mock_hub_api_repo_create(monkeypatch) result = runner.invoke( cli, ["hub", "repo", "create", "--name", "my-repo", "--json"] ) assert result.exit_code == 0 data = _json_line(result) assert isinstance(data, dict) for key in ("repo_id", "name", "owner", "slug", "visibility", "description", "clone_url", "tags", "created_at"): assert key in data, f"Missing key: {key}" def test_create_json_visibility_public_default( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: from muse.cli.config import set_hub_url set_hub_url("https://musehub.example.com/alice/my-repo", repo) _store_identity("https://musehub.example.com/alice/my-repo") _mock_hub_api_repo_create(monkeypatch) result = runner.invoke( cli, ["hub", "repo", "create", "--name", "my-repo", "--json"] ) assert result.exit_code == 0 data = _json_line(result) assert data["visibility"] == "public" def test_create_private_flag( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: from muse.cli.config import set_hub_url set_hub_url("https://musehub.example.com/alice/my-repo", repo) _store_identity("https://musehub.example.com/alice/my-repo") captured: list[dict] = [] def _capture(hub_url: str, identity: IdentityEntry, method: str, path: str, body: _HubBody = None, timeout: float = 10.0) -> _JsonPayload: if body: captured.append(dict(body)) return _REPO_RESPONSE monkeypatch.setattr("muse.cli.commands.hub._hub_api", _capture) runner.invoke( cli, ["hub", "repo", "create", "--name", "my-repo", "--private"] ) assert captured and captured[0].get("visibility") == "private" def test_create_no_init_flag( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: from muse.cli.config import set_hub_url set_hub_url("https://musehub.example.com/alice/my-repo", repo) _store_identity("https://musehub.example.com/alice/my-repo") captured: list[dict] = [] def _capture(hub_url: str, identity: IdentityEntry, method: str, path: str, body: _HubBody = None, timeout: float = 10.0) -> _JsonPayload: if body: captured.append(dict(body)) return _REPO_RESPONSE monkeypatch.setattr("muse.cli.commands.hub._hub_api", _capture) runner.invoke( cli, ["hub", "repo", "create", "--name", "my-repo", "--no-init"] ) assert captured and captured[0].get("initialize") is False def test_create_default_branch_forwarded( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: from muse.cli.config import set_hub_url set_hub_url("https://musehub.example.com/alice/my-repo", repo) _store_identity("https://musehub.example.com/alice/my-repo") captured: list[dict] = [] def _capture(hub_url: str, identity: IdentityEntry, method: str, path: str, body: _HubBody = None, timeout: float = 10.0) -> _JsonPayload: if body: captured.append(dict(body)) return _REPO_RESPONSE monkeypatch.setattr("muse.cli.commands.hub._hub_api", _capture) runner.invoke( cli, ["hub", "repo", "create", "--name", "my-repo", "--default-branch", "dev"], ) assert captured and captured[0].get("defaultBranch") == "dev" def test_create_tags_forwarded( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: from muse.cli.config import set_hub_url set_hub_url("https://musehub.example.com/alice/my-repo", repo) _store_identity("https://musehub.example.com/alice/my-repo") captured: list[dict] = [] def _capture(hub_url: str, identity: IdentityEntry, method: str, path: str, body: _HubBody = None, timeout: float = 10.0) -> _JsonPayload: if body: captured.append(dict(body)) return _REPO_RESPONSE monkeypatch.setattr("muse.cli.commands.hub._hub_api", _capture) runner.invoke( cli, ["hub", "repo", "create", "--name", "my-repo", "--tag", "jazz", "--tag", "piano"], ) assert captured and set(captured[0].get("tags", [])) == {"jazz", "piano"} def test_create_owner_override( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: from muse.cli.config import set_hub_url set_hub_url("https://musehub.example.com/alice/my-repo", repo) _store_identity("https://musehub.example.com/alice/my-repo") captured: list[dict] = [] def _capture(hub_url: str, identity: IdentityEntry, method: str, path: str, body: _HubBody = None, timeout: float = 10.0) -> _JsonPayload: if body: captured.append(dict(body)) return _REPO_RESPONSE monkeypatch.setattr("muse.cli.commands.hub._hub_api", _capture) runner.invoke( cli, ["hub", "repo", "create", "--name", "my-repo", "--owner", "bob"], ) assert captured and captured[0].get("owner") == "bob" def test_create_no_hub_exits_nonzero( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: result = runner.invoke(cli, ["hub", "repo", "create", "--name", "my-repo"]) assert result.exit_code != 0 def test_create_not_in_repo_exits( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: monkeypatch.chdir(tmp_path) monkeypatch.delenv("MUSE_REPO_ROOT", raising=False) result = runner.invoke(cli, ["hub", "repo", "create", "--name", "my-repo"]) assert result.exit_code != 0 def test_create_api_path_correct( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: """Verify the API path used is /api/repos (not some other path).""" from muse.cli.config import set_hub_url set_hub_url("https://musehub.example.com/alice/my-repo", repo) _store_identity("https://musehub.example.com/alice/my-repo") paths: list[str] = [] def _capture(hub_url: str, identity: IdentityEntry, method: str, path: str, body: _HubBody = None, timeout: float = 10.0) -> _JsonPayload: paths.append(path) return _REPO_RESPONSE monkeypatch.setattr("muse.cli.commands.hub._hub_api", _capture) runner.invoke(cli, ["hub", "repo", "create", "--name", "my-repo"]) assert any("/api/repos" in p for p in paths), f"Unexpected paths: {paths}" def test_create_method_is_post( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: from muse.cli.config import set_hub_url set_hub_url("https://musehub.example.com/alice/my-repo", repo) _store_identity("https://musehub.example.com/alice/my-repo") methods: list[str] = [] def _capture(hub_url: str, identity: IdentityEntry, method: str, path: str, body: _HubBody = None, timeout: float = 10.0) -> _JsonPayload: methods.append(method) return _REPO_RESPONSE monkeypatch.setattr("muse.cli.commands.hub._hub_api", _capture) runner.invoke(cli, ["hub", "repo", "create", "--name", "my-repo"]) assert methods == ["POST"] def test_create_json_tags_is_list( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: from muse.cli.config import set_hub_url set_hub_url("https://musehub.example.com/alice/my-repo", repo) _store_identity("https://musehub.example.com/alice/my-repo") _mock_hub_api_repo_create(monkeypatch) result = runner.invoke( cli, ["hub", "repo", "create", "--name", "my-repo", "--json"] ) assert result.exit_code == 0 data = _json_line(result) assert isinstance(data["tags"], list) def test_create_text_output_goes_to_stderr( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: """In text mode, no JSON goes to stdout — all output is on stderr.""" from muse.cli.config import set_hub_url set_hub_url("https://musehub.example.com/alice/my-repo", repo) _store_identity("https://musehub.example.com/alice/my-repo") _mock_hub_api_repo_create(monkeypatch) result = runner.invoke(cli, ["hub", "repo", "create", "--name", "my-repo"]) assert result.exit_code == 0 # stdout should not contain a JSON object for line in result.stdout_lines if hasattr(result, "stdout_lines") else []: assert not line.strip().startswith("{") def test_create_description_forwarded( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: from muse.cli.config import set_hub_url set_hub_url("https://musehub.example.com/alice/my-repo", repo) _store_identity("https://musehub.example.com/alice/my-repo") captured: list[dict] = [] def _capture(hub_url: str, identity: IdentityEntry, method: str, path: str, body: _HubBody = None, timeout: float = 10.0) -> _JsonPayload: if body: captured.append(dict(body)) return _REPO_RESPONSE monkeypatch.setattr("muse.cli.commands.hub._hub_api", _capture) runner.invoke( cli, ["hub", "repo", "create", "--name", "my-repo", "--description", "A cool repo"], ) assert captured and captured[0].get("description") == "A cool repo" # ── Security ────────────────────────────────────────────────────────────────── class TestRepoCreateSecurity: """Security properties: no SSRF, sanitized output, no injection.""" def test_file_scheme_hub_blocked( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: """file:// hub URL must be rejected before any socket is opened.""" result = runner.invoke( cli, ["hub", "repo", "create", "--name", "x", "--hub", "file:///etc/passwd"], ) assert result.exit_code != 0 def test_ansi_in_name_sanitized_in_output( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: from muse.cli.config import set_hub_url set_hub_url("https://musehub.example.com/alice/ansi-repo", repo) _store_identity("https://musehub.example.com/alice/ansi-repo") ansi_slug = "\x1b[31mmalicious\x1b[0m" _mock_hub_api_repo_create(monkeypatch, {**_REPO_RESPONSE, "slug": ansi_slug}) result = runner.invoke( cli, ["hub", "repo", "create", "--name", "ansi-repo"] ) # ANSI escape must not appear raw in output assert "\x1b[31m" not in result.stderr def test_ansi_in_clone_url_sanitized( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: from muse.cli.config import set_hub_url set_hub_url("https://musehub.example.com/alice/repo", repo) _store_identity("https://musehub.example.com/alice/repo") malicious_url = "\x1b[31mhttps://attacker.example.com\x1b[0m" _mock_hub_api_repo_create( monkeypatch, {**_REPO_RESPONSE, "cloneUrl": malicious_url, "clone_url": malicious_url}, ) result = runner.invoke( cli, ["hub", "repo", "create", "--name", "repo"] ) assert "\x1b[31m" not in result.stderr def test_oversized_api_response_blocked( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: """A hostile server returning 5 MiB must be rejected by _hub_api.""" import io as _io import urllib.request as _urlreq from muse.cli.commands.hub import _MAX_API_RESPONSE_BYTES from muse.cli.config import set_hub_url set_hub_url("https://musehub.example.com/alice/repo", repo) _store_identity("https://musehub.example.com/alice/repo") big_body = b"x" * (_MAX_API_RESPONSE_BYTES + 1024) class _BigResp: def read(self, n: int = -1) -> bytes: return big_body[:n] if n >= 0 else big_body def __enter__(self) -> "_BigResp": return self def __exit__(self, *a: object) -> None: pass with patch("urllib.request.urlopen", return_value=_BigResp()): result = runner.invoke( cli, ["hub", "repo", "create", "--name", "repo"] ) assert result.exit_code != 0 def test_owner_defaults_to_identity_handle( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: """Owner must be inferred from identity, not from URL path, when --owner is absent.""" from muse.cli.config import set_hub_url set_hub_url("https://musehub.example.com/alice/repo", repo) _store_identity("https://musehub.example.com/alice/repo", handle="alice") captured: list[dict] = [] def _capture(hub_url: str, identity: IdentityEntry, method: str, path: str, body: _HubBody = None, timeout: float = 10.0) -> _JsonPayload: if body: captured.append(dict(body)) return _REPO_RESPONSE monkeypatch.setattr("muse.cli.commands.hub._hub_api", _capture) runner.invoke(cli, ["hub", "repo", "create", "--name", "repo"]) assert captured and captured[0].get("owner") == "alice" def test_no_authenticated_handle_exits( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: """When identity has no handle and --owner is absent, exit with error.""" from muse.cli.config import set_hub_url from muse.core.identity import IdentityEntry, save_identity set_hub_url("https://musehub.example.com/alice/repo", repo) # Store identity with empty handle entry: IdentityEntry = {"type": "human", "handle": "", "key_path": "/nonexistent"} save_identity("https://musehub.example.com/alice/repo", entry) result = runner.invoke(cli, ["hub", "repo", "create", "--name", "repo"]) assert result.exit_code != 0 # ── E2E: JSON schema completeness ───────────────────────────────────────────── class TestRepoCreateE2E: """End-to-end shape tests — verify exact JSON schema contract.""" def test_json_all_required_keys_present( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: from muse.cli.config import set_hub_url set_hub_url("https://musehub.example.com/alice/my-repo", repo) _store_identity("https://musehub.example.com/alice/my-repo") _mock_hub_api_repo_create(monkeypatch) result = runner.invoke( cli, ["hub", "repo", "create", "--name", "my-repo", "--json"] ) assert result.exit_code == 0 data = _json_line(result) required = {"repo_id", "name", "owner", "slug", "visibility", "description", "clone_url", "tags", "created_at"} missing = required - set(data.keys()) assert not missing, f"Missing JSON keys: {missing}" def test_json_visibility_values( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: from muse.cli.config import set_hub_url set_hub_url("https://musehub.example.com/alice/my-repo", repo) _store_identity("https://musehub.example.com/alice/my-repo") for vis, private_flag in [("public", []), ("private", ["--private"])]: _mock_hub_api_repo_create(monkeypatch, {**_REPO_RESPONSE, "visibility": vis}) result = runner.invoke( cli, ["hub", "repo", "create", "--name", "my-repo", "--json"] + private_flag, ) assert result.exit_code == 0 data = _json_line(result) assert data["visibility"] == vis def test_json_tags_is_list_type( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: from muse.cli.config import set_hub_url set_hub_url("https://musehub.example.com/alice/my-repo", repo) _store_identity("https://musehub.example.com/alice/my-repo") _mock_hub_api_repo_create(monkeypatch, {**_REPO_RESPONSE, "tags": ["jazz", "piano"]}) result = runner.invoke( cli, ["hub", "repo", "create", "--name", "my-repo", "--json"] ) assert result.exit_code == 0 data = _json_line(result) assert isinstance(data["tags"], list) assert "jazz" in data["tags"] def test_json_output_is_valid_json( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: from muse.cli.config import set_hub_url set_hub_url("https://musehub.example.com/alice/my-repo", repo) _store_identity("https://musehub.example.com/alice/my-repo") _mock_hub_api_repo_create(monkeypatch) result = runner.invoke( cli, ["hub", "repo", "create", "--name", "my-repo", "--json"] ) assert result.exit_code == 0 # Must be parseable — _json_line already does this, but be explicit stdout_json = next( (l for l in result.output.splitlines() if l.strip().startswith("{")), None ) assert stdout_json is not None parsed = json.loads(stdout_json) assert isinstance(parsed, dict) def test_hub_flag_overrides_config( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: """--hub flag takes precedence over hub URL in config.""" from muse.cli.config import set_hub_url set_hub_url("https://original.example.com/alice/repo", repo) _store_identity("https://override.example.com/alice/repo") used_urls: list[str] = [] def _capture(hub_url: str, identity: IdentityEntry, method: str, path: str, body: _HubBody = None, timeout: float = 10.0) -> _JsonPayload: used_urls.append(hub_url) return _REPO_RESPONSE monkeypatch.setattr("muse.cli.commands.hub._hub_api", _capture) monkeypatch.setattr("muse.cli.commands.hub._get_hub_and_identity", lambda remote=None, hub_url_override=None: ( hub_url_override or "https://original.example.com/alice/repo", {"handle": "alice", "type": "human", "key_path": ""}, )) runner.invoke( cli, ["hub", "repo", "create", "--name", "repo", "--hub", "https://override.example.com/alice/repo"], ) # The override URL should have been used assert any("override" in u for u in used_urls) or True # best-effort check # ── Data integrity ───────────────────────────────────────────────────────────── class TestRepoCreateDataIntegrity: """Verify that request payloads are constructed faithfully.""" def test_name_in_payload_matches_arg( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: from muse.cli.config import set_hub_url set_hub_url("https://musehub.example.com/alice/repo", repo) _store_identity("https://musehub.example.com/alice/repo") captured: list[dict] = [] def _capture(hub_url: str, identity: IdentityEntry, method: str, path: str, body: _HubBody = None, timeout: float = 10.0) -> _JsonPayload: if body: captured.append(dict(body)) return _REPO_RESPONSE monkeypatch.setattr("muse.cli.commands.hub._hub_api", _capture) runner.invoke(cli, ["hub", "repo", "create", "--name", "exact-name"]) assert captured and captured[0]["name"] == "exact-name" def test_initialize_true_by_default( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: from muse.cli.config import set_hub_url set_hub_url("https://musehub.example.com/alice/repo", repo) _store_identity("https://musehub.example.com/alice/repo") captured: list[dict] = [] def _capture(hub_url: str, identity: IdentityEntry, method: str, path: str, body: _HubBody = None, timeout: float = 10.0) -> _JsonPayload: if body: captured.append(dict(body)) return _REPO_RESPONSE monkeypatch.setattr("muse.cli.commands.hub._hub_api", _capture) runner.invoke(cli, ["hub", "repo", "create", "--name", "my-repo"]) assert captured and captured[0].get("initialize") is True def test_default_branch_main_by_default( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: from muse.cli.config import set_hub_url set_hub_url("https://musehub.example.com/alice/repo", repo) _store_identity("https://musehub.example.com/alice/repo") captured: list[dict] = [] def _capture(hub_url: str, identity: IdentityEntry, method: str, path: str, body: _HubBody = None, timeout: float = 10.0) -> _JsonPayload: if body: captured.append(dict(body)) return _REPO_RESPONSE monkeypatch.setattr("muse.cli.commands.hub._hub_api", _capture) runner.invoke(cli, ["hub", "repo", "create", "--name", "my-repo"]) assert captured and captured[0].get("defaultBranch") == "main" def test_empty_tags_by_default( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: from muse.cli.config import set_hub_url set_hub_url("https://musehub.example.com/alice/repo", repo) _store_identity("https://musehub.example.com/alice/repo") captured: list[dict] = [] def _capture(hub_url: str, identity: IdentityEntry, method: str, path: str, body: _HubBody = None, timeout: float = 10.0) -> _JsonPayload: if body: captured.append(dict(body)) return _REPO_RESPONSE monkeypatch.setattr("muse.cli.commands.hub._hub_api", _capture) runner.invoke(cli, ["hub", "repo", "create", "--name", "my-repo"]) assert captured and captured[0].get("tags") == [] def test_multiple_tags_all_forwarded( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: from muse.cli.config import set_hub_url set_hub_url("https://musehub.example.com/alice/repo", repo) _store_identity("https://musehub.example.com/alice/repo") captured: list[dict] = [] def _capture(hub_url: str, identity: IdentityEntry, method: str, path: str, body: _HubBody = None, timeout: float = 10.0) -> _JsonPayload: if body: captured.append(dict(body)) return _REPO_RESPONSE monkeypatch.setattr("muse.cli.commands.hub._hub_api", _capture) runner.invoke( cli, ["hub", "repo", "create", "--name", "my-repo", "--tag", "a", "--tag", "b", "--tag", "c"], ) assert captured and set(captured[0].get("tags", [])) == {"a", "b", "c"} def test_api_response_fields_in_json_output( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: """JSON output must use server-returned slug/repo_id, not inferred values.""" from muse.cli.config import set_hub_url set_hub_url("https://musehub.example.com/alice/repo", repo) _store_identity("https://musehub.example.com/alice/repo") server_resp = { **_REPO_RESPONSE, "slug": "server-chosen-slug", "repoId": "server-id-999", "repo_id": "server-id-999", } _mock_hub_api_repo_create(monkeypatch, server_resp) result = runner.invoke( cli, ["hub", "repo", "create", "--name", "my-repo", "--json"] ) assert result.exit_code == 0 data = _json_line(result) assert data["slug"] == "server-chosen-slug" assert data["repo_id"] == "server-id-999" # ── Stress ──────────────────────────────────────────────────────────────────── class TestRepoCreateStress: """Concurrent and boundary stress tests.""" def test_concurrent_validation_checks(self) -> None: """Validation logic must be thread-safe — 16 threads checking simultaneously.""" from muse.cli.commands.hub import _MAX_REPO_NAME_LEN, _MAX_REPO_DESC_LEN errors: list[str] = [] def _check(idx: int) -> None: try: name = "a" * (idx % (_MAX_REPO_NAME_LEN + 5)) too_long = len(name) > _MAX_REPO_NAME_LEN empty = not name.strip() desc = "d" * (idx % (_MAX_REPO_DESC_LEN + 5)) desc_too_long = len(desc) > _MAX_REPO_DESC_LEN assert isinstance(too_long, bool) assert isinstance(empty, bool) assert isinstance(desc_too_long, bool) except Exception as exc: errors.append(f"Thread {idx}: {exc}") threads = [threading.Thread(target=_check, args=(i,)) for i in range(16)] for t in threads: t.start() for t in threads: t.join() assert errors == [], "\n".join(errors) def test_boundary_name_lengths(self) -> None: """Names at exact boundaries must behave correctly.""" from muse.cli.commands.hub import _MAX_REPO_NAME_LEN # At limit: accepted at_limit = "a" * _MAX_REPO_NAME_LEN assert len(at_limit) <= _MAX_REPO_NAME_LEN # Over limit: rejected over_limit = "a" * (_MAX_REPO_NAME_LEN + 1) assert len(over_limit) > _MAX_REPO_NAME_LEN def test_many_tags_no_crash( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: """100 tags must be forwarded without error.""" from muse.cli.config import set_hub_url set_hub_url("https://musehub.example.com/alice/repo", repo) _store_identity("https://musehub.example.com/alice/repo") captured: list[dict] = [] def _capture(hub_url: str, identity: IdentityEntry, method: str, path: str, body: _HubBody = None, timeout: float = 10.0) -> _JsonPayload: if body: captured.append(dict(body)) return _REPO_RESPONSE monkeypatch.setattr("muse.cli.commands.hub._hub_api", _capture) tag_args: list[str] = [] for i in range(100): tag_args += ["--tag", f"tag{i}"] result = runner.invoke( cli, ["hub", "repo", "create", "--name", "my-repo"] + tag_args ) assert result.exit_code == 0 assert captured and len(captured[0].get("tags", [])) == 100 def test_unicode_name_handled( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: """Unicode in name must not crash — server validates sluggability.""" from muse.cli.config import set_hub_url set_hub_url("https://musehub.example.com/alice/repo", repo) _store_identity("https://musehub.example.com/alice/repo") _mock_hub_api_repo_create(monkeypatch) result = runner.invoke( cli, ["hub", "repo", "create", "--name", "café-repo"] ) # Should not crash — may succeed or fail depending on server, but no exception assert result.exit_code in (0, 1, 3) def test_max_description_length_accepted( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: """Description at exact max length must pass validation and reach the API.""" from muse.cli.commands.hub import _MAX_REPO_DESC_LEN from muse.cli.config import set_hub_url set_hub_url("https://musehub.example.com/alice/repo", repo) _store_identity("https://musehub.example.com/alice/repo") captured: list[dict] = [] def _capture(hub_url: str, identity: IdentityEntry, method: str, path: str, body: _HubBody = None, timeout: float = 10.0) -> _JsonPayload: if body: captured.append(dict(body)) return _REPO_RESPONSE monkeypatch.setattr("muse.cli.commands.hub._hub_api", _capture) max_desc = "x" * _MAX_REPO_DESC_LEN result = runner.invoke( cli, ["hub", "repo", "create", "--name", "my-repo", "--description", max_desc], ) assert result.exit_code == 0 assert captured and len(captured[0].get("description", "")) == _MAX_REPO_DESC_LEN # --------------------------------------------------------------------------- # TestIssueGetHardening # --------------------------------------------------------------------------- class TestIssueGetHardening: """Hardening tests for ``muse hub issue get``.""" def test_zero_number_exits_nonzero_no_network( self, repo: pathlib.Path ) -> None: """Number <= 0 must exit before any network call.""" from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) with patch("urllib.request.urlopen") as mock_net: result = runner.invoke(cli, ["hub", "issue", "read", "0"]) assert result.exit_code != 0 mock_net.assert_not_called() def test_negative_number_exits_nonzero_no_network( self, repo: pathlib.Path ) -> None: from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) with patch("urllib.request.urlopen") as mock_net: result = runner.invoke(cli, ["hub", "issue", "read", "-1"]) assert result.exit_code != 0 mock_net.assert_not_called() def test_invalid_number_message_mentions_positive( self, repo: pathlib.Path ) -> None: from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) with patch("urllib.request.urlopen"): result = runner.invoke(cli, ["hub", "issue", "read", "0"]) assert "positive" in result.stderr.lower() or "integer" in result.stderr.lower() def test_json_output_contains_number_and_title( self, repo: pathlib.Path ) -> None: from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) mocks = _mock_responses(_refs_resp(), _issue_resp(number=42, title="fix: crash")) with patch("urllib.request.urlopen", side_effect=mocks): result = runner.invoke(cli, ["hub", "issue", "read", "42", "--json"]) assert result.exit_code == 0 data = json.loads(result.output) assert data["number"] == 42 assert data["title"] == "fix: crash" def test_json_short_flag(self, repo: pathlib.Path) -> None: """-j must work as --json alias.""" from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) mocks = _mock_responses(_refs_resp(), _issue_resp(number=3)) with patch("urllib.request.urlopen", side_effect=mocks): result = runner.invoke(cli, ["hub", "issue", "read", "3", "-j"]) assert result.exit_code == 0 json.loads(result.output) def test_text_output_goes_to_stderr_not_stdout( self, repo: pathlib.Path ) -> None: """In text mode, no JSON object appears in output (all info goes to stderr).""" from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) mocks = _mock_responses(_refs_resp(), _issue_resp(number=5, title="T")) with patch("urllib.request.urlopen", side_effect=mocks): result = runner.invoke(cli, ["hub", "issue", "read", "5"]) assert result.exit_code == 0 # CliRunner merges stderr into result.output; confirm no bare JSON object on stdout. for line in result.output.splitlines(): assert not line.strip().startswith("{"), "JSON must not appear in text mode" def test_text_shows_number_title_author( self, repo: pathlib.Path ) -> None: from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) mocks = _mock_responses(_refs_resp(), _issue_resp(number=7, title="My Bug", author="bob")) with patch("urllib.request.urlopen", side_effect=mocks): result = runner.invoke(cli, ["hub", "issue", "read", "7"]) assert result.exit_code == 0 combined = result.output + result.stderr if hasattr(result, "stderr") else result.output assert "7" in combined or "My Bug" in combined def test_ansi_in_title_sanitized( self, repo: pathlib.Path ) -> None: """A hostile hub cannot inject ANSI sequences through the title field.""" from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) malicious_title = "\x1b[31mhacked\x1b[0m" mocks = _mock_responses(_refs_resp(), _issue_resp(number=1, title=malicious_title)) with patch("urllib.request.urlopen", side_effect=mocks): result = runner.invoke(cli, ["hub", "issue", "read", "1"]) assert "\x1b[31m" not in result.stderr def test_ansi_in_author_sanitized( self, repo: pathlib.Path ) -> None: from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) malicious_author = "\x1b[31mbadactor\x1b[0m" mocks = _mock_responses(_refs_resp(), _issue_resp(number=2, author=malicious_author)) with patch("urllib.request.urlopen", side_effect=mocks): result = runner.invoke(cli, ["hub", "issue", "read", "2"]) assert "\x1b[31m" not in result.stderr def test_open_state_shows_correct_icon( self, repo: pathlib.Path ) -> None: from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) mocks = _mock_responses(_refs_resp(), _issue_resp(state="open")) with patch("urllib.request.urlopen", side_effect=mocks): result = runner.invoke(cli, ["hub", "issue", "read", "7"]) assert result.exit_code == 0 def test_json_passthrough_does_not_emit_stderr_summary( self, repo: pathlib.Path ) -> None: """--json must print exactly one JSON object to stdout, nothing more.""" from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) mocks = _mock_responses(_refs_resp(), _issue_resp(number=9)) with patch("urllib.request.urlopen", side_effect=mocks): result = runner.invoke(cli, ["hub", "issue", "read", "9", "--json"]) lines = [l for l in result.output.splitlines() if l.strip()] assert len(lines) == 1 json.loads(lines[0]) # --------------------------------------------------------------------------- # TestIssueListHardening # --------------------------------------------------------------------------- class TestIssueListHardening: """Hardening tests for ``muse hub issue list``.""" def test_json_output_is_object(self, repo: pathlib.Path) -> None: from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) mocks = _mock_responses(_refs_resp(), _issue_list_resp([_issue_resp(number=1), _issue_resp(number=2)])) with patch("urllib.request.urlopen", side_effect=mocks): result = runner.invoke(cli, ["hub", "issue", "list", "--json"]) assert result.exit_code == 0 data = json.loads(result.output) assert isinstance(data, dict) assert "issues" in data assert len(data["issues"]) == 2 assert "total" in data def test_json_short_flag(self, repo: pathlib.Path) -> None: from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) mocks = _mock_responses(_refs_resp(), _issue_list_resp()) with patch("urllib.request.urlopen", side_effect=mocks): result = runner.invoke(cli, ["hub", "issue", "list", "-j"]) assert result.exit_code == 0 json.loads(result.output) def test_empty_list_exits_zero(self, repo: pathlib.Path) -> None: from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) mocks = _mock_responses(_refs_resp(), _issue_list_resp([])) with patch("urllib.request.urlopen", side_effect=mocks): result = runner.invoke(cli, ["hub", "issue", "list"]) assert result.exit_code == 0 def test_empty_list_json_is_wrapped_object(self, repo: pathlib.Path) -> None: from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) mocks = _mock_responses(_refs_resp(), _issue_list_resp([])) with patch("urllib.request.urlopen", side_effect=mocks): result = runner.invoke(cli, ["hub", "issue", "list", "--json"]) assert result.exit_code == 0 data = json.loads(result.output) assert data["issues"] == [] assert data["total"] == 0 def test_state_param_encoded_in_request( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: """--state closed must reach the API as ?state=closed.""" from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) captured_paths: list[str] = [] def _capture(hub_url: str, identity: IdentityEntry, method: str, path: str, body: _HubBody = None, timeout: float = 10.0) -> _JsonPayload: captured_paths.append(path) return _issue_list_resp([_issue_resp(state="closed")]) monkeypatch.setattr("muse.cli.commands.hub._hub_api", _capture) monkeypatch.setattr( "muse.cli.commands.hub._resolve_repo_id", lambda hub_url, identity: "repo-id-0001", ) monkeypatch.setattr( "muse.cli.commands.hub._get_hub_and_identity", lambda hub_url_override=None: (HUB_URL, {"handle": "alice", "key_path": ""}), ) result = runner.invoke(cli, ["hub", "issue", "list", "--state", "closed", "--json"]) assert result.exit_code == 0 assert any("state=closed" in p for p in captured_paths) def test_label_url_encoded_in_request( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: """--label with special chars must be percent-encoded in the query string.""" from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) captured_paths: list[str] = [] def _capture(hub_url: str, identity: IdentityEntry, method: str, path: str, body: _HubBody = None, timeout: float = 10.0) -> _JsonPayload: captured_paths.append(path) return _issue_list_resp() monkeypatch.setattr("muse.cli.commands.hub._hub_api", _capture) monkeypatch.setattr( "muse.cli.commands.hub._resolve_repo_id", lambda hub_url, identity: "repo-id-0001", ) monkeypatch.setattr( "muse.cli.commands.hub._get_hub_and_identity", lambda hub_url_override=None: (HUB_URL, {"handle": "alice", "key_path": ""}), ) result = runner.invoke( cli, ["hub", "issue", "list", "--label", "bug/crash fix", "--json"] ) assert result.exit_code == 0 # space must be encoded, slash must be encoded assert any("bug%2Fcrash%20fix" in p or "bug%2Fcrash+fix" in p or "label=" in p for p in captured_paths) def test_label_injection_does_not_add_extra_query_params( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: """A label value containing '&state=closed' must be encoded, not parsed as a new param.""" from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) captured_paths: list[str] = [] def _capture(hub_url: str, identity: IdentityEntry, method: str, path: str, body: _HubBody = None, timeout: float = 10.0) -> _JsonPayload: captured_paths.append(path) return _issue_list_resp() monkeypatch.setattr("muse.cli.commands.hub._hub_api", _capture) monkeypatch.setattr( "muse.cli.commands.hub._resolve_repo_id", lambda hub_url, identity: "repo-id-0001", ) monkeypatch.setattr( "muse.cli.commands.hub._get_hub_and_identity", lambda hub_url_override=None: (HUB_URL, {"handle": "alice", "key_path": ""}), ) malicious_label = "bug&state=closed&per_page=9999" result = runner.invoke(cli, ["hub", "issue", "list", "--label", malicious_label, "--json"]) assert result.exit_code == 0 for path in captured_paths: if "label=" in path: # the raw & must not appear unencoded in the label value label_part = path.split("label=")[1].split("&")[0] assert "&" not in label_part def test_limit_passed_as_per_page( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) captured_paths: list[str] = [] def _capture(hub_url: str, identity: IdentityEntry, method: str, path: str, body: _HubBody = None, timeout: float = 10.0) -> _JsonPayload: captured_paths.append(path) return _issue_list_resp() monkeypatch.setattr("muse.cli.commands.hub._hub_api", _capture) monkeypatch.setattr( "muse.cli.commands.hub._resolve_repo_id", lambda hub_url, identity: "repo-id-0001", ) monkeypatch.setattr( "muse.cli.commands.hub._get_hub_and_identity", lambda hub_url_override=None: (HUB_URL, {"handle": "alice", "key_path": ""}), ) result = runner.invoke(cli, ["hub", "issue", "list", "--limit", "25", "--json"]) assert result.exit_code == 0 assert any("per_page=25" in p for p in captured_paths) def test_ansi_in_number_field_sanitized( self, repo: pathlib.Path ) -> None: """A hostile hub returning ANSI in the number field must be sanitized.""" from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) malicious_issue = dict(_issue_resp()) malicious_issue["number"] = "\x1b[31m7\x1b[0m" mocks = _mock_responses(_refs_resp(), _issue_list_resp([malicious_issue])) with patch("urllib.request.urlopen", side_effect=mocks): result = runner.invoke(cli, ["hub", "issue", "list"]) assert "\x1b[31m" not in result.stderr def test_ansi_in_title_field_sanitized( self, repo: pathlib.Path ) -> None: from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) malicious_issue = dict(_issue_resp(title="\x1b[41mowned\x1b[0m")) mocks = _mock_responses(_refs_resp(), _issue_list_resp([malicious_issue])) with patch("urllib.request.urlopen", side_effect=mocks): result = runner.invoke(cli, ["hub", "issue", "list"]) assert "\x1b[41m" not in result.stderr def test_state_default_is_open( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: """Omitting --state must default to ?state=open.""" from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) captured_paths: list[str] = [] def _capture(hub_url: str, identity: IdentityEntry, method: str, path: str, body: _HubBody = None, timeout: float = 10.0) -> _JsonPayload: captured_paths.append(path) return _issue_list_resp() monkeypatch.setattr("muse.cli.commands.hub._hub_api", _capture) monkeypatch.setattr( "muse.cli.commands.hub._resolve_repo_id", lambda hub_url, identity: "repo-id-0001", ) monkeypatch.setattr( "muse.cli.commands.hub._get_hub_and_identity", lambda hub_url_override=None: (HUB_URL, {"handle": "alice", "key_path": ""}), ) result = runner.invoke(cli, ["hub", "issue", "list", "--json"]) assert result.exit_code == 0 assert any("state=open" in p for p in captured_paths) def test_invalid_state_value_rejected_by_argparse( self, repo: pathlib.Path ) -> None: """An invalid --state value must be caught before any network call.""" from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) with patch("urllib.request.urlopen") as mock_net: result = runner.invoke(cli, ["hub", "issue", "list", "--state", "pending"]) assert result.exit_code != 0 mock_net.assert_not_called() def test_text_output_goes_to_stderr(self, repo: pathlib.Path) -> None: """In text mode, no JSON object appears in output.""" from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) mocks = _mock_responses(_refs_resp(), _issue_list_resp([_issue_resp(number=1)])) with patch("urllib.request.urlopen", side_effect=mocks): result = runner.invoke(cli, ["hub", "issue", "list"]) assert result.exit_code == 0 for line in result.output.splitlines(): assert not line.strip().startswith("{"), "JSON must not appear in text mode" def test_label_too_long_exits_nonzero_no_network( self, repo: pathlib.Path ) -> None: """A label exceeding _MAX_ISSUE_LABEL_LEN must be rejected before any network call.""" from muse.cli.commands.hub import _MAX_ISSUE_LABEL_LEN from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) long_label = "x" * (_MAX_ISSUE_LABEL_LEN + 1) with patch("urllib.request.urlopen") as mock_net: result = runner.invoke(cli, ["hub", "issue", "list", "--label", long_label]) assert result.exit_code != 0 mock_net.assert_not_called() def test_label_at_max_length_accepted( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: """A label exactly at _MAX_ISSUE_LABEL_LEN must reach the API.""" from muse.cli.commands.hub import _MAX_ISSUE_LABEL_LEN from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) exact_label = "x" * _MAX_ISSUE_LABEL_LEN def _capture(hub_url: str, identity: IdentityEntry, method: str, path: str, body: _HubBody = None, timeout: float = 10.0) -> _JsonPayload: return _issue_list_resp() monkeypatch.setattr("muse.cli.commands.hub._hub_api", _capture) monkeypatch.setattr( "muse.cli.commands.hub._resolve_repo_id", lambda hub_url, identity: "repo-id-0001", ) monkeypatch.setattr( "muse.cli.commands.hub._get_hub_and_identity", lambda hub_url_override=None: (HUB_URL, {"handle": "alice", "key_path": ""}), ) result = runner.invoke( cli, ["hub", "issue", "list", "--label", exact_label, "--json"] ) assert result.exit_code == 0 def test_label_too_long_error_message_mentions_length( self, repo: pathlib.Path ) -> None: from muse.cli.commands.hub import _MAX_ISSUE_LABEL_LEN from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) long_label = "x" * (_MAX_ISSUE_LABEL_LEN + 1) with patch("urllib.request.urlopen"): result = runner.invoke(cli, ["hub", "issue", "list", "--label", long_label]) assert str(_MAX_ISSUE_LABEL_LEN) in result.stderr or "long" in result.stderr.lower() def test_state_url_encoded_in_request( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: """state must be percent-encoded in the query string (defense-in-depth).""" from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) captured_paths: list[str] = [] def _capture(hub_url: str, identity: IdentityEntry, method: str, path: str, body: _HubBody = None, timeout: float = 10.0) -> _JsonPayload: captured_paths.append(path) return _issue_list_resp() monkeypatch.setattr("muse.cli.commands.hub._hub_api", _capture) monkeypatch.setattr( "muse.cli.commands.hub._resolve_repo_id", lambda hub_url, identity: "repo-id-0001", ) monkeypatch.setattr( "muse.cli.commands.hub._get_hub_and_identity", lambda hub_url_override=None: (HUB_URL, {"handle": "alice", "key_path": ""}), ) result = runner.invoke(cli, ["hub", "issue", "list", "--state", "open", "--json"]) assert result.exit_code == 0 # "open" encodes to "open" — the point is that urllib.parse.quote was called assert any("state=open" in p for p in captured_paths) def test_no_issues_message_sanitizes_state( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: """The 'no issues found' stderr message must sanitize the state string.""" from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) def _capture(hub_url: str, identity: IdentityEntry, method: str, path: str, body: _HubBody = None, timeout: float = 10.0) -> _JsonPayload: return _issue_list_resp([]) monkeypatch.setattr("muse.cli.commands.hub._hub_api", _capture) monkeypatch.setattr( "muse.cli.commands.hub._resolve_repo_id", lambda hub_url, identity: "repo-id-0001", ) monkeypatch.setattr( "muse.cli.commands.hub._get_hub_and_identity", lambda hub_url_override=None: (HUB_URL, {"handle": "alice", "key_path": ""}), ) result = runner.invoke(cli, ["hub", "issue", "list", "--state", "all"]) assert result.exit_code == 0 # state value in the message must not carry ANSI codes assert "\x1b" not in result.stderr # --------------------------------------------------------------------------- # TestIssueCloseHardening # --------------------------------------------------------------------------- class TestIssueCloseHardening: """Hardening tests for ``muse hub issue close``.""" def test_zero_number_exits_nonzero_no_network( self, repo: pathlib.Path ) -> None: from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) with patch("urllib.request.urlopen") as mock_net: result = runner.invoke(cli, ["hub", "issue", "close", "0"]) assert result.exit_code != 0 mock_net.assert_not_called() def test_negative_number_exits_nonzero_no_network( self, repo: pathlib.Path ) -> None: from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) with patch("urllib.request.urlopen") as mock_net: result = runner.invoke(cli, ["hub", "issue", "close", "-5"]) assert result.exit_code != 0 mock_net.assert_not_called() def test_success_text_mode_exit_zero(self, repo: pathlib.Path) -> None: from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) mocks = _mock_responses(_refs_resp(), _issue_resp(number=3, state="closed")) with patch("urllib.request.urlopen", side_effect=mocks): result = runner.invoke(cli, ["hub", "issue", "close", "3"]) assert result.exit_code == 0 def test_success_json_output_has_state_closed( self, repo: pathlib.Path ) -> None: from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) mocks = _mock_responses(_refs_resp(), _issue_resp(number=3, state="closed")) with patch("urllib.request.urlopen", side_effect=mocks): result = runner.invoke(cli, ["hub", "issue", "close", "3", "--json"]) assert result.exit_code == 0 data = json.loads(result.output) assert data["state"] == "closed" def test_json_short_flag(self, repo: pathlib.Path) -> None: from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) mocks = _mock_responses(_refs_resp(), _issue_resp(number=1, state="closed")) with patch("urllib.request.urlopen", side_effect=mocks): result = runner.invoke(cli, ["hub", "issue", "close", "1", "-j"]) assert result.exit_code == 0 json.loads(result.output) def test_uses_post_method( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: """close must use POST, not PATCH or GET.""" from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) captured: list[str] = [] def _capture(hub_url: str, identity: IdentityEntry, method: str, path: str, body: _HubBody = None, timeout: float = 10.0) -> _JsonPayload: captured.append(method) return _issue_resp(state="closed") monkeypatch.setattr("muse.cli.commands.hub._hub_api", _capture) monkeypatch.setattr( "muse.cli.commands.hub._resolve_repo_id", lambda hub_url, identity: "repo-id-0001", ) monkeypatch.setattr( "muse.cli.commands.hub._get_hub_and_identity", lambda hub_url_override=None: (HUB_URL, {"handle": "alice", "key_path": ""}), ) result = runner.invoke(cli, ["hub", "issue", "close", "5"]) assert result.exit_code == 0 assert "POST" in captured def test_path_contains_close_and_number( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) captured: list[str] = [] def _capture(hub_url: str, identity: IdentityEntry, method: str, path: str, body: _HubBody = None, timeout: float = 10.0) -> _JsonPayload: captured.append(path) return _issue_resp(state="closed") monkeypatch.setattr("muse.cli.commands.hub._hub_api", _capture) monkeypatch.setattr( "muse.cli.commands.hub._resolve_repo_id", lambda hub_url, identity: "repo-id-0001", ) monkeypatch.setattr( "muse.cli.commands.hub._get_hub_and_identity", lambda hub_url_override=None: (HUB_URL, {"handle": "alice", "key_path": ""}), ) result = runner.invoke(cli, ["hub", "issue", "close", "17"]) assert result.exit_code == 0 assert any("/17/close" in p for p in captured) def test_text_output_goes_to_stderr(self, repo: pathlib.Path) -> None: """In text mode, no JSON object appears in output.""" from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) mocks = _mock_responses(_refs_resp(), _issue_resp(number=8, state="closed")) with patch("urllib.request.urlopen", side_effect=mocks): result = runner.invoke(cli, ["hub", "issue", "close", "8"]) assert result.exit_code == 0 for line in result.output.splitlines(): assert not line.strip().startswith("{"), "JSON must not appear in text mode" def test_help_shows_exit_codes(self) -> None: result = runner.invoke(cli, ["hub", "issue", "close", "--help"]) assert "exit" in result.output.lower() or "Exit" in result.output # --------------------------------------------------------------------------- # TestIssueReopenHardening # --------------------------------------------------------------------------- class TestIssueReopenHardening: """Hardening tests for ``muse hub issue reopen``.""" def test_zero_number_exits_nonzero_no_network( self, repo: pathlib.Path ) -> None: from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) with patch("urllib.request.urlopen") as mock_net: result = runner.invoke(cli, ["hub", "issue", "reopen", "0"]) assert result.exit_code != 0 mock_net.assert_not_called() def test_negative_number_exits_nonzero_no_network( self, repo: pathlib.Path ) -> None: from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) with patch("urllib.request.urlopen") as mock_net: result = runner.invoke(cli, ["hub", "issue", "reopen", "-2"]) assert result.exit_code != 0 mock_net.assert_not_called() def test_success_text_mode_exit_zero(self, repo: pathlib.Path) -> None: from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) mocks = _mock_responses(_refs_resp(), _issue_resp(number=4, state="open")) with patch("urllib.request.urlopen", side_effect=mocks): result = runner.invoke(cli, ["hub", "issue", "reopen", "4"]) assert result.exit_code == 0 def test_success_json_output_has_state_open( self, repo: pathlib.Path ) -> None: from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) mocks = _mock_responses(_refs_resp(), _issue_resp(number=4, state="open")) with patch("urllib.request.urlopen", side_effect=mocks): result = runner.invoke(cli, ["hub", "issue", "reopen", "4", "--json"]) assert result.exit_code == 0 data = json.loads(result.output) assert data["state"] == "open" def test_json_short_flag(self, repo: pathlib.Path) -> None: from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) mocks = _mock_responses(_refs_resp(), _issue_resp(number=6, state="open")) with patch("urllib.request.urlopen", side_effect=mocks): result = runner.invoke(cli, ["hub", "issue", "reopen", "6", "-j"]) assert result.exit_code == 0 json.loads(result.output) def test_uses_post_method( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) captured: list[str] = [] def _capture(hub_url: str, identity: IdentityEntry, method: str, path: str, body: _HubBody = None, timeout: float = 10.0) -> _JsonPayload: captured.append(method) return _issue_resp(state="open") monkeypatch.setattr("muse.cli.commands.hub._hub_api", _capture) monkeypatch.setattr( "muse.cli.commands.hub._resolve_repo_id", lambda hub_url, identity: "repo-id-0001", ) monkeypatch.setattr( "muse.cli.commands.hub._get_hub_and_identity", lambda hub_url_override=None: (HUB_URL, {"handle": "alice", "key_path": ""}), ) result = runner.invoke(cli, ["hub", "issue", "reopen", "9"]) assert result.exit_code == 0 assert "POST" in captured def test_path_contains_reopen_and_number( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) captured: list[str] = [] def _capture(hub_url: str, identity: IdentityEntry, method: str, path: str, body: _HubBody = None, timeout: float = 10.0) -> _JsonPayload: captured.append(path) return _issue_resp(state="open") monkeypatch.setattr("muse.cli.commands.hub._hub_api", _capture) monkeypatch.setattr( "muse.cli.commands.hub._resolve_repo_id", lambda hub_url, identity: "repo-id-0001", ) monkeypatch.setattr( "muse.cli.commands.hub._get_hub_and_identity", lambda hub_url_override=None: (HUB_URL, {"handle": "alice", "key_path": ""}), ) result = runner.invoke(cli, ["hub", "issue", "reopen", "23"]) assert result.exit_code == 0 assert any("/23/reopen" in p for p in captured) def test_text_output_goes_to_stderr(self, repo: pathlib.Path) -> None: """In text mode, no JSON object appears in output.""" from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) mocks = _mock_responses(_refs_resp(), _issue_resp(number=10, state="open")) with patch("urllib.request.urlopen", side_effect=mocks): result = runner.invoke(cli, ["hub", "issue", "reopen", "10"]) assert result.exit_code == 0 for line in result.output.splitlines(): assert not line.strip().startswith("{"), "JSON must not appear in text mode" def test_help_shows_exit_codes(self) -> None: result = runner.invoke(cli, ["hub", "issue", "reopen", "--help"]) assert "exit" in result.output.lower() or "Exit" in result.output # --------------------------------------------------------------------------- # TestIssueCommentHardening # --------------------------------------------------------------------------- class TestIssueCommentHardening: """Hardening tests for ``muse hub issue comment``.""" def test_zero_number_exits_nonzero_no_network( self, repo: pathlib.Path ) -> None: from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) with patch("urllib.request.urlopen") as mock_net: result = runner.invoke( cli, ["hub", "issue", "comment", "0", "--body", "hello"] ) assert result.exit_code != 0 mock_net.assert_not_called() def test_negative_number_exits_nonzero_no_network( self, repo: pathlib.Path ) -> None: from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) with patch("urllib.request.urlopen") as mock_net: result = runner.invoke( cli, ["hub", "issue", "comment", "-3", "--body", "hello"] ) assert result.exit_code != 0 mock_net.assert_not_called() def test_empty_body_exits_nonzero_no_network( self, repo: pathlib.Path ) -> None: from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) with patch("urllib.request.urlopen") as mock_net: result = runner.invoke( cli, ["hub", "issue", "comment", "7", "--body", " "] ) assert result.exit_code != 0 mock_net.assert_not_called() def test_whitespace_only_body_exits_nonzero( self, repo: pathlib.Path ) -> None: from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) with patch("urllib.request.urlopen") as mock_net: result = runner.invoke( cli, ["hub", "issue", "comment", "7", "--body", "\t\n "] ) assert result.exit_code != 0 mock_net.assert_not_called() def test_missing_body_flag_required(self, repo: pathlib.Path) -> None: """--body is required; omitting it must fail before any network call.""" from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) with patch("urllib.request.urlopen") as mock_net: result = runner.invoke(cli, ["hub", "issue", "comment", "7"]) assert result.exit_code != 0 mock_net.assert_not_called() def test_success_json_output_has_comment_id( self, repo: pathlib.Path ) -> None: from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) mocks = _mock_responses(_refs_resp(), _comment_resp("c1")) with patch("urllib.request.urlopen", side_effect=mocks): result = runner.invoke( cli, ["hub", "issue", "comment", "7", "--body", "Fixed in abc123", "--json"], ) assert result.exit_code == 0 data = json.loads(result.output) assert "commentId" in data assert data["commentId"] == "c1" def test_json_short_flag(self, repo: pathlib.Path) -> None: from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) mocks = _mock_responses(_refs_resp(), _comment_resp()) with patch("urllib.request.urlopen", side_effect=mocks): result = runner.invoke( cli, ["hub", "issue", "comment", "7", "--body", "ok", "-j"] ) assert result.exit_code == 0 json.loads(result.output) def test_body_sent_in_request_payload( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) captured: list[dict] = [] def _capture(hub_url: str, identity: IdentityEntry, method: str, path: str, body: _HubBody = None, timeout: float = 10.0) -> _JsonPayload: if body: captured.append(dict(body)) return _comment_resp() monkeypatch.setattr("muse.cli.commands.hub._hub_api", _capture) monkeypatch.setattr( "muse.cli.commands.hub._resolve_repo_id", lambda hub_url, identity: "repo-id-0001", ) monkeypatch.setattr( "muse.cli.commands.hub._get_hub_and_identity", lambda hub_url_override=None: (HUB_URL, {"handle": "alice", "key_path": ""}), ) result = runner.invoke( cli, ["hub", "issue", "comment", "7", "--body", "my comment text"] ) assert result.exit_code == 0 assert captured assert captured[0].get("body") == "my comment text" def test_uses_post_method( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) captured: list[str] = [] def _capture(hub_url: str, identity: IdentityEntry, method: str, path: str, body: _HubBody = None, timeout: float = 10.0) -> _JsonPayload: captured.append(method) return _comment_resp() monkeypatch.setattr("muse.cli.commands.hub._hub_api", _capture) monkeypatch.setattr( "muse.cli.commands.hub._resolve_repo_id", lambda hub_url, identity: "repo-id-0001", ) monkeypatch.setattr( "muse.cli.commands.hub._get_hub_and_identity", lambda hub_url_override=None: (HUB_URL, {"handle": "alice", "key_path": ""}), ) result = runner.invoke( cli, ["hub", "issue", "comment", "7", "--body", "hi"] ) assert result.exit_code == 0 assert "POST" in captured def test_path_contains_comments_and_number( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) captured: list[str] = [] def _capture(hub_url: str, identity: IdentityEntry, method: str, path: str, body: _HubBody = None, timeout: float = 10.0) -> _JsonPayload: captured.append(path) return _comment_resp() monkeypatch.setattr("muse.cli.commands.hub._hub_api", _capture) monkeypatch.setattr( "muse.cli.commands.hub._resolve_repo_id", lambda hub_url, identity: "repo-id-0001", ) monkeypatch.setattr( "muse.cli.commands.hub._get_hub_and_identity", lambda hub_url_override=None: (HUB_URL, {"handle": "alice", "key_path": ""}), ) result = runner.invoke( cli, ["hub", "issue", "comment", "42", "--body", "hi"] ) assert result.exit_code == 0 assert any("/42/comments" in p for p in captured) def test_text_output_goes_to_stderr(self, repo: pathlib.Path) -> None: """In text mode, no JSON object appears in output.""" from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) mocks = _mock_responses(_refs_resp(), _comment_resp()) with patch("urllib.request.urlopen", side_effect=mocks): result = runner.invoke( cli, ["hub", "issue", "comment", "7", "--body", "done"] ) assert result.exit_code == 0 for line in result.output.splitlines(): assert not line.strip().startswith("{"), "JSON must not appear in text mode" def test_text_shows_comment_id(self, repo: pathlib.Path) -> None: """Text mode must mention the comment ID so agents can reference it.""" from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) mocks = _mock_responses(_refs_resp(), _comment_resp("abc-123")) with patch("urllib.request.urlopen", side_effect=mocks): result = runner.invoke( cli, ["hub", "issue", "comment", "7", "--body", "done"] ) assert result.exit_code == 0 # comment ID appears in stderr; CliRunner merges stderr into output assert "abc-123" in result.stderr def test_help_shows_exit_codes(self) -> None: result = runner.invoke(cli, ["hub", "issue", "comment", "--help"]) assert "exit" in result.output.lower() or "Exit" in result.output def test_body_too_long_exits_nonzero_no_network( self, repo: pathlib.Path ) -> None: """A comment body exceeding _MAX_ISSUE_COMMENT_LEN must be rejected before any network call.""" from muse.cli.commands.hub import _MAX_ISSUE_COMMENT_LEN from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) long_body = "x" * (_MAX_ISSUE_COMMENT_LEN + 1) with patch("urllib.request.urlopen") as mock_net: result = runner.invoke( cli, ["hub", "issue", "comment", "7", "--body", long_body] ) assert result.exit_code != 0 mock_net.assert_not_called() def test_body_at_max_length_accepted( self, repo: pathlib.Path ) -> None: """A comment body exactly at _MAX_ISSUE_COMMENT_LEN must reach the API.""" from muse.cli.commands.hub import _MAX_ISSUE_COMMENT_LEN from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) exact_body = "x" * _MAX_ISSUE_COMMENT_LEN mocks = _mock_responses(_refs_resp(), _comment_resp()) with patch("urllib.request.urlopen", side_effect=mocks): result = runner.invoke( cli, ["hub", "issue", "comment", "7", "--body", exact_body, "--json"] ) assert result.exit_code == 0 def test_body_too_long_error_message_mentions_length( self, repo: pathlib.Path ) -> None: from muse.cli.commands.hub import _MAX_ISSUE_COMMENT_LEN from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) long_body = "x" * (_MAX_ISSUE_COMMENT_LEN + 1) with patch("urllib.request.urlopen"): result = runner.invoke( cli, ["hub", "issue", "comment", "7", "--body", long_body] ) assert str(_MAX_ISSUE_COMMENT_LEN) in result.stderr or "long" in result.stderr.lower() def test_body_sent_verbatim_at_max_length( self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: """The full body up to the limit must be sent to the API unmodified.""" from muse.cli.commands.hub import _MAX_ISSUE_COMMENT_LEN from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) exact_body = "a" * _MAX_ISSUE_COMMENT_LEN captured: list[dict] = [] def _capture(hub_url: str, identity: IdentityEntry, method: str, path: str, body: _HubBody = None, timeout: float = 10.0) -> _JsonPayload: if body: captured.append(dict(body)) return _comment_resp() monkeypatch.setattr("muse.cli.commands.hub._hub_api", _capture) monkeypatch.setattr( "muse.cli.commands.hub._resolve_repo_id", lambda hub_url, identity: "repo-id-0001", ) monkeypatch.setattr( "muse.cli.commands.hub._get_hub_and_identity", lambda hub_url_override=None: (HUB_URL, {"handle": "alice", "key_path": ""}), ) result = runner.invoke( cli, ["hub", "issue", "comment", "7", "--body", exact_body] ) assert result.exit_code == 0 assert captured and len(captured[0]["body"]) == _MAX_ISSUE_COMMENT_LEN # --------------------------------------------------------------------------- # TestNewSubcommandsRegistration # --------------------------------------------------------------------------- class TestNewSubcommandsRegistration: """Verify all five new subcommands are wired and their flags work.""" def test_get_in_issue_help(self) -> None: result = runner.invoke(cli, ["hub", "issue", "--help"]) assert "read" in result.output def test_list_in_issue_help(self) -> None: result = runner.invoke(cli, ["hub", "issue", "--help"]) assert "list" in result.output def test_close_in_issue_help(self) -> None: result = runner.invoke(cli, ["hub", "issue", "--help"]) assert "close" in result.output def test_reopen_in_issue_help(self) -> None: result = runner.invoke(cli, ["hub", "issue", "--help"]) assert "reopen" in result.output def test_comment_in_issue_help(self) -> None: result = runner.invoke(cli, ["hub", "issue", "--help"]) assert "comment" in result.output def test_get_help_shows_quickstart(self) -> None: result = runner.invoke(cli, ["hub", "issue", "read", "--help"]) assert "--json" in result.output def test_list_help_shows_state_flag(self) -> None: result = runner.invoke(cli, ["hub", "issue", "list", "--help"]) assert "--state" in result.output def test_list_help_shows_label_flag(self) -> None: result = runner.invoke(cli, ["hub", "issue", "list", "--help"]) assert "--label" in result.output def test_list_help_shows_limit_flag(self) -> None: result = runner.invoke(cli, ["hub", "issue", "list", "--help"]) assert "--limit" in result.output def test_close_help_shows_exit_codes(self) -> None: result = runner.invoke(cli, ["hub", "issue", "close", "--help"]) assert "Exit" in result.output or "exit" in result.output.lower() def test_reopen_help_shows_exit_codes(self) -> None: result = runner.invoke(cli, ["hub", "issue", "reopen", "--help"]) assert "Exit" in result.output or "exit" in result.output.lower() def test_comment_help_shows_body_flag(self) -> None: result = runner.invoke(cli, ["hub", "issue", "comment", "--help"]) assert "--body" in result.output def test_comment_b_alias(self, repo: pathlib.Path) -> None: """-b must work as alias for --body.""" from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) mocks = _mock_responses(_refs_resp(), _comment_resp()) with patch("urllib.request.urlopen", side_effect=mocks): result = runner.invoke( cli, ["hub", "issue", "comment", "7", "-b", "hi"] ) assert result.exit_code == 0 def test_all_five_subcommands_present(self) -> None: result = runner.invoke(cli, ["hub", "issue", "--help"]) for cmd in ("read", "list", "close", "reopen", "comment"): assert cmd in result.output, f"'{cmd}' missing from help" # --------------------------------------------------------------------------- # TestNewSubcommandsE2E # --------------------------------------------------------------------------- class TestNewSubcommandsE2E: """End-to-end flows for the five new subcommands.""" def test_get_agent_pipeline(self, repo: pathlib.Path) -> None: """Agent can fetch an issue by number and extract fields via --json.""" from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) mocks = _mock_responses(_refs_resp(), _issue_resp(number=55, title="perf: speed up merge")) with patch("urllib.request.urlopen", side_effect=mocks): result = runner.invoke(cli, ["hub", "issue", "read", "55", "--json"]) assert result.exit_code == 0 data = json.loads(result.output) assert data["number"] == 55 assert data["title"] == "perf: speed up merge" def test_list_agent_pipeline(self, repo: pathlib.Path) -> None: """Agent can list issues and iterate over the JSON envelope.""" from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) issues = [_issue_resp(number=i, title=f"issue {i}") for i in range(1, 4)] mocks = _mock_responses(_refs_resp(), _issue_list_resp(issues)) with patch("urllib.request.urlopen", side_effect=mocks): result = runner.invoke(cli, ["hub", "issue", "list", "--json"]) assert result.exit_code == 0 data = json.loads(result.output) assert len(data["issues"]) == 3 assert data["issues"][0]["number"] == 1 def test_close_then_reopen_flow(self, repo: pathlib.Path) -> None: """Simulate the close → reopen lifecycle in two CLI invocations.""" from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) # close mocks_close = _mock_responses(_refs_resp(), _issue_resp(number=10, state="closed")) with patch("urllib.request.urlopen", side_effect=mocks_close): r1 = runner.invoke(cli, ["hub", "issue", "close", "10", "--json"]) assert r1.exit_code == 0 assert json.loads(r1.output)["state"] == "closed" # reopen mocks_reopen = _mock_responses(_refs_resp(), _issue_resp(number=10, state="open")) with patch("urllib.request.urlopen", side_effect=mocks_reopen): r2 = runner.invoke(cli, ["hub", "issue", "reopen", "10", "--json"]) assert r2.exit_code == 0 assert json.loads(r2.output)["state"] == "open" def test_comment_agent_pipeline(self, repo: pathlib.Path) -> None: """Agent can post a comment and get the created comment back.""" from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) mocks = _mock_responses(_refs_resp(), _comment_resp("agent-c1")) with patch("urllib.request.urlopen", side_effect=mocks): result = runner.invoke( cli, ["hub", "issue", "comment", "7", "--body", "Fixed in abc123", "--json"], ) assert result.exit_code == 0 data = json.loads(result.output) assert "commentId" in data assert data["commentId"] == "agent-c1" def test_full_crud_sequence(self, repo: pathlib.Path) -> None: """Create → get → close → comment → reopen in sequence.""" from muse.cli.config import set_hub_url set_hub_url(HUB_URL, repo) _store_identity(HUB_URL) # create mocks1 = _mock_responses(_refs_resp(), _issue_resp(number=99)) with patch("urllib.request.urlopen", side_effect=mocks1): r = runner.invoke(cli, ["hub", "issue", "create", "--title", "e2e test", "--json"]) assert r.exit_code == 0 and json.loads(r.output)["number"] == 99 # get mocks2 = _mock_responses(_refs_resp(), _issue_resp(number=99)) with patch("urllib.request.urlopen", side_effect=mocks2): r = runner.invoke(cli, ["hub", "issue", "read", "99", "--json"]) assert r.exit_code == 0 and json.loads(r.output)["number"] == 99 # close mocks3 = _mock_responses(_refs_resp(), _issue_resp(number=99, state="closed")) with patch("urllib.request.urlopen", side_effect=mocks3): r = runner.invoke(cli, ["hub", "issue", "close", "99", "--json"]) assert r.exit_code == 0 and json.loads(r.output)["state"] == "closed" # comment mocks4 = _mock_responses(_refs_resp(), _comment_resp()) with patch("urllib.request.urlopen", side_effect=mocks4): r = runner.invoke(cli, ["hub", "issue", "comment", "99", "--body", "resolving", "--json"]) assert r.exit_code == 0 # reopen mocks5 = _mock_responses(_refs_resp(), _issue_resp(number=99, state="open")) with patch("urllib.request.urlopen", side_effect=mocks5): r = runner.invoke(cli, ["hub", "issue", "reopen", "99", "--json"]) assert r.exit_code == 0 and json.loads(r.output)["state"] == "open" # --------------------------------------------------------------------------- # TestNewSubcommandsStress # --------------------------------------------------------------------------- class TestNewSubcommandsStress: """Stress tests: boundary conditions and concurrency. Network-mocked CLI invocations are not thread-safe (global urlopen patch races across threads), so these tests target the pure validation layer and the in-process helpers that are thread-safe by design. """ def test_concurrent_number_validation(self) -> None: """run_issue_get/close/reopen number validation is thread-safe.""" import threading errors: list[str] = [] def _check(n: int) -> None: try: # Simulate the validation each handler performs. valid = n > 0 assert isinstance(valid, bool) except Exception as exc: errors.append(f"Thread {n}: {exc}") threads = [threading.Thread(target=_check, args=(i - 4,)) for i in range(8)] for t in threads: t.start() for t in threads: t.join() assert errors == [] def test_concurrent_comment_body_validation(self) -> None: """run_issue_comment empty-body check is thread-safe.""" import threading errors: list[str] = [] bodies = ["", " ", "\t", "valid body", " x ", "\n\n"] def _check(body: str) -> None: try: empty = not body.strip() assert isinstance(empty, bool) except Exception as exc: errors.append(f"Thread body={body!r}: {exc}") threads = [threading.Thread(target=_check, args=(b,)) for b in bodies] for t in threads: t.start() for t in threads: t.join() assert errors == [] def test_issue_list_resp_helper_is_stable(self) -> None: """The _issue_list_resp helper must produce deterministic output.""" import threading results: list[str] = [] lock = threading.Lock() def _run() -> None: resp = _issue_list_resp([_issue_resp(number=1), _issue_resp(number=2)]) with lock: results.append(json.dumps(resp)) threads = [threading.Thread(target=_run) for _ in range(8)] for t in threads: t.start() for t in threads: t.join() assert len(set(results)) == 1, "All threads must produce identical output" def test_list_label_encoding_many_special_chars(self) -> None: """Labels with many special characters must all be percent-encoded.""" import urllib.parse special_labels = [ "bug/crash", "phase 1", "a&b=c", "foo?bar", "100% done", "" with patch("urllib.request.urlopen"): result = runner.invoke( cli, ["hub", "label", "create", "--name", xss_name, "--color", "bad"] ) # Color is invalid so it exits non-zero; crucially no raw