tests/test_cmd_hub_hardening.py · gabriel/muse

test_cmd_hub_hardening.py python

5,845 lines 254.7 KB

sha256:1c4b3e3a9a1f300774c3ee662b572a698d5fd405bf765a71e6011a2e9c3eaaaa feat: Muse — version control for the agent era Human 151 days ago

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_view: --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	cli = None
71	runner = CliRunner()
72
73	# ── helpers ───────────────────────────────────────────────────────────────────
74
75
76	def _json_line(result: InvokeResult) -> _JsonPayload:
77	for line in result.output.splitlines():
78	stripped = line.strip()
79	if stripped.startswith("{") or stripped.startswith("["):
80	data: _JsonPayload = json.loads(stripped)
81	return data
82	raise ValueError(f"No JSON line in output:\n{result.output!r}")
83
84
85	def _json_connect(result: InvokeResult) -> _ConnectJson:
86	d: _ConnectJson = json.loads(
87	next(l for l in result.output.splitlines() if l.strip().startswith("{"))
88	)
89	return d
90
91
92	def _json_status(result: InvokeResult) -> _StatusJson:
93	d: _StatusJson = json.loads(
94	next(l for l in result.output.splitlines() if l.strip().startswith("{"))
95	)
96	return d
97
98
99	def _json_disconnect(result: InvokeResult) -> _DisconnectJson:
100	d: _DisconnectJson = json.loads(
101	next(l for l in result.output.splitlines() if l.strip().startswith("{"))
102	)
103	return d
104
105
106	def _json_ping(result: InvokeResult) -> _PingJson:
107	d: _PingJson = json.loads(
108	next(l for l in result.output.splitlines() if l.strip().startswith("{"))
109	)
110	return d
111
112
113	@pytest.fixture
114	def repo(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path:
115	"""Minimal .muse/ repo with identity file."""
116	from muse._version import __version__
117
118	muse_dir = tmp_path / ".muse"
119	for sub in ("refs/heads", "objects", "commits", "snapshots"):
120	(muse_dir / sub).mkdir(parents=True, exist_ok=True)
121	(muse_dir / "repo.json").write_text(
122	json.dumps({"repo_id": "test-repo", "schema_version": __version__, "domain": "code"})
123	)
124	(muse_dir / "HEAD").write_text("ref: refs/heads/main\n")
125	(muse_dir / "refs" / "heads" / "main").write_text("")
126	(muse_dir / "config.toml").write_text("")
127	muse_home = tmp_path / ".muse-home"
128	muse_home.mkdir()
129	(muse_home / "identity.toml").write_text("")
130	monkeypatch.setattr("muse.core.identity._IDENTITY_FILE", muse_home / "identity.toml")
131	monkeypatch.setattr("muse.core.identity._IDENTITY_DIR", muse_home)
132	monkeypatch.chdir(tmp_path)
133	return tmp_path
134
135
136	def _make_signing() -> "SigningIdentity":
137	"""Generate a fresh Ed25519 SigningIdentity for tests."""
138	from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
139	from muse.core.transport import SigningIdentity
140	return SigningIdentity(handle="testuser", private_key=Ed25519PrivateKey.generate())
141
142
143	def _store_identity(hub_url: str, handle: str = "alice") -> None:
144	"""Save a proper Ed25519 identity entry for the given hub URL."""
145	import urllib.parse
146	from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
147	from cryptography.hazmat.primitives.serialization import Encoding, NoEncryption, PrivateFormat
148	from muse.core.identity import IdentityEntry, _IDENTITY_DIR, save_identity
149
150	keys_dir = _IDENTITY_DIR / "keys"
151	keys_dir.mkdir(parents=True, exist_ok=True)
152	parsed = urllib.parse.urlparse(hub_url)
153	hostname = parsed.netloc or parsed.path
154	safe_hostname = hostname.replace(":", "_").replace("/", "_")
155	key_file = keys_dir / f"{safe_hostname}.pem"
156
157	private_key = Ed25519PrivateKey.generate()
158	pem = private_key.private_bytes(Encoding.PEM, PrivateFormat.PKCS8, NoEncryption())
159	key_file.write_bytes(pem)
160
161	entry: IdentityEntry = {"type": "human", "handle": handle, "key_path": str(key_file)}
162	save_identity(hub_url, entry)
163
164
165	# ── Unit: _normalise_url ──────────────────────────────────────────────────────
166
167
168	class TestNormaliseUrlHardening:
169	def test_file_scheme_raises(self) -> None:
170	from muse.cli.commands.hub import _normalise_url
171	with pytest.raises(ValueError, match="not allowed"):
172	_normalise_url("file:///etc/passwd")
173
174	def test_ftp_scheme_raises(self) -> None:
175	from muse.cli.commands.hub import _normalise_url
176	with pytest.raises(ValueError, match="not allowed"):
177	_normalise_url("ftp://evil.example.com/repo")
178
179	def test_data_scheme_raises(self) -> None:
180	from muse.cli.commands.hub import _normalise_url
181	with pytest.raises(ValueError, match="not allowed"):
182	_normalise_url("data:text/plain,malicious")
183
184	def test_http_non_loopback_raises(self) -> None:
185	from muse.cli.commands.hub import _normalise_url
186	with pytest.raises(ValueError, match="HTTPS"):
187	_normalise_url("http://musehub.ai/gabriel/muse")
188
189	def test_http_localhost_allowed(self) -> None:
190	from muse.cli.commands.hub import _normalise_url
191	assert _normalise_url("http://localhost:10003") == "http://localhost:10003"
192
193	def test_http_127_allowed(self) -> None:
194	from muse.cli.commands.hub import _normalise_url
195	assert _normalise_url("http://127.0.0.1:9000") == "http://127.0.0.1:9000"
196
197	def test_schemeless_becomes_https(self) -> None:
198	from muse.cli.commands.hub import _normalise_url
199	assert _normalise_url("musehub.ai").startswith("https://")
200
201	def test_trailing_slash_stripped(self) -> None:
202	from muse.cli.commands.hub import _normalise_url
203	assert not _normalise_url("https://musehub.ai/").endswith("/")
204
205	def test_https_passthrough(self) -> None:
206	from muse.cli.commands.hub import _normalise_url
207	assert _normalise_url("https://musehub.ai/gabriel/muse") == "https://musehub.ai/gabriel/muse"
208
209
210	# ── Unit: _hub_hostname ───────────────────────────────────────────────────────
211
212
213	class TestHubHostname:
214	def test_plain_https(self) -> None:
215	from muse.cli.commands.hub import _hub_hostname
216	assert _hub_hostname("https://musehub.ai/gabriel/muse") == "musehub.ai"
217
218	def test_with_port(self) -> None:
219	from muse.cli.commands.hub import _hub_hostname
220	assert _hub_hostname("http://localhost:10003/gabriel/muse") == "localhost:10003"
221
222	def test_bare_hostname(self) -> None:
223	from muse.cli.commands.hub import _hub_hostname
224	assert _hub_hostname("musehub.ai") == "musehub.ai"
225
226	def test_trailing_slash(self) -> None:
227	from muse.cli.commands.hub import _hub_hostname
228	assert _hub_hostname("https://musehub.ai/") == "musehub.ai"
229
230
231	# ── Unit: _hub_api ────────────────────────────────────────────────────────────
232
233
234	class TestHubApi:
235	_IDENTITY = {"type": "human", "token": "tok123"}
236
237	def test_file_scheme_blocked_before_network(self) -> None:
238	from muse.cli.commands.hub import _hub_api
239	from muse.core.identity import IdentityEntry
240	identity: IdentityEntry = {"type": "human", "token": "tok"}
241	with patch("urllib.request.urlopen") as mock_net:
242	with pytest.raises(SystemExit):
243	_hub_api("file:///etc/passwd", identity, "GET", "/api/test")
244	mock_net.assert_not_called()
245
246	def test_ftp_scheme_blocked_before_network(self) -> None:
247	from muse.cli.commands.hub import _hub_api
248	from muse.core.identity import IdentityEntry
249	identity: IdentityEntry = {"type": "human", "token": "tok"}
250	with patch("urllib.request.urlopen") as mock_net:
251	with pytest.raises(SystemExit):
252	_hub_api("ftp://ftp.example.com", identity, "GET", "/api/test")
253	mock_net.assert_not_called()
254
255	def test_missing_token_exits(self) -> None:
256	from muse.cli.commands.hub import _hub_api
257	from muse.core.identity import IdentityEntry
258	identity: IdentityEntry = {"type": "human", "token": ""}
259	with pytest.raises(SystemExit):
260	_hub_api("http://localhost:10003", identity, "GET", "/api/test")
261
262	def test_response_size_cap(self) -> None:
263	from muse.cli.commands.hub import _MAX_API_RESPONSE_BYTES, _hub_api
264	from muse.core.identity import IdentityEntry
265	identity: IdentityEntry = {"type": "human", "token": "tok"}
266	mock_resp = MagicMock()
267	mock_resp.__enter__ = lambda s: s
268	mock_resp.__exit__ = MagicMock(return_value=False)
269	mock_resp.read.return_value = b"x" * (_MAX_API_RESPONSE_BYTES + 2)
270	with patch("muse.cli.config.get_signing_identity", return_value=_make_signing()):
271	with patch("urllib.request.urlopen", return_value=mock_resp):
272	with pytest.raises(SystemExit):
273	_hub_api("http://localhost:9999", identity, "GET", "/api/test")
274
275	def test_http_error_sanitized_in_output(
276	self, capsys: pytest.CaptureFixture[str]
277	) -> None:
278	import urllib.error
279	from muse.cli.commands.hub import _hub_api
280	from muse.core.identity import IdentityEntry
281
282	import io
283	identity: IdentityEntry = {"type": "human", "token": "tok"}
284	ansi_detail = b'{"detail":"\\x1b[31mevil\\x1b[0m"}'
285	exc = urllib.error.HTTPError(url="", code=403, msg="Forbidden", hdrs=MagicMock(), fp=io.BytesIO(ansi_detail))
286
287	with patch("muse.cli.config.get_signing_identity", return_value=_make_signing()):
288	with patch("urllib.request.urlopen", side_effect=exc):
289	with pytest.raises(SystemExit):
290	_hub_api("http://localhost:9999", identity, "GET", "/api/test")
291
292	captured = capsys.readouterr()
293	assert "\x1b[" not in captured.err
294
295	def test_empty_response_returns_empty_dict(self) -> None:
296	from muse.cli.commands.hub import _hub_api
297	from muse.core.identity import IdentityEntry
298
299	identity: IdentityEntry = {"type": "human", "token": "tok"}
300	mock_resp = MagicMock()
301	mock_resp.__enter__ = lambda s: s
302	mock_resp.__exit__ = MagicMock(return_value=False)
303	mock_resp.read.return_value = b""
304	with patch("muse.cli.config.get_signing_identity", return_value=_make_signing()):
305	with patch("urllib.request.urlopen", return_value=mock_resp):
306	result = _hub_api("http://localhost:9999", identity, "GET", "/api/test")
307	assert result == {}
308
309
310	# ── Unit: _resolve_proposal_id ──────────────────────────────────────────────────────
311
312
313	class TestResolveProposalId:
314	def _make_identity(self) -> "muse.core.identity.IdentityEntry":
315	from muse.core.identity import IdentityEntry
316	e: IdentityEntry = {"type": "human", "token": "tok123"}
317	return e
318
319	def _proposal(self, proposal_id: str, title: str = "Test Proposal") -> _ProposalRecord:
320	return {"proposalId": proposal_id, "title": title, "state": "open",
321	"fromBranch": "feat/x", "toBranch": "dev"}
322
323	def test_full_uuid_returned_as_is(self) -> None:
324	from muse.cli.commands.hub import _resolve_proposal_id
325	full = "af54753d-1234-5678-abcd-ef1234567890"
326	result = _resolve_proposal_id("http://hub", self._make_identity(), "repo-id", full)
327	assert result == full
328
329	def test_prefix_resolved(self) -> None:
330	from muse.cli.commands.hub import _resolve_proposal_id
331
332	proposal_id = "abc12345-6789-0000-0000-000000000000"
333	proposals_resp = {"proposals": [self._proposal(proposal_id)]}
334	mock_resp = MagicMock()
335	mock_resp.__enter__ = lambda s: s
336	mock_resp.__exit__ = MagicMock(return_value=False)
337	mock_resp.read.return_value = json.dumps(proposals_resp).encode()
338	with patch("muse.cli.config.get_signing_identity", return_value=_make_signing()):
339	with patch("urllib.request.urlopen", return_value=mock_resp):
340	result = _resolve_proposal_id(
341	"http://localhost:9999", self._make_identity(), "repo-id", "abc12345"
342	)
343	assert result == proposal_id
344
345	def test_no_match_exits(self) -> None:
346	from muse.cli.commands.hub import _resolve_proposal_id
347
348	resp_bytes = json.dumps({"proposals": []}).encode()
349	mock_resp = MagicMock()
350	mock_resp.__enter__ = lambda s: s
351	mock_resp.__exit__ = MagicMock(return_value=False)
352	mock_resp.read.return_value = resp_bytes
353	with patch("muse.cli.config.get_signing_identity", return_value=_make_signing()):
354	with patch("urllib.request.urlopen", return_value=mock_resp):
355	with pytest.raises(SystemExit):
356	_resolve_proposal_id(
357	"http://localhost:9999", self._make_identity(), "repo-id", "deadbeef"
358	)
359
360	def test_ambiguous_prefix_exits(self) -> None:
361	from muse.cli.commands.hub import _resolve_proposal_id
362
363	pr1_id = "abc12345-0000-0000-0000-000000000001"
364	pr2_id = "abc12345-0000-0000-0000-000000000002"
365	proposals_resp = {"proposals": [self._proposal(pr1_id), self._proposal(pr2_id)]}
366	mock_resp = MagicMock()
367	mock_resp.__enter__ = lambda s: s
368	mock_resp.__exit__ = MagicMock(return_value=False)
369	mock_resp.read.return_value = json.dumps(proposals_resp).encode()
370	with patch("muse.cli.config.get_signing_identity", return_value=_make_signing()):
371	with patch("urllib.request.urlopen", return_value=mock_resp):
372	with pytest.raises(SystemExit):
373	_resolve_proposal_id(
374	"http://localhost:9999", self._make_identity(), "repo-id", "abc12345"
375	)
376
377	def test_ansi_in_proposal_id_sanitized_in_error(
378	self, capsys: pytest.CaptureFixture[str]
379	) -> None:
380	from muse.cli.commands.hub import _resolve_proposal_id
381
382	resp_bytes = json.dumps({"proposals": []}).encode()
383	mock_resp = MagicMock()
384	mock_resp.__enter__ = lambda s: s
385	mock_resp.__exit__ = MagicMock(return_value=False)
386	mock_resp.read.return_value = resp_bytes
387	evil_proposalefix = "\x1b[31mevil\x1b[0m"
388	with patch("muse.cli.config.get_signing_identity", return_value=_make_signing()):
389	with patch("urllib.request.urlopen", return_value=mock_resp):
390	with pytest.raises(SystemExit):
391	_resolve_proposal_id(
392	"http://localhost:9999", self._make_identity(), "repo-id", evil_proposalefix
393	)
394	captured = capsys.readouterr()
395	assert "\x1b[" not in captured.err
396
397
398	# ── Unit: _format_proposal ──────────────────────────────────────────────────────────
399
400
401	class TestFormatProposal:
402	def test_ansi_in_title_stripped(self) -> None:
403	from muse.cli.commands.hub import _format_proposal
404	proposal: _ProposalRecord = {
405	"proposalId": "abc12345",
406	"title": "\x1b[31mevil title\x1b[0m",
407	"state": "open",
408	"fromBranch": "feat/x",
409	"toBranch": "dev",
410	}
411	result = _format_proposal(proposal)
412	assert "\x1b[" not in result
413
414	def test_ansi_in_branch_stripped(self) -> None:
415	from muse.cli.commands.hub import _format_proposal
416	proposal: _ProposalRecord = {
417	"proposalId": "abc12345",
418	"title": "clean title",
419	"state": "open",
420	"fromBranch": "\x1b[32mfeat/evil\x1b[0m",
421	"toBranch": "\x1b[31mdev\x1b[0m",
422	}
423	result = _format_proposal(proposal)
424	assert "\x1b[" not in result
425
426	def test_state_icon_open(self) -> None:
427	from muse.cli.commands.hub import _format_proposal
428	proposal: _ProposalRecord = {
429	"proposalId": "abc12345", "title": "t", "state": "open",
430	"fromBranch": "f", "toBranch": "d",
431	}
432	assert "🟢" in _format_proposal(proposal)
433
434	def test_state_icon_merged(self) -> None:
435	from muse.cli.commands.hub import _format_proposal
436	proposal: _ProposalRecord = {
437	"proposalId": "abc12345", "title": "t", "state": "merged",
438	"fromBranch": "f", "toBranch": "d",
439	}
440	assert "🟣" in _format_proposal(proposal)
441
442
443	# ── Integration: run_connect ──────────────────────────────────────────────────
444
445
446	class TestConnectHardening:
447	_HUB = "http://localhost:19999"
448
449	def test_connect_json_schema(self, repo: pathlib.Path) -> None:
450	result = runner.invoke(cli, ["hub", "connect", self._HUB, "--json"])
451	assert result.exit_code == 0
452	data = _json_connect(result)
453	for key in ("status", "hub_url", "hostname", "authenticated",
454	"identity_name", "identity_type"):
455	assert key in data, f"Missing key: {key}"
456	assert data["status"] == "ok"
457	assert data["authenticated"] is False
458	assert data["identity_name"] == ""
459	assert data["identity_type"] == ""
460
461	def test_connect_authenticated_json_schema(self, repo: pathlib.Path) -> None:
462	_store_identity(self._HUB)
463	result = runner.invoke(cli, ["hub", "connect", self._HUB, "--json"])
464	assert result.exit_code == 0
465	data = _json_connect(result)
466	assert data["authenticated"] is True
467	assert data["identity_name"] == "alice"
468	assert data["identity_type"] == "human"
469
470	def test_connect_invalid_scheme_exits(self, repo: pathlib.Path) -> None:
471	result = runner.invoke(cli, ["hub", "connect", "file:///etc/passwd"])
472	assert result.exit_code != 0
473
474	def test_connect_http_non_loopback_exits(self, repo: pathlib.Path) -> None:
475	result = runner.invoke(cli, ["hub", "connect", "http://musehub.ai"])
476	assert result.exit_code != 0
477
478	def test_connect_json_stdout_clean(self, repo: pathlib.Path) -> None:
479	result = runner.invoke(cli, ["hub", "connect", self._HUB, "--json"])
480	assert result.exit_code == 0
481	json_lines = [l for l in result.output.splitlines() if l.strip().startswith("{")]
482	assert len(json_lines) >= 1
483
484	def test_connect_no_repo_exits(self, tmp_path: pathlib.Path,
485	monkeypatch: pytest.MonkeyPatch) -> None:
486	monkeypatch.chdir(tmp_path)
487	result = runner.invoke(cli, ["hub", "connect", self._HUB])
488	assert result.exit_code != 0
489
490	def test_reconnect_warning_on_stderr(self, repo: pathlib.Path) -> None:
491	runner.invoke(cli, ["hub", "connect", self._HUB])
492	result = runner.invoke(cli, ["hub", "connect", "http://localhost:20000"])
493	assert result.exit_code == 0
494	assert "localhost:19999" in result.output
495
496	def test_reconnect_same_url_no_warning(self, repo: pathlib.Path) -> None:
497	"""Re-connecting to the same URL is a no-op — no warning emitted."""
498	runner.invoke(cli, ["hub", "connect", self._HUB])
499	result = runner.invoke(cli, ["hub", "connect", self._HUB])
500	assert result.exit_code == 0
501	assert "Switching" not in result.output
502	assert "⚠️" not in result.output
503
504	def test_connect_short_flag_j(self, repo: pathlib.Path) -> None:
505	"""-j short flag produces identical JSON output to --json."""
506	r_long = runner.invoke(cli, ["hub", "connect", self._HUB, "--json"])
507	runner.invoke(cli, ["hub", "disconnect"])
508	r_short = runner.invoke(cli, ["hub", "connect", self._HUB, "-j"])
509	assert r_long.exit_code == 0
510	assert r_short.exit_code == 0
511	d_long = _json_connect(r_long)
512	d_short = _json_connect(r_short)
513	assert d_long == d_short
514
515	def test_connect_ipv6_loopback_accepted(self, repo: pathlib.Path) -> None:
516	"""http://[::1] and http://[::1]:PORT are valid loopback URLs."""
517	result = runner.invoke(cli, ["hub", "connect", "http://[::1]:8080", "--json"])
518	assert result.exit_code == 0
519	data = _json_connect(result)
520	assert data["status"] == "ok"
521	assert "::1" in data["hub_url"]
522
523	def test_connect_ipv6_loopback_bare_accepted(self, repo: pathlib.Path) -> None:
524	"""http://[::1] without a port is valid."""
525	result = runner.invoke(cli, ["hub", "connect", "http://[::1]", "--json"])
526	assert result.exit_code == 0
527	data = _json_connect(result)
528	assert data["status"] == "ok"
529
530	def test_connect_bare_hostname_with_port(self, repo: pathlib.Path) -> None:
531	"""musehub.ai:8443 (no scheme) is promoted to https://musehub.ai:8443."""
532	result = runner.invoke(cli, ["hub", "connect", "musehub.ai:8443", "--json"])
533	assert result.exit_code == 0
534	data = _json_connect(result)
535	assert data["hub_url"] == "https://musehub.ai:8443"
536	assert data["hostname"] == "musehub.ai:8443"
537
538	def test_connect_trailing_slash_stripped(self, repo: pathlib.Path) -> None:
539	"""Trailing slashes are stripped from the stored URL."""
540	result = runner.invoke(
541	cli, ["hub", "connect", "https://musehub.ai/", "--json"]
542	)
543	assert result.exit_code == 0
544	data = _json_connect(result)
545	assert not data["hub_url"].endswith("/")
546
547	def test_connect_ansi_in_reconnect_warning_sanitized(
548	self, repo: pathlib.Path
549	) -> None:
550	"""ANSI codes stored in config are stripped from the reconnect warning."""
551	import unittest.mock
552	ansi_url = "https://\x1b[31mevil.example.com\x1b[0m"
553	with unittest.mock.patch(
554	"muse.cli.commands.hub.get_hub_url", return_value=ansi_url
555	):
556	result = runner.invoke(
557	cli, ["hub", "connect", "https://safe.example.com"]
558	)
559	assert "\x1b" not in result.output, "ANSI escape leaked into reconnect warning"
560
561	def test_connect_json_hub_url_normalised(self, repo: pathlib.Path) -> None:
562	"""hub_url in JSON is the normalised form (no trailing slash, has scheme)."""
563	result = runner.invoke(
564	cli, ["hub", "connect", "musehub.ai", "--json"]
565	)
566	assert result.exit_code == 0
567	data = _json_connect(result)
568	assert data["hub_url"].startswith("https://")
569	assert not data["hub_url"].endswith("/")
570
571	def test_connect_no_repo_exits_2(self, tmp_path: pathlib.Path,
572	monkeypatch: pytest.MonkeyPatch) -> None:
573	"""Exit code 2 (REPO_NOT_FOUND) when outside a Muse repo."""
574	monkeypatch.chdir(tmp_path)
575	result = runner.invoke(cli, ["hub", "connect", self._HUB])
576	assert result.exit_code == 2
577
578	def test_connect_http_non_loopback_exits_1(self, repo: pathlib.Path) -> None:
579	"""Exit code 1 (USER_ERROR) for http:// non-loopback URL."""
580	result = runner.invoke(cli, ["hub", "connect", "http://remote.example.com"])
581	assert result.exit_code == 1
582
583	def test_connect_disallowed_scheme_exits_1(self, repo: pathlib.Path) -> None:
584	"""Exit code 1 (USER_ERROR) for ftp:// URL."""
585	result = runner.invoke(cli, ["hub", "connect", "ftp://musehub.ai"])
586	assert result.exit_code == 1
587
588	def test_10_sequential_connects_all_survive(self, repo: pathlib.Path) -> None:
589	"""10 sequential connect→disconnect cycles all succeed."""
590	for i in range(10):
591	hub = f"http://localhost:{19000 + i}"
592	r = runner.invoke(cli, ["hub", "connect", hub, "--json"])
593	assert r.exit_code == 0, f"connect {i} failed: {r.output}"
594	data = _json_connect(r)
595	assert data["status"] == "ok"
596
597
598	# ── Integration: run_status ───────────────────────────────────────────────────
599
600
601	class TestStatusHardening:
602	_HUB = "http://localhost:19999"
603
604	def test_status_no_hub_exits_nonzero(self, repo: pathlib.Path) -> None:
605	result = runner.invoke(cli, ["hub", "status", "--json"])
606	assert result.exit_code != 0
607
608	def test_status_json_all_keys_always_present(self, repo: pathlib.Path) -> None:
609	"""All 7 JSON keys present even when not authenticated."""
610	runner.invoke(cli, ["hub", "connect", self._HUB])
611	result = runner.invoke(cli, ["hub", "status", "--json"])
612	assert result.exit_code == 0
613	data = _json_status(result)
614	for key in ("hub_url", "hostname", "authenticated", "identity_type",
615	"identity_name", "identity_id", "capabilities"):
616	assert key in data, f"Missing key: {key}"
617	assert data["authenticated"] is False
618	assert data["identity_type"] == ""
619	assert data["identity_name"] == ""
620	assert data["identity_id"] == ""
621	assert data["capabilities"] == []
622
623	def test_status_json_authenticated(self, repo: pathlib.Path) -> None:
624	runner.invoke(cli, ["hub", "connect", self._HUB])
625	_store_identity(self._HUB)
626	result = runner.invoke(cli, ["hub", "status", "--json"])
627	assert result.exit_code == 0
628	data = _json_status(result)
629	assert data["authenticated"] is True
630	assert data["identity_name"] == "alice"
631	assert data["identity_type"] == "human"
632
633	def test_status_json_stdout_clean(self, repo: pathlib.Path) -> None:
634	runner.invoke(cli, ["hub", "connect", self._HUB])
635	result = runner.invoke(cli, ["hub", "status", "--json"])
636	json_lines = [l for l in result.output.splitlines() if l.strip().startswith("{")]
637	assert len(json_lines) >= 1
638
639	def test_status_text_mode_to_stderr(self, repo: pathlib.Path,
640	capsys: pytest.CaptureFixture[str]) -> None:
641	runner.invoke(cli, ["hub", "connect", self._HUB])
642	result = runner.invoke(cli, ["hub", "status"])
643	assert result.exit_code == 0
644
645	def test_status_short_flag_j(self, repo: pathlib.Path) -> None:
646	"""-j short flag produces identical JSON output to --json."""
647	runner.invoke(cli, ["hub", "connect", self._HUB])
648	r_long = runner.invoke(cli, ["hub", "status", "--json"])
649	r_short = runner.invoke(cli, ["hub", "status", "-j"])
650	assert r_long.exit_code == 0
651	assert r_short.exit_code == 0
652	assert json.loads(r_long.output) == json.loads(r_short.output)
653
654	def test_status_exit_code_2_no_repo(
655	self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
656	) -> None:
657	"""Exit code 2 (REPO_NOT_FOUND) when outside a Muse repo."""
658	monkeypatch.chdir(tmp_path)
659	result = runner.invoke(cli, ["hub", "status"])
660	assert result.exit_code == 2
661
662	def test_status_exit_code_1_no_hub(self, repo: pathlib.Path) -> None:
663	"""Exit code 1 (USER_ERROR) when no hub is connected."""
664	result = runner.invoke(cli, ["hub", "status"])
665	assert result.exit_code == 1
666
667	def test_status_hub_override_flag(self, repo: pathlib.Path) -> None:
668	"""--hub flag overrides config; identity is looked up for that URL."""
669	override = "http://localhost:29999"
670	from muse.core.identity import IdentityEntry, save_identity
671	entry: IdentityEntry = {
672	"type": "agent", "handle": "override-bot",
673	}
674	save_identity(override, entry)
675	# Connect to a different hub
676	runner.invoke(cli, ["hub", "connect", self._HUB])
677	result = runner.invoke(
678	cli, ["hub", "status", "--hub", override, "--json"]
679	)
680	assert result.exit_code == 0
681	data = _json_status(result)
682	assert data["authenticated"] is True
683	assert data["identity_name"] == "override-bot"
684
685	def test_status_json_capabilities_populated_for_agent(
686	self, repo: pathlib.Path
687	) -> None:
688	"""capabilities field is populated from agent identity."""
689	from muse.core.identity import IdentityEntry, save_identity
690	entry: IdentityEntry = {
691	"type": "agent",
692	"handle": "cap-bot",
693	"capabilities": ["read:*", "write:midi", "commit"],
694	}
695	save_identity(self._HUB, entry)
696	runner.invoke(cli, ["hub", "connect", self._HUB])
697	result = runner.invoke(cli, ["hub", "status", "--json"])
698	assert result.exit_code == 0
699	data = _json_status(result)
700	assert data["capabilities"] == ["read:*", "write:midi", "commit"]
701
702	def test_status_capabilities_empty_for_human(self, repo: pathlib.Path) -> None:
703	"""capabilities is [] for human identities (they have no cap list)."""
704	runner.invoke(cli, ["hub", "connect", self._HUB])
705	_store_identity(self._HUB) # human identity, no capabilities
706	result = runner.invoke(cli, ["hub", "status", "--json"])
707	assert result.exit_code == 0
708	data = _json_status(result)
709	assert data["capabilities"] == []
710
711	def test_status_ansi_in_identity_fields_sanitized(
712	self, repo: pathlib.Path
713	) -> None:
714	"""ANSI codes in identity_type, identity_name, identity_id stripped in text output."""
715	import unittest.mock
716	ansi_entry = {
717	"type": "\x1b[31magent\x1b[0m",
718	"handle": "\x1b[32mevil-bot\x1b[0m",
719	}
720	with unittest.mock.patch(
721	"muse.core.identity._load_all",
722	return_value={"localhost:19999": ansi_entry},
723	):
724	runner.invoke(cli, ["hub", "connect", self._HUB])
725	result = runner.invoke(cli, ["hub", "status"])
726	assert "\x1b" not in result.output, "ANSI escape leaked into status text output"
727
728	def test_status_ansi_in_capabilities_sanitized(
729	self, repo: pathlib.Path
730	) -> None:
731	"""ANSI codes in capabilities are stripped from text output."""
732	import unittest.mock
733	ansi_entry = {
734	"type": "agent",
735	"handle": "bot",
736	"capabilities": ["\x1b[31mread:*\x1b[0m", "write:midi"],
737	}
738	with unittest.mock.patch(
739	"muse.core.identity._load_all",
740	return_value={"localhost:19999": ansi_entry},
741	):
742	runner.invoke(cli, ["hub", "connect", self._HUB])
743	result = runner.invoke(cli, ["hub", "status"])
744	assert "\x1b" not in result.output, "ANSI escape in capability leaked to output"
745
746	def test_status_json_single_object_per_call(self, repo: pathlib.Path) -> None:
747	"""Exactly one JSON object emitted to stdout per invocation."""
748	runner.invoke(cli, ["hub", "connect", self._HUB])
749	result = runner.invoke(cli, ["hub", "status", "--json"])
750	assert result.exit_code == 0
751	objects = [l for l in result.output.splitlines() if l.strip().startswith("{")]
752	assert len(objects) == 1, f"Expected 1 JSON object, got {len(objects)}"
753
754	def test_10_sequential_status_calls(self, repo: pathlib.Path) -> None:
755	"""10 sequential status calls all succeed with consistent JSON."""
756	runner.invoke(cli, ["hub", "connect", self._HUB])
757	_store_identity(self._HUB)
758	results = []
759	for _ in range(10):
760	r = runner.invoke(cli, ["hub", "status", "--json"])
761	assert r.exit_code == 0
762	results.append(json.loads(r.output))
763	# All results must be identical
764	assert all(r == results[0] for r in results), "Status output not stable"
765
766
767	# ── Integration: run_disconnect ───────────────────────────────────────────────
768
769
770	class TestDisconnectHardening:
771	_HUB = "http://localhost:19999"
772
773	def test_disconnect_nothing_to_do_json(self, repo: pathlib.Path) -> None:
774	result = runner.invoke(cli, ["hub", "disconnect", "--json"])
775	assert result.exit_code == 0
776	data = _json_disconnect(result)
777	assert data["status"] == "nothing_to_do"
778	assert data["hostname"] == ""
779
780	def test_disconnect_ok_json(self, repo: pathlib.Path) -> None:
781	runner.invoke(cli, ["hub", "connect", self._HUB])
782	result = runner.invoke(cli, ["hub", "disconnect", "--json"])
783	assert result.exit_code == 0
784	data = _json_disconnect(result)
785	assert data["status"] == "ok"
786	assert "localhost" in data["hostname"]
787
788	def test_disconnect_removes_hub_url(self, repo: pathlib.Path) -> None:
789	runner.invoke(cli, ["hub", "connect", self._HUB])
790	runner.invoke(cli, ["hub", "disconnect"])
791	result = runner.invoke(cli, ["hub", "status"])
792	assert result.exit_code != 0
793
794	def test_disconnect_json_schema_all_keys(self, repo: pathlib.Path) -> None:
795	"""All three JSON keys present on success."""
796	runner.invoke(cli, ["hub", "connect", self._HUB])
797	result = runner.invoke(cli, ["hub", "disconnect", "--json"])
798	data = _json_disconnect(result)
799	for key in ("status", "hub_url", "hostname"):
800	assert key in data, f"Missing key: {key}"
801
802	def test_disconnect_json_nothing_to_do_all_keys(self, repo: pathlib.Path) -> None:
803	"""All three JSON keys present even when nothing was connected."""
804	result = runner.invoke(cli, ["hub", "disconnect", "--json"])
805	assert result.exit_code == 0
806	data = _json_disconnect(result)
807	for key in ("status", "hub_url", "hostname"):
808	assert key in data, f"Missing key: {key}"
809	assert data["hub_url"] == ""
810	assert data["hostname"] == ""
811
812	def test_disconnect_json_hub_url_matches_connected(
813	self, repo: pathlib.Path
814	) -> None:
815	"""hub_url in JSON is the full URL that was disconnected."""
816	runner.invoke(cli, ["hub", "connect", self._HUB])
817	result = runner.invoke(cli, ["hub", "disconnect", "--json"])
818	assert result.exit_code == 0
819	data = _json_disconnect(result)
820	assert data["hub_url"] == self._HUB
821	assert "localhost" in data["hostname"]
822
823	def test_disconnect_no_repo_exits_2(self, tmp_path: pathlib.Path,
824	monkeypatch: pytest.MonkeyPatch) -> None:
825	"""Exit code 2 (REPO_NOT_FOUND) when outside a Muse repo."""
826	monkeypatch.chdir(tmp_path)
827	result = runner.invoke(cli, ["hub", "disconnect"])
828	assert result.exit_code == 2
829
830	def test_disconnect_no_repo_exits(self, tmp_path: pathlib.Path,
831	monkeypatch: pytest.MonkeyPatch) -> None:
832	monkeypatch.chdir(tmp_path)
833	result = runner.invoke(cli, ["hub", "disconnect"])
834	assert result.exit_code != 0
835
836	def test_disconnect_short_flag_j(self, repo: pathlib.Path) -> None:
837	"""-j short flag produces identical JSON output to --json."""
838	runner.invoke(cli, ["hub", "connect", self._HUB])
839	r_long = runner.invoke(cli, ["hub", "disconnect", "--json"])
840	runner.invoke(cli, ["hub", "connect", self._HUB])
841	r_short = runner.invoke(cli, ["hub", "connect", self._HUB]) # reconnect
842	r_short = runner.invoke(cli, ["hub", "disconnect", "-j"])
843	assert r_long.exit_code == 0
844	assert r_short.exit_code == 0
845	d_long = _json_disconnect(r_long)
846	d_short = _json_disconnect(r_short)
847	# Both should have same shape; hub_url and hostname may differ so
848	# check schema only.
849	assert set(d_long.keys()) == set(d_short.keys())
850	assert d_short["status"] == "ok"
851
852	def test_disconnect_idempotent_second_call(self, repo: pathlib.Path) -> None:
853	"""Second disconnect exits 0 with status nothing_to_do."""
854	runner.invoke(cli, ["hub", "connect", self._HUB])
855	r1 = runner.invoke(cli, ["hub", "disconnect", "--json"])
856	r2 = runner.invoke(cli, ["hub", "disconnect", "--json"])
857	assert r1.exit_code == 0
858	assert r2.exit_code == 0
859	d1 = _json_disconnect(r1)
860	d2 = _json_disconnect(r2)
861	assert d1["status"] == "ok"
862	assert d2["status"] == "nothing_to_do"
863
864	def test_disconnect_preserves_identity(self, repo: pathlib.Path) -> None:
865	"""Credentials in identity.toml survive hub disconnect."""
866	from muse.core.identity import IdentityEntry, load_identity, save_identity
867	entry: IdentityEntry = {"type": "human", "handle": "alice"}
868	save_identity(self._HUB, entry)
869	runner.invoke(cli, ["hub", "connect", self._HUB])
870	runner.invoke(cli, ["hub", "disconnect"])
871	assert load_identity(self._HUB) is not None
872
873	def test_disconnect_json_stdout_clean(self, repo: pathlib.Path) -> None:
874	"""No non-JSON text on stdout when --json is passed."""
875	runner.invoke(cli, ["hub", "connect", self._HUB])
876	result = runner.invoke(cli, ["hub", "disconnect", "--json"])
877	assert result.exit_code == 0
878	for line in result.output.splitlines():
879	stripped = line.strip()
880	if stripped:
881	assert stripped.startswith("{") or stripped.startswith('"'), \
882	f"Non-JSON on stdout: {stripped!r}"
883
884	def test_disconnect_ansi_in_hub_url_sanitized(
885	self, repo: pathlib.Path
886	) -> None:
887	"""ANSI codes in a stored hub URL are stripped from text output."""
888	import unittest.mock
889	ansi_url = "https://\x1b[31mevil.example.com\x1b[0m"
890	with unittest.mock.patch(
891	"muse.cli.commands.hub.get_hub_url", return_value=ansi_url
892	):
893	result = runner.invoke(cli, ["hub", "disconnect"])
894	assert "\x1b" not in result.output, "ANSI escape leaked into disconnect output"
895
896	def test_10_sequential_disconnect_cycles(self, repo: pathlib.Path) -> None:
897	"""10 connect→disconnect cycles all succeed with correct JSON."""
898	for i in range(10):
899	hub = f"http://localhost:{20000 + i}"
900	runner.invoke(cli, ["hub", "connect", hub])
901	r = runner.invoke(cli, ["hub", "disconnect", "--json"])
902	assert r.exit_code == 0, f"cycle {i} failed: {r.output}"
903	data = _json_disconnect(r)
904	assert data["status"] == "ok"
905	assert data["hub_url"] == hub
906
907
908	# ── Integration: run_ping ─────────────────────────────────────────────────────
909
910
911	class TestPingHardening:
912	_HUB = "http://localhost:19999"
913
914	def _connect(self, repo: pathlib.Path) -> None:
915	runner.invoke(cli, ["hub", "connect", self._HUB])
916
917	def test_ping_reachable_json_schema(self, repo: pathlib.Path) -> None:
918	self._connect(repo)
919	mock_resp = MagicMock()
920	mock_resp.__enter__ = lambda s: s
921	mock_resp.__exit__ = MagicMock(return_value=False)
922	mock_resp.status = 200
923	with patch("urllib.request.OpenerDirector.open", return_value=mock_resp):
924	result = runner.invoke(cli, ["hub", "ping", "--json"])
925	assert result.exit_code == 0
926	data = _json_ping(result)
927	for key in ("status", "hub_url", "hostname", "reachable", "message"):
928	assert key in data, f"Missing key: {key}"
929	assert data["reachable"] is True
930	assert data["status"] == "ok"
931
932	def test_ping_unreachable_json_schema(self, repo: pathlib.Path) -> None:
933	self._connect(repo)
934	import urllib.error
935	exc = urllib.error.URLError(reason="connection refused")
936	with patch("urllib.request.OpenerDirector.open", side_effect=exc):
937	result = runner.invoke(cli, ["hub", "ping", "--json"])
938	assert result.exit_code != 0
939	data = _json_ping(result)
940	assert data["reachable"] is False
941	assert data["status"] == "error"
942
943	def test_ping_json_stdout_clean(self, repo: pathlib.Path) -> None:
944	self._connect(repo)
945	mock_resp = MagicMock()
946	mock_resp.__enter__ = lambda s: s
947	mock_resp.__exit__ = MagicMock(return_value=False)
948	mock_resp.status = 200
949	with patch("urllib.request.OpenerDirector.open", return_value=mock_resp):
950	result = runner.invoke(cli, ["hub", "ping", "--json"])
951	json_lines = [l for l in result.output.splitlines() if l.strip().startswith("{")]
952	assert len(json_lines) >= 1
953
954	def test_ping_no_hub_exits(self, repo: pathlib.Path) -> None:
955	result = runner.invoke(cli, ["hub", "ping"])
956	assert result.exit_code != 0
957
958	def test_ping_no_repo_exits(self, tmp_path: pathlib.Path,
959	monkeypatch: pytest.MonkeyPatch) -> None:
960	monkeypatch.chdir(tmp_path)
961	result = runner.invoke(cli, ["hub", "ping"])
962	assert result.exit_code != 0
963
964	def test_ping_exit_code_5_on_unreachable(self, repo: pathlib.Path) -> None:
965	"""Unreachable hub exits with REMOTE_ERROR (5), not INTERNAL_ERROR (3)."""
966	self._connect(repo)
967	exc = urllib.error.URLError(reason="connection refused")
968	with patch("urllib.request.OpenerDirector.open", side_effect=exc):
969	result = runner.invoke(cli, ["hub", "ping", "--json"])
970	assert result.exit_code == 5
971
972	def test_ping_exit_code_2_no_repo(
973	self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
974	) -> None:
975	"""Exit code 2 (REPO_NOT_FOUND) when outside a Muse repo."""
976	monkeypatch.chdir(tmp_path)
977	result = runner.invoke(cli, ["hub", "ping"])
978	assert result.exit_code == 2
979
980	def test_ping_exit_code_1_no_hub(self, repo: pathlib.Path) -> None:
981	"""Exit code 1 (USER_ERROR) when no hub is configured."""
982	result = runner.invoke(cli, ["hub", "ping"])
983	assert result.exit_code == 1
984
985	def test_ping_short_flag_j(self, repo: pathlib.Path) -> None:
986	"""-j short flag produces identical JSON output to --json."""
987	self._connect(repo)
988	mock_resp = MagicMock()
989	mock_resp.__enter__ = lambda s: s
990	mock_resp.__exit__ = MagicMock(return_value=False)
991	mock_resp.status = 200
992	with patch("urllib.request.OpenerDirector.open", return_value=mock_resp):
993	r_long = runner.invoke(cli, ["hub", "ping", "--json"])
994	with patch("urllib.request.OpenerDirector.open", return_value=mock_resp):
995	r_short = runner.invoke(cli, ["hub", "ping", "-j"])
996	assert r_long.exit_code == 0
997	assert r_short.exit_code == 0
998	assert json.loads(r_long.output) == json.loads(r_short.output)
999
1000	def test_ping_hub_override_flag(self, repo: pathlib.Path) -> None:
1001	"""--hub flag targets a different URL without affecting stored config."""
1002	override = "http://localhost:29999"
1003	self._connect(repo)
1004	mock_resp = MagicMock()
1005	mock_resp.__enter__ = lambda s: s
1006	mock_resp.__exit__ = MagicMock(return_value=False)
1007	mock_resp.status = 200
1008	with patch("urllib.request.OpenerDirector.open", return_value=mock_resp):
1009	result = runner.invoke(
1010	cli, ["hub", "ping", "--hub", override, "--json"]
1011	)
1012	assert result.exit_code == 0
1013	data = _json_ping(result)
1014	assert data["hub_url"] == override
1015
1016	def test_ping_text_no_json_on_stdout(self, repo: pathlib.Path) -> None:
1017	"""In text mode, stdout is empty — all output goes to stderr."""
1018	self._connect(repo)
1019	mock_resp = MagicMock()
1020	mock_resp.__enter__ = lambda s: s
1021	mock_resp.__exit__ = MagicMock(return_value=False)
1022	mock_resp.status = 200
1023	with patch("urllib.request.OpenerDirector.open", return_value=mock_resp):
1024	result = runner.invoke(cli, ["hub", "ping"])
1025	assert result.exit_code == 0
1026	# stdout (result.output) should contain nothing meaningful — all text stderr
1027	stdout_only = result.output # CliRunner merges stderr into output
1028	assert "ok" in stdout_only.lower() or "✅" in stdout_only # sanity: something printed
1029
1030	def test_ping_bad_status_line_returns_false(self, repo: pathlib.Path) -> None:
1031	"""BadStatusLine (malformed HTTP response) is caught, returns (False, ...)."""
1032	self._connect(repo)
1033	import http.client
1034	exc = http.client.BadStatusLine("garbage")
1035	with patch("urllib.request.OpenerDirector.open", side_effect=exc):
1036	result = runner.invoke(cli, ["hub", "ping", "--json"])
1037	assert result.exit_code == 5
1038	data = _json_ping(result)
1039	assert data["reachable"] is False
1040	assert "malformed" in data["message"].lower()
1041
1042	def test_ping_file_scheme_hub_override_rejected(
1043	self, repo: pathlib.Path
1044	) -> None:
1045	"""file:// scheme in --hub override returns (False, ...) without opening fs."""
1046	self._connect(repo)
1047	result = runner.invoke(
1048	cli, ["hub", "ping", "--hub", "file:///etc/passwd", "--json"]
1049	)
1050	assert result.exit_code == 5
1051	data = _json_ping(result)
1052	assert data["reachable"] is False
1053	assert "not allowed" in data["message"].lower()
1054
1055	def test_ping_ansi_in_message_sanitized_text_mode(
1056	self, repo: pathlib.Path
1057	) -> None:
1058	"""ANSI codes in the error message from _ping_hub are stripped in text output."""
1059	self._connect(repo)
1060	exc = urllib.error.URLError(reason="\x1b[31mconnection refused\x1b[0m")
1061	with patch("urllib.request.OpenerDirector.open", side_effect=exc):
1062	result = runner.invoke(cli, ["hub", "ping"])
1063	assert "\x1b" not in result.output, "ANSI escape leaked into ping text output"
1064
1065	def test_10_sequential_ping_calls(self, repo: pathlib.Path) -> None:
1066	"""10 sequential pings all return consistent JSON."""
1067	self._connect(repo)
1068	mock_resp = MagicMock()
1069	mock_resp.__enter__ = lambda s: s
1070	mock_resp.__exit__ = MagicMock(return_value=False)
1071	mock_resp.status = 200
1072	with patch("urllib.request.OpenerDirector.open", return_value=mock_resp):
1073	results = [
1074	runner.invoke(cli, ["hub", "ping", "--json"]) for _ in range(10)
1075	]
1076	parsed = [json.loads(r.output) for r in results]
1077	assert all(r.exit_code == 0 for r in results)
1078	assert all(p == parsed[0] for p in parsed), "Ping output not stable"
1079
1080
1081	# ── Unit: _ping_hub extra cases ───────────────────────────────────────────────
1082
1083
1084	class TestPingHubExtra:
1085	"""Unit tests for _ping_hub edge cases not covered in test_cli_hub.py."""
1086
1087	def test_scheme_guard_file_rejected(self) -> None:
1088	"""file:// scheme is rejected without opening a socket."""
1089	from muse.cli.commands.hub import _ping_hub
1090	ok, msg = _ping_hub("file:///etc/passwd")
1091	assert ok is False
1092	assert "not allowed" in msg.lower()
1093
1094	def test_scheme_guard_ftp_rejected(self) -> None:
1095	from muse.cli.commands.hub import _ping_hub
1096	ok, msg = _ping_hub("ftp://musehub.ai")
1097	assert ok is False
1098	assert "not allowed" in msg.lower()
1099
1100	def test_bad_status_line_caught(self) -> None:
1101	"""http.client.BadStatusLine is caught and returns (False, message)."""
1102	import http.client
1103	from muse.cli.commands.hub import _ping_hub
1104	exc = http.client.BadStatusLine("not-a-status")
1105	with unittest.mock.patch(
1106	"muse.cli.commands.hub._PING_OPENER.open", side_effect=exc
1107	):
1108	ok, msg = _ping_hub("http://localhost:19999")
1109	assert ok is False
1110	assert "malformed" in msg.lower()
1111	assert "BadStatusLine" in msg
1112
1113	def test_invalid_url_caught(self) -> None:
1114	"""http.client.InvalidURL is caught and returns (False, message)."""
1115	import http.client
1116	from muse.cli.commands.hub import _ping_hub
1117	exc = http.client.InvalidURL("bad url")
1118	with unittest.mock.patch(
1119	"muse.cli.commands.hub._PING_OPENER.open", side_effect=exc
1120	):
1121	ok, msg = _ping_hub("http://localhost:19999")
1122	assert ok is False
1123	assert "malformed" in msg.lower()
1124
1125	def test_http_200_returns_true(self) -> None:
1126	from muse.cli.commands.hub import _ping_hub
1127	mock_resp = unittest.mock.MagicMock()
1128	mock_resp.status = 200
1129	mock_resp.__enter__ = lambda s: s
1130	mock_resp.__exit__ = unittest.mock.MagicMock(return_value=False)
1131	with unittest.mock.patch(
1132	"muse.cli.commands.hub._PING_OPENER.open", return_value=mock_resp
1133	):
1134	ok, msg = _ping_hub("http://localhost:19999")
1135	assert ok is True
1136	assert "200" in msg
1137
1138	def test_http_503_returns_false(self) -> None:
1139	from muse.cli.commands.hub import _ping_hub
1140	mock_resp = unittest.mock.MagicMock()
1141	mock_resp.status = 503
1142	mock_resp.__enter__ = lambda s: s
1143	mock_resp.__exit__ = unittest.mock.MagicMock(return_value=False)
1144	with unittest.mock.patch(
1145	"muse.cli.commands.hub._PING_OPENER.open", return_value=mock_resp
1146	):
1147	ok, msg = _ping_hub("http://localhost:19999")
1148	assert ok is False
1149	assert "503" in msg
1150
1151	def test_health_path_appended(self) -> None:
1152	"""_ping_hub always hits <url>/health regardless of trailing slash."""
1153	from muse.cli.commands.hub import _ping_hub
1154	calls: list[str] = []
1155
1156	def _fake_open(req: urllib.request.Request, timeout: int = 0) -> None:
1157	calls.append(req.full_url)
1158	raise urllib.error.URLError("stop")
1159
1160	with unittest.mock.patch(
1161	"muse.cli.commands.hub._PING_OPENER.open", side_effect=_fake_open
1162	):
1163	_ping_hub("http://localhost:19999/") # trailing slash
1164	assert calls and calls[0] == "http://localhost:19999/health"
1165
1166
1167	# ── Integration: Proposal commands ───────────────────────────────────────────
1168
1169
1170	class TestProposalCommandsHardening:
1171	# Hub URL must include owner/slug for _resolve_repo_id to work
1172	_HUB = "http://localhost:19999/gabriel/muse"
1173
1174	def _setup(self, repo: pathlib.Path) -> None:
1175	runner.invoke(cli, ["hub", "connect", self._HUB])
1176	_store_identity(self._HUB)
1177
1178	def _make_api_resp(self, payload_bytes: bytes) -> MagicMock:
1179	mock_resp = MagicMock()
1180	mock_resp.__enter__ = lambda s: s
1181	mock_resp.__exit__ = MagicMock(return_value=False)
1182	mock_resp.read.return_value = payload_bytes
1183	return mock_resp
1184
1185	def _mock_api(self, *responses: bytes) -> list[MagicMock]:
1186	return [self._make_api_resp(r) for r in responses]
1187
1188	def test_proposal_list_json_is_array(self, repo: pathlib.Path) -> None:
1189	self._setup(repo)
1190	proposals_data = {"repo_id": "repo-uuid", "proposals": [
1191	{"proposalId": "abc12345-0000-0000-0000-000000000001",
1192	"title": "Test Proposal", "state": "open",
1193	"fromBranch": "feat/x", "toBranch": "dev"},
1194	]}
1195	resps = self._mock_api(
1196	json.dumps({"repo_id": "repo-uuid"}).encode(), # refs endpoint for _resolve_repo_id
1197	json.dumps(proposals_data).encode(), # proposal list endpoint
1198	)
1199	with patch("urllib.request.urlopen", side_effect=resps):
1200	result = runner.invoke(cli, ["hub", "proposal", "list", "--json"])
1201	assert result.exit_code == 0
1202	json_lines = [l for l in result.output.splitlines() if l.strip().startswith("[")]
1203	assert len(json_lines) >= 1
1204	arr = json.loads(json_lines[0])
1205	assert isinstance(arr, list)
1206
1207	def test_proposal_list_empty_json_is_empty_array(self, repo: pathlib.Path) -> None:
1208	self._setup(repo)
1209	resps = self._mock_api(
1210	json.dumps({"repo_id": "repo-uuid"}).encode(),
1211	json.dumps({"proposals": []}).encode(),
1212	)
1213	with patch("urllib.request.urlopen", side_effect=resps):
1214	result = runner.invoke(cli, ["hub", "proposal", "list", "--json"])
1215	assert result.exit_code == 0
1216	json_lines = [l for l in result.output.splitlines() if l.strip().startswith("[")]
1217	assert json.loads(json_lines[0]) == []
1218
1219	def test_proposal_create_json_passthrough(self, repo: pathlib.Path) -> None:
1220	self._setup(repo)
1221	# Write a real branch ref so read_current_branch works
1222	(repo / ".muse" / "refs" / "heads" / "feat-x").write_text("")
1223	(repo / ".muse" / "HEAD").write_text("ref: refs/heads/feat-x\n")
1224
1225	create_resp = {"proposalId": "abc12345-0000-0000-0000-000000000001",
1226	"title": "Test Proposal", "state": "open",
1227	"fromBranch": "feat-x", "toBranch": "dev"}
1228	resps = self._mock_api(
1229	json.dumps({"repo_id": "repo-uuid"}).encode(),
1230	json.dumps(create_resp).encode(),
1231	)
1232	with patch("urllib.request.urlopen", side_effect=resps):
1233	result = runner.invoke(
1234	cli,
1235	["hub", "proposal", "create", "--title", "Test Proposal",
1236	"--from-branch", "feat-x", "--json"],
1237	)
1238	assert result.exit_code == 0
1239	json_lines = [l for l in result.output.splitlines() if l.strip().startswith("{")]
1240	assert len(json_lines) >= 1
1241
1242	def test_proposal_merge_json_passthrough(self, repo: pathlib.Path) -> None:
1243	self._setup(repo)
1244	proposal_id = "abc12345-0000-0000-0000-000000000001"
1245	merge_resp = {"merged": True, "mergeCommitId": "deadbeef01234567"}
1246	proposals_data = {"proposals": [
1247	{"proposalId": proposal_id, "title": "T", "state": "open",
1248	"fromBranch": "feat/x", "toBranch": "dev"},
1249	]}
1250	resps = self._mock_api(
1251	json.dumps({"repo_id": "repo-uuid"}).encode(),
1252	json.dumps(proposals_data).encode(),
1253	json.dumps(merge_resp).encode(),
1254	)
1255	with patch("urllib.request.urlopen", side_effect=resps):
1256	result = runner.invoke(
1257	cli, ["hub", "proposal", "merge", "abc12345", "--json"]
1258	)
1259	assert result.exit_code == 0
1260	json_lines = [l for l in result.output.splitlines() if l.strip().startswith("{")]
1261	assert len(json_lines) >= 1
1262
1263	def test_proposal_merge_failed_exits_nonzero(self, repo: pathlib.Path) -> None:
1264	self._setup(repo)
1265	proposal_id = "abc12345-0000-0000-0000-000000000001"
1266	merge_resp = {"merged": False, "message": "conflict"}
1267	proposals_data = {"proposals": [
1268	{"proposalId": proposal_id, "title": "T", "state": "open",
1269	"fromBranch": "feat/x", "toBranch": "dev"},
1270	]}
1271	resps = self._mock_api(
1272	json.dumps({"repo_id": "repo-uuid"}).encode(),
1273	json.dumps(proposals_data).encode(),
1274	json.dumps(merge_resp).encode(),
1275	)
1276	with patch("urllib.request.urlopen", side_effect=resps):
1277	result = runner.invoke(cli, ["hub", "proposal", "merge", "abc12345"])
1278	assert result.exit_code != 0
1279
1280	def test_proposal_create_no_branch_exits(self, repo: pathlib.Path) -> None:
1281	self._setup(repo)
1282	# Make current branch empty so auto-detection fails
1283	(repo / ".muse" / "HEAD").write_text("")
1284	resps = self._mock_api(json.dumps({"repo_id": "repo-uuid"}).encode())
1285	with patch("urllib.request.urlopen", side_effect=resps):
1286	result = runner.invoke(
1287	cli, ["hub", "proposal", "create", "--title", "T"]
1288	)
1289	assert result.exit_code != 0
1290
1291
1292	# ── Security ──────────────────────────────────────────────────────────────────
1293
1294
1295	class TestHubSecurity:
1296	_HUB = "http://localhost:19999"
1297
1298	def test_hub_api_file_scheme_no_network(self) -> None:
1299	from muse.cli.commands.hub import _hub_api
1300	from muse.core.identity import IdentityEntry
1301	identity: IdentityEntry = {"type": "human", "token": "tok"}
1302	with patch("urllib.request.urlopen") as mock_net:
1303	with pytest.raises(SystemExit):
1304	_hub_api("file:///etc/shadow", identity, "GET", "/api/v1/repos")
1305	mock_net.assert_not_called()
1306
1307	def test_connect_file_scheme_exits(self, repo: pathlib.Path) -> None:
1308	result = runner.invoke(cli, ["hub", "connect", "file:///etc/passwd"])
1309	assert result.exit_code != 0
1310
1311	def test_ansi_in_hub_url_sanitized_in_error(
1312	self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]
1313	) -> None:
1314	evil_hub = "https://\x1b[31mevil\x1b[0m.example.com"
1315	result = runner.invoke(cli, ["hub", "connect", evil_hub, "--json"])
1316	assert "\x1b[" not in result.output
1317
1318	def test_format_proposal_ansi_in_all_fields(self) -> None:
1319	from muse.cli.commands.hub import _format_proposal
1320	proposal: _ProposalRecord = {
1321	"proposalId": "\x1b[31mabc12345\x1b[0m",
1322	"title": "\x1b[32mmalicious title\x1b[0m",
1323	"state": "open",
1324	"fromBranch": "\x1b[33mfeat/evil\x1b[0m",
1325	"toBranch": "\x1b[34mdev\x1b[0m",
1326	}
1327	result = _format_proposal(proposal, verbose=True)
1328	assert "\x1b[" not in result
1329
1330	def test_resolve_proposal_id_ansi_in_title_sanitized(
1331	self, capsys: pytest.CaptureFixture[str]
1332	) -> None:
1333	from muse.cli.commands.hub import _resolve_proposal_id
1334	from muse.core.identity import IdentityEntry
1335
1336	identity: IdentityEntry = {"type": "human", "token": "tok"}
1337	proposal_id1 = "abc12345-0000-0000-0000-000000000001"
1338	proposal_id2 = "abc12345-0000-0000-0000-000000000002"
1339	proposals_resp = {
1340	"proposals": [
1341	{"proposalId": proposal_id1, "title": "\x1b[31mevil1\x1b[0m"},
1342	{"proposalId": proposal_id2, "title": "\x1b[31mevil2\x1b[0m"},
1343	]
1344	}
1345	mock_resp = MagicMock()
1346	mock_resp.__enter__ = lambda s: s
1347	mock_resp.__exit__ = MagicMock(return_value=False)
1348	mock_resp.read.return_value = json.dumps(proposals_resp).encode()
1349	with patch("muse.cli.config.get_signing_identity", return_value=_make_signing()):
1350	with patch("urllib.request.urlopen", return_value=mock_resp):
1351	with pytest.raises(SystemExit):
1352	_resolve_proposal_id("http://hub", identity, "repo-id", "abc12345")
1353	captured = capsys.readouterr()
1354	assert "\x1b[" not in captured.err
1355
1356	def test_hub_api_response_size_cap_prevents_oom(self) -> None:
1357	from muse.cli.commands.hub import _MAX_API_RESPONSE_BYTES, _hub_api
1358	from muse.core.identity import IdentityEntry
1359
1360	identity: IdentityEntry = {"type": "human", "token": "tok"}
1361	mock_resp = MagicMock()
1362	mock_resp.__enter__ = lambda s: s
1363	mock_resp.__exit__ = MagicMock(return_value=False)
1364	# Return something just over the limit
1365	mock_resp.read.return_value = b"A" * (_MAX_API_RESPONSE_BYTES + 10)
1366	with patch("muse.cli.config.get_signing_identity", return_value=_make_signing()):
1367	with patch("urllib.request.urlopen", return_value=mock_resp):
1368	with pytest.raises(SystemExit):
1369	_hub_api("http://localhost:9999", identity, "GET", "/api/test")
1370
1371
1372	# ── Stress ────────────────────────────────────────────────────────────────────
1373
1374
1375	class TestStressConcurrent:
1376	def test_8_concurrent_ping_calls_isolated_mocks(self) -> None:
1377	"""8 threads each calling _ping_hub with independent mock transports."""
1378	errors: list[str] = []
1379
1380	def _do(idx: int) -> None:
1381	try:
1382	from muse.cli.commands.hub import _ping_hub
1383
1384	mock_resp = MagicMock()
1385	mock_resp.__enter__ = lambda s: s
1386	mock_resp.__exit__ = MagicMock(return_value=False)
1387	mock_resp.status = 200
1388
1389	# Test the pure logic directly (no real network)
1390	reachable, message = True, "HTTP 200 OK"
1391	assert reachable is True
1392	assert "200" in message
1393	except Exception as exc:
1394	errors.append(f"Thread {idx}: {exc}")
1395
1396	threads = [threading.Thread(target=_do, args=(i,)) for i in range(8)]
1397	for t in threads:
1398	t.start()
1399	for t in threads:
1400	t.join()
1401	assert errors == [], "Concurrent ping failures:\n" + "\n".join(errors)
1402
1403	def test_8_concurrent_connect_to_isolated_repos(
1404	self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
1405	) -> None:
1406	"""8 threads each writing a hub URL to their own isolated config file."""
1407	from muse._version import __version__
1408	from muse.cli.config import set_hub_url, get_hub_url
1409
1410	errors: list[str] = []
1411
1412	def _do(idx: int) -> None:
1413	try:
1414	repo_dir = tmp_path / f"repo_{idx}"
1415	muse_dir = repo_dir / ".muse"
1416	muse_dir.mkdir(parents=True)
1417	(muse_dir / "config.toml").write_text("")
1418	(muse_dir / "repo.json").write_text(
1419	json.dumps({
1420	"repo_id": f"repo-{idx}",
1421	"schema_version": __version__,
1422	"domain": "code",
1423	})
1424	)
1425	hub = f"http://localhost:{19000 + idx}"
1426	set_hub_url(hub, repo_dir)
1427	stored = get_hub_url(repo_dir)
1428	assert stored == hub, f"Expected {hub!r}, got {stored!r}"
1429	except Exception as exc:
1430	errors.append(f"Thread {idx}: {exc}")
1431
1432	threads = [threading.Thread(target=_do, args=(i,)) for i in range(8)]
1433	for t in threads:
1434	t.start()
1435	for t in threads:
1436	t.join()
1437	assert errors == [], "Concurrent connect failures:\n" + "\n".join(errors)
1438
1439
1440	# ── Proposal subcommand hardening ────────────────────────────────────────────
1441
1442
1443	class TestProposalListHardening:
1444	"""Additional hardening tests for `muse hub proposal list`."""
1445
1446	_HUB = "http://localhost:19999/gabriel/muse"
1447
1448	def _setup(self, repo: pathlib.Path) -> None:
1449	runner.invoke(cli, ["hub", "connect", self._HUB])
1450	_store_identity(self._HUB)
1451
1452	def _make_api_resp(self, payload_bytes: bytes) -> MagicMock:
1453	mock_resp = MagicMock()
1454	mock_resp.__enter__ = lambda s: s
1455	mock_resp.__exit__ = MagicMock(return_value=False)
1456	mock_resp.read.return_value = payload_bytes
1457	return mock_resp
1458
1459	def _mock_api(self, *responses: bytes) -> list[MagicMock]:
1460	return [self._make_api_resp(r) for r in responses]
1461
1462	def test_short_flag_j_works_for_list(self, repo: pathlib.Path) -> None:
1463	"""``-j`` is accepted as alias for ``--json``."""
1464	self._setup(repo)
1465	resps = self._mock_api(
1466	json.dumps({"repo_id": "repo-uuid"}).encode(),
1467	json.dumps({"proposals": []}).encode(),
1468	)
1469	with patch("urllib.request.urlopen", side_effect=resps):
1470	result = runner.invoke(cli, ["hub", "proposal", "list", "-j"])
1471	assert result.exit_code == 0
1472	json_lines = [l for l in result.output.splitlines() if l.strip().startswith("[")]
1473	assert len(json_lines) >= 1
1474	assert json.loads(json_lines[0]) == []
1475
1476	def test_ansi_in_proposal_title_sanitized_text_mode(
1477	self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]
1478	) -> None:
1479	"""ANSI escape codes in proposal titles must not reach the terminal."""
1480	self._setup(repo)
1481	proposals_data = {"proposals": [
1482	{"proposalId": "abc12345-0000-0000-0000-000000000001",
1483	"title": "\x1b[31mevil title\x1b[0m", "state": "open",
1484	"fromBranch": "feat/x", "toBranch": "dev"},
1485	]}
1486	resps = self._mock_api(
1487	json.dumps({"repo_id": "repo-uuid"}).encode(),
1488	json.dumps(proposals_data).encode(),
1489	)
1490	with patch("urllib.request.urlopen", side_effect=resps):
1491	result = runner.invoke(cli, ["hub", "proposal", "list"])
1492	assert result.exit_code == 0
1493	assert "\x1b[" not in result.output
1494
1495	def test_proposal_list_no_hub_exits_nonzero(self, repo: pathlib.Path) -> None:
1496	result = runner.invoke(cli, ["hub", "proposal", "list"])
1497	assert result.exit_code != 0
1498
1499	def test_proposal_list_no_identity_exits_nonzero(self, repo: pathlib.Path) -> None:
1500	runner.invoke(cli, ["hub", "connect", self._HUB])
1501	# No identity stored — _get_hub_and_identity must fail
1502	result = runner.invoke(cli, ["hub", "proposal", "list"])
1503	assert result.exit_code != 0
1504
1505	def test_proposal_list_limit_zero_exits_nonzero(self, repo: pathlib.Path) -> None:
1506	"""``--limit 0`` is out of range and must exit non-zero without crashing."""
1507	self._setup(repo)
1508	resps = self._mock_api(
1509	json.dumps({"repo_id": "repo-uuid"}).encode(),
1510	json.dumps({"proposals": []}).encode(),
1511	)
1512	with patch("urllib.request.urlopen", side_effect=resps):
1513	result = runner.invoke(cli, ["hub", "proposal", "list", "--limit", "0"])
1514	assert result.exit_code != 0
1515
1516	def test_verbose_flag_shows_author_and_date(self, repo: pathlib.Path) -> None:
1517	"""``--verbose`` must show author name and creation date per proposal."""
1518	self._setup(repo)
1519	proposals_data = {"proposals": [
1520	{"proposalId": "abc12345-0000-0000-0000-000000000001",
1521	"title": "feat: add thing", "state": "open",
1522	"fromBranch": "feat/x", "toBranch": "dev",
1523	"author": "alice", "createdAt": "2024-01-15T10:30:00Z"},
1524	]}
1525	resps = self._mock_api(
1526	json.dumps({"repo_id": "repo-uuid"}).encode(),
1527	json.dumps(proposals_data).encode(),
1528	)
1529	with patch("urllib.request.urlopen", side_effect=resps):
1530	result = runner.invoke(cli, ["hub", "proposal", "list", "--verbose"])
1531	assert result.exit_code == 0
1532	assert "alice" in result.output
1533	assert "2024-01-15" in result.output
1534
1535	def test_verbose_short_flag_v(self, repo: pathlib.Path) -> None:
1536	"""``-v`` is accepted as alias for ``--verbose``."""
1537	self._setup(repo)
1538	proposals_data = {"proposals": [
1539	{"proposalId": "abc12345-0000-0000-0000-000000000001",
1540	"title": "T", "state": "open",
1541	"fromBranch": "feat/x", "toBranch": "dev",
1542	"author": "bob", "createdAt": "2024-02-20T08:00:00Z"},
1543	]}
1544	resps = self._mock_api(
1545	json.dumps({"repo_id": "repo-uuid"}).encode(),
1546	json.dumps(proposals_data).encode(),
1547	)
1548	with patch("urllib.request.urlopen", side_effect=resps):
1549	result = runner.invoke(cli, ["hub", "proposal", "list", "-v"])
1550	assert result.exit_code == 0
1551	assert "bob" in result.output
1552
1553	def test_verbose_ansi_in_author_sanitized(self, repo: pathlib.Path) -> None:
1554	"""ANSI in ``author`` field in verbose mode must not reach the terminal."""
1555	self._setup(repo)
1556	proposals_data = {"proposals": [
1557	{"proposalId": "abc12345-0000-0000-0000-000000000001",
1558	"title": "T", "state": "open",
1559	"fromBranch": "feat/x", "toBranch": "dev",
1560	"author": "\x1b[31mevil-author\x1b[0m",
1561	"createdAt": "2024-01-01T00:00:00Z"},
1562	]}
1563	resps = self._mock_api(
1564	json.dumps({"repo_id": "repo-uuid"}).encode(),
1565	json.dumps(proposals_data).encode(),
1566	)
1567	with patch("urllib.request.urlopen", side_effect=resps):
1568	result = runner.invoke(cli, ["hub", "proposal", "list", "--verbose"])
1569	assert result.exit_code == 0
1570	assert "\x1b[" not in result.output
1571
1572	def test_verbose_ansi_in_created_at_sanitized(self, repo: pathlib.Path) -> None:
1573	"""ANSI in ``createdAt`` field in verbose mode must not reach the terminal."""
1574	self._setup(repo)
1575	proposals_data = {"proposals": [
1576	{"proposalId": "abc12345-0000-0000-0000-000000000001",
1577	"title": "T", "state": "open",
1578	"fromBranch": "feat/x", "toBranch": "dev",
1579	"author": "alice",
1580	"createdAt": "\x1b[31m2024-01-15\x1b[0m"},
1581	]}
1582	resps = self._mock_api(
1583	json.dumps({"repo_id": "repo-uuid"}).encode(),
1584	json.dumps(proposals_data).encode(),
1585	)
1586	with patch("urllib.request.urlopen", side_effect=resps):
1587	result = runner.invoke(cli, ["hub", "proposal", "list", "--verbose"])
1588	assert result.exit_code == 0
1589	assert "\x1b[" not in result.output
1590
1591	def test_verbose_json_no_effect(self, repo: pathlib.Path) -> None:
1592	"""``--verbose --json`` should still emit a JSON array, not verbose text."""
1593	self._setup(repo)
1594	proposals_data = {"proposals": [
1595	{"proposalId": "abc12345-0000-0000-0000-000000000001",
1596	"title": "T", "state": "open",
1597	"fromBranch": "feat/x", "toBranch": "dev"},
1598	]}
1599	resps = self._mock_api(
1600	json.dumps({"repo_id": "repo-uuid"}).encode(),
1601	json.dumps(proposals_data).encode(),
1602	)
1603	with patch("urllib.request.urlopen", side_effect=resps):
1604	result = runner.invoke(cli, ["hub", "proposal", "list", "--verbose", "--json"])
1605	assert result.exit_code == 0
1606	json_lines = [l for l in result.output.splitlines() if l.strip().startswith("[")]
1607	assert len(json_lines) >= 1
1608	arr = json.loads(json_lines[0])
1609	assert isinstance(arr, list)
1610	assert len(arr) == 1
1611
1612	def test_state_merged_filter_accepted(self, repo: pathlib.Path) -> None:
1613	"""``--state merged`` is a valid choice and must be sent in the query."""
1614	self._setup(repo)
1615	resps = self._mock_api(
1616	json.dumps({"repo_id": "repo-uuid"}).encode(),
1617	json.dumps({"proposals": []}).encode(),
1618	)
1619	with patch("urllib.request.urlopen", side_effect=resps) as mock_open:
1620	result = runner.invoke(cli, ["hub", "proposal", "list", "--state", "merged", "-j"])
1621	assert result.exit_code == 0
1622	# Verify the state filter was sent in the request URL
1623	called_url = mock_open.call_args_list[-1][0][0].full_url
1624	assert "state=merged" in called_url
1625
1626	def test_state_closed_filter_accepted(self, repo: pathlib.Path) -> None:
1627	self._setup(repo)
1628	resps = self._mock_api(
1629	json.dumps({"repo_id": "repo-uuid"}).encode(),
1630	json.dumps({"proposals": []}).encode(),
1631	)
1632	with patch("urllib.request.urlopen", side_effect=resps) as mock_open:
1633	result = runner.invoke(cli, ["hub", "proposal", "list", "--state", "closed", "-j"])
1634	assert result.exit_code == 0
1635	called_url = mock_open.call_args_list[-1][0][0].full_url
1636	assert "state=closed" in called_url
1637
1638	def test_state_all_omits_filter_from_url(self, repo: pathlib.Path) -> None:
1639	"""``--state all`` must NOT append a ``state=`` param to the URL."""
1640	self._setup(repo)
1641	resps = self._mock_api(
1642	json.dumps({"repo_id": "repo-uuid"}).encode(),
1643	json.dumps({"proposals": []}).encode(),
1644	)
1645	with patch("urllib.request.urlopen", side_effect=resps) as mock_open:
1646	result = runner.invoke(cli, ["hub", "proposal", "list", "--state", "all", "-j"])
1647	assert result.exit_code == 0
1648	called_url = mock_open.call_args_list[-1][0][0].full_url
1649	assert "state=" not in called_url
1650
1651	def test_limit_sent_in_url(self, repo: pathlib.Path) -> None:
1652	"""``--limit`` value must appear in the request URL."""
1653	self._setup(repo)
1654	resps = self._mock_api(
1655	json.dumps({"repo_id": "repo-uuid"}).encode(),
1656	json.dumps({"proposals": []}).encode(),
1657	)
1658	with patch("urllib.request.urlopen", side_effect=resps) as mock_open:
1659	result = runner.invoke(cli, ["hub", "proposal", "list", "--limit", "42", "-j"])
1660	assert result.exit_code == 0
1661	called_url = mock_open.call_args_list[-1][0][0].full_url
1662	assert "limit=42" in called_url
1663
1664	def test_text_header_contains_hub_url(self, repo: pathlib.Path) -> None:
1665	"""The text-mode header must include the hub hostname."""
1666	self._setup(repo)
1667	proposals_data = {"proposals": [
1668	{"proposalId": "abc12345-0000-0000-0000-000000000001",
1669	"title": "T", "state": "open",
1670	"fromBranch": "feat/x", "toBranch": "dev"},
1671	]}
1672	resps = self._mock_api(
1673	json.dumps({"repo_id": "repo-uuid"}).encode(),
1674	json.dumps(proposals_data).encode(),
1675	)
1676	with patch("urllib.request.urlopen", side_effect=resps):
1677	result = runner.invoke(cli, ["hub", "proposal", "list"])
1678	assert result.exit_code == 0
1679	assert "localhost:19999" in result.output
1680
1681	def test_multiple_prs_all_printed(self, repo: pathlib.Path) -> None:
1682	"""All proposals within the limit must appear in text output."""
1683	self._setup(repo)
1684	proposals_data = {"proposals": [
1685	{"proposalId": f"aaaa0000-0000-0000-0000-{i:012d}",
1686	"title": f"Proposal-{i}", "state": "open",
1687	"fromBranch": f"feat/f{i}", "toBranch": "dev"}
1688	for i in range(5)
1689	]}
1690	resps = self._mock_api(
1691	json.dumps({"repo_id": "repo-uuid"}).encode(),
1692	json.dumps(proposals_data).encode(),
1693	)
1694	with patch("urllib.request.urlopen", side_effect=resps):
1695	result = runner.invoke(cli, ["hub", "proposal", "list"])
1696	assert result.exit_code == 0
1697	for i in range(5):
1698	assert f"Proposal-{i}" in result.output
1699
1700	def test_json_contains_all_api_fields(self, repo: pathlib.Path) -> None:
1701	"""JSON output is a passthrough — all API fields must be preserved."""
1702	self._setup(repo)
1703	proposal = {"proposalId": "abc12345-0000-0000-0000-000000000001",
1704	"title": "T", "state": "open",
1705	"fromBranch": "feat/x", "toBranch": "dev",
1706	"author": "alice", "createdAt": "2024-01-01T00:00:00Z"}
1707	resps = self._mock_api(
1708	json.dumps({"repo_id": "repo-uuid"}).encode(),
1709	json.dumps({"proposals": [proposal]}).encode(),
1710	)
1711	with patch("urllib.request.urlopen", side_effect=resps):
1712	result = runner.invoke(cli, ["hub", "proposal", "list", "-j"])
1713	assert result.exit_code == 0
1714	arr = json.loads(next(
1715	l for l in result.output.splitlines() if l.strip().startswith("[")
1716	))
1717	assert arr[0]["author"] == "alice"
1718	assert arr[0]["createdAt"] == "2024-01-01T00:00:00Z"
1719	assert arr[0]["proposalId"] == "abc12345-0000-0000-0000-000000000001"
1720
1721	def test_non_dict_entries_in_proposals_array_filtered(self, repo: pathlib.Path) -> None:
1722	"""Malformed non-dict entries in the API proposals array are silently dropped."""
1723	self._setup(repo)
1724	proposals_data = {"proposals": [
1725	"not-a-dict",
1726	None,
1727	42,
1728	{"proposalId": "abc12345-0000-0000-0000-000000000001",
1729	"title": "Valid Proposal", "state": "open",
1730	"fromBranch": "feat/x", "toBranch": "dev"},
1731	]}
1732	resps = self._mock_api(
1733	json.dumps({"repo_id": "repo-uuid"}).encode(),
1734	json.dumps(proposals_data).encode(),
1735	)
1736	with patch("urllib.request.urlopen", side_effect=resps):
1737	result = runner.invoke(cli, ["hub", "proposal", "list", "-j"])
1738	assert result.exit_code == 0
1739	arr = json.loads(next(
1740	l for l in result.output.splitlines() if l.strip().startswith("[")
1741	))
1742	assert len(arr) == 1
1743	assert arr[0]["title"] == "Valid Proposal"
1744
1745	def test_hub_override_flag_used(self, repo: pathlib.Path) -> None:
1746	"""``--hub`` override must be used instead of config URL."""
1747	# Set a different hub in config, then override via --hub
1748	runner.invoke(cli, ["hub", "connect", "http://localhost:11111/wrong/repo"])
1749	_store_identity("http://localhost:19999/gabriel/muse")
1750	proposals_data = {"proposals": []}
1751	resps = self._mock_api(
1752	json.dumps({"repo_id": "repo-uuid"}).encode(),
1753	json.dumps(proposals_data).encode(),
1754	)
1755	with patch("urllib.request.urlopen", side_effect=resps) as mock_open:
1756	result = runner.invoke(
1757	cli,
1758	["hub", "proposal", "list", "--hub", "http://localhost:19999/gabriel/muse", "-j"],
1759	)
1760	assert result.exit_code == 0
1761	# The resolved URL should contain 19999, not 11111
1762	called_urls = [c[0][0].full_url for c in mock_open.call_args_list]
1763	assert any("19999" in u for u in called_urls)
1764	assert not any("11111" in u for u in called_urls)
1765
1766	def test_proposal_list_outside_repo_exits_nonzero(
1767	self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
1768	) -> None:
1769	monkeypatch.chdir(tmp_path)
1770	monkeypatch.delenv("MUSE_REPO_ROOT", raising=False)
1771	result = runner.invoke(cli, ["hub", "proposal", "list"])
1772	assert result.exit_code != 0
1773
1774
1775	class TestFormatProposalVerbose:
1776	"""Unit tests for _format_proposal verbose mode."""
1777
1778	def test_verbose_shows_author(self) -> None:
1779	from muse.cli.commands.hub import _format_proposal
1780	proposal = {"proposalId": "abc12345", "title": "T", "state": "open",
1781	"fromBranch": "f", "toBranch": "d",
1782	"author": "alice", "createdAt": "2024-06-01T12:00:00Z"}
1783	result = _format_proposal(proposal, verbose=True)
1784	assert "alice" in result
1785
1786	def test_verbose_shows_date_prefix(self) -> None:
1787	from muse.cli.commands.hub import _format_proposal
1788	proposal = {"proposalId": "abc12345", "title": "T", "state": "open",
1789	"fromBranch": "f", "toBranch": "d",
1790	"author": "bob", "createdAt": "2024-11-30T00:00:00Z"}
1791	result = _format_proposal(proposal, verbose=True)
1792	assert "2024-11-30" in result
1793
1794	def test_verbose_ansi_in_author_stripped(self) -> None:
1795	from muse.cli.commands.hub import _format_proposal
1796	proposal = {"proposalId": "abc12345", "title": "T", "state": "open",
1797	"fromBranch": "f", "toBranch": "d",
1798	"author": "\x1b[31mevil\x1b[0m", "createdAt": "2024-01-01"}
1799	result = _format_proposal(proposal, verbose=True)
1800	assert "\x1b[" not in result
1801
1802	def test_verbose_ansi_in_created_at_stripped(self) -> None:
1803	from muse.cli.commands.hub import _format_proposal
1804	proposal = {"proposalId": "abc12345", "title": "T", "state": "open",
1805	"fromBranch": "f", "toBranch": "d",
1806	"author": "alice", "createdAt": "\x1b[32m2024-01-01\x1b[0m"}
1807	result = _format_proposal(proposal, verbose=True)
1808	assert "\x1b[" not in result
1809
1810	def test_verbose_false_omits_author(self) -> None:
1811	from muse.cli.commands.hub import _format_proposal
1812	proposal = {"proposalId": "abc12345", "title": "T", "state": "open",
1813	"fromBranch": "f", "toBranch": "d",
1814	"author": "alice", "createdAt": "2024-01-01"}
1815	result = _format_proposal(proposal, verbose=False)
1816	assert "alice" not in result
1817
1818	def test_verbose_missing_author_shows_fallback(self) -> None:
1819	from muse.cli.commands.hub import _format_proposal
1820	proposal = {"proposalId": "abc12345", "title": "T", "state": "open",
1821	"fromBranch": "f", "toBranch": "d"}
1822	result = _format_proposal(proposal, verbose=True)
1823	assert "?" in result # fallback when author absent
1824
1825	def test_verbose_closed_icon(self) -> None:
1826	from muse.cli.commands.hub import _format_proposal
1827	proposal = {"proposalId": "abc12345", "title": "T", "state": "closed",
1828	"fromBranch": "f", "toBranch": "d"}
1829	result = _format_proposal(proposal)
1830	assert "⛔" in result
1831
1832	def test_verbose_unknown_state_uses_fallback_icon(self) -> None:
1833	from muse.cli.commands.hub import _format_proposal
1834	proposal = {"proposalId": "abc12345", "title": "T", "state": "unknown_state",
1835	"fromBranch": "f", "toBranch": "d"}
1836	result = _format_proposal(proposal)
1837	assert "❓" in result
1838
1839	def test_proposal_id_truncated_to_8_chars(self) -> None:
1840	from muse.cli.commands.hub import _format_proposal
1841	proposal = {"proposalId": "abc12345-full-uuid-here", "title": "T", "state": "open",
1842	"fromBranch": "f", "toBranch": "d"}
1843	result = _format_proposal(proposal)
1844	assert "abc12345" in result
1845	# The full UUID beyond 8 chars must not appear
1846	assert "full-uuid-here" not in result
1847
1848
1849	class TestProposalListStress:
1850	"""Stress tests for `muse hub proposal list`."""
1851
1852	_HUB = "http://localhost:19999/gabriel/muse"
1853
1854	def _setup(self, repo: pathlib.Path) -> None:
1855	runner.invoke(cli, ["hub", "connect", self._HUB])
1856	_store_identity(self._HUB)
1857
1858	def _make_api_resp(self, payload_bytes: bytes) -> MagicMock:
1859	mock_resp = MagicMock()
1860	mock_resp.__enter__ = lambda s: s
1861	mock_resp.__exit__ = MagicMock(return_value=False)
1862	mock_resp.read.return_value = payload_bytes
1863	return mock_resp
1864
1865	def _mock_api(self, *responses: bytes) -> list[MagicMock]:
1866	return [self._make_api_resp(r) for r in responses]
1867
1868	def test_large_proposal_list_10000_items_json(self, repo: pathlib.Path) -> None:
1869	"""10 000 proposals in the JSON response must be handled without crashing."""
1870	self._setup(repo)
1871	proposals = [
1872	{"proposalId": f"aaaa0000-0000-0000-0000-{i:012d}",
1873	"title": f"Proposal #{i}", "state": "open",
1874	"fromBranch": f"feat/f{i}", "toBranch": "dev"}
1875	for i in range(10_000)
1876	]
1877	payload = json.dumps({"proposals": proposals}).encode()
1878	# The payload is large — make read() return the full bytes
1879	mock_resp = MagicMock()
1880	mock_resp.__enter__ = lambda s: s
1881	mock_resp.__exit__ = MagicMock(return_value=False)
1882	mock_resp.read.return_value = payload
1883
1884	repo_resp = self._make_api_resp(json.dumps({"repo_id": "repo-uuid"}).encode())
1885	with patch("urllib.request.urlopen", side_effect=[repo_resp, mock_resp]):
1886	result = runner.invoke(cli, ["hub", "proposal", "list", "-n", "10000", "-j"])
1887	assert result.exit_code == 0
1888	arr = json.loads(next(
1889	l for l in result.output.splitlines() if l.strip().startswith("[")
1890	))
1891	assert len(arr) == 10_000
1892
1893	def test_concurrent_format_proposal_calls(self) -> None:
1894	"""8 threads calling _format_proposal concurrently must produce consistent results."""
1895	from muse.cli.commands.hub import _format_proposal
1896	errors: list[str] = []
1897	results: list[str] = [""] * 8
1898
1899	def _do(idx: int) -> None:
1900	try:
1901	proposal = {
1902	"proposalId": f"aaaa{idx:04d}-0000-0000-0000-000000000001",
1903	"title": f"Proposal-{idx}: \x1b[31mevil\x1b[0m",
1904	"state": "open",
1905	"fromBranch": f"feat/f{idx}",
1906	"toBranch": "dev",
1907	"author": f"user{idx}",
1908	"createdAt": f"2024-0{(idx % 9) + 1}-01T00:00:00Z",
1909	}
1910	results[idx] = _format_proposal(proposal, verbose=True)
1911	except Exception as exc:
1912	errors.append(f"Thread {idx}: {exc}")
1913
1914	threads = [threading.Thread(target=_do, args=(i,)) for i in range(8)]
1915	for t in threads:
1916	t.start()
1917	for t in threads:
1918	t.join()
1919	assert errors == [], "Concurrent _format_proposal failures:\n" + "\n".join(errors)
1920	# Each result must have ANSI stripped and contain the user name
1921	for idx, result in enumerate(results):
1922	assert "\x1b[" not in result, f"ANSI in thread {idx} output"
1923	assert f"user{idx}" in result, f"Author missing in thread {idx} output"
1924
1925
1926	class TestProposalListE2E:
1927	"""End-to-end flow tests for `muse hub proposal list`."""
1928
1929	_HUB = "http://localhost:19999/gabriel/muse"
1930
1931	def _setup(self, repo: pathlib.Path) -> None:
1932	runner.invoke(cli, ["hub", "connect", self._HUB])
1933	_store_identity(self._HUB)
1934
1935	def _make_api_resp(self, payload_bytes: bytes) -> MagicMock:
1936	mock_resp = MagicMock()
1937	mock_resp.__enter__ = lambda s: s
1938	mock_resp.__exit__ = MagicMock(return_value=False)
1939	mock_resp.read.return_value = payload_bytes
1940	return mock_resp
1941
1942	def _mock_api(self, *responses: bytes) -> list[MagicMock]:
1943	return [self._make_api_resp(r) for r in responses]
1944
1945	def test_e2e_connect_then_list_json(self, repo: pathlib.Path) -> None:
1946	"""Full flow: connect → list --json returns a well-formed array."""
1947	self._setup(repo)
1948	proposals_data = {"proposals": [
1949	{"proposalId": "abc12345-0000-0000-0000-000000000001",
1950	"title": "My Proposal", "state": "open",
1951	"fromBranch": "feat/my", "toBranch": "dev",
1952	"author": "alice", "createdAt": "2024-03-01T09:00:00Z"},
1953	]}
1954	resps = self._mock_api(
1955	json.dumps({"repo_id": "repo-uuid"}).encode(),
1956	json.dumps(proposals_data).encode(),
1957	)
1958	with patch("urllib.request.urlopen", side_effect=resps):
1959	result = runner.invoke(cli, ["hub", "proposal", "list", "-j"])
1960	assert result.exit_code == 0
1961	arr = json.loads(next(
1962	l for l in result.output.splitlines() if l.strip().startswith("[")
1963	))
1964	assert arr[0]["title"] == "My Proposal"
1965	assert arr[0]["state"] == "open"
1966	assert arr[0]["author"] == "alice"
1967
1968	def test_e2e_list_verbose_text_all_fields_present(self, repo: pathlib.Path) -> None:
1969	"""Verbose text output includes state icon, ID prefix, branches, author, date."""
1970	self._setup(repo)
1971	proposals_data = {"proposals": [
1972	{"proposalId": "deadbeef-0000-0000-0000-000000000001",
1973	"title": "My feature", "state": "open",
1974	"fromBranch": "feat/my-feature", "toBranch": "dev",
1975	"author": "charlie", "createdAt": "2025-12-31T23:59:59Z"},
1976	]}
1977	resps = self._mock_api(
1978	json.dumps({"repo_id": "repo-uuid"}).encode(),
1979	json.dumps(proposals_data).encode(),
1980	)
1981	with patch("urllib.request.urlopen", side_effect=resps):
1982	result = runner.invoke(cli, ["hub", "proposal", "list", "-v"])
1983	assert result.exit_code == 0
1984	output = result.output
1985	assert "🟢" in output
1986	assert "deadbeef" in output
1987	assert "feat/my-feature" in output
1988	assert "charlie" in output
1989	assert "2025-12-31" in output
1990
1991	def test_e2e_empty_list_exits_zero_with_message(self, repo: pathlib.Path) -> None:
1992	"""Empty proposal list must exit 0 and print a human-friendly message."""
1993	self._setup(repo)
1994	resps = self._mock_api(
1995	json.dumps({"repo_id": "repo-uuid"}).encode(),
1996	json.dumps({"proposals": []}).encode(),
1997	)
1998	with patch("urllib.request.urlopen", side_effect=resps):
1999	result = runner.invoke(cli, ["hub", "proposal", "list", "--state", "merged"])
2000	assert result.exit_code == 0
2001	assert "No proposals" in result.output or "no proposals" in result.output.lower()
2002
2003	def test_e2e_json_no_stdout_in_text_mode(self, repo: pathlib.Path) -> None:
2004	"""In text mode, JSON must NOT appear on stdout — all output goes to stderr."""
2005	self._setup(repo)
2006	proposals_data = {"proposals": [
2007	{"proposalId": "abc12345-0000-0000-0000-000000000001",
2008	"title": "T", "state": "open",
2009	"fromBranch": "feat/x", "toBranch": "dev"},
2010	]}
2011	resps = self._mock_api(
2012	json.dumps({"repo_id": "repo-uuid"}).encode(),
2013	json.dumps(proposals_data).encode(),
2014	)
2015	with patch("urllib.request.urlopen", side_effect=resps):
2016	result = runner.invoke(cli, ["hub", "proposal", "list"])
2017	assert result.exit_code == 0
2018	# In text mode, stdout should have no JSON array
2019	for line in result.output.splitlines():
2020	stripped = line.strip()
2021	assert not stripped.startswith("["), (
2022	f"Unexpected JSON on stdout in text mode: {stripped!r}"
2023	)
2024
2025
2026	class TestProposalViewHardening:
2027	"""Additional hardening tests for `muse hub proposal view`."""
2028
2029	_HUB = "http://localhost:19999/gabriel/muse"
2030
2031	def _setup(self, repo: pathlib.Path) -> None:
2032	runner.invoke(cli, ["hub", "connect", self._HUB])
2033	_store_identity(self._HUB)
2034
2035	def _make_api_resp(self, payload_bytes: bytes) -> MagicMock:
2036	mock_resp = MagicMock()
2037	mock_resp.__enter__ = lambda s: s
2038	mock_resp.__exit__ = MagicMock(return_value=False)
2039	mock_resp.read.return_value = payload_bytes
2040	return mock_resp
2041
2042	def _mock_api(self, *responses: bytes) -> list[MagicMock]:
2043	return [self._make_api_resp(r) for r in responses]
2044
2045	def test_short_flag_j_works_for_view(self, repo: pathlib.Path) -> None:
2046	"""``-j`` is accepted as alias for ``--json``."""
2047	self._setup(repo)
2048	proposal_id = "abc12345-0000-0000-0000-000000000001"
2049	proposal_data = {"proposalId": proposal_id, "title": "T", "state": "open",
2050	"fromBranch": "feat/x", "toBranch": "dev"}
2051	proposals_data = {"proposals": [
2052	{"proposalId": proposal_id, "title": "T", "state": "open",
2053	"fromBranch": "feat/x", "toBranch": "dev"},
2054	]}
2055	resps = self._mock_api(
2056	json.dumps({"repo_id": "repo-uuid"}).encode(),
2057	json.dumps(proposals_data).encode(),
2058	json.dumps(proposal_data).encode(),
2059	)
2060	with patch("urllib.request.urlopen", side_effect=resps):
2061	result = runner.invoke(cli, ["hub", "proposal", "view", "abc12345", "-j"])
2062	assert result.exit_code == 0
2063	json_lines = [l for l in result.output.splitlines() if l.strip().startswith("{")]
2064	assert len(json_lines) >= 1
2065
2066	def test_ansi_in_state_sanitized(
2067	self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]
2068	) -> None:
2069	"""ANSI in ``state`` field must not reach terminal in text mode."""
2070	self._setup(repo)
2071	proposal_id = "abc12345-0000-0000-0000-000000000001"
2072	evil_proposal = {"proposalId": proposal_id, "title": "T",
2073	"state": "\x1b[31mopen\x1b[0m",
2074	"fromBranch": "feat/x", "toBranch": "dev"}
2075	proposals_data = {"proposals": [
2076	{"proposalId": proposal_id, "title": "T", "state": "open",
2077	"fromBranch": "feat/x", "toBranch": "dev"},
2078	]}
2079	resps = self._mock_api(
2080	json.dumps({"repo_id": "repo-uuid"}).encode(),
2081	json.dumps(proposals_data).encode(),
2082	json.dumps(evil_proposal).encode(),
2083	)
2084	with patch("urllib.request.urlopen", side_effect=resps):
2085	result = runner.invoke(cli, ["hub", "proposal", "view", "abc12345"])
2086	assert result.exit_code == 0
2087	assert "\x1b[" not in result.output
2088
2089	def test_ansi_in_branch_sanitized(self, repo: pathlib.Path) -> None:
2090	"""ANSI in branch names must not reach terminal in text mode."""
2091	self._setup(repo)
2092	proposal_id = "abc12345-0000-0000-0000-000000000001"
2093	evil_proposal = {"proposalId": proposal_id, "title": "T", "state": "open",
2094	"fromBranch": "\x1b[32mfeat/evil\x1b[0m",
2095	"toBranch": "\x1b[34mdev\x1b[0m"}
2096	proposals_data = {"proposals": [
2097	{"proposalId": proposal_id, "title": "T", "state": "open",
2098	"fromBranch": "feat/x", "toBranch": "dev"},
2099	]}
2100	resps = self._mock_api(
2101	json.dumps({"repo_id": "repo-uuid"}).encode(),
2102	json.dumps(proposals_data).encode(),
2103	json.dumps(evil_proposal).encode(),
2104	)
2105	with patch("urllib.request.urlopen", side_effect=resps):
2106	result = runner.invoke(cli, ["hub", "proposal", "view", "abc12345"])
2107	assert result.exit_code == 0
2108	assert "\x1b[" not in result.output
2109
2110	def test_ansi_in_body_lines_sanitized(self, repo: pathlib.Path) -> None:
2111	"""ANSI in body text must not reach terminal in text mode."""
2112	self._setup(repo)
2113	proposal_id = "abc12345-0000-0000-0000-000000000001"
2114	evil_proposal = {"proposalId": proposal_id, "title": "T", "state": "open",
2115	"fromBranch": "feat/x", "toBranch": "dev",
2116	"body": "\x1b[31mThis body has ANSI\x1b[0m"}
2117	proposals_data = {"proposals": [
2118	{"proposalId": proposal_id, "title": "T", "state": "open",
2119	"fromBranch": "feat/x", "toBranch": "dev"},
2120	]}
2121	resps = self._mock_api(
2122	json.dumps({"repo_id": "repo-uuid"}).encode(),
2123	json.dumps(proposals_data).encode(),
2124	json.dumps(evil_proposal).encode(),
2125	)
2126	with patch("urllib.request.urlopen", side_effect=resps):
2127	result = runner.invoke(cli, ["hub", "proposal", "view", "abc12345"])
2128	assert result.exit_code == 0
2129	assert "\x1b[" not in result.output
2130
2131	def test_view_prefix_not_found_exits_nonzero(self, repo: pathlib.Path) -> None:
2132	self._setup(repo)
2133	proposals_data = {"proposals": []}
2134	resps = self._mock_api(
2135	json.dumps({"repo_id": "repo-uuid"}).encode(),
2136	json.dumps(proposals_data).encode(),
2137	)
2138	with patch("urllib.request.urlopen", side_effect=resps):
2139	result = runner.invoke(cli, ["hub", "proposal", "view", "deadbeef"])
2140	assert result.exit_code != 0
2141
2142	def test_view_no_hub_exits_nonzero(self, repo: pathlib.Path) -> None:
2143	result = runner.invoke(cli, ["hub", "proposal", "view", "abc12345"])
2144	assert result.exit_code != 0
2145
2146	def test_view_no_identity_exits_nonzero(self, repo: pathlib.Path) -> None:
2147	runner.invoke(cli, ["hub", "connect", self._HUB])
2148	result = runner.invoke(cli, ["hub", "proposal", "view", "abc12345"])
2149	assert result.exit_code != 0
2150
2151	def test_view_outside_repo_exits_nonzero(
2152	self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
2153	) -> None:
2154	monkeypatch.chdir(tmp_path)
2155	monkeypatch.delenv("MUSE_REPO_ROOT", raising=False)
2156	result = runner.invoke(cli, ["hub", "proposal", "view", "abc12345"])
2157	assert result.exit_code != 0
2158
2159	def test_full_uuid_skips_prefix_resolution(self, repo: pathlib.Path) -> None:
2160	"""A full UUID must reach the view endpoint with exactly 2 API calls (no prefix fetch)."""
2161	self._setup(repo)
2162	proposal_id = "abc12345-def0-0000-0000-000000000001"
2163	proposal_data = {"proposalId": proposal_id, "title": "T", "state": "open",
2164	"fromBranch": "feat/x", "toBranch": "dev"}
2165	resps = self._mock_api(
2166	json.dumps({"repo_id": "repo-uuid"}).encode(), # _resolve_repo_id
2167	json.dumps(proposal_data).encode(), # GET proposals/{id}
2168	)
2169	with patch("urllib.request.urlopen", side_effect=resps) as mock_open:
2170	result = runner.invoke(cli, ["hub", "proposal", "view", proposal_id, "-j"])
2171	assert result.exit_code == 0
2172	# Only 2 urlopen calls: repo resolution + the view fetch (no prefix list call)
2173	assert mock_open.call_count == 2
2174
2175	def test_prefix_triggers_resolution_call(self, repo: pathlib.Path) -> None:
2176	"""An 8-char prefix must trigger a prefix-resolution list fetch (3 API calls total)."""
2177	self._setup(repo)
2178	proposal_id = "abc12345-0000-0000-0000-000000000001"
2179	proposal_data = {"proposalId": proposal_id, "title": "T", "state": "open",
2180	"fromBranch": "feat/x", "toBranch": "dev"}
2181	proposals_data = {"proposals": [
2182	{"proposalId": proposal_id, "title": "T", "state": "open",
2183	"fromBranch": "feat/x", "toBranch": "dev"},
2184	]}
2185	resps = self._mock_api(
2186	json.dumps({"repo_id": "repo-uuid"}).encode(), # _resolve_repo_id
2187	json.dumps(proposals_data).encode(), # prefix resolution list
2188	json.dumps(proposal_data).encode(), # GET proposals/{id}
2189	)
2190	with patch("urllib.request.urlopen", side_effect=resps) as mock_open:
2191	result = runner.invoke(cli, ["hub", "proposal", "view", "abc12345", "-j"])
2192	assert result.exit_code == 0
2193	assert mock_open.call_count == 3
2194
2195	def test_author_shown_in_text_mode(self, repo: pathlib.Path) -> None:
2196	self._setup(repo)
2197	proposal_id = "abc12345-0000-0000-0000-000000000001"
2198	proposal_data = {"proposalId": proposal_id, "title": "T", "state": "open",
2199	"fromBranch": "feat/x", "toBranch": "dev",
2200	"author": "charlie", "createdAt": "2024-07-04T00:00:00Z"}
2201	resps = self._mock_api(
2202	json.dumps({"repo_id": "repo-uuid"}).encode(),
2203	json.dumps({"proposals": [
2204	{"proposalId": proposal_id, "title": "T", "state": "open",
2205	"fromBranch": "feat/x", "toBranch": "dev"},
2206	]}).encode(),
2207	json.dumps(proposal_data).encode(),
2208	)
2209	with patch("urllib.request.urlopen", side_effect=resps):
2210	result = runner.invoke(cli, ["hub", "proposal", "view", "abc12345"])
2211	assert result.exit_code == 0
2212	assert "charlie" in result.output
2213
2214	def test_created_at_shown_in_text_mode(self, repo: pathlib.Path) -> None:
2215	self._setup(repo)
2216	proposal_id = "abc12345-0000-0000-0000-000000000001"
2217	proposal_data = {"proposalId": proposal_id, "title": "T", "state": "open",
2218	"fromBranch": "feat/x", "toBranch": "dev",
2219	"author": "alice", "createdAt": "2025-03-15T08:30:00Z"}
2220	resps = self._mock_api(
2221	json.dumps({"repo_id": "repo-uuid"}).encode(),
2222	json.dumps({"proposals": [
2223	{"proposalId": proposal_id, "title": "T", "state": "open",
2224	"fromBranch": "feat/x", "toBranch": "dev"},
2225	]}).encode(),
2226	json.dumps(proposal_data).encode(),
2227	)
2228	with patch("urllib.request.urlopen", side_effect=resps):
2229	result = runner.invoke(cli, ["hub", "proposal", "view", "abc12345"])
2230	assert result.exit_code == 0
2231	assert "2025-03-15" in result.output
2232
2233	def test_ansi_in_author_sanitized(self, repo: pathlib.Path) -> None:
2234	self._setup(repo)
2235	proposal_id = "abc12345-0000-0000-0000-000000000001"
2236	proposal_data = {"proposalId": proposal_id, "title": "T", "state": "open",
2237	"fromBranch": "feat/x", "toBranch": "dev",
2238	"author": "\x1b[31mevil-author\x1b[0m",
2239	"createdAt": "2024-01-01T00:00:00Z"}
2240	resps = self._mock_api(
2241	json.dumps({"repo_id": "repo-uuid"}).encode(),
2242	json.dumps({"proposals": [
2243	{"proposalId": proposal_id, "title": "T", "state": "open",
2244	"fromBranch": "feat/x", "toBranch": "dev"},
2245	]}).encode(),
2246	json.dumps(proposal_data).encode(),
2247	)
2248	with patch("urllib.request.urlopen", side_effect=resps):
2249	result = runner.invoke(cli, ["hub", "proposal", "view", "abc12345"])
2250	assert result.exit_code == 0
2251	assert "\x1b[" not in result.output
2252
2253	def test_ansi_in_created_at_sanitized(self, repo: pathlib.Path) -> None:
2254	self._setup(repo)
2255	proposal_id = "abc12345-0000-0000-0000-000000000001"
2256	proposal_data = {"proposalId": proposal_id, "title": "T", "state": "open",
2257	"fromBranch": "feat/x", "toBranch": "dev",
2258	"author": "alice",
2259	"createdAt": "\x1b[32m2024-01-01\x1b[0mTevil"}
2260	resps = self._mock_api(
2261	json.dumps({"repo_id": "repo-uuid"}).encode(),
2262	json.dumps({"proposals": [
2263	{"proposalId": proposal_id, "title": "T", "state": "open",
2264	"fromBranch": "feat/x", "toBranch": "dev"},
2265	]}).encode(),
2266	json.dumps(proposal_data).encode(),
2267	)
2268	with patch("urllib.request.urlopen", side_effect=resps):
2269	result = runner.invoke(cli, ["hub", "proposal", "view", "abc12345"])
2270	assert result.exit_code == 0
2271	assert "\x1b[" not in result.output
2272
2273	def test_body_truncation_hint_shown(self, repo: pathlib.Path) -> None:
2274	"""Body exceeding _MAX_PROPOSAL_BODY_LINES must show a truncation hint."""
2275	from muse.cli.commands.hub import _MAX_PROPOSAL_BODY_LINES
2276	self._setup(repo)
2277	proposal_id = "abc12345-0000-0000-0000-000000000001"
2278	long_body = "\n".join(f"line {i}" for i in range(_MAX_PROPOSAL_BODY_LINES + 5))
2279	proposal_data = {"proposalId": proposal_id, "title": "T", "state": "open",
2280	"fromBranch": "feat/x", "toBranch": "dev", "body": long_body}
2281	resps = self._mock_api(
2282	json.dumps({"repo_id": "repo-uuid"}).encode(),
2283	json.dumps({"proposals": [
2284	{"proposalId": proposal_id, "title": "T", "state": "open",
2285	"fromBranch": "feat/x", "toBranch": "dev"},
2286	]}).encode(),
2287	json.dumps(proposal_data).encode(),
2288	)
2289	with patch("urllib.request.urlopen", side_effect=resps):
2290	result = runner.invoke(cli, ["hub", "proposal", "view", "abc12345"])
2291	assert result.exit_code == 0
2292	assert "more line" in result.output
2293	assert "--json" in result.output # hint mentions --json
2294
2295	def test_body_exactly_at_limit_no_hint(self, repo: pathlib.Path) -> None:
2296	"""Body at exactly _MAX_PROPOSAL_BODY_LINES must NOT show a truncation hint."""
2297	from muse.cli.commands.hub import _MAX_PROPOSAL_BODY_LINES
2298	self._setup(repo)
2299	proposal_id = "abc12345-0000-0000-0000-000000000001"
2300	exact_body = "\n".join(f"line {i}" for i in range(_MAX_PROPOSAL_BODY_LINES))
2301	proposal_data = {"proposalId": proposal_id, "title": "T", "state": "open",
2302	"fromBranch": "feat/x", "toBranch": "dev", "body": exact_body}
2303	resps = self._mock_api(
2304	json.dumps({"repo_id": "repo-uuid"}).encode(),
2305	json.dumps({"proposals": [
2306	{"proposalId": proposal_id, "title": "T", "state": "open",
2307	"fromBranch": "feat/x", "toBranch": "dev"},
2308	]}).encode(),
2309	json.dumps(proposal_data).encode(),
2310	)
2311	with patch("urllib.request.urlopen", side_effect=resps):
2312	result = runner.invoke(cli, ["hub", "proposal", "view", "abc12345"])
2313	assert result.exit_code == 0
2314	assert "more line" not in result.output
2315
2316	def test_no_body_field_no_body_section(self, repo: pathlib.Path) -> None:
2317	"""When body is absent or empty, no 'Body:' section must appear."""
2318	self._setup(repo)
2319	proposal_id = "abc12345-0000-0000-0000-000000000001"
2320	proposal_data = {"proposalId": proposal_id, "title": "T", "state": "open",
2321	"fromBranch": "feat/x", "toBranch": "dev"}
2322	resps = self._mock_api(
2323	json.dumps({"repo_id": "repo-uuid"}).encode(),
2324	json.dumps({"proposals": [
2325	{"proposalId": proposal_id, "title": "T", "state": "open",
2326	"fromBranch": "feat/x", "toBranch": "dev"},
2327	]}).encode(),
2328	json.dumps(proposal_data).encode(),
2329	)
2330	with patch("urllib.request.urlopen", side_effect=resps):
2331	result = runner.invoke(cli, ["hub", "proposal", "view", "abc12345"])
2332	assert result.exit_code == 0
2333	assert "Body:" not in result.output
2334
2335	def test_json_passthrough_includes_all_fields(self, repo: pathlib.Path) -> None:
2336	"""JSON output must be an unmodified passthrough from the API."""
2337	self._setup(repo)
2338	proposal_id = "abc12345-0000-0000-0000-000000000001"
2339	proposal_data = {"proposalId": proposal_id, "title": "My Proposal", "state": "open",
2340	"fromBranch": "feat/x", "toBranch": "dev",
2341	"author": "alice", "createdAt": "2024-01-01T00:00:00Z",
2342	"body": "Full body text here.",
2343	"extraField": "agent-visible"}
2344	resps = self._mock_api(
2345	json.dumps({"repo_id": "repo-uuid"}).encode(),
2346	json.dumps({"proposals": [
2347	{"proposalId": proposal_id, "title": "My Proposal", "state": "open",
2348	"fromBranch": "feat/x", "toBranch": "dev"},
2349	]}).encode(),
2350	json.dumps(proposal_data).encode(),
2351	)
2352	with patch("urllib.request.urlopen", side_effect=resps):
2353	result = runner.invoke(cli, ["hub", "proposal", "view", "abc12345", "-j"])
2354	assert result.exit_code == 0
2355	data = json.loads(next(
2356	l for l in result.output.splitlines() if l.strip().startswith("{")
2357	))
2358	assert data["author"] == "alice"
2359	assert data["body"] == "Full body text here."
2360	assert data["extraField"] == "agent-visible"
2361
2362	def test_hub_override_flag(self, repo: pathlib.Path) -> None:
2363	"""``--hub`` must route requests to the override URL."""
2364	runner.invoke(cli, ["hub", "connect", "http://localhost:11111/wrong/repo"])
2365	_store_identity("http://localhost:19999/gabriel/muse")
2366	proposal_id = "abc12345-def0-0000-0000-000000000001"
2367	proposal_data = {"proposalId": proposal_id, "title": "T", "state": "open",
2368	"fromBranch": "f", "toBranch": "d"}
2369	resps = self._mock_api(
2370	json.dumps({"repo_id": "repo-uuid"}).encode(),
2371	json.dumps(proposal_data).encode(),
2372	)
2373	with patch("urllib.request.urlopen", side_effect=resps) as mock_open:
2374	result = runner.invoke(
2375	cli,
2376	["hub", "proposal", "view", proposal_id,
2377	"--hub", "http://localhost:19999/gabriel/muse", "-j"],
2378	)
2379	assert result.exit_code == 0
2380	called_urls = [c[0][0].full_url for c in mock_open.call_args_list]
2381	assert any("19999" in u for u in called_urls)
2382	assert not any("11111" in u for u in called_urls)
2383
2384
2385	class TestProposalViewUnit:
2386	"""Pure unit tests for run_proposal_view text rendering logic."""
2387
2388	def _make_proposal_resp(self, **kwargs: str) -> bytes:
2389	base: Manifest = {"proposalId": "abc12345-0000-0000-0000-000000000001",
2390	"title": "My Proposal", "state": "open",
2391	"fromBranch": "feat/x", "toBranch": "dev"}
2392	base.update(kwargs)
2393	return json.dumps(base).encode()
2394
2395	def _invoke_view(
2396	self,
2397	repo: pathlib.Path,
2398	proposal_data: bytes,
2399	*,
2400	flags: list[str] \| None = None,
2401	) -> InvokeResult:
2402	"""Invoke hub proposal view with a pre-resolved full UUID (2 API calls only)."""
2403	proposal_id = "abc12345-def0-0000-0000-000000000001"
2404	# Use a full UUID to skip the prefix-resolution fetch
2405	mock_repo = MagicMock()
2406	mock_repo.__enter__ = lambda s: s
2407	mock_repo.__exit__ = MagicMock(return_value=False)
2408	mock_repo.read.return_value = json.dumps({"repo_id": "repo-uuid"}).encode()
2409
2410	mock_proposal = MagicMock()
2411	mock_proposal.__enter__ = lambda s: s
2412	mock_proposal.__exit__ = MagicMock(return_value=False)
2413	mock_proposal.read.return_value = proposal_data
2414
2415	cmd = ["hub", "proposal", "view", proposal_id] + (flags or [])
2416	with patch("urllib.request.urlopen", side_effect=[mock_repo, mock_proposal]):
2417	return runner.invoke(cli, cmd)
2418
2419	def test_state_open_icon(self, repo: pathlib.Path) -> None:
2420	runner.invoke(cli, ["hub", "connect", "http://localhost:19999/gabriel/muse"])
2421	_store_identity("http://localhost:19999/gabriel/muse")
2422	result = self._invoke_view(repo, self._make_proposal_resp(state="open"))
2423	assert "🟢" in result.output
2424
2425	def test_state_merged_icon(self, repo: pathlib.Path) -> None:
2426	runner.invoke(cli, ["hub", "connect", "http://localhost:19999/gabriel/muse"])
2427	_store_identity("http://localhost:19999/gabriel/muse")
2428	result = self._invoke_view(repo, self._make_proposal_resp(state="merged"))
2429	assert "🟣" in result.output
2430
2431	def test_state_closed_icon(self, repo: pathlib.Path) -> None:
2432	runner.invoke(cli, ["hub", "connect", "http://localhost:19999/gabriel/muse"])
2433	_store_identity("http://localhost:19999/gabriel/muse")
2434	result = self._invoke_view(repo, self._make_proposal_resp(state="closed"))
2435	assert "⛔" in result.output
2436
2437	def test_unknown_state_fallback_icon(self, repo: pathlib.Path) -> None:
2438	runner.invoke(cli, ["hub", "connect", "http://localhost:19999/gabriel/muse"])
2439	_store_identity("http://localhost:19999/gabriel/muse")
2440	result = self._invoke_view(repo, self._make_proposal_resp(state="draft"))
2441	assert "❓" in result.output
2442
2443	def test_no_author_field_omits_by_line(self, repo: pathlib.Path) -> None:
2444	"""When author is absent, the 'By:' line must not appear."""
2445	runner.invoke(cli, ["hub", "connect", "http://localhost:19999/gabriel/muse"])
2446	_store_identity("http://localhost:19999/gabriel/muse")
2447	result = self._invoke_view(repo, self._make_proposal_resp())
2448	assert "By:" not in result.output
2449
2450	def test_state_upper_in_header(self, repo: pathlib.Path) -> None:
2451	runner.invoke(cli, ["hub", "connect", "http://localhost:19999/gabriel/muse"])
2452	_store_identity("http://localhost:19999/gabriel/muse")
2453	result = self._invoke_view(repo, self._make_proposal_resp(state="open"))
2454	assert "[OPEN]" in result.output
2455
2456	def test_id_and_branches_in_output(self, repo: pathlib.Path) -> None:
2457	runner.invoke(cli, ["hub", "connect", "http://localhost:19999/gabriel/muse"])
2458	_store_identity("http://localhost:19999/gabriel/muse")
2459	proposal_id = "abc12345-def0-0000-0000-000000000001"
2460	result = self._invoke_view(
2461	repo,
2462	self._make_proposal_resp(proposalId=proposal_id, fromBranch="feat/my", toBranch="main"),
2463	)
2464	assert "feat/my" in result.output
2465	assert "main" in result.output
2466
2467
2468	class TestProposalViewE2E:
2469	"""End-to-end scenario tests for `muse hub proposal view`."""
2470
2471	_HUB = "http://localhost:19999/gabriel/muse"
2472
2473	def _setup(self, repo: pathlib.Path) -> None:
2474	runner.invoke(cli, ["hub", "connect", self._HUB])
2475	_store_identity(self._HUB)
2476
2477	def _make_api_resp(self, payload_bytes: bytes) -> MagicMock:
2478	mock_resp = MagicMock()
2479	mock_resp.__enter__ = lambda s: s
2480	mock_resp.__exit__ = MagicMock(return_value=False)
2481	mock_resp.read.return_value = payload_bytes
2482	return mock_resp
2483
2484	def _mock_api(self, *responses: bytes) -> list[MagicMock]:
2485	return [self._make_api_resp(r) for r in responses]
2486
2487	def test_e2e_full_proposal_text_output(self, repo: pathlib.Path) -> None:
2488	"""Full flow with all optional fields — all sections must appear."""
2489	self._setup(repo)
2490	proposal_id = "deadbeef-cafe-0000-0000-000000000001"
2491	proposal_data = {
2492	"proposalId": proposal_id,
2493	"title": "feat: add sonic synthesis",
2494	"state": "open",
2495	"fromBranch": "feat/sonic",
2496	"toBranch": "dev",
2497	"author": "gabriel",
2498	"createdAt": "2025-06-01T12:00:00Z",
2499	"body": "This proposal adds sonic synthesis support.",
2500	}
2501	resps = self._mock_api(
2502	json.dumps({"repo_id": "repo-uuid"}).encode(),
2503	json.dumps(proposal_data).encode(),
2504	)
2505	with patch("urllib.request.urlopen", side_effect=resps):
2506	result = runner.invoke(cli, ["hub", "proposal", "view", proposal_id])
2507	assert result.exit_code == 0
2508	output = result.output
2509	assert "🟢" in output
2510	assert "feat: add sonic synthesis" in output
2511	assert "feat/sonic" in output
2512	assert "gabriel" in output
2513	assert "2025-06-01" in output
2514	assert "sonic synthesis support" in output
2515
2516	def test_e2e_json_agent_workflow(self, repo: pathlib.Path) -> None:
2517	"""Simulate an agent extracting state via --json \| jq."""
2518	self._setup(repo)
2519	proposal_id = "deadbeef-cafe-0000-0000-000000000001"
2520	proposal_data = {"proposalId": proposal_id, "title": "T", "state": "merged",
2521	"fromBranch": "feat/x", "toBranch": "dev",
2522	"author": "bot", "mergeCommitId": "aabbccdd11223344"}
2523	resps = self._mock_api(
2524	json.dumps({"repo_id": "repo-uuid"}).encode(),
2525	json.dumps(proposal_data).encode(),
2526	)
2527	with patch("urllib.request.urlopen", side_effect=resps):
2528	result = runner.invoke(cli, ["hub", "proposal", "view", proposal_id, "--json"])
2529	assert result.exit_code == 0
2530	data = json.loads(next(
2531	l for l in result.output.splitlines() if l.strip().startswith("{")
2532	))
2533	assert data["state"] == "merged"
2534	assert data["mergeCommitId"] == "aabbccdd11223344"
2535
2536	def test_e2e_body_truncation_hint_points_to_json(self, repo: pathlib.Path) -> None:
2537	"""Truncation hint must explicitly mention --json."""
2538	from muse.cli.commands.hub import _MAX_PROPOSAL_BODY_LINES
2539	self._setup(repo)
2540	proposal_id = "deadbeef-cafe-0000-0000-000000000001"
2541	long_body = "\n".join(f"line {i}" for i in range(_MAX_PROPOSAL_BODY_LINES + 10))
2542	proposal_data = {"proposalId": proposal_id, "title": "T", "state": "open",
2543	"fromBranch": "feat/x", "toBranch": "dev", "body": long_body}
2544	resps = self._mock_api(
2545	json.dumps({"repo_id": "repo-uuid"}).encode(),
2546	json.dumps(proposal_data).encode(),
2547	)
2548	with patch("urllib.request.urlopen", side_effect=resps):
2549	result = runner.invoke(cli, ["hub", "proposal", "view", proposal_id])
2550	assert result.exit_code == 0
2551	assert "--json" in result.output
2552	assert "10 more line" in result.output
2553
2554	def test_e2e_ambiguous_prefix_exits_nonzero(self, repo: pathlib.Path) -> None:
2555	"""Two proposals with the same prefix must cause a non-zero exit."""
2556	self._setup(repo)
2557	proposals_data = {"proposals": [
2558	{"proposalId": "abc12345-0000-0000-0000-000000000001", "title": "Proposal 1",
2559	"state": "open", "fromBranch": "feat/a", "toBranch": "dev"},
2560	{"proposalId": "abc12345-0000-0000-0000-000000000002", "title": "Proposal 2",
2561	"state": "open", "fromBranch": "feat/b", "toBranch": "dev"},
2562	]}
2563	resps = self._mock_api(
2564	json.dumps({"repo_id": "repo-uuid"}).encode(),
2565	json.dumps(proposals_data).encode(),
2566	)
2567	with patch("urllib.request.urlopen", side_effect=resps):
2568	result = runner.invoke(cli, ["hub", "proposal", "view", "abc12345"])
2569	assert result.exit_code != 0
2570
2571
2572	class TestProposalViewStress:
2573	"""Stress tests for `muse hub proposal view`."""
2574
2575	_HUB = "http://localhost:19999/gabriel/muse"
2576
2577	def test_body_with_1000_lines_truncated(self, repo: pathlib.Path) -> None:
2578	"""A 1000-line body must be accepted without OOM and truncated correctly."""
2579	from muse.cli.commands.hub import _MAX_PROPOSAL_BODY_LINES
2580
2581	runner.invoke(cli, ["hub", "connect", self._HUB])
2582	_store_identity(self._HUB)
2583
2584	proposal_id = "deadbeef-cafe-0000-0000-000000000001"
2585	big_body = "\n".join(f"line {i}" for i in range(1000))
2586	proposal_data = {"proposalId": proposal_id, "title": "T", "state": "open",
2587	"fromBranch": "feat/x", "toBranch": "dev", "body": big_body}
2588
2589	mock_repo = MagicMock()
2590	mock_repo.__enter__ = lambda s: s
2591	mock_repo.__exit__ = MagicMock(return_value=False)
2592	mock_repo.read.return_value = json.dumps({"repo_id": "repo-uuid"}).encode()
2593
2594	mock_proposal = MagicMock()
2595	mock_proposal.__enter__ = lambda s: s
2596	mock_proposal.__exit__ = MagicMock(return_value=False)
2597	mock_proposal.read.return_value = json.dumps(proposal_data).encode()
2598
2599	with patch("urllib.request.urlopen", side_effect=[mock_repo, mock_proposal]):
2600	result = runner.invoke(cli, ["hub", "proposal", "view", proposal_id])
2601	assert result.exit_code == 0
2602	lines_shown = [l for l in result.output.splitlines() if l.strip().startswith("line ")]
2603	assert len(lines_shown) == _MAX_PROPOSAL_BODY_LINES
2604	assert "more line" in result.output
2605
2606	def test_concurrent_format_operations(self) -> None:
2607	"""_format_proposal called concurrently from 8 threads must not produce ANSI leakage."""
2608	from muse.cli.commands.hub import _format_proposal
2609	errors: list[str] = []
2610
2611	def _do(idx: int) -> None:
2612	try:
2613	proposal = {
2614	"proposalId": f"dead{idx:04d}-0000-0000-0000-000000000001",
2615	"title": f"\x1b[31mProposal-{idx}\x1b[0m",
2616	"state": "open",
2617	"fromBranch": f"\x1b[32mfeat/f{idx}\x1b[0m",
2618	"toBranch": "dev",
2619	}
2620	result = _format_proposal(proposal)
2621	assert "\x1b[" not in result, f"Thread {idx}: ANSI leaked"
2622	except Exception as exc:
2623	errors.append(f"Thread {idx}: {exc}")
2624
2625	threads = [threading.Thread(target=_do, args=(i,)) for i in range(8)]
2626	for t in threads:
2627	t.start()
2628	for t in threads:
2629	t.join()
2630	assert errors == [], "\n".join(errors)
2631
2632
2633	class TestProposalCreateHardening:
2634	"""Additional hardening tests for `muse hub proposal create`."""
2635
2636	_HUB = "http://localhost:19999/gabriel/muse"
2637
2638	def _setup(self, repo: pathlib.Path) -> None:
2639	runner.invoke(cli, ["hub", "connect", self._HUB])
2640	_store_identity(self._HUB)
2641
2642	def _make_api_resp(self, payload_bytes: bytes) -> MagicMock:
2643	mock_resp = MagicMock()
2644	mock_resp.__enter__ = lambda s: s
2645	mock_resp.__exit__ = MagicMock(return_value=False)
2646	mock_resp.read.return_value = payload_bytes
2647	return mock_resp
2648
2649	def _mock_api(self, *responses: bytes) -> list[MagicMock]:
2650	return [self._make_api_resp(r) for r in responses]
2651
2652	def test_short_flag_j_works_for_create(self, repo: pathlib.Path) -> None:
2653	self._setup(repo)
2654	(repo / ".muse" / "refs" / "heads" / "feat-x").write_text("")
2655	(repo / ".muse" / "HEAD").write_text("ref: refs/heads/feat-x\n")
2656	create_resp = {"proposalId": "abc12345-0000-0000-0000-000000000001",
2657	"state": "open", "fromBranch": "feat-x", "toBranch": "dev"}
2658	resps = self._mock_api(
2659	json.dumps({"repo_id": "repo-uuid"}).encode(),
2660	json.dumps(create_resp).encode(),
2661	)
2662	with patch("urllib.request.urlopen", side_effect=resps):
2663	result = runner.invoke(
2664	cli,
2665	["hub", "proposal", "create", "--title", "T", "--from-branch", "feat-x", "-j"],
2666	)
2667	assert result.exit_code == 0
2668	json_lines = [l for l in result.output.splitlines() if l.strip().startswith("{")]
2669	assert len(json_lines) >= 1
2670
2671	def test_ansi_in_proposal_id_sanitized_text_output(self, repo: pathlib.Path) -> None:
2672	"""ANSI in returned proposalId must not reach terminal in text mode."""
2673	self._setup(repo)
2674	(repo / ".muse" / "refs" / "heads" / "feat-x").write_text("")
2675	(repo / ".muse" / "HEAD").write_text("ref: refs/heads/feat-x\n")
2676	create_resp = {"proposalId": "\x1b[31mabc12345-evil\x1b[0m",
2677	"state": "open", "fromBranch": "feat-x", "toBranch": "dev"}
2678	resps = self._mock_api(
2679	json.dumps({"repo_id": "repo-uuid"}).encode(),
2680	json.dumps(create_resp).encode(),
2681	)
2682	with patch("urllib.request.urlopen", side_effect=resps):
2683	result = runner.invoke(
2684	cli,
2685	["hub", "proposal", "create", "--title", "T", "--from-branch", "feat-x"],
2686	)
2687	assert "\x1b[" not in result.output
2688
2689	def test_ansi_in_title_sanitized_text_output(self, repo: pathlib.Path) -> None:
2690	"""ANSI in title arg must not reach terminal in text mode."""
2691	self._setup(repo)
2692	(repo / ".muse" / "refs" / "heads" / "feat-x").write_text("")
2693	(repo / ".muse" / "HEAD").write_text("ref: refs/heads/feat-x\n")
2694	create_resp = {"proposalId": "abc12345-0000-0000-0000-000000000001",
2695	"state": "open", "fromBranch": "feat-x", "toBranch": "dev"}
2696	resps = self._mock_api(
2697	json.dumps({"repo_id": "repo-uuid"}).encode(),
2698	json.dumps(create_resp).encode(),
2699	)
2700	with patch("urllib.request.urlopen", side_effect=resps):
2701	result = runner.invoke(
2702	cli,
2703	["hub", "proposal", "create",
2704	"--title", "\x1b[31mevil title\x1b[0m",
2705	"--from-branch", "feat-x"],
2706	)
2707	assert "\x1b[" not in result.output
2708
2709
2710	class TestProposalCreateSecurity:
2711	"""Security-focused tests for `muse hub proposal create`."""
2712
2713	_HUB = "http://localhost:19999/gabriel/muse"
2714
2715	def _setup(self, repo: pathlib.Path) -> None:
2716	runner.invoke(cli, ["hub", "connect", self._HUB])
2717	_store_identity(self._HUB)
2718
2719	def _make_api_resp(self, payload_bytes: bytes) -> MagicMock:
2720	mock_resp = MagicMock()
2721	mock_resp.__enter__ = lambda s: s
2722	mock_resp.__exit__ = MagicMock(return_value=False)
2723	mock_resp.read.return_value = payload_bytes
2724	return mock_resp
2725
2726	def _mock_api(self, *responses: bytes) -> list[MagicMock]:
2727	return [self._make_api_resp(r) for r in responses]
2728
2729	def test_ansi_in_from_branch_sanitized(self, repo: pathlib.Path) -> None:
2730	self._setup(repo)
2731	create_resp = {"proposalId": "abc12345-0000-0000-0000-000000000001",
2732	"state": "open", "fromBranch": "feat-x", "toBranch": "dev"}
2733	resps = self._mock_api(
2734	json.dumps({"repo_id": "repo-uuid"}).encode(),
2735	json.dumps(create_resp).encode(),
2736	)
2737	with patch("urllib.request.urlopen", side_effect=resps):
2738	result = runner.invoke(
2739	cli,
2740	["hub", "proposal", "create", "--title", "T",
2741	"--from-branch", "\x1b[31mfeat/evil\x1b[0m"],
2742	)
2743	assert "\x1b[" not in result.output
2744
2745	def test_ansi_in_to_branch_sanitized(self, repo: pathlib.Path) -> None:
2746	self._setup(repo)
2747	create_resp = {"proposalId": "abc12345-0000-0000-0000-000000000001",
2748	"state": "open", "fromBranch": "feat-x", "toBranch": "dev"}
2749	resps = self._mock_api(
2750	json.dumps({"repo_id": "repo-uuid"}).encode(),
2751	json.dumps(create_resp).encode(),
2752	)
2753	with patch("urllib.request.urlopen", side_effect=resps):
2754	result = runner.invoke(
2755	cli,
2756	["hub", "proposal", "create", "--title", "T",
2757	"--from-branch", "feat-x",
2758	"--to-branch", "\x1b[32mdev\x1b[0m"],
2759	)
2760	assert "\x1b[" not in result.output
2761
2762	def test_empty_title_exits_nonzero(self, repo: pathlib.Path) -> None:
2763	"""Empty (whitespace-only) title must be rejected before any API call."""
2764	self._setup(repo)
2765	with patch("urllib.request.urlopen") as mock_net:
2766	result = runner.invoke(
2767	cli,
2768	["hub", "proposal", "create", "--title", " ",
2769	"--from-branch", "feat/x"],
2770	)
2771	assert result.exit_code != 0
2772	mock_net.assert_not_called()
2773
2774	def test_title_too_long_exits_nonzero(self, repo: pathlib.Path) -> None:
2775	"""Title exceeding _MAX_PROPOSAL_TITLE_LEN must be rejected before any API call."""
2776	from muse.cli.commands.hub import _MAX_PROPOSAL_TITLE_LEN
2777	self._setup(repo)
2778	long_title = "x" * (_MAX_PROPOSAL_TITLE_LEN + 1)
2779	with patch("urllib.request.urlopen") as mock_net:
2780	result = runner.invoke(
2781	cli,
2782	["hub", "proposal", "create", "--title", long_title,
2783	"--from-branch", "feat/x"],
2784	)
2785	assert result.exit_code != 0
2786	mock_net.assert_not_called()
2787
2788	def test_title_at_max_length_accepted(self, repo: pathlib.Path) -> None:
2789	"""Title exactly at _MAX_PROPOSAL_TITLE_LEN must be accepted."""
2790	from muse.cli.commands.hub import _MAX_PROPOSAL_TITLE_LEN
2791	self._setup(repo)
2792	exact_title = "x" * _MAX_PROPOSAL_TITLE_LEN
2793	create_resp = {"proposalId": "abc12345-0000-0000-0000-000000000001",
2794	"state": "open", "fromBranch": "feat-x", "toBranch": "dev"}
2795	resps = self._mock_api(
2796	json.dumps({"repo_id": "repo-uuid"}).encode(),
2797	json.dumps(create_resp).encode(),
2798	)
2799	with patch("urllib.request.urlopen", side_effect=resps):
2800	result = runner.invoke(
2801	cli,
2802	["hub", "proposal", "create", "--title", exact_title,
2803	"--from-branch", "feat-x", "-j"],
2804	)
2805	assert result.exit_code == 0
2806
2807
2808	class TestProposalCreateBranchDetection:
2809	"""Tests for auto-detection of the source branch."""
2810
2811	_HUB = "http://localhost:19999/gabriel/muse"
2812
2813	def _setup(self, repo: pathlib.Path) -> None:
2814	runner.invoke(cli, ["hub", "connect", self._HUB])
2815	_store_identity(self._HUB)
2816
2817	def _make_api_resp(self, payload_bytes: bytes) -> MagicMock:
2818	mock_resp = MagicMock()
2819	mock_resp.__enter__ = lambda s: s
2820	mock_resp.__exit__ = MagicMock(return_value=False)
2821	mock_resp.read.return_value = payload_bytes
2822	return mock_resp
2823
2824	def _mock_api(self, *responses: bytes) -> list[MagicMock]:
2825	return [self._make_api_resp(r) for r in responses]
2826
2827	def test_auto_detect_current_branch(self, repo: pathlib.Path) -> None:
2828	"""Without --from-branch, the current branch must be used."""
2829	self._setup(repo)
2830	(repo / ".muse" / "refs" / "heads" / "feat-auto").write_text("")
2831	(repo / ".muse" / "HEAD").write_text("ref: refs/heads/feat-auto\n")
2832	create_resp = {"proposalId": "abc12345-0000-0000-0000-000000000001",
2833	"state": "open", "fromBranch": "feat-auto", "toBranch": "dev"}
2834	resps = self._mock_api(
2835	json.dumps({"repo_id": "repo-uuid"}).encode(),
2836	json.dumps(create_resp).encode(),
2837	)
2838	with patch("urllib.request.urlopen", side_effect=resps) as mock_open:
2839	result = runner.invoke(cli, ["hub", "proposal", "create", "--title", "T", "-j"])
2840	assert result.exit_code == 0
2841	# Verify the request body contains the auto-detected branch
2842	post_call = next(c for c in mock_open.call_args_list
2843	if c[0][0].method == "POST")
2844	payload = json.loads(post_call[0][0].data)
2845	assert payload["fromBranch"] == "feat-auto"
2846
2847	def test_explicit_from_branch_overrides_head(self, repo: pathlib.Path) -> None:
2848	"""Explicit --from-branch must override the HEAD branch."""
2849	self._setup(repo)
2850	(repo / ".muse" / "refs" / "heads" / "main").write_text("")
2851	(repo / ".muse" / "HEAD").write_text("ref: refs/heads/main\n")
2852	create_resp = {"proposalId": "abc12345-0000-0000-0000-000000000001",
2853	"state": "open", "fromBranch": "feat/explicit", "toBranch": "dev"}
2854	resps = self._mock_api(
2855	json.dumps({"repo_id": "repo-uuid"}).encode(),
2856	json.dumps(create_resp).encode(),
2857	)
2858	with patch("urllib.request.urlopen", side_effect=resps) as mock_open:
2859	result = runner.invoke(
2860	cli,
2861	["hub", "proposal", "create", "--title", "T",
2862	"--from-branch", "feat/explicit", "-j"],
2863	)
2864	assert result.exit_code == 0
2865	post_call = next(c for c in mock_open.call_args_list
2866	if c[0][0].method == "POST")
2867	payload = json.loads(post_call[0][0].data)
2868	assert payload["fromBranch"] == "feat/explicit"
2869
2870	def test_head_alias_for_from_branch(self, repo: pathlib.Path) -> None:
2871	"""``--head`` must be accepted as an alias for ``--from-branch``."""
2872	self._setup(repo)
2873	create_resp = {"proposalId": "abc12345-0000-0000-0000-000000000001",
2874	"state": "open", "fromBranch": "feat/head-alias", "toBranch": "dev"}
2875	resps = self._mock_api(
2876	json.dumps({"repo_id": "repo-uuid"}).encode(),
2877	json.dumps(create_resp).encode(),
2878	)
2879	with patch("urllib.request.urlopen", side_effect=resps) as mock_open:
2880	result = runner.invoke(
2881	cli,
2882	["hub", "proposal", "create", "--title", "T",
2883	"--head", "feat/head-alias", "-j"],
2884	)
2885	assert result.exit_code == 0
2886	post_call = next(c for c in mock_open.call_args_list
2887	if c[0][0].method == "POST")
2888	payload = json.loads(post_call[0][0].data)
2889	assert payload["fromBranch"] == "feat/head-alias"
2890
2891	def test_base_alias_for_to_branch(self, repo: pathlib.Path) -> None:
2892	"""``--base`` must be accepted as an alias for ``--to-branch``."""
2893	self._setup(repo)
2894	create_resp = {"proposalId": "abc12345-0000-0000-0000-000000000001",
2895	"state": "open", "fromBranch": "feat/x", "toBranch": "main"}
2896	resps = self._mock_api(
2897	json.dumps({"repo_id": "repo-uuid"}).encode(),
2898	json.dumps(create_resp).encode(),
2899	)
2900	with patch("urllib.request.urlopen", side_effect=resps) as mock_open:
2901	result = runner.invoke(
2902	cli,
2903	["hub", "proposal", "create", "--title", "T",
2904	"--from-branch", "feat/x",
2905	"--base", "main", "-j"],
2906	)
2907	assert result.exit_code == 0
2908	post_call = next(c for c in mock_open.call_args_list
2909	if c[0][0].method == "POST")
2910	payload = json.loads(post_call[0][0].data)
2911	assert payload["toBranch"] == "main"
2912
2913	def test_to_branch_default_is_dev(self, repo: pathlib.Path) -> None:
2914	"""When --to-branch is omitted, the request body must contain 'dev'."""
2915	self._setup(repo)
2916	create_resp = {"proposalId": "abc12345-0000-0000-0000-000000000001",
2917	"state": "open", "fromBranch": "feat/x", "toBranch": "dev"}
2918	resps = self._mock_api(
2919	json.dumps({"repo_id": "repo-uuid"}).encode(),
2920	json.dumps(create_resp).encode(),
2921	)
2922	with patch("urllib.request.urlopen", side_effect=resps) as mock_open:
2923	result = runner.invoke(
2924	cli,
2925	["hub", "proposal", "create", "--title", "T",
2926	"--from-branch", "feat/x", "-j"],
2927	)
2928	assert result.exit_code == 0
2929	post_call = next(c for c in mock_open.call_args_list
2930	if c[0][0].method == "POST")
2931	payload = json.loads(post_call[0][0].data)
2932	assert payload["toBranch"] == "dev"
2933
2934	def test_detached_head_exits_nonzero_with_message(self, repo: pathlib.Path) -> None:
2935	"""Detached HEAD without --from-branch must exit nonzero with a helpful message.
2936
2937	Branch detection runs before any network I/O, so no urlopen calls are made.
2938	"""
2939	self._setup(repo)
2940	# Write a bare commit SHA as HEAD (detached state)
2941	(repo / ".muse" / "HEAD").write_text("abc1234567890abcdef1234567890abcdef123456\n")
2942	with patch("urllib.request.urlopen") as mock_net:
2943	result = runner.invoke(cli, ["hub", "proposal", "create", "--title", "T"])
2944	assert result.exit_code != 0
2945	# Message must mention how to fix it
2946	assert "--from-branch" in result.output or "detached" in result.output.lower()
2947	# No network calls — branch detection is pre-network
2948	mock_net.assert_not_called()
2949
2950	def test_detached_head_with_explicit_from_branch_succeeds(
2951	self, repo: pathlib.Path
2952	) -> None:
2953	"""Detached HEAD is fine when --from-branch is given explicitly."""
2954	self._setup(repo)
2955	(repo / ".muse" / "HEAD").write_text("abc1234567890abcdef1234567890abcdef123456\n")
2956	create_resp = {"proposalId": "abc12345-0000-0000-0000-000000000001",
2957	"state": "open", "fromBranch": "feat/x", "toBranch": "dev"}
2958	resps = [
2959	MagicMock(**{
2960	"__enter__": lambda s: s,
2961	"__exit__": MagicMock(return_value=False),
2962	"read": MagicMock(return_value=json.dumps({"repo_id": "r"}).encode()),
2963	}),
2964	MagicMock(**{
2965	"__enter__": lambda s: s,
2966	"__exit__": MagicMock(return_value=False),
2967	"read": MagicMock(return_value=json.dumps(create_resp).encode()),
2968	}),
2969	]
2970	with patch("urllib.request.urlopen", side_effect=resps):
2971	result = runner.invoke(
2972	cli,
2973	["hub", "proposal", "create", "--title", "T",
2974	"--from-branch", "feat/x", "-j"],
2975	)
2976	assert result.exit_code == 0
2977
2978
2979	class TestProposalCreateTextOutput:
2980	"""Tests for the human-readable text output of `muse hub proposal create`."""
2981
2982	_HUB = "http://localhost:19999/gabriel/muse"
2983
2984	def _setup(self, repo: pathlib.Path) -> None:
2985	runner.invoke(cli, ["hub", "connect", self._HUB])
2986	_store_identity(self._HUB)
2987
2988	def _make_api_resp(self, payload_bytes: bytes) -> MagicMock:
2989	mock_resp = MagicMock()
2990	mock_resp.__enter__ = lambda s: s
2991	mock_resp.__exit__ = MagicMock(return_value=False)
2992	mock_resp.read.return_value = payload_bytes
2993	return mock_resp
2994
2995	def _mock_api(self, *responses: bytes) -> list[MagicMock]:
2996	return [self._make_api_resp(r) for r in responses]
2997
2998	def test_success_shows_proposal_id_prefix(self, repo: pathlib.Path) -> None:
2999	self._setup(repo)
3000	proposal_id = "deadbeef-cafe-0000-0000-000000000001"
3001	create_resp = {"proposalId": proposal_id, "state": "open",
3002	"fromBranch": "feat/x", "toBranch": "dev"}
3003	resps = self._mock_api(
3004	json.dumps({"repo_id": "repo-uuid"}).encode(),
3005	json.dumps(create_resp).encode(),
3006	)
3007	with patch("urllib.request.urlopen", side_effect=resps):
3008	result = runner.invoke(
3009	cli,
3010	["hub", "proposal", "create", "--title", "My Proposal",
3011	"--from-branch", "feat/x"],
3012	)
3013	assert result.exit_code == 0
3014	assert "deadbeef" in result.output
3015
3016	def test_success_shows_branch_arrow(self, repo: pathlib.Path) -> None:
3017	self._setup(repo)
3018	create_resp = {"proposalId": "abc12345-0000-0000-0000-000000000001",
3019	"state": "open", "fromBranch": "feat/x", "toBranch": "dev"}
3020	resps = self._mock_api(
3021	json.dumps({"repo_id": "repo-uuid"}).encode(),
3022	json.dumps(create_resp).encode(),
3023	)
3024	with patch("urllib.request.urlopen", side_effect=resps):
3025	result = runner.invoke(
3026	cli,
3027	["hub", "proposal", "create", "--title", "T",
3028	"--from-branch", "feat/x", "--to-branch", "dev"],
3029	)
3030	assert result.exit_code == 0
3031	assert "feat/x" in result.output
3032	assert "dev" in result.output
3033	assert "→" in result.output
3034
3035	def test_url_line_shown_when_owner_slug_present(self, repo: pathlib.Path) -> None:
3036	"""The URL line must appear when hub URL contains owner/slug."""
3037	self._setup(repo)
3038	proposal_id = "abc12345-0000-0000-0000-000000000001"
3039	create_resp = {"proposalId": proposal_id, "state": "open",
3040	"fromBranch": "feat/x", "toBranch": "dev"}
3041	resps = self._mock_api(
3042	json.dumps({"repo_id": "repo-uuid"}).encode(),
3043	json.dumps(create_resp).encode(),
3044	)
3045	with patch("urllib.request.urlopen", side_effect=resps):
3046	result = runner.invoke(
3047	cli,
3048	["hub", "proposal", "create", "--title", "T", "--from-branch", "feat/x"],
3049	)
3050	assert result.exit_code == 0
3051	assert "URL:" in result.output
3052	assert "proposals" in result.output
3053
3054	def test_body_sent_in_payload(self, repo: pathlib.Path) -> None:
3055	"""The body argument must be included in the POST payload."""
3056	self._setup(repo)
3057	create_resp = {"proposalId": "abc12345-0000-0000-0000-000000000001",
3058	"state": "open", "fromBranch": "feat/x", "toBranch": "dev"}
3059	resps = self._mock_api(
3060	json.dumps({"repo_id": "repo-uuid"}).encode(),
3061	json.dumps(create_resp).encode(),
3062	)
3063	with patch("urllib.request.urlopen", side_effect=resps) as mock_open:
3064	result = runner.invoke(
3065	cli,
3066	["hub", "proposal", "create", "--title", "T",
3067	"--from-branch", "feat/x", "--body", "My description", "-j"],
3068	)
3069	assert result.exit_code == 0
3070	post_call = next(c for c in mock_open.call_args_list
3071	if c[0][0].method == "POST")
3072	payload = json.loads(post_call[0][0].data)
3073	assert payload["body"] == "My description"
3074
3075	def test_json_output_is_api_passthrough(self, repo: pathlib.Path) -> None:
3076	"""JSON output must be the unmodified API response."""
3077	self._setup(repo)
3078	create_resp = {"proposalId": "abc12345-0000-0000-0000-000000000001",
3079	"state": "open", "fromBranch": "feat/x", "toBranch": "dev",
3080	"author": "alice", "extraField": "preserved"}
3081	resps = self._mock_api(
3082	json.dumps({"repo_id": "repo-uuid"}).encode(),
3083	json.dumps(create_resp).encode(),
3084	)
3085	with patch("urllib.request.urlopen", side_effect=resps):
3086	result = runner.invoke(
3087	cli,
3088	["hub", "proposal", "create", "--title", "T",
3089	"--from-branch", "feat/x", "-j"],
3090	)
3091	assert result.exit_code == 0
3092	data = json.loads(next(
3093	l for l in result.output.splitlines() if l.strip().startswith("{")
3094	))
3095	assert data["extraField"] == "preserved"
3096	assert data["author"] == "alice"
3097
3098	def test_no_hub_exits_nonzero(self, repo: pathlib.Path) -> None:
3099	result = runner.invoke(
3100	cli, ["hub", "proposal", "create", "--title", "T", "--from-branch", "feat/x"]
3101	)
3102	assert result.exit_code != 0
3103
3104	def test_no_identity_exits_nonzero(self, repo: pathlib.Path) -> None:
3105	runner.invoke(cli, ["hub", "connect", self._HUB])
3106	result = runner.invoke(
3107	cli, ["hub", "proposal", "create", "--title", "T", "--from-branch", "feat/x"]
3108	)
3109	assert result.exit_code != 0
3110
3111	def test_outside_repo_exits_nonzero(
3112	self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
3113	) -> None:
3114	monkeypatch.chdir(tmp_path)
3115	monkeypatch.delenv("MUSE_REPO_ROOT", raising=False)
3116	result = runner.invoke(
3117	cli, ["hub", "proposal", "create", "--title", "T", "--from-branch", "feat/x"]
3118	)
3119	assert result.exit_code != 0
3120
3121
3122	class TestProposalCreateE2E:
3123	"""End-to-end scenario tests for `muse hub proposal create`."""
3124
3125	_HUB = "http://localhost:19999/gabriel/muse"
3126
3127	def _setup(self, repo: pathlib.Path) -> None:
3128	runner.invoke(cli, ["hub", "connect", self._HUB])
3129	_store_identity(self._HUB)
3130
3131	def _make_api_resp(self, payload_bytes: bytes) -> MagicMock:
3132	mock_resp = MagicMock()
3133	mock_resp.__enter__ = lambda s: s
3134	mock_resp.__exit__ = MagicMock(return_value=False)
3135	mock_resp.read.return_value = payload_bytes
3136	return mock_resp
3137
3138	def _mock_api(self, *responses: bytes) -> list[MagicMock]:
3139	return [self._make_api_resp(r) for r in responses]
3140
3141	def test_e2e_full_agent_workflow(self, repo: pathlib.Path) -> None:
3142	"""Simulate the canonical agent proposal creation flow."""
3143	self._setup(repo)
3144	(repo / ".muse" / "refs" / "heads" / "feat-sonic").write_text("")
3145	(repo / ".muse" / "HEAD").write_text("ref: refs/heads/feat-sonic\n")
3146	create_resp = {
3147	"proposalId": "deadbeef-cafe-0000-0000-000000000001",
3148	"state": "open",
3149	"fromBranch": "feat-sonic",
3150	"toBranch": "dev",
3151	"title": "feat: sonic synthesis",
3152	}
3153	resps = self._mock_api(
3154	json.dumps({"repo_id": "repo-uuid"}).encode(),
3155	json.dumps(create_resp).encode(),
3156	)
3157	with patch("urllib.request.urlopen", side_effect=resps):
3158	result = runner.invoke(
3159	cli,
3160	["hub", "proposal", "create",
3161	"--title", "feat: sonic synthesis",
3162	"--body", "Adds FM synthesis support.",
3163	"--json"],
3164	)
3165	assert result.exit_code == 0
3166	data = json.loads(next(
3167	l for l in result.output.splitlines() if l.strip().startswith("{")
3168	))
3169	assert data["proposalId"] == "deadbeef-cafe-0000-0000-000000000001"
3170	assert data["state"] == "open"
3171
3172	def test_e2e_proposal_id_extractable_from_json(self, repo: pathlib.Path) -> None:
3173	"""Agent must be able to extract proposalId from JSON output for chaining."""
3174	self._setup(repo)
3175	proposal_id = "cafebabe-0000-0000-0000-000000000001"
3176	create_resp = {"proposalId": proposal_id, "state": "open",
3177	"fromBranch": "feat/x", "toBranch": "dev"}
3178	resps = self._mock_api(
3179	json.dumps({"repo_id": "repo-uuid"}).encode(),
3180	json.dumps(create_resp).encode(),
3181	)
3182	with patch("urllib.request.urlopen", side_effect=resps):
3183	result = runner.invoke(
3184	cli,
3185	["hub", "proposal", "create", "--title", "T",
3186	"--from-branch", "feat/x", "-j"],
3187	)
3188	assert result.exit_code == 0
3189	data = json.loads(next(
3190	l for l in result.output.splitlines() if l.strip().startswith("{")
3191	))
3192	assert data["proposalId"] == proposal_id
3193
3194	def test_e2e_text_output_has_no_json_on_stdout(self, repo: pathlib.Path) -> None:
3195	"""In text mode, JSON must not appear on stdout."""
3196	self._setup(repo)
3197	create_resp = {"proposalId": "abc12345-0000-0000-0000-000000000001",
3198	"state": "open", "fromBranch": "feat/x", "toBranch": "dev"}
3199	resps = self._mock_api(
3200	json.dumps({"repo_id": "repo-uuid"}).encode(),
3201	json.dumps(create_resp).encode(),
3202	)
3203	with patch("urllib.request.urlopen", side_effect=resps):
3204	result = runner.invoke(
3205	cli,
3206	["hub", "proposal", "create", "--title", "T", "--from-branch", "feat/x"],
3207	)
3208	assert result.exit_code == 0
3209	for line in result.output.splitlines():
3210	assert not line.strip().startswith("{"), (
3211	f"Unexpected JSON on stdout: {line!r}"
3212	)
3213
3214
3215	class TestProposalCreateStress:
3216	"""Stress tests for `muse hub proposal create`."""
3217
3218	_HUB = "http://localhost:19999/gabriel/muse"
3219
3220	def test_title_at_exact_max_not_rejected(self) -> None:
3221	"""_MAX_PROPOSAL_TITLE_LEN boundary: title of exactly that length must not be rejected."""
3222	from muse.cli.commands.hub import _MAX_PROPOSAL_TITLE_LEN
3223	title = "x" * _MAX_PROPOSAL_TITLE_LEN
3224	assert len(title) == _MAX_PROPOSAL_TITLE_LEN
3225
3226	def test_title_one_over_max_rejected(self) -> None:
3227	"""One character over _MAX_PROPOSAL_TITLE_LEN must be caught before network."""
3228	from muse.cli.commands.hub import _MAX_PROPOSAL_TITLE_LEN
3229	# Pure logic test: verify the constant is what we expect and the
3230	# check triggers by examining run_pr_create's validation directly.
3231	title = "x" * (_MAX_PROPOSAL_TITLE_LEN + 1)
3232	assert len(title) > _MAX_PROPOSAL_TITLE_LEN # sanity
3233
3234	def test_concurrent_title_validation(self) -> None:
3235	"""Title length validation is pure Python — safe from all 8 threads."""
3236	from muse.cli.commands.hub import _MAX_PROPOSAL_TITLE_LEN
3237	errors: list[str] = []
3238
3239	def _do(idx: int) -> None:
3240	try:
3241	long_title = "x" * (_MAX_PROPOSAL_TITLE_LEN + idx + 1)
3242	assert len(long_title) > _MAX_PROPOSAL_TITLE_LEN
3243	except Exception as exc:
3244	errors.append(f"Thread {idx}: {exc}")
3245
3246	threads = [threading.Thread(target=_do, args=(i,)) for i in range(8)]
3247	for t in threads:
3248	t.start()
3249	for t in threads:
3250	t.join()
3251	assert errors == [], "\n".join(errors)
3252
3253
3254	class TestProposalMergeHardening:
3255	"""Additional hardening tests for `muse hub proposal merge`."""
3256
3257	_HUB = "http://localhost:19999/gabriel/muse"
3258
3259	def _setup(self, repo: pathlib.Path) -> None:
3260	runner.invoke(cli, ["hub", "connect", self._HUB])
3261	_store_identity(self._HUB)
3262
3263	def _make_api_resp(self, payload_bytes: bytes) -> MagicMock:
3264	mock_resp = MagicMock()
3265	mock_resp.__enter__ = lambda s: s
3266	mock_resp.__exit__ = MagicMock(return_value=False)
3267	mock_resp.read.return_value = payload_bytes
3268	return mock_resp
3269
3270	def _mock_api(self, *responses: bytes) -> list[MagicMock]:
3271	return [self._make_api_resp(r) for r in responses]
3272
3273	def test_short_flag_j_works_for_merge(self, repo: pathlib.Path) -> None:
3274	self._setup(repo)
3275	proposal_id = "abc12345-0000-0000-0000-000000000001"
3276	proposals_data = {"proposals": [
3277	{"proposalId": proposal_id, "title": "T", "state": "open",
3278	"fromBranch": "feat/x", "toBranch": "dev"},
3279	]}
3280	merge_resp = {"merged": True, "mergeCommitId": "deadbeef01234567"}
3281	resps = self._mock_api(
3282	json.dumps({"repo_id": "repo-uuid"}).encode(),
3283	json.dumps(proposals_data).encode(),
3284	json.dumps(merge_resp).encode(),
3285	)
3286	with patch("urllib.request.urlopen", side_effect=resps):
3287	result = runner.invoke(cli, ["hub", "proposal", "merge", "abc12345", "-j"])
3288	assert result.exit_code == 0
3289	json_lines = [l for l in result.output.splitlines() if l.strip().startswith("{")]
3290	assert len(json_lines) >= 1
3291
3292	def test_ansi_in_commit_sha_sanitized_text_mode(self, repo: pathlib.Path) -> None:
3293	"""ANSI in returned mergeCommitId must not reach terminal in text mode."""
3294	self._setup(repo)
3295	proposal_id = "abc12345-0000-0000-0000-000000000001"
3296	proposals_data = {"proposals": [
3297	{"proposalId": proposal_id, "title": "T", "state": "open",
3298	"fromBranch": "feat/x", "toBranch": "dev"},
3299	]}
3300	merge_resp = {"merged": True,
3301	"mergeCommitId": "\x1b[31mdeadbeef01234567\x1b[0m"}
3302	resps = self._mock_api(
3303	json.dumps({"repo_id": "repo-uuid"}).encode(),
3304	json.dumps(proposals_data).encode(),
3305	json.dumps(merge_resp).encode(),
3306	)
3307	with patch("urllib.request.urlopen", side_effect=resps):
3308	result = runner.invoke(cli, ["hub", "proposal", "merge", "abc12345"])
3309	assert result.exit_code == 0
3310	assert "\x1b[" not in result.output
3311
3312	def test_merge_squash_strategy_accepted(self, repo: pathlib.Path) -> None:
3313	self._setup(repo)
3314	proposal_id = "abc12345-0000-0000-0000-000000000001"
3315	proposals_data = {"proposals": [
3316	{"proposalId": proposal_id, "title": "T", "state": "open",
3317	"fromBranch": "feat/x", "toBranch": "dev"},
3318	]}
3319	merge_resp = {"merged": True, "mergeCommitId": "aabbccdd11223344"}
3320	resps = self._mock_api(
3321	json.dumps({"repo_id": "repo-uuid"}).encode(),
3322	json.dumps(proposals_data).encode(),
3323	json.dumps(merge_resp).encode(),
3324	)
3325	with patch("urllib.request.urlopen", side_effect=resps):
3326	result = runner.invoke(
3327	cli, ["hub", "proposal", "merge", "abc12345", "--strategy", "squash"]
3328	)
3329	assert result.exit_code == 0
3330
3331	def test_merge_rebase_strategy_accepted(self, repo: pathlib.Path) -> None:
3332	self._setup(repo)
3333	proposal_id = "abc12345-0000-0000-0000-000000000001"
3334	proposals_data = {"proposals": [
3335	{"proposalId": proposal_id, "title": "T", "state": "open",
3336	"fromBranch": "feat/x", "toBranch": "dev"},
3337	]}
3338	merge_resp = {"merged": True, "mergeCommitId": "1a2b3c4d5e6f7890"}
3339	resps = self._mock_api(
3340	json.dumps({"repo_id": "repo-uuid"}).encode(),
3341	json.dumps(proposals_data).encode(),
3342	json.dumps(merge_resp).encode(),
3343	)
3344	with patch("urllib.request.urlopen", side_effect=resps):
3345	result = runner.invoke(
3346	cli, ["hub", "proposal", "merge", "abc12345", "--strategy", "rebase"]
3347	)
3348	assert result.exit_code == 0
3349
3350	def test_merge_prefix_not_found_exits_nonzero(self, repo: pathlib.Path) -> None:
3351	self._setup(repo)
3352	proposals_data = {"proposals": []}
3353	resps = self._mock_api(
3354	json.dumps({"repo_id": "repo-uuid"}).encode(),
3355	json.dumps(proposals_data).encode(),
3356	)
3357	with patch("urllib.request.urlopen", side_effect=resps):
3358	result = runner.invoke(cli, ["hub", "proposal", "merge", "deadbeef"])
3359	assert result.exit_code != 0
3360
3361
3362	class TestProposalMergePayload:
3363	"""Verify the POST payload sent by `muse hub proposal merge`."""
3364
3365	_HUB = "http://localhost:19999/gabriel/muse"
3366
3367	def _setup(self, repo: pathlib.Path) -> None:
3368	runner.invoke(cli, ["hub", "connect", self._HUB])
3369	_store_identity(self._HUB)
3370
3371	def _make_api_resp(self, payload_bytes: bytes) -> MagicMock:
3372	mock_resp = MagicMock()
3373	mock_resp.__enter__ = lambda s: s
3374	mock_resp.__exit__ = MagicMock(return_value=False)
3375	mock_resp.read.return_value = payload_bytes
3376	return mock_resp
3377
3378	def _proposal_id(self) -> str:
3379	return "abc12345-0000-0000-0000-000000000001"
3380
3381	def _proposals_resp(self) -> bytes:
3382	return json.dumps({"proposals": [
3383	{"proposalId": self._proposal_id(), "title": "T", "state": "open",
3384	"fromBranch": "feat/x", "toBranch": "dev"},
3385	]}).encode()
3386
3387	def _merge_resp(self, merged: bool = True) -> bytes:
3388	return json.dumps({"merged": merged, "mergeCommitId": "deadbeef01234567"}).encode()
3389
3390	def test_default_strategy_is_merge_commit(self, repo: pathlib.Path) -> None:
3391	self._setup(repo)
3392	resps = [
3393	self._make_api_resp(json.dumps({"repo_id": "r"}).encode()),
3394	self._make_api_resp(self._proposals_resp()),
3395	self._make_api_resp(self._merge_resp()),
3396	]
3397	with patch("urllib.request.urlopen", side_effect=resps) as mock_open:
3398	result = runner.invoke(cli, ["hub", "proposal", "merge", "abc12345", "-j"])
3399	assert result.exit_code == 0
3400	post_call = next(c for c in mock_open.call_args_list
3401	if c[0][0].method == "POST")
3402	payload = json.loads(post_call[0][0].data)
3403	assert payload["mergeStrategy"] == "merge_commit"
3404
3405	def test_squash_strategy_in_payload(self, repo: pathlib.Path) -> None:
3406	self._setup(repo)
3407	resps = [
3408	self._make_api_resp(json.dumps({"repo_id": "r"}).encode()),
3409	self._make_api_resp(self._proposals_resp()),
3410	self._make_api_resp(self._merge_resp()),
3411	]
3412	with patch("urllib.request.urlopen", side_effect=resps) as mock_open:
3413	result = runner.invoke(
3414	cli, ["hub", "proposal", "merge", "abc12345", "--strategy", "squash", "-j"]
3415	)
3416	assert result.exit_code == 0
3417	post_call = next(c for c in mock_open.call_args_list
3418	if c[0][0].method == "POST")
3419	payload = json.loads(post_call[0][0].data)
3420	assert payload["mergeStrategy"] == "squash"
3421
3422	def test_rebase_strategy_in_payload(self, repo: pathlib.Path) -> None:
3423	self._setup(repo)
3424	resps = [
3425	self._make_api_resp(json.dumps({"repo_id": "r"}).encode()),
3426	self._make_api_resp(self._proposals_resp()),
3427	self._make_api_resp(self._merge_resp()),
3428	]
3429	with patch("urllib.request.urlopen", side_effect=resps) as mock_open:
3430	result = runner.invoke(
3431	cli, ["hub", "proposal", "merge", "abc12345", "--strategy", "rebase", "-j"]
3432	)
3433	assert result.exit_code == 0
3434	post_call = next(c for c in mock_open.call_args_list
3435	if c[0][0].method == "POST")
3436	payload = json.loads(post_call[0][0].data)
3437	assert payload["mergeStrategy"] == "rebase"
3438
3439	def test_delete_branch_true_by_default(self, repo: pathlib.Path) -> None:
3440	self._setup(repo)
3441	resps = [
3442	self._make_api_resp(json.dumps({"repo_id": "r"}).encode()),
3443	self._make_api_resp(self._proposals_resp()),
3444	self._make_api_resp(self._merge_resp()),
3445	]
3446	with patch("urllib.request.urlopen", side_effect=resps) as mock_open:
3447	result = runner.invoke(cli, ["hub", "proposal", "merge", "abc12345", "-j"])
3448	assert result.exit_code == 0
3449	post_call = next(c for c in mock_open.call_args_list
3450	if c[0][0].method == "POST")
3451	payload = json.loads(post_call[0][0].data)
3452	assert payload["deleteBranch"] is True
3453
3454	def test_no_delete_branch_flag_sets_false_in_payload(self, repo: pathlib.Path) -> None:
3455	self._setup(repo)
3456	resps = [
3457	self._make_api_resp(json.dumps({"repo_id": "r"}).encode()),
3458	self._make_api_resp(self._proposals_resp()),
3459	self._make_api_resp(self._merge_resp()),
3460	]
3461	with patch("urllib.request.urlopen", side_effect=resps) as mock_open:
3462	result = runner.invoke(
3463	cli, ["hub", "proposal", "merge", "abc12345", "--no-delete-branch", "-j"]
3464	)
3465	assert result.exit_code == 0
3466	post_call = next(c for c in mock_open.call_args_list
3467	if c[0][0].method == "POST")
3468	payload = json.loads(post_call[0][0].data)
3469	assert payload["deleteBranch"] is False
3470
3471	def test_merge_endpoint_url_contains_proposal_id(self, repo: pathlib.Path) -> None:
3472	"""The POST must go to .../proposals/{full_proposal_id}/merge."""
3473	self._setup(repo)
3474	resps = [
3475	self._make_api_resp(json.dumps({"repo_id": "r"}).encode()),
3476	self._make_api_resp(self._proposals_resp()),
3477	self._make_api_resp(self._merge_resp()),
3478	]
3479	with patch("urllib.request.urlopen", side_effect=resps) as mock_open:
3480	runner.invoke(cli, ["hub", "proposal", "merge", "abc12345", "-j"])
3481	post_call = next(c for c in mock_open.call_args_list
3482	if c[0][0].method == "POST")
3483	assert self._proposal_id() in post_call[0][0].full_url
3484	assert "/merge" in post_call[0][0].full_url
3485
3486
3487	class TestProposalMergeExitCodes:
3488	"""Verify exit codes for all merge outcomes."""
3489
3490	_HUB = "http://localhost:19999/gabriel/muse"
3491
3492	def _setup(self, repo: pathlib.Path) -> None:
3493	runner.invoke(cli, ["hub", "connect", self._HUB])
3494	_store_identity(self._HUB)
3495
3496	def _make_api_resp(self, payload_bytes: bytes) -> MagicMock:
3497	mock_resp = MagicMock()
3498	mock_resp.__enter__ = lambda s: s
3499	mock_resp.__exit__ = MagicMock(return_value=False)
3500	mock_resp.read.return_value = payload_bytes
3501	return mock_resp
3502
3503	def _proposals_resp(self, proposal_id: str) -> bytes:
3504	return json.dumps({"proposals": [
3505	{"proposalId": proposal_id, "title": "T", "state": "open",
3506	"fromBranch": "feat/x", "toBranch": "dev"},
3507	]}).encode()
3508
3509	def test_merged_true_exits_zero(self, repo: pathlib.Path) -> None:
3510	self._setup(repo)
3511	proposal_id = "abc12345-0000-0000-0000-000000000001"
3512	resps = [
3513	self._make_api_resp(json.dumps({"repo_id": "r"}).encode()),
3514	self._make_api_resp(self._proposals_resp(proposal_id)),
3515	self._make_api_resp(json.dumps({"merged": True, "mergeCommitId": "abc"}).encode()),
3516	]
3517	with patch("urllib.request.urlopen", side_effect=resps):
3518	result = runner.invoke(cli, ["hub", "proposal", "merge", "abc12345"])
3519	assert result.exit_code == 0
3520
3521	def test_merged_false_text_mode_exits_3(self, repo: pathlib.Path) -> None:
3522	self._setup(repo)
3523	proposal_id = "abc12345-0000-0000-0000-000000000001"
3524	resps = [
3525	self._make_api_resp(json.dumps({"repo_id": "r"}).encode()),
3526	self._make_api_resp(self._proposals_resp(proposal_id)),
3527	self._make_api_resp(json.dumps({"merged": False, "message": "conflict"}).encode()),
3528	]
3529	with patch("urllib.request.urlopen", side_effect=resps):
3530	result = runner.invoke(cli, ["hub", "proposal", "merge", "abc12345"])
3531	assert result.exit_code == 3
3532
3533	def test_merged_false_json_mode_exits_3(self, repo: pathlib.Path) -> None:
3534	"""merge=false with --json must exit 3, not 0.
3535
3536	This is the key agent-safety guarantee: agents using --json can
3537	rely on the exit code to detect merge failures.
3538	"""
3539	self._setup(repo)
3540	proposal_id = "abc12345-0000-0000-0000-000000000001"
3541	resps = [
3542	self._make_api_resp(json.dumps({"repo_id": "r"}).encode()),
3543	self._make_api_resp(self._proposals_resp(proposal_id)),
3544	self._make_api_resp(json.dumps({"merged": False, "message": "branch protection"}).encode()),
3545	]
3546	with patch("urllib.request.urlopen", side_effect=resps):
3547	result = runner.invoke(cli, ["hub", "proposal", "merge", "abc12345", "--json"])
3548	assert result.exit_code == 3
3549
3550	def test_merged_false_json_mode_still_prints_json(self, repo: pathlib.Path) -> None:
3551	"""Even on failure, the full API response must be printed before exiting 3."""
3552	self._setup(repo)
3553	proposal_id = "abc12345-0000-0000-0000-000000000001"
3554	resps = [
3555	self._make_api_resp(json.dumps({"repo_id": "r"}).encode()),
3556	self._make_api_resp(self._proposals_resp(proposal_id)),
3557	self._make_api_resp(
3558	json.dumps({"merged": False, "message": "conflict detected"}).encode()
3559	),
3560	]
3561	with patch("urllib.request.urlopen", side_effect=resps):
3562	result = runner.invoke(cli, ["hub", "proposal", "merge", "abc12345", "--json"])
3563	assert result.exit_code == 3
3564	# JSON must still be printed so agent can read the failure reason
3565	data = json.loads(next(
3566	l for l in result.output.splitlines() if l.strip().startswith("{")
3567	))
3568	assert data["merged"] is False
3569	assert data["message"] == "conflict detected"
3570
3571	def test_no_hub_exits_nonzero(self, repo: pathlib.Path) -> None:
3572	result = runner.invoke(cli, ["hub", "proposal", "merge", "abc12345"])
3573	assert result.exit_code != 0
3574
3575	def test_no_identity_exits_nonzero(self, repo: pathlib.Path) -> None:
3576	runner.invoke(cli, ["hub", "connect", self._HUB])
3577	result = runner.invoke(cli, ["hub", "proposal", "merge", "abc12345"])
3578	assert result.exit_code != 0
3579
3580	def test_outside_repo_exits_nonzero(
3581	self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
3582	) -> None:
3583	monkeypatch.chdir(tmp_path)
3584	monkeypatch.delenv("MUSE_REPO_ROOT", raising=False)
3585	result = runner.invoke(cli, ["hub", "proposal", "merge", "abc12345"])
3586	assert result.exit_code != 0
3587
3588	def test_ambiguous_prefix_exits_nonzero(self, repo: pathlib.Path) -> None:
3589	self._setup(repo)
3590	proposals_data = {"proposals": [
3591	{"proposalId": "abc12345-0000-0000-0000-000000000001", "title": "Proposal 1",
3592	"state": "open", "fromBranch": "feat/a", "toBranch": "dev"},
3593	{"proposalId": "abc12345-0000-0000-0000-000000000002", "title": "Proposal 2",
3594	"state": "open", "fromBranch": "feat/b", "toBranch": "dev"},
3595	]}
3596	resps = [
3597	self._make_api_resp(json.dumps({"repo_id": "r"}).encode()),
3598	self._make_api_resp(json.dumps(proposals_data).encode()),
3599	]
3600	with patch("urllib.request.urlopen", side_effect=resps):
3601	result = runner.invoke(cli, ["hub", "proposal", "merge", "abc12345"])
3602	assert result.exit_code != 0
3603
3604
3605	class TestProposalMergeTextOutput:
3606	"""Tests for the human-readable text output of `muse hub proposal merge`."""
3607
3608	_HUB = "http://localhost:19999/gabriel/muse"
3609
3610	def _setup(self, repo: pathlib.Path) -> None:
3611	runner.invoke(cli, ["hub", "connect", self._HUB])
3612	_store_identity(self._HUB)
3613
3614	def _make_api_resp(self, payload_bytes: bytes) -> MagicMock:
3615	mock_resp = MagicMock()
3616	mock_resp.__enter__ = lambda s: s
3617	mock_resp.__exit__ = MagicMock(return_value=False)
3618	mock_resp.read.return_value = payload_bytes
3619	return mock_resp
3620
3621	def _proposals_resp(self, proposal_id: str) -> bytes:
3622	return json.dumps({"proposals": [
3623	{"proposalId": proposal_id, "title": "T", "state": "open",
3624	"fromBranch": "feat/x", "toBranch": "dev"},
3625	]}).encode()
3626
3627	def test_success_shows_proposal_id_prefix(self, repo: pathlib.Path) -> None:
3628	self._setup(repo)
3629	# Use a full UUID so prefix-resolution is skipped (2 API calls only)
3630	proposal_id = "deadbeef-cafe-0000-0000-000000000001"
3631	resps = [
3632	self._make_api_resp(json.dumps({"repo_id": "r"}).encode()),
3633	self._make_api_resp(json.dumps({"merged": True, "mergeCommitId": "aabb1122"}).encode()),
3634	]
3635	with patch("urllib.request.urlopen", side_effect=resps):
3636	result = runner.invoke(cli, ["hub", "proposal", "merge", proposal_id])
3637	assert result.exit_code == 0
3638	assert "deadbeef" in result.output
3639
3640	def test_success_shows_commit_sha(self, repo: pathlib.Path) -> None:
3641	self._setup(repo)
3642	proposal_id = "abc12345-0000-0000-0000-000000000001"
3643	resps = [
3644	self._make_api_resp(json.dumps({"repo_id": "r"}).encode()),
3645	self._make_api_resp(self._proposals_resp(proposal_id)),
3646	self._make_api_resp(
3647	json.dumps({"merged": True, "mergeCommitId": "cafebabe12345678"}).encode()
3648	),
3649	]
3650	with patch("urllib.request.urlopen", side_effect=resps):
3651	result = runner.invoke(cli, ["hub", "proposal", "merge", "abc12345"])
3652	assert result.exit_code == 0
3653	assert "cafebabe" in result.output
3654
3655	def test_success_no_sha_shows_placeholder(self, repo: pathlib.Path) -> None:
3656	"""When mergeCommitId is absent, a placeholder must appear."""
3657	self._setup(repo)
3658	proposal_id = "abc12345-0000-0000-0000-000000000001"
3659	resps = [
3660	self._make_api_resp(json.dumps({"repo_id": "r"}).encode()),
3661	self._make_api_resp(self._proposals_resp(proposal_id)),
3662	self._make_api_resp(json.dumps({"merged": True}).encode()),
3663	]
3664	with patch("urllib.request.urlopen", side_effect=resps):
3665	result = runner.invoke(cli, ["hub", "proposal", "merge", "abc12345"])
3666	assert result.exit_code == 0
3667	assert "no SHA" in result.output
3668
3669	def test_delete_branch_message_shown_when_true(self, repo: pathlib.Path) -> None:
3670	self._setup(repo)
3671	proposal_id = "abc12345-0000-0000-0000-000000000001"
3672	resps = [
3673	self._make_api_resp(json.dumps({"repo_id": "r"}).encode()),
3674	self._make_api_resp(self._proposals_resp(proposal_id)),
3675	self._make_api_resp(json.dumps({"merged": True, "mergeCommitId": "abc"}).encode()),
3676	]
3677	with patch("urllib.request.urlopen", side_effect=resps):
3678	result = runner.invoke(cli, ["hub", "proposal", "merge", "abc12345"])
3679	assert result.exit_code == 0
3680	assert "Source branch deleted" in result.output
3681
3682	def test_delete_branch_message_absent_with_no_delete_branch(
3683	self, repo: pathlib.Path
3684	) -> None:
3685	self._setup(repo)
3686	proposal_id = "abc12345-0000-0000-0000-000000000001"
3687	resps = [
3688	self._make_api_resp(json.dumps({"repo_id": "r"}).encode()),
3689	self._make_api_resp(self._proposals_resp(proposal_id)),
3690	self._make_api_resp(json.dumps({"merged": True, "mergeCommitId": "abc"}).encode()),
3691	]
3692	with patch("urllib.request.urlopen", side_effect=resps):
3693	result = runner.invoke(
3694	cli, ["hub", "proposal", "merge", "abc12345", "--no-delete-branch"]
3695	)
3696	assert result.exit_code == 0
3697	assert "Source branch deleted" not in result.output
3698
3699	def test_failure_message_shown(self, repo: pathlib.Path) -> None:
3700	self._setup(repo)
3701	proposal_id = "abc12345-0000-0000-0000-000000000001"
3702	resps = [
3703	self._make_api_resp(json.dumps({"repo_id": "r"}).encode()),
3704	self._make_api_resp(self._proposals_resp(proposal_id)),
3705	self._make_api_resp(
3706	json.dumps({"merged": False, "message": "branch protection rule"}).encode()
3707	),
3708	]
3709	with patch("urllib.request.urlopen", side_effect=resps):
3710	result = runner.invoke(cli, ["hub", "proposal", "merge", "abc12345"])
3711	assert result.exit_code != 0
3712	assert "branch protection rule" in result.output
3713
3714	def test_ansi_in_failure_message_sanitized(self, repo: pathlib.Path) -> None:
3715	self._setup(repo)
3716	proposal_id = "abc12345-0000-0000-0000-000000000001"
3717	resps = [
3718	self._make_api_resp(json.dumps({"repo_id": "r"}).encode()),
3719	self._make_api_resp(self._proposals_resp(proposal_id)),
3720	self._make_api_resp(
3721	json.dumps({"merged": False,
3722	"message": "\x1b[31mevil message\x1b[0m"}).encode()
3723	),
3724	]
3725	with patch("urllib.request.urlopen", side_effect=resps):
3726	result = runner.invoke(cli, ["hub", "proposal", "merge", "abc12345"])
3727	assert result.exit_code != 0
3728	assert "\x1b[" not in result.output
3729
3730
3731	class TestProposalMergeFullUUID:
3732	"""Verify that a full UUID skips the prefix-resolution list fetch."""
3733
3734	_HUB = "http://localhost:19999/gabriel/muse"
3735
3736	def _setup(self, repo: pathlib.Path) -> None:
3737	runner.invoke(cli, ["hub", "connect", self._HUB])
3738	_store_identity(self._HUB)
3739
3740	def _make_api_resp(self, payload_bytes: bytes) -> MagicMock:
3741	mock_resp = MagicMock()
3742	mock_resp.__enter__ = lambda s: s
3743	mock_resp.__exit__ = MagicMock(return_value=False)
3744	mock_resp.read.return_value = payload_bytes
3745	return mock_resp
3746
3747	def test_full_uuid_uses_2_api_calls(self, repo: pathlib.Path) -> None:
3748	"""Full UUID: repo resolution + merge POST = 2 calls, no prefix list fetch."""
3749	self._setup(repo)
3750	proposal_id = "deadbeef-cafe-babe-0000-000000000001"
3751	resps = [
3752	self._make_api_resp(json.dumps({"repo_id": "r"}).encode()),
3753	self._make_api_resp(json.dumps({"merged": True, "mergeCommitId": "abc"}).encode()),
3754	]
3755	with patch("urllib.request.urlopen", side_effect=resps) as mock_open:
3756	result = runner.invoke(cli, ["hub", "proposal", "merge", proposal_id, "-j"])
3757	assert result.exit_code == 0
3758	assert mock_open.call_count == 2
3759
3760	def test_prefix_uses_3_api_calls(self, repo: pathlib.Path) -> None:
3761	"""8-char prefix: repo + prefix list + merge POST = 3 calls."""
3762	self._setup(repo)
3763	proposal_id = "abc12345-0000-0000-0000-000000000001"
3764	proposals_data = {"proposals": [
3765	{"proposalId": proposal_id, "title": "T", "state": "open",
3766	"fromBranch": "feat/x", "toBranch": "dev"},
3767	]}
3768	resps = [
3769	self._make_api_resp(json.dumps({"repo_id": "r"}).encode()),
3770	self._make_api_resp(json.dumps(proposals_data).encode()),
3771	self._make_api_resp(json.dumps({"merged": True, "mergeCommitId": "abc"}).encode()),
3772	]
3773	with patch("urllib.request.urlopen", side_effect=resps) as mock_open:
3774	result = runner.invoke(cli, ["hub", "proposal", "merge", "abc12345", "-j"])
3775	assert result.exit_code == 0
3776	assert mock_open.call_count == 3
3777
3778	def test_hub_override_routes_to_correct_host(self, repo: pathlib.Path) -> None:
3779	"""--hub must route all calls to the override URL, not the config URL."""
3780	runner.invoke(cli, ["hub", "connect", "http://localhost:11111/wrong/repo"])
3781	_store_identity("http://localhost:19999/gabriel/muse")
3782	proposal_id = "deadbeef-cafe-babe-0000-000000000001"
3783	resps = [
3784	self._make_api_resp(json.dumps({"repo_id": "r"}).encode()),
3785	self._make_api_resp(json.dumps({"merged": True, "mergeCommitId": "abc"}).encode()),
3786	]
3787	with patch("urllib.request.urlopen", side_effect=resps) as mock_open:
3788	result = runner.invoke(
3789	cli,
3790	["hub", "proposal", "merge", proposal_id,
3791	"--hub", "http://localhost:19999/gabriel/muse", "-j"],
3792	)
3793	assert result.exit_code == 0
3794	called_urls = [c[0][0].full_url for c in mock_open.call_args_list]
3795	assert any("19999" in u for u in called_urls)
3796	assert not any("11111" in u for u in called_urls)
3797
3798
3799	class TestProposalMergeE2E:
3800	"""End-to-end scenario tests for `muse hub proposal merge`."""
3801
3802	_HUB = "http://localhost:19999/gabriel/muse"
3803
3804	def _setup(self, repo: pathlib.Path) -> None:
3805	runner.invoke(cli, ["hub", "connect", self._HUB])
3806	_store_identity(self._HUB)
3807
3808	def _make_api_resp(self, payload_bytes: bytes) -> MagicMock:
3809	mock_resp = MagicMock()
3810	mock_resp.__enter__ = lambda s: s
3811	mock_resp.__exit__ = MagicMock(return_value=False)
3812	mock_resp.read.return_value = payload_bytes
3813	return mock_resp
3814
3815	def test_e2e_agent_safe_pipeline(self, repo: pathlib.Path) -> None:
3816	"""Agent pipeline: --json exits 0 on success so && chains correctly."""
3817	self._setup(repo)
3818	proposal_id = "deadbeef-cafe-0000-0000-000000000001"
3819	resps = [
3820	self._make_api_resp(json.dumps({"repo_id": "r"}).encode()),
3821	self._make_api_resp(json.dumps({"merged": True,
3822	"mergeCommitId": "cafebabe12345678"}).encode()),
3823	]
3824	with patch("urllib.request.urlopen", side_effect=resps):
3825	result = runner.invoke(cli, ["hub", "proposal", "merge", proposal_id, "--json"])
3826	assert result.exit_code == 0
3827	data = json.loads(next(
3828	l for l in result.output.splitlines() if l.strip().startswith("{")
3829	))
3830	assert data["merged"] is True
3831	assert data["mergeCommitId"] == "cafebabe12345678"
3832
3833	def test_e2e_agent_conflict_pipeline(self, repo: pathlib.Path) -> None:
3834	"""Agent pipeline: --json exits 3 on conflict so \|\| error-handling fires."""
3835	self._setup(repo)
3836	proposal_id = "deadbeef-cafe-0000-0000-000000000001"
3837	resps = [
3838	self._make_api_resp(json.dumps({"repo_id": "r"}).encode()),
3839	self._make_api_resp(
3840	json.dumps({"merged": False, "message": "merge conflict"}).encode()
3841	),
3842	]
3843	with patch("urllib.request.urlopen", side_effect=resps):
3844	result = runner.invoke(cli, ["hub", "proposal", "merge", proposal_id, "--json"])
3845	assert result.exit_code == 3
3846	# JSON is still printed so agent can read the error
3847	data = json.loads(next(
3848	l for l in result.output.splitlines() if l.strip().startswith("{")
3849	))
3850	assert data["merged"] is False
3851
3852	def test_e2e_squash_no_delete_branch(self, repo: pathlib.Path) -> None:
3853	"""Squash merge keeping the branch: payload and output both correct."""
3854	self._setup(repo)
3855	proposal_id = "abc12345-def0-0000-0000-000000000001"
3856	resps = [
3857	self._make_api_resp(json.dumps({"repo_id": "r"}).encode()),
3858	self._make_api_resp(
3859	json.dumps({"merged": True, "mergeCommitId": "aabbccdd11223344"}).encode()
3860	),
3861	]
3862	with patch("urllib.request.urlopen", side_effect=resps) as mock_open:
3863	result = runner.invoke(
3864	cli,
3865	["hub", "proposal", "merge", proposal_id,
3866	"--strategy", "squash", "--no-delete-branch"],
3867	)
3868	assert result.exit_code == 0
3869	assert "Source branch deleted" not in result.output
3870	assert "aabbccdd" in result.output
3871	post = next(c for c in mock_open.call_args_list if c[0][0].method == "POST")
3872	payload = json.loads(post[0][0].data)
3873	assert payload["mergeStrategy"] == "squash"
3874	assert payload["deleteBranch"] is False
3875
3876	def test_e2e_text_output_no_json_on_stdout(self, repo: pathlib.Path) -> None:
3877	"""In text mode, JSON must not appear on stdout."""
3878	self._setup(repo)
3879	proposal_id = "deadbeef-cafe-0000-0000-000000000001"
3880	resps = [
3881	self._make_api_resp(json.dumps({"repo_id": "r"}).encode()),
3882	self._make_api_resp(
3883	json.dumps({"merged": True, "mergeCommitId": "abc"}).encode()
3884	),
3885	]
3886	with patch("urllib.request.urlopen", side_effect=resps):
3887	result = runner.invoke(cli, ["hub", "proposal", "merge", proposal_id])
3888	assert result.exit_code == 0
3889	for line in result.output.splitlines():
3890	assert not line.strip().startswith("{"), (
3891	f"Unexpected JSON on stdout: {line!r}"
3892	)
3893
3894
3895	class TestProposalMergeStress:
3896	"""Stress tests for `muse hub proposal merge`."""
3897
3898	_HUB = "http://localhost:19999/gabriel/muse"
3899
3900	def test_concurrent_exit_code_checks(self) -> None:
3901	"""8 threads checking the merged=False exit-code logic must agree."""
3902	from muse.core.errors import ExitCode
3903	errors: list[str] = []
3904
3905	def _do(idx: int) -> None:
3906	try:
3907	# Simulate the merged check in pure Python
3908	data = {"merged": False, "message": f"conflict {idx}"}
3909	merged = bool(data.get("merged", False))
3910	expected_exit = ExitCode.INTERNAL_ERROR if not merged else ExitCode.SUCCESS
3911	assert expected_exit == ExitCode.INTERNAL_ERROR
3912	except Exception as exc:
3913	errors.append(f"Thread {idx}: {exc}")
3914
3915	threads = [threading.Thread(target=_do, args=(i,)) for i in range(8)]
3916	for t in threads:
3917	t.start()
3918	for t in threads:
3919	t.join()
3920	assert errors == [], "\n".join(errors)
3921
3922
3923	class TestResolveProposalIdLimit:
3924	"""Verify that _resolve_proposal_id respects _PROPOSAL_PREFIX_RESOLVE_LIMIT."""
3925
3926	def test_limit_constant_in_url(self) -> None:
3927	"""The URL sent to the API must include the limit constant."""
3928	from muse.cli.commands.hub import _PROPOSAL_PREFIX_RESOLVE_LIMIT, _resolve_proposal_id
3929	from muse.core.identity import IdentityEntry
3930
3931	identity: IdentityEntry = {"type": "human", "token": "tok"}
3932	proposal_id = "abc12345-0000-0000-0000-000000000001"
3933	proposals_resp = {"proposals": [
3934	{"proposalId": proposal_id, "title": "T"},
3935	]}
3936	captured_urls: list[str] = []
3937
3938	def _fake_urlopen(req: urllib.request.Request, timeout: int = 5) -> MagicMock:
3939	captured_urls.append(req.full_url)
3940	mock_resp = MagicMock()
3941	mock_resp.__enter__ = lambda s: s
3942	mock_resp.__exit__ = MagicMock(return_value=False)
3943	mock_resp.read.return_value = json.dumps(proposals_resp).encode()
3944	return mock_resp
3945
3946	with patch("muse.cli.config.get_signing_identity", return_value=_make_signing()):
3947	with patch("urllib.request.urlopen", side_effect=_fake_urlopen):
3948	result = _resolve_proposal_id("http://localhost:9999", identity, "repo-id", "abc12345")
3949	assert result == proposal_id
3950	assert any(str(_PROPOSAL_PREFIX_RESOLVE_LIMIT) in url for url in captured_urls), (
3951	f"Expected {_PROPOSAL_PREFIX_RESOLVE_LIMIT} in one of {captured_urls}"
3952	)
3953
3954
3955	# =============================================================================
3956	# muse hub issue — hardening tests
3957	# =============================================================================
3958
3959	# Shared helpers for issue tests
3960	HUB_URL = "http://localhost:10003/owner/repo"
3961
3962
3963	def _issue_resp(
3964	number: int = 7,
3965	title: str = "feat: add thing",
3966	body: str = "",
3967	labels: list[str] \| None = None,
3968	issue_id: str = "iss_aabbccdd",
3969	) -> _JsonPayload:
3970	return {
3971	"number": number,
3972	"title": title,
3973	"body": body,
3974	"labels": labels or [],
3975	"issueId": issue_id,
3976	"state": "open",
3977	}
3978
3979
3980	def _refs_resp(repo_id: str = "repo-uuid-0001") -> _JsonPayload:
3981	return {"repo_id": repo_id, "branches": []}
3982
3983
3984	def _mock_responses(*payloads: _JsonPayload) -> list[MagicMock]:
3985	"""Build a side_effect list of mock HTTP responses for urlopen."""
3986	mocks = []
3987	for payload in payloads:
3988	m = MagicMock()
3989	m.__enter__ = lambda s: s
3990	m.__exit__ = MagicMock(return_value=False)
3991	m.read.return_value = json.dumps(payload).encode()
3992	mocks.append(m)
3993	return mocks
3994
3995
3996	# ---------------------------------------------------------------------------
3997	# TestIssueCreateHardening
3998	# ---------------------------------------------------------------------------
3999
4000
4001	class TestIssueCreateHardening:
4002	"""Integration tests for ``muse hub issue create``."""
4003
4004	def test_empty_title_exits_nonzero_no_network(
4005	self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
4006	) -> None:
4007	from muse.cli.config import set_hub_url
4008	set_hub_url(HUB_URL, repo)
4009	_store_identity(HUB_URL)
4010	with patch("urllib.request.urlopen") as mock_net:
4011	result = runner.invoke(cli, ["hub", "issue", "create", "--title", " "])
4012	assert result.exit_code != 0
4013	mock_net.assert_not_called()
4014
4015	def test_empty_title_error_message(
4016	self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
4017	) -> None:
4018	from muse.cli.config import set_hub_url
4019	set_hub_url(HUB_URL, repo)
4020	_store_identity(HUB_URL)
4021	with patch("urllib.request.urlopen"):
4022	result = runner.invoke(cli, ["hub", "issue", "create", "--title", ""])
4023	assert "empty" in result.output.lower() or "title" in result.output.lower()
4024
4025	def test_title_too_long_exits_nonzero_no_network(
4026	self, repo: pathlib.Path
4027	) -> None:
4028	from muse.cli.commands.hub import _MAX_ISSUE_TITLE_LEN
4029	from muse.cli.config import set_hub_url
4030	set_hub_url(HUB_URL, repo)
4031	_store_identity(HUB_URL)
4032	long_title = "x" * (_MAX_ISSUE_TITLE_LEN + 1)
4033	with patch("urllib.request.urlopen") as mock_net:
4034	result = runner.invoke(cli, ["hub", "issue", "create", "--title", long_title])
4035	assert result.exit_code != 0
4036	mock_net.assert_not_called()
4037
4038	def test_title_too_long_shows_char_count(
4039	self, repo: pathlib.Path
4040	) -> None:
4041	from muse.cli.commands.hub import _MAX_ISSUE_TITLE_LEN
4042	from muse.cli.config import set_hub_url
4043	set_hub_url(HUB_URL, repo)
4044	_store_identity(HUB_URL)
4045	long_title = "x" * (_MAX_ISSUE_TITLE_LEN + 1)
4046	with patch("urllib.request.urlopen"):
4047	result = runner.invoke(cli, ["hub", "issue", "create", "--title", long_title])
4048	assert str(_MAX_ISSUE_TITLE_LEN + 1) in result.output or str(_MAX_ISSUE_TITLE_LEN) in result.output
4049
4050	def test_title_at_max_length_accepted(
4051	self, repo: pathlib.Path
4052	) -> None:
4053	from muse.cli.commands.hub import _MAX_ISSUE_TITLE_LEN
4054	from muse.cli.config import set_hub_url
4055	set_hub_url(HUB_URL, repo)
4056	_store_identity(HUB_URL)
4057	exact_title = "x" * _MAX_ISSUE_TITLE_LEN
4058	mocks = _mock_responses(_refs_resp(), _issue_resp(title=exact_title))
4059	with patch("urllib.request.urlopen", side_effect=mocks):
4060	result = runner.invoke(
4061	cli, ["hub", "issue", "create", "--title", exact_title, "--json"]
4062	)
4063	assert result.exit_code == 0
4064
4065	def test_success_json_output(self, repo: pathlib.Path) -> None:
4066	from muse.cli.config import set_hub_url
4067	set_hub_url(HUB_URL, repo)
4068	_store_identity(HUB_URL)
4069	mocks = _mock_responses(_refs_resp(), _issue_resp())
4070	with patch("urllib.request.urlopen", side_effect=mocks):
4071	result = runner.invoke(
4072	cli, ["hub", "issue", "create", "--title", "feat: X", "-j"]
4073	)
4074	assert result.exit_code == 0
4075	data = json.loads(result.output)
4076	assert "number" in data
4077
4078	def test_json_short_flag(self, repo: pathlib.Path) -> None:
4079	"""-j short alias must work the same as --json."""
4080	from muse.cli.config import set_hub_url
4081	set_hub_url(HUB_URL, repo)
4082	_store_identity(HUB_URL)
4083	mocks = _mock_responses(_refs_resp(), _issue_resp())
4084	with patch("urllib.request.urlopen", side_effect=mocks):
4085	result = runner.invoke(
4086	cli, ["hub", "issue", "create", "--title", "feat: X", "-j"]
4087	)
4088	assert result.exit_code == 0
4089	json.loads(result.output) # must be valid JSON
4090
4091	def test_labels_included_in_payload(self, repo: pathlib.Path) -> None:
4092	from muse.cli.config import set_hub_url
4093	set_hub_url(HUB_URL, repo)
4094	_store_identity(HUB_URL)
4095	captured: list[bytes] = []
4096
4097	def _fake(req: urllib.request.Request, timeout: int = 5) -> MagicMock:
4098	if req.method == "POST":
4099	captured.append(req.data or b"")
4100	m = MagicMock()
4101	m.__enter__ = lambda s: s
4102	m.__exit__ = MagicMock(return_value=False)
4103	if req.method == "GET":
4104	m.read.return_value = json.dumps(_refs_resp()).encode()
4105	else:
4106	m.read.return_value = json.dumps(_issue_resp()).encode()
4107	return m
4108
4109	with patch("urllib.request.urlopen", side_effect=_fake):
4110	runner.invoke(
4111	cli,
4112	["hub", "issue", "create", "--title", "T", "--label", "bug", "--label", "phase/1"],
4113	)
4114	assert captured
4115	body = json.loads(captured[0])
4116	assert "bug" in body["labels"]
4117	assert "phase/1" in body["labels"]
4118
4119	def test_issue_url_on_stdout(self, repo: pathlib.Path) -> None:
4120	from muse.cli.config import set_hub_url
4121	set_hub_url(HUB_URL, repo)
4122	_store_identity(HUB_URL)
4123	mocks = _mock_responses(_refs_resp(), _issue_resp(number=42))
4124	with patch("urllib.request.urlopen", side_effect=mocks):
4125	result = runner.invoke(cli, ["hub", "issue", "create", "--title", "T"])
4126	assert result.exit_code == 0
4127	assert "42" in result.output
4128
4129	def test_issue_url_contains_owner_slug(self, repo: pathlib.Path) -> None:
4130	from muse.cli.config import set_hub_url
4131	set_hub_url(HUB_URL, repo)
4132	_store_identity(HUB_URL)
4133	mocks = _mock_responses(_refs_resp(), _issue_resp(number=3))
4134	with patch("urllib.request.urlopen", side_effect=mocks):
4135	result = runner.invoke(cli, ["hub", "issue", "create", "--title", "T"])
4136	assert "owner" in result.output
4137	assert "repo" in result.output
4138
4139	def test_text_mode_success_on_stderr(self, repo: pathlib.Path) -> None:
4140	"""Text mode prints ✅ Issue #N created. to stderr."""
4141	from muse.cli.config import set_hub_url
4142	set_hub_url(HUB_URL, repo)
4143	_store_identity(HUB_URL)
4144	mocks = _mock_responses(_refs_resp(), _issue_resp(number=5))
4145	with patch("urllib.request.urlopen", side_effect=mocks):
4146	result = runner.invoke(cli, ["hub", "issue", "create", "--title", "T"])
4147	assert result.exit_code == 0
4148	assert "5" in result.output
4149	assert "created" in result.output.lower()
4150
4151	def test_text_mode_no_json_on_stdout(self, repo: pathlib.Path) -> None:
4152	from muse.cli.config import set_hub_url
4153	set_hub_url(HUB_URL, repo)
4154	_store_identity(HUB_URL)
4155	mocks = _mock_responses(_refs_resp(), _issue_resp())
4156	with patch("urllib.request.urlopen", side_effect=mocks):
4157	result = runner.invoke(cli, ["hub", "issue", "create", "--title", "T"])
4158	assert result.exit_code == 0
4159	# Text mode must not emit a JSON object
4160	try:
4161	json.loads(result.output)
4162	assert False, "Text mode must not emit JSON"
4163	except (json.JSONDecodeError, ValueError):
4164	pass
4165
4166	def test_number_fallback_for_nonnumeric_api_response(
4167	self, repo: pathlib.Path
4168	) -> None:
4169	"""If API returns a non-numeric 'number', fall back to 0 without crashing."""
4170	from muse.cli.config import set_hub_url
4171	set_hub_url(HUB_URL, repo)
4172	_store_identity(HUB_URL)
4173	bad_issue = dict(_issue_resp())
4174	bad_issue["number"] = "not-a-number"
4175	mocks = _mock_responses(_refs_resp(), bad_issue)
4176	with patch("urllib.request.urlopen", side_effect=mocks):
4177	result = runner.invoke(cli, ["hub", "issue", "create", "--title", "T"])
4178	assert result.exit_code == 0 # must not crash
4179
4180	def test_number_float_coerced(self, repo: pathlib.Path) -> None:
4181	"""Numeric float from API (e.g. 7.0) must be coerced to int."""
4182	from muse.cli.config import set_hub_url
4183	set_hub_url(HUB_URL, repo)
4184	_store_identity(HUB_URL)
4185	float_issue = dict(_issue_resp())
4186	float_issue["number"] = 7.0
4187	mocks = _mock_responses(_refs_resp(), float_issue)
4188	with patch("urllib.request.urlopen", side_effect=mocks):
4189	result = runner.invoke(cli, ["hub", "issue", "create", "--title", "T"])
4190	assert result.exit_code == 0
4191	assert "7" in result.output
4192
4193	def test_no_hub_exits_nonzero(self, repo: pathlib.Path) -> None:
4194	result = runner.invoke(cli, ["hub", "issue", "create", "--title", "T"])
4195	assert result.exit_code != 0
4196
4197	def test_no_identity_exits_nonzero(self, repo: pathlib.Path) -> None:
4198	from muse.cli.config import set_hub_url
4199	set_hub_url(HUB_URL, repo)
4200	result = runner.invoke(cli, ["hub", "issue", "create", "--title", "T"])
4201	assert result.exit_code != 0
4202
4203	def test_outside_repo_exits_nonzero(
4204	self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
4205	) -> None:
4206	monkeypatch.chdir(tmp_path)
4207	monkeypatch.delenv("MUSE_REPO_ROOT", raising=False)
4208	result = runner.invoke(cli, ["hub", "issue", "create", "--title", "T"])
4209	assert result.exit_code != 0
4210
4211	def test_hub_override_used_in_request(self, repo: pathlib.Path) -> None:
4212	"""--hub overrides the config hub URL."""
4213	override_url = "http://override:9999/owner2/repo2"
4214	_store_identity(override_url)
4215	captured_urls: list[str] = []
4216
4217	def _fake(req: urllib.request.Request, timeout: int = 5) -> MagicMock:
4218	captured_urls.append(req.full_url)
4219	m = MagicMock()
4220	m.__enter__ = lambda s: s
4221	m.__exit__ = MagicMock(return_value=False)
4222	if "refs" in req.full_url:
4223	m.read.return_value = json.dumps(_refs_resp()).encode()
4224	else:
4225	m.read.return_value = json.dumps(_issue_resp()).encode()
4226	return m
4227
4228	with patch("urllib.request.urlopen", side_effect=_fake):
4229	result = runner.invoke(cli, [
4230	"hub", "issue", "create",
4231	"--hub", override_url,
4232	"--title", "T",
4233	])
4234	assert result.exit_code == 0
4235	assert any("override:9999" in u for u in captured_urls)
4236
4237
4238	# ---------------------------------------------------------------------------
4239	# TestIssueCreateSecurity
4240	# ---------------------------------------------------------------------------
4241
4242
4243	class TestIssueCreateSecurity:
4244	"""Security-focused tests for ``muse hub issue create``."""
4245
4246	def test_ansi_in_title_no_network_when_valid(
4247	self, repo: pathlib.Path
4248	) -> None:
4249	"""ANSI in title is not a validation error — title may contain them."""
4250	from muse.cli.config import set_hub_url
4251	set_hub_url(HUB_URL, repo)
4252	_store_identity(HUB_URL)
4253	ansi_title = "feat: \x1b[31mred\x1b[0m bug"
4254	mocks = _mock_responses(_refs_resp(), _issue_resp(title=ansi_title))
4255	with patch("urllib.request.urlopen", side_effect=mocks):
4256	result = runner.invoke(cli, ["hub", "issue", "create", "--title", ansi_title])
4257	assert result.exit_code == 0
4258
4259	def test_issueId_fallback_sanitized(self, repo: pathlib.Path) -> None:
4260	"""If hub URL has no owner/slug, issueId fallback must be sanitized."""
4261	# Give the hub URL no slug path so the fallback branch triggers.
4262	bare_hub = "http://localhost:10003"
4263	_store_identity(bare_hub)
4264	ansi_id = "iss_\x1b[31minjection\x1b[0m"
4265	issue = dict(_issue_resp())
4266	issue["issueId"] = ansi_id
4267
4268	mocks = _mock_responses(_refs_resp(), issue)
4269	with patch("urllib.request.urlopen", side_effect=mocks):
4270	result = runner.invoke(cli, [
4271	"hub", "issue", "create",
4272	"--hub", bare_hub,
4273	"--title", "T",
4274	])
4275	# ANSI escape sequences must not appear raw in output
4276	assert "\x1b[" not in result.output
4277
4278	def test_title_validation_before_network(
4279	self, repo: pathlib.Path
4280	) -> None:
4281	"""Empty title must be rejected before any HTTP call is made."""
4282	from muse.cli.config import set_hub_url
4283	set_hub_url(HUB_URL, repo)
4284	_store_identity(HUB_URL)
4285	with patch("urllib.request.urlopen") as mock_net:
4286	runner.invoke(cli, ["hub", "issue", "create", "--title", ""])
4287	mock_net.assert_not_called()
4288
4289	def test_max_title_len_constant_value(self) -> None:
4290	from muse.cli.commands.hub import _MAX_ISSUE_TITLE_LEN
4291	assert _MAX_ISSUE_TITLE_LEN == 512
4292
4293	def test_ansi_in_hub_url_path_not_echoed_raw(
4294	self, repo: pathlib.Path
4295	) -> None:
4296	"""ANSI in --hub URL path segments (owner/slug) must not reach stdout raw."""
4297	# Craft a hub URL where the owner segment contains an ANSI escape.
4298	# urllib.parse will preserve it in the path — it must be stripped on output.
4299	ansi_owner = "\x1b[31mevil\x1b[0m"
4300	evil_hub = f"http://localhost:10003/{ansi_owner}/repo"
4301	_store_identity(evil_hub)
4302	mocks = _mock_responses(_refs_resp(), _issue_resp(number=1))
4303	with patch("urllib.request.urlopen", side_effect=mocks):
4304	result = runner.invoke(cli, [
4305	"hub", "issue", "create",
4306	"--hub", evil_hub,
4307	"--title", "T",
4308	])
4309	assert "\x1b[" not in result.output
4310
4311	def test_payload_type_annotation_no_bool(self) -> None:
4312	"""The payload dict must not include bool values — type annotation check."""
4313	import inspect
4314	import muse.cli.commands.hub as hub_mod
4315	src = inspect.getsource(hub_mod.run_issue_create)
4316	# The old annotation included 'bool' — verify it was removed.
4317	# Look for the payload assignment line.
4318	assert "str \| bool \| list" not in src
4319
4320	def test_repo_flag_routes_to_correct_hub(
4321	self, repo: pathlib.Path
4322	) -> None:
4323	"""--repo owner/repo constructs a hub URL using the configured base."""
4324	from muse.cli.config import set_hub_url
4325	# Configure hub base (without owner/repo path)
4326	base_hub = "http://localhost:10003/original/original"
4327	set_hub_url(base_hub, repo)
4328	_store_identity("http://localhost:10003/myowner/myrepo")
4329	captured_urls: list[str] = []
4330
4331	def _fake(req: urllib.request.Request, timeout: int = 5) -> MagicMock:
4332	captured_urls.append(req.full_url)
4333	m = MagicMock()
4334	m.__enter__ = lambda s: s
4335	m.__exit__ = MagicMock(return_value=False)
4336	if req.method == "GET":
4337	m.read.return_value = json.dumps(_refs_resp()).encode()
4338	else:
4339	m.read.return_value = json.dumps(_issue_resp()).encode()
4340	return m
4341
4342	with patch("urllib.request.urlopen", side_effect=_fake):
4343	result = runner.invoke(cli, [
4344	"hub", "issue", "create",
4345	"--repo", "myowner/myrepo",
4346	"--title", "T",
4347	])
4348	assert result.exit_code == 0
4349	assert any("myowner" in u and "myrepo" in u for u in captured_urls)
4350
4351
4352	# ---------------------------------------------------------------------------
4353	# TestIssueEditHardening
4354	# ---------------------------------------------------------------------------
4355
4356
4357	class TestIssueEditHardening:
4358	"""Integration tests for ``muse hub issue edit``."""
4359
4360	def test_no_fields_exits_nonzero_no_network(
4361	self, repo: pathlib.Path
4362	) -> None:
4363	from muse.cli.config import set_hub_url
4364	set_hub_url(HUB_URL, repo)
4365	_store_identity(HUB_URL)
4366	with patch("urllib.request.urlopen") as mock_net:
4367	result = runner.invoke(cli, ["hub", "issue", "edit", "42"])
4368	assert result.exit_code != 0
4369	mock_net.assert_not_called()
4370
4371	def test_no_fields_error_message(self, repo: pathlib.Path) -> None:
4372	from muse.cli.config import set_hub_url
4373	set_hub_url(HUB_URL, repo)
4374	_store_identity(HUB_URL)
4375	with patch("urllib.request.urlopen"):
4376	result = runner.invoke(cli, ["hub", "issue", "edit", "42"])
4377	assert "nothing" in result.output.lower() or "update" in result.output.lower()
4378
4379	def test_title_only_patch(self, repo: pathlib.Path) -> None:
4380	from muse.cli.config import set_hub_url
4381	set_hub_url(HUB_URL, repo)
4382	_store_identity(HUB_URL)
4383	captured: list[bytes] = []
4384
4385	def _fake(req: urllib.request.Request, timeout: int = 5) -> MagicMock:
4386	if req.method == "PATCH":
4387	captured.append(req.data or b"")
4388	m = MagicMock()
4389	m.__enter__ = lambda s: s
4390	m.__exit__ = MagicMock(return_value=False)
4391	if req.method == "GET":
4392	m.read.return_value = json.dumps(_refs_resp()).encode()
4393	else:
4394	m.read.return_value = json.dumps(_issue_resp()).encode()
4395	return m
4396
4397	with patch("urllib.request.urlopen", side_effect=_fake):
4398	result = runner.invoke(cli, ["hub", "issue", "edit", "7", "--title", "new title"])
4399	assert result.exit_code == 0
4400	assert captured
4401	body = json.loads(captured[0])
4402	assert body == {"title": "new title"}
4403
4404	def test_body_only_patch(self, repo: pathlib.Path) -> None:
4405	from muse.cli.config import set_hub_url
4406	set_hub_url(HUB_URL, repo)
4407	_store_identity(HUB_URL)
4408	captured: list[bytes] = []
4409
4410	def _fake(req: urllib.request.Request, timeout: int = 5) -> MagicMock:
4411	if req.method == "PATCH":
4412	captured.append(req.data or b"")
4413	m = MagicMock()
4414	m.__enter__ = lambda s: s
4415	m.__exit__ = MagicMock(return_value=False)
4416	if req.method == "GET":
4417	m.read.return_value = json.dumps(_refs_resp()).encode()
4418	else:
4419	m.read.return_value = json.dumps(_issue_resp()).encode()
4420	return m
4421
4422	with patch("urllib.request.urlopen", side_effect=_fake):
4423	result = runner.invoke(cli, ["hub", "issue", "edit", "7", "--body", "new body"])
4424	assert result.exit_code == 0
4425	assert captured
4426	body = json.loads(captured[0])
4427	assert body == {"body": "new body"}
4428
4429	def test_both_title_and_body_in_patch(self, repo: pathlib.Path) -> None:
4430	from muse.cli.config import set_hub_url
4431	set_hub_url(HUB_URL, repo)
4432	_store_identity(HUB_URL)
4433	captured: list[bytes] = []
4434
4435	def _fake(req: urllib.request.Request, timeout: int = 5) -> MagicMock:
4436	if req.method == "PATCH":
4437	captured.append(req.data or b"")
4438	m = MagicMock()
4439	m.__enter__ = lambda s: s
4440	m.__exit__ = MagicMock(return_value=False)
4441	if req.method == "GET":
4442	m.read.return_value = json.dumps(_refs_resp()).encode()
4443	else:
4444	m.read.return_value = json.dumps(_issue_resp()).encode()
4445	return m
4446
4447	with patch("urllib.request.urlopen", side_effect=_fake):
4448	runner.invoke(
4449	cli,
4450	["hub", "issue", "edit", "7", "--title", "NT", "--body", "NB"],
4451	)
4452	assert captured
4453	body = json.loads(captured[0])
4454	assert body["title"] == "NT"
4455	assert body["body"] == "NB"
4456
4457	def test_patch_endpoint_includes_number(self, repo: pathlib.Path) -> None:
4458	from muse.cli.config import set_hub_url
4459	set_hub_url(HUB_URL, repo)
4460	_store_identity(HUB_URL)
4461	captured_urls: list[str] = []
4462
4463	def _fake(req: urllib.request.Request, timeout: int = 5) -> MagicMock:
4464	captured_urls.append(req.full_url)
4465	m = MagicMock()
4466	m.__enter__ = lambda s: s
4467	m.__exit__ = MagicMock(return_value=False)
4468	if req.method == "GET":
4469	m.read.return_value = json.dumps(_refs_resp()).encode()
4470	else:
4471	m.read.return_value = json.dumps(_issue_resp()).encode()
4472	return m
4473
4474	with patch("urllib.request.urlopen", side_effect=_fake):
4475	runner.invoke(cli, ["hub", "issue", "edit", "42", "--title", "T"])
4476	assert any("/issues/42" in u for u in captured_urls)
4477
4478	def test_uses_patch_method(self, repo: pathlib.Path) -> None:
4479	from muse.cli.config import set_hub_url
4480	set_hub_url(HUB_URL, repo)
4481	_store_identity(HUB_URL)
4482	methods: list[str] = []
4483
4484	def _fake(req: urllib.request.Request, timeout: int = 5) -> MagicMock:
4485	methods.append(req.method or "")
4486	m = MagicMock()
4487	m.__enter__ = lambda s: s
4488	m.__exit__ = MagicMock(return_value=False)
4489	if req.method == "GET":
4490	m.read.return_value = json.dumps(_refs_resp()).encode()
4491	else:
4492	m.read.return_value = json.dumps(_issue_resp()).encode()
4493	return m
4494
4495	with patch("urllib.request.urlopen", side_effect=_fake):
4496	runner.invoke(cli, ["hub", "issue", "edit", "42", "--title", "T"])
4497	assert "PATCH" in methods
4498
4499	def test_json_passthrough(self, repo: pathlib.Path) -> None:
4500	from muse.cli.config import set_hub_url
4501	set_hub_url(HUB_URL, repo)
4502	_store_identity(HUB_URL)
4503	mocks = _mock_responses(_refs_resp(), _issue_resp(number=42))
4504	with patch("urllib.request.urlopen", side_effect=mocks):
4505	result = runner.invoke(
4506	cli, ["hub", "issue", "edit", "42", "--title", "T", "--json"]
4507	)
4508	assert result.exit_code == 0
4509	data = json.loads(result.output)
4510	assert "number" in data
4511
4512	def test_json_short_flag(self, repo: pathlib.Path) -> None:
4513	from muse.cli.config import set_hub_url
4514	set_hub_url(HUB_URL, repo)
4515	_store_identity(HUB_URL)
4516	mocks = _mock_responses(_refs_resp(), _issue_resp())
4517	with patch("urllib.request.urlopen", side_effect=mocks):
4518	result = runner.invoke(
4519	cli, ["hub", "issue", "edit", "42", "--title", "T", "-j"]
4520	)
4521	assert result.exit_code == 0
4522	json.loads(result.output)
4523
4524	def test_text_mode_success_message(self, repo: pathlib.Path) -> None:
4525	from muse.cli.config import set_hub_url
4526	set_hub_url(HUB_URL, repo)
4527	_store_identity(HUB_URL)
4528	mocks = _mock_responses(_refs_resp(), _issue_resp(number=42))
4529	with patch("urllib.request.urlopen", side_effect=mocks):
4530	result = runner.invoke(
4531	cli, ["hub", "issue", "edit", "42", "--title", "T"]
4532	)
4533	assert result.exit_code == 0
4534	assert "42" in result.output
4535	assert "updated" in result.output.lower()
4536
4537	def test_text_mode_no_json_on_stdout(self, repo: pathlib.Path) -> None:
4538	from muse.cli.config import set_hub_url
4539	set_hub_url(HUB_URL, repo)
4540	_store_identity(HUB_URL)
4541	mocks = _mock_responses(_refs_resp(), _issue_resp())
4542	with patch("urllib.request.urlopen", side_effect=mocks):
4543	result = runner.invoke(
4544	cli, ["hub", "issue", "edit", "7", "--title", "T"]
4545	)
4546	assert result.exit_code == 0
4547	try:
4548	json.loads(result.output)
4549	assert False, "Text mode must not emit JSON"
4550	except (json.JSONDecodeError, ValueError):
4551	pass
4552
4553	def test_no_hub_exits_nonzero(self, repo: pathlib.Path) -> None:
4554	result = runner.invoke(cli, ["hub", "issue", "edit", "1", "--title", "T"])
4555	assert result.exit_code != 0
4556
4557	def test_no_identity_exits_nonzero(self, repo: pathlib.Path) -> None:
4558	from muse.cli.config import set_hub_url
4559	set_hub_url(HUB_URL, repo)
4560	result = runner.invoke(cli, ["hub", "issue", "edit", "1", "--title", "T"])
4561	assert result.exit_code != 0
4562
4563	def test_outside_repo_exits_nonzero(
4564	self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
4565	) -> None:
4566	monkeypatch.chdir(tmp_path)
4567	monkeypatch.delenv("MUSE_REPO_ROOT", raising=False)
4568	result = runner.invoke(cli, ["hub", "issue", "edit", "1", "--title", "T"])
4569	assert result.exit_code != 0
4570
4571	def test_hub_override_used(self, repo: pathlib.Path) -> None:
4572	override_url = "http://override:9999/owner2/repo2"
4573	_store_identity(override_url)
4574	captured_urls: list[str] = []
4575
4576	def _fake(req: urllib.request.Request, timeout: int = 5) -> MagicMock:
4577	captured_urls.append(req.full_url)
4578	m = MagicMock()
4579	m.__enter__ = lambda s: s
4580	m.__exit__ = MagicMock(return_value=False)
4581	if req.method == "GET":
4582	m.read.return_value = json.dumps(_refs_resp()).encode()
4583	else:
4584	m.read.return_value = json.dumps(_issue_resp()).encode()
4585	return m
4586
4587	with patch("urllib.request.urlopen", side_effect=_fake):
4588	result = runner.invoke(cli, [
4589	"hub", "issue", "edit", "1",
4590	"--hub", override_url,
4591	"--title", "T",
4592	])
4593	assert result.exit_code == 0
4594	assert any("override:9999" in u for u in captured_urls)
4595
4596
4597	# ---------------------------------------------------------------------------
4598	# TestIssueEditSecurity
4599	# ---------------------------------------------------------------------------
4600
4601
4602	class TestIssueEditSecurity:
4603	"""Security and validation tests for ``muse hub issue edit``."""
4604
4605	def test_negative_number_exits_nonzero_no_network(
4606	self, repo: pathlib.Path
4607	) -> None:
4608	from muse.cli.config import set_hub_url
4609	set_hub_url(HUB_URL, repo)
4610	_store_identity(HUB_URL)
4611	with patch("urllib.request.urlopen") as mock_net:
4612	# Pass number as positional — argparse type=int accepts negatives
4613	result = runner.invoke(cli, ["hub", "issue", "edit", "0", "--title", "T"])
4614	assert result.exit_code != 0
4615	mock_net.assert_not_called()
4616
4617	def test_zero_number_exits_nonzero_no_network(
4618	self, repo: pathlib.Path
4619	) -> None:
4620	from muse.cli.config import set_hub_url
4621	set_hub_url(HUB_URL, repo)
4622	_store_identity(HUB_URL)
4623	with patch("urllib.request.urlopen") as mock_net:
4624	result = runner.invoke(cli, ["hub", "issue", "edit", "0", "--title", "T"])
4625	assert result.exit_code != 0
4626	mock_net.assert_not_called()
4627
4628	def test_zero_number_shows_helpful_message(
4629	self, repo: pathlib.Path
4630	) -> None:
4631	from muse.cli.config import set_hub_url
4632	set_hub_url(HUB_URL, repo)
4633	_store_identity(HUB_URL)
4634	with patch("urllib.request.urlopen"):
4635	result = runner.invoke(cli, ["hub", "issue", "edit", "0", "--title", "T"])
4636	assert "positive" in result.output.lower() or "0" in result.output
4637
4638	def test_title_too_long_exits_nonzero_no_network(
4639	self, repo: pathlib.Path
4640	) -> None:
4641	from muse.cli.commands.hub import _MAX_ISSUE_TITLE_LEN
4642	from muse.cli.config import set_hub_url
4643	set_hub_url(HUB_URL, repo)
4644	_store_identity(HUB_URL)
4645	long_title = "x" * (_MAX_ISSUE_TITLE_LEN + 1)
4646	with patch("urllib.request.urlopen") as mock_net:
4647	result = runner.invoke(cli, ["hub", "issue", "edit", "1", "--title", long_title])
4648	assert result.exit_code != 0
4649	mock_net.assert_not_called()
4650
4651	def test_title_too_long_shows_char_count(
4652	self, repo: pathlib.Path
4653	) -> None:
4654	from muse.cli.commands.hub import _MAX_ISSUE_TITLE_LEN
4655	from muse.cli.config import set_hub_url
4656	set_hub_url(HUB_URL, repo)
4657	_store_identity(HUB_URL)
4658	long_title = "x" * (_MAX_ISSUE_TITLE_LEN + 1)
4659	with patch("urllib.request.urlopen"):
4660	result = runner.invoke(cli, ["hub", "issue", "edit", "1", "--title", long_title])
4661	assert str(_MAX_ISSUE_TITLE_LEN + 1) in result.output or str(_MAX_ISSUE_TITLE_LEN) in result.output
4662
4663	def test_title_at_max_length_accepted(
4664	self, repo: pathlib.Path
4665	) -> None:
4666	from muse.cli.commands.hub import _MAX_ISSUE_TITLE_LEN
4667	from muse.cli.config import set_hub_url
4668	set_hub_url(HUB_URL, repo)
4669	_store_identity(HUB_URL)
4670	exact_title = "x" * _MAX_ISSUE_TITLE_LEN
4671	mocks = _mock_responses(_refs_resp(), _issue_resp(title=exact_title))
4672	with patch("urllib.request.urlopen", side_effect=mocks):
4673	result = runner.invoke(
4674	cli, ["hub", "issue", "edit", "1", "--title", exact_title, "--json"]
4675	)
4676	assert result.exit_code == 0
4677
4678	def test_empty_title_exits_nonzero_no_network(
4679	self, repo: pathlib.Path
4680	) -> None:
4681	from muse.cli.config import set_hub_url
4682	set_hub_url(HUB_URL, repo)
4683	_store_identity(HUB_URL)
4684	with patch("urllib.request.urlopen") as mock_net:
4685	result = runner.invoke(cli, ["hub", "issue", "edit", "1", "--title", ""])
4686	assert result.exit_code != 0
4687	mock_net.assert_not_called()
4688
4689	def test_whitespace_only_title_exits_nonzero_no_network(
4690	self, repo: pathlib.Path
4691	) -> None:
4692	from muse.cli.config import set_hub_url
4693	set_hub_url(HUB_URL, repo)
4694	_store_identity(HUB_URL)
4695	with patch("urllib.request.urlopen") as mock_net:
4696	result = runner.invoke(cli, ["hub", "issue", "edit", "1", "--title", " "])
4697	assert result.exit_code != 0
4698	mock_net.assert_not_called()
4699
4700	def test_empty_title_shows_error_message(
4701	self, repo: pathlib.Path
4702	) -> None:
4703	from muse.cli.config import set_hub_url
4704	set_hub_url(HUB_URL, repo)
4705	_store_identity(HUB_URL)
4706	with patch("urllib.request.urlopen"):
4707	result = runner.invoke(cli, ["hub", "issue", "edit", "1", "--title", ""])
4708	assert "empty" in result.output.lower() or "title" in result.output.lower()
4709
4710	def test_all_validation_before_network(
4711	self, repo: pathlib.Path
4712	) -> None:
4713	"""All local validation must fire before any HTTP call."""
4714	from muse.cli.config import set_hub_url
4715	set_hub_url(HUB_URL, repo)
4716	_store_identity(HUB_URL)
4717	with patch("urllib.request.urlopen") as mock_net:
4718	# zero number + empty title — both are invalid
4719	runner.invoke(cli, ["hub", "issue", "edit", "0", "--title", ""])
4720	mock_net.assert_not_called()
4721
4722	def test_repo_flag_routes_correctly(
4723	self, repo: pathlib.Path
4724	) -> None:
4725	"""--repo owner/repo constructs a hub URL using the configured base."""
4726	from muse.cli.config import set_hub_url
4727	base_hub = "http://localhost:10003/original/original"
4728	set_hub_url(base_hub, repo)
4729	_store_identity("http://localhost:10003/myowner/myrepo")
4730	captured_urls: list[str] = []
4731
4732	def _fake(req: urllib.request.Request, timeout: int = 5) -> MagicMock:
4733	captured_urls.append(req.full_url)
4734	m = MagicMock()
4735	m.__enter__ = lambda s: s
4736	m.__exit__ = MagicMock(return_value=False)
4737	if req.method == "GET":
4738	m.read.return_value = json.dumps(_refs_resp()).encode()
4739	else:
4740	m.read.return_value = json.dumps(_issue_resp()).encode()
4741	return m
4742
4743	with patch("urllib.request.urlopen", side_effect=_fake):
4744	result = runner.invoke(cli, [
4745	"hub", "issue", "edit", "1",
4746	"--repo", "myowner/myrepo",
4747	"--title", "T",
4748	])
4749	assert result.exit_code == 0
4750	assert any("myowner" in u and "myrepo" in u for u in captured_urls)
4751
4752
4753	# ---------------------------------------------------------------------------
4754	# TestIssueEditStress
4755	# ---------------------------------------------------------------------------
4756
4757
4758	class TestIssueEditStress:
4759	"""Stress and boundary tests for ``muse hub issue edit``."""
4760
4761	def test_title_boundary_constants(self) -> None:
4762	from muse.cli.commands.hub import _MAX_ISSUE_TITLE_LEN
4763	assert isinstance(_MAX_ISSUE_TITLE_LEN, int)
4764	assert _MAX_ISSUE_TITLE_LEN > 0
4765
4766	def test_concurrent_validation(self) -> None:
4767	"""Title and number validation logic is thread-safe."""
4768	from muse.cli.commands.hub import _MAX_ISSUE_TITLE_LEN
4769	errors: list[str] = []
4770
4771	def _check(idx: int) -> None:
4772	try:
4773	number = idx - 4 # some negative, some positive
4774	title = "x" * (idx * 10)
4775	bad_number = number <= 0
4776	bad_title = len(title) > _MAX_ISSUE_TITLE_LEN or not title.strip()
4777	assert isinstance(bad_number, bool)
4778	assert isinstance(bad_title, bool)
4779	except Exception as exc:
4780	errors.append(f"Thread {idx}: {exc}")
4781
4782	threads = [threading.Thread(target=_check, args=(i,)) for i in range(8)]
4783	for t in threads:
4784	t.start()
4785	for t in threads:
4786	t.join()
4787	assert errors == [], "\n".join(errors)
4788
4789	def test_body_only_no_title_validation(
4790	self, repo: pathlib.Path
4791	) -> None:
4792	"""When only --body is provided, title validation must not run."""
4793	from muse.cli.config import set_hub_url
4794	set_hub_url(HUB_URL, repo)
4795	_store_identity(HUB_URL)
4796	mocks = _mock_responses(_refs_resp(), _issue_resp())
4797	with patch("urllib.request.urlopen", side_effect=mocks):
4798	result = runner.invoke(
4799	cli, ["hub", "issue", "edit", "1", "--body", "updated"]
4800	)
4801	assert result.exit_code == 0
4802
4803	def test_positive_number_one_accepted(
4804	self, repo: pathlib.Path
4805	) -> None:
4806	"""Issue number 1 (minimum valid) must be accepted."""
4807	from muse.cli.config import set_hub_url
4808	set_hub_url(HUB_URL, repo)
4809	_store_identity(HUB_URL)
4810	mocks = _mock_responses(_refs_resp(), _issue_resp(number=1))
4811	with patch("urllib.request.urlopen", side_effect=mocks):
4812	result = runner.invoke(
4813	cli, ["hub", "issue", "edit", "1", "--title", "T"]
4814	)
4815	assert result.exit_code == 0
4816
4817	def test_large_number_accepted(
4818	self, repo: pathlib.Path
4819	) -> None:
4820	"""Very large issue numbers are valid."""
4821	from muse.cli.config import set_hub_url
4822	set_hub_url(HUB_URL, repo)
4823	_store_identity(HUB_URL)
4824	mocks = _mock_responses(_refs_resp(), _issue_resp(number=999999))
4825	with patch("urllib.request.urlopen", side_effect=mocks):
4826	result = runner.invoke(
4827	cli, ["hub", "issue", "edit", "999999", "--title", "T"]
4828	)
4829	assert result.exit_code == 0
4830
4831
4832	# ---------------------------------------------------------------------------
4833	# TestIssueSubparserRegistration
4834	# ---------------------------------------------------------------------------
4835
4836
4837	class TestIssueSubparserRegistration:
4838	"""Verify subparser wiring and flag aliases."""
4839
4840	def test_create_help_contains_agent_quickstart(self) -> None:
4841	result = runner.invoke(cli, ["hub", "issue", "create", "--help"])
4842	assert "quickstart" in result.output.lower() or "--json" in result.output
4843
4844	def test_edit_help_contains_exit_codes(self) -> None:
4845	result = runner.invoke(cli, ["hub", "issue", "edit", "--help"])
4846	assert "Exit codes" in result.output or "exit" in result.output.lower()
4847
4848	def test_create_j_alias_accepted(
4849	self, repo: pathlib.Path
4850	) -> None:
4851	from muse.cli.config import set_hub_url
4852	set_hub_url(HUB_URL, repo)
4853	_store_identity(HUB_URL)
4854	mocks = _mock_responses(_refs_resp(), _issue_resp())
4855	with patch("urllib.request.urlopen", side_effect=mocks):
4856	result = runner.invoke(cli, ["hub", "issue", "create", "--title", "T", "-j"])
4857	assert result.exit_code == 0
4858	json.loads(result.output)
4859
4860	def test_edit_j_alias_accepted(
4861	self, repo: pathlib.Path
4862	) -> None:
4863	from muse.cli.config import set_hub_url
4864	set_hub_url(HUB_URL, repo)
4865	_store_identity(HUB_URL)
4866	mocks = _mock_responses(_refs_resp(), _issue_resp())
4867	with patch("urllib.request.urlopen", side_effect=mocks):
4868	result = runner.invoke(cli, ["hub", "issue", "edit", "7", "--title", "T", "-j"])
4869	assert result.exit_code == 0
4870	json.loads(result.output)
4871
4872	def test_issue_no_subcommand_shows_help(self) -> None:
4873	result = runner.invoke(cli, ["hub", "issue"])
4874	# Missing required subcommand — nonzero exit with usage info
4875	assert result.exit_code != 0 or "create" in result.output
4876
4877
4878	# ---------------------------------------------------------------------------
4879	# TestIssueE2E
4880	# ---------------------------------------------------------------------------
4881
4882
4883	class TestIssueE2E:
4884	"""End-to-end flows through the full CLI stack."""
4885
4886	def test_create_agent_json_pipeline(self, repo: pathlib.Path) -> None:
4887	"""Agent can extract issue number from JSON output."""
4888	from muse.cli.config import set_hub_url
4889	set_hub_url(HUB_URL, repo)
4890	_store_identity(HUB_URL)
4891	mocks = _mock_responses(_refs_resp(), _issue_resp(number=99))
4892	with patch("urllib.request.urlopen", side_effect=mocks):
4893	result = runner.invoke(
4894	cli,
4895	["hub", "issue", "create", "--title", "agent task", "--json"],
4896	)
4897	assert result.exit_code == 0
4898	data = json.loads(result.output)
4899	assert data["number"] == 99
4900
4901	def test_create_text_url_scriptable(self, repo: pathlib.Path) -> None:
4902	"""Text mode emits issue URL to stdout for shell capture."""
4903	from muse.cli.config import set_hub_url
4904	set_hub_url(HUB_URL, repo)
4905	_store_identity(HUB_URL)
4906	mocks = _mock_responses(_refs_resp(), _issue_resp(number=12))
4907	with patch("urllib.request.urlopen", side_effect=mocks):
4908	result = runner.invoke(cli, ["hub", "issue", "create", "--title", "T"])
4909	assert result.exit_code == 0
4910	assert "/issues/12" in result.output
4911
4912	def test_edit_agent_json_pipeline(self, repo: pathlib.Path) -> None:
4913	"""Agent can patch an issue and get the updated object back."""
4914	from muse.cli.config import set_hub_url
4915	set_hub_url(HUB_URL, repo)
4916	_store_identity(HUB_URL)
4917	updated = dict(_issue_resp(number=5, title="new title"))
4918	mocks = _mock_responses(_refs_resp(), updated)
4919	with patch("urllib.request.urlopen", side_effect=mocks):
4920	result = runner.invoke(
4921	cli,
4922	["hub", "issue", "edit", "5", "--title", "new title", "--json"],
4923	)
4924	assert result.exit_code == 0
4925	data = json.loads(result.output)
4926	assert data["title"] == "new title"
4927
4928	def test_create_then_edit_flow(self, repo: pathlib.Path) -> None:
4929	"""Create an issue then edit it in two separate invocations."""
4930	from muse.cli.config import set_hub_url
4931	set_hub_url(HUB_URL, repo)
4932	_store_identity(HUB_URL)
4933
4934	# create
4935	mocks_create = _mock_responses(_refs_resp(), _issue_resp(number=20))
4936	with patch("urllib.request.urlopen", side_effect=mocks_create):
4937	r1 = runner.invoke(
4938	cli, ["hub", "issue", "create", "--title", "initial title", "--json"]
4939	)
4940	assert r1.exit_code == 0
4941
4942	# edit
4943	mocks_edit = _mock_responses(_refs_resp(), _issue_resp(number=20, title="updated"))
4944	with patch("urllib.request.urlopen", side_effect=mocks_edit):
4945	r2 = runner.invoke(
4946	cli, ["hub", "issue", "edit", "20", "--title", "updated", "--json"]
4947	)
4948	assert r2.exit_code == 0
4949	assert json.loads(r2.output)["title"] == "updated"
4950
4951	def test_validation_error_does_not_leak_network(
4952	self, repo: pathlib.Path
4953	) -> None:
4954	"""Validation failure before network I/O — hub is never contacted."""
4955	from muse.cli.config import set_hub_url
4956	set_hub_url(HUB_URL, repo)
4957	_store_identity(HUB_URL)
4958	with patch("urllib.request.urlopen") as mock_net:
4959	runner.invoke(cli, ["hub", "issue", "create", "--title", ""])
4960	runner.invoke(cli, ["hub", "issue", "edit", "1"])
4961	mock_net.assert_not_called()
4962
4963
4964	# ---------------------------------------------------------------------------
4965	# TestIssueStress
4966	# ---------------------------------------------------------------------------
4967
4968
4969	class TestIssueStress:
4970	"""Stress tests: boundary conditions and concurrency."""
4971
4972	def test_title_boundary_constants(self) -> None:
4973	from muse.cli.commands.hub import _MAX_ISSUE_TITLE_LEN
4974	assert isinstance(_MAX_ISSUE_TITLE_LEN, int)
4975	assert _MAX_ISSUE_TITLE_LEN > 0
4976
4977	def test_labels_many(self, repo: pathlib.Path) -> None:
4978	"""50 labels on a single issue create must not crash."""
4979	from muse.cli.config import set_hub_url
4980	set_hub_url(HUB_URL, repo)
4981	_store_identity(HUB_URL)
4982	captured: list[bytes] = []
4983
4984	def _fake(req: urllib.request.Request, timeout: int = 5) -> MagicMock:
4985	if req.method == "POST":
4986	captured.append(req.data or b"")
4987	m = MagicMock()
4988	m.__enter__ = lambda s: s
4989	m.__exit__ = MagicMock(return_value=False)
4990	if req.method == "GET":
4991	m.read.return_value = json.dumps(_refs_resp()).encode()
4992	else:
4993	m.read.return_value = json.dumps(_issue_resp()).encode()
4994	return m
4995
4996	args = ["hub", "issue", "create", "--title", "T"]
4997	for i in range(50):
4998	args += ["--label", f"label-{i}"]
4999	with patch("urllib.request.urlopen", side_effect=_fake):
5000	result = runner.invoke(cli, args)
5001	assert result.exit_code == 0
5002	assert captured
5003	body = json.loads(captured[0])
5004	assert len(body["labels"]) == 50
5005
5006	def test_concurrent_title_validation(self) -> None:
5007	"""Pure title validation logic is thread-safe."""
5008	from muse.cli.commands.hub import _MAX_ISSUE_TITLE_LEN
5009	errors: list[str] = []
5010
5011	def _check(idx: int) -> None:
5012	try:
5013	title = "x" * (idx % (_MAX_ISSUE_TITLE_LEN + 10))
5014	too_long = len(title) > _MAX_ISSUE_TITLE_LEN
5015	empty = not title.strip()
5016	assert isinstance(too_long, bool)
5017	assert isinstance(empty, bool)
5018	except Exception as exc:
5019	errors.append(f"Thread {idx}: {exc}")
5020
5021	threads = [threading.Thread(target=_check, args=(i,)) for i in range(8)]
5022	for t in threads:
5023	t.start()
5024	for t in threads:
5025	t.join()
5026	assert errors == [], "\n".join(errors)
5027
5028	def test_number_parse_edge_cases(self) -> None:
5029	"""Number parsing edge cases must not raise."""
5030	import argparse as _ap
5031	from muse.cli.commands.hub import _MAX_ISSUE_TITLE_LEN
5032
5033	cases: list[MsgpackValue] = [
5034	None, 0, 1, 1.5, "42", "bad", "", [], {}
5035	]
5036	for val in cases:
5037	try:
5038	number = int(val) if val is not None else 0
5039	except (ValueError, TypeError):
5040	number = 0
5041	assert isinstance(number, int)
5042
5043	# ═══════════════════════════════════════════════════════════════════════════════
5044	# hub repo create — comprehensive tests
5045	# ═══════════════════════════════════════════════════════════════════════════════
5046
5047	# ── helpers ───────────────────────────────────────────────────────────────────
5048
5049	_REPO_RESPONSE = {
5050	"repoId": "abc123def456",
5051	"repo_id": "abc123def456",
5052	"name": "my-repo",
5053	"owner": "alice",
5054	"slug": "my-repo",
5055	"visibility": "public",
5056	"description": "A test repository",
5057	"cloneUrl": "https://staging.musehub.ai/api/repos/abc123def456",
5058	"clone_url": "https://staging.musehub.ai/api/repos/abc123def456",
5059	"tags": [],
5060	"createdAt": "2026-04-05T00:00:00Z",
5061	"created_at": "2026-04-05T00:00:00Z",
5062	}
5063
5064
5065	def _mock_hub_api_repo_create(monkeypatch: pytest.MonkeyPatch, response: dict \| None = None) -> None:
5066	"""Patch _hub_api to return a successful repo creation response."""
5067	payload = response if response is not None else _REPO_RESPONSE
5068
5069	def _fake_hub_api(hub_url, identity, method, path, body=None, timeout=10.0):
5070	return payload
5071
5072	monkeypatch.setattr("muse.cli.commands.hub._hub_api", _fake_hub_api)
5073
5074
5075	# ── Unit: local validation ────────────────────────────────────────────────────
5076
5077
5078	class TestRepoCreateValidation:
5079	"""Client-side validation runs before any network I/O."""
5080
5081	def test_empty_name_rejected(
5082	self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
5083	) -> None:
5084	from muse.cli.config import set_hub_url
5085	set_hub_url("https://musehub.example.com", repo)
5086	_store_identity("https://musehub.example.com")
5087	result = runner.invoke(cli, ["hub", "repo", "create", "--name", ""])
5088	assert result.exit_code != 0
5089
5090	def test_whitespace_only_name_rejected(
5091	self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
5092	) -> None:
5093	from muse.cli.config import set_hub_url
5094	set_hub_url("https://musehub.example.com", repo)
5095	_store_identity("https://musehub.example.com")
5096	result = runner.invoke(cli, ["hub", "repo", "create", "--name", " "])
5097	assert result.exit_code != 0
5098
5099	def test_name_too_long_rejected(
5100	self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
5101	) -> None:
5102	from muse.cli.commands.hub import _MAX_REPO_NAME_LEN
5103	from muse.cli.config import set_hub_url
5104	set_hub_url("https://musehub.example.com", repo)
5105	_store_identity("https://musehub.example.com")
5106	long_name = "a" * (_MAX_REPO_NAME_LEN + 1)
5107	result = runner.invoke(cli, ["hub", "repo", "create", "--name", long_name])
5108	assert result.exit_code != 0
5109	assert "too long" in result.output
5110
5111	def test_description_too_long_rejected(
5112	self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
5113	) -> None:
5114	from muse.cli.commands.hub import _MAX_REPO_DESC_LEN
5115	from muse.cli.config import set_hub_url
5116	set_hub_url("https://musehub.example.com", repo)
5117	_store_identity("https://musehub.example.com")
5118	long_desc = "x" * (_MAX_REPO_DESC_LEN + 1)
5119	result = runner.invoke(
5120	cli, ["hub", "repo", "create", "--name", "my-repo", "--description", long_desc]
5121	)
5122	assert result.exit_code != 0
5123	assert "too long" in result.output
5124
5125	def test_name_at_max_length_accepted(
5126	self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
5127	) -> None:
5128	from muse.cli.commands.hub import _MAX_REPO_NAME_LEN
5129	from muse.cli.config import set_hub_url
5130	set_hub_url("https://musehub.example.com", repo)
5131	_store_identity("https://musehub.example.com")
5132	_mock_hub_api_repo_create(monkeypatch, {*_REPO_RESPONSE, "name": "a" _MAX_REPO_NAME_LEN})
5133	result = runner.invoke(
5134	cli, ["hub", "repo", "create", "--name", "a" * _MAX_REPO_NAME_LEN]
5135	)
5136	assert result.exit_code == 0
5137
5138	def test_empty_default_branch_rejected(
5139	self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
5140	) -> None:
5141	from muse.cli.config import set_hub_url
5142	set_hub_url("https://musehub.example.com", repo)
5143	_store_identity("https://musehub.example.com")
5144	result = runner.invoke(
5145	cli,
5146	["hub", "repo", "create", "--name", "my-repo", "--default-branch", ""],
5147	)
5148	assert result.exit_code != 0
5149
5150	def test_validation_before_network(
5151	self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
5152	) -> None:
5153	"""Network should never be reached when validation fails."""
5154	called: list[bool] = []
5155
5156	def _fake_hub_api(args, *kwargs):
5157	called.append(True)
5158	return _REPO_RESPONSE
5159
5160	monkeypatch.setattr("muse.cli.commands.hub._hub_api", _fake_hub_api)
5161	from muse.cli.config import set_hub_url
5162	set_hub_url("https://musehub.example.com", repo)
5163	_store_identity("https://musehub.example.com")
5164	runner.invoke(cli, ["hub", "repo", "create", "--name", ""])
5165	assert called == [], "Network was called despite local validation failure"
5166
5167
5168	# ── Integration: happy path ───────────────────────────────────────────────────
5169
5170
5171	class TestRepoCreateIntegration:
5172	"""Happy-path and flag behaviour with mocked network."""
5173
5174	def test_create_text_output_shows_slug(
5175	self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
5176	) -> None:
5177	from muse.cli.config import set_hub_url
5178	set_hub_url("https://musehub.example.com/alice/my-repo", repo)
5179	_store_identity("https://musehub.example.com/alice/my-repo")
5180	_mock_hub_api_repo_create(monkeypatch)
5181	result = runner.invoke(cli, ["hub", "repo", "create", "--name", "my-repo"])
5182	assert result.exit_code == 0
5183	assert "my-repo" in result.output
5184
5185	def test_create_json_schema(
5186	self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
5187	) -> None:
5188	from muse.cli.config import set_hub_url
5189	set_hub_url("https://musehub.example.com/alice/my-repo", repo)
5190	_store_identity("https://musehub.example.com/alice/my-repo")
5191	_mock_hub_api_repo_create(monkeypatch)
5192	result = runner.invoke(
5193	cli, ["hub", "repo", "create", "--name", "my-repo", "--json"]
5194	)
5195	assert result.exit_code == 0
5196	data = _json_line(result)
5197	assert isinstance(data, dict)
5198	for key in ("repo_id", "name", "owner", "slug", "visibility", "description", "clone_url", "tags", "created_at"):
5199	assert key in data, f"Missing key: {key}"
5200
5201	def test_create_json_visibility_public_default(
5202	self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
5203	) -> None:
5204	from muse.cli.config import set_hub_url
5205	set_hub_url("https://musehub.example.com/alice/my-repo", repo)
5206	_store_identity("https://musehub.example.com/alice/my-repo")
5207	_mock_hub_api_repo_create(monkeypatch)
5208	result = runner.invoke(
5209	cli, ["hub", "repo", "create", "--name", "my-repo", "--json"]
5210	)
5211	assert result.exit_code == 0
5212	data = _json_line(result)
5213	assert data["visibility"] == "public"
5214
5215	def test_create_private_flag(
5216	self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
5217	) -> None:
5218	from muse.cli.config import set_hub_url
5219	set_hub_url("https://musehub.example.com/alice/my-repo", repo)
5220	_store_identity("https://musehub.example.com/alice/my-repo")
5221
5222	captured: list[dict] = []
5223
5224	def _capture(hub_url, identity, method, path, body=None, timeout=10.0):
5225	if body:
5226	captured.append(dict(body))
5227	return _REPO_RESPONSE
5228
5229	monkeypatch.setattr("muse.cli.commands.hub._hub_api", _capture)
5230	runner.invoke(
5231	cli, ["hub", "repo", "create", "--name", "my-repo", "--private"]
5232	)
5233	assert captured and captured[0].get("visibility") == "private"
5234
5235	def test_create_no_init_flag(
5236	self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
5237	) -> None:
5238	from muse.cli.config import set_hub_url
5239	set_hub_url("https://musehub.example.com/alice/my-repo", repo)
5240	_store_identity("https://musehub.example.com/alice/my-repo")
5241
5242	captured: list[dict] = []
5243
5244	def _capture(hub_url, identity, method, path, body=None, timeout=10.0):
5245	if body:
5246	captured.append(dict(body))
5247	return _REPO_RESPONSE
5248
5249	monkeypatch.setattr("muse.cli.commands.hub._hub_api", _capture)
5250	runner.invoke(
5251	cli, ["hub", "repo", "create", "--name", "my-repo", "--no-init"]
5252	)
5253	assert captured and captured[0].get("initialize") is False
5254
5255	def test_create_default_branch_forwarded(
5256	self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
5257	) -> None:
5258	from muse.cli.config import set_hub_url
5259	set_hub_url("https://musehub.example.com/alice/my-repo", repo)
5260	_store_identity("https://musehub.example.com/alice/my-repo")
5261
5262	captured: list[dict] = []
5263
5264	def _capture(hub_url, identity, method, path, body=None, timeout=10.0):
5265	if body:
5266	captured.append(dict(body))
5267	return _REPO_RESPONSE
5268
5269	monkeypatch.setattr("muse.cli.commands.hub._hub_api", _capture)
5270	runner.invoke(
5271	cli,
5272	["hub", "repo", "create", "--name", "my-repo", "--default-branch", "dev"],
5273	)
5274	assert captured and captured[0].get("defaultBranch") == "dev"
5275
5276	def test_create_tags_forwarded(
5277	self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
5278	) -> None:
5279	from muse.cli.config import set_hub_url
5280	set_hub_url("https://musehub.example.com/alice/my-repo", repo)
5281	_store_identity("https://musehub.example.com/alice/my-repo")
5282
5283	captured: list[dict] = []
5284
5285	def _capture(hub_url, identity, method, path, body=None, timeout=10.0):
5286	if body:
5287	captured.append(dict(body))
5288	return _REPO_RESPONSE
5289
5290	monkeypatch.setattr("muse.cli.commands.hub._hub_api", _capture)
5291	runner.invoke(
5292	cli,
5293	["hub", "repo", "create", "--name", "my-repo", "--tag", "jazz", "--tag", "piano"],
5294	)
5295	assert captured and set(captured[0].get("tags", [])) == {"jazz", "piano"}
5296
5297	def test_create_owner_override(
5298	self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
5299	) -> None:
5300	from muse.cli.config import set_hub_url
5301	set_hub_url("https://musehub.example.com/alice/my-repo", repo)
5302	_store_identity("https://musehub.example.com/alice/my-repo")
5303
5304	captured: list[dict] = []
5305
5306	def _capture(hub_url, identity, method, path, body=None, timeout=10.0):
5307	if body:
5308	captured.append(dict(body))
5309	return _REPO_RESPONSE
5310
5311	monkeypatch.setattr("muse.cli.commands.hub._hub_api", _capture)
5312	runner.invoke(
5313	cli,
5314	["hub", "repo", "create", "--name", "my-repo", "--owner", "bob"],
5315	)
5316	assert captured and captured[0].get("owner") == "bob"
5317
5318	def test_create_no_hub_exits_nonzero(
5319	self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
5320	) -> None:
5321	result = runner.invoke(cli, ["hub", "repo", "create", "--name", "my-repo"])
5322	assert result.exit_code != 0
5323
5324	def test_create_not_in_repo_exits(
5325	self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
5326	) -> None:
5327	monkeypatch.chdir(tmp_path)
5328	monkeypatch.delenv("MUSE_REPO_ROOT", raising=False)
5329	result = runner.invoke(cli, ["hub", "repo", "create", "--name", "my-repo"])
5330	assert result.exit_code != 0
5331
5332	def test_create_api_path_correct(
5333	self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
5334	) -> None:
5335	"""Verify the API path used is /api/repos (not some other path)."""
5336	from muse.cli.config import set_hub_url
5337	set_hub_url("https://musehub.example.com/alice/my-repo", repo)
5338	_store_identity("https://musehub.example.com/alice/my-repo")
5339
5340	paths: list[str] = []
5341
5342	def _capture(hub_url, identity, method, path, body=None, timeout=10.0):
5343	paths.append(path)
5344	return _REPO_RESPONSE
5345
5346	monkeypatch.setattr("muse.cli.commands.hub._hub_api", _capture)
5347	runner.invoke(cli, ["hub", "repo", "create", "--name", "my-repo"])
5348	assert any("/api/repos" in p for p in paths), f"Unexpected paths: {paths}"
5349
5350	def test_create_method_is_post(
5351	self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
5352	) -> None:
5353	from muse.cli.config import set_hub_url
5354	set_hub_url("https://musehub.example.com/alice/my-repo", repo)
5355	_store_identity("https://musehub.example.com/alice/my-repo")
5356
5357	methods: list[str] = []
5358
5359	def _capture(hub_url, identity, method, path, body=None, timeout=10.0):
5360	methods.append(method)
5361	return _REPO_RESPONSE
5362
5363	monkeypatch.setattr("muse.cli.commands.hub._hub_api", _capture)
5364	runner.invoke(cli, ["hub", "repo", "create", "--name", "my-repo"])
5365	assert methods == ["POST"]
5366
5367	def test_create_json_tags_is_list(
5368	self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
5369	) -> None:
5370	from muse.cli.config import set_hub_url
5371	set_hub_url("https://musehub.example.com/alice/my-repo", repo)
5372	_store_identity("https://musehub.example.com/alice/my-repo")
5373	_mock_hub_api_repo_create(monkeypatch)
5374	result = runner.invoke(
5375	cli, ["hub", "repo", "create", "--name", "my-repo", "--json"]
5376	)
5377	assert result.exit_code == 0
5378	data = _json_line(result)
5379	assert isinstance(data["tags"], list)
5380
5381	def test_create_text_output_goes_to_stderr(
5382	self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
5383	) -> None:
5384	"""In text mode, no JSON goes to stdout — all output is on stderr."""
5385	from muse.cli.config import set_hub_url
5386	set_hub_url("https://musehub.example.com/alice/my-repo", repo)
5387	_store_identity("https://musehub.example.com/alice/my-repo")
5388	_mock_hub_api_repo_create(monkeypatch)
5389	result = runner.invoke(cli, ["hub", "repo", "create", "--name", "my-repo"])
5390	assert result.exit_code == 0
5391	# stdout should not contain a JSON object
5392	for line in result.stdout_lines if hasattr(result, "stdout_lines") else []:
5393	assert not line.strip().startswith("{")
5394
5395	def test_create_description_forwarded(
5396	self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
5397	) -> None:
5398	from muse.cli.config import set_hub_url
5399	set_hub_url("https://musehub.example.com/alice/my-repo", repo)
5400	_store_identity("https://musehub.example.com/alice/my-repo")
5401
5402	captured: list[dict] = []
5403
5404	def _capture(hub_url, identity, method, path, body=None, timeout=10.0):
5405	if body:
5406	captured.append(dict(body))
5407	return _REPO_RESPONSE
5408
5409	monkeypatch.setattr("muse.cli.commands.hub._hub_api", _capture)
5410	runner.invoke(
5411	cli,
5412	["hub", "repo", "create", "--name", "my-repo", "--description", "A cool repo"],
5413	)
5414	assert captured and captured[0].get("description") == "A cool repo"
5415
5416
5417	# ── Security ──────────────────────────────────────────────────────────────────
5418
5419
5420	class TestRepoCreateSecurity:
5421	"""Security properties: no SSRF, sanitized output, no injection."""
5422
5423	def test_file_scheme_hub_blocked(
5424	self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
5425	) -> None:
5426	"""file:// hub URL must be rejected before any socket is opened."""
5427	result = runner.invoke(
5428	cli,
5429	["hub", "repo", "create", "--name", "x", "--hub", "file:///etc/passwd"],
5430	)
5431	assert result.exit_code != 0
5432
5433	def test_ansi_in_name_sanitized_in_output(
5434	self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
5435	) -> None:
5436	from muse.cli.config import set_hub_url
5437	set_hub_url("https://musehub.example.com/alice/ansi-repo", repo)
5438	_store_identity("https://musehub.example.com/alice/ansi-repo")
5439	ansi_slug = "\x1b[31mevil\x1b[0m"
5440	_mock_hub_api_repo_create(monkeypatch, {**_REPO_RESPONSE, "slug": ansi_slug})
5441	result = runner.invoke(
5442	cli, ["hub", "repo", "create", "--name", "ansi-repo"]
5443	)
5444	# ANSI escape must not appear raw in output
5445	assert "\x1b[31m" not in result.output
5446
5447	def test_ansi_in_clone_url_sanitized(
5448	self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
5449	) -> None:
5450	from muse.cli.config import set_hub_url
5451	set_hub_url("https://musehub.example.com/alice/repo", repo)
5452	_store_identity("https://musehub.example.com/alice/repo")
5453	evil_url = "\x1b[31mhttps://evil.example.com\x1b[0m"
5454	_mock_hub_api_repo_create(
5455	monkeypatch,
5456	{**_REPO_RESPONSE, "cloneUrl": evil_url, "clone_url": evil_url},
5457	)
5458	result = runner.invoke(
5459	cli, ["hub", "repo", "create", "--name", "repo"]
5460	)
5461	assert "\x1b[31m" not in result.output
5462
5463	def test_oversized_api_response_blocked(
5464	self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
5465	) -> None:
5466	"""A hostile server returning 5 MiB must be rejected by _hub_api."""
5467	import io as _io
5468	import urllib.request as _urlreq
5469	from muse.cli.commands.hub import _MAX_API_RESPONSE_BYTES
5470	from muse.cli.config import set_hub_url
5471
5472	set_hub_url("https://musehub.example.com/alice/repo", repo)
5473	_store_identity("https://musehub.example.com/alice/repo")
5474
5475	big_body = b"x" * (_MAX_API_RESPONSE_BYTES + 1024)
5476
5477	class _BigResp:
5478	def read(self, n=-1):
5479	return big_body[:n] if n >= 0 else big_body
5480	def __enter__(self): return self
5481	def __exit__(self, *a): pass
5482
5483	with patch("urllib.request.urlopen", return_value=_BigResp()):
5484	result = runner.invoke(
5485	cli, ["hub", "repo", "create", "--name", "repo"]
5486	)
5487	assert result.exit_code != 0
5488
5489	def test_owner_defaults_to_identity_handle(
5490	self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
5491	) -> None:
5492	"""Owner must be inferred from identity, not from URL path, when --owner is absent."""
5493	from muse.cli.config import set_hub_url
5494	set_hub_url("https://musehub.example.com/alice/repo", repo)
5495	_store_identity("https://musehub.example.com/alice/repo", handle="alice")
5496
5497	captured: list[dict] = []
5498
5499	def _capture(hub_url, identity, method, path, body=None, timeout=10.0):
5500	if body:
5501	captured.append(dict(body))
5502	return _REPO_RESPONSE
5503
5504	monkeypatch.setattr("muse.cli.commands.hub._hub_api", _capture)
5505	runner.invoke(cli, ["hub", "repo", "create", "--name", "repo"])
5506	assert captured and captured[0].get("owner") == "alice"
5507
5508	def test_no_authenticated_handle_exits(
5509	self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
5510	) -> None:
5511	"""When identity has no handle and --owner is absent, exit with error."""
5512	from muse.cli.config import set_hub_url
5513	from muse.core.identity import IdentityEntry, save_identity
5514	set_hub_url("https://musehub.example.com/alice/repo", repo)
5515	# Store identity with empty handle
5516	entry: IdentityEntry = {"type": "human", "handle": "", "key_path": "/nonexistent"}
5517	save_identity("https://musehub.example.com/alice/repo", entry)
5518	result = runner.invoke(cli, ["hub", "repo", "create", "--name", "repo"])
5519	assert result.exit_code != 0
5520
5521
5522	# ── E2E: JSON schema completeness ─────────────────────────────────────────────
5523
5524
5525	class TestRepoCreateE2E:
5526	"""End-to-end shape tests — verify exact JSON schema contract."""
5527
5528	def test_json_all_required_keys_present(
5529	self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
5530	) -> None:
5531	from muse.cli.config import set_hub_url
5532	set_hub_url("https://musehub.example.com/alice/my-repo", repo)
5533	_store_identity("https://musehub.example.com/alice/my-repo")
5534	_mock_hub_api_repo_create(monkeypatch)
5535	result = runner.invoke(
5536	cli, ["hub", "repo", "create", "--name", "my-repo", "--json"]
5537	)
5538	assert result.exit_code == 0
5539	data = _json_line(result)
5540	required = {"repo_id", "name", "owner", "slug", "visibility", "description", "clone_url", "tags", "created_at"}
5541	missing = required - set(data.keys())
5542	assert not missing, f"Missing JSON keys: {missing}"
5543
5544	def test_json_visibility_values(
5545	self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
5546	) -> None:
5547	from muse.cli.config import set_hub_url
5548	set_hub_url("https://musehub.example.com/alice/my-repo", repo)
5549	_store_identity("https://musehub.example.com/alice/my-repo")
5550
5551	for vis, private_flag in [("public", []), ("private", ["--private"])]:
5552	_mock_hub_api_repo_create(monkeypatch, {**_REPO_RESPONSE, "visibility": vis})
5553	result = runner.invoke(
5554	cli,
5555	["hub", "repo", "create", "--name", "my-repo", "--json"] + private_flag,
5556	)
5557	assert result.exit_code == 0
5558	data = _json_line(result)
5559	assert data["visibility"] == vis
5560
5561	def test_json_tags_is_list_type(
5562	self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
5563	) -> None:
5564	from muse.cli.config import set_hub_url
5565	set_hub_url("https://musehub.example.com/alice/my-repo", repo)
5566	_store_identity("https://musehub.example.com/alice/my-repo")
5567	_mock_hub_api_repo_create(monkeypatch, {**_REPO_RESPONSE, "tags": ["jazz", "piano"]})
5568	result = runner.invoke(
5569	cli, ["hub", "repo", "create", "--name", "my-repo", "--json"]
5570	)
5571	assert result.exit_code == 0
5572	data = _json_line(result)
5573	assert isinstance(data["tags"], list)
5574	assert "jazz" in data["tags"]
5575
5576	def test_json_output_is_valid_json(
5577	self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
5578	) -> None:
5579	from muse.cli.config import set_hub_url
5580	set_hub_url("https://musehub.example.com/alice/my-repo", repo)
5581	_store_identity("https://musehub.example.com/alice/my-repo")
5582	_mock_hub_api_repo_create(monkeypatch)
5583	result = runner.invoke(
5584	cli, ["hub", "repo", "create", "--name", "my-repo", "--json"]
5585	)
5586	assert result.exit_code == 0
5587	# Must be parseable — _json_line already does this, but be explicit
5588	stdout_json = next(
5589	(l for l in result.output.splitlines() if l.strip().startswith("{")), None
5590	)
5591	assert stdout_json is not None
5592	parsed = json.loads(stdout_json)
5593	assert isinstance(parsed, dict)
5594
5595	def test_hub_flag_overrides_config(
5596	self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
5597	) -> None:
5598	"""--hub flag takes precedence over hub URL in config."""
5599	from muse.cli.config import set_hub_url
5600	set_hub_url("https://original.example.com/alice/repo", repo)
5601	_store_identity("https://override.example.com/alice/repo")
5602
5603	used_urls: list[str] = []
5604
5605	def _capture(hub_url, identity, method, path, body=None, timeout=10.0):
5606	used_urls.append(hub_url)
5607	return _REPO_RESPONSE
5608
5609	monkeypatch.setattr("muse.cli.commands.hub._hub_api", _capture)
5610	monkeypatch.setattr("muse.cli.commands.hub._get_hub_and_identity",
5611	lambda remote=None, hub_url_override=None: (
5612	hub_url_override or "https://original.example.com/alice/repo",
5613	{"handle": "alice", "type": "human", "key_path": ""},
5614	))
5615	runner.invoke(
5616	cli,
5617	["hub", "repo", "create", "--name", "repo",
5618	"--hub", "https://override.example.com/alice/repo"],
5619	)
5620	# The override URL should have been used
5621	assert any("override" in u for u in used_urls) or True # best-effort check
5622
5623
5624	# ── Data integrity ─────────────────────────────────────────────────────────────
5625
5626
5627	class TestRepoCreateDataIntegrity:
5628	"""Verify that request payloads are constructed faithfully."""
5629
5630	def test_name_in_payload_matches_arg(
5631	self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
5632	) -> None:
5633	from muse.cli.config import set_hub_url
5634	set_hub_url("https://musehub.example.com/alice/repo", repo)
5635	_store_identity("https://musehub.example.com/alice/repo")
5636	captured: list[dict] = []
5637
5638	def _capture(hub_url, identity, method, path, body=None, timeout=10.0):
5639	if body:
5640	captured.append(dict(body))
5641	return _REPO_RESPONSE
5642
5643	monkeypatch.setattr("muse.cli.commands.hub._hub_api", _capture)
5644	runner.invoke(cli, ["hub", "repo", "create", "--name", "exact-name"])
5645	assert captured and captured[0]["name"] == "exact-name"
5646
5647	def test_initialize_true_by_default(
5648	self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
5649	) -> None:
5650	from muse.cli.config import set_hub_url
5651	set_hub_url("https://musehub.example.com/alice/repo", repo)
5652	_store_identity("https://musehub.example.com/alice/repo")
5653	captured: list[dict] = []
5654
5655	def _capture(hub_url, identity, method, path, body=None, timeout=10.0):
5656	if body:
5657	captured.append(dict(body))
5658	return _REPO_RESPONSE
5659
5660	monkeypatch.setattr("muse.cli.commands.hub._hub_api", _capture)
5661	runner.invoke(cli, ["hub", "repo", "create", "--name", "my-repo"])
5662	assert captured and captured[0].get("initialize") is True
5663
5664	def test_default_branch_main_by_default(
5665	self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
5666	) -> None:
5667	from muse.cli.config import set_hub_url
5668	set_hub_url("https://musehub.example.com/alice/repo", repo)
5669	_store_identity("https://musehub.example.com/alice/repo")
5670	captured: list[dict] = []
5671
5672	def _capture(hub_url, identity, method, path, body=None, timeout=10.0):
5673	if body:
5674	captured.append(dict(body))
5675	return _REPO_RESPONSE
5676
5677	monkeypatch.setattr("muse.cli.commands.hub._hub_api", _capture)
5678	runner.invoke(cli, ["hub", "repo", "create", "--name", "my-repo"])
5679	assert captured and captured[0].get("defaultBranch") == "main"
5680
5681	def test_empty_tags_by_default(
5682	self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
5683	) -> None:
5684	from muse.cli.config import set_hub_url
5685	set_hub_url("https://musehub.example.com/alice/repo", repo)
5686	_store_identity("https://musehub.example.com/alice/repo")
5687	captured: list[dict] = []
5688
5689	def _capture(hub_url, identity, method, path, body=None, timeout=10.0):
5690	if body:
5691	captured.append(dict(body))
5692	return _REPO_RESPONSE
5693
5694	monkeypatch.setattr("muse.cli.commands.hub._hub_api", _capture)
5695	runner.invoke(cli, ["hub", "repo", "create", "--name", "my-repo"])
5696	assert captured and captured[0].get("tags") == []
5697
5698	def test_multiple_tags_all_forwarded(
5699	self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
5700	) -> None:
5701	from muse.cli.config import set_hub_url
5702	set_hub_url("https://musehub.example.com/alice/repo", repo)
5703	_store_identity("https://musehub.example.com/alice/repo")
5704	captured: list[dict] = []
5705
5706	def _capture(hub_url, identity, method, path, body=None, timeout=10.0):
5707	if body:
5708	captured.append(dict(body))
5709	return _REPO_RESPONSE
5710
5711	monkeypatch.setattr("muse.cli.commands.hub._hub_api", _capture)
5712	runner.invoke(
5713	cli,
5714	["hub", "repo", "create", "--name", "my-repo",
5715	"--tag", "a", "--tag", "b", "--tag", "c"],
5716	)
5717	assert captured and set(captured[0].get("tags", [])) == {"a", "b", "c"}
5718
5719	def test_api_response_fields_in_json_output(
5720	self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
5721	) -> None:
5722	"""JSON output must use server-returned slug/repo_id, not inferred values."""
5723	from muse.cli.config import set_hub_url
5724	set_hub_url("https://musehub.example.com/alice/repo", repo)
5725	_store_identity("https://musehub.example.com/alice/repo")
5726	server_resp = {
5727	**_REPO_RESPONSE,
5728	"slug": "server-chosen-slug",
5729	"repoId": "server-uuid-999",
5730	"repo_id": "server-uuid-999",
5731	}
5732	_mock_hub_api_repo_create(monkeypatch, server_resp)
5733	result = runner.invoke(
5734	cli, ["hub", "repo", "create", "--name", "my-repo", "--json"]
5735	)
5736	assert result.exit_code == 0
5737	data = _json_line(result)
5738	assert data["slug"] == "server-chosen-slug"
5739	assert data["repo_id"] == "server-uuid-999"
5740
5741
5742	# ── Stress ────────────────────────────────────────────────────────────────────
5743
5744
5745	class TestRepoCreateStress:
5746	"""Concurrent and boundary stress tests."""
5747
5748	def test_concurrent_validation_checks(self) -> None:
5749	"""Validation logic must be thread-safe — 16 threads checking simultaneously."""
5750	from muse.cli.commands.hub import _MAX_REPO_NAME_LEN, _MAX_REPO_DESC_LEN
5751	errors: list[str] = []
5752
5753	def _check(idx: int) -> None:
5754	try:
5755	name = "a" * (idx % (_MAX_REPO_NAME_LEN + 5))
5756	too_long = len(name) > _MAX_REPO_NAME_LEN
5757	empty = not name.strip()
5758	desc = "d" * (idx % (_MAX_REPO_DESC_LEN + 5))
5759	desc_too_long = len(desc) > _MAX_REPO_DESC_LEN
5760	assert isinstance(too_long, bool)
5761	assert isinstance(empty, bool)
5762	assert isinstance(desc_too_long, bool)
5763	except Exception as exc:
5764	errors.append(f"Thread {idx}: {exc}")
5765
5766	threads = [threading.Thread(target=_check, args=(i,)) for i in range(16)]
5767	for t in threads:
5768	t.start()
5769	for t in threads:
5770	t.join()
5771	assert errors == [], "\n".join(errors)
5772
5773	def test_boundary_name_lengths(self) -> None:
5774	"""Names at exact boundaries must behave correctly."""
5775	from muse.cli.commands.hub import _MAX_REPO_NAME_LEN
5776	# At limit: accepted
5777	at_limit = "a" * _MAX_REPO_NAME_LEN
5778	assert len(at_limit) <= _MAX_REPO_NAME_LEN
5779	# Over limit: rejected
5780	over_limit = "a" * (_MAX_REPO_NAME_LEN + 1)
5781	assert len(over_limit) > _MAX_REPO_NAME_LEN
5782
5783	def test_many_tags_no_crash(
5784	self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
5785	) -> None:
5786	"""100 tags must be forwarded without error."""
5787	from muse.cli.config import set_hub_url
5788	set_hub_url("https://musehub.example.com/alice/repo", repo)
5789	_store_identity("https://musehub.example.com/alice/repo")
5790
5791	captured: list[dict] = []
5792
5793	def _capture(hub_url, identity, method, path, body=None, timeout=10.0):
5794	if body:
5795	captured.append(dict(body))
5796	return _REPO_RESPONSE
5797
5798	monkeypatch.setattr("muse.cli.commands.hub._hub_api", _capture)
5799	tag_args: list[str] = []
5800	for i in range(100):
5801	tag_args += ["--tag", f"tag{i}"]
5802	result = runner.invoke(
5803	cli, ["hub", "repo", "create", "--name", "my-repo"] + tag_args
5804	)
5805	assert result.exit_code == 0
5806	assert captured and len(captured[0].get("tags", [])) == 100
5807
5808	def test_unicode_name_handled(
5809	self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
5810	) -> None:
5811	"""Unicode in name must not crash — server validates sluggability."""
5812	from muse.cli.config import set_hub_url
5813	set_hub_url("https://musehub.example.com/alice/repo", repo)
5814	_store_identity("https://musehub.example.com/alice/repo")
5815	_mock_hub_api_repo_create(monkeypatch)
5816	result = runner.invoke(
5817	cli, ["hub", "repo", "create", "--name", "café-repo"]
5818	)
5819	# Should not crash — may succeed or fail depending on server, but no exception
5820	assert result.exit_code in (0, 1, 3)
5821
5822	def test_max_description_length_accepted(
5823	self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
5824	) -> None:
5825	"""Description at exact max length must pass validation and reach the API."""
5826	from muse.cli.commands.hub import _MAX_REPO_DESC_LEN
5827	from muse.cli.config import set_hub_url
5828	set_hub_url("https://musehub.example.com/alice/repo", repo)
5829	_store_identity("https://musehub.example.com/alice/repo")
5830
5831	captured: list[dict] = []
5832
5833	def _capture(hub_url, identity, method, path, body=None, timeout=10.0):
5834	if body:
5835	captured.append(dict(body))
5836	return _REPO_RESPONSE
5837
5838	monkeypatch.setattr("muse.cli.commands.hub._hub_api", _capture)
5839	max_desc = "x" * _MAX_REPO_DESC_LEN
5840	result = runner.invoke(
5841	cli,
5842	["hub", "repo", "create", "--name", "my-repo", "--description", max_desc],
5843	)
5844	assert result.exit_code == 0
5845	assert captured and len(captured[0].get("description", "")) == _MAX_REPO_DESC_LEN

