gabriel/muse — blame/dev/test_cmd_hub_hardening.py

1 files

1 commits

0 hotspots

0 🧊 dead

0 💥 blast risk

sha256:e Merge 'bump/muse-v0.2.1' into 'dev' — proposal: chore: bump muse to v0.2.1 · gabriel · Aug 28, 2026

1	"""Comprehensive hardening tests for ``muse hub``.
2
3	Coverage
4	--------
5	Unit
6	- _normalise_url: scheme injection (file://, ftp://), http non-loopback rejected,
7	loopback allowed, scheme-less normalised to https, trailing slash stripped
8	- _hub_hostname: standard URL, URL with port, URL with path, bare hostname
9	- _ping_hub: reachable, HTTP error, URLError, timeout, redirect refused
10	- _hub_api: file:// scheme blocked, response size cap, error detail sanitized,
11	missing token exits, None-value payload keys stripped
12	- _resolve_proposal_id: full UUID passthrough, prefix match, no match, ambiguous match
13	- _format_proposal: all fields sanitized
14
15	Integration (CliRunner + mock hub)
16	- run_connect: --json schema, re-connect warns, normalisation, invalid scheme exits
17	- run_disconnect: --json ok, --json nothing_to_do, text mode to stderr
18	- run_status: --json all keys present, no-hub exits, not-authenticated exits
19	- run_ping: --json ok, --json error, text mode to stderr, unreachable exits nonzero
20	- run_proposal_list: --json is a JSON array, text mode to stderr, no-proposals message
21	- run_proposal_create: --json schema, missing branch exits, sanitizes output
22	- run_proposal_merge: --json schema, merge=false exits nonzero
23	- run_proposal_show: --json passthrough
24
25	Security
26	- file:// hub URL blocked in _hub_api before network
27	- ANSI in proposal title/branch sanitized in _format_proposal
28	- ANSI in proposal ID sanitized in _resolve_proposal_id errors
29	- hub URL in errors sanitized in _get_hub_and_identity
30	- Response body size cap prevents OOM
31
32	E2E (via CliRunner)
33	- connect --json schema includes all required keys
34	- disconnect --json schema correct for both ok and nothing_to_do
35	- ping --json schema with all required keys
36	- status --json all keys always present (no missing keys when not authenticated)
37
38	Stress
39	- 8 concurrent ping checks against isolated mock responses
40	"""
41
42	from __future__ import annotations
43
44	import json
45	import pathlib
46	import threading
47	import unittest.mock
48	import urllib.error
49	import ssl
50	import urllib.request
51	from collections.abc import Mapping
52	from typing import TYPE_CHECKING
53	from unittest.mock import MagicMock, patch
54
55	import pytest
56
57	from tests.cli_test_helper import CliRunner, InvokeResult
58
59	if TYPE_CHECKING:
60	pass
61
62	from muse.cli.commands.hub.connection import (
63	_ConnectJson,
64	_DisconnectJson,
65	_PingJson,
66	_StatusJson,
67	)
68	from muse.core.types import Manifest, MsgpackDict, MsgpackValue
69	from muse.core.identity import IdentityEntry
70	from muse.core.paths import head_path, heads_dir, muse_dir
71
72	type _JsonPayload = MsgpackDict
73	type _ProposalRecord = dict[str, str]
74	type _RepoResponse = dict[str, str]
75	type _HubBody = Mapping[str, str \| bool \| list[str] \| None] \| None
76	cli = None
77	runner = CliRunner()
78
79	# ── helpers ───────────────────────────────────────────────────────────────────
80
81
82	def _json_line(result: InvokeResult) -> _JsonPayload:
83	for line in result.output.splitlines():
84	stripped = line.strip()
85	if stripped.startswith("{") or stripped.startswith("["):
86	data: _JsonPayload = json.loads(stripped)
87	return data
88	raise ValueError(f"No JSON line in output:\n{result.output!r}")
89
90
91	def _json_connect(result: InvokeResult) -> _ConnectJson:
92	d: _ConnectJson = json.loads(
93	next(l for l in result.output.splitlines() if l.strip().startswith("{"))
94	)
95	return d
96
97
98	def _json_status(result: InvokeResult) -> _StatusJson:
99	d: _StatusJson = json.loads(
100	next(l for l in result.output.splitlines() if l.strip().startswith("{"))
101	)
102	return d
103
104
105	def _json_disconnect(result: InvokeResult) -> _DisconnectJson:
106	d: _DisconnectJson = json.loads(
107	next(l for l in result.output.splitlines() if l.strip().startswith("{"))
108	)
109	return d
110
111
112	def _json_ping(result: InvokeResult) -> _PingJson:
113	d: _PingJson = json.loads(
114	next(l for l in result.output.splitlines() if l.strip().startswith("{"))
115	)
116	return d
117
118
119	@pytest.fixture
120	def repo(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path:
121	"""Minimal .muse/ repo with identity file."""
122	from muse._version import __version__
123
124	dot_muse = muse_dir(tmp_path)
125	for sub in ("refs/heads", "objects", "commits", "snapshots"):
126	(dot_muse / sub).mkdir(parents=True, exist_ok=True)
127	(dot_muse / "repo.json").write_text(
128	json.dumps({"repo_id": "test-repo", "schema_version": __version__, "domain": "code"})
129	)
130	(dot_muse / "HEAD").write_text("ref: refs/heads/main\n")
131	(dot_muse / "refs" / "heads" / "main").write_text("")
132	(dot_muse / "config.toml").write_text("")
133	muse_home = tmp_path / ".muse-home"
134	muse_home.mkdir()
135	(muse_home / "identity.toml").write_text("")
136	monkeypatch.setattr("muse.core.identity._IDENTITY_FILE", muse_home / "identity.toml")
137	monkeypatch.setattr("muse.core.identity._IDENTITY_DIR", muse_home)
138	monkeypatch.chdir(tmp_path)
139	return tmp_path
140
141
142	def _make_signing() -> "SigningIdentity":
143	"""Generate a fresh Ed25519 SigningIdentity for tests."""
144	from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
145	from muse.core.transport import SigningIdentity
146	return SigningIdentity(handle="testuser", private_key=Ed25519PrivateKey.generate())
147
148
149	def _store_identity(hub_url: str, handle: str = "alice") -> None:
150	"""Save a human identity entry with hd_path + keychain mnemonic."""
151	from muse.core.identity import IdentityEntry, save_identity
152	from muse.core.hdkeys import muse_path, DOMAIN_IDENTITY, ENTITY_HUMAN, ROLE_SIGN
153
154	hd_path = muse_path(DOMAIN_IDENTITY, ENTITY_HUMAN, 0)
155	entry: IdentityEntry = {
156	"type": "human",
157	"handle": handle,
158	"algorithm": "ed25519",
159	"fingerprint": f"test-fp-{handle}",
160	"hd_path": hd_path,
161	}
162	# Use a fixed test mnemonic stored in the keychain so resolve_signing_identity works.
163	_TEST_MNEMONIC = (
164	"abandon abandon abandon abandon abandon abandon abandon abandon "
165	"abandon abandon abandon about"
166	)
167	save_identity(hub_url, entry, mnemonic=_TEST_MNEMONIC)
168
169
170	# ── Unit: _normalise_url ──────────────────────────────────────────────────────
171
172
173	class TestNormaliseUrlHardening:
174	def test_file_scheme_raises(self) -> None:
175	from muse.cli.commands.hub import _normalise_url
176	with pytest.raises(ValueError, match="not allowed"):
177	_normalise_url("file:///etc/passwd")
178
179	def test_ftp_scheme_raises(self) -> None:
180	from muse.cli.commands.hub import _normalise_url
181	with pytest.raises(ValueError, match="not allowed"):
182	_normalise_url("ftp://attacker.example.com/repo")
183
184	def test_data_scheme_raises(self) -> None:
185	from muse.cli.commands.hub import _normalise_url
186	with pytest.raises(ValueError, match="not allowed"):
187	_normalise_url("data:text/plain,malicious")
188
189	def test_http_non_loopback_raises(self) -> None:
190	from muse.cli.commands.hub import _normalise_url
191	with pytest.raises(ValueError, match="HTTPS"):
192	_normalise_url("http://musehub.ai/gabriel/muse")
193
194	def test_http_localhost_allowed(self) -> None:
195	from muse.cli.commands.hub import _normalise_url
196	assert _normalise_url("https://localhost:1337") == "https://localhost:1337"
197
198	def test_http_127_allowed(self) -> None:
199	from muse.cli.commands.hub import _normalise_url
200	assert _normalise_url("http://127.0.0.1:9000") == "http://127.0.0.1:9000"
201
202	def test_schemeless_becomes_https(self) -> None:
203	from muse.cli.commands.hub import _normalise_url
204	assert _normalise_url("musehub.ai").startswith("https://")
205
206	def test_trailing_slash_stripped(self) -> None:
207	from muse.cli.commands.hub import _normalise_url
208	assert not _normalise_url("https://musehub.ai/").endswith("/")
209
210	def test_https_passthrough(self) -> None:
211	from muse.cli.commands.hub import _normalise_url
212	assert _normalise_url("https://musehub.ai/gabriel/muse") == "https://musehub.ai/gabriel/muse"
213
214
215	# ── Unit: _hub_hostname ───────────────────────────────────────────────────────
216
217
218	class TestHubHostname:
219	def test_plain_https(self) -> None:
220	from muse.cli.commands.hub.connection import _hub_hostname
221	assert _hub_hostname("https://musehub.ai/gabriel/muse") == "musehub.ai"
222
223	def test_with_port(self) -> None:
224	from muse.cli.commands.hub.connection import _hub_hostname
225	assert _hub_hostname("https://localhost:1337/gabriel/muse") == "localhost:1337"
226
227	def test_bare_hostname(self) -> None:
228	from muse.cli.commands.hub.connection import _hub_hostname
229	assert _hub_hostname("musehub.ai") == "musehub.ai"
230
231	def test_trailing_slash(self) -> None:
232	from muse.cli.commands.hub.connection import _hub_hostname
233	assert _hub_hostname("https://musehub.ai/") == "musehub.ai"
234
235
236	# ── Unit: _hub_api ────────────────────────────────────────────────────────────
237
238
239	class TestHubApi:
240	_IDENTITY = {"type": "human", "token": "tok123"}
241
242	def test_file_scheme_blocked_before_network(self) -> None:
243	from muse.cli.commands.hub import _hub_api
244	from muse.core.identity import IdentityEntry
245	identity: IdentityEntry = {"type": "human", "token": "tok"}
246	with patch("urllib.request.urlopen") as mock_net:
247	with pytest.raises(SystemExit):
248	_hub_api("file:///etc/passwd", identity, "GET", "/api/test")
249	mock_net.assert_not_called()
250
251	def test_ftp_scheme_blocked_before_network(self) -> None:
252	from muse.cli.commands.hub import _hub_api
253	from muse.core.identity import IdentityEntry
254	identity: IdentityEntry = {"type": "human", "token": "tok"}
255	with patch("urllib.request.urlopen") as mock_net:
256	with pytest.raises(SystemExit):
257	_hub_api("ftp://ftp.example.com", identity, "GET", "/api/test")
258	mock_net.assert_not_called()
259
260	def test_missing_token_exits(self) -> None:
261	"""No signing identity on a mutating method → SystemExit before any network I/O.
262
263	GET requests on public resources are allowed without auth; POST/PUT/DELETE/PATCH
264	still require a signing identity. identity carries no handle/key_path so
265	Ed25519 key loading is skipped. get_signing_identity is patched to return None.
266	"""
267	from muse.cli.commands.hub import _hub_api
268	from muse.core.identity import IdentityEntry
269	identity: IdentityEntry = {"type": "human"}
270	with patch("muse.cli.config.get_signing_identity", return_value=None):
271	with patch("urllib.request.urlopen") as mock_net:
272	with pytest.raises(SystemExit):
273	_hub_api("https://localhost:1337", identity, "POST", "/api/test")
274	mock_net.assert_not_called()
275
276	def test_response_size_cap(self) -> None:
277	from muse.cli.commands.hub import _MAX_API_RESPONSE_BYTES, _hub_api
278	from muse.core.identity import IdentityEntry
279	identity: IdentityEntry = {"type": "human", "token": "tok"}
280	mock_resp = MagicMock()
281	mock_resp.__enter__ = lambda s: s
282	mock_resp.__exit__ = MagicMock(return_value=False)
283	mock_resp.read.return_value = b"x" * (_MAX_API_RESPONSE_BYTES + 2)
284	with patch("muse.cli.config.get_signing_identity", return_value=_make_signing()):
285	with patch("urllib.request.urlopen", return_value=mock_resp):
286	with pytest.raises(SystemExit):
287	_hub_api("http://localhost:9999", identity, "GET", "/api/test")
288
289	def test_http_error_sanitized_in_output(
290	self, capsys: pytest.CaptureFixture[str]
291	) -> None:
292	import urllib.error
293	from muse.cli.commands.hub import _hub_api
294	from muse.core.identity import IdentityEntry
295
296	import io
297	identity: IdentityEntry = {"type": "human", "token": "tok"}
298	ansi_detail = b'{"detail":"\\x1b[31mmalicious\\x1b[0m"}'
299	exc = urllib.error.HTTPError(url="", code=403, msg="Forbidden", hdrs=MagicMock(), fp=io.BytesIO(ansi_detail))
300
301	with patch("muse.cli.config.get_signing_identity", return_value=_make_signing()):
302	with patch("urllib.request.urlopen", side_effect=exc):
303	with pytest.raises(SystemExit):
304	_hub_api("http://localhost:9999", identity, "GET", "/api/test")
305
306	captured = capsys.readouterr()
307	assert "\x1b[" not in captured.err
308
309	def test_empty_response_returns_empty_dict(self) -> None:
310	from muse.cli.commands.hub import _hub_api
311	from muse.core.identity import IdentityEntry
312
313	identity: IdentityEntry = {"type": "human", "token": "tok"}
314	mock_resp = MagicMock()
315	mock_resp.__enter__ = lambda s: s
316	mock_resp.__exit__ = MagicMock(return_value=False)
317	mock_resp.read.return_value = b""
318	with patch("muse.cli.config.get_signing_identity", return_value=_make_signing()):
319	with patch("urllib.request.urlopen", return_value=mock_resp):
320	result = _hub_api("http://localhost:9999", identity, "GET", "/api/test")
321	assert result == {}
322
323
324	# ── TDD: PEM-load path removed from _hub_api (H1) ───────────────────────────
325
326
327	class TestHubApiPemLoadRemoved:
328	"""H1: _hub_api must use get_signing_identity exclusively — no PEM reads.
329
330	Before fix: identity.get("key_path") was used as primary signing path;
331	load_pem_private_key was called if the file existed.
332	After fix: get_signing_identity(remote_url=...) is the only signing path.
333	"""
334
335	def test_H1_load_pem_private_key_never_called_even_with_key_path(
336	self, tmp_path: pathlib.Path
337	) -> None:
338	"""Even when identity contains key_path pointing to a real file,
339	load_pem_private_key must NOT be called — only get_signing_identity."""
340	from muse.cli.commands.hub import _hub_api
341	from muse.core.identity import IdentityEntry
342	from unittest.mock import MagicMock, patch, call
343
344	# Create a fake PEM file so is_file() returns True
345	fake_pem = tmp_path / "fake.pem"
346	fake_pem.write_bytes(b"-----BEGIN PRIVATE KEY-----\ngarbage\n-----END PRIVATE KEY-----\n")
347
348	identity: IdentityEntry = {
349	"type": "human",
350	"handle": "alice",
351	"key_path": str(fake_pem),
352	}
353
354	mock_resp = MagicMock()
355	mock_resp.__enter__ = lambda s: s
356	mock_resp.__exit__ = MagicMock(return_value=False)
357	mock_resp.read.return_value = b"{}"
358
359	mock_load_pem = MagicMock(return_value=MagicMock())
360
361	with (
362	patch("muse.cli.config.get_signing_identity", return_value=_make_signing()) as mock_gsi,
363	patch("urllib.request.urlopen", return_value=mock_resp),
364	patch("cryptography.hazmat.primitives.serialization.load_pem_private_key", mock_load_pem),
365	):
366	_hub_api("http://localhost:9999", identity, "GET", "/api/test")
367
368	mock_load_pem.assert_not_called()
369	mock_gsi.assert_called_once()
370
371
372	# ── Unit: _resolve_proposal_id ──────────────────────────────────────────────────────
373
374
375	class TestResolveProposalId:
376	def _make_identity(self) -> "muse.core.identity.IdentityEntry":
377	from muse.core.identity import IdentityEntry
378	e: IdentityEntry = {"type": "human", "token": "tok123"}
379	return e
380
381	def _proposal(self, proposal_id: str, title: str = "Test Proposal") -> _ProposalRecord:
382	return {"proposalId": proposal_id, "title": title, "state": "open",
383	"fromBranch": "feat/x", "toBranch": "dev"}
384
385	def test_full_id_returned_as_is(self) -> None:
386	from muse.cli.commands.hub import _resolve_proposal_id
387	full = "af54753d-1234-5678-abcd-ef1234567890"
388	result = _resolve_proposal_id("http://hub", self._make_identity(), "repo-id", full)
389	assert result == full
390
391	def test_prefix_resolved(self) -> None:
392	from muse.cli.commands.hub import _resolve_proposal_id
393
394	proposal_id = "abc12345-6789-0000-0000-000000000000"
395	proposals_resp = {"proposals": [self._proposal(proposal_id)]}
396	mock_resp = MagicMock()
397	mock_resp.__enter__ = lambda s: s
398	mock_resp.__exit__ = MagicMock(return_value=False)
399	mock_resp.read.return_value = json.dumps(proposals_resp).encode()
400	with patch("muse.cli.config.get_signing_identity", return_value=_make_signing()):
401	with patch("urllib.request.urlopen", return_value=mock_resp):
402	result = _resolve_proposal_id(
403	"http://localhost:9999", self._make_identity(), "repo-id", "abc12345"
404	)
405	assert result == proposal_id
406
407	def test_no_match_exits(self) -> None:
408	from muse.cli.commands.hub import _resolve_proposal_id
409
410	resp_bytes = json.dumps({"proposals": []}).encode()
411	mock_resp = MagicMock()
412	mock_resp.__enter__ = lambda s: s
413	mock_resp.__exit__ = MagicMock(return_value=False)
414	mock_resp.read.return_value = resp_bytes
415	with patch("muse.cli.config.get_signing_identity", return_value=_make_signing()):
416	with patch("urllib.request.urlopen", return_value=mock_resp):
417	with pytest.raises(SystemExit):
418	_resolve_proposal_id(
419	"http://localhost:9999", self._make_identity(), "repo-id", "deadbeef"
420	)
421
422	def test_ambiguous_prefix_exits(self) -> None:
423	from muse.cli.commands.hub import _resolve_proposal_id
424
425	pr1_id = "abc12345-0000-0000-0000-000000000001"
426	pr2_id = "abc12345-0000-0000-0000-000000000002"
427	proposals_resp = {"proposals": [self._proposal(pr1_id), self._proposal(pr2_id)]}
428	mock_resp = MagicMock()
429	mock_resp.__enter__ = lambda s: s
430	mock_resp.__exit__ = MagicMock(return_value=False)
431	mock_resp.read.return_value = json.dumps(proposals_resp).encode()
432	with patch("muse.cli.config.get_signing_identity", return_value=_make_signing()):
433	with patch("urllib.request.urlopen", return_value=mock_resp):
434	with pytest.raises(SystemExit):
435	_resolve_proposal_id(
436	"http://localhost:9999", self._make_identity(), "repo-id", "abc12345"
437	)
438
439	def test_ansi_in_proposal_id_sanitized_in_error(
440	self, capsys: pytest.CaptureFixture[str]
441	) -> None:
442	from muse.cli.commands.hub import _resolve_proposal_id
443
444	resp_bytes = json.dumps({"proposals": []}).encode()
445	mock_resp = MagicMock()
446	mock_resp.__enter__ = lambda s: s
447	mock_resp.__exit__ = MagicMock(return_value=False)
448	mock_resp.read.return_value = resp_bytes
449	malicious_proposalefix = "\x1b[31mmalicious\x1b[0m"
450	with patch("muse.cli.config.get_signing_identity", return_value=_make_signing()):
451	with patch("urllib.request.urlopen", return_value=mock_resp):
452	with pytest.raises(SystemExit):
453	_resolve_proposal_id(
454	"http://localhost:9999", self._make_identity(), "repo-id", malicious_proposalefix
455	)
456	captured = capsys.readouterr()
457	assert "\x1b[" not in captured.err
458
459
460	# ── Unit: _format_proposal ──────────────────────────────────────────────────────────
461
462
463	class TestFormatProposal:
464	def test_ansi_in_title_stripped(self) -> None:
465	from muse.cli.commands.hub import _format_proposal
466	proposal: _ProposalRecord = {
467	"proposalId": "abc12345",
468	"title": "\x1b[31mmalicious title\x1b[0m",
469	"state": "open",
470	"fromBranch": "feat/x",
471	"toBranch": "dev",
472	}
473	result = _format_proposal(proposal)
474	assert "\x1b[" not in result
475
476	def test_ansi_in_branch_stripped(self) -> None:
477	from muse.cli.commands.hub import _format_proposal
478	proposal: _ProposalRecord = {
479	"proposalId": "abc12345",
480	"title": "clean title",
481	"state": "open",
482	"fromBranch": "\x1b[32mfeat/malicious\x1b[0m",
483	"toBranch": "\x1b[31mdev\x1b[0m",
484	}
485	result = _format_proposal(proposal)
486	assert "\x1b[" not in result
487
488	def test_state_icon_open(self) -> None:
489	from muse.cli.commands.hub import _format_proposal
490	proposal: _ProposalRecord = {
491	"proposalId": "abc12345", "title": "t", "state": "open",
492	"fromBranch": "f", "toBranch": "d",
493	}
494	assert "🟢" in _format_proposal(proposal)
495
496	def test_state_icon_merged(self) -> None:
497	from muse.cli.commands.hub import _format_proposal
498	proposal: _ProposalRecord = {
499	"proposalId": "abc12345", "title": "t", "state": "merged",
500	"fromBranch": "f", "toBranch": "d",
501	}
502	assert "🟣" in _format_proposal(proposal)
503
504
505	# ── Integration: run_connect ──────────────────────────────────────────────────
506
507
508	class TestConnectHardening:
509	_HUB = "http://localhost:19999"
510
511	def test_connect_json_schema(self, repo: pathlib.Path) -> None:
512	result = runner.invoke(cli, ["hub", "connect", self._HUB, "--json"])
513	assert result.exit_code == 0
514	data = _json_connect(result)
515	for key in ("status", "hub_url", "hostname", "authenticated",
516	"identity_name", "identity_type"):
517	assert key in data, f"Missing key: {key}"
518	assert data["status"] == "ok"
519	assert data["authenticated"] is False
520	assert data["identity_name"] == ""
521	assert data["identity_type"] == ""
522
523	def test_connect_authenticated_json_schema(self, repo: pathlib.Path) -> None:
524	_store_identity(self._HUB)
525	result = runner.invoke(cli, ["hub", "connect", self._HUB, "--json"])
526	assert result.exit_code == 0
527	data = _json_connect(result)
528	assert data["authenticated"] is True
529	assert data["identity_name"] == "alice"
530	assert data["identity_type"] == "human"
531
532	def test_connect_invalid_scheme_exits(self, repo: pathlib.Path) -> None:
533	result = runner.invoke(cli, ["hub", "connect", "file:///etc/passwd"])
534	assert result.exit_code != 0
535
536	def test_connect_http_non_loopback_exits(self, repo: pathlib.Path) -> None:
537	result = runner.invoke(cli, ["hub", "connect", "http://musehub.ai"])
538	assert result.exit_code != 0
539
540	def test_connect_json_stdout_clean(self, repo: pathlib.Path) -> None:
541	result = runner.invoke(cli, ["hub", "connect", self._HUB, "--json"])
542	assert result.exit_code == 0
543	json_lines = [l for l in result.output.splitlines() if l.strip().startswith("{")]
544	assert len(json_lines) >= 1
545
546	def test_connect_no_repo_exits(self, tmp_path: pathlib.Path,
547	monkeypatch: pytest.MonkeyPatch) -> None:
548	monkeypatch.chdir(tmp_path)
549	result = runner.invoke(cli, ["hub", "connect", self._HUB])
550	assert result.exit_code != 0
551
552	def test_reconnect_warning_on_stderr(self, repo: pathlib.Path) -> None:
553	runner.invoke(cli, ["hub", "connect", self._HUB])
554	result = runner.invoke(cli, ["hub", "connect", "http://localhost:20000"])
555	assert result.exit_code == 0
556	assert "localhost:19999" in result.stderr
557
558	def test_reconnect_same_url_no_warning(self, repo: pathlib.Path) -> None:
559	"""Re-connecting to the same URL is a no-op — no warning emitted."""
560	runner.invoke(cli, ["hub", "connect", self._HUB])
561	result = runner.invoke(cli, ["hub", "connect", self._HUB])
562	assert result.exit_code == 0
563	assert "Switching" not in result.stderr
564	assert "⚠️" not in result.stderr
565
566	def test_connect_short_flag_j(self, repo: pathlib.Path) -> None:
567	"""-j short flag produces identical JSON output to --json."""
568	r_long = runner.invoke(cli, ["hub", "connect", self._HUB, "--json"])
569	runner.invoke(cli, ["hub", "disconnect"])
570	r_short = runner.invoke(cli, ["hub", "connect", self._HUB, "-j"])
571	assert r_long.exit_code == 0
572	assert r_short.exit_code == 0
573	d_long = _json_connect(r_long)
574	d_short = _json_connect(r_short)
575	assert d_long == d_short
576
577	def test_connect_ipv6_loopback_accepted(self, repo: pathlib.Path) -> None:
578	"""http://[::1] and http://[::1]:PORT are valid loopback URLs."""
579	result = runner.invoke(cli, ["hub", "connect", "http://[::1]:8080", "--json"])
580	assert result.exit_code == 0
581	data = _json_connect(result)
582	assert data["status"] == "ok"
583	assert "::1" in data["hub_url"]
584
585	def test_connect_ipv6_loopback_bare_accepted(self, repo: pathlib.Path) -> None:
586	"""http://[::1] without a port is valid."""
587	result = runner.invoke(cli, ["hub", "connect", "http://[::1]", "--json"])
588	assert result.exit_code == 0
589	data = _json_connect(result)
590	assert data["status"] == "ok"
591
592	def test_connect_bare_hostname_with_port(self, repo: pathlib.Path) -> None:
593	"""musehub.ai:8443 (no scheme) is promoted to https://musehub.ai:8443."""
594	result = runner.invoke(cli, ["hub", "connect", "musehub.ai:8443", "--json"])
595	assert result.exit_code == 0
596	data = _json_connect(result)
597	assert data["hub_url"] == "https://musehub.ai:8443"
598	assert data["hostname"] == "musehub.ai:8443"
599
600	def test_connect_trailing_slash_stripped(self, repo: pathlib.Path) -> None:
601	"""Trailing slashes are stripped from the stored URL."""
602	result = runner.invoke(
603	cli, ["hub", "connect", "https://musehub.ai/", "--json"]
604	)
605	assert result.exit_code == 0
606	data = _json_connect(result)
607	assert not data["hub_url"].endswith("/")
608
609	def test_connect_ansi_in_reconnect_warning_sanitized(
610	self, repo: pathlib.Path
611	) -> None:
612	"""ANSI codes stored in config are stripped from the reconnect warning."""
613	import unittest.mock
614	ansi_url = "https://\x1b[31mattacker.example.com\x1b[0m"
615	with unittest.mock.patch(
616	"muse.cli.commands.hub.get_hub_url", return_value=ansi_url
617	):
618	result = runner.invoke(
619	cli, ["hub", "connect", "https://safe.example.com"]
620	)
621	assert "\x1b" not in result.stderr, "ANSI escape leaked into reconnect warning"
622
623	def test_connect_json_hub_url_normalised(self, repo: pathlib.Path) -> None:
624	"""hub_url in JSON is the normalised form (no trailing slash, has scheme)."""
625	result = runner.invoke(
626	cli, ["hub", "connect", "musehub.ai", "--json"]
627	)
628	assert result.exit_code == 0
629	data = _json_connect(result)
630	assert data["hub_url"].startswith("https://")
631	assert not data["hub_url"].endswith("/")
632
633	def test_connect_no_repo_exits_2(self, tmp_path: pathlib.Path,
634	monkeypatch: pytest.MonkeyPatch) -> None:
635	"""Exit code 2 (REPO_NOT_FOUND) when outside a Muse repo."""
636	monkeypatch.chdir(tmp_path)
637	result = runner.invoke(cli, ["hub", "connect", self._HUB])
638	assert result.exit_code == 2
639
640	def test_connect_http_non_loopback_exits_1(self, repo: pathlib.Path) -> None:
641	"""Exit code 1 (USER_ERROR) for http:// non-loopback URL."""
642	result = runner.invoke(cli, ["hub", "connect", "http://remote.example.com"])
643	assert result.exit_code == 1
644
645	def test_connect_disallowed_scheme_exits_1(self, repo: pathlib.Path) -> None:
646	"""Exit code 1 (USER_ERROR) for ftp:// URL."""
647	result = runner.invoke(cli, ["hub", "connect", "ftp://musehub.ai"])
648	assert result.exit_code == 1
649
650	def test_10_sequential_connects_all_survive(self, repo: pathlib.Path) -> None:
651	"""10 sequential connect→disconnect cycles all succeed."""
652	for i in range(10):
653	hub = f"http://localhost:{19000 + i}"
654	r = runner.invoke(cli, ["hub", "connect", hub, "--json"])
655	assert r.exit_code == 0, f"connect {i} failed: {r.output}"
656	data = _json_connect(r)
657	assert data["status"] == "ok"
658
659
660	# ── Integration: run_status ───────────────────────────────────────────────────
661
662
663	class TestStatusHardening:
664	_HUB = "http://localhost:19999"
665
666	def test_status_no_hub_exits_nonzero(self, repo: pathlib.Path) -> None:
667	result = runner.invoke(cli, ["hub", "status", "--json"])
668	assert result.exit_code != 0
669
670	def test_status_json_all_keys_always_present(self, repo: pathlib.Path) -> None:
671	"""All 7 JSON keys present even when not authenticated."""
672	runner.invoke(cli, ["hub", "connect", self._HUB])
673	result = runner.invoke(cli, ["hub", "status", "--json"])
674	assert result.exit_code == 0
675	data = _json_status(result)
676	for key in ("hub_url", "hostname", "authenticated", "identity_type",
677	"identity_name", "identity_id", "capabilities"):
678	assert key in data, f"Missing key: {key}"
679	assert data["authenticated"] is False
680	assert data["identity_type"] == ""
681	assert data["identity_name"] == ""
682	assert data["identity_id"] == ""
683	assert data["capabilities"] == []
684
685	def test_status_json_authenticated(self, repo: pathlib.Path) -> None:
686	runner.invoke(cli, ["hub", "connect", self._HUB])
687	_store_identity(self._HUB)
688	result = runner.invoke(cli, ["hub", "status", "--json"])
689	assert result.exit_code == 0
690	data = _json_status(result)
691	assert data["authenticated"] is True
692	assert data["identity_name"] == "alice"
693	assert data["identity_type"] == "human"
694
695	def test_status_json_stdout_clean(self, repo: pathlib.Path) -> None:
696	runner.invoke(cli, ["hub", "connect", self._HUB])
697	result = runner.invoke(cli, ["hub", "status", "--json"])
698	json_lines = [l for l in result.output.splitlines() if l.strip().startswith("{")]
699	assert len(json_lines) >= 1
700
701	def test_status_text_mode_to_stderr(self, repo: pathlib.Path,
702	capsys: pytest.CaptureFixture[str]) -> None:
703	runner.invoke(cli, ["hub", "connect", self._HUB])
704	result = runner.invoke(cli, ["hub", "status"])
705	assert result.exit_code == 0
706
707	def test_status_short_flag_j(self, repo: pathlib.Path) -> None:
708	"""-j short flag produces identical JSON output to --json."""
709	runner.invoke(cli, ["hub", "connect", self._HUB])
710	r_long = runner.invoke(cli, ["hub", "status", "--json"])
711	r_short = runner.invoke(cli, ["hub", "status", "-j"])
712	assert r_long.exit_code == 0
713	assert r_short.exit_code == 0
714	assert json.loads(r_long.output) == json.loads(r_short.output)
715
716	def test_status_exit_code_2_no_repo(
717	self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
718	) -> None:
719	"""Exit code 2 (REPO_NOT_FOUND) when outside a Muse repo."""
720	monkeypatch.chdir(tmp_path)
721	result = runner.invoke(cli, ["hub", "status"])
722	assert result.exit_code == 2
723
724	def test_status_exit_code_1_no_hub(self, repo: pathlib.Path) -> None:
725	"""Exit code 1 (USER_ERROR) when no hub is connected."""
726	result = runner.invoke(cli, ["hub", "status"])
727	assert result.exit_code == 1
728
729	def test_status_hub_override_flag(self, repo: pathlib.Path) -> None:
730	"""--hub flag overrides config; identity is looked up for that URL."""
731	override = "http://localhost:29999"
732	from muse.core.identity import IdentityEntry, save_identity
733	entry: IdentityEntry = {
734	"type": "agent", "handle": "override-bot",
735	}
736	save_identity(override, entry)
737	# Connect to a different hub
738	runner.invoke(cli, ["hub", "connect", self._HUB])
739	result = runner.invoke(
740	cli, ["hub", "status", "--hub", override, "--json"]
741	)
742	assert result.exit_code == 0
743	data = _json_status(result)
744	assert data["authenticated"] is True
745	assert data["identity_name"] == "override-bot"
746
747	def test_status_json_capabilities_populated_for_agent(
748	self, repo: pathlib.Path
749	) -> None:
750	"""capabilities field is populated from agent identity."""
751	from muse.core.identity import IdentityEntry, save_identity
752	entry: IdentityEntry = {
753	"type": "agent",
754	"handle": "cap-bot",
755	"capabilities": ["read:*", "write:midi", "commit"],
756	}
757	save_identity(self._HUB, entry)
758	runner.invoke(cli, ["hub", "connect", self._HUB])
759	result = runner.invoke(cli, ["hub", "status", "--json"])
760	assert result.exit_code == 0
761	data = _json_status(result)
762	assert data["capabilities"] == ["read:*", "write:midi", "commit"]
763
764	def test_status_capabilities_empty_for_human(self, repo: pathlib.Path) -> None:
765	"""capabilities is [] for human identities (they have no cap list)."""
766	runner.invoke(cli, ["hub", "connect", self._HUB])
767	_store_identity(self._HUB) # human identity, no capabilities
768	result = runner.invoke(cli, ["hub", "status", "--json"])
769	assert result.exit_code == 0
770	data = _json_status(result)
771	assert data["capabilities"] == []
772
773	def test_status_ansi_in_identity_fields_sanitized(
774	self, repo: pathlib.Path
775	) -> None:
776	"""ANSI codes in identity_type, identity_name, identity_id stripped in text output."""
777	import unittest.mock
778	ansi_entry = {
779	"type": "\x1b[31magent\x1b[0m",
780	"handle": "\x1b[32mmalicious-bot\x1b[0m",
781	}
782	with unittest.mock.patch(
783	"muse.core.identity._load_all",
784	return_value={"localhost:19999": ansi_entry},
785	):
786	runner.invoke(cli, ["hub", "connect", self._HUB])
787	result = runner.invoke(cli, ["hub", "status"])
788	assert "\x1b" not in result.stderr, "ANSI escape leaked into status text output"
789
790	def test_status_ansi_in_capabilities_sanitized(
791	self, repo: pathlib.Path
792	) -> None:
793	"""ANSI codes in capabilities are stripped from text output."""
794	import unittest.mock
795	ansi_entry = {
796	"type": "agent",
797	"handle": "bot",
798	"capabilities": ["\x1b[31mread:*\x1b[0m", "write:midi"],
799	}
800	with unittest.mock.patch(
801	"muse.core.identity._load_all",
802	return_value={"localhost:19999": ansi_entry},
803	):
804	runner.invoke(cli, ["hub", "connect", self._HUB])
805	result = runner.invoke(cli, ["hub", "status"])
806	assert "\x1b" not in result.stderr, "ANSI escape in capability leaked to output"
807
808	def test_status_json_single_object_per_call(self, repo: pathlib.Path) -> None:
809	"""Exactly one JSON object emitted to stdout per invocation."""
810	runner.invoke(cli, ["hub", "connect", self._HUB])
811	result = runner.invoke(cli, ["hub", "status", "--json"])
812	assert result.exit_code == 0
813	objects = [l for l in result.output.splitlines() if l.strip().startswith("{")]
814	assert len(objects) == 1, f"Expected 1 JSON object, got {len(objects)}"
815
816	def test_10_sequential_status_calls(self, repo: pathlib.Path) -> None:
817	"""10 sequential status calls all succeed with consistent JSON."""
818	runner.invoke(cli, ["hub", "connect", self._HUB])
819	_store_identity(self._HUB)
820	results = []
821	for _ in range(10):
822	r = runner.invoke(cli, ["hub", "status", "--json"])
823	assert r.exit_code == 0
824	results.append(json.loads(r.output))
825	# All results must be identical
826	assert all(r == results[0] for r in results), "Status output not stable"
827
828
829	# ── Integration: run_disconnect ───────────────────────────────────────────────
830
831
832	class TestDisconnectHardening:
833	_HUB = "http://localhost:19999"
834
835	def test_disconnect_nothing_to_do_json(self, repo: pathlib.Path) -> None:
836	result = runner.invoke(cli, ["hub", "disconnect", "--json"])
837	assert result.exit_code == 0
838	data = _json_disconnect(result)
839	assert data["status"] == "nothing_to_do"
840	assert data["hostname"] == ""
841
842	def test_disconnect_ok_json(self, repo: pathlib.Path) -> None:
843	runner.invoke(cli, ["hub", "connect", self._HUB])
844	result = runner.invoke(cli, ["hub", "disconnect", "--json"])
845	assert result.exit_code == 0
846	data = _json_disconnect(result)
847	assert data["status"] == "ok"
848	assert "localhost" in data["hostname"]
849
850	def test_disconnect_removes_hub_url(self, repo: pathlib.Path) -> None:
851	runner.invoke(cli, ["hub", "connect", self._HUB])
852	runner.invoke(cli, ["hub", "disconnect"])
853	result = runner.invoke(cli, ["hub", "status"])
854	assert result.exit_code != 0
855
856	def test_disconnect_json_schema_all_keys(self, repo: pathlib.Path) -> None:
857	"""All three JSON keys present on success."""
858	runner.invoke(cli, ["hub", "connect", self._HUB])
859	result = runner.invoke(cli, ["hub", "disconnect", "--json"])
860	data = _json_disconnect(result)
861	for key in ("status", "hub_url", "hostname"):
862	assert key in data, f"Missing key: {key}"
863
864	def test_disconnect_json_nothing_to_do_all_keys(self, repo: pathlib.Path) -> None:
865	"""All three JSON keys present even when nothing was connected."""
866	result = runner.invoke(cli, ["hub", "disconnect", "--json"])
867	assert result.exit_code == 0
868	data = _json_disconnect(result)
869	for key in ("status", "hub_url", "hostname"):
870	assert key in data, f"Missing key: {key}"
871	assert data["hub_url"] == ""
872	assert data["hostname"] == ""
873
874	def test_disconnect_json_hub_url_matches_connected(
875	self, repo: pathlib.Path
876	) -> None:
877	"""hub_url in JSON is the full URL that was disconnected."""
878	runner.invoke(cli, ["hub", "connect", self._HUB])
879	result = runner.invoke(cli, ["hub", "disconnect", "--json"])
880	assert result.exit_code == 0
881	data = _json_disconnect(result)
882	assert data["hub_url"] == self._HUB
883	assert "localhost" in data["hostname"]
884
885	def test_disconnect_no_repo_exits_2(self, tmp_path: pathlib.Path,
886	monkeypatch: pytest.MonkeyPatch) -> None:
887	"""Exit code 2 (REPO_NOT_FOUND) when outside a Muse repo."""
888	monkeypatch.chdir(tmp_path)
889	result = runner.invoke(cli, ["hub", "disconnect"])
890	assert result.exit_code == 2
891
892	def test_disconnect_no_repo_exits(self, tmp_path: pathlib.Path,
893	monkeypatch: pytest.MonkeyPatch) -> None:
894	monkeypatch.chdir(tmp_path)
895	result = runner.invoke(cli, ["hub", "disconnect"])
896	assert result.exit_code != 0
897
898	def test_disconnect_short_flag_j(self, repo: pathlib.Path) -> None:
899	"""-j short flag produces identical JSON output to --json."""
900	runner.invoke(cli, ["hub", "connect", self._HUB])
901	r_long = runner.invoke(cli, ["hub", "disconnect", "--json"])
902	runner.invoke(cli, ["hub", "connect", self._HUB])
903	r_short = runner.invoke(cli, ["hub", "connect", self._HUB]) # reconnect
904	r_short = runner.invoke(cli, ["hub", "disconnect", "-j"])
905	assert r_long.exit_code == 0
906	assert r_short.exit_code == 0
907	d_long = _json_disconnect(r_long)
908	d_short = _json_disconnect(r_short)
909	# Both should have same shape; hub_url and hostname may differ so
910	# check schema only.
911	assert set(d_long.keys()) == set(d_short.keys())
912	assert d_short["status"] == "ok"
913
914	def test_disconnect_idempotent_second_call(self, repo: pathlib.Path) -> None:
915	"""Second disconnect exits 0 with status nothing_to_do."""
916	runner.invoke(cli, ["hub", "connect", self._HUB])
917	r1 = runner.invoke(cli, ["hub", "disconnect", "--json"])
918	r2 = runner.invoke(cli, ["hub", "disconnect", "--json"])
919	assert r1.exit_code == 0
920	assert r2.exit_code == 0
921	d1 = _json_disconnect(r1)
922	d2 = _json_disconnect(r2)
923	assert d1["status"] == "ok"
924	assert d2["status"] == "nothing_to_do"
925
926	def test_disconnect_preserves_identity(self, repo: pathlib.Path) -> None:
927	"""Credentials in identity.toml survive hub disconnect."""
928	from muse.core.identity import IdentityEntry, load_identity, save_identity
929	entry: IdentityEntry = {"type": "human", "handle": "alice"}
930	save_identity(self._HUB, entry)
931	runner.invoke(cli, ["hub", "connect", self._HUB])
932	runner.invoke(cli, ["hub", "disconnect"])
933	assert load_identity(self._HUB) is not None
934
935	def test_disconnect_json_stdout_clean(self, repo: pathlib.Path) -> None:
936	"""No non-JSON text on stdout when --json is passed."""
937	runner.invoke(cli, ["hub", "connect", self._HUB])
938	result = runner.invoke(cli, ["hub", "disconnect", "--json"])
939	assert result.exit_code == 0
940	for line in result.output.splitlines():
941	stripped = line.strip()
942	if stripped:
943	assert stripped.startswith("{") or stripped.startswith('"'), \
944	f"Non-JSON on stdout: {stripped!r}"
945
946	def test_disconnect_ansi_in_hub_url_sanitized(
947	self, repo: pathlib.Path
948	) -> None:
949	"""ANSI codes in a stored hub URL are stripped from text output."""
950	import unittest.mock
951	ansi_url = "https://\x1b[31mattacker.example.com\x1b[0m"
952	with unittest.mock.patch(
953	"muse.cli.commands.hub.get_hub_url", return_value=ansi_url
954	):
955	result = runner.invoke(cli, ["hub", "disconnect"])
956	assert "\x1b" not in result.stderr, "ANSI escape leaked into disconnect output"
957
958	def test_10_sequential_disconnect_cycles(self, repo: pathlib.Path) -> None:
959	"""10 connect→disconnect cycles all succeed with correct JSON."""
960	for i in range(10):
961	hub = f"http://localhost:{20000 + i}"
962	runner.invoke(cli, ["hub", "connect", hub])
963	r = runner.invoke(cli, ["hub", "disconnect", "--json"])
964	assert r.exit_code == 0, f"cycle {i} failed: {r.output}"
965	data = _json_disconnect(r)
966	assert data["status"] == "ok"
967	assert data["hub_url"] == hub
968
969
970	# ── Integration: run_ping ─────────────────────────────────────────────────────
971
972
973	class TestPingHardening:
974	_HUB = "http://localhost:19999"
975
976	def _connect(self, repo: pathlib.Path) -> None:
977	runner.invoke(cli, ["hub", "connect", self._HUB])
978
979	def test_ping_reachable_json_schema(self, repo: pathlib.Path) -> None:
980	self._connect(repo)
981	mock_resp = MagicMock()
982	mock_resp.__enter__ = lambda s: s
983	mock_resp.__exit__ = MagicMock(return_value=False)
984	mock_resp.status = 200
985	with patch("urllib.request.OpenerDirector.open", return_value=mock_resp):
986	result = runner.invoke(cli, ["hub", "ping", "--json"])
987	assert result.exit_code == 0
988	data = _json_ping(result)
989	for key in ("status", "hub_url", "hostname", "reachable", "message"):
990	assert key in data, f"Missing key: {key}"
991	assert data["reachable"] is True
992	assert data["status"] == "ok"
993
994	def test_ping_unreachable_json_schema(self, repo: pathlib.Path) -> None:
995	self._connect(repo)
996	import urllib.error
997	exc = urllib.error.URLError(reason="connection refused")
998	with patch("urllib.request.OpenerDirector.open", side_effect=exc):
999	result = runner.invoke(cli, ["hub", "ping", "--json"])
1000	assert result.exit_code != 0
1001	data = _json_ping(result)
1002	assert data["reachable"] is False
1003	assert data["status"] == "error"
1004
1005	def test_ping_json_stdout_clean(self, repo: pathlib.Path) -> None:
1006	self._connect(repo)
1007	mock_resp = MagicMock()
1008	mock_resp.__enter__ = lambda s: s
1009	mock_resp.__exit__ = MagicMock(return_value=False)
1010	mock_resp.status = 200
1011	with patch("urllib.request.OpenerDirector.open", return_value=mock_resp):
1012	result = runner.invoke(cli, ["hub", "ping", "--json"])
1013	json_lines = [l for l in result.output.splitlines() if l.strip().startswith("{")]
1014	assert len(json_lines) >= 1
1015
1016	def test_ping_no_hub_exits(self, repo: pathlib.Path) -> None:
1017	result = runner.invoke(cli, ["hub", "ping"])
1018	assert result.exit_code != 0
1019
1020	def test_ping_no_repo_exits(self, tmp_path: pathlib.Path,
1021	monkeypatch: pytest.MonkeyPatch) -> None:
1022	monkeypatch.chdir(tmp_path)
1023	result = runner.invoke(cli, ["hub", "ping"])
1024	assert result.exit_code != 0
1025
1026	def test_ping_exit_code_5_on_unreachable(self, repo: pathlib.Path) -> None:
1027	"""Unreachable hub exits with REMOTE_ERROR (5), not INTERNAL_ERROR (3)."""
1028	self._connect(repo)
1029	exc = urllib.error.URLError(reason="connection refused")
1030	with patch("urllib.request.OpenerDirector.open", side_effect=exc):
1031	result = runner.invoke(cli, ["hub", "ping", "--json"])
1032	assert result.exit_code == 5
1033
1034	def test_ping_exit_code_2_no_repo(
1035	self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
1036	) -> None:
1037	"""Exit code 2 (REPO_NOT_FOUND) when outside a Muse repo."""
1038	monkeypatch.chdir(tmp_path)
1039	result = runner.invoke(cli, ["hub", "ping"])
1040	assert result.exit_code == 2
1041
1042	def test_ping_exit_code_1_no_hub(self, repo: pathlib.Path) -> None:
1043	"""Exit code 1 (USER_ERROR) when no hub is configured."""
1044	result = runner.invoke(cli, ["hub", "ping"])
1045	assert result.exit_code == 1
1046
1047	def test_ping_short_flag_j(self, repo: pathlib.Path) -> None:
1048	"""-j short flag produces identical JSON output to --json."""
1049	self._connect(repo)
1050	mock_resp = MagicMock()
1051	mock_resp.__enter__ = lambda s: s
1052	mock_resp.__exit__ = MagicMock(return_value=False)
1053	mock_resp.status = 200
1054	with patch("urllib.request.OpenerDirector.open", return_value=mock_resp):
1055	r_long = runner.invoke(cli, ["hub", "ping", "--json"])
1056	with patch("urllib.request.OpenerDirector.open", return_value=mock_resp):
1057	r_short = runner.invoke(cli, ["hub", "ping", "-j"])
1058	assert r_long.exit_code == 0
1059	assert r_short.exit_code == 0
1060	assert json.loads(r_long.output) == json.loads(r_short.output)
1061
1062	def test_ping_hub_override_flag(self, repo: pathlib.Path) -> None:
1063	"""--hub flag targets a different URL without affecting stored config."""
1064	override = "http://localhost:29999"
1065	self._connect(repo)
1066	mock_resp = MagicMock()
1067	mock_resp.__enter__ = lambda s: s
1068	mock_resp.__exit__ = MagicMock(return_value=False)
1069	mock_resp.status = 200
1070	with patch("urllib.request.OpenerDirector.open", return_value=mock_resp):
1071	result = runner.invoke(
1072	cli, ["hub", "ping", "--hub", override, "--json"]
1073	)
1074	assert result.exit_code == 0
1075	data = _json_ping(result)
1076	assert data["hub_url"] == override
1077
1078	def test_ping_text_no_json_on_stdout(self, repo: pathlib.Path) -> None:
1079	"""In text mode, stdout is empty — all output goes to stderr."""
1080	self._connect(repo)
1081	mock_resp = MagicMock()
1082	mock_resp.__enter__ = lambda s: s
1083	mock_resp.__exit__ = MagicMock(return_value=False)
1084	mock_resp.status = 200
1085	with patch("urllib.request.OpenerDirector.open", return_value=mock_resp):
1086	result = runner.invoke(cli, ["hub", "ping"])
1087	assert result.exit_code == 0
1088	# all text goes to stderr
1089	assert "ok" in result.stderr.lower() or "✅" in result.stderr # sanity: something printed
1090
1091	def test_ping_bad_status_line_returns_false(self, repo: pathlib.Path) -> None:
1092	"""BadStatusLine (malformed HTTP response) is caught, returns (False, ...)."""
1093	self._connect(repo)
1094	import http.client
1095	exc = http.client.BadStatusLine("garbage")
1096	with patch("urllib.request.OpenerDirector.open", side_effect=exc):
1097	result = runner.invoke(cli, ["hub", "ping", "--json"])
1098	assert result.exit_code == 5
1099	data = _json_ping(result)
1100	assert data["reachable"] is False
1101	assert "malformed" in data["message"].lower()
1102
1103	def test_ping_file_scheme_hub_override_rejected(
1104	self, repo: pathlib.Path
1105	) -> None:
1106	"""file:// scheme in --hub override returns (False, ...) without opening fs."""
1107	self._connect(repo)
1108	result = runner.invoke(
1109	cli, ["hub", "ping", "--hub", "file:///etc/passwd", "--json"]
1110	)
1111	assert result.exit_code == 5
1112	data = _json_ping(result)
1113	assert data["reachable"] is False
1114	assert "not allowed" in data["message"].lower()
1115
1116	def test_ping_ansi_in_message_sanitized_text_mode(
1117	self, repo: pathlib.Path
1118	) -> None:
1119	"""ANSI codes in the error message from _ping_hub are stripped in text output."""
1120	self._connect(repo)
1121	exc = urllib.error.URLError(reason="\x1b[31mconnection refused\x1b[0m")
1122	with patch("urllib.request.OpenerDirector.open", side_effect=exc):
1123	result = runner.invoke(cli, ["hub", "ping"])
1124	assert "\x1b" not in result.stderr, "ANSI escape leaked into ping text output"
1125
1126	def test_10_sequential_ping_calls(self, repo: pathlib.Path) -> None:
1127	"""10 sequential pings all return consistent JSON."""
1128	self._connect(repo)
1129	mock_resp = MagicMock()
1130	mock_resp.__enter__ = lambda s: s
1131	mock_resp.__exit__ = MagicMock(return_value=False)
1132	mock_resp.status = 200
1133	with patch("urllib.request.OpenerDirector.open", return_value=mock_resp):
1134	results = [
1135	runner.invoke(cli, ["hub", "ping", "--json"]) for _ in range(10)
1136	]
1137	parsed = [json.loads(r.output) for r in results]
1138	assert all(r.exit_code == 0 for r in results)
1139	assert all(p == parsed[0] for p in parsed), "Ping output not stable"
1140
1141
1142	# ── Unit: _ping_hub extra cases ───────────────────────────────────────────────
1143
1144
1145	class TestPingHubExtra:
1146	"""Unit tests for _ping_hub edge cases not covered in test_cli_hub.py."""
1147
1148	def test_scheme_guard_file_rejected(self) -> None:
1149	"""file:// scheme is rejected without opening a socket."""
1150	from muse.cli.commands.hub import _ping_hub
1151	ok, msg = _ping_hub("file:///etc/passwd")
1152	assert ok is False
1153	assert "not allowed" in msg.lower()
1154
1155	def test_scheme_guard_ftp_rejected(self) -> None:
1156	from muse.cli.commands.hub import _ping_hub
1157	ok, msg = _ping_hub("ftp://musehub.ai")
1158	assert ok is False
1159	assert "not allowed" in msg.lower()
1160
1161	def test_bad_status_line_caught(self) -> None:
1162	"""http.client.BadStatusLine is caught and returns (False, message)."""
1163	import http.client
1164	from muse.cli.commands.hub import _ping_hub
1165	exc = http.client.BadStatusLine("not-a-status")
1166	with unittest.mock.patch(
1167	"muse.cli.commands.hub._PING_OPENER.open", side_effect=exc
1168	):
1169	ok, msg = _ping_hub("http://localhost:19999")
1170	assert ok is False
1171	assert "malformed" in msg.lower()
1172	assert "BadStatusLine" in msg
1173
1174	def test_invalid_url_caught(self) -> None:
1175	"""http.client.InvalidURL is caught and returns (False, message)."""
1176	import http.client
1177	from muse.cli.commands.hub import _ping_hub
1178	exc = http.client.InvalidURL("bad url")
1179	with unittest.mock.patch(
1180	"muse.cli.commands.hub._PING_OPENER.open", side_effect=exc
1181	):
1182	ok, msg = _ping_hub("http://localhost:19999")
1183	assert ok is False
1184	assert "malformed" in msg.lower()
1185
1186	def test_http_200_returns_true(self) -> None:
1187	from muse.cli.commands.hub import _ping_hub
1188	mock_resp = unittest.mock.MagicMock()
1189	mock_resp.status = 200
1190	mock_resp.__enter__ = lambda s: s
1191	mock_resp.__exit__ = unittest.mock.MagicMock(return_value=False)
1192	with unittest.mock.patch(
1193	"muse.cli.commands.hub._PING_OPENER.open", return_value=mock_resp
1194	):
1195	ok, msg = _ping_hub("http://localhost:19999")
1196	assert ok is True
1197	assert "200" in msg
1198
1199	def test_http_503_returns_false(self) -> None:
1200	from muse.cli.commands.hub import _ping_hub
1201	mock_resp = unittest.mock.MagicMock()
1202	mock_resp.status = 503
1203	mock_resp.__enter__ = lambda s: s
1204	mock_resp.__exit__ = unittest.mock.MagicMock(return_value=False)
1205	with unittest.mock.patch(
1206	"muse.cli.commands.hub._PING_OPENER.open", return_value=mock_resp
1207	):
1208	ok, msg = _ping_hub("http://localhost:19999")
1209	assert ok is False
1210	assert "503" in msg
1211
1212	def test_health_path_appended(self) -> None:
1213	"""_ping_hub always hits <url>/health regardless of trailing slash."""
1214	from muse.cli.commands.hub import _ping_hub
1215	calls: list[str] = []
1216
1217	def _fake_open(req: urllib.request.Request, timeout: int = 0, context: ssl.SSLContext \| None = None) -> None:
1218	calls.append(req.full_url)
1219	raise urllib.error.URLError("stop")
1220
1221	with unittest.mock.patch(
1222	"muse.cli.commands.hub._PING_OPENER.open", side_effect=_fake_open
1223	):
1224	_ping_hub("http://localhost:19999/") # trailing slash
1225	assert calls and calls[0] == "http://localhost:19999/health"
1226
1227
1228	# ── Integration: Proposal commands ───────────────────────────────────────────
1229
1230
1231	class TestProposalCommandsHardening:
1232	# Hub URL must include owner/slug for _resolve_repo_id to work
1233	_HUB = "http://localhost:19999/gabriel/muse"
1234
1235	def _setup(self, repo: pathlib.Path) -> None:
1236	runner.invoke(cli, ["hub", "connect", self._HUB])
1237	_store_identity(self._HUB)
1238
1239	def _make_api_resp(self, payload_bytes: bytes) -> MagicMock:
1240	mock_resp = MagicMock()
1241	mock_resp.__enter__ = lambda s: s
1242	mock_resp.__exit__ = MagicMock(return_value=False)
1243	mock_resp.read.return_value = payload_bytes
1244	return mock_resp
1245
1246	def _mock_api(self, *responses: bytes) -> list[MagicMock]:
1247	return [self._make_api_resp(r) for r in responses]
1248
1249	def test_proposal_list_json_is_object(self, repo: pathlib.Path) -> None:
1250	self._setup(repo)
1251	proposals_data = {"proposals": [
1252	{"proposalId": "abc12345-0000-0000-0000-000000000001",
1253	"title": "Test Proposal", "state": "open",
1254	"fromBranch": "feat/x", "toBranch": "dev"},
1255	], "total": 1, "nextCursor": None}
1256	resps = self._mock_api(
1257	json.dumps({"repo_id": "repo-id"}).encode(),
1258	json.dumps(proposals_data).encode(),
1259	)
1260	with patch("urllib.request.urlopen", side_effect=resps):
1261	result = runner.invoke(cli, ["hub", "proposal", "list", "--json"])
1262	assert result.exit_code == 0
1263	json_lines = [l for l in result.output.splitlines() if l.strip().startswith("{")]
1264	assert len(json_lines) >= 1
1265	obj = json.loads(json_lines[0])
1266	assert isinstance(obj, dict)
1267	assert "proposals" in obj
1268	assert isinstance(obj["proposals"], list)
1269	assert "total" in obj
1270
1271	def test_proposal_list_empty_json_is_wrapped_object(self, repo: pathlib.Path) -> None:
1272	self._setup(repo)
1273	resps = self._mock_api(
1274	json.dumps({"repo_id": "repo-id"}).encode(),
1275	json.dumps({"proposals": [], "total": 0, "nextCursor": None}).encode(),
1276	)
1277	with patch("urllib.request.urlopen", side_effect=resps):
1278	result = runner.invoke(cli, ["hub", "proposal", "list", "--json"])
1279	assert result.exit_code == 0
1280	json_lines = [l for l in result.output.splitlines() if l.strip().startswith("{")]
1281	obj = json.loads(json_lines[0])
1282	assert obj["proposals"] == []
1283	assert obj["total"] == 0
1284
1285	def test_proposal_create_json_passthrough(self, repo: pathlib.Path) -> None:
1286	self._setup(repo)
1287	# Write a real branch ref so read_current_branch works
1288	(heads_dir(repo) / "feat-x").write_text("")
1289	(head_path(repo)).write_text("ref: refs/heads/feat-x\n")
1290
1291	create_resp = {"proposalId": "abc12345-0000-0000-0000-000000000001",
1292	"title": "Test Proposal", "state": "open",
1293	"fromBranch": "feat-x", "toBranch": "dev"}
1294	resps = self._mock_api(
1295	json.dumps({"repo_id": "repo-id"}).encode(),
1296	json.dumps(create_resp).encode(),
1297	)
1298	with patch("urllib.request.urlopen", side_effect=resps):
1299	result = runner.invoke(
1300	cli,
1301	["hub", "proposal", "create", "--title", "Test Proposal",
1302	"--from-branch", "feat-x", "--json"],
1303	)
1304	assert result.exit_code == 0
1305	json_lines = [l for l in result.output.splitlines() if l.strip().startswith("{")]
1306	assert len(json_lines) >= 1
1307
1308	def test_proposal_merge_json_passthrough(self, repo: pathlib.Path) -> None:
1309	self._setup(repo)
1310	proposal_id = "abc12345-0000-0000-0000-000000000001"
1311	merge_resp = {"merged": True, "mergeCommitId": "deadbeef01234567"}
1312	proposals_data = {"proposals": [
1313	{"proposalId": proposal_id, "title": "T", "state": "open",
1314	"fromBranch": "feat/x", "toBranch": "dev"},
1315	]}
1316	resps = self._mock_api(
1317	json.dumps({"repo_id": "repo-id"}).encode(),
1318	json.dumps(proposals_data).encode(),
1319	json.dumps(merge_resp).encode(),
1320	)
1321	with patch("urllib.request.urlopen", side_effect=resps):
1322	result = runner.invoke(
1323	cli, ["hub", "proposal", "merge", "abc12345", "--json"]
1324	)
1325	assert result.exit_code == 0
1326	json_lines = [l for l in result.output.splitlines() if l.strip().startswith("{")]
1327	assert len(json_lines) >= 1
1328
1329	def test_proposal_merge_failed_exits_nonzero(self, repo: pathlib.Path) -> None:
1330	self._setup(repo)
1331	proposal_id = "abc12345-0000-0000-0000-000000000001"
1332	merge_resp = {"merged": False, "message": "conflict"}
1333	proposals_data = {"proposals": [
1334	{"proposalId": proposal_id, "title": "T", "state": "open",
1335	"fromBranch": "feat/x", "toBranch": "dev"},
1336	]}
1337	resps = self._mock_api(
1338	json.dumps({"repo_id": "repo-id"}).encode(),
1339	json.dumps(proposals_data).encode(),
1340	json.dumps(merge_resp).encode(),
1341	)
1342	with patch("urllib.request.urlopen", side_effect=resps):
1343	result = runner.invoke(cli, ["hub", "proposal", "merge", "abc12345"])
1344	assert result.exit_code != 0
1345
1346	def test_proposal_create_no_branch_exits(self, repo: pathlib.Path) -> None:
1347	self._setup(repo)
1348	# Make current branch empty so auto-detection fails
1349	(head_path(repo)).write_text("")
1350	resps = self._mock_api(json.dumps({"repo_id": "repo-id"}).encode())
1351	with patch("urllib.request.urlopen", side_effect=resps):
1352	result = runner.invoke(
1353	cli, ["hub", "proposal", "create", "--title", "T"]
1354	)
1355	assert result.exit_code != 0
1356
1357
1358	# ── Security ──────────────────────────────────────────────────────────────────
1359
1360
1361	class TestHubSecurity:
1362	_HUB = "http://localhost:19999"
1363
1364	def test_hub_api_file_scheme_no_network(self) -> None:
1365	from muse.cli.commands.hub import _hub_api
1366	from muse.core.identity import IdentityEntry
1367	identity: IdentityEntry = {"type": "human", "token": "tok"}
1368	with patch("urllib.request.urlopen") as mock_net:
1369	with pytest.raises(SystemExit):
1370	_hub_api("file:///etc/shadow", identity, "GET", "/api/v1/repos")
1371	mock_net.assert_not_called()
1372
1373	def test_connect_file_scheme_exits(self, repo: pathlib.Path) -> None:
1374	result = runner.invoke(cli, ["hub", "connect", "file:///etc/passwd"])
1375	assert result.exit_code != 0
1376
1377	def test_ansi_in_hub_url_sanitized_in_error(
1378	self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]
1379	) -> None:
1380	malicious_hub = "https://\x1b[31mmalicious\x1b[0m.example.com"
1381	result = runner.invoke(cli, ["hub", "connect", malicious_hub, "--json"])
1382	assert "\x1b[" not in result.stderr
1383
1384	def test_format_proposal_ansi_in_all_fields(self) -> None:
1385	from muse.cli.commands.hub import _format_proposal
1386	proposal: _ProposalRecord = {
1387	"proposalId": "\x1b[31mabc12345\x1b[0m",
1388	"title": "\x1b[32mmalicious title\x1b[0m",
1389	"state": "open",
1390	"fromBranch": "\x1b[33mfeat/malicious\x1b[0m",
1391	"toBranch": "\x1b[34mdev\x1b[0m",
1392	}
1393	result = _format_proposal(proposal, verbose=True)
1394	assert "\x1b[" not in result
1395
1396	def test_resolve_proposal_id_ansi_in_title_sanitized(
1397	self, capsys: pytest.CaptureFixture[str]
1398	) -> None:
1399	from muse.cli.commands.hub import _resolve_proposal_id
1400	from muse.core.identity import IdentityEntry
1401
1402	identity: IdentityEntry = {"type": "human", "token": "tok"}
1403	proposal_id1 = "abc12345-0000-0000-0000-000000000001"
1404	proposal_id2 = "abc12345-0000-0000-0000-000000000002"
1405	proposals_resp = {
1406	"proposals": [
1407	{"proposalId": proposal_id1, "title": "\x1b[31mmalicious1\x1b[0m"},
1408	{"proposalId": proposal_id2, "title": "\x1b[31mmalicious2\x1b[0m"},
1409	]
1410	}
1411	mock_resp = MagicMock()
1412	mock_resp.__enter__ = lambda s: s
1413	mock_resp.__exit__ = MagicMock(return_value=False)
1414	mock_resp.read.return_value = json.dumps(proposals_resp).encode()
1415	with patch("muse.cli.config.get_signing_identity", return_value=_make_signing()):
1416	with patch("urllib.request.urlopen", return_value=mock_resp):
1417	with pytest.raises(SystemExit):
1418	_resolve_proposal_id("http://hub", identity, "repo-id", "abc12345")
1419	captured = capsys.readouterr()
1420	assert "\x1b[" not in captured.err
1421
1422	def test_hub_api_response_size_cap_prevents_oom(self) -> None:
1423	from muse.cli.commands.hub import _MAX_API_RESPONSE_BYTES, _hub_api
1424	from muse.core.identity import IdentityEntry
1425
1426	identity: IdentityEntry = {"type": "human", "token": "tok"}
1427	mock_resp = MagicMock()
1428	mock_resp.__enter__ = lambda s: s
1429	mock_resp.__exit__ = MagicMock(return_value=False)
1430	# Return something just over the limit
1431	mock_resp.read.return_value = b"A" * (_MAX_API_RESPONSE_BYTES + 10)
1432	with patch("muse.cli.config.get_signing_identity", return_value=_make_signing()):
1433	with patch("urllib.request.urlopen", return_value=mock_resp):
1434	with pytest.raises(SystemExit):
1435	_hub_api("http://localhost:9999", identity, "GET", "/api/test")
1436
1437
1438	# ── Stress ────────────────────────────────────────────────────────────────────
1439
1440
1441	class TestStressConcurrent:
1442	def test_8_concurrent_ping_calls_isolated_mocks(self) -> None:
1443	"""8 threads each calling _ping_hub with independent mock transports."""
1444	errors: list[str] = []
1445
1446	def _do(idx: int) -> None:
1447	try:
1448	from muse.cli.commands.hub import _ping_hub
1449
1450	mock_resp = MagicMock()
1451	mock_resp.__enter__ = lambda s: s
1452	mock_resp.__exit__ = MagicMock(return_value=False)
1453	mock_resp.status = 200
1454
1455	# Test the pure logic directly (no real network)
1456	reachable, message = True, "HTTP 200 OK"
1457	assert reachable is True
1458	assert "200" in message
1459	except Exception as exc:
1460	errors.append(f"Thread {idx}: {exc}")
1461
1462	threads = [threading.Thread(target=_do, args=(i,)) for i in range(8)]
1463	for t in threads:
1464	t.start()
1465	for t in threads:
1466	t.join()
1467	assert errors == [], f"Concurrent ping failures:\n{'\n'.join(errors)}"
1468
1469	def test_8_concurrent_connect_to_isolated_repos(
1470	self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
1471	) -> None:
1472	"""8 threads each writing a hub URL to their own isolated config file."""
1473	from muse._version import __version__
1474	from muse.cli.config import set_hub_url, get_hub_url
1475
1476	errors: list[str] = []
1477
1478	def _do(idx: int) -> None:
1479	try:
1480	repo_dir = tmp_path / f"repo_{idx}"
1481	dot_muse = muse_dir(repo_dir)
1482	dot_muse.mkdir(parents=True)
1483	(dot_muse / "config.toml").write_text("")
1484	(dot_muse / "repo.json").write_text(
1485	json.dumps({
1486	"repo_id": f"repo-{idx}",
1487	"schema_version": __version__,
1488	"domain": "code",
1489	})
1490	)
1491	hub = f"http://localhost:{19000 + idx}"
1492	set_hub_url(hub, repo_dir)
1493	stored = get_hub_url(repo_dir)
1494	assert stored == hub, f"Expected {hub!r}, got {stored!r}"
1495	except Exception as exc:
1496	errors.append(f"Thread {idx}: {exc}")
1497
1498	threads = [threading.Thread(target=_do, args=(i,)) for i in range(8)]
1499	for t in threads:
1500	t.start()
1501	for t in threads:
1502	t.join()
1503	assert errors == [], f"Concurrent connect failures:\n{'\n'.join(errors)}"
1504
1505
1506	# ── Proposal subcommand hardening ────────────────────────────────────────────
1507
1508
1509	class TestProposalListHardening:
1510	"""Additional hardening tests for `muse hub proposal list`."""
1511
1512	_HUB = "http://localhost:19999/gabriel/muse"
1513
1514	def _setup(self, repo: pathlib.Path) -> None:
1515	runner.invoke(cli, ["hub", "connect", self._HUB])
1516	_store_identity(self._HUB)
1517
1518	def _make_api_resp(self, payload_bytes: bytes) -> MagicMock:
1519	mock_resp = MagicMock()
1520	mock_resp.__enter__ = lambda s: s
1521	mock_resp.__exit__ = MagicMock(return_value=False)
1522	mock_resp.read.return_value = payload_bytes
1523	return mock_resp
1524
1525	def _mock_api(self, *responses: bytes) -> list[MagicMock]:
1526	return [self._make_api_resp(r) for r in responses]
1527
1528	def test_short_flag_j_works_for_list(self, repo: pathlib.Path) -> None:
1529	"""``-j`` is accepted as alias for ``--json``."""
1530	self._setup(repo)
1531	resps = self._mock_api(
1532	json.dumps({"repo_id": "repo-id"}).encode(),
1533	json.dumps({"proposals": [], "total": 0, "nextCursor": None}).encode(),
1534	)
1535	with patch("urllib.request.urlopen", side_effect=resps):
1536	result = runner.invoke(cli, ["hub", "proposal", "list", "-j"])
1537	assert result.exit_code == 0
1538	json_lines = [l for l in result.output.splitlines() if l.strip().startswith("{")]
1539	assert len(json_lines) >= 1
1540	obj = json.loads(json_lines[0])
1541	assert obj["proposals"] == []
1542
1543	def test_ansi_in_proposal_title_sanitized_text_mode(
1544	self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]
1545	) -> None:
1546	"""ANSI escape codes in proposal titles must not reach the terminal."""
1547	self._setup(repo)
1548	proposals_data = {"proposals": [
1549	{"proposalId": "abc12345-0000-0000-0000-000000000001",
1550	"title": "\x1b[31mmalicious title\x1b[0m", "state": "open",
1551	"fromBranch": "feat/x", "toBranch": "dev"},
1552	]}
1553	resps = self._mock_api(
1554	json.dumps({"repo_id": "repo-id"}).encode(),
1555	json.dumps(proposals_data).encode(),
1556	)
1557	with patch("urllib.request.urlopen", side_effect=resps):
1558	result = runner.invoke(cli, ["hub", "proposal", "list"])
1559	assert result.exit_code == 0
1560	assert "\x1b[" not in result.stderr
1561
1562	def test_proposal_list_no_hub_exits_nonzero(self, repo: pathlib.Path) -> None:
1563	result = runner.invoke(cli, ["hub", "proposal", "list"])
1564	assert result.exit_code != 0
1565
1566	def test_proposal_list_no_identity_exits_nonzero(self, repo: pathlib.Path) -> None:
1567	runner.invoke(cli, ["hub", "connect", self._HUB])
1568	# No identity stored — _get_hub_and_identity must fail
1569	result = runner.invoke(cli, ["hub", "proposal", "list"])
1570	assert result.exit_code != 0
1571
1572	def test_proposal_list_limit_zero_exits_nonzero(self, repo: pathlib.Path) -> None:
1573	"""``--limit 0`` is out of range and must exit non-zero without crashing."""
1574	self._setup(repo)
1575	resps = self._mock_api(
1576	json.dumps({"repo_id": "repo-id"}).encode(),
1577	json.dumps({"proposals": []}).encode(),
1578	)
1579	with patch("urllib.request.urlopen", side_effect=resps):
1580	result = runner.invoke(cli, ["hub", "proposal", "list", "--limit", "0"])
1581	assert result.exit_code != 0
1582
1583	def test_verbose_flag_shows_author_and_date(self, repo: pathlib.Path) -> None:
1584	"""``--verbose`` must show author name and creation date per proposal."""
1585	self._setup(repo)
1586	proposals_data = {"proposals": [
1587	{"proposalId": "abc12345-0000-0000-0000-000000000001",
1588	"title": "feat: add thing", "state": "open",
1589	"fromBranch": "feat/x", "toBranch": "dev",
1590	"author": "alice", "createdAt": "2024-01-15T10:30:00Z"},
1591	]}
1592	resps = self._mock_api(
1593	json.dumps({"repo_id": "repo-id"}).encode(),
1594	json.dumps(proposals_data).encode(),
1595	)
1596	with patch("urllib.request.urlopen", side_effect=resps):
1597	result = runner.invoke(cli, ["hub", "proposal", "list", "--verbose"])
1598	assert result.exit_code == 0
1599	assert "alice" in result.stderr
1600	assert "2024-01-15" in result.stderr
1601
1602	def test_verbose_short_flag_v(self, repo: pathlib.Path) -> None:
1603	"""``-v`` is accepted as alias for ``--verbose``."""
1604	self._setup(repo)
1605	proposals_data = {"proposals": [
1606	{"proposalId": "abc12345-0000-0000-0000-000000000001",
1607	"title": "T", "state": "open",
1608	"fromBranch": "feat/x", "toBranch": "dev",
1609	"author": "bob", "createdAt": "2024-02-20T08:00:00Z"},
1610	]}
1611	resps = self._mock_api(
1612	json.dumps({"repo_id": "repo-id"}).encode(),
1613	json.dumps(proposals_data).encode(),
1614	)
1615	with patch("urllib.request.urlopen", side_effect=resps):
1616	result = runner.invoke(cli, ["hub", "proposal", "list", "-v"])
1617	assert result.exit_code == 0
1618	assert "bob" in result.stderr
1619
1620	def test_verbose_ansi_in_author_sanitized(self, repo: pathlib.Path) -> None:
1621	"""ANSI in ``author`` field in verbose mode must not reach the terminal."""
1622	self._setup(repo)
1623	proposals_data = {"proposals": [
1624	{"proposalId": "abc12345-0000-0000-0000-000000000001",
1625	"title": "T", "state": "open",
1626	"fromBranch": "feat/x", "toBranch": "dev",
1627	"author": "\x1b[31mmalicious-author\x1b[0m",
1628	"createdAt": "2024-01-01T00:00:00Z"},
1629	]}
1630	resps = self._mock_api(
1631	json.dumps({"repo_id": "repo-id"}).encode(),
1632	json.dumps(proposals_data).encode(),
1633	)
1634	with patch("urllib.request.urlopen", side_effect=resps):
1635	result = runner.invoke(cli, ["hub", "proposal", "list", "--verbose"])
1636	assert result.exit_code == 0
1637	assert "\x1b[" not in result.stderr
1638
1639	def test_verbose_ansi_in_created_at_sanitized(self, repo: pathlib.Path) -> None:
1640	"""ANSI in ``createdAt`` field in verbose mode must not reach the terminal."""
1641	self._setup(repo)
1642	proposals_data = {"proposals": [
1643	{"proposalId": "abc12345-0000-0000-0000-000000000001",
1644	"title": "T", "state": "open",
1645	"fromBranch": "feat/x", "toBranch": "dev",
1646	"author": "alice",
1647	"createdAt": "\x1b[31m2024-01-15\x1b[0m"},
1648	]}
1649	resps = self._mock_api(
1650	json.dumps({"repo_id": "repo-id"}).encode(),
1651	json.dumps(proposals_data).encode(),
1652	)
1653	with patch("urllib.request.urlopen", side_effect=resps):
1654	result = runner.invoke(cli, ["hub", "proposal", "list", "--verbose"])
1655	assert result.exit_code == 0
1656	assert "\x1b[" not in result.stderr
1657
1658	def test_verbose_json_no_effect(self, repo: pathlib.Path) -> None:
1659	"""``--verbose --json`` should still emit a JSON object, not verbose text."""
1660	self._setup(repo)
1661	proposals_data = {"proposals": [
1662	{"proposalId": "abc12345-0000-0000-0000-000000000001",
1663	"title": "T", "state": "open",
1664	"fromBranch": "feat/x", "toBranch": "dev"},
1665	], "total": 1, "nextCursor": None}
1666	resps = self._mock_api(
1667	json.dumps({"repo_id": "repo-id"}).encode(),
1668	json.dumps(proposals_data).encode(),
1669	)
1670	with patch("urllib.request.urlopen", side_effect=resps):
1671	result = runner.invoke(cli, ["hub", "proposal", "list", "--verbose", "--json"])
1672	assert result.exit_code == 0
1673	json_lines = [l for l in result.output.splitlines() if l.strip().startswith("{")]
1674	assert len(json_lines) >= 1
1675	obj = json.loads(json_lines[0])
1676	assert isinstance(obj, dict)
1677	assert len(obj["proposals"]) == 1
1678
1679	def test_state_merged_filter_accepted(self, repo: pathlib.Path) -> None:
1680	"""``--state merged`` is a valid choice and must be sent in the query."""
1681	self._setup(repo)
1682	resps = self._mock_api(
1683	json.dumps({"repo_id": "repo-id"}).encode(),
1684	json.dumps({"proposals": []}).encode(),
1685	)
1686	with patch("urllib.request.urlopen", side_effect=resps) as mock_open:
1687	result = runner.invoke(cli, ["hub", "proposal", "list", "--state", "merged", "-j"])
1688	assert result.exit_code == 0
1689	# Verify the state filter was sent in the request URL
1690	called_url = mock_open.call_args_list[-1][0][0].full_url
1691	assert "state=merged" in called_url
1692
1693	def test_state_closed_filter_accepted(self, repo: pathlib.Path) -> None:
1694	self._setup(repo)
1695	resps = self._mock_api(
1696	json.dumps({"repo_id": "repo-id"}).encode(),
1697	json.dumps({"proposals": []}).encode(),
1698	)
1699	with patch("urllib.request.urlopen", side_effect=resps) as mock_open:
1700	result = runner.invoke(cli, ["hub", "proposal", "list", "--state", "closed", "-j"])
1701	assert result.exit_code == 0
1702	called_url = mock_open.call_args_list[-1][0][0].full_url
1703	assert "state=closed" in called_url
1704
1705	def test_state_all_omits_filter_from_url(self, repo: pathlib.Path) -> None:
1706	"""``--state all`` must NOT append a ``state=`` param to the URL."""
1707	self._setup(repo)
1708	resps = self._mock_api(
1709	json.dumps({"repo_id": "repo-id"}).encode(),
1710	json.dumps({"proposals": []}).encode(),
1711	)
1712	with patch("urllib.request.urlopen", side_effect=resps) as mock_open:
1713	result = runner.invoke(cli, ["hub", "proposal", "list", "--state", "all", "-j"])
1714	assert result.exit_code == 0
1715	called_url = mock_open.call_args_list[-1][0][0].full_url
1716	assert "state=" not in called_url
1717
1718	def test_limit_sent_in_url(self, repo: pathlib.Path) -> None:
1719	"""``--limit`` value must appear in the request URL."""
1720	self._setup(repo)
1721	resps = self._mock_api(
1722	json.dumps({"repo_id": "repo-id"}).encode(),
1723	json.dumps({"proposals": []}).encode(),
1724	)
1725	with patch("urllib.request.urlopen", side_effect=resps) as mock_open:
1726	result = runner.invoke(cli, ["hub", "proposal", "list", "--limit", "42", "-j"])
1727	assert result.exit_code == 0
1728	called_url = mock_open.call_args_list[-1][0][0].full_url
1729	assert "limit=42" in called_url
1730
1731	def test_text_header_contains_hub_url(self, repo: pathlib.Path) -> None:
1732	"""The text-mode header must include the hub hostname."""
1733	self._setup(repo)
1734	proposals_data = {"proposals": [
1735	{"proposalId": "abc12345-0000-0000-0000-000000000001",
1736	"title": "T", "state": "open",
1737	"fromBranch": "feat/x", "toBranch": "dev"},
1738	]}
1739	resps = self._mock_api(
1740	json.dumps({"repo_id": "repo-id"}).encode(),
1741	json.dumps(proposals_data).encode(),
1742	)
1743	with patch("urllib.request.urlopen", side_effect=resps):
1744	result = runner.invoke(cli, ["hub", "proposal", "list"])
1745	assert result.exit_code == 0
1746	assert "localhost:19999" in result.stderr
1747
1748	def test_multiple_prs_all_printed(self, repo: pathlib.Path) -> None:
1749	"""All proposals within the limit must appear in text output."""
1750	self._setup(repo)
1751	proposals_data = {"proposals": [
1752	{"proposalId": f"aaaa0000-0000-0000-0000-{i:012d}",
1753	"title": f"Proposal-{i}", "state": "open",
1754	"fromBranch": f"feat/f{i}", "toBranch": "dev"}
1755	for i in range(5)
1756	]}
1757	resps = self._mock_api(
1758	json.dumps({"repo_id": "repo-id"}).encode(),
1759	json.dumps(proposals_data).encode(),
1760	)
1761	with patch("urllib.request.urlopen", side_effect=resps):
1762	result = runner.invoke(cli, ["hub", "proposal", "list"])
1763	assert result.exit_code == 0
1764	for i in range(5):
1765	assert f"Proposal-{i}" in result.stderr
1766
1767	def test_json_contains_all_api_fields(self, repo: pathlib.Path) -> None:
1768	"""JSON output is a passthrough — all API fields must be preserved."""
1769	self._setup(repo)
1770	proposal = {"proposalId": "abc12345-0000-0000-0000-000000000001",
1771	"title": "T", "state": "open",
1772	"fromBranch": "feat/x", "toBranch": "dev",
1773	"author": "alice", "createdAt": "2024-01-01T00:00:00Z"}
1774	resps = self._mock_api(
1775	json.dumps({"repo_id": "repo-id"}).encode(),
1776	json.dumps({"proposals": [proposal], "total": 1, "nextCursor": None}).encode(),
1777	)
1778	with patch("urllib.request.urlopen", side_effect=resps):
1779	result = runner.invoke(cli, ["hub", "proposal", "list", "-j"])
1780	assert result.exit_code == 0
1781	obj = json.loads(next(
1782	l for l in result.output.splitlines() if l.strip().startswith("{")
1783	))
1784	arr = obj["proposals"]
1785	assert arr[0]["author"] == "alice"
1786	assert arr[0]["createdAt"] == "2024-01-01T00:00:00Z"
1787	assert arr[0]["proposalId"] == "abc12345-0000-0000-0000-000000000001"
1788
1789	def test_non_dict_entries_in_proposals_array_filtered(self, repo: pathlib.Path) -> None:
1790	"""Malformed non-dict entries in the API proposals array are silently dropped."""
1791	self._setup(repo)
1792	proposals_data = {"proposals": [
1793	"not-a-dict",
1794	None,
1795	42,
1796	{"proposalId": "abc12345-0000-0000-0000-000000000001",
1797	"title": "Valid Proposal", "state": "open",
1798	"fromBranch": "feat/x", "toBranch": "dev"},
1799	], "total": 1, "nextCursor": None}
1800	resps = self._mock_api(
1801	json.dumps({"repo_id": "repo-id"}).encode(),
1802	json.dumps(proposals_data).encode(),
1803	)
1804	with patch("urllib.request.urlopen", side_effect=resps):
1805	result = runner.invoke(cli, ["hub", "proposal", "list", "-j"])
1806	assert result.exit_code == 0
1807	obj = json.loads(next(
1808	l for l in result.output.splitlines() if l.strip().startswith("{")
1809	))
1810	assert len(obj["proposals"]) == 1
1811	assert obj["proposals"][0]["title"] == "Valid Proposal"
1812
1813	def test_hub_override_flag_used(self, repo: pathlib.Path) -> None:
1814	"""``--hub`` override must be used instead of config URL."""
1815	# Set a different hub in config, then override via --hub
1816	runner.invoke(cli, ["hub", "connect", "http://localhost:11111/wrong/repo"])
1817	_store_identity("http://localhost:19999/gabriel/muse")
1818	proposals_data = {"proposals": []}
1819	resps = self._mock_api(
1820	json.dumps({"repo_id": "repo-id"}).encode(),
1821	json.dumps(proposals_data).encode(),
1822	)
1823	with patch("urllib.request.urlopen", side_effect=resps) as mock_open:
1824	result = runner.invoke(
1825	cli,
1826	["hub", "proposal", "list", "--hub", "http://localhost:19999/gabriel/muse", "-j"],
1827	)
1828	assert result.exit_code == 0
1829	# The resolved URL should contain 19999, not 11111
1830	called_urls = [c[0][0].full_url for c in mock_open.call_args_list]
1831	assert any("19999" in u for u in called_urls)
1832	assert not any("11111" in u for u in called_urls)
1833
1834	def test_proposal_list_outside_repo_exits_nonzero(
1835	self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
1836	) -> None:
1837	monkeypatch.chdir(tmp_path)
1838	monkeypatch.delenv("MUSE_REPO_ROOT", raising=False)
1839	result = runner.invoke(cli, ["hub", "proposal", "list"])
1840	assert result.exit_code != 0
1841
1842
1843	class TestFormatProposalVerbose:
1844	"""Unit tests for _format_proposal verbose mode."""
1845
1846	def test_verbose_shows_author(self) -> None:
1847	from muse.cli.commands.hub import _format_proposal
1848	proposal = {"proposalId": "abc12345", "title": "T", "state": "open",
1849	"fromBranch": "f", "toBranch": "d",
1850	"author": "alice", "createdAt": "2024-06-01T12:00:00Z"}
1851	result = _format_proposal(proposal, verbose=True)
1852	assert "alice" in result
1853
1854	def test_verbose_shows_date_prefix(self) -> None:
1855	from muse.cli.commands.hub import _format_proposal
1856	proposal = {"proposalId": "abc12345", "title": "T", "state": "open",
1857	"fromBranch": "f", "toBranch": "d",
1858	"author": "bob", "createdAt": "2024-11-30T00:00:00Z"}
1859	result = _format_proposal(proposal, verbose=True)
1860	assert "2024-11-30" in result
1861
1862	def test_verbose_ansi_in_author_stripped(self) -> None:
1863	from muse.cli.commands.hub import _format_proposal
1864	proposal = {"proposalId": "abc12345", "title": "T", "state": "open",
1865	"fromBranch": "f", "toBranch": "d",
1866	"author": "\x1b[31mmalicious\x1b[0m", "createdAt": "2024-01-01"}
1867	result = _format_proposal(proposal, verbose=True)
1868	assert "\x1b[" not in result
1869
1870	def test_verbose_ansi_in_created_at_stripped(self) -> None:
1871	from muse.cli.commands.hub import _format_proposal
1872	proposal = {"proposalId": "abc12345", "title": "T", "state": "open",
1873	"fromBranch": "f", "toBranch": "d",
1874	"author": "alice", "createdAt": "\x1b[32m2024-01-01\x1b[0m"}
1875	result = _format_proposal(proposal, verbose=True)
1876	assert "\x1b[" not in result
1877
1878	def test_verbose_false_omits_author(self) -> None:
1879	from muse.cli.commands.hub import _format_proposal
1880	proposal = {"proposalId": "abc12345", "title": "T", "state": "open",
1881	"fromBranch": "f", "toBranch": "d",
1882	"author": "alice", "createdAt": "2024-01-01"}
1883	result = _format_proposal(proposal, verbose=False)
1884	assert "alice" not in result
1885
1886	def test_verbose_missing_author_shows_fallback(self) -> None:
1887	from muse.cli.commands.hub import _format_proposal
1888	proposal = {"proposalId": "abc12345", "title": "T", "state": "open",
1889	"fromBranch": "f", "toBranch": "d"}
1890	result = _format_proposal(proposal, verbose=True)
1891	assert "?" in result # fallback when author absent
1892
1893	def test_verbose_closed_icon(self) -> None:
1894	from muse.cli.commands.hub import _format_proposal
1895	proposal = {"proposalId": "abc12345", "title": "T", "state": "closed",
1896	"fromBranch": "f", "toBranch": "d"}
1897	result = _format_proposal(proposal)
1898	assert "⛔" in result
1899
1900	def test_verbose_unknown_state_uses_fallback_icon(self) -> None:
1901	from muse.cli.commands.hub import _format_proposal
1902	proposal = {"proposalId": "abc12345", "title": "T", "state": "unknown_state",
1903	"fromBranch": "f", "toBranch": "d"}
1904	result = _format_proposal(proposal)
1905	assert "❓" in result
1906
1907	def test_proposal_id_shown_in_full(self) -> None:
1908	# Per "show full cryptographic IDs in all human-readable CLI output"
1909	# (commit 1ddad, 2026-06-12), proposal IDs are displayed in full — not
1910	# truncated — so the full ID is copyable and prefix-resolvable.
1911	from muse.cli.commands.hub import _format_proposal
1912	proposal = {"proposalId": "abc12345-full-id-here", "title": "T", "state": "open",
1913	"fromBranch": "f", "toBranch": "d"}
1914	result = _format_proposal(proposal)
1915	assert "abc12345-full-id-here" in result
1916
1917
1918	class TestProposalListStress:
1919	"""Stress tests for `muse hub proposal list`."""
1920
1921	_HUB = "http://localhost:19999/gabriel/muse"
1922
1923	def _setup(self, repo: pathlib.Path) -> None:
1924	runner.invoke(cli, ["hub", "connect", self._HUB])
1925	_store_identity(self._HUB)
1926
1927	def _make_api_resp(self, payload_bytes: bytes) -> MagicMock:
1928	mock_resp = MagicMock()
1929	mock_resp.__enter__ = lambda s: s
1930	mock_resp.__exit__ = MagicMock(return_value=False)
1931	mock_resp.read.return_value = payload_bytes
1932	return mock_resp
1933
1934	def _mock_api(self, *responses: bytes) -> list[MagicMock]:
1935	return [self._make_api_resp(r) for r in responses]
1936
1937	def test_large_proposal_list_10000_items_json(self, repo: pathlib.Path) -> None:
1938	"""10 000 proposals in the JSON response must be handled without crashing."""
1939	self._setup(repo)
1940	proposals = [
1941	{"proposalId": f"aaaa0000-0000-0000-0000-{i:012d}",
1942	"title": f"Proposal #{i}", "state": "open",
1943	"fromBranch": f"feat/f{i}", "toBranch": "dev"}
1944	for i in range(10_000)
1945	]
1946	payload = json.dumps({"proposals": proposals, "total": 10_000, "nextCursor": None}).encode()
1947	mock_resp = MagicMock()
1948	mock_resp.__enter__ = lambda s: s
1949	mock_resp.__exit__ = MagicMock(return_value=False)
1950	mock_resp.read.return_value = payload
1951
1952	repo_resp = self._make_api_resp(json.dumps({"repo_id": "repo-id"}).encode())
1953	with patch("urllib.request.urlopen", side_effect=[repo_resp, mock_resp]):
1954	result = runner.invoke(cli, ["hub", "proposal", "list", "-n", "10000", "-j"])
1955	assert result.exit_code == 0
1956	obj = json.loads(next(
1957	l for l in result.output.splitlines() if l.strip().startswith("{")
1958	))
1959	assert len(obj["proposals"]) == 10_000
1960
1961	def test_concurrent_format_proposal_calls(self) -> None:
1962	"""8 threads calling _format_proposal concurrently must produce consistent results."""
1963	from muse.cli.commands.hub import _format_proposal
1964	errors: list[str] = []
1965	results: list[str] = [""] * 8
1966
1967	def _do(idx: int) -> None:
1968	try:
1969	proposal = {
1970	"proposalId": f"aaaa{idx:04d}-0000-0000-0000-000000000001",
1971	"title": f"Proposal-{idx}: \x1b[31mmalicious\x1b[0m",
1972	"state": "open",
1973	"fromBranch": f"feat/f{idx}",
1974	"toBranch": "dev",
1975	"author": f"user{idx}",
1976	"createdAt": f"2024-0{(idx % 9) + 1}-01T00:00:00Z",
1977	}
1978	results[idx] = _format_proposal(proposal, verbose=True)
1979	except Exception as exc:
1980	errors.append(f"Thread {idx}: {exc}")
1981
1982	threads = [threading.Thread(target=_do, args=(i,)) for i in range(8)]
1983	for t in threads:
1984	t.start()
1985	for t in threads:
1986	t.join()
1987	assert errors == [], f"Concurrent _format_proposal failures:\n{'\n'.join(errors)}"
1988	# Each result must have ANSI stripped and contain the user name
1989	for idx, result in enumerate(results):
1990	assert "\x1b[" not in result, f"ANSI in thread {idx} output"
1991	assert f"user{idx}" in result, f"Author missing in thread {idx} output"
1992
1993
1994	class TestProposalListE2E:
1995	"""End-to-end flow tests for `muse hub proposal list`."""
1996
1997	_HUB = "http://localhost:19999/gabriel/muse"
1998
1999	def _setup(self, repo: pathlib.Path) -> None:
2000	runner.invoke(cli, ["hub", "connect", self._HUB])
2001	_store_identity(self._HUB)
2002
2003	def _make_api_resp(self, payload_bytes: bytes) -> MagicMock:
2004	mock_resp = MagicMock()
2005	mock_resp.__enter__ = lambda s: s
2006	mock_resp.__exit__ = MagicMock(return_value=False)
2007	mock_resp.read.return_value = payload_bytes
2008	return mock_resp
2009
2010	def _mock_api(self, *responses: bytes) -> list[MagicMock]:
2011	return [self._make_api_resp(r) for r in responses]
2012
2013	def test_e2e_connect_then_list_json(self, repo: pathlib.Path) -> None:
2014	"""Full flow: connect → list --json returns a well-formed envelope object."""
2015	self._setup(repo)
2016	proposals_data = {"proposals": [
2017	{"proposalId": "abc12345-0000-0000-0000-000000000001",
2018	"title": "My Proposal", "state": "open",
2019	"fromBranch": "feat/my", "toBranch": "dev",
2020	"author": "alice", "createdAt": "2024-03-01T09:00:00Z"},
2021	], "total": 1, "nextCursor": None}
2022	resps = self._mock_api(
2023	json.dumps({"repo_id": "repo-id"}).encode(),
2024	json.dumps(proposals_data).encode(),
2025	)
2026	with patch("urllib.request.urlopen", side_effect=resps):
2027	result = runner.invoke(cli, ["hub", "proposal", "list", "-j"])
2028	assert result.exit_code == 0
2029	obj = json.loads(next(
2030	l for l in result.output.splitlines() if l.strip().startswith("{")
2031	))
2032	arr = obj["proposals"]
2033	assert arr[0]["title"] == "My Proposal"
2034	assert arr[0]["state"] == "open"
2035	assert arr[0]["author"] == "alice"
2036
2037	def test_e2e_list_verbose_text_all_fields_present(self, repo: pathlib.Path) -> None:
2038	"""Verbose text output includes state icon, ID prefix, branches, author, date."""
2039	self._setup(repo)
2040	proposals_data = {"proposals": [
2041	{"proposalId": "deadbeef-0000-0000-0000-000000000001",
2042	"title": "My feature", "state": "open",
2043	"fromBranch": "feat/my-feature", "toBranch": "dev",
2044	"author": "charlie", "createdAt": "2025-12-31T23:59:59Z"},
2045	]}
2046	resps = self._mock_api(
2047	json.dumps({"repo_id": "repo-id"}).encode(),
2048	json.dumps(proposals_data).encode(),
2049	)
2050	with patch("urllib.request.urlopen", side_effect=resps):
2051	result = runner.invoke(cli, ["hub", "proposal", "list", "-v"])
2052	assert result.exit_code == 0
2053	output = result.stderr
2054	assert "🟢" in output
2055	assert "deadbeef" in output
2056	assert "feat/my-feature" in output
2057	assert "charlie" in output
2058	assert "2025-12-31" in output
2059
2060	def test_e2e_empty_list_exits_zero_with_message(self, repo: pathlib.Path) -> None:
2061	"""Empty proposal list must exit 0 and print a human-friendly message."""
2062	self._setup(repo)
2063	resps = self._mock_api(
2064	json.dumps({"repo_id": "repo-id"}).encode(),
2065	json.dumps({"proposals": []}).encode(),
2066	)
2067	with patch("urllib.request.urlopen", side_effect=resps):
2068	result = runner.invoke(cli, ["hub", "proposal", "list", "--state", "merged"])
2069	assert result.exit_code == 0
2070	assert "No proposals" in result.stderr or "no proposals" in result.stderr.lower()
2071
2072	def test_e2e_json_no_stdout_in_text_mode(self, repo: pathlib.Path) -> None:
2073	"""In text mode, JSON must NOT appear on stdout — all output goes to stderr."""
2074	self._setup(repo)
2075	proposals_data = {"proposals": [
2076	{"proposalId": "abc12345-0000-0000-0000-000000000001",
2077	"title": "T", "state": "open",
2078	"fromBranch": "feat/x", "toBranch": "dev"},
2079	]}
2080	resps = self._mock_api(
2081	json.dumps({"repo_id": "repo-id"}).encode(),
2082	json.dumps(proposals_data).encode(),
2083	)
2084	with patch("urllib.request.urlopen", side_effect=resps):
2085	result = runner.invoke(cli, ["hub", "proposal", "list"])
2086	assert result.exit_code == 0
2087	# In text mode, stdout should have no JSON array
2088	for line in result.output.splitlines():
2089	stripped = line.strip()
2090	assert not stripped.startswith("["), (
2091	f"Unexpected JSON on stdout in text mode: {stripped!r}"
2092	)
2093
2094
2095	class TestProposalViewHardening:
2096	"""Additional hardening tests for `muse hub proposal show`."""
2097
2098	_HUB = "http://localhost:19999/gabriel/muse"
2099
2100	def _setup(self, repo: pathlib.Path) -> None:
2101	runner.invoke(cli, ["hub", "connect", self._HUB])
2102	_store_identity(self._HUB)
2103
2104	def _make_api_resp(self, payload_bytes: bytes) -> MagicMock:
2105	mock_resp = MagicMock()
2106	mock_resp.__enter__ = lambda s: s
2107	mock_resp.__exit__ = MagicMock(return_value=False)
2108	mock_resp.read.return_value = payload_bytes
2109	return mock_resp
2110
2111	def _mock_api(self, *responses: bytes) -> list[MagicMock]:
2112	return [self._make_api_resp(r) for r in responses]
2113
2114	def test_short_flag_j_works_for_view(self, repo: pathlib.Path) -> None:
2115	"""``-j`` is accepted as alias for ``--json``."""
2116	self._setup(repo)
2117	proposal_id = "abc12345-0000-0000-0000-000000000001"
2118	proposal_data = {"proposalId": proposal_id, "title": "T", "state": "open",
2119	"fromBranch": "feat/x", "toBranch": "dev"}
2120	proposals_data = {"proposals": [
2121	{"proposalId": proposal_id, "title": "T", "state": "open",
2122	"fromBranch": "feat/x", "toBranch": "dev"},
2123	]}
2124	resps = self._mock_api(
2125	json.dumps({"repo_id": "repo-id"}).encode(),
2126	json.dumps(proposals_data).encode(),
2127	json.dumps(proposal_data).encode(),
2128	)
2129	with patch("urllib.request.urlopen", side_effect=resps):
2130	result = runner.invoke(cli, ["hub", "proposal", "read", "abc12345", "-j"])
2131	assert result.exit_code == 0
2132	json_lines = [l for l in result.output.splitlines() if l.strip().startswith("{")]
2133	assert len(json_lines) >= 1
2134
2135	def test_ansi_in_state_sanitized(
2136	self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]
2137	) -> None:
2138	"""ANSI in ``state`` field must not reach terminal in text mode."""
2139	self._setup(repo)
2140	proposal_id = "abc12345-0000-0000-0000-000000000001"
2141	malicious_proposal = {"proposalId": proposal_id, "title": "T",
2142	"state": "\x1b[31mopen\x1b[0m",
2143	"fromBranch": "feat/x", "toBranch": "dev"}
2144	proposals_data = {"proposals": [
2145	{"proposalId": proposal_id, "title": "T", "state": "open",
2146	"fromBranch": "feat/x", "toBranch": "dev"},
2147	]}
2148	resps = self._mock_api(
2149	json.dumps({"repo_id": "repo-id"}).encode(),
2150	json.dumps(proposals_data).encode(),
2151	json.dumps(malicious_proposal).encode(),
2152	)
2153	with patch("urllib.request.urlopen", side_effect=resps):
2154	result = runner.invoke(cli, ["hub", "proposal", "read", "abc12345"])
2155	assert result.exit_code == 0
2156	assert "\x1b[" not in result.stderr
2157
2158	def test_ansi_in_branch_sanitized(self, repo: pathlib.Path) -> None:
2159	"""ANSI in branch names must not reach terminal in text mode."""
2160	self._setup(repo)
2161	proposal_id = "abc12345-0000-0000-0000-000000000001"
2162	malicious_proposal = {"proposalId": proposal_id, "title": "T", "state": "open",
2163	"fromBranch": "\x1b[32mfeat/malicious\x1b[0m",
2164	"toBranch": "\x1b[34mdev\x1b[0m"}
2165	proposals_data = {"proposals": [
2166	{"proposalId": proposal_id, "title": "T", "state": "open",
2167	"fromBranch": "feat/x", "toBranch": "dev"},
2168	]}
2169	resps = self._mock_api(
2170	json.dumps({"repo_id": "repo-id"}).encode(),
2171	json.dumps(proposals_data).encode(),
2172	json.dumps(malicious_proposal).encode(),
2173	)
2174	with patch("urllib.request.urlopen", side_effect=resps):
2175	result = runner.invoke(cli, ["hub", "proposal", "read", "abc12345"])
2176	assert result.exit_code == 0
2177	assert "\x1b[" not in result.stderr
2178
2179	def test_ansi_in_body_lines_sanitized(self, repo: pathlib.Path) -> None:
2180	"""ANSI in body text must not reach terminal in text mode."""
2181	self._setup(repo)
2182	proposal_id = "abc12345-0000-0000-0000-000000000001"
2183	malicious_proposal = {"proposalId": proposal_id, "title": "T", "state": "open",
2184	"fromBranch": "feat/x", "toBranch": "dev",
2185	"body": "\x1b[31mThis body has ANSI\x1b[0m"}
2186	proposals_data = {"proposals": [
2187	{"proposalId": proposal_id, "title": "T", "state": "open",
2188	"fromBranch": "feat/x", "toBranch": "dev"},
2189	]}
2190	resps = self._mock_api(
2191	json.dumps({"repo_id": "repo-id"}).encode(),
2192	json.dumps(proposals_data).encode(),
2193	json.dumps(malicious_proposal).encode(),
2194	)
2195	with patch("urllib.request.urlopen", side_effect=resps):
2196	result = runner.invoke(cli, ["hub", "proposal", "read", "abc12345"])
2197	assert result.exit_code == 0
2198	assert "\x1b[" not in result.stderr
2199
2200	def test_view_prefix_not_found_exits_nonzero(self, repo: pathlib.Path) -> None:
2201	self._setup(repo)
2202	proposals_data = {"proposals": []}
2203	resps = self._mock_api(
2204	json.dumps({"repo_id": "repo-id"}).encode(),
2205	json.dumps(proposals_data).encode(),
2206	)
2207	with patch("urllib.request.urlopen", side_effect=resps):
2208	result = runner.invoke(cli, ["hub", "proposal", "read", "deadbeef"])
2209	assert result.exit_code != 0
2210
2211	def test_view_no_hub_exits_nonzero(self, repo: pathlib.Path) -> None:
2212	result = runner.invoke(cli, ["hub", "proposal", "read", "abc12345"])
2213	assert result.exit_code != 0
2214
2215	def test_view_no_identity_exits_nonzero(self, repo: pathlib.Path) -> None:
2216	runner.invoke(cli, ["hub", "connect", self._HUB])
2217	result = runner.invoke(cli, ["hub", "proposal", "read", "abc12345"])
2218	assert result.exit_code != 0
2219
2220	def test_view_outside_repo_exits_nonzero(
2221	self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
2222	) -> None:
2223	monkeypatch.chdir(tmp_path)
2224	monkeypatch.delenv("MUSE_REPO_ROOT", raising=False)
2225	result = runner.invoke(cli, ["hub", "proposal", "read", "abc12345"])
2226	assert result.exit_code != 0
2227
2228	def test_full_id_skips_prefix_resolution(self, repo: pathlib.Path) -> None:
2229	"""A full proposal ID must reach the view endpoint with exactly 2 API calls (no prefix fetch)."""
2230	self._setup(repo)
2231	proposal_id = "abc12345-def0-0000-0000-000000000001"
2232	proposal_data = {"proposalId": proposal_id, "title": "T", "state": "open",
2233	"fromBranch": "feat/x", "toBranch": "dev"}
2234	resps = self._mock_api(
2235	json.dumps({"repo_id": "repo-id"}).encode(), # _resolve_repo_id
2236	json.dumps(proposal_data).encode(), # GET proposals/{id}
2237	)
2238	with patch("urllib.request.urlopen", side_effect=resps) as mock_open:
2239	result = runner.invoke(cli, ["hub", "proposal", "read", proposal_id, "-j"])
2240	assert result.exit_code == 0
2241	# Only 2 urlopen calls: repo resolution + the view fetch (no prefix list call)
2242	assert mock_open.call_count == 2
2243
2244	def test_prefix_triggers_resolution_call(self, repo: pathlib.Path) -> None:
2245	"""An 8-char prefix must trigger a prefix-resolution list fetch (3 API calls total)."""
2246	self._setup(repo)
2247	proposal_id = "abc12345-0000-0000-0000-000000000001"
2248	proposal_data = {"proposalId": proposal_id, "title": "T", "state": "open",
2249	"fromBranch": "feat/x", "toBranch": "dev"}
2250	proposals_data = {"proposals": [
2251	{"proposalId": proposal_id, "title": "T", "state": "open",
2252	"fromBranch": "feat/x", "toBranch": "dev"},
2253	]}
2254	resps = self._mock_api(
2255	json.dumps({"repo_id": "repo-id"}).encode(), # _resolve_repo_id
2256	json.dumps(proposals_data).encode(), # prefix resolution list
2257	json.dumps(proposal_data).encode(), # GET proposals/{id}
2258	)
2259	with patch("urllib.request.urlopen", side_effect=resps) as mock_open:
2260	result = runner.invoke(cli, ["hub", "proposal", "read", "abc12345", "-j"])
2261	assert result.exit_code == 0
2262	assert mock_open.call_count == 3
2263
2264	def test_author_shown_in_text_mode(self, repo: pathlib.Path) -> None:
2265	self._setup(repo)
2266	proposal_id = "abc12345-0000-0000-0000-000000000001"
2267	proposal_data = {"proposalId": proposal_id, "title": "T", "state": "open",
2268	"fromBranch": "feat/x", "toBranch": "dev",
2269	"author": "charlie", "createdAt": "2024-07-04T00:00:00Z"}
2270	resps = self._mock_api(
2271	json.dumps({"repo_id": "repo-id"}).encode(),
2272	json.dumps({"proposals": [
2273	{"proposalId": proposal_id, "title": "T", "state": "open",
2274	"fromBranch": "feat/x", "toBranch": "dev"},
2275	]}).encode(),
2276	json.dumps(proposal_data).encode(),
2277	)
2278	with patch("urllib.request.urlopen", side_effect=resps):
2279	result = runner.invoke(cli, ["hub", "proposal", "read", "abc12345"])
2280	assert result.exit_code == 0
2281	assert "charlie" in result.stderr
2282
2283	def test_created_at_shown_in_text_mode(self, repo: pathlib.Path) -> None:
2284	self._setup(repo)
2285	proposal_id = "abc12345-0000-0000-0000-000000000001"
2286	proposal_data = {"proposalId": proposal_id, "title": "T", "state": "open",
2287	"fromBranch": "feat/x", "toBranch": "dev",
2288	"author": "alice", "createdAt": "2025-03-15T08:30:00Z"}
2289	resps = self._mock_api(
2290	json.dumps({"repo_id": "repo-id"}).encode(),
2291	json.dumps({"proposals": [
2292	{"proposalId": proposal_id, "title": "T", "state": "open",
2293	"fromBranch": "feat/x", "toBranch": "dev"},
2294	]}).encode(),
2295	json.dumps(proposal_data).encode(),
2296	)
2297	with patch("urllib.request.urlopen", side_effect=resps):
2298	result = runner.invoke(cli, ["hub", "proposal", "read", "abc12345"])
2299	assert result.exit_code == 0
2300	assert "2025-03-15" in result.stderr
2301
2302	def test_ansi_in_author_sanitized(self, repo: pathlib.Path) -> None:
2303	self._setup(repo)
2304	proposal_id = "abc12345-0000-0000-0000-000000000001"
2305	proposal_data = {"proposalId": proposal_id, "title": "T", "state": "open",
2306	"fromBranch": "feat/x", "toBranch": "dev",
2307	"author": "\x1b[31mmalicious-author\x1b[0m",
2308	"createdAt": "2024-01-01T00:00:00Z"}
2309	resps = self._mock_api(
2310	json.dumps({"repo_id": "repo-id"}).encode(),
2311	json.dumps({"proposals": [
2312	{"proposalId": proposal_id, "title": "T", "state": "open",
2313	"fromBranch": "feat/x", "toBranch": "dev"},
2314	]}).encode(),
2315	json.dumps(proposal_data).encode(),
2316	)
2317	with patch("urllib.request.urlopen", side_effect=resps):
2318	result = runner.invoke(cli, ["hub", "proposal", "read", "abc12345"])
2319	assert result.exit_code == 0
2320	assert "\x1b[" not in result.stderr
2321
2322	def test_ansi_in_created_at_sanitized(self, repo: pathlib.Path) -> None:
2323	self._setup(repo)
2324	proposal_id = "abc12345-0000-0000-0000-000000000001"
2325	proposal_data = {"proposalId": proposal_id, "title": "T", "state": "open",
2326	"fromBranch": "feat/x", "toBranch": "dev",
2327	"author": "alice",
2328	"createdAt": "\x1b[32m2024-01-01\x1b[0mTmalicious"}
2329	resps = self._mock_api(
2330	json.dumps({"repo_id": "repo-id"}).encode(),
2331	json.dumps({"proposals": [
2332	{"proposalId": proposal_id, "title": "T", "state": "open",
2333	"fromBranch": "feat/x", "toBranch": "dev"},
2334	]}).encode(),
2335	json.dumps(proposal_data).encode(),
2336	)
2337	with patch("urllib.request.urlopen", side_effect=resps):
2338	result = runner.invoke(cli, ["hub", "proposal", "read", "abc12345"])
2339	assert result.exit_code == 0
2340	assert "\x1b[" not in result.stderr
2341
2342	def test_body_truncation_hint_shown(self, repo: pathlib.Path) -> None:
2343	"""Body exceeding _MAX_PROPOSAL_BODY_LINES must show a truncation hint."""
2344	from muse.cli.commands.hub import _MAX_PROPOSAL_BODY_LINES
2345	self._setup(repo)
2346	proposal_id = "abc12345-0000-0000-0000-000000000001"
2347	long_body = "\n".join(f"line {i}" for i in range(_MAX_PROPOSAL_BODY_LINES + 5))
2348	proposal_data = {"proposalId": proposal_id, "title": "T", "state": "open",
2349	"fromBranch": "feat/x", "toBranch": "dev", "body": long_body}
2350	resps = self._mock_api(
2351	json.dumps({"repo_id": "repo-id"}).encode(),
2352	json.dumps({"proposals": [
2353	{"proposalId": proposal_id, "title": "T", "state": "open",
2354	"fromBranch": "feat/x", "toBranch": "dev"},
2355	]}).encode(),
2356	json.dumps(proposal_data).encode(),
2357	)
2358	with patch("urllib.request.urlopen", side_effect=resps):
2359	result = runner.invoke(cli, ["hub", "proposal", "read", "abc12345"])
2360	assert result.exit_code == 0
2361	assert "more line" in result.stderr
2362	assert "--json" in result.stderr # hint mentions --json
2363
2364	def test_body_exactly_at_limit_no_hint(self, repo: pathlib.Path) -> None:
2365	"""Body at exactly _MAX_PROPOSAL_BODY_LINES must NOT show a truncation hint."""
2366	from muse.cli.commands.hub import _MAX_PROPOSAL_BODY_LINES
2367	self._setup(repo)
2368	proposal_id = "abc12345-0000-0000-0000-000000000001"
2369	exact_body = "\n".join(f"line {i}" for i in range(_MAX_PROPOSAL_BODY_LINES))
2370	proposal_data = {"proposalId": proposal_id, "title": "T", "state": "open",
2371	"fromBranch": "feat/x", "toBranch": "dev", "body": exact_body}
2372	resps = self._mock_api(
2373	json.dumps({"repo_id": "repo-id"}).encode(),
2374	json.dumps({"proposals": [
2375	{"proposalId": proposal_id, "title": "T", "state": "open",
2376	"fromBranch": "feat/x", "toBranch": "dev"},
2377	]}).encode(),
2378	json.dumps(proposal_data).encode(),
2379	)
2380	with patch("urllib.request.urlopen", side_effect=resps):
2381	result = runner.invoke(cli, ["hub", "proposal", "read", "abc12345"])
2382	assert result.exit_code == 0
2383	assert "more line" not in result.stderr
2384
2385	def test_no_body_field_no_body_section(self, repo: pathlib.Path) -> None:
2386	"""When body is absent or empty, no 'Body:' section must appear."""
2387	self._setup(repo)
2388	proposal_id = "abc12345-0000-0000-0000-000000000001"
2389	proposal_data = {"proposalId": proposal_id, "title": "T", "state": "open",
2390	"fromBranch": "feat/x", "toBranch": "dev"}
2391	resps = self._mock_api(
2392	json.dumps({"repo_id": "repo-id"}).encode(),
2393	json.dumps({"proposals": [
2394	{"proposalId": proposal_id, "title": "T", "state": "open",
2395	"fromBranch": "feat/x", "toBranch": "dev"},
2396	]}).encode(),
2397	json.dumps(proposal_data).encode(),
2398	)
2399	with patch("urllib.request.urlopen", side_effect=resps):
2400	result = runner.invoke(cli, ["hub", "proposal", "read", "abc12345"])
2401	assert result.exit_code == 0
2402	assert "Body:" not in result.stderr
2403
2404	def test_json_passthrough_includes_all_fields(self, repo: pathlib.Path) -> None:
2405	"""JSON output must be an unmodified passthrough from the API."""
2406	self._setup(repo)
2407	proposal_id = "abc12345-0000-0000-0000-000000000001"
2408	proposal_data = {"proposalId": proposal_id, "title": "My Proposal", "state": "open",
2409	"fromBranch": "feat/x", "toBranch": "dev",
2410	"author": "alice", "createdAt": "2024-01-01T00:00:00Z",
2411	"body": "Full body text here.",
2412	"extraField": "agent-visible"}
2413	resps = self._mock_api(
2414	json.dumps({"repo_id": "repo-id"}).encode(),
2415	json.dumps({"proposals": [
2416	{"proposalId": proposal_id, "title": "My Proposal", "state": "open",
2417	"fromBranch": "feat/x", "toBranch": "dev"},
2418	]}).encode(),
2419	json.dumps(proposal_data).encode(),
2420	)
2421	with patch("urllib.request.urlopen", side_effect=resps):
2422	result = runner.invoke(cli, ["hub", "proposal", "read", "abc12345", "-j"])
2423	assert result.exit_code == 0
2424	data = json.loads(next(
2425	l for l in result.output.splitlines() if l.strip().startswith("{")
2426	))
2427	assert data["author"] == "alice"
2428	assert data["body"] == "Full body text here."
2429	assert data["extraField"] == "agent-visible"
2430
2431	def test_hub_override_flag(self, repo: pathlib.Path) -> None:
2432	"""``--hub`` must route requests to the override URL."""
2433	runner.invoke(cli, ["hub", "connect", "http://localhost:11111/wrong/repo"])
2434	_store_identity("http://localhost:19999/gabriel/muse")
2435	proposal_id = "abc12345-def0-0000-0000-000000000001"
2436	proposal_data = {"proposalId": proposal_id, "title": "T", "state": "open",
2437	"fromBranch": "f", "toBranch": "d"}
2438	resps = self._mock_api(
2439	json.dumps({"repo_id": "repo-id"}).encode(),
2440	json.dumps(proposal_data).encode(),
2441	)
2442	with patch("urllib.request.urlopen", side_effect=resps) as mock_open:
2443	result = runner.invoke(
2444	cli,
2445	["hub", "proposal", "read", proposal_id,
2446	"--hub", "http://localhost:19999/gabriel/muse", "-j"],
2447	)
2448	assert result.exit_code == 0
2449	called_urls = [c[0][0].full_url for c in mock_open.call_args_list]
2450	assert any("19999" in u for u in called_urls)
2451	assert not any("11111" in u for u in called_urls)
2452
2453
2454	class TestProposalViewUnit:
2455	"""Pure unit tests for run_proposal_show text rendering logic."""
2456
2457	def _make_proposal_resp(self, **kwargs: str) -> bytes:
2458	base: Manifest = {"proposalId": "abc12345-0000-0000-0000-000000000001",
2459	"title": "My Proposal", "state": "open",
2460	"fromBranch": "feat/x", "toBranch": "dev"}
2461	base.update(kwargs)
2462	return json.dumps(base).encode()
2463
2464	def _invoke_view(
2465	self,
2466	repo: pathlib.Path,
2467	proposal_data: bytes,
2468	*,
2469	flags: list[str] \| None = None,
2470	) -> InvokeResult:
2471	"""Invoke hub proposal show with a pre-resolved full UUID (2 API calls only)."""
2472	proposal_id = "abc12345-def0-0000-0000-000000000001"
2473	# Use a full UUID to skip the prefix-resolution fetch
2474	mock_repo = MagicMock()
2475	mock_repo.__enter__ = lambda s: s
2476	mock_repo.__exit__ = MagicMock(return_value=False)
2477	mock_repo.read.return_value = json.dumps({"repo_id": "repo-id"}).encode()
2478
2479	mock_proposal = MagicMock()
2480	mock_proposal.__enter__ = lambda s: s
2481	mock_proposal.__exit__ = MagicMock(return_value=False)
2482	mock_proposal.read.return_value = proposal_data
2483
2484	cmd = ["hub", "proposal", "read", proposal_id] + (flags or [])
2485	with patch("urllib.request.urlopen", side_effect=[mock_repo, mock_proposal]):
2486	return runner.invoke(cli, cmd)
2487
2488	def test_state_open_icon(self, repo: pathlib.Path) -> None:
2489	runner.invoke(cli, ["hub", "connect", "http://localhost:19999/gabriel/muse"])
2490	_store_identity("http://localhost:19999/gabriel/muse")
2491	result = self._invoke_view(repo, self._make_proposal_resp(state="open"))
2492	assert "🟢" in result.stderr
2493
2494	def test_state_merged_icon(self, repo: pathlib.Path) -> None:
2495	runner.invoke(cli, ["hub", "connect", "http://localhost:19999/gabriel/muse"])
2496	_store_identity("http://localhost:19999/gabriel/muse")
2497	result = self._invoke_view(repo, self._make_proposal_resp(state="merged"))
2498	assert "🟣" in result.stderr
2499
2500	def test_state_closed_icon(self, repo: pathlib.Path) -> None:
2501	runner.invoke(cli, ["hub", "connect", "http://localhost:19999/gabriel/muse"])
2502	_store_identity("http://localhost:19999/gabriel/muse")
2503	result = self._invoke_view(repo, self._make_proposal_resp(state="closed"))
2504	assert "⛔" in result.stderr
2505
2506	def test_unknown_state_fallback_icon(self, repo: pathlib.Path) -> None:
2507	runner.invoke(cli, ["hub", "connect", "http://localhost:19999/gabriel/muse"])
2508	_store_identity("http://localhost:19999/gabriel/muse")
2509	result = self._invoke_view(repo, self._make_proposal_resp(state="draft"))
2510	assert "❓" in result.stderr
2511
2512	def test_no_author_field_omits_by_line(self, repo: pathlib.Path) -> None:
2513	"""When author is absent, the 'By:' line must not appear."""
2514	runner.invoke(cli, ["hub", "connect", "http://localhost:19999/gabriel/muse"])
2515	_store_identity("http://localhost:19999/gabriel/muse")
2516	result = self._invoke_view(repo, self._make_proposal_resp())
2517	assert "By:" not in result.stderr
2518
2519	def test_state_upper_in_header(self, repo: pathlib.Path) -> None:
2520	runner.invoke(cli, ["hub", "connect", "http://localhost:19999/gabriel/muse"])
2521	_store_identity("http://localhost:19999/gabriel/muse")
2522	result = self._invoke_view(repo, self._make_proposal_resp(state="open"))
2523	assert "[OPEN]" in result.stderr
2524
2525	def test_id_and_branches_in_output(self, repo: pathlib.Path) -> None:
2526	runner.invoke(cli, ["hub", "connect", "http://localhost:19999/gabriel/muse"])
2527	_store_identity("http://localhost:19999/gabriel/muse")
2528	proposal_id = "abc12345-def0-0000-0000-000000000001"
2529	result = self._invoke_view(
2530	repo,
2531	self._make_proposal_resp(proposalId=proposal_id, fromBranch="feat/my", toBranch="main"),
2532	)
2533	assert "feat/my" in result.stderr
2534	assert "main" in result.stderr
2535
2536
2537	class TestProposalViewE2E:
2538	"""End-to-end scenario tests for `muse hub proposal show`."""
2539
2540	_HUB = "http://localhost:19999/gabriel/muse"
2541
2542	def _setup(self, repo: pathlib.Path) -> None:
2543	runner.invoke(cli, ["hub", "connect", self._HUB])
2544	_store_identity(self._HUB)
2545
2546	def _make_api_resp(self, payload_bytes: bytes) -> MagicMock:
2547	mock_resp = MagicMock()
2548	mock_resp.__enter__ = lambda s: s
2549	mock_resp.__exit__ = MagicMock(return_value=False)
2550	mock_resp.read.return_value = payload_bytes
2551	return mock_resp
2552
2553	def _mock_api(self, *responses: bytes) -> list[MagicMock]:
2554	return [self._make_api_resp(r) for r in responses]
2555
2556	def test_e2e_full_proposal_text_output(self, repo: pathlib.Path) -> None:
2557	"""Full flow with all optional fields — all sections must appear."""
2558	self._setup(repo)
2559	proposal_id = "deadbeef-cafe-0000-0000-000000000001"
2560	proposal_data = {
2561	"proposalId": proposal_id,
2562	"title": "feat: add sonic synthesis",
2563	"state": "open",
2564	"fromBranch": "feat/sonic",
2565	"toBranch": "dev",
2566	"author": "gabriel",
2567	"createdAt": "2025-06-01T12:00:00Z",
2568	"body": "This proposal adds sonic synthesis support.",
2569	}
2570	resps = self._mock_api(
2571	json.dumps({"repo_id": "repo-id"}).encode(),
2572	json.dumps(proposal_data).encode(),
2573	)
2574	with patch("urllib.request.urlopen", side_effect=resps):
2575	result = runner.invoke(cli, ["hub", "proposal", "read", proposal_id])
2576	assert result.exit_code == 0
2577	output = result.stderr
2578	assert "🟢" in output
2579	assert "feat: add sonic synthesis" in output
2580	assert "feat/sonic" in output
2581	assert "gabriel" in output
2582	assert "2025-06-01" in output
2583	assert "sonic synthesis support" in output
2584
2585	def test_e2e_json_agent_workflow(self, repo: pathlib.Path) -> None:
2586	"""Simulate an agent extracting state via --json \| jq."""
2587	self._setup(repo)
2588	proposal_id = "deadbeef-cafe-0000-0000-000000000001"
2589	proposal_data = {"proposalId": proposal_id, "title": "T", "state": "merged",
2590	"fromBranch": "feat/x", "toBranch": "dev",
2591	"author": "bot", "mergeCommitId": "aabbccdd11223344"}
2592	resps = self._mock_api(
2593	json.dumps({"repo_id": "repo-id"}).encode(),
2594	json.dumps(proposal_data).encode(),
2595	)
2596	with patch("urllib.request.urlopen", side_effect=resps):
2597	result = runner.invoke(cli, ["hub", "proposal", "read", proposal_id, "--json"])
2598	assert result.exit_code == 0
2599	data = json.loads(next(
2600	l for l in result.output.splitlines() if l.strip().startswith("{")
2601	))
2602	assert data["state"] == "merged"
2603	assert data["mergeCommitId"] == "aabbccdd11223344"
2604
2605	def test_e2e_body_truncation_hint_points_to_json(self, repo: pathlib.Path) -> None:
2606	"""Truncation hint must explicitly mention --json."""
2607	from muse.cli.commands.hub import _MAX_PROPOSAL_BODY_LINES
2608	self._setup(repo)
2609	proposal_id = "deadbeef-cafe-0000-0000-000000000001"
2610	long_body = "\n".join(f"line {i}" for i in range(_MAX_PROPOSAL_BODY_LINES + 10))
2611	proposal_data = {"proposalId": proposal_id, "title": "T", "state": "open",
2612	"fromBranch": "feat/x", "toBranch": "dev", "body": long_body}
2613	resps = self._mock_api(
2614	json.dumps({"repo_id": "repo-id"}).encode(),
2615	json.dumps(proposal_data).encode(),
2616	)
2617	with patch("urllib.request.urlopen", side_effect=resps):
2618	result = runner.invoke(cli, ["hub", "proposal", "read", proposal_id])
2619	assert result.exit_code == 0
2620	assert "--json" in result.stderr
2621	assert "10 more line" in result.stderr
2622
2623	def test_e2e_ambiguous_prefix_exits_nonzero(self, repo: pathlib.Path) -> None:
2624	"""Two proposals with the same prefix must cause a non-zero exit."""
2625	self._setup(repo)
2626	proposals_data = {"proposals": [
2627	{"proposalId": "abc12345-0000-0000-0000-000000000001", "title": "Proposal 1",
2628	"state": "open", "fromBranch": "feat/a", "toBranch": "dev"},
2629	{"proposalId": "abc12345-0000-0000-0000-000000000002", "title": "Proposal 2",
2630	"state": "open", "fromBranch": "feat/b", "toBranch": "dev"},
2631	]}
2632	resps = self._mock_api(
2633	json.dumps({"repo_id": "repo-id"}).encode(),
2634	json.dumps(proposals_data).encode(),
2635	)
2636	with patch("urllib.request.urlopen", side_effect=resps):
2637	result = runner.invoke(cli, ["hub", "proposal", "read", "abc12345"])
2638	assert result.exit_code != 0
2639
2640
2641	class TestProposalViewStress:
2642	"""Stress tests for `muse hub proposal show`."""
2643
2644	_HUB = "http://localhost:19999/gabriel/muse"
2645
2646	def test_body_with_1000_lines_truncated(self, repo: pathlib.Path) -> None:
2647	"""A 1000-line body must be accepted without OOM and truncated correctly."""
2648	from muse.cli.commands.hub import _MAX_PROPOSAL_BODY_LINES
2649
2650	runner.invoke(cli, ["hub", "connect", self._HUB])
2651	_store_identity(self._HUB)
2652
2653	proposal_id = "deadbeef-cafe-0000-0000-000000000001"
2654	big_body = "\n".join(f"line {i}" for i in range(1000))
2655	proposal_data = {"proposalId": proposal_id, "title": "T", "state": "open",
2656	"fromBranch": "feat/x", "toBranch": "dev", "body": big_body}
2657
2658	mock_repo = MagicMock()
2659	mock_repo.__enter__ = lambda s: s
2660	mock_repo.__exit__ = MagicMock(return_value=False)
2661	mock_repo.read.return_value = json.dumps({"repo_id": "repo-id"}).encode()
2662
2663	mock_proposal = MagicMock()
2664	mock_proposal.__enter__ = lambda s: s
2665	mock_proposal.__exit__ = MagicMock(return_value=False)
2666	mock_proposal.read.return_value = json.dumps(proposal_data).encode()
2667
2668	with patch("urllib.request.urlopen", side_effect=[mock_repo, mock_proposal]):
2669	result = runner.invoke(cli, ["hub", "proposal", "read", proposal_id])
2670	assert result.exit_code == 0
2671	lines_shown = [l for l in result.stderr.splitlines() if l.strip().startswith("line ")]
2672	assert len(lines_shown) == _MAX_PROPOSAL_BODY_LINES
2673	assert "more line" in result.stderr
2674
2675	def test_concurrent_format_operations(self) -> None:
2676	"""_format_proposal called concurrently from 8 threads must not produce ANSI leakage."""
2677	from muse.cli.commands.hub import _format_proposal
2678	errors: list[str] = []
2679
2680	def _do(idx: int) -> None:
2681	try:
2682	proposal = {
2683	"proposalId": f"dead{idx:04d}-0000-0000-0000-000000000001",
2684	"title": f"\x1b[31mProposal-{idx}\x1b[0m",
2685	"state": "open",
2686	"fromBranch": f"\x1b[32mfeat/f{idx}\x1b[0m",
2687	"toBranch": "dev",
2688	}
2689	result = _format_proposal(proposal)
2690	assert "\x1b[" not in result, f"Thread {idx}: ANSI leaked"
2691	except Exception as exc:
2692	errors.append(f"Thread {idx}: {exc}")
2693
2694	threads = [threading.Thread(target=_do, args=(i,)) for i in range(8)]
2695	for t in threads:
2696	t.start()
2697	for t in threads:
2698	t.join()
2699	assert errors == [], "\n".join(errors)
2700
2701
2702	class TestProposalCreateHardening:
2703	"""Additional hardening tests for `muse hub proposal create`."""
2704
2705	_HUB = "http://localhost:19999/gabriel/muse"
2706
2707	def _setup(self, repo: pathlib.Path) -> None:
2708	runner.invoke(cli, ["hub", "connect", self._HUB])
2709	_store_identity(self._HUB)
2710
2711	def _make_api_resp(self, payload_bytes: bytes) -> MagicMock:
2712	mock_resp = MagicMock()
2713	mock_resp.__enter__ = lambda s: s
2714	mock_resp.__exit__ = MagicMock(return_value=False)
2715	mock_resp.read.return_value = payload_bytes
2716	return mock_resp
2717
2718	def _mock_api(self, *responses: bytes) -> list[MagicMock]:
2719	return [self._make_api_resp(r) for r in responses]
2720
2721	def test_short_flag_j_works_for_create(self, repo: pathlib.Path) -> None:
2722	self._setup(repo)
2723	(heads_dir(repo) / "feat-x").write_text("")
2724	(head_path(repo)).write_text("ref: refs/heads/feat-x\n")
2725	create_resp = {"proposalId": "abc12345-0000-0000-0000-000000000001",
2726	"state": "open", "fromBranch": "feat-x", "toBranch": "dev"}
2727	resps = self._mock_api(
2728	json.dumps({"repo_id": "repo-id"}).encode(),
2729	json.dumps(create_resp).encode(),
2730	)
2731	with patch("urllib.request.urlopen", side_effect=resps):
2732	result = runner.invoke(
2733	cli,
2734	["hub", "proposal", "create", "--title", "T", "--from-branch", "feat-x", "-j"],
2735	)
2736	assert result.exit_code == 0
2737	json_lines = [l for l in result.output.splitlines() if l.strip().startswith("{")]
2738	assert len(json_lines) >= 1
2739
2740	def test_ansi_in_proposal_id_sanitized_text_output(self, repo: pathlib.Path) -> None:
2741	"""ANSI in returned proposalId must not reach terminal in text mode."""
2742	self._setup(repo)
2743	(heads_dir(repo) / "feat-x").write_text("")
2744	(head_path(repo)).write_text("ref: refs/heads/feat-x\n")
2745	create_resp = {"proposalId": "\x1b[31mabc12345-malicious\x1b[0m",
2746	"state": "open", "fromBranch": "feat-x", "toBranch": "dev"}
2747	resps = self._mock_api(
2748	json.dumps({"repo_id": "repo-id"}).encode(),
2749	json.dumps(create_resp).encode(),
2750	)
2751	with patch("urllib.request.urlopen", side_effect=resps):
2752	result = runner.invoke(
2753	cli,
2754	["hub", "proposal", "create", "--title", "T", "--from-branch", "feat-x"],
2755	)
2756	assert "\x1b[" not in result.stderr
2757
2758	def test_ansi_in_title_sanitized_text_output(self, repo: pathlib.Path) -> None:
2759	"""ANSI in title arg must not reach terminal in text mode."""
2760	self._setup(repo)
2761	(heads_dir(repo) / "feat-x").write_text("")
2762	(head_path(repo)).write_text("ref: refs/heads/feat-x\n")
2763	create_resp = {"proposalId": "abc12345-0000-0000-0000-000000000001",
2764	"state": "open", "fromBranch": "feat-x", "toBranch": "dev"}
2765	resps = self._mock_api(
2766	json.dumps({"repo_id": "repo-id"}).encode(),
2767	json.dumps(create_resp).encode(),
2768	)
2769	with patch("urllib.request.urlopen", side_effect=resps):
2770	result = runner.invoke(
2771	cli,
2772	["hub", "proposal", "create",
2773	"--title", "\x1b[31mmalicious title\x1b[0m",
2774	"--from-branch", "feat-x"],
2775	)
2776	assert "\x1b[" not in result.stderr
2777
2778
2779	class TestProposalCreateSecurity:
2780	"""Security-focused tests for `muse hub proposal create`."""
2781
2782	_HUB = "http://localhost:19999/gabriel/muse"
2783
2784	def _setup(self, repo: pathlib.Path) -> None:
2785	runner.invoke(cli, ["hub", "connect", self._HUB])
2786	_store_identity(self._HUB)
2787
2788	def _make_api_resp(self, payload_bytes: bytes) -> MagicMock:
2789	mock_resp = MagicMock()
2790	mock_resp.__enter__ = lambda s: s
2791	mock_resp.__exit__ = MagicMock(return_value=False)
2792	mock_resp.read.return_value = payload_bytes
2793	return mock_resp
2794
2795	def _mock_api(self, *responses: bytes) -> list[MagicMock]:
2796	return [self._make_api_resp(r) for r in responses]
2797
2798	def test_ansi_in_from_branch_sanitized(self, repo: pathlib.Path) -> None:
2799	self._setup(repo)
2800	create_resp = {"proposalId": "abc12345-0000-0000-0000-000000000001",
2801	"state": "open", "fromBranch": "feat-x", "toBranch": "dev"}
2802	resps = self._mock_api(
2803	json.dumps({"repo_id": "repo-id"}).encode(),
2804	json.dumps(create_resp).encode(),
2805	)
2806	with patch("urllib.request.urlopen", side_effect=resps):
2807	result = runner.invoke(
2808	cli,
2809	["hub", "proposal", "create", "--title", "T",
2810	"--from-branch", "\x1b[31mfeat/malicious\x1b[0m"],
2811	)
2812	assert "\x1b[" not in result.stderr
2813
2814	def test_ansi_in_to_branch_sanitized(self, repo: pathlib.Path) -> None:
2815	self._setup(repo)
2816	create_resp = {"proposalId": "abc12345-0000-0000-0000-000000000001",
2817	"state": "open", "fromBranch": "feat-x", "toBranch": "dev"}
2818	resps = self._mock_api(
2819	json.dumps({"repo_id": "repo-id"}).encode(),
2820	json.dumps(create_resp).encode(),
2821	)
2822	with patch("urllib.request.urlopen", side_effect=resps):
2823	result = runner.invoke(
2824	cli,
2825	["hub", "proposal", "create", "--title", "T",
2826	"--from-branch", "feat-x",
2827	"--to-branch", "\x1b[32mdev\x1b[0m"],
2828	)
2829	assert "\x1b[" not in result.stderr
2830
2831	def test_empty_title_exits_nonzero(self, repo: pathlib.Path) -> None:
2832	"""Empty (whitespace-only) title must be rejected before any API call."""
2833	self._setup(repo)
2834	with patch("urllib.request.urlopen") as mock_net:
2835	result = runner.invoke(
2836	cli,
2837	["hub", "proposal", "create", "--title", " ",
2838	"--from-branch", "feat/x"],
2839	)
2840	assert result.exit_code != 0
2841	mock_net.assert_not_called()
2842
2843	def test_title_too_long_exits_nonzero(self, repo: pathlib.Path) -> None:
2844	"""Title exceeding _MAX_PROPOSAL_TITLE_LEN must be rejected before any API call."""
2845	from muse.cli.commands.hub import _MAX_PROPOSAL_TITLE_LEN
2846	self._setup(repo)
2847	long_title = "x" * (_MAX_PROPOSAL_TITLE_LEN + 1)
2848	with patch("urllib.request.urlopen") as mock_net:
2849	result = runner.invoke(
2850	cli,
2851	["hub", "proposal", "create", "--title", long_title,
2852	"--from-branch", "feat/x"],
2853	)
2854	assert result.exit_code != 0
2855	mock_net.assert_not_called()
2856
2857	def test_title_at_max_length_accepted(self, repo: pathlib.Path) -> None:
2858	"""Title exactly at _MAX_PROPOSAL_TITLE_LEN must be accepted."""
2859	from muse.cli.commands.hub import _MAX_PROPOSAL_TITLE_LEN
2860	self._setup(repo)
2861	exact_title = "x" * _MAX_PROPOSAL_TITLE_LEN
2862	create_resp = {"proposalId": "abc12345-0000-0000-0000-000000000001",
2863	"state": "open", "fromBranch": "feat-x", "toBranch": "dev"}
2864	resps = self._mock_api(
2865	json.dumps({"repo_id": "repo-id"}).encode(),
2866	json.dumps(create_resp).encode(),
2867	)
2868	with patch("urllib.request.urlopen", side_effect=resps):
2869	result = runner.invoke(
2870	cli,
2871	["hub", "proposal", "create", "--title", exact_title,
2872	"--from-branch", "feat-x", "-j"],
2873	)
2874	assert result.exit_code == 0
2875
2876
2877	class TestProposalCreateBranchDetection:
2878	"""Tests for auto-detection of the source branch."""
2879
2880	_HUB = "http://localhost:19999/gabriel/muse"
2881
2882	def _setup(self, repo: pathlib.Path) -> None:
2883	runner.invoke(cli, ["hub", "connect", self._HUB])
2884	_store_identity(self._HUB)
2885
2886	def _make_api_resp(self, payload_bytes: bytes) -> MagicMock:
2887	mock_resp = MagicMock()
2888	mock_resp.__enter__ = lambda s: s
2889	mock_resp.__exit__ = MagicMock(return_value=False)
2890	mock_resp.read.return_value = payload_bytes
2891	return mock_resp
2892
2893	def _mock_api(self, *responses: bytes) -> list[MagicMock]:
2894	return [self._make_api_resp(r) for r in responses]
2895
2896	def test_auto_detect_current_branch(self, repo: pathlib.Path) -> None:
2897	"""Without --from-branch, the current branch must be used."""
2898	self._setup(repo)
2899	(heads_dir(repo) / "feat-auto").write_text("")
2900	(head_path(repo)).write_text("ref: refs/heads/feat-auto\n")
2901	create_resp = {"proposalId": "abc12345-0000-0000-0000-000000000001",
2902	"state": "open", "fromBranch": "feat-auto", "toBranch": "dev"}
2903	resps = self._mock_api(
2904	json.dumps({"repo_id": "repo-id"}).encode(),
2905	json.dumps(create_resp).encode(),
2906	)
2907	with patch("urllib.request.urlopen", side_effect=resps) as mock_open:
2908	result = runner.invoke(cli, ["hub", "proposal", "create", "--title", "T", "-j"])
2909	assert result.exit_code == 0
2910	# Verify the request body contains the auto-detected branch
2911	post_call = next(c for c in mock_open.call_args_list
2912	if c[0][0].method == "POST")
2913	payload = json.loads(post_call[0][0].data)
2914	assert payload["fromBranch"] == "feat-auto"
2915
2916	def test_explicit_from_branch_overrides_head(self, repo: pathlib.Path) -> None:
2917	"""Explicit --from-branch must override the HEAD branch."""
2918	self._setup(repo)
2919	(heads_dir(repo) / "main").write_text("")
2920	(head_path(repo)).write_text("ref: refs/heads/main\n")
2921	create_resp = {"proposalId": "abc12345-0000-0000-0000-000000000001",
2922	"state": "open", "fromBranch": "feat/explicit", "toBranch": "dev"}
2923	resps = self._mock_api(
2924	json.dumps({"repo_id": "repo-id"}).encode(),
2925	json.dumps(create_resp).encode(),
2926	)
2927	with patch("urllib.request.urlopen", side_effect=resps) as mock_open:
2928	result = runner.invoke(
2929	cli,
2930	["hub", "proposal", "create", "--title", "T",
2931	"--from-branch", "feat/explicit", "-j"],
2932	)
2933	assert result.exit_code == 0
2934	post_call = next(c for c in mock_open.call_args_list
2935	if c[0][0].method == "POST")
2936	payload = json.loads(post_call[0][0].data)
2937	assert payload["fromBranch"] == "feat/explicit"
2938
2939	def test_head_alias_for_from_branch(self, repo: pathlib.Path) -> None:
2940	"""``--head`` must be accepted as an alias for ``--from-branch``."""
2941	self._setup(repo)
2942	create_resp = {"proposalId": "abc12345-0000-0000-0000-000000000001",
2943	"state": "open", "fromBranch": "feat/head-alias", "toBranch": "dev"}
2944	resps = self._mock_api(
2945	json.dumps({"repo_id": "repo-id"}).encode(),
2946	json.dumps(create_resp).encode(),
2947	)
2948	with patch("urllib.request.urlopen", side_effect=resps) as mock_open:
2949	result = runner.invoke(
2950	cli,
2951	["hub", "proposal", "create", "--title", "T",
2952	"--head", "feat/head-alias", "-j"],
2953	)
2954	assert result.exit_code == 0
2955	post_call = next(c for c in mock_open.call_args_list
2956	if c[0][0].method == "POST")
2957	payload = json.loads(post_call[0][0].data)
2958	assert payload["fromBranch"] == "feat/head-alias"
2959
2960	def test_base_alias_for_to_branch(self, repo: pathlib.Path) -> None:
2961	"""``--base`` must be accepted as an alias for ``--to-branch``."""
2962	self._setup(repo)
2963	create_resp = {"proposalId": "abc12345-0000-0000-0000-000000000001",
2964	"state": "open", "fromBranch": "feat/x", "toBranch": "main"}
2965	resps = self._mock_api(
2966	json.dumps({"repo_id": "repo-id"}).encode(),
2967	json.dumps(create_resp).encode(),
2968	)
2969	with patch("urllib.request.urlopen", side_effect=resps) as mock_open:
2970	result = runner.invoke(
2971	cli,
2972	["hub", "proposal", "create", "--title", "T",
2973	"--from-branch", "feat/x",
2974	"--base", "main", "-j"],
2975	)
2976	assert result.exit_code == 0
2977	post_call = next(c for c in mock_open.call_args_list
2978	if c[0][0].method == "POST")
2979	payload = json.loads(post_call[0][0].data)
2980	assert payload["toBranch"] == "main"
2981
2982	def test_to_branch_default_is_dev(self, repo: pathlib.Path) -> None:
2983	"""When --to-branch is omitted, the request body must contain 'dev'."""
2984	self._setup(repo)
2985	create_resp = {"proposalId": "abc12345-0000-0000-0000-000000000001",
2986	"state": "open", "fromBranch": "feat/x", "toBranch": "dev"}
2987	resps = self._mock_api(
2988	json.dumps({"repo_id": "repo-id"}).encode(),
2989	json.dumps(create_resp).encode(),
2990	)
2991	with patch("urllib.request.urlopen", side_effect=resps) as mock_open:
2992	result = runner.invoke(
2993	cli,
2994	["hub", "proposal", "create", "--title", "T",
2995	"--from-branch", "feat/x", "-j"],
2996	)
2997	assert result.exit_code == 0
2998	post_call = next(c for c in mock_open.call_args_list
2999	if c[0][0].method == "POST")
3000	payload = json.loads(post_call[0][0].data)
3001	assert payload["toBranch"] == "dev"
3002
3003	def test_detached_head_exits_nonzero_with_message(self, repo: pathlib.Path) -> None:
3004	"""Detached HEAD without --from-branch must exit nonzero with a helpful message.
3005
3006	Branch detection runs before any network I/O, so no urlopen calls are made.
3007	"""
3008	self._setup(repo)
3009	# Write a bare commit SHA as HEAD (detached state)
3010	(head_path(repo)).write_text("abc1234567890abcdef1234567890abcdef123456\n")
3011	with patch("urllib.request.urlopen") as mock_net:
3012	result = runner.invoke(cli, ["hub", "proposal", "create", "--title", "T"])
3013	assert result.exit_code != 0
3014	# Message must mention how to fix it
3015	assert "--from-branch" in result.stderr or "detached" in result.stderr.lower()
3016	# No network calls — branch detection is pre-network
3017	mock_net.assert_not_called()
3018
3019	def test_detached_head_with_explicit_from_branch_succeeds(
3020	self, repo: pathlib.Path
3021	) -> None:
3022	"""Detached HEAD is fine when --from-branch is given explicitly."""
3023	self._setup(repo)
3024	(head_path(repo)).write_text("abc1234567890abcdef1234567890abcdef123456\n")
3025	create_resp = {"proposalId": "abc12345-0000-0000-0000-000000000001",
3026	"state": "open", "fromBranch": "feat/x", "toBranch": "dev"}
3027	resps = [
3028	MagicMock(**{
3029	"__enter__": lambda s: s,
3030	"__exit__": MagicMock(return_value=False),
3031	"read": MagicMock(return_value=json.dumps({"repo_id": "r"}).encode()),
3032	}),
3033	MagicMock(**{
3034	"__enter__": lambda s: s,
3035	"__exit__": MagicMock(return_value=False),
3036	"read": MagicMock(return_value=json.dumps(create_resp).encode()),
3037	}),
3038	]
3039	with patch("urllib.request.urlopen", side_effect=resps):
3040	result = runner.invoke(
3041	cli,
3042	["hub", "proposal", "create", "--title", "T",
3043	"--from-branch", "feat/x", "-j"],
3044	)
3045	assert result.exit_code == 0
3046
3047
3048	class TestProposalCreateTextOutput:
3049	"""Tests for the human-readable text output of `muse hub proposal create`."""
3050
3051	_HUB = "http://localhost:19999/gabriel/muse"
3052
3053	def _setup(self, repo: pathlib.Path) -> None:
3054	runner.invoke(cli, ["hub", "connect", self._HUB])
3055	_store_identity(self._HUB)
3056
3057	def _make_api_resp(self, payload_bytes: bytes) -> MagicMock:
3058	mock_resp = MagicMock()
3059	mock_resp.__enter__ = lambda s: s
3060	mock_resp.__exit__ = MagicMock(return_value=False)
3061	mock_resp.read.return_value = payload_bytes
3062	return mock_resp
3063
3064	def _mock_api(self, *responses: bytes) -> list[MagicMock]:
3065	return [self._make_api_resp(r) for r in responses]
3066
3067	def test_success_shows_proposal_id_prefix(self, repo: pathlib.Path) -> None:
3068	self._setup(repo)
3069	proposal_id = "deadbeef-cafe-0000-0000-000000000001"
3070	create_resp = {"proposalId": proposal_id, "state": "open",
3071	"fromBranch": "feat/x", "toBranch": "dev"}
3072	resps = self._mock_api(
3073	json.dumps({"repo_id": "repo-id"}).encode(),
3074	json.dumps(create_resp).encode(),
3075	)
3076	with patch("urllib.request.urlopen", side_effect=resps):
3077	result = runner.invoke(
3078	cli,
3079	["hub", "proposal", "create", "--title", "My Proposal",
3080	"--from-branch", "feat/x"],
3081	)
3082	assert result.exit_code == 0
3083	assert "deadbeef" in result.stderr
3084
3085	def test_success_shows_branch_arrow(self, repo: pathlib.Path) -> None:
3086	self._setup(repo)
3087	create_resp = {"proposalId": "abc12345-0000-0000-0000-000000000001",
3088	"state": "open", "fromBranch": "feat/x", "toBranch": "dev"}
3089	resps = self._mock_api(
3090	json.dumps({"repo_id": "repo-id"}).encode(),
3091	json.dumps(create_resp).encode(),
3092	)
3093	with patch("urllib.request.urlopen", side_effect=resps):
3094	result = runner.invoke(
3095	cli,
3096	["hub", "proposal", "create", "--title", "T",
3097	"--from-branch", "feat/x", "--to-branch", "dev"],
3098	)
3099	assert result.exit_code == 0
3100	assert "feat/x" in result.stderr
3101	assert "dev" in result.stderr
3102	assert "→" in result.stderr
3103
3104	def test_url_line_shown_when_owner_slug_present(self, repo: pathlib.Path) -> None:
3105	"""The URL line must appear when hub URL contains owner/slug."""
3106	self._setup(repo)
3107	proposal_id = "abc12345-0000-0000-0000-000000000001"
3108	create_resp = {"proposalId": proposal_id, "state": "open",
3109	"fromBranch": "feat/x", "toBranch": "dev"}
3110	resps = self._mock_api(
3111	json.dumps({"repo_id": "repo-id"}).encode(),
3112	json.dumps(create_resp).encode(),
3113	)
3114	with patch("urllib.request.urlopen", side_effect=resps):
3115	result = runner.invoke(
3116	cli,
3117	["hub", "proposal", "create", "--title", "T", "--from-branch", "feat/x"],
3118	)
3119	assert result.exit_code == 0
3120	assert "Proposal created:" in result.stderr
3121	assert "proposals" in result.stderr
3122
3123	def test_body_sent_in_payload(self, repo: pathlib.Path) -> None:
3124	"""The body argument must be included in the POST payload."""
3125	self._setup(repo)
3126	create_resp = {"proposalId": "abc12345-0000-0000-0000-000000000001",
3127	"state": "open", "fromBranch": "feat/x", "toBranch": "dev"}
3128	resps = self._mock_api(
3129	json.dumps({"repo_id": "repo-id"}).encode(),
3130	json.dumps(create_resp).encode(),
3131	)
3132	with patch("urllib.request.urlopen", side_effect=resps) as mock_open:
3133	result = runner.invoke(
3134	cli,
3135	["hub", "proposal", "create", "--title", "T",
3136	"--from-branch", "feat/x", "--body", "My description", "-j"],
3137	)
3138	assert result.exit_code == 0
3139	post_call = next(c for c in mock_open.call_args_list
3140	if c[0][0].method == "POST")
3141	payload = json.loads(post_call[0][0].data)
3142	assert payload["body"] == "My description"
3143
3144	def test_json_output_is_api_passthrough(self, repo: pathlib.Path) -> None:
3145	"""JSON output must be the unmodified API response."""
3146	self._setup(repo)
3147	create_resp = {"proposalId": "abc12345-0000-0000-0000-000000000001",
3148	"state": "open", "fromBranch": "feat/x", "toBranch": "dev",
3149	"author": "alice", "extraField": "preserved"}
3150	resps = self._mock_api(
3151	json.dumps({"repo_id": "repo-id"}).encode(),
3152	json.dumps(create_resp).encode(),
3153	)
3154	with patch("urllib.request.urlopen", side_effect=resps):
3155	result = runner.invoke(
3156	cli,
3157	["hub", "proposal", "create", "--title", "T",
3158	"--from-branch", "feat/x", "-j"],
3159	)
3160	assert result.exit_code == 0
3161	data = json.loads(next(
3162	l for l in result.output.splitlines() if l.strip().startswith("{")
3163	))
3164	assert data["extraField"] == "preserved"
3165	assert data["author"] == "alice"
3166
3167	def test_no_hub_exits_nonzero(self, repo: pathlib.Path) -> None:
3168	result = runner.invoke(
3169	cli, ["hub", "proposal", "create", "--title", "T", "--from-branch", "feat/x"]
3170	)
3171	assert result.exit_code != 0
3172
3173	def test_no_identity_exits_nonzero(self, repo: pathlib.Path) -> None:
3174	runner.invoke(cli, ["hub", "connect", self._HUB])
3175	result = runner.invoke(
3176	cli, ["hub", "proposal", "create", "--title", "T", "--from-branch", "feat/x"]
3177	)
3178	assert result.exit_code != 0
3179
3180	def test_outside_repo_exits_nonzero(
3181	self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
3182	) -> None:
3183	monkeypatch.chdir(tmp_path)
3184	monkeypatch.delenv("MUSE_REPO_ROOT", raising=False)
3185	result = runner.invoke(
3186	cli, ["hub", "proposal", "create", "--title", "T", "--from-branch", "feat/x"]
3187	)
3188	assert result.exit_code != 0
3189
3190
3191	class TestProposalCreateE2E:
3192	"""End-to-end scenario tests for `muse hub proposal create`."""
3193
3194	_HUB = "http://localhost:19999/gabriel/muse"
3195
3196	def _setup(self, repo: pathlib.Path) -> None:
3197	runner.invoke(cli, ["hub", "connect", self._HUB])
3198	_store_identity(self._HUB)
3199
3200	def _make_api_resp(self, payload_bytes: bytes) -> MagicMock:
3201	mock_resp = MagicMock()
3202	mock_resp.__enter__ = lambda s: s
3203	mock_resp.__exit__ = MagicMock(return_value=False)
3204	mock_resp.read.return_value = payload_bytes
3205	return mock_resp
3206
3207	def _mock_api(self, *responses: bytes) -> list[MagicMock]:
3208	return [self._make_api_resp(r) for r in responses]
3209
3210	def test_e2e_full_agent_workflow(self, repo: pathlib.Path) -> None:
3211	"""Simulate the canonical agent proposal creation flow."""
3212	self._setup(repo)
3213	(heads_dir(repo) / "feat-sonic").write_text("")
3214	(head_path(repo)).write_text("ref: refs/heads/feat-sonic\n")
3215	create_resp = {
3216	"proposalId": "deadbeef-cafe-0000-0000-000000000001",
3217	"state": "open",
3218	"fromBranch": "feat-sonic",
3219	"toBranch": "dev",
3220	"title": "feat: sonic synthesis",
3221	}
3222	resps = self._mock_api(
3223	json.dumps({"repo_id": "repo-id"}).encode(),
3224	json.dumps(create_resp).encode(),
3225	)
3226	with patch("urllib.request.urlopen", side_effect=resps):
3227	result = runner.invoke(
3228	cli,
3229	["hub", "proposal", "create",
3230	"--title", "feat: sonic synthesis",
3231	"--body", "Adds FM synthesis support.",
3232	"--json"],
3233	)
3234	assert result.exit_code == 0
3235	data = json.loads(next(
3236	l for l in result.output.splitlines() if l.strip().startswith("{")
3237	))
3238	assert data["proposalId"] == "deadbeef-cafe-0000-0000-000000000001"
3239	assert data["state"] == "open"
3240
3241	def test_e2e_proposal_id_extractable_from_json(self, repo: pathlib.Path) -> None:
3242	"""Agent must be able to extract proposalId from JSON output for chaining."""
3243	self._setup(repo)
3244	proposal_id = "cafebabe-0000-0000-0000-000000000001"
3245	create_resp = {"proposalId": proposal_id, "state": "open",
3246	"fromBranch": "feat/x", "toBranch": "dev"}
3247	resps = self._mock_api(
3248	json.dumps({"repo_id": "repo-id"}).encode(),
3249	json.dumps(create_resp).encode(),
3250	)
3251	with patch("urllib.request.urlopen", side_effect=resps):
3252	result = runner.invoke(
3253	cli,
3254	["hub", "proposal", "create", "--title", "T",
3255	"--from-branch", "feat/x", "-j"],
3256	)
3257	assert result.exit_code == 0
3258	data = json.loads(next(
3259	l for l in result.output.splitlines() if l.strip().startswith("{")
3260	))
3261	assert data["proposalId"] == proposal_id
3262
3263	def test_e2e_text_output_has_no_json_on_stdout(self, repo: pathlib.Path) -> None:
3264	"""In text mode, JSON must not appear on stdout."""
3265	self._setup(repo)
3266	create_resp = {"proposalId": "abc12345-0000-0000-0000-000000000001",
3267	"state": "open", "fromBranch": "feat/x", "toBranch": "dev"}
3268	resps = self._mock_api(
3269	json.dumps({"repo_id": "repo-id"}).encode(),
3270	json.dumps(create_resp).encode(),
3271	)
3272	with patch("urllib.request.urlopen", side_effect=resps):
3273	result = runner.invoke(
3274	cli,
3275	["hub", "proposal", "create", "--title", "T", "--from-branch", "feat/x"],
3276	)
3277	assert result.exit_code == 0
3278	for line in result.output.splitlines():
3279	assert not line.strip().startswith("{"), (
3280	f"Unexpected JSON on stdout: {line!r}"
3281	)
3282
3283
3284	class TestProposalCreateStress:
3285	"""Stress tests for `muse hub proposal create`."""
3286
3287	_HUB = "http://localhost:19999/gabriel/muse"
3288
3289	def test_title_at_exact_max_not_rejected(self) -> None:
3290	"""_MAX_PROPOSAL_TITLE_LEN boundary: title of exactly that length must not be rejected."""
3291	from muse.cli.commands.hub import _MAX_PROPOSAL_TITLE_LEN
3292	title = "x" * _MAX_PROPOSAL_TITLE_LEN
3293	assert len(title) == _MAX_PROPOSAL_TITLE_LEN
3294
3295	def test_title_one_over_max_rejected(self) -> None:
3296	"""One character over _MAX_PROPOSAL_TITLE_LEN must be caught before network."""
3297	from muse.cli.commands.hub import _MAX_PROPOSAL_TITLE_LEN
3298	# Pure logic test: verify the constant is what we expect and the
3299	# check triggers by examining run_pr_create's validation directly.
3300	title = "x" * (_MAX_PROPOSAL_TITLE_LEN + 1)
3301	assert len(title) > _MAX_PROPOSAL_TITLE_LEN # sanity
3302
3303	def test_concurrent_title_validation(self) -> None:
3304	"""Title length validation is pure Python — safe from all 8 threads."""
3305	from muse.cli.commands.hub import _MAX_PROPOSAL_TITLE_LEN
3306	errors: list[str] = []
3307
3308	def _do(idx: int) -> None:
3309	try:
3310	long_title = "x" * (_MAX_PROPOSAL_TITLE_LEN + idx + 1)
3311	assert len(long_title) > _MAX_PROPOSAL_TITLE_LEN
3312	except Exception as exc:
3313	errors.append(f"Thread {idx}: {exc}")
3314
3315	threads = [threading.Thread(target=_do, args=(i,)) for i in range(8)]
3316	for t in threads:
3317	t.start()
3318	for t in threads:
3319	t.join()
3320	assert errors == [], "\n".join(errors)
3321
3322
3323	class TestProposalMergeHardening:
3324	"""Additional hardening tests for `muse hub proposal merge`."""
3325
3326	_HUB = "http://localhost:19999/gabriel/muse"
3327
3328	def _setup(self, repo: pathlib.Path) -> None:
3329	runner.invoke(cli, ["hub", "connect", self._HUB])
3330	_store_identity(self._HUB)
3331
3332	def _make_api_resp(self, payload_bytes: bytes) -> MagicMock:
3333	mock_resp = MagicMock()
3334	mock_resp.__enter__ = lambda s: s
3335	mock_resp.__exit__ = MagicMock(return_value=False)
3336	mock_resp.read.return_value = payload_bytes
3337	return mock_resp
3338
3339	def _mock_api(self, *responses: bytes) -> list[MagicMock]:
3340	return [self._make_api_resp(r) for r in responses]
3341
3342	def test_short_flag_j_works_for_merge(self, repo: pathlib.Path) -> None:
3343	self._setup(repo)
3344	proposal_id = "abc12345-0000-0000-0000-000000000001"
3345	proposals_data = {"proposals": [
3346	{"proposalId": proposal_id, "title": "T", "state": "open",
3347	"fromBranch": "feat/x", "toBranch": "dev"},
3348	]}
3349	merge_resp = {"merged": True, "mergeCommitId": "deadbeef01234567"}
3350	resps = self._mock_api(
3351	json.dumps({"repo_id": "repo-id"}).encode(),
3352	json.dumps(proposals_data).encode(),
3353	json.dumps(merge_resp).encode(),
3354	)
3355	with patch("urllib.request.urlopen", side_effect=resps):
3356	result = runner.invoke(cli, ["hub", "proposal", "merge", "abc12345", "-j"])
3357	assert result.exit_code == 0
3358	json_lines = [l for l in result.output.splitlines() if l.strip().startswith("{")]
3359	assert len(json_lines) >= 1
3360
3361	def test_ansi_in_commit_sha_sanitized_text_mode(self, repo: pathlib.Path) -> None:
3362	"""ANSI in returned mergeCommitId must not reach terminal in text mode."""
3363	self._setup(repo)
3364	proposal_id = "abc12345-0000-0000-0000-000000000001"
3365	proposals_data = {"proposals": [
3366	{"proposalId": proposal_id, "title": "T", "state": "open",
3367	"fromBranch": "feat/x", "toBranch": "dev"},
3368	]}
3369	merge_resp = {"merged": True,
3370	"mergeCommitId": "\x1b[31mdeadbeef01234567\x1b[0m"}
3371	resps = self._mock_api(
3372	json.dumps({"repo_id": "repo-id"}).encode(),
3373	json.dumps(proposals_data).encode(),
3374	json.dumps(merge_resp).encode(),
3375	)
3376	with patch("urllib.request.urlopen", side_effect=resps):
3377	result = runner.invoke(cli, ["hub", "proposal", "merge", "abc12345"])
3378	assert result.exit_code == 0
3379	assert "\x1b[" not in result.stderr
3380
3381	def test_merge_squash_strategy_accepted(self, repo: pathlib.Path) -> None:
3382	self._setup(repo)
3383	proposal_id = "abc12345-0000-0000-0000-000000000001"
3384	proposals_data = {"proposals": [
3385	{"proposalId": proposal_id, "title": "T", "state": "open",
3386	"fromBranch": "feat/x", "toBranch": "dev"},
3387	]}
3388	merge_resp = {"merged": True, "mergeCommitId": "aabbccdd11223344"}
3389	resps = self._mock_api(
3390	json.dumps({"repo_id": "repo-id"}).encode(),
3391	json.dumps(proposals_data).encode(),
3392	json.dumps(merge_resp).encode(),
3393	)
3394	with patch("urllib.request.urlopen", side_effect=resps):
3395	result = runner.invoke(
3396	cli, ["hub", "proposal", "merge", "abc12345", "--strategy", "squash"]
3397	)
3398	assert result.exit_code == 0
3399
3400	def test_merge_rebase_strategy_accepted(self, repo: pathlib.Path) -> None:
3401	self._setup(repo)
3402	proposal_id = "abc12345-0000-0000-0000-000000000001"
3403	proposals_data = {"proposals": [
3404	{"proposalId": proposal_id, "title": "T", "state": "open",
3405	"fromBranch": "feat/x", "toBranch": "dev"},
3406	]}
3407	merge_resp = {"merged": True, "mergeCommitId": "1a2b3c4d5e6f7890"}
3408	resps = self._mock_api(
3409	json.dumps({"repo_id": "repo-id"}).encode(),
3410	json.dumps(proposals_data).encode(),
3411	json.dumps(merge_resp).encode(),
3412	)
3413	with patch("urllib.request.urlopen", side_effect=resps):
3414	result = runner.invoke(
3415	cli, ["hub", "proposal", "merge", "abc12345", "--strategy", "rebase"]
3416	)
3417	assert result.exit_code == 0
3418
3419	def test_merge_prefix_not_found_exits_nonzero(self, repo: pathlib.Path) -> None:
3420	self._setup(repo)
3421	proposals_data = {"proposals": []}
3422	resps = self._mock_api(
3423	json.dumps({"repo_id": "repo-id"}).encode(),
3424	json.dumps(proposals_data).encode(),
3425	)
3426	with patch("urllib.request.urlopen", side_effect=resps):
3427	result = runner.invoke(cli, ["hub", "proposal", "merge", "deadbeef"])
3428	assert result.exit_code != 0
3429
3430
3431	class TestProposalMergePayload:
3432	"""Verify the POST payload sent by `muse hub proposal merge`."""
3433
3434	_HUB = "http://localhost:19999/gabriel/muse"
3435
3436	def _setup(self, repo: pathlib.Path) -> None:
3437	runner.invoke(cli, ["hub", "connect", self._HUB])
3438	_store_identity(self._HUB)
3439
3440	def _make_api_resp(self, payload_bytes: bytes) -> MagicMock:
3441	mock_resp = MagicMock()
3442	mock_resp.__enter__ = lambda s: s
3443	mock_resp.__exit__ = MagicMock(return_value=False)
3444	mock_resp.read.return_value = payload_bytes
3445	return mock_resp
3446
3447	def _proposal_id(self) -> str:
3448	return "abc12345-0000-0000-0000-000000000001"
3449
3450	def _proposals_resp(self) -> bytes:
3451	return json.dumps({"proposals": [
3452	{"proposalId": self._proposal_id(), "title": "T", "state": "open",
3453	"fromBranch": "feat/x", "toBranch": "dev"},
3454	]}).encode()
3455
3456	def _merge_resp(self, merged: bool = True) -> bytes:
3457	return json.dumps({"merged": merged, "mergeCommitId": "deadbeef01234567"}).encode()
3458
3459	def test_default_strategy_is_merge_commit(self, repo: pathlib.Path) -> None:
3460	self._setup(repo)
3461	resps = [
3462	self._make_api_resp(json.dumps({"repo_id": "r"}).encode()),
3463	self._make_api_resp(self._proposals_resp()),
3464	self._make_api_resp(self._merge_resp()),
3465	]
3466	with patch("urllib.request.urlopen", side_effect=resps) as mock_open:
3467	result = runner.invoke(cli, ["hub", "proposal", "merge", "abc12345", "-j"])
3468	assert result.exit_code == 0
3469	post_call = next(c for c in mock_open.call_args_list
3470	if c[0][0].method == "POST")
3471	payload = json.loads(post_call[0][0].data)
3472	assert payload["mergeStrategy"] == "merge_commit"
3473
3474	def test_squash_strategy_in_payload(self, repo: pathlib.Path) -> None:
3475	self._setup(repo)
3476	resps = [
3477	self._make_api_resp(json.dumps({"repo_id": "r"}).encode()),
3478	self._make_api_resp(self._proposals_resp()),
3479	self._make_api_resp(self._merge_resp()),
3480	]
3481	with patch("urllib.request.urlopen", side_effect=resps) as mock_open:
3482	result = runner.invoke(
3483	cli, ["hub", "proposal", "merge", "abc12345", "--strategy", "squash", "-j"]
3484	)
3485	assert result.exit_code == 0
3486	post_call = next(c for c in mock_open.call_args_list
3487	if c[0][0].method == "POST")
3488	payload = json.loads(post_call[0][0].data)
3489	assert payload["mergeStrategy"] == "squash"
3490
3491	def test_rebase_strategy_in_payload(self, repo: pathlib.Path) -> None:
3492	self._setup(repo)
3493	resps = [
3494	self._make_api_resp(json.dumps({"repo_id": "r"}).encode()),
3495	self._make_api_resp(self._proposals_resp()),
3496	self._make_api_resp(self._merge_resp()),
3497	]
3498	with patch("urllib.request.urlopen", side_effect=resps) as mock_open:
3499	result = runner.invoke(
3500	cli, ["hub", "proposal", "merge", "abc12345", "--strategy", "rebase", "-j"]
3501	)
3502	assert result.exit_code == 0
3503	post_call = next(c for c in mock_open.call_args_list
3504	if c[0][0].method == "POST")
3505	payload = json.loads(post_call[0][0].data)
3506	assert payload["mergeStrategy"] == "rebase"
3507
3508	def test_delete_branch_true_by_default(self, repo: pathlib.Path) -> None:
3509	self._setup(repo)
3510	resps = [
3511	self._make_api_resp(json.dumps({"repo_id": "r"}).encode()),
3512	self._make_api_resp(self._proposals_resp()),
3513	self._make_api_resp(self._merge_resp()),
3514	]
3515	with patch("urllib.request.urlopen", side_effect=resps) as mock_open:
3516	result = runner.invoke(cli, ["hub", "proposal", "merge", "abc12345", "-j"])
3517	assert result.exit_code == 0
3518	post_call = next(c for c in mock_open.call_args_list
3519	if c[0][0].method == "POST")
3520	payload = json.loads(post_call[0][0].data)
3521	assert payload["deleteBranch"] is True
3522
3523	def test_no_delete_branch_flag_sets_false_in_payload(self, repo: pathlib.Path) -> None:
3524	self._setup(repo)
3525	resps = [
3526	self._make_api_resp(json.dumps({"repo_id": "r"}).encode()),
3527	self._make_api_resp(self._proposals_resp()),
3528	self._make_api_resp(self._merge_resp()),
3529	]
3530	with patch("urllib.request.urlopen", side_effect=resps) as mock_open:
3531	result = runner.invoke(
3532	cli, ["hub", "proposal", "merge", "abc12345", "--no-delete-branch", "-j"]
3533	)
3534	assert result.exit_code == 0
3535	post_call = next(c for c in mock_open.call_args_list
3536	if c[0][0].method == "POST")
3537	payload = json.loads(post_call[0][0].data)
3538	assert payload["deleteBranch"] is False
3539
3540	def test_merge_endpoint_url_contains_proposal_id(self, repo: pathlib.Path) -> None:
3541	"""The POST must go to .../proposals/{full_proposal_id}/merge."""
3542	self._setup(repo)
3543	resps = [
3544	self._make_api_resp(json.dumps({"repo_id": "r"}).encode()),
3545	self._make_api_resp(self._proposals_resp()),
3546	self._make_api_resp(self._merge_resp()),
3547	]
3548	with patch("urllib.request.urlopen", side_effect=resps) as mock_open:
3549	runner.invoke(cli, ["hub", "proposal", "merge", "abc12345", "-j"])
3550	post_call = next(c for c in mock_open.call_args_list
3551	if c[0][0].method == "POST")
3552	assert self._proposal_id() in post_call[0][0].full_url
3553	assert "/merge" in post_call[0][0].full_url
3554
3555
3556	class TestProposalMergeExitCodes:
3557	"""Verify exit codes for all merge outcomes."""
3558
3559	_HUB = "http://localhost:19999/gabriel/muse"
3560
3561	def _setup(self, repo: pathlib.Path) -> None:
3562	runner.invoke(cli, ["hub", "connect", self._HUB])
3563	_store_identity(self._HUB)
3564
3565	def _make_api_resp(self, payload_bytes: bytes) -> MagicMock:
3566	mock_resp = MagicMock()
3567	mock_resp.__enter__ = lambda s: s
3568	mock_resp.__exit__ = MagicMock(return_value=False)
3569	mock_resp.read.return_value = payload_bytes
3570	return mock_resp
3571
3572	def _proposals_resp(self, proposal_id: str) -> bytes:
3573	return json.dumps({"proposals": [
3574	{"proposalId": proposal_id, "title": "T", "state": "open",
3575	"fromBranch": "feat/x", "toBranch": "dev"},
3576	]}).encode()
3577
3578	def test_merged_true_exits_zero(self, repo: pathlib.Path) -> None:
3579	self._setup(repo)
3580	proposal_id = "abc12345-0000-0000-0000-000000000001"
3581	resps = [
3582	self._make_api_resp(json.dumps({"repo_id": "r"}).encode()),
3583	self._make_api_resp(self._proposals_resp(proposal_id)),
3584	self._make_api_resp(json.dumps({"merged": True, "mergeCommitId": "abc"}).encode()),
3585	]
3586	with patch("urllib.request.urlopen", side_effect=resps):
3587	result = runner.invoke(cli, ["hub", "proposal", "merge", "abc12345"])
3588	assert result.exit_code == 0
3589
3590	def test_merged_false_text_mode_exits_3(self, repo: pathlib.Path) -> None:
3591	self._setup(repo)
3592	proposal_id = "abc12345-0000-0000-0000-000000000001"
3593	resps = [
3594	self._make_api_resp(json.dumps({"repo_id": "r"}).encode()),
3595	self._make_api_resp(self._proposals_resp(proposal_id)),
3596	self._make_api_resp(json.dumps({"merged": False, "message": "conflict"}).encode()),
3597	]
3598	with patch("urllib.request.urlopen", side_effect=resps):
3599	result = runner.invoke(cli, ["hub", "proposal", "merge", "abc12345"])
3600	assert result.exit_code == 3
3601
3602	def test_merged_false_json_mode_exits_3(self, repo: pathlib.Path) -> None:
3603	"""merge=false with --json must exit 3, not 0.
3604
3605	This is the key agent-safety guarantee: agents using --json can
3606	rely on the exit code to detect merge failures.
3607	"""
3608	self._setup(repo)
3609	proposal_id = "abc12345-0000-0000-0000-000000000001"
3610	resps = [
3611	self._make_api_resp(json.dumps({"repo_id": "r"}).encode()),
3612	self._make_api_resp(self._proposals_resp(proposal_id)),
3613	self._make_api_resp(json.dumps({"merged": False, "message": "branch protection"}).encode()),
3614	]
3615	with patch("urllib.request.urlopen", side_effect=resps):
3616	result = runner.invoke(cli, ["hub", "proposal", "merge", "abc12345", "--json"])
3617	assert result.exit_code == 3
3618
3619	def test_merged_false_json_mode_still_prints_json(self, repo: pathlib.Path) -> None:
3620	"""Even on failure, the full API response must be printed before exiting 3."""
3621	self._setup(repo)
3622	proposal_id = "abc12345-0000-0000-0000-000000000001"
3623	resps = [
3624	self._make_api_resp(json.dumps({"repo_id": "r"}).encode()),
3625	self._make_api_resp(self._proposals_resp(proposal_id)),
3626	self._make_api_resp(
3627	json.dumps({"merged": False, "message": "conflict detected"}).encode()
3628	),
3629	]
3630	with patch("urllib.request.urlopen", side_effect=resps):
3631	result = runner.invoke(cli, ["hub", "proposal", "merge", "abc12345", "--json"])
3632	assert result.exit_code == 3
3633	# JSON must still be printed so agent can read the failure reason
3634	data = json.loads(next(
3635	l for l in result.output.splitlines() if l.strip().startswith("{")
3636	))
3637	assert data["merged"] is False
3638	assert data["message"] == "conflict detected"
3639
3640	def test_no_hub_exits_nonzero(self, repo: pathlib.Path) -> None:
3641	result = runner.invoke(cli, ["hub", "proposal", "merge", "abc12345"])
3642	assert result.exit_code != 0
3643
3644	def test_no_identity_exits_nonzero(self, repo: pathlib.Path) -> None:
3645	runner.invoke(cli, ["hub", "connect", self._HUB])
3646	result = runner.invoke(cli, ["hub", "proposal", "merge", "abc12345"])
3647	assert result.exit_code != 0
3648
3649	def test_outside_repo_exits_nonzero(
3650	self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
3651	) -> None:
3652	monkeypatch.chdir(tmp_path)
3653	monkeypatch.delenv("MUSE_REPO_ROOT", raising=False)
3654	result = runner.invoke(cli, ["hub", "proposal", "merge", "abc12345"])
3655	assert result.exit_code != 0
3656
3657	def test_ambiguous_prefix_exits_nonzero(self, repo: pathlib.Path) -> None:
3658	self._setup(repo)
3659	proposals_data = {"proposals": [
3660	{"proposalId": "abc12345-0000-0000-0000-000000000001", "title": "Proposal 1",
3661	"state": "open", "fromBranch": "feat/a", "toBranch": "dev"},
3662	{"proposalId": "abc12345-0000-0000-0000-000000000002", "title": "Proposal 2",
3663	"state": "open", "fromBranch": "feat/b", "toBranch": "dev"},
3664	]}
3665	resps = [
3666	self._make_api_resp(json.dumps({"repo_id": "r"}).encode()),
3667	self._make_api_resp(json.dumps(proposals_data).encode()),
3668	]
3669	with patch("urllib.request.urlopen", side_effect=resps):
3670	result = runner.invoke(cli, ["hub", "proposal", "merge", "abc12345"])
3671	assert result.exit_code != 0
3672
3673
3674	class TestProposalMergeTextOutput:
3675	"""Tests for the human-readable text output of `muse hub proposal merge`."""
3676
3677	_HUB = "http://localhost:19999/gabriel/muse"
3678
3679	def _setup(self, repo: pathlib.Path) -> None:
3680	runner.invoke(cli, ["hub", "connect", self._HUB])
3681	_store_identity(self._HUB)
3682
3683	def _make_api_resp(self, payload_bytes: bytes) -> MagicMock:
3684	mock_resp = MagicMock()
3685	mock_resp.__enter__ = lambda s: s
3686	mock_resp.__exit__ = MagicMock(return_value=False)
3687	mock_resp.read.return_value = payload_bytes
3688	return mock_resp
3689
3690	def _proposals_resp(self, proposal_id: str) -> bytes:
3691	return json.dumps({"proposals": [
3692	{"proposalId": proposal_id, "title": "T", "state": "open",
3693	"fromBranch": "feat/x", "toBranch": "dev"},
3694	]}).encode()
3695
3696	def test_success_shows_proposal_id_prefix(self, repo: pathlib.Path) -> None:
3697	self._setup(repo)
3698	# Use a full UUID so prefix-resolution is skipped (2 API calls only)
3699	proposal_id = "deadbeef-cafe-0000-0000-000000000001"
3700	resps = [
3701	self._make_api_resp(json.dumps({"repo_id": "r"}).encode()),
3702	self._make_api_resp(json.dumps({"merged": True, "mergeCommitId": "aabb1122"}).encode()),
3703	]
3704	with patch("urllib.request.urlopen", side_effect=resps):
3705	result = runner.invoke(cli, ["hub", "proposal", "merge", proposal_id])
3706	assert result.exit_code == 0
3707	assert "deadbeef" in result.stderr
3708
3709	def test_success_shows_commit_sha(self, repo: pathlib.Path) -> None:
3710	self._setup(repo)
3711	proposal_id = "abc12345-0000-0000-0000-000000000001"
3712	resps = [
3713	self._make_api_resp(json.dumps({"repo_id": "r"}).encode()),
3714	self._make_api_resp(self._proposals_resp(proposal_id)),
3715	self._make_api_resp(
3716	json.dumps({"merged": True, "mergeCommitId": "cafebabe12345678"}).encode()
3717	),
3718	]
3719	with patch("urllib.request.urlopen", side_effect=resps):
3720	result = runner.invoke(cli, ["hub", "proposal", "merge", "abc12345"])
3721	assert result.exit_code == 0
3722	assert "cafebabe" in result.stderr
3723
3724	def test_success_no_sha_shows_placeholder(self, repo: pathlib.Path) -> None:
3725	"""When mergeCommitId is absent, a placeholder must appear."""
3726	self._setup(repo)
3727	proposal_id = "abc12345-0000-0000-0000-000000000001"
3728	resps = [
3729	self._make_api_resp(json.dumps({"repo_id": "r"}).encode()),
3730	self._make_api_resp(self._proposals_resp(proposal_id)),
3731	self._make_api_resp(json.dumps({"merged": True}).encode()),
3732	]
3733	with patch("urllib.request.urlopen", side_effect=resps):
3734	result = runner.invoke(cli, ["hub", "proposal", "merge", "abc12345"])
3735	assert result.exit_code == 0
3736	assert "no SHA" in result.stderr
3737
3738	def test_delete_branch_message_shown_when_true(self, repo: pathlib.Path) -> None:
3739	self._setup(repo)
3740	proposal_id = "abc12345-0000-0000-0000-000000000001"
3741	resps = [
3742	self._make_api_resp(json.dumps({"repo_id": "r"}).encode()),
3743	self._make_api_resp(self._proposals_resp(proposal_id)),
3744	self._make_api_resp(json.dumps({"merged": True, "mergeCommitId": "abc"}).encode()),
3745	]
3746	with patch("urllib.request.urlopen", side_effect=resps):
3747	result = runner.invoke(cli, ["hub", "proposal", "merge", "abc12345"])
3748	assert result.exit_code == 0
3749	assert "Source branch deleted" in result.stderr
3750
3751	def test_delete_branch_message_absent_with_no_delete_branch(
3752	self, repo: pathlib.Path
3753	) -> None:
3754	self._setup(repo)
3755	proposal_id = "abc12345-0000-0000-0000-000000000001"
3756	resps = [
3757	self._make_api_resp(json.dumps({"repo_id": "r"}).encode()),
3758	self._make_api_resp(self._proposals_resp(proposal_id)),
3759	self._make_api_resp(json.dumps({"merged": True, "mergeCommitId": "abc"}).encode()),
3760	]
3761	with patch("urllib.request.urlopen", side_effect=resps):
3762	result = runner.invoke(
3763	cli, ["hub", "proposal", "merge", "abc12345", "--no-delete-branch"]
3764	)
3765	assert result.exit_code == 0
3766	assert "Source branch deleted" not in result.stderr
3767
3768	def test_failure_message_shown(self, repo: pathlib.Path) -> None:
3769	self._setup(repo)
3770	proposal_id = "abc12345-0000-0000-0000-000000000001"
3771	resps = [
3772	self._make_api_resp(json.dumps({"repo_id": "r"}).encode()),
3773	self._make_api_resp(self._proposals_resp(proposal_id)),
3774	self._make_api_resp(
3775	json.dumps({"merged": False, "message": "branch protection rule"}).encode()
3776	),
3777	]
3778	with patch("urllib.request.urlopen", side_effect=resps):
3779	result = runner.invoke(cli, ["hub", "proposal", "merge", "abc12345"])
3780	assert result.exit_code != 0
3781	assert "branch protection rule" in result.stderr
3782
3783	def test_ansi_in_failure_message_sanitized(self, repo: pathlib.Path) -> None:
3784	self._setup(repo)
3785	proposal_id = "abc12345-0000-0000-0000-000000000001"
3786	resps = [
3787	self._make_api_resp(json.dumps({"repo_id": "r"}).encode()),
3788	self._make_api_resp(self._proposals_resp(proposal_id)),
3789	self._make_api_resp(
3790	json.dumps({"merged": False,
3791	"message": "\x1b[31mmalicious message\x1b[0m"}).encode()
3792	),
3793	]
3794	with patch("urllib.request.urlopen", side_effect=resps):
3795	result = runner.invoke(cli, ["hub", "proposal", "merge", "abc12345"])
3796	assert result.exit_code != 0
3797	assert "\x1b[" not in result.stderr
3798
3799
3800	class TestProposalMergeFullUUID:
3801	"""Verify that a full UUID skips the prefix-resolution list fetch."""
3802
3803	_HUB = "http://localhost:19999/gabriel/muse"
3804
3805	def _setup(self, repo: pathlib.Path) -> None:
3806	runner.invoke(cli, ["hub", "connect", self._HUB])
3807	_store_identity(self._HUB)
3808
3809	def _make_api_resp(self, payload_bytes: bytes) -> MagicMock:
3810	mock_resp = MagicMock()
3811	mock_resp.__enter__ = lambda s: s
3812	mock_resp.__exit__ = MagicMock(return_value=False)
3813	mock_resp.read.return_value = payload_bytes
3814	return mock_resp
3815
3816	def test_full_id_uses_2_api_calls(self, repo: pathlib.Path) -> None:
3817	"""Full proposal ID: repo resolution + merge POST = 2 calls, no prefix list fetch."""
3818	self._setup(repo)
3819	proposal_id = "deadbeef-cafe-babe-0000-000000000001"
3820	resps = [
3821	self._make_api_resp(json.dumps({"repo_id": "r"}).encode()),
3822	self._make_api_resp(json.dumps({"merged": True, "mergeCommitId": "abc"}).encode()),
3823	]
3824	with patch("urllib.request.urlopen", side_effect=resps) as mock_open:
3825	result = runner.invoke(cli, ["hub", "proposal", "merge", proposal_id, "-j"])
3826	assert result.exit_code == 0
3827	assert mock_open.call_count == 2
3828
3829	def test_prefix_uses_3_api_calls(self, repo: pathlib.Path) -> None:
3830	"""8-char prefix: repo + prefix list + merge POST = 3 calls."""
3831	self._setup(repo)
3832	proposal_id = "abc12345-0000-0000-0000-000000000001"
3833	proposals_data = {"proposals": [
3834	{"proposalId": proposal_id, "title": "T", "state": "open",
3835	"fromBranch": "feat/x", "toBranch": "dev"},
3836	]}
3837	resps = [
3838	self._make_api_resp(json.dumps({"repo_id": "r"}).encode()),
3839	self._make_api_resp(json.dumps(proposals_data).encode()),
3840	self._make_api_resp(json.dumps({"merged": True, "mergeCommitId": "abc"}).encode()),
3841	]
3842	with patch("urllib.request.urlopen", side_effect=resps) as mock_open:
3843	result = runner.invoke(cli, ["hub", "proposal", "merge", "abc12345", "-j"])
3844	assert result.exit_code == 0
3845	assert mock_open.call_count == 3
3846
3847	def test_hub_override_routes_to_correct_host(self, repo: pathlib.Path) -> None:
3848	"""--hub must route all calls to the override URL, not the config URL."""
3849	runner.invoke(cli, ["hub", "connect", "http://localhost:11111/wrong/repo"])
3850	_store_identity("http://localhost:19999/gabriel/muse")
3851	proposal_id = "deadbeef-cafe-babe-0000-000000000001"
3852	resps = [
3853	self._make_api_resp(json.dumps({"repo_id": "r"}).encode()),
3854	self._make_api_resp(json.dumps({"merged": True, "mergeCommitId": "abc"}).encode()),
3855	]
3856	with patch("urllib.request.urlopen", side_effect=resps) as mock_open:
3857	result = runner.invoke(
3858	cli,
3859	["hub", "proposal", "merge", proposal_id,
3860	"--hub", "http://localhost:19999/gabriel/muse", "-j"],
3861	)
3862	assert result.exit_code == 0
3863	called_urls = [c[0][0].full_url for c in mock_open.call_args_list]
3864	assert any("19999" in u for u in called_urls)
3865	assert not any("11111" in u for u in called_urls)
3866
3867
3868	class TestProposalMergeE2E:
3869	"""End-to-end scenario tests for `muse hub proposal merge`."""
3870
3871	_HUB = "http://localhost:19999/gabriel/muse"
3872
3873	def _setup(self, repo: pathlib.Path) -> None:
3874	runner.invoke(cli, ["hub", "connect", self._HUB])
3875	_store_identity(self._HUB)
3876
3877	def _make_api_resp(self, payload_bytes: bytes) -> MagicMock:
3878	mock_resp = MagicMock()
3879	mock_resp.__enter__ = lambda s: s
3880	mock_resp.__exit__ = MagicMock(return_value=False)
3881	mock_resp.read.return_value = payload_bytes
3882	return mock_resp
3883
3884	def test_e2e_agent_safe_pipeline(self, repo: pathlib.Path) -> None:
3885	"""Agent pipeline: --json exits 0 on success so && chains correctly."""
3886	self._setup(repo)
3887	proposal_id = "deadbeef-cafe-0000-0000-000000000001"
3888	resps = [
3889	self._make_api_resp(json.dumps({"repo_id": "r"}).encode()),
3890	self._make_api_resp(json.dumps({"merged": True,
3891	"mergeCommitId": "cafebabe12345678"}).encode()),
3892	]
3893	with patch("urllib.request.urlopen", side_effect=resps):
3894	result = runner.invoke(cli, ["hub", "proposal", "merge", proposal_id, "--json"])
3895	assert result.exit_code == 0
3896	data = json.loads(next(
3897	l for l in result.output.splitlines() if l.strip().startswith("{")
3898	))
3899	assert data["merged"] is True
3900	assert data["mergeCommitId"] == "cafebabe12345678"
3901
3902	def test_e2e_agent_conflict_pipeline(self, repo: pathlib.Path) -> None:
3903	"""Agent pipeline: --json exits 3 on conflict so \|\| error-handling fires."""
3904	self._setup(repo)
3905	proposal_id = "deadbeef-cafe-0000-0000-000000000001"
3906	resps = [
3907	self._make_api_resp(json.dumps({"repo_id": "r"}).encode()),
3908	self._make_api_resp(
3909	json.dumps({"merged": False, "message": "merge conflict"}).encode()
3910	),
3911	]
3912	with patch("urllib.request.urlopen", side_effect=resps):
3913	result = runner.invoke(cli, ["hub", "proposal", "merge", proposal_id, "--json"])
3914	assert result.exit_code == 3
3915	# JSON is still printed so agent can read the error
3916	data = json.loads(next(
3917	l for l in result.output.splitlines() if l.strip().startswith("{")
3918	))
3919	assert data["merged"] is False
3920
3921	def test_e2e_squash_no_delete_branch(self, repo: pathlib.Path) -> None:
3922	"""Squash merge keeping the branch: payload and output both correct."""
3923	self._setup(repo)
3924	proposal_id = "abc12345-def0-0000-0000-000000000001"
3925	resps = [
3926	self._make_api_resp(json.dumps({"repo_id": "r"}).encode()),
3927	self._make_api_resp(
3928	json.dumps({"merged": True, "mergeCommitId": "aabbccdd11223344"}).encode()
3929	),
3930	]
3931	with patch("urllib.request.urlopen", side_effect=resps) as mock_open:
3932	result = runner.invoke(
3933	cli,
3934	["hub", "proposal", "merge", proposal_id,
3935	"--strategy", "squash", "--no-delete-branch"],
3936	)
3937	assert result.exit_code == 0
3938	assert "Source branch deleted" not in result.stderr
3939	assert "aabbccdd" in result.stderr
3940	post = next(c for c in mock_open.call_args_list if c[0][0].method == "POST")
3941	payload = json.loads(post[0][0].data)
3942	assert payload["mergeStrategy"] == "squash"
3943	assert payload["deleteBranch"] is False
3944
3945	def test_e2e_text_output_no_json_on_stdout(self, repo: pathlib.Path) -> None:
3946	"""In text mode, JSON must not appear on stdout."""
3947	self._setup(repo)
3948	proposal_id = "deadbeef-cafe-0000-0000-000000000001"
3949	resps = [
3950	self._make_api_resp(json.dumps({"repo_id": "r"}).encode()),
3951	self._make_api_resp(
3952	json.dumps({"merged": True, "mergeCommitId": "abc"}).encode()
3953	),
3954	]
3955	with patch("urllib.request.urlopen", side_effect=resps):
3956	result = runner.invoke(cli, ["hub", "proposal", "merge", proposal_id])
3957	assert result.exit_code == 0
3958	for line in result.output.splitlines():
3959	assert not line.strip().startswith("{"), (
3960	f"Unexpected JSON on stdout: {line!r}"
3961	)
3962
3963
3964	class TestProposalMergeStress:
3965	"""Stress tests for `muse hub proposal merge`."""
3966
3967	_HUB = "http://localhost:19999/gabriel/muse"
3968
3969	def test_concurrent_exit_code_checks(self) -> None:
3970	"""8 threads checking the merged=False exit-code logic must agree."""
3971	from muse.core.errors import ExitCode
3972	errors: list[str] = []
3973
3974	def _do(idx: int) -> None:
3975	try:
3976	# Simulate the merged check in pure Python
3977	data = {"merged": False, "message": f"conflict {idx}"}
3978	merged = bool(data.get("merged", False))
3979	expected_exit = ExitCode.INTERNAL_ERROR if not merged else ExitCode.SUCCESS
3980	assert expected_exit == ExitCode.INTERNAL_ERROR
3981	except Exception as exc:
3982	errors.append(f"Thread {idx}: {exc}")
3983
3984	threads = [threading.Thread(target=_do, args=(i,)) for i in range(8)]
3985	for t in threads:
3986	t.start()
3987	for t in threads:
3988	t.join()
3989	assert errors == [], "\n".join(errors)
3990
3991
3992	class TestResolveProposalIdLimit:
3993	"""Verify that _resolve_proposal_id respects _PROPOSAL_PREFIX_RESOLVE_LIMIT."""
3994
3995	def test_limit_constant_in_url(self) -> None:
3996	"""The URL sent to the API must include the limit constant."""
3997	from muse.cli.commands.hub import _PROPOSAL_PREFIX_RESOLVE_LIMIT, _resolve_proposal_id
3998	from muse.core.identity import IdentityEntry
3999
4000	identity: IdentityEntry = {"type": "human", "token": "tok"}
4001	proposal_id = "abc12345-0000-0000-0000-000000000001"
4002	proposals_resp = {"proposals": [
4003	{"proposalId": proposal_id, "title": "T"},
4004	]}
4005	captured_urls: list[str] = []
4006
4007	def _fake_urlopen(req: urllib.request.Request, timeout: int = 5, context: ssl.SSLContext \| None = None) -> MagicMock:
4008	captured_urls.append(req.full_url)
4009	mock_resp = MagicMock()
4010	mock_resp.__enter__ = lambda s: s
4011	mock_resp.__exit__ = MagicMock(return_value=False)
4012	mock_resp.read.return_value = json.dumps(proposals_resp).encode()
4013	return mock_resp
4014
4015	with patch("muse.cli.config.get_signing_identity", return_value=_make_signing()):
4016	with patch("urllib.request.urlopen", side_effect=_fake_urlopen):
4017	result = _resolve_proposal_id("http://localhost:9999", identity, "repo-id", "abc12345")
4018	assert result == proposal_id
4019	assert any(str(_PROPOSAL_PREFIX_RESOLVE_LIMIT) in url for url in captured_urls), (
4020	f"Expected {_PROPOSAL_PREFIX_RESOLVE_LIMIT} in one of {captured_urls}"
4021	)
4022
4023
4024	class TestResolveProposalIdSha256Passthrough:
4025	"""sha256-prefixed full IDs must be returned as-is without hitting the list endpoint.
4026
4027	Regression: the old full-ID check required a hyphen (`-`), so sha256:<hex>
4028	IDs always fell through to the list fetch with limit=200. Servers that cap
4029	the limit lower than 200 returned 422, making every `hub proposal read
4030	sha256:...` call fail on those hubs.
4031	"""
4032
4033	def _make_identity(self) -> "muse.core.identity.IdentityEntry":
4034	from muse.core.identity import IdentityEntry
4035	e: IdentityEntry = {"type": "human", "token": "tok123"}
4036	return e
4037
4038	def test_full_sha256_id_returned_as_is_no_network(self) -> None:
4039	"""A full sha256:<64-hex> ID must be returned without any network call."""
4040	from muse.cli.commands.hub import _resolve_proposal_id
4041
4042	full = "sha256:" + "a" * 64
4043	captured: list[str] = []
4044
4045	def _fail_urlopen(a: str, *kw: str) -> None:
4046	captured.append("called")
4047	raise AssertionError("urlopen must not be called for a full sha256 ID")
4048
4049	with patch("urllib.request.urlopen", side_effect=_fail_urlopen):
4050	result = _resolve_proposal_id("http://hub", self._make_identity(), "repo-id", full)
4051
4052	assert result == full
4053	assert captured == [], "urlopen was called — full sha256 ID was not detected as complete"
4054
4055	def test_sha256_prefix_still_resolves_via_list(self) -> None:
4056	"""A short sha256 prefix (fewer than 71 chars) still fetches the list."""
4057	from muse.cli.commands.hub import _resolve_proposal_id
4058
4059	full = "sha256:" + "b" * 64
4060	proposals_resp = {"proposals": [{"proposalId": full, "title": "T", "state": "open",
4061	"fromBranch": "feat/x", "toBranch": "dev"}]}
4062	mock_resp = MagicMock()
4063	mock_resp.__enter__ = lambda s: s
4064	mock_resp.__exit__ = MagicMock(return_value=False)
4065	mock_resp.read.return_value = json.dumps(proposals_resp).encode()
4066
4067	with patch("muse.cli.config.get_signing_identity", return_value=_make_signing()):
4068	with patch("urllib.request.urlopen", return_value=mock_resp):
4069	result = _resolve_proposal_id(
4070	"http://localhost:9999", self._make_identity(), "repo-id", "sha256:bbbb"
4071	)
4072	assert result == full
4073
4074	def test_hyphenated_uuid_still_returned_as_is(self) -> None:
4075	"""Regression: existing UUID-style full IDs must not be broken."""
4076	from muse.cli.commands.hub import _resolve_proposal_id
4077
4078	full = "af54753d-1234-5678-abcd-ef1234567890"
4079	with patch("urllib.request.urlopen", side_effect=AssertionError("must not call network")):
4080	result = _resolve_proposal_id("http://hub", self._make_identity(), "repo-id", full)
4081	assert result == full
4082
4083
4084	class TestProposalMerge422Regression:
4085	"""Regression for issue #54: hub proposal merge with a full sha256 ID must
4086	not call the proposals list endpoint.
4087
4088	Root cause: the old full-ID check in _resolve_proposal_id required a
4089	hyphen, so sha256:<hex> IDs fell through to the list fetch (?limit=200).
4090	Servers that capped limit at 100 returned 422 on that call, blocking every
4091	CLI merge regardless of strategy.
4092
4093	Fix: _resolve_proposal_id now calls split_id() first; sha256-prefixed IDs
4094	are returned as-is without any network round-trip, so the 422 can never
4095	occur on that path.
4096	"""
4097
4098	def test_merge_sha256_id_makes_no_list_call(self) -> None:
4099	"""run_proposal_merge with a full sha256 proposal ID must POST to
4100	/merge and never touch the proposals list endpoint."""
4101	import argparse
4102	from muse.cli.commands.hub.proposals import run_proposal_merge
4103
4104	proposal_id = "sha256:" + "c" * 64
4105	list_urls: list[str] = []
4106	merge_urls: list[str] = []
4107
4108	def _fake_urlopen(req: urllib.request.Request, timeout: int = 5,
4109	context: ssl.SSLContext \| None = None) -> MagicMock:
4110	url = req.full_url
4111	if "proposals?" in url:
4112	list_urls.append(url)
4113	if "/merge" in url and req.method == "POST":
4114	merge_urls.append(url)
4115	mock_resp = MagicMock()
4116	mock_resp.__enter__ = lambda s: s
4117	mock_resp.__exit__ = MagicMock(return_value=False)
4118	mock_resp.read.return_value = json.dumps({
4119	"merged": True,
4120	"mergeCommitId": "sha256:" + "d" * 64,
4121	}).encode()
4122	return mock_resp
4123
4124	args = argparse.Namespace(
4125	proposal_id=proposal_id,
4126	strategy="squash",
4127	delete_branch=False,
4128	json_output=False,
4129	hub="http://localhost:9999/owner/repo",
4130	)
4131
4132	with (
4133	patch("muse.cli.commands.hub.proposals._get_hub_and_identity",
4134	return_value=("http://localhost:9999/owner/repo",
4135	{"type": "human", "token": "tok"})),
4136	patch("muse.cli.commands.hub.proposals._resolve_repo_id",
4137	return_value="test-repo-id"),
4138	patch("muse.cli.config.get_signing_identity",
4139	return_value=_make_signing()),
4140	patch("urllib.request.urlopen", side_effect=_fake_urlopen),
4141	):
4142	run_proposal_merge(args)
4143
4144	assert not list_urls, (
4145	"run_proposal_merge must not call the proposals list endpoint "
4146	f"when given a full sha256 ID (triggers 422 on servers with "
4147	f"limit cap). Called: {list_urls}"
4148	)
4149	assert merge_urls, (
4150	"run_proposal_merge must POST to the /merge endpoint"
4151	)
4152
4153	def test_merge_prefix_id_calls_list_but_not_with_limit_exceeding_server_cap(
4154	self,
4155	) -> None:
4156	"""Short prefix IDs still resolve via the list endpoint, but the
4157	limit used must not exceed the server's PaginationParams cap (200)."""
4158	import argparse
4159	from muse.cli.commands.hub import _PROPOSAL_PREFIX_RESOLVE_LIMIT
4160
4161	assert _PROPOSAL_PREFIX_RESOLVE_LIMIT <= 200, (
4162	f"_PROPOSAL_PREFIX_RESOLVE_LIMIT is {_PROPOSAL_PREFIX_RESOLVE_LIMIT}, "
4163	"which exceeds the server's PaginationParams cap of 200. "
4164	"Lower the constant or raise the server cap to fix issue #54."
4165	)
4166
4167
4168	# =============================================================================
4169	# muse hub issue — hardening tests
4170	# =============================================================================
4171
4172	# Shared helpers for issue tests
4173	HUB_URL = "https://localhost:1337/owner/repo"
4174
4175
4176	def _issue_resp(
4177	number: int = 7,
4178	title: str = "feat: add thing",
4179	body: str = "",
4180	labels: list[str] \| None = None,
4181	issue_id: str = "iss_aabbccdd",
4182	state: str = "open",
4183	author: str = "alice",
4184	) -> _JsonPayload:
4185	return {
4186	"number": number,
4187	"title": title,
4188	"body": body,
4189	"labels": labels or [],
4190	"issueId": issue_id,
4191	"state": state,
4192	"author": author,
4193	"createdAt": "2026-04-09T00:00:00Z",
4194	}
4195
4196
4197	def _issue_list_resp(issues: list[_JsonPayload] \| None = None) -> _JsonPayload:
4198	"""Wrap issues in the list-response envelope."""
4199	items = issues if issues is not None else [_issue_resp()]
4200	return {"issues": items, "total": len(items)}
4201
4202
4203	def _comment_resp(comment_id: str = "c0") -> _JsonPayload:
4204	"""A single-comment response as returned by POST .../comments."""
4205	return {
4206	"commentId": comment_id,
4207	"issueId": "issue-id-0001",
4208	"author": "alice",
4209	"body": "test comment",
4210	"parentId": None,
4211	"isDeleted": False,
4212	"createdAt": "2026-04-14T00:00:00Z",
4213	"updatedAt": "2026-04-14T00:00:00Z",
4214	}
4215
4216
4217	def _refs_resp(repo_id: str = "repo-id-0001") -> _JsonPayload:
4218	return {"repo_id": repo_id, "branches": []}
4219
4220
4221	def _mock_responses(*payloads: _JsonPayload) -> list[MagicMock]:
4222	"""Build a side_effect list of mock HTTP responses for urlopen."""
4223	mocks = []
4224	for payload in payloads:
4225	m = MagicMock()
4226	m.__enter__ = lambda s: s
4227	m.__exit__ = MagicMock(return_value=False)
4228	m.read.return_value = json.dumps(payload).encode()
4229	mocks.append(m)
4230	return mocks
4231
4232
4233	# ---------------------------------------------------------------------------
4234	# TestIssueCreateHardening
4235	# ---------------------------------------------------------------------------
4236
4237
4238	class TestIssueCreateHardening:
4239	"""Integration tests for ``muse hub issue create``."""
4240
4241	def test_empty_title_exits_nonzero_no_network(
4242	self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
4243	) -> None:
4244	from muse.cli.config import set_hub_url
4245	set_hub_url(HUB_URL, repo)
4246	_store_identity(HUB_URL)
4247	with patch("urllib.request.urlopen") as mock_net:
4248	result = runner.invoke(cli, ["hub", "issue", "create", "--title", " "])
4249	assert result.exit_code != 0
4250	mock_net.assert_not_called()
4251
4252	def test_empty_title_error_message(
4253	self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
4254	) -> None:
4255	from muse.cli.config import set_hub_url
4256	set_hub_url(HUB_URL, repo)
4257	_store_identity(HUB_URL)
4258	with patch("urllib.request.urlopen"):
4259	result = runner.invoke(cli, ["hub", "issue", "create", "--title", ""])
4260	assert "empty" in result.stderr.lower() or "title" in result.stderr.lower()
4261
4262	def test_title_too_long_exits_nonzero_no_network(
4263	self, repo: pathlib.Path
4264	) -> None:
4265	from muse.cli.commands.hub import _MAX_ISSUE_TITLE_LEN
4266	from muse.cli.config import set_hub_url
4267	set_hub_url(HUB_URL, repo)
4268	_store_identity(HUB_URL)
4269	long_title = "x" * (_MAX_ISSUE_TITLE_LEN + 1)
4270	with patch("urllib.request.urlopen") as mock_net:
4271	result = runner.invoke(cli, ["hub", "issue", "create", "--title", long_title])
4272	assert result.exit_code != 0
4273	mock_net.assert_not_called()
4274
4275	def test_title_too_long_shows_char_count(
4276	self, repo: pathlib.Path
4277	) -> None:
4278	from muse.cli.commands.hub import _MAX_ISSUE_TITLE_LEN
4279	from muse.cli.config import set_hub_url
4280	set_hub_url(HUB_URL, repo)
4281	_store_identity(HUB_URL)
4282	long_title = "x" * (_MAX_ISSUE_TITLE_LEN + 1)
4283	with patch("urllib.request.urlopen"):
4284	result = runner.invoke(cli, ["hub", "issue", "create", "--title", long_title])
4285	assert str(_MAX_ISSUE_TITLE_LEN + 1) in result.stderr or str(_MAX_ISSUE_TITLE_LEN) in result.stderr
4286
4287	def test_title_at_max_length_accepted(
4288	self, repo: pathlib.Path
4289	) -> None:
4290	from muse.cli.commands.hub import _MAX_ISSUE_TITLE_LEN
4291	from muse.cli.config import set_hub_url
4292	set_hub_url(HUB_URL, repo)
4293	_store_identity(HUB_URL)
4294	exact_title = "x" * _MAX_ISSUE_TITLE_LEN
4295	mocks = _mock_responses(_refs_resp(), _issue_resp(title=exact_title))
4296	with patch("urllib.request.urlopen", side_effect=mocks):
4297	result = runner.invoke(
4298	cli, ["hub", "issue", "create", "--title", exact_title, "--json"]
4299	)
4300	assert result.exit_code == 0
4301
4302	def test_success_json_output(self, repo: pathlib.Path) -> None:
4303	from muse.cli.config import set_hub_url
4304	set_hub_url(HUB_URL, repo)
4305	_store_identity(HUB_URL)
4306	mocks = _mock_responses(_refs_resp(), _issue_resp())
4307	with patch("urllib.request.urlopen", side_effect=mocks):
4308	result = runner.invoke(
4309	cli, ["hub", "issue", "create", "--title", "feat: X", "-j"]
4310	)
4311	assert result.exit_code == 0
4312	data = json.loads(result.output)
4313	assert "number" in data
4314
4315	def test_json_short_flag(self, repo: pathlib.Path) -> None:
4316	"""-j short alias must work the same as --json."""
4317	from muse.cli.config import set_hub_url
4318	set_hub_url(HUB_URL, repo)
4319	_store_identity(HUB_URL)
4320	mocks = _mock_responses(_refs_resp(), _issue_resp())
4321	with patch("urllib.request.urlopen", side_effect=mocks):
4322	result = runner.invoke(
4323	cli, ["hub", "issue", "create", "--title", "feat: X", "-j"]
4324	)
4325	assert result.exit_code == 0
4326	json.loads(result.output) # must be valid JSON
4327
4328	def test_labels_included_in_payload(self, repo: pathlib.Path) -> None:
4329	from muse.cli.config import set_hub_url
4330	set_hub_url(HUB_URL, repo)
4331	_store_identity(HUB_URL)
4332	captured: list[bytes] = []
4333
4334	def _fake(req: urllib.request.Request, timeout: int = 5, context: ssl.SSLContext \| None = None) -> MagicMock:
4335	if req.method == "POST":
4336	captured.append(req.data or b"")
4337	m = MagicMock()
4338	m.__enter__ = lambda s: s
4339	m.__exit__ = MagicMock(return_value=False)
4340	if req.method == "GET":
4341	m.read.return_value = json.dumps(_refs_resp()).encode()
4342	else:
4343	m.read.return_value = json.dumps(_issue_resp()).encode()
4344	return m
4345
4346	with patch("urllib.request.urlopen", side_effect=_fake):
4347	runner.invoke(
4348	cli,
4349	["hub", "issue", "create", "--title", "T", "--label", "bug", "--label", "phase/1"],
4350	)
4351	assert captured
4352	body = json.loads(captured[0])
4353	assert "bug" in body["labels"]
4354	assert "phase/1" in body["labels"]
4355
4356	def test_issue_url_on_stdout(self, repo: pathlib.Path) -> None:
4357	from muse.cli.config import set_hub_url
4358	set_hub_url(HUB_URL, repo)
4359	_store_identity(HUB_URL)
4360	mocks = _mock_responses(_refs_resp(), _issue_resp(number=42))
4361	with patch("urllib.request.urlopen", side_effect=mocks):
4362	result = runner.invoke(cli, ["hub", "issue", "create", "--title", "T"])
4363	assert result.exit_code == 0
4364	assert "42" in result.stderr
4365
4366	def test_issue_url_contains_owner_slug(self, repo: pathlib.Path) -> None:
4367	from muse.cli.config import set_hub_url
4368	set_hub_url(HUB_URL, repo)
4369	_store_identity(HUB_URL)
4370	mocks = _mock_responses(_refs_resp(), _issue_resp(number=3))
4371	with patch("urllib.request.urlopen", side_effect=mocks):
4372	result = runner.invoke(cli, ["hub", "issue", "create", "--title", "T"])
4373	assert "owner" in result.output
4374	assert "repo" in result.output
4375
4376	def test_text_mode_success_on_stderr(self, repo: pathlib.Path) -> None:
4377	"""Text mode prints ✅ Issue #N created. to stderr."""
4378	from muse.cli.config import set_hub_url
4379	set_hub_url(HUB_URL, repo)
4380	_store_identity(HUB_URL)
4381	mocks = _mock_responses(_refs_resp(), _issue_resp(number=5))
4382	with patch("urllib.request.urlopen", side_effect=mocks):
4383	result = runner.invoke(cli, ["hub", "issue", "create", "--title", "T"])
4384	assert result.exit_code == 0
4385	assert "5" in result.stderr
4386	assert "created" in result.stderr.lower()
4387
4388	def test_text_mode_no_json_on_stdout(self, repo: pathlib.Path) -> None:
4389	from muse.cli.config import set_hub_url
4390	set_hub_url(HUB_URL, repo)
4391	_store_identity(HUB_URL)
4392	mocks = _mock_responses(_refs_resp(), _issue_resp())
4393	with patch("urllib.request.urlopen", side_effect=mocks):
4394	result = runner.invoke(cli, ["hub", "issue", "create", "--title", "T"])
4395	assert result.exit_code == 0
4396	# Text mode must not emit a JSON object
4397	try:
4398	json.loads(result.output)
4399	assert False, "Text mode must not emit JSON"
4400	except (json.JSONDecodeError, ValueError):
4401	pass
4402
4403	def test_number_fallback_for_nonnumeric_api_response(
4404	self

File truncated at 200 KB — view full file ↗

test_cmd_hub_hardening.py file-level

`test_cmd_hub_hardening.py` file-level