aaronrene/gabriel-muse — blame/main/test_cmd_hub_hardening.py

1 files

1 commits

0 hotspots

0 🧊 dead

0 💥 blast risk

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

File truncated at 200 KB — view full file ↗

test_cmd_hub_hardening.py file-level

`test_cmd_hub_hardening.py` file-level