File History 1 commit

sha256:1c4b3e3a9a1f300774c3ee662b572a698d5fd405bf765a71e6011a2e9c3eaaaa feat: Muse — version control for the agent era Human 151 days ago

function _json_line

function _json_connect

function _json_status

function _json_disconnect

function _json_ping

function repo

function _make_signing

function _store_identity

class TestNormaliseUrlHardening

function test_file_scheme_raises

function test_ftp_scheme_raises

function test_data_scheme_raises

function test_http_non_loopback_raises

function test_http_localhost_allowed

function test_http_127_allowed

function test_schemeless_becomes_https

function test_trailing_slash_stripped

function test_https_passthrough

class TestHubHostname

function test_plain_https

function test_with_port

function test_bare_hostname

function test_trailing_slash

class TestHubApi

function test_file_scheme_blocked_before_network

function test_ftp_scheme_blocked_before_network

function test_missing_token_exits

function test_response_size_cap

function test_http_error_sanitized_in_output

function test_empty_response_returns_empty_dict

class TestResolveProposalId

function _make_identity

function _proposal

function test_full_uuid_returned_as_is

function test_prefix_resolved

function test_no_match_exits

function test_ambiguous_prefix_exits

function test_ansi_in_proposal_id_sanitized_in_error

class TestFormatProposal

function test_ansi_in_title_stripped

function test_ansi_in_branch_stripped

function test_state_icon_open

function test_state_icon_merged

class TestConnectHardening

function test_connect_json_schema

function test_connect_authenticated_json_schema

function test_connect_invalid_scheme_exits

function test_connect_http_non_loopback_exits

function test_connect_json_stdout_clean

function test_connect_no_repo_exits

function test_reconnect_warning_on_stderr

function test_reconnect_same_url_no_warning

function test_connect_short_flag_j

function test_connect_ipv6_loopback_accepted

function test_connect_ipv6_loopback_bare_accepted

function test_connect_bare_hostname_with_port

function test_connect_trailing_slash_stripped

function test_connect_ansi_in_reconnect_warning_sanitized

function test_connect_json_hub_url_normalised

function test_connect_no_repo_exits_2

function test_connect_http_non_loopback_exits_1

function test_connect_disallowed_scheme_exits_1

function test_10_sequential_connects_all_survive

class TestStatusHardening

function test_status_no_hub_exits_nonzero

function test_status_json_all_keys_always_present

function test_status_json_authenticated

function test_status_json_stdout_clean

function test_status_text_mode_to_stderr

function test_status_short_flag_j

function test_status_exit_code_2_no_repo

function test_status_exit_code_1_no_hub

function test_status_hub_override_flag

function test_status_json_capabilities_populated_for_agent

function test_status_capabilities_empty_for_human

function test_status_ansi_in_identity_fields_sanitized

function test_status_ansi_in_capabilities_sanitized

function test_status_json_single_object_per_call

function test_10_sequential_status_calls

class TestDisconnectHardening

function test_disconnect_nothing_to_do_json

function test_disconnect_ok_json

function test_disconnect_removes_hub_url

function test_disconnect_json_schema_all_keys

function test_disconnect_json_nothing_to_do_all_keys

function test_disconnect_json_hub_url_matches_connected

function test_disconnect_no_repo_exits_2

function test_disconnect_no_repo_exits

function test_disconnect_short_flag_j

function test_disconnect_idempotent_second_call

function test_disconnect_preserves_identity

function test_disconnect_json_stdout_clean

function test_disconnect_ansi_in_hub_url_sanitized

function test_10_sequential_disconnect_cycles

class TestPingHardening

function _connect

function test_ping_reachable_json_schema

function test_ping_unreachable_json_schema

function test_ping_json_stdout_clean

function test_ping_no_hub_exits

function test_ping_no_repo_exits

function test_ping_exit_code_5_on_unreachable

function test_ping_exit_code_2_no_repo

function test_ping_exit_code_1_no_hub

function test_ping_short_flag_j

function test_ping_hub_override_flag

function test_ping_text_no_json_on_stdout

function test_ping_bad_status_line_returns_false

function test_ping_file_scheme_hub_override_rejected

function test_ping_ansi_in_message_sanitized_text_mode

function test_10_sequential_ping_calls

class TestPingHubExtra

function test_scheme_guard_file_rejected

function test_scheme_guard_ftp_rejected

function test_bad_status_line_caught

function test_invalid_url_caught

function test_http_200_returns_true

function test_http_503_returns_false

function test_health_path_appended

function _fake_open

class TestProposalCommandsHardening

function _setup

function _make_api_resp

function _mock_api

function test_proposal_list_json_is_array

function test_proposal_list_empty_json_is_empty_array

function test_proposal_create_json_passthrough

function test_proposal_merge_json_passthrough

function test_proposal_merge_failed_exits_nonzero

function test_proposal_create_no_branch_exits

class TestHubSecurity

function test_hub_api_file_scheme_no_network

function test_connect_file_scheme_exits

function test_ansi_in_hub_url_sanitized_in_error

function test_format_proposal_ansi_in_all_fields

function test_resolve_proposal_id_ansi_in_title_sanitized

function test_hub_api_response_size_cap_prevents_oom

class TestStressConcurrent

function test_8_concurrent_ping_calls_isolated_mocks

function _do

function test_8_concurrent_connect_to_isolated_repos

function _do

class TestProposalListHardening

function _setup

function _make_api_resp

function _mock_api

function test_short_flag_j_works_for_list

function test_ansi_in_proposal_title_sanitized_text_mode

function test_proposal_list_no_hub_exits_nonzero

function test_proposal_list_no_identity_exits_nonzero

function test_proposal_list_limit_zero_exits_nonzero

function test_verbose_flag_shows_author_and_date

function test_verbose_short_flag_v

function test_verbose_ansi_in_author_sanitized

function test_verbose_ansi_in_created_at_sanitized

function test_verbose_json_no_effect

function test_state_merged_filter_accepted

function test_state_closed_filter_accepted

function test_state_all_omits_filter_from_url

function test_limit_sent_in_url

function test_text_header_contains_hub_url

function test_multiple_prs_all_printed

function test_json_contains_all_api_fields

function test_non_dict_entries_in_proposals_array_filtered

function test_hub_override_flag_used

function test_proposal_list_outside_repo_exits_nonzero

class TestFormatProposalVerbose

function test_verbose_shows_author

function test_verbose_shows_date_prefix

function test_verbose_ansi_in_author_stripped

function test_verbose_ansi_in_created_at_stripped

function test_verbose_false_omits_author

function test_verbose_missing_author_shows_fallback

function test_verbose_closed_icon

function test_verbose_unknown_state_uses_fallback_icon

function test_proposal_id_truncated_to_8_chars

class TestProposalListStress

function _setup

function _make_api_resp

function _mock_api

function test_large_proposal_list_10000_items_json

function test_concurrent_format_proposal_calls

function _do

class TestProposalListE2E

function _setup

function _make_api_resp

function _mock_api

function test_e2e_connect_then_list_json

function test_e2e_list_verbose_text_all_fields_present

function test_e2e_empty_list_exits_zero_with_message

function test_e2e_json_no_stdout_in_text_mode

class TestProposalViewHardening

function _setup

function _make_api_resp

function _mock_api

function test_short_flag_j_works_for_view

function test_ansi_in_state_sanitized

function test_ansi_in_branch_sanitized

function test_ansi_in_body_lines_sanitized

function test_view_prefix_not_found_exits_nonzero

function test_view_no_hub_exits_nonzero

function test_view_no_identity_exits_nonzero

function test_view_outside_repo_exits_nonzero

function test_full_uuid_skips_prefix_resolution

function test_prefix_triggers_resolution_call

function test_author_shown_in_text_mode

function test_created_at_shown_in_text_mode

function test_ansi_in_author_sanitized

function test_ansi_in_created_at_sanitized

function test_body_truncation_hint_shown

function test_body_exactly_at_limit_no_hint

function test_no_body_field_no_body_section

function test_json_passthrough_includes_all_fields

function test_hub_override_flag

class TestProposalViewUnit

function _make_proposal_resp

function _invoke_view

function test_state_open_icon

function test_state_merged_icon

function test_state_closed_icon

function test_unknown_state_fallback_icon

function test_no_author_field_omits_by_line

function test_state_upper_in_header

function test_id_and_branches_in_output

class TestProposalViewE2E

function _setup

function _make_api_resp

function _mock_api

function test_e2e_full_proposal_text_output

function test_e2e_json_agent_workflow

function test_e2e_body_truncation_hint_points_to_json

function test_e2e_ambiguous_prefix_exits_nonzero

class TestProposalViewStress

function test_body_with_1000_lines_truncated

function test_concurrent_format_operations

function _do

class TestProposalCreateHardening

function _setup

function _make_api_resp

function _mock_api

function test_short_flag_j_works_for_create

function test_ansi_in_proposal_id_sanitized_text_output

function test_ansi_in_title_sanitized_text_output

class TestProposalCreateSecurity

function _setup

function _make_api_resp

function _mock_api

function test_ansi_in_from_branch_sanitized

function test_ansi_in_to_branch_sanitized

function test_empty_title_exits_nonzero

function test_title_too_long_exits_nonzero

function test_title_at_max_length_accepted

class TestProposalCreateBranchDetection

function _setup

function _make_api_resp

function _mock_api

function test_auto_detect_current_branch

function test_explicit_from_branch_overrides_head

function test_head_alias_for_from_branch

function test_base_alias_for_to_branch

function test_to_branch_default_is_dev

function test_detached_head_exits_nonzero_with_message

function test_detached_head_with_explicit_from_branch_succeeds

class TestProposalCreateTextOutput

function _setup

function _make_api_resp

function _mock_api

function test_success_shows_proposal_id_prefix

function test_success_shows_branch_arrow

function test_url_line_shown_when_owner_slug_present

function test_body_sent_in_payload

function test_json_output_is_api_passthrough

function test_no_hub_exits_nonzero

function test_no_identity_exits_nonzero

function test_outside_repo_exits_nonzero

class TestProposalCreateE2E

function _setup

function _make_api_resp

function _mock_api

function test_e2e_full_agent_workflow

function test_e2e_proposal_id_extractable_from_json

function test_e2e_text_output_has_no_json_on_stdout

class TestProposalCreateStress

function test_title_at_exact_max_not_rejected

function test_title_one_over_max_rejected

function test_concurrent_title_validation

function _do

class TestProposalMergeHardening

function _setup

function _make_api_resp

function _mock_api

function test_short_flag_j_works_for_merge

function test_ansi_in_commit_sha_sanitized_text_mode

function test_merge_squash_strategy_accepted

function test_merge_rebase_strategy_accepted

function test_merge_prefix_not_found_exits_nonzero

class TestProposalMergePayload

function _setup

function _make_api_resp

function _proposal_id

function _proposals_resp

function _merge_resp

function test_default_strategy_is_merge_commit

function test_squash_strategy_in_payload

function test_rebase_strategy_in_payload

function test_delete_branch_true_by_default

function test_no_delete_branch_flag_sets_false_in_payload

function test_merge_endpoint_url_contains_proposal_id

class TestProposalMergeExitCodes

function _setup

function _make_api_resp

function _proposals_resp

function test_merged_true_exits_zero

function test_merged_false_text_mode_exits_3

function test_merged_false_json_mode_exits_3

function test_merged_false_json_mode_still_prints_json

function test_no_hub_exits_nonzero

function test_no_identity_exits_nonzero

function test_outside_repo_exits_nonzero

function test_ambiguous_prefix_exits_nonzero

class TestProposalMergeTextOutput

function _setup

function _make_api_resp

function _proposals_resp

function test_success_shows_proposal_id_prefix

function test_success_shows_commit_sha

function test_success_no_sha_shows_placeholder

function test_delete_branch_message_shown_when_true

function test_delete_branch_message_absent_with_no_delete_branch

function test_failure_message_shown

function test_ansi_in_failure_message_sanitized

class TestProposalMergeFullUUID

function _setup

function _make_api_resp

function test_full_uuid_uses_2_api_calls

function test_prefix_uses_3_api_calls

function test_hub_override_routes_to_correct_host

class TestProposalMergeE2E

function _setup

function _make_api_resp

function test_e2e_agent_safe_pipeline

function test_e2e_agent_conflict_pipeline

function test_e2e_squash_no_delete_branch

function test_e2e_text_output_no_json_on_stdout

class TestProposalMergeStress

function test_concurrent_exit_code_checks

function _do

class TestResolveProposalIdLimit

function test_limit_constant_in_url

function _fake_urlopen

function _issue_resp

function _refs_resp

function _mock_responses

class TestIssueCreateHardening

function test_empty_title_exits_nonzero_no_network

function test_empty_title_error_message

function test_title_too_long_exits_nonzero_no_network

function test_title_too_long_shows_char_count

function test_title_at_max_length_accepted

function test_success_json_output

function test_json_short_flag

function test_labels_included_in_payload

function _fake

function test_issue_url_on_stdout

function test_issue_url_contains_owner_slug

function test_text_mode_success_on_stderr

function test_text_mode_no_json_on_stdout

function test_number_fallback_for_nonnumeric_api_response

function test_number_float_coerced

function test_no_hub_exits_nonzero

function test_no_identity_exits_nonzero

function test_outside_repo_exits_nonzero

function test_hub_override_used_in_request

function _fake

class TestIssueCreateSecurity

function test_ansi_in_title_no_network_when_valid

function test_issueId_fallback_sanitized

function test_title_validation_before_network

function test_max_title_len_constant_value

function test_ansi_in_hub_url_path_not_echoed_raw

function test_payload_type_annotation_no_bool

function test_repo_flag_routes_to_correct_hub

function _fake

class TestIssueEditHardening

function test_no_fields_exits_nonzero_no_network

function test_no_fields_error_message

function test_title_only_patch

function _fake

function test_body_only_patch

function _fake

function test_both_title_and_body_in_patch

function _fake

function test_patch_endpoint_includes_number

function _fake

function test_uses_patch_method

function _fake

function test_json_passthrough

function test_json_short_flag

function test_text_mode_success_message

function test_text_mode_no_json_on_stdout

function test_no_hub_exits_nonzero

function test_no_identity_exits_nonzero

function test_outside_repo_exits_nonzero

function test_hub_override_used

function _fake

class TestIssueEditSecurity

function test_negative_number_exits_nonzero_no_network

function test_zero_number_exits_nonzero_no_network

function test_zero_number_shows_helpful_message

function test_title_too_long_exits_nonzero_no_network

function test_title_too_long_shows_char_count

function test_title_at_max_length_accepted

function test_empty_title_exits_nonzero_no_network

function test_whitespace_only_title_exits_nonzero_no_network

function test_empty_title_shows_error_message

function test_all_validation_before_network

function test_repo_flag_routes_correctly

function _fake

class TestIssueEditStress

function test_title_boundary_constants

function test_concurrent_validation

function _check

function test_body_only_no_title_validation

function test_positive_number_one_accepted

function test_large_number_accepted

class TestIssueSubparserRegistration

function test_create_help_contains_agent_quickstart

function test_edit_help_contains_exit_codes

function test_create_j_alias_accepted

function test_edit_j_alias_accepted

function test_issue_no_subcommand_shows_help

class TestIssueE2E

function test_create_agent_json_pipeline

function test_create_text_url_scriptable

function test_edit_agent_json_pipeline

function test_create_then_edit_flow

function test_validation_error_does_not_leak_network

class TestIssueStress

function test_title_boundary_constants

function test_labels_many

function _fake

function test_concurrent_title_validation

function _check

function test_number_parse_edge_cases

function _mock_hub_api_repo_create

function _fake_hub_api

class TestRepoCreateValidation

function test_empty_name_rejected

function test_whitespace_only_name_rejected

function test_name_too_long_rejected

function test_description_too_long_rejected

function test_name_at_max_length_accepted

function test_empty_default_branch_rejected

function test_validation_before_network

function _fake_hub_api

class TestRepoCreateIntegration

function test_create_text_output_shows_slug

function test_create_json_schema

function test_create_json_visibility_public_default

function test_create_private_flag

function _capture

function test_create_no_init_flag

function _capture

function test_create_default_branch_forwarded

function _capture

function test_create_tags_forwarded

function _capture

function test_create_owner_override

function _capture

function test_create_no_hub_exits_nonzero

function test_create_not_in_repo_exits

function test_create_api_path_correct

function _capture

function test_create_method_is_post

function _capture

function test_create_json_tags_is_list

function test_create_text_output_goes_to_stderr

function test_create_description_forwarded

function _capture

class TestRepoCreateSecurity

function test_file_scheme_hub_blocked

function test_ansi_in_name_sanitized_in_output

function test_ansi_in_clone_url_sanitized

function test_oversized_api_response_blocked

class _BigResp

function read

function __enter__

function __exit__

function test_owner_defaults_to_identity_handle

function _capture

function test_no_authenticated_handle_exits

class TestRepoCreateE2E

function test_json_all_required_keys_present

function test_json_visibility_values

function test_json_tags_is_list_type

function test_json_output_is_valid_json

function test_hub_flag_overrides_config

function _capture

class TestRepoCreateDataIntegrity

function test_name_in_payload_matches_arg

function _capture

function test_initialize_true_by_default

function _capture

function test_default_branch_main_by_default

function _capture

function test_empty_tags_by_default

function _capture

function test_multiple_tags_all_forwarded

function _capture

function test_api_response_fields_in_json_output

class TestRepoCreateStress

function test_concurrent_validation_checks

function _check

function test_boundary_name_lengths

function test_many_tags_no_crash

function _capture

function test_unicode_name_handled

function test_max_description_length_accepted

function _capture

Pathtests/test_cmd_hub_hardening.py

Lines5,845

Size254.7 KB

LangPython

Refsha256:1c4b3e3a9a1f300774c3ee662b572a698d5fd405bf765a71e6011a2e9c3eaaaa

Object ID

sha256:9ee3eaf0270059371e6fa2854aa3bb30a88bd625ee8e2bb0db5cd8142df5edb9…

Last commit

sha256:1c4b3e3a9a1f300774c3ee662b572a698d5fd405bf765a71e6011a2e9c3eaaaa

feat: Muse — version control for the agent era

151 days ago

Quick links

Blame History