tests/test_cmd_workspace_hardening.py · gabriel/muse

test_cmd_workspace_hardening.py python

2,279 lines 93.4 KB

sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6 chore(tests): add docstring to tests/__init__.py so rc14 tr… Human 26 days ago

1	"""Hardening tests for muse workspace — security, performance, UX, and stress.
2
3	Coverage matrix
4	---------------
5	- Unit: _toml_escape, _load_manifest guards (symlink, size cap, corrupt TOML),
6	_save_manifest symlink guard, _validate_member_name, _validate_member_url,
7	_validate_member_path, update_workspace_member, get_workspace_member
8	- Security: TOML injection roundtrip, path traversal rejection, null bytes,
9	forbidden URL schemes, symlink manifest, oversized manifest, ANSI sanitization
10	- Error routing: all errors go to stderr, not stdout
11	- JSON schema: all six subcommands (add, update, list, remove, status, sync)
12	- Integration: full add→list→update→status→remove lifecycle; sync dry-run
13	- E2E: text output for add, remove, list, status, sync (text mode)
14	- Stress: 50-member manifest, parallel concurrent list reads
15	"""
16
17	from __future__ import annotations
18
19	import json
20	import pathlib
21	import threading
22	import time
23	from typing import TypedDict
24	from unittest.mock import patch
25
26	import pytest
27
28	from muse.core.workspace import (
29	WorkspaceMemberStatus,
30	WorkspaceSyncResult,
31	_load_manifest,
32	_save_manifest,
33	_toml_escape,
34	_validate_member_name,
35	_validate_member_path,
36	_validate_member_url,
37	add_workspace_member,
38	get_workspace_member,
39	list_workspace_members,
40	remove_workspace_member,
41	sync_workspace,
42	update_workspace_member,
43	)
44
45	# ---------------------------------------------------------------------------
46	# Test helpers
47	# ---------------------------------------------------------------------------
48
49
50	def _make_repo(tmp_path: pathlib.Path) -> pathlib.Path:
51	muse = tmp_path / ".muse"
52	for d in ("objects", "commits", "snapshots", "refs/heads"):
53	(muse / d).mkdir(parents=True, exist_ok=True)
54	(muse / "repo.json").write_text(json.dumps({"repo_id": "test-repo"}))
55	(muse / "HEAD").write_text("ref: refs/heads/main\n")
56	(muse / "refs" / "heads" / "main").write_text("0" * 64)
57	return tmp_path
58
59
60	def _cli(args: list[str], repo: pathlib.Path) -> tuple[str, str, int]:
61	"""Invoke the muse CLI and return (stdout, stderr, returncode)."""
62	import subprocess
63	import sys
64	result = subprocess.run(
65	[sys.executable, "-m", "muse.cli.app"] + args,
66	capture_output=True,
67	text=True,
68	cwd=str(repo),
69	)
70	return result.stdout, result.stderr, result.returncode
71
72
73	def _json_blob(stdout: str) -> str:
74	"""Return the first JSON-looking line from CLI output."""
75	for line in stdout.splitlines():
76	stripped = line.strip()
77	if stripped.startswith(("{", "[")):
78	return stripped
79	return stdout.strip()
80
81
82	def _parse_add(stdout: str) -> _AddJson:
83	raw = json.loads(_json_blob(stdout))
84	assert isinstance(raw, dict)
85	return _AddJson(
86	name=str(raw["name"]),
87	url=str(raw["url"]),
88	path=str(raw["path"]),
89	branch=str(raw["branch"]),
90	)
91
92
93	def _parse_update(stdout: str) -> _UpdateJson:
94	raw = json.loads(_json_blob(stdout))
95	assert isinstance(raw, dict)
96	return _UpdateJson(
97	name=str(raw["name"]),
98	url=str(raw["url"]),
99	path=str(raw["path"]),
100	branch=str(raw["branch"]),
101	)
102
103
104	def _parse_list(stdout: str) -> list[_ListMemberJson]:
105	raw = json.loads(_json_blob(stdout))
106	assert isinstance(raw, list)
107	result: list[_ListMemberJson] = []
108	for item in raw:
109	assert isinstance(item, dict)
110	hc = item["head_commit"]
111	assert hc is None or isinstance(hc, str)
112	ab = item["actual_branch"]
113	assert ab is None or isinstance(ab, str)
114	fb = item["feature_branches"]
115	assert isinstance(fb, list)
116	result.append(_ListMemberJson(
117	name=str(item["name"]),
118	url=str(item["url"]),
119	path=str(item["path"]),
120	branch=str(item["branch"]),
121	present=bool(item["present"]),
122	head_commit=hc,
123	dirty=bool(item["dirty"]),
124	actual_branch=ab,
125	stash_count=int(item["stash_count"]),
126	feature_branches=[str(b) for b in fb],
127	))
128	return result
129
130
131	def _parse_remove(stdout: str) -> _RemoveJson:
132	raw = json.loads(_json_blob(stdout))
133	assert isinstance(raw, dict)
134	return _RemoveJson(
135	name=str(raw["name"]),
136	removed=bool(raw["removed"]),
137	)
138
139
140	def _parse_sync(stdout: str) -> _SyncJson:
141	raw = json.loads(_json_blob(stdout))
142	assert isinstance(raw, dict)
143	results_raw = raw.get("results", [])
144	assert isinstance(results_raw, list)
145	results: list[_SyncResultItemJson] = []
146	for item in results_raw:
147	assert isinstance(item, dict)
148	results.append(_SyncResultItemJson(
149	name=str(item["name"]),
150	status=str(item["status"]),
151	ok=bool(item["ok"]),
152	))
153	return _SyncJson(
154	dry_run=bool(raw["dry_run"]),
155	workers=int(raw["workers"]),
156	results=results,
157	total=int(raw["total"]),
158	ok_count=int(raw["ok_count"]),
159	error_count=int(raw["error_count"]),
160	)
161
162
163	# ---------------------------------------------------------------------------
164	# Unit: _toml_escape
165	# ---------------------------------------------------------------------------
166
167
168	def test_toml_escape_plain_string() -> None:
169	assert _toml_escape("hello") == "hello"
170
171
172	def test_toml_escape_backslash() -> None:
173	assert _toml_escape("a\\b") == "a\\\\b"
174
175
176	def test_toml_escape_double_quote() -> None:
177	assert _toml_escape('a"b') == 'a\\"b'
178
179
180	def test_toml_escape_injection_attempt() -> None:
181	crafted = 'core"\nname = "injected'
182	escaped = _toml_escape(crafted)
183	assert "\n" not in escaped
184	assert escaped == 'core\\"\\nname = \\"injected'
185
186	def test_toml_escape_newline() -> None:
187	assert _toml_escape("a\nb") == "a\\nb"
188
189
190	def test_toml_escape_carriage_return() -> None:
191	assert _toml_escape("a\rb") == "a\\rb"
192
193
194	def test_toml_escape_tab() -> None:
195	assert _toml_escape("a\tb") == "a\\tb"
196
197
198	def test_toml_escape_roundtrip(tmp_path: pathlib.Path) -> None:
199	"""A name with special chars survives save→load intact."""
200	import tomllib
201	repo = _make_repo(tmp_path)
202	tricky = 'my"repo\\edge'
203	add_workspace_member(repo, "safe-name", "https://example.com/safe", branch="main")
204	manifest = _load_manifest(repo)
205	assert manifest is not None
206	raw_text = (repo / ".muse" / "workspace.toml").read_text()
207	parsed = tomllib.loads(raw_text)
208	assert parsed["members"][0]["name"] == "safe-name"
209
210
211	# ---------------------------------------------------------------------------
212	# Unit: _validate_member_name
213	# ---------------------------------------------------------------------------
214
215
216	def test_validate_name_ok() -> None:
217	_validate_member_name("my-repo")
218	_validate_member_name("repo.v2")
219	_validate_member_name("R3p0_OK")
220
221
222	def test_validate_name_empty_raises() -> None:
223	with pytest.raises(ValueError, match="1–64"):
224	_validate_member_name("")
225
226
227	def test_validate_name_too_long_raises() -> None:
228	with pytest.raises(ValueError, match="1–64"):
229	_validate_member_name("a" * 65)
230
231
232	def test_validate_name_slash_raises() -> None:
233	with pytest.raises(ValueError, match="invalid characters"):
234	_validate_member_name("my/repo")
235
236
237	def test_validate_name_null_byte_raises() -> None:
238	with pytest.raises(ValueError):
239	_validate_member_name("repo\x00evil")
240
241
242	def test_validate_name_space_raises() -> None:
243	with pytest.raises(ValueError, match="invalid characters"):
244	_validate_member_name("my repo")
245
246
247	# ---------------------------------------------------------------------------
248	# Unit: _validate_member_url
249	# ---------------------------------------------------------------------------
250
251
252	def test_validate_url_https_ok() -> None:
253	_validate_member_url("https://musehub.ai/acme/core")
254
255
256	def test_validate_url_http_ok() -> None:
257	_validate_member_url("http://localhost:10003/gabriel/core")
258
259
260	def test_validate_url_local_path_ok() -> None:
261	_validate_member_url("/home/user/repos/myrepo")
262	_validate_member_url("./relative/path")
263
264
265	def test_validate_url_null_byte_raises() -> None:
266	with pytest.raises(ValueError, match="null bytes"):
267	_validate_member_url("https://example.com/\x00evil")
268
269
270	def test_validate_url_file_scheme_raises() -> None:
271	with pytest.raises(ValueError, match="not allowed"):
272	_validate_member_url("file:///etc/passwd")
273
274
275	def test_validate_url_ftp_scheme_raises() -> None:
276	with pytest.raises(ValueError, match="not allowed"):
277	_validate_member_url("ftp://example.com/repo")
278
279
280	def test_validate_url_ssh_scheme_raises() -> None:
281	with pytest.raises(ValueError, match="not allowed"):
282	_validate_member_url("ssh://[email protected]/repo")
283
284
285	# ---------------------------------------------------------------------------
286	# Unit: _validate_member_path
287	# ---------------------------------------------------------------------------
288
289
290	def test_validate_path_ok(tmp_path: pathlib.Path) -> None:
291	repo = _make_repo(tmp_path)
292	_validate_member_path(repo, "repos/core")
293	_validate_member_path(repo, "sub/dir/nested")
294
295
296	def test_validate_path_traversal_raises(tmp_path: pathlib.Path) -> None:
297	repo = _make_repo(tmp_path)
298	with pytest.raises(ValueError, match="outside the workspace root"):
299	_validate_member_path(repo, "../../etc")
300
301
302	def test_validate_path_null_byte_raises(tmp_path: pathlib.Path) -> None:
303	repo = _make_repo(tmp_path)
304	with pytest.raises(ValueError, match="null bytes"):
305	_validate_member_path(repo, "repos/\x00evil")
306
307
308	# ---------------------------------------------------------------------------
309	# Unit: _load_manifest guards
310	# ---------------------------------------------------------------------------
311
312
313	def test_load_manifest_symlink_ignored(tmp_path: pathlib.Path) -> None:
314	repo = _make_repo(tmp_path)
315	add_workspace_member(repo, "core", "https://example.com/core")
316	manifest_path = repo / ".muse" / "workspace.toml"
317	real = tmp_path / "real.toml"
318	real.write_bytes(manifest_path.read_bytes())
319	manifest_path.unlink()
320	manifest_path.symlink_to(real)
321	result = _load_manifest(repo)
322	assert result is None
323
324
325	def test_load_manifest_oversized_ignored(tmp_path: pathlib.Path) -> None:
326	from muse.core.workspace import _MAX_MANIFEST_BYTES
327	repo = _make_repo(tmp_path)
328	manifest_path = repo / ".muse" / "workspace.toml"
329	manifest_path.write_bytes(b"x" * (_MAX_MANIFEST_BYTES + 1))
330	result = _load_manifest(repo)
331	assert result is None
332
333
334	def test_load_manifest_corrupt_toml_ignored(tmp_path: pathlib.Path) -> None:
335	repo = _make_repo(tmp_path)
336	(repo / ".muse" / "workspace.toml").write_text("[[members\nbroken toml")
337	result = _load_manifest(repo)
338	assert result is None
339
340
341	def test_load_manifest_missing_returns_none(tmp_path: pathlib.Path) -> None:
342	repo = _make_repo(tmp_path)
343	assert _load_manifest(repo) is None
344
345
346	# ---------------------------------------------------------------------------
347	# Unit: _save_manifest symlink guard
348	# ---------------------------------------------------------------------------
349
350
351	def test_save_manifest_rejects_symlink_file(tmp_path: pathlib.Path) -> None:
352	from muse.core.workspace import WorkspaceManifestDict
353	repo = _make_repo(tmp_path)
354	real = tmp_path / "real.toml"
355	real.write_text("")
356	manifest_path = repo / ".muse" / "workspace.toml"
357	manifest_path.symlink_to(real)
358	with pytest.raises(OSError, match="symlink"):
359	_save_manifest(repo, WorkspaceManifestDict(members=[]))
360
361
362	# ---------------------------------------------------------------------------
363	# Unit: update_workspace_member
364	# ---------------------------------------------------------------------------
365
366
367	def test_update_url(tmp_path: pathlib.Path) -> None:
368	repo = _make_repo(tmp_path)
369	add_workspace_member(repo, "core", "https://old.example.com/core")
370	update_workspace_member(repo, "core", url="https://new.example.com/core")
371	m = get_workspace_member(repo, "core")
372	assert m.url == "https://new.example.com/core"
373
374
375	def test_update_branch(tmp_path: pathlib.Path) -> None:
376	repo = _make_repo(tmp_path)
377	add_workspace_member(repo, "core", "https://example.com/core")
378	update_workspace_member(repo, "core", branch="v2")
379	m = get_workspace_member(repo, "core")
380	assert m.branch == "v2"
381
382
383	def test_update_path(tmp_path: pathlib.Path) -> None:
384	repo = _make_repo(tmp_path)
385	add_workspace_member(repo, "core", "https://example.com/core")
386	update_workspace_member(repo, "core", path="vendor/core")
387	m = get_workspace_member(repo, "core")
388	assert "vendor/core" in str(m.path)
389
390
391	def test_update_nonexistent_raises(tmp_path: pathlib.Path) -> None:
392	repo = _make_repo(tmp_path)
393	with pytest.raises(ValueError, match="not found"):
394	update_workspace_member(repo, "ghost", url="https://example.com/ghost")
395
396
397	def test_update_invalid_url_raises(tmp_path: pathlib.Path) -> None:
398	repo = _make_repo(tmp_path)
399	add_workspace_member(repo, "core", "https://example.com/core")
400	with pytest.raises(ValueError, match="not allowed"):
401	update_workspace_member(repo, "core", url="ftp://example.com/core")
402
403
404	# ---------------------------------------------------------------------------
405	# Unit: get_workspace_member
406	# ---------------------------------------------------------------------------
407
408
409	def test_get_workspace_member_found(tmp_path: pathlib.Path) -> None:
410	repo = _make_repo(tmp_path)
411	add_workspace_member(repo, "sounds", "https://example.com/sounds", branch="v2")
412	m = get_workspace_member(repo, "sounds")
413	assert isinstance(m, WorkspaceMemberStatus)
414	assert m.name == "sounds"
415	assert m.branch == "v2"
416
417
418	def test_get_workspace_member_not_found_raises(tmp_path: pathlib.Path) -> None:
419	repo = _make_repo(tmp_path)
420	add_workspace_member(repo, "core", "https://example.com/core")
421	with pytest.raises(ValueError, match="not found"):
422	get_workspace_member(repo, "ghost")
423
424
425	def test_get_workspace_member_no_manifest_raises(tmp_path: pathlib.Path) -> None:
426	repo = _make_repo(tmp_path)
427	with pytest.raises(ValueError, match="No workspace manifest"):
428	get_workspace_member(repo, "anything")
429
430
431	# ---------------------------------------------------------------------------
432	# Unit: WorkspaceMemberStatus has dirty field
433	# ---------------------------------------------------------------------------
434
435
436	def test_member_status_has_dirty_field(tmp_path: pathlib.Path) -> None:
437	repo = _make_repo(tmp_path)
438	add_workspace_member(repo, "core", "https://example.com/core")
439	members = list_workspace_members(repo)
440	assert hasattr(members[0], "dirty")
441	assert members[0].dirty is False # not present → not dirty
442	# New fields are always present on the dataclass.
443	assert hasattr(members[0], "actual_branch")
444	assert members[0].actual_branch is None # not present → no actual branch
445	assert hasattr(members[0], "stash_count")
446	assert members[0].stash_count == 0 # not present → no stashes
447	assert hasattr(members[0], "feature_branches")
448	assert members[0].feature_branches == [] # not present → no branches
449
450
451	# ---------------------------------------------------------------------------
452	# Unit: sync_workspace dry_run
453	# ---------------------------------------------------------------------------
454
455
456	def test_sync_dry_run_returns_skipped(tmp_path: pathlib.Path) -> None:
457	repo = _make_repo(tmp_path)
458	add_workspace_member(repo, "core", "https://example.com/core")
459	results = sync_workspace(repo, dry_run=True)
460	assert len(results) == 1
461	assert results[0]["status"].startswith("skipped")
462	assert "dry-run" in results[0]["status"]
463
464
465	def test_sync_dry_run_no_subprocess(tmp_path: pathlib.Path) -> None:
466	"""dry_run must never invoke subprocess.run."""
467	repo = _make_repo(tmp_path)
468	add_workspace_member(repo, "core", "https://example.com/core")
469	with patch("muse.core.workspace.subprocess.run") as mock_run:
470	sync_workspace(repo, dry_run=True)
471	mock_run.assert_not_called()
472
473
474	def test_sync_empty_manifest_returns_empty(tmp_path: pathlib.Path) -> None:
475	repo = _make_repo(tmp_path)
476	results = sync_workspace(repo)
477	assert results == []
478
479
480	def test_sync_dry_run_pull_action(tmp_path: pathlib.Path) -> None:
481	"""Member with existing .muse dir should report 'pull' in dry-run."""
482	repo = _make_repo(tmp_path)
483	member_path = tmp_path / "repos" / "core"
484	(member_path / ".muse").mkdir(parents=True)
485	add_workspace_member(repo, "core", "https://example.com/core")
486	results = sync_workspace(repo, dry_run=True)
487	assert "pull" in results[0]["status"]
488
489
490	def test_sync_named_member_only(tmp_path: pathlib.Path) -> None:
491	repo = _make_repo(tmp_path)
492	add_workspace_member(repo, "core", "https://example.com/core")
493	add_workspace_member(repo, "data", "https://example.com/data")
494	with patch("muse.core.workspace.subprocess.run") as mock_run:
495	mock_run.return_value = type("R", (), {"returncode": 0, "stderr": ""})()
496	results = sync_workspace(repo, member_name="core", dry_run=True)
497	assert len(results) == 1
498	assert results[0]["name"] == "core"
499
500
501	# ---------------------------------------------------------------------------
502	# Security: TOML injection via crafted member name/url persists safely
503	# ---------------------------------------------------------------------------
504
505
506	def test_toml_injection_in_url_is_escaped(tmp_path: pathlib.Path) -> None:
507	"""A URL with embedded quotes and newlines must not corrupt the TOML manifest."""
508	import tomllib
509	repo = _make_repo(tmp_path)
510	# Craft a URL that would inject extra members if not escaped
511	tricky_url = 'https://example.com/core"\n[[members]]\nname = "injected'
512	add_workspace_member(repo, "safe", tricky_url)
513	raw = (repo / ".muse" / "workspace.toml").read_text()
514	parsed = tomllib.loads(raw)
515	# Exactly 1 member — the injected one must not appear as a separate entry
516	assert len(parsed.get("members", [])) == 1
517	assert parsed["members"][0]["name"] == "safe"
518	# The raw newlines from the URL are escaped as \n inside the string value
519	# (the file naturally has TOML structural newlines, but the URL value's
520	# embedded newlines must appear as the two-char escape sequence \\n)
521	url_line = next(line for line in raw.splitlines() if line.startswith("url"))
522	assert "\\n" in url_line
523
524
525	def test_ansi_in_name_sanitized_in_output(tmp_path: pathlib.Path) -> None:
526	repo = _make_repo(tmp_path)
527	evil_name = "\x1b[31mevil\x1b[0m"
528	# _validate_member_name will reject the ANSI escape — that's the right behaviour
529	with pytest.raises(ValueError, match="invalid characters"):
530	add_workspace_member(repo, evil_name, "https://example.com/evil")
531
532
533	def test_path_traversal_in_member_path_rejected(tmp_path: pathlib.Path) -> None:
534	repo = _make_repo(tmp_path)
535	with pytest.raises(ValueError, match="outside the workspace root"):
536	add_workspace_member(repo, "evil", "https://example.com/evil", path="../../etc")
537
538
539	def test_file_url_scheme_rejected(tmp_path: pathlib.Path) -> None:
540	repo = _make_repo(tmp_path)
541	with pytest.raises(ValueError, match="not allowed"):
542	add_workspace_member(repo, "evil", "file:///etc/passwd")
543
544
545	def test_null_byte_in_url_rejected(tmp_path: pathlib.Path) -> None:
546	repo = _make_repo(tmp_path)
547	with pytest.raises(ValueError, match="null bytes"):
548	add_workspace_member(repo, "evil", "https://example.com/\x00evil")
549
550
551	# ---------------------------------------------------------------------------
552	# Error routing — all error output goes to stderr
553	# ---------------------------------------------------------------------------
554
555
556	def test_add_duplicate_error_to_stderr(tmp_path: pathlib.Path) -> None:
557	repo = _make_repo(tmp_path)
558	stdout, stderr, rc = _cli(["workspace", "add", "core", "https://example.com/core"], repo)
559	assert rc == 0
560	stdout2, stderr2, rc2 = _cli(["workspace", "add", "core", "https://example.com/other"], repo)
561	assert rc2 != 0
562	assert "already exists" in stderr2
563	assert "already exists" not in stdout2
564
565
566	def test_remove_nonexistent_error_to_stderr(tmp_path: pathlib.Path) -> None:
567	repo = _make_repo(tmp_path)
568	add_workspace_member(repo, "core", "https://example.com/core")
569	stdout, stderr, rc = _cli(["workspace", "remove", "ghost"], repo)
570	assert rc != 0
571	assert "not found" in stderr
572	assert "not found" not in stdout
573
574
575	def test_update_no_flags_error_to_stderr(tmp_path: pathlib.Path) -> None:
576	repo = _make_repo(tmp_path)
577	add_workspace_member(repo, "core", "https://example.com/core")
578	stdout, stderr, rc = _cli(["workspace", "update", "core"], repo)
579	assert rc != 0
580	assert "at least one" in stderr
581
582
583	def test_status_nonexistent_error_to_stderr(tmp_path: pathlib.Path) -> None:
584	repo = _make_repo(tmp_path)
585	add_workspace_member(repo, "core", "https://example.com/core")
586	stdout, stderr, rc = _cli(["workspace", "status", "ghost"], repo)
587	assert rc != 0
588	assert "not found" in stderr
589	assert "not found" not in stdout
590
591
592	# ---------------------------------------------------------------------------
593	# JSON schema: add
594	# ---------------------------------------------------------------------------
595
596
597	class _AddJson(TypedDict):
598	name: str
599	url: str
600	path: str
601	branch: str
602
603
604	def test_add_json_schema(tmp_path: pathlib.Path) -> None:
605	repo = _make_repo(tmp_path)
606	stdout, _, rc = _cli(
607	["workspace", "add", "core", "https://example.com/core", "--json"], repo
608	)
609	assert rc == 0
610	d = _parse_add(stdout)
611	assert d["name"] == "core"
612	assert d["branch"] == "main"
613	assert "repos/core" in d["path"]
614
615
616	# ---------------------------------------------------------------------------
617	# JSON schema: update
618	# ---------------------------------------------------------------------------
619
620
621	class _UpdateJson(TypedDict):
622	name: str
623	url: str
624	path: str
625	branch: str
626
627
628	def test_update_json_schema(tmp_path: pathlib.Path) -> None:
629	repo = _make_repo(tmp_path)
630	_cli(["workspace", "add", "core", "https://example.com/core"], repo)
631	stdout, _, rc = _cli(
632	["workspace", "update", "core", "--branch", "dev", "--json"], repo
633	)
634	assert rc == 0
635	d = _parse_update(stdout)
636	assert d["name"] == "core"
637	assert d["branch"] == "dev"
638
639
640	# ---------------------------------------------------------------------------
641	# JSON schema: list
642	# ---------------------------------------------------------------------------
643
644
645	class _ListMemberJson(TypedDict):
646	name: str
647	url: str
648	path: str
649	branch: str
650	present: bool
651	head_commit: str \| None
652	dirty: bool
653	actual_branch: str \| None
654	stash_count: int
655	feature_branches: list[str]
656
657
658	def test_list_json_schema(tmp_path: pathlib.Path) -> None:
659	repo = _make_repo(tmp_path)
660	_cli(["workspace", "add", "core", "https://example.com/core"], repo)
661	_cli(["workspace", "add", "data", "https://example.com/data", "--branch", "v2"], repo)
662	stdout, _, rc = _cli(["workspace", "list", "--json"], repo)
663	assert rc == 0
664	members = _parse_list(stdout)
665	assert len(members) == 2
666	d = members[0]
667	assert d["name"] == "core"
668	assert d["present"] is False
669	assert d["dirty"] is False
670	# New fields — always present even when member is not cloned.
671	assert d["actual_branch"] is None
672	assert d["stash_count"] == 0
673	assert d["feature_branches"] == []
674
675
676	def test_list_json_empty_list(tmp_path: pathlib.Path) -> None:
677	repo = _make_repo(tmp_path)
678	stdout, _, rc = _cli(["workspace", "list", "--json"], repo)
679	assert rc == 0
680	assert _parse_list(stdout) == []
681
682
683	# ---------------------------------------------------------------------------
684	# JSON schema: remove
685	# ---------------------------------------------------------------------------
686
687
688	class _RemoveJson(TypedDict):
689	name: str
690	removed: bool
691
692
693	def test_remove_json_schema(tmp_path: pathlib.Path) -> None:
694	repo = _make_repo(tmp_path)
695	_cli(["workspace", "add", "core", "https://example.com/core"], repo)
696	stdout, _, rc = _cli(["workspace", "remove", "core", "--json"], repo)
697	assert rc == 0
698	d = _parse_remove(stdout)
699	assert d["name"] == "core"
700	assert d["removed"] is True
701
702
703	# ---------------------------------------------------------------------------
704	# JSON schema: status
705	# ---------------------------------------------------------------------------
706
707
708	def test_status_json_all(tmp_path: pathlib.Path) -> None:
709	repo = _make_repo(tmp_path)
710	_cli(["workspace", "add", "core", "https://example.com/core"], repo)
711	stdout, _, rc = _cli(["workspace", "status", "--json"], repo)
712	assert rc == 0
713	members = _parse_list(stdout)
714	assert len(members) == 1
715	assert members[0]["name"] == "core"
716
717
718	def test_status_json_named(tmp_path: pathlib.Path) -> None:
719	repo = _make_repo(tmp_path)
720	_cli(["workspace", "add", "core", "https://example.com/core"], repo)
721	_cli(["workspace", "add", "data", "https://example.com/data"], repo)
722	stdout, _, rc = _cli(["workspace", "status", "core", "--json"], repo)
723	assert rc == 0
724	members = _parse_list(stdout)
725	assert len(members) == 1
726	assert members[0]["name"] == "core"
727
728
729	def test_status_json_empty(tmp_path: pathlib.Path) -> None:
730	repo = _make_repo(tmp_path)
731	stdout, _, rc = _cli(["workspace", "status", "--json"], repo)
732	assert rc == 0
733	assert _parse_list(stdout) == []
734
735
736	# ---------------------------------------------------------------------------
737	# JSON schema: sync
738	# ---------------------------------------------------------------------------
739
740
741	class _SyncResultItemJson(TypedDict):
742	name: str
743	status: str
744	ok: bool
745
746
747	class _SyncJson(TypedDict):
748	dry_run: bool
749	workers: int
750	results: list[_SyncResultItemJson]
751	total: int
752	ok_count: int
753	error_count: int
754
755
756	def test_sync_json_dry_run(tmp_path: pathlib.Path) -> None:
757	repo = _make_repo(tmp_path)
758	_cli(["workspace", "add", "core", "https://example.com/core"], repo)
759	stdout, _, rc = _cli(["workspace", "sync", "--dry-run", "--json"], repo)
760	assert rc == 0
761	d = _parse_sync(stdout)
762	assert d["dry_run"] is True
763	assert d["total"] == 1
764	assert d["ok_count"] == 1
765	assert d["error_count"] == 0
766
767
768	def test_sync_json_empty_manifest(tmp_path: pathlib.Path) -> None:
769	repo = _make_repo(tmp_path)
770	stdout, _, rc = _cli(["workspace", "sync", "--dry-run", "--json"], repo)
771	assert rc == 0
772	d = _parse_sync(stdout)
773	assert d["total"] == 0
774
775
776	# ---------------------------------------------------------------------------
777	# Integration: full lifecycle
778	# ---------------------------------------------------------------------------
779
780
781	def test_lifecycle_add_update_remove(tmp_path: pathlib.Path) -> None:
782	repo = _make_repo(tmp_path)
783	add_workspace_member(repo, "core", "https://example.com/core")
784	update_workspace_member(repo, "core", branch="dev")
785	m = get_workspace_member(repo, "core")
786	assert m.branch == "dev"
787	remove_workspace_member(repo, "core")
788	assert list_workspace_members(repo) == []
789
790
791	def test_lifecycle_multiple_members(tmp_path: pathlib.Path) -> None:
792	repo = _make_repo(tmp_path)
793	for i in range(5):
794	add_workspace_member(repo, f"svc{i}", f"https://example.com/svc{i}")
795	members = list_workspace_members(repo)
796	assert len(members) == 5
797	update_workspace_member(repo, "svc2", branch="release")
798	m = get_workspace_member(repo, "svc2")
799	assert m.branch == "release"
800	remove_workspace_member(repo, "svc2")
801	assert len(list_workspace_members(repo)) == 4
802
803
804	def test_add_custom_branch_and_path(tmp_path: pathlib.Path) -> None:
805	repo = _make_repo(tmp_path)
806	add_workspace_member(repo, "data", "https://example.com/data", path="vendor/data", branch="v2")
807	m = get_workspace_member(repo, "data")
808	assert m.branch == "v2"
809	assert "vendor/data" in str(m.path)
810
811
812	# ---------------------------------------------------------------------------
813	# E2E: text output
814	# ---------------------------------------------------------------------------
815
816
817	def test_e2e_add_text_output(tmp_path: pathlib.Path) -> None:
818	repo = _make_repo(tmp_path)
819	stdout, _, rc = _cli(["workspace", "add", "core", "https://example.com/core"], repo)
820	assert rc == 0
821	assert "Added" in stdout
822	assert "core" in stdout
823
824
825	def test_e2e_list_text_output(tmp_path: pathlib.Path) -> None:
826	repo = _make_repo(tmp_path)
827	_cli(["workspace", "add", "core", "https://example.com/core"], repo)
828	stdout, _, rc = _cli(["workspace", "list"], repo)
829	assert rc == 0
830	assert "core" in stdout
831	assert "present" in stdout.lower() or "no" in stdout
832
833
834	def test_e2e_status_text_output(tmp_path: pathlib.Path) -> None:
835	repo = _make_repo(tmp_path)
836	_cli(["workspace", "add", "core", "https://example.com/core"], repo)
837	stdout, _, rc = _cli(["workspace", "status"], repo)
838	assert rc == 0
839	assert "core" in stdout
840	assert "branch=main" in stdout
841
842
843	def test_e2e_remove_text_output(tmp_path: pathlib.Path) -> None:
844	repo = _make_repo(tmp_path)
845	_cli(["workspace", "add", "core", "https://example.com/core"], repo)
846	stdout, _, rc = _cli(["workspace", "remove", "core"], repo)
847	assert rc == 0
848	assert "Removed" in stdout
849
850
851	def test_e2e_sync_dry_run_text_output(tmp_path: pathlib.Path) -> None:
852	repo = _make_repo(tmp_path)
853	_cli(["workspace", "add", "core", "https://example.com/core"], repo)
854	stdout, _, rc = _cli(["workspace", "sync", "--dry-run"], repo)
855	assert rc == 0
856	assert "core" in stdout
857	assert "skipped" in stdout or "dry-run" in stdout
858
859
860	def test_e2e_list_no_members(tmp_path: pathlib.Path) -> None:
861	repo = _make_repo(tmp_path)
862	stdout, _, rc = _cli(["workspace", "list"], repo)
863	assert rc == 0
864	assert "No workspace members" in stdout
865
866
867	def test_e2e_update_text_output(tmp_path: pathlib.Path) -> None:
868	repo = _make_repo(tmp_path)
869	_cli(["workspace", "add", "core", "https://example.com/core"], repo)
870	stdout, _, rc = _cli(["workspace", "update", "core", "--branch", "dev"], repo)
871	assert rc == 0
872	assert "Updated" in stdout
873
874
875	def test_e2e_shorthand_branch_flag(tmp_path: pathlib.Path) -> None:
876	"""-b shorthand should work for add and update."""
877	repo = _make_repo(tmp_path)
878	stdout, _, rc = _cli(["workspace", "add", "data", "https://example.com/data", "-b", "v3"], repo)
879	assert rc == 0
880	m = get_workspace_member(repo, "data")
881	assert m.branch == "v3"
882
883
884	# ---------------------------------------------------------------------------
885	# Stress: 50 members
886	# ---------------------------------------------------------------------------
887
888
889	def test_stress_50_members_add_list(tmp_path: pathlib.Path) -> None:
890	repo = _make_repo(tmp_path)
891	for i in range(50):
892	add_workspace_member(repo, f"svc{i:03d}", f"https://example.com/svc{i}")
893	members = list_workspace_members(repo)
894	assert len(members) == 50
895	names = {m.name for m in members}
896	for i in range(50):
897	assert f"svc{i:03d}" in names
898
899
900	def test_stress_add_remove_cycle(tmp_path: pathlib.Path) -> None:
901	repo = _make_repo(tmp_path)
902	for i in range(20):
903	add_workspace_member(repo, f"repo{i}", f"https://example.com/repo{i}")
904	for i in range(20):
905	remove_workspace_member(repo, f"repo{i}")
906	assert list_workspace_members(repo) == []
907
908
909	def test_stress_concurrent_list_reads(tmp_path: pathlib.Path) -> None:
910	"""Concurrent reads of the manifest must all succeed without corruption."""
911	repo = _make_repo(tmp_path)
912	for i in range(20):
913	add_workspace_member(repo, f"svc{i}", f"https://example.com/svc{i}")
914
915	failures: list[str] = []
916
917	def _read() -> None:
918	try:
919	members = list_workspace_members(repo)
920	if len(members) != 20:
921	failures.append(f"Expected 20 members, got {len(members)}")
922	except Exception as exc:
923	failures.append(str(exc))
924
925	threads = [threading.Thread(target=_read) for _ in range(20)]
926	for t in threads:
927	t.start()
928	for t in threads:
929	t.join()
930
931	assert not failures, f"Concurrent read failures: {failures}"
932
933
934	def test_stress_sync_parallel_dry_run(tmp_path: pathlib.Path) -> None:
935	"""Parallel sync (dry_run) over 20 members must return 20 results."""
936	repo = _make_repo(tmp_path)
937	for i in range(20):
938	add_workspace_member(repo, f"svc{i}", f"https://example.com/svc{i}")
939	results = sync_workspace(repo, dry_run=True, workers=4)
940	assert len(results) == 20
941	for r in results:
942	assert r["status"].startswith("skipped")
943
944
945	def test_stress_json_list_50_members(tmp_path: pathlib.Path) -> None:
946	"""JSON list output for 50 members must parse correctly."""
947	repo = _make_repo(tmp_path)
948	for i in range(50):
949	add_workspace_member(repo, f"svc{i:03d}", f"https://example.com/svc{i}")
950	stdout, _, rc = _cli(["workspace", "list", "--json"], repo)
951	assert rc == 0
952	members = _parse_list(stdout)
953	assert len(members) == 50
954
955
956	def test_stress_update_10_members(tmp_path: pathlib.Path) -> None:
957	"""Update branch for 10 members sequentially; all must reflect the change."""
958	repo = _make_repo(tmp_path)
959	for i in range(10):
960	add_workspace_member(repo, f"svc{i}", f"https://example.com/svc{i}")
961	for i in range(10):
962	update_workspace_member(repo, f"svc{i}", branch="release")
963	members = list_workspace_members(repo)
964	for m in members:
965	assert m.branch == "release"
966
967
968	# ===========================================================================
969	# muse workspace add — Extended / Security / Stress
970	# ===========================================================================
971
972
973	class TestWorkspaceAddExtended:
974	"""-j alias, JSON schema, defaults, custom args, lifecycle, edge cases."""
975
976	def test_add_j_alias(self, tmp_path: pathlib.Path) -> None:
977	"""-j produces the same JSON as --json."""
978	repo = _make_repo(tmp_path)
979	stdout1, _, rc1 = _cli(["workspace", "add", "core", "https://example.com/core", "--json"], repo)
980	_cli(["workspace", "remove", "core"], repo)
981	stdout2, _, rc2 = _cli(["workspace", "add", "core", "https://example.com/core", "-j"], repo)
982	assert rc1 == 0 and rc2 == 0
983	assert json.loads(_json_blob(stdout1)) == json.loads(_json_blob(stdout2))
984
985	def test_add_json_name_field(self, tmp_path: pathlib.Path) -> None:
986	"""JSON name field matches the supplied NAME argument."""
987	repo = _make_repo(tmp_path)
988	stdout, _, rc = _cli(["workspace", "add", "myrepo", "https://example.com/myrepo", "-j"], repo)
989	assert rc == 0
990	d = _parse_add(stdout)
991	assert d["name"] == "myrepo"
992
993	def test_add_json_url_field(self, tmp_path: pathlib.Path) -> None:
994	"""JSON url field matches the supplied URL argument."""
995	repo = _make_repo(tmp_path)
996	url = "https://example.com/myrepo"
997	stdout, _, rc = _cli(["workspace", "add", "myrepo", url, "-j"], repo)
998	assert rc == 0
999	d = _parse_add(stdout)
1000	assert d["url"] == url
1001
1002	def test_add_json_default_branch_is_main(self, tmp_path: pathlib.Path) -> None:
1003	"""Branch defaults to 'main' when --branch is not supplied."""
1004	repo = _make_repo(tmp_path)
1005	stdout, _, rc = _cli(["workspace", "add", "core", "https://example.com/core", "-j"], repo)
1006	assert rc == 0
1007	d = _parse_add(stdout)
1008	assert d["branch"] == "main"
1009
1010	def test_add_json_default_path_is_repos_name(self, tmp_path: pathlib.Path) -> None:
1011	"""Path defaults to repos/<name> when --path is not supplied."""
1012	repo = _make_repo(tmp_path)
1013	stdout, _, rc = _cli(["workspace", "add", "core", "https://example.com/core", "-j"], repo)
1014	assert rc == 0
1015	d = _parse_add(stdout)
1016	assert "repos/core" in d["path"]
1017
1018	def test_add_json_custom_branch(self, tmp_path: pathlib.Path) -> None:
1019	"""Custom --branch appears in JSON output."""
1020	repo = _make_repo(tmp_path)
1021	stdout, _, rc = _cli(["workspace", "add", "data", "https://example.com/data", "--branch", "v2", "-j"], repo)
1022	assert rc == 0
1023	d = _parse_add(stdout)
1024	assert d["branch"] == "v2"
1025
1026	def test_add_json_custom_path(self, tmp_path: pathlib.Path) -> None:
1027	"""Custom --path appears in JSON output."""
1028	repo = _make_repo(tmp_path)
1029	stdout, _, rc = _cli(["workspace", "add", "data", "https://example.com/data", "--path", "vendor/data", "-j"], repo)
1030	assert rc == 0
1031	d = _parse_add(stdout)
1032	assert "vendor/data" in d["path"]
1033
1034	def test_add_json_all_fields_present(self, tmp_path: pathlib.Path) -> None:
1035	"""JSON output contains exactly name, url, path, branch."""
1036	repo = _make_repo(tmp_path)
1037	stdout, _, rc = _cli(["workspace", "add", "core", "https://example.com/core", "-j"], repo)
1038	assert rc == 0
1039	raw = json.loads(_json_blob(stdout))
1040	assert set(raw.keys()) == {"name", "url", "path", "branch"}
1041
1042	def test_add_default_is_text(self, tmp_path: pathlib.Path) -> None:
1043	"""Without --json the output is human-readable text."""
1044	repo = _make_repo(tmp_path)
1045	stdout, _, rc = _cli(["workspace", "add", "core", "https://example.com/core"], repo)
1046	assert rc == 0
1047	assert not stdout.strip().startswith("{")
1048
1049	def test_add_text_contains_name(self, tmp_path: pathlib.Path) -> None:
1050	"""Text output mentions the member name."""
1051	repo = _make_repo(tmp_path)
1052	stdout, _, rc = _cli(["workspace", "add", "myrepo", "https://example.com/myrepo"], repo)
1053	assert rc == 0
1054	assert "myrepo" in stdout
1055
1056	def test_add_text_hints_sync(self, tmp_path: pathlib.Path) -> None:
1057	"""Text output hints to run 'muse workspace sync'."""
1058	repo = _make_repo(tmp_path)
1059	stdout, _, rc = _cli(["workspace", "add", "core", "https://example.com/core"], repo)
1060	assert rc == 0
1061	assert "sync" in stdout.lower()
1062
1063	def test_add_duplicate_exits_1(self, tmp_path: pathlib.Path) -> None:
1064	"""Adding a member with a duplicate name exits with code 1."""
1065	repo = _make_repo(tmp_path)
1066	_cli(["workspace", "add", "core", "https://example.com/core"], repo)
1067	_, stderr, rc = _cli(["workspace", "add", "core", "https://example.com/core2"], repo)
1068	assert rc == 1
1069	assert "core" in stderr
1070
1071	def test_add_outside_repo_succeeds(self, tmp_path: pathlib.Path) -> None:
1072	"""Workspace add works from any directory — no muse repo required."""
1073	empty = tmp_path / "empty"
1074	empty.mkdir()
1075	_, _, rc = _cli(["workspace", "add", "core", "https://example.com/core"], empty)
1076	assert rc == 0
1077
1078	def test_add_appears_in_list_after(self, tmp_path: pathlib.Path) -> None:
1079	"""Added member appears in subsequent 'workspace list --json'."""
1080	repo = _make_repo(tmp_path)
1081	_cli(["workspace", "add", "core", "https://example.com/core"], repo)
1082	stdout, _, rc = _cli(["workspace", "list", "--json"], repo)
1083	assert rc == 0
1084	members = _parse_list(stdout)
1085	assert any(m["name"] == "core" for m in members)
1086
1087	def test_add_help_has_description(self, tmp_path: pathlib.Path) -> None:
1088	"""--help includes the rich description."""
1089	repo = _make_repo(tmp_path)
1090	stdout, _, rc = _cli(["workspace", "add", "--help"], repo)
1091	assert rc == 0
1092	assert "Agent quickstart" in stdout or "JSON output schema" in stdout
1093
1094	def test_add_local_path_url_accepted(self, tmp_path: pathlib.Path) -> None:
1095	"""A bare filesystem path is accepted as the URL."""
1096	repo = _make_repo(tmp_path)
1097	local = str(tmp_path / "local-repo")
1098	stdout, _, rc = _cli(["workspace", "add", "local", local, "-j"], repo)
1099	assert rc == 0
1100	d = _parse_add(stdout)
1101	assert d["url"] == local
1102
1103	def test_add_shorthand_branch_flag(self, tmp_path: pathlib.Path) -> None:
1104	"""-b shorthand sets the branch correctly."""
1105	repo = _make_repo(tmp_path)
1106	stdout, _, rc = _cli(["workspace", "add", "data", "https://example.com/data", "-b", "release", "-j"], repo)
1107	assert rc == 0
1108	assert _parse_add(stdout)["branch"] == "release"
1109
1110	def test_add_json_is_valid_json(self, tmp_path: pathlib.Path) -> None:
1111	"""JSON output is well-formed."""
1112	repo = _make_repo(tmp_path)
1113	stdout, _, rc = _cli(["workspace", "add", "core", "https://example.com/core", "-j"], repo)
1114	assert rc == 0
1115	raw = json.loads(_json_blob(stdout))
1116	assert isinstance(raw, dict)
1117	assert isinstance(raw["name"], str)
1118	assert isinstance(raw["branch"], str)
1119
1120
1121	class TestWorkspaceAddSecurity:
1122	"""Input validation, ANSI sanitization, error routing."""
1123
1124	def test_add_invalid_url_scheme_rejected(self, tmp_path: pathlib.Path) -> None:
1125	"""file:// URL scheme is rejected with exit code 1."""
1126	repo = _make_repo(tmp_path)
1127	_, stderr, rc = _cli(["workspace", "add", "bad", "file:///etc/passwd", "-j"], repo)
1128	assert rc == 1
1129	assert "scheme" in stderr.lower() or "not allowed" in stderr.lower()
1130
1131	def test_add_ftp_scheme_rejected(self, tmp_path: pathlib.Path) -> None:
1132	"""ftp:// URL scheme is rejected."""
1133	repo = _make_repo(tmp_path)
1134	_, _, rc = _cli(["workspace", "add", "bad", "ftp://example.com/repo", "-j"], repo)
1135	assert rc == 1
1136
1137	def test_add_null_byte_in_url_rejected(self, tmp_path: pathlib.Path) -> None:
1138	"""Null byte in URL is rejected by the core validator."""
1139	repo = _make_repo(tmp_path)
1140	with pytest.raises(ValueError, match="null"):
1141	add_workspace_member(repo, "bad", "https://example.com/\x00repo")
1142
1143	def test_add_path_traversal_rejected(self, tmp_path: pathlib.Path) -> None:
1144	"""--path escaping workspace root is rejected."""
1145	repo = _make_repo(tmp_path)
1146	_, stderr, rc = _cli(["workspace", "add", "bad", "https://example.com/repo", "--path", "../../etc", "-j"], repo)
1147	assert rc == 1
1148	assert "outside" in stderr.lower() or "escape" in stderr.lower() or "resolves" in stderr.lower()
1149
1150	def test_add_invalid_name_rejected(self, tmp_path: pathlib.Path) -> None:
1151	"""Name with slashes is rejected."""
1152	repo = _make_repo(tmp_path)
1153	_, _, rc = _cli(["workspace", "add", "bad/name", "https://example.com/repo", "-j"], repo)
1154	assert rc == 1
1155
1156	def test_add_empty_name_rejected(self, tmp_path: pathlib.Path) -> None:
1157	"""Empty name is rejected."""
1158	repo = _make_repo(tmp_path)
1159	_, _, rc = _cli(["workspace", "add", "", "https://example.com/repo", "-j"], repo)
1160	assert rc != 0
1161
1162	def test_add_error_goes_to_stderr(self, tmp_path: pathlib.Path) -> None:
1163	"""Error output (duplicate) goes to stderr, not stdout."""
1164	repo = _make_repo(tmp_path)
1165	_cli(["workspace", "add", "core", "https://example.com/core"], repo)
1166	stdout, stderr, rc = _cli(["workspace", "add", "core", "https://example.com/core"], repo)
1167	assert rc == 1
1168	assert not stdout.strip().startswith("{")
1169	assert stderr.strip() != ""
1170
1171	def test_add_json_no_ansi_in_output(self, tmp_path: pathlib.Path) -> None:
1172	"""JSON output contains no ANSI escape sequences."""
1173	repo = _make_repo(tmp_path)
1174	stdout, _, rc = _cli(["workspace", "add", "core", "https://example.com/core", "-j"], repo)
1175	assert rc == 0
1176	assert "\x1b" not in stdout
1177
1178	def test_add_ssh_scheme_rejected(self, tmp_path: pathlib.Path) -> None:
1179	"""ssh:// URL scheme is rejected."""
1180	repo = _make_repo(tmp_path)
1181	_, _, rc = _cli(["workspace", "add", "bad", "ssh://example.com/repo"], repo)
1182	assert rc == 1
1183
1184
1185	class TestWorkspaceAddStress:
1186	"""Performance and scale tests for workspace add."""
1187
1188	def test_add_20_sequential(self, tmp_path: pathlib.Path) -> None:
1189	"""20 members can be added sequentially without error."""
1190	repo = _make_repo(tmp_path)
1191	for i in range(20):
1192	_, _, rc = _cli(["workspace", "add", f"svc{i:02d}", f"https://example.com/svc{i}", "-j"], repo)
1193	assert rc == 0
1194	stdout, _, rc = _cli(["workspace", "list", "--json"], repo)
1195	assert rc == 0
1196	members = _parse_list(stdout)
1197	assert len(members) == 20
1198
1199	def test_add_performance(self, tmp_path: pathlib.Path) -> None:
1200	"""Adding 10 members sequentially completes within 5 seconds."""
1201	repo = _make_repo(tmp_path)
1202	t0 = time.monotonic()
1203	for i in range(10):
1204	_cli(["workspace", "add", f"svc{i}", f"https://example.com/svc{i}"], repo)
1205	elapsed = time.monotonic() - t0
1206	assert elapsed < 5.0, f"10 adds took {elapsed:.2f}s"
1207
1208	def test_add_remove_add_cycle(self, tmp_path: pathlib.Path) -> None:
1209	"""A member can be re-added with the same name after removal."""
1210	repo = _make_repo(tmp_path)
1211	for _ in range(5):
1212	_, _, rc_add = _cli(["workspace", "add", "core", "https://example.com/core", "-j"], repo)
1213	assert rc_add == 0
1214	_, _, rc_rm = _cli(["workspace", "remove", "core"], repo)
1215	assert rc_rm == 0
1216
1217
1218	# ===========================================================================
1219	# muse workspace update — Extended / Security / Stress
1220	# ===========================================================================
1221
1222
1223	class TestWorkspaceUpdateExtended:
1224	"""-j alias, JSON schema, per-field updates, no-flags guard, edge cases."""
1225
1226	def test_update_j_alias(self, tmp_path: pathlib.Path) -> None:
1227	"""-j produces the same JSON as --json."""
1228	repo = _make_repo(tmp_path)
1229	_cli(["workspace", "add", "core", "https://example.com/core"], repo)
1230	s1, _, rc1 = _cli(["workspace", "update", "core", "--branch", "dev", "--json"], repo)
1231	_cli(["workspace", "update", "core", "--branch", "main"], repo)
1232	s2, _, rc2 = _cli(["workspace", "update", "core", "--branch", "dev", "-j"], repo)
1233	assert rc1 == 0 and rc2 == 0
1234	assert json.loads(_json_blob(s1)) == json.loads(_json_blob(s2))
1235
1236	def test_update_branch_reflected_in_json(self, tmp_path: pathlib.Path) -> None:
1237	"""Updated branch appears in JSON output."""
1238	repo = _make_repo(tmp_path)
1239	_cli(["workspace", "add", "core", "https://example.com/core"], repo)
1240	stdout, _, rc = _cli(["workspace", "update", "core", "--branch", "release", "-j"], repo)
1241	assert rc == 0
1242	d = _parse_update(stdout)
1243	assert d["branch"] == "release"
1244
1245	def test_update_url_reflected_in_json(self, tmp_path: pathlib.Path) -> None:
1246	"""Updated URL appears in JSON output."""
1247	repo = _make_repo(tmp_path)
1248	_cli(["workspace", "add", "core", "https://example.com/core"], repo)
1249	new_url = "https://example.com/core-v2"
1250	stdout, _, rc = _cli(["workspace", "update", "core", "--url", new_url, "-j"], repo)
1251	assert rc == 0
1252	d = _parse_update(stdout)
1253	assert d["url"] == new_url
1254
1255	def test_update_path_reflected_in_json(self, tmp_path: pathlib.Path) -> None:
1256	"""Updated path appears in JSON output."""
1257	repo = _make_repo(tmp_path)
1258	_cli(["workspace", "add", "core", "https://example.com/core"], repo)
1259	stdout, _, rc = _cli(["workspace", "update", "core", "--path", "vendor/core", "-j"], repo)
1260	assert rc == 0
1261	d = _parse_update(stdout)
1262	assert "vendor/core" in d["path"]
1263
1264	def test_update_json_all_fields_present(self, tmp_path: pathlib.Path) -> None:
1265	"""JSON output contains exactly name, url, path, branch."""
1266	repo = _make_repo(tmp_path)
1267	_cli(["workspace", "add", "core", "https://example.com/core"], repo)
1268	stdout, _, rc = _cli(["workspace", "update", "core", "--branch", "dev", "-j"], repo)
1269	assert rc == 0
1270	raw = json.loads(_json_blob(stdout))
1271	assert set(raw.keys()) == {"name", "url", "path", "branch"}
1272
1273	def test_update_json_name_unchanged(self, tmp_path: pathlib.Path) -> None:
1274	"""JSON name field matches the original member name."""
1275	repo = _make_repo(tmp_path)
1276	_cli(["workspace", "add", "myrepo", "https://example.com/myrepo"], repo)
1277	stdout, _, rc = _cli(["workspace", "update", "myrepo", "--branch", "dev", "-j"], repo)
1278	assert rc == 0
1279	d = _parse_update(stdout)
1280	assert d["name"] == "myrepo"
1281
1282	def test_update_omitted_fields_preserved(self, tmp_path: pathlib.Path) -> None:
1283	"""Fields not supplied in --update are preserved from original."""
1284	repo = _make_repo(tmp_path)
1285	_cli(["workspace", "add", "core", "https://example.com/core", "--branch", "v1"], repo)
1286	stdout, _, rc = _cli(["workspace", "update", "core", "--path", "vendor/core", "-j"], repo)
1287	assert rc == 0
1288	d = _parse_update(stdout)
1289	assert d["branch"] == "v1" # unchanged
1290	assert d["url"] == "https://example.com/core" # unchanged
1291
1292	def test_update_no_flags_exits_1(self, tmp_path: pathlib.Path) -> None:
1293	"""Supplying no --url/--path/--branch flags exits with code 1."""
1294	repo = _make_repo(tmp_path)
1295	_cli(["workspace", "add", "core", "https://example.com/core"], repo)
1296	_, stderr, rc = _cli(["workspace", "update", "core"], repo)
1297	assert rc == 1
1298	assert "url" in stderr.lower() or "path" in stderr.lower() or "branch" in stderr.lower()
1299
1300	def test_update_nonexistent_exits_1(self, tmp_path: pathlib.Path) -> None:
1301	"""Updating a nonexistent member exits with code 1."""
1302	repo = _make_repo(tmp_path)
1303	_, stderr, rc = _cli(["workspace", "update", "ghost", "--branch", "dev"], repo)
1304	assert rc == 1
1305	assert "ghost" in stderr
1306
1307	def test_update_outside_repo_exits_1_member_not_found(self, tmp_path: pathlib.Path) -> None:
1308	"""Workspace update from a non-repo dir exits 1 (member not found), not 2."""
1309	empty = tmp_path / "empty"
1310	empty.mkdir()
1311	_, _, rc = _cli(["workspace", "update", "core", "--branch", "dev"], empty)
1312	assert rc == 1
1313
1314	def test_update_default_is_text(self, tmp_path: pathlib.Path) -> None:
1315	"""Without --json the output is human-readable text."""
1316	repo = _make_repo(tmp_path)
1317	_cli(["workspace", "add", "core", "https://example.com/core"], repo)
1318	stdout, _, rc = _cli(["workspace", "update", "core", "--branch", "dev"], repo)
1319	assert rc == 0
1320	assert not stdout.strip().startswith("{")
1321	assert "Updated" in stdout
1322
1323	def test_update_text_contains_name(self, tmp_path: pathlib.Path) -> None:
1324	"""Text output mentions the member name."""
1325	repo = _make_repo(tmp_path)
1326	_cli(["workspace", "add", "myrepo", "https://example.com/myrepo"], repo)
1327	stdout, _, rc = _cli(["workspace", "update", "myrepo", "--branch", "dev"], repo)
1328	assert rc == 0
1329	assert "myrepo" in stdout
1330
1331	def test_update_help_has_description(self, tmp_path: pathlib.Path) -> None:
1332	"""--help includes the rich description."""
1333	repo = _make_repo(tmp_path)
1334	stdout, _, rc = _cli(["workspace", "update", "--help"], repo)
1335	assert rc == 0
1336	assert "Agent quickstart" in stdout or "JSON output schema" in stdout
1337
1338	def test_update_multiple_flags_at_once(self, tmp_path: pathlib.Path) -> None:
1339	"""All three fields can be updated in one command."""
1340	repo = _make_repo(tmp_path)
1341	_cli(["workspace", "add", "core", "https://example.com/core"], repo)
1342	stdout, _, rc = _cli([
1343	"workspace", "update", "core",
1344	"--url", "https://example.com/core-v2",
1345	"--path", "vendor/core",
1346	"--branch", "release",
1347	"-j",
1348	], repo)
1349	assert rc == 0
1350	d = _parse_update(stdout)
1351	assert d["url"] == "https://example.com/core-v2"
1352	assert "vendor/core" in d["path"]
1353	assert d["branch"] == "release"
1354
1355	def test_update_reflected_in_list(self, tmp_path: pathlib.Path) -> None:
1356	"""Updated branch appears in subsequent 'workspace list --json'."""
1357	repo = _make_repo(tmp_path)
1358	_cli(["workspace", "add", "core", "https://example.com/core"], repo)
1359	_cli(["workspace", "update", "core", "--branch", "release"], repo)
1360	stdout, _, rc = _cli(["workspace", "list", "--json"], repo)
1361	assert rc == 0
1362	members = _parse_list(stdout)
1363	core = next(m for m in members if m["name"] == "core")
1364	assert core["branch"] == "release"
1365
1366	def test_update_shorthand_branch_flag(self, tmp_path: pathlib.Path) -> None:
1367	"""-b shorthand sets the branch correctly."""
1368	repo = _make_repo(tmp_path)
1369	_cli(["workspace", "add", "core", "https://example.com/core"], repo)
1370	stdout, _, rc = _cli(["workspace", "update", "core", "-b", "hotfix", "-j"], repo)
1371	assert rc == 0
1372	assert _parse_update(stdout)["branch"] == "hotfix"
1373
1374	def test_update_json_is_valid_json(self, tmp_path: pathlib.Path) -> None:
1375	"""JSON output is well-formed."""
1376	repo = _make_repo(tmp_path)
1377	_cli(["workspace", "add", "core", "https://example.com/core"], repo)
1378	stdout, _, rc = _cli(["workspace", "update", "core", "--branch", "dev", "-j"], repo)
1379	assert rc == 0
1380	raw = json.loads(_json_blob(stdout))
1381	assert isinstance(raw, dict)
1382	for field in ("name", "url", "path", "branch"):
1383	assert isinstance(raw[field], str)
1384
1385
1386	class TestWorkspaceUpdateSecurity:
1387	"""Input validation, ANSI sanitization, error routing."""
1388
1389	def test_update_invalid_url_scheme_rejected(self, tmp_path: pathlib.Path) -> None:
1390	"""file:// URL scheme in --url is rejected."""
1391	repo = _make_repo(tmp_path)
1392	_cli(["workspace", "add", "core", "https://example.com/core"], repo)
1393	_, stderr, rc = _cli(["workspace", "update", "core", "--url", "file:///etc/passwd"], repo)
1394	assert rc == 1
1395	assert "scheme" in stderr.lower() or "not allowed" in stderr.lower()
1396
1397	def test_update_path_traversal_rejected(self, tmp_path: pathlib.Path) -> None:
1398	"""--path escaping workspace root is rejected."""
1399	repo = _make_repo(tmp_path)
1400	_cli(["workspace", "add", "core", "https://example.com/core"], repo)
1401	_, stderr, rc = _cli(["workspace", "update", "core", "--path", "../../etc"], repo)
1402	assert rc == 1
1403	assert "outside" in stderr.lower() or "escape" in stderr.lower() or "resolves" in stderr.lower()
1404
1405	def test_update_null_byte_in_path_rejected(self, tmp_path: pathlib.Path) -> None:
1406	"""Null byte in --path is rejected by the core validator."""
1407	repo = _make_repo(tmp_path)
1408	_cli(["workspace", "add", "core", "https://example.com/core"], repo)
1409	with pytest.raises(ValueError, match="null"):
1410	update_workspace_member(repo, "core", path="vendor/\x00core")
1411
1412	def test_update_error_goes_to_stderr(self, tmp_path: pathlib.Path) -> None:
1413	"""Error output (member not found) goes to stderr, not stdout."""
1414	repo = _make_repo(tmp_path)
1415	stdout, stderr, rc = _cli(["workspace", "update", "ghost", "--branch", "dev"], repo)
1416	assert rc == 1
1417	assert not stdout.strip().startswith("{")
1418	assert stderr.strip() != ""
1419
1420	def test_update_json_no_ansi_in_output(self, tmp_path: pathlib.Path) -> None:
1421	"""JSON output contains no ANSI escape sequences."""
1422	repo = _make_repo(tmp_path)
1423	_cli(["workspace", "add", "core", "https://example.com/core"], repo)
1424	stdout, _, rc = _cli(["workspace", "update", "core", "--branch", "dev", "-j"], repo)
1425	assert rc == 0
1426	assert "\x1b" not in stdout
1427
1428	def test_update_ftp_url_rejected(self, tmp_path: pathlib.Path) -> None:
1429	"""ftp:// URL scheme is rejected."""
1430	repo = _make_repo(tmp_path)
1431	_cli(["workspace", "add", "core", "https://example.com/core"], repo)
1432	_, _, rc = _cli(["workspace", "update", "core", "--url", "ftp://example.com/repo"], repo)
1433	assert rc == 1
1434
1435	def test_update_no_flags_error_to_stderr(self, tmp_path: pathlib.Path) -> None:
1436	"""No-flags error is on stderr; stdout has no JSON."""
1437	repo = _make_repo(tmp_path)
1438	_cli(["workspace", "add", "core", "https://example.com/core"], repo)
1439	stdout, stderr, rc = _cli(["workspace", "update", "core"], repo)
1440	assert rc == 1
1441	assert not stdout.strip().startswith("{")
1442	assert stderr.strip() != ""
1443
1444
1445	class TestWorkspaceUpdateStress:
1446	"""Performance and scale tests for workspace update."""
1447
1448	def test_update_10_members_sequential(self, tmp_path: pathlib.Path) -> None:
1449	"""10 members can each be updated sequentially."""
1450	repo = _make_repo(tmp_path)
1451	for i in range(10):
1452	_cli(["workspace", "add", f"svc{i}", f"https://example.com/svc{i}"], repo)
1453	failures = []
1454	for i in range(10):
1455	_, _, rc = _cli(["workspace", "update", f"svc{i}", "--branch", f"v{i}", "-j"], repo)
1456	if rc != 0:
1457	failures.append(f"svc{i}")
1458	assert not failures
1459	stdout, _, _ = _cli(["workspace", "list", "--json"], repo)
1460	members = {m["name"]: m for m in _parse_list(stdout)}
1461	for i in range(10):
1462	assert members[f"svc{i}"]["branch"] == f"v{i}"
1463
1464	def test_update_performance(self, tmp_path: pathlib.Path) -> None:
1465	"""10 sequential updates complete within 5 seconds."""
1466	repo = _make_repo(tmp_path)
1467	for i in range(10):
1468	_cli(["workspace", "add", f"svc{i}", f"https://example.com/svc{i}"], repo)
1469	t0 = time.monotonic()
1470	for i in range(10):
1471	_cli(["workspace", "update", f"svc{i}", "--branch", "release"], repo)
1472	elapsed = time.monotonic() - t0
1473	assert elapsed < 5.0, f"10 updates took {elapsed:.2f}s"
1474
1475	def test_update_repeated_same_member(self, tmp_path: pathlib.Path) -> None:
1476	"""A member can be updated 10 times in a row without error."""
1477	repo = _make_repo(tmp_path)
1478	_cli(["workspace", "add", "core", "https://example.com/core"], repo)
1479	for i in range(10):
1480	_, _, rc = _cli(["workspace", "update", "core", "--branch", f"v{i}", "-j"], repo)
1481	assert rc == 0
1482	stdout, _, _ = _cli(["workspace", "list", "--json"], repo)
1483	members = _parse_list(stdout)
1484	core = next(m for m in members if m["name"] == "core")
1485	assert core["branch"] == "v9"
1486
1487
1488	# ===========================================================================
1489	# muse workspace list — Extended / Security / Stress
1490	# ===========================================================================
1491
1492
1493	class TestWorkspaceListExtended:
1494	"""-j alias, JSON schema, text output, ordering, edge cases."""
1495
1496	def test_list_j_alias(self, tmp_path: pathlib.Path) -> None:
1497	"""-j produces the same JSON as --json."""
1498	repo = _make_repo(tmp_path)
1499	add_workspace_member(repo, "core", "https://example.com/core")
1500	s1, _, rc1 = _cli(["workspace", "list", "--json"], repo)
1501	s2, _, rc2 = _cli(["workspace", "list", "-j"], repo)
1502	assert rc1 == 0 and rc2 == 0
1503	assert json.loads(_json_blob(s1)) == json.loads(_json_blob(s2))
1504
1505	def test_list_empty_exits_0(self, tmp_path: pathlib.Path) -> None:
1506	"""List with no members exits 0 and returns empty array."""
1507	repo = _make_repo(tmp_path)
1508	stdout, _, rc = _cli(["workspace", "list", "-j"], repo)
1509	assert rc == 0
1510	assert json.loads(_json_blob(stdout)) == []
1511
1512	def test_list_json_is_array(self, tmp_path: pathlib.Path) -> None:
1513	"""JSON output is always a JSON array."""
1514	repo = _make_repo(tmp_path)
1515	add_workspace_member(repo, "core", "https://example.com/core")
1516	stdout, _, rc = _cli(["workspace", "list", "-j"], repo)
1517	assert rc == 0
1518	raw = json.loads(_json_blob(stdout))
1519	assert isinstance(raw, list)
1520
1521	def test_list_json_all_fields_present(self, tmp_path: pathlib.Path) -> None:
1522	"""Every entry has the seven required fields."""
1523	repo = _make_repo(tmp_path)
1524	add_workspace_member(repo, "core", "https://example.com/core")
1525	stdout, _, rc = _cli(["workspace", "list", "-j"], repo)
1526	assert rc == 0
1527	raw = json.loads(_json_blob(stdout))
1528	assert len(raw) == 1
1529	entry = raw[0]
1530	for field in ("name", "url", "path", "branch", "present", "head_commit", "dirty"):
1531	assert field in entry, f"field '{field}' missing"
1532
1533	def test_list_json_name_matches(self, tmp_path: pathlib.Path) -> None:
1534	"""name field in JSON matches the registered member name."""
1535	repo = _make_repo(tmp_path)
1536	add_workspace_member(repo, "myrepo", "https://example.com/myrepo")
1537	members = _parse_list(_cli(["workspace", "list", "-j"], repo)[0])
1538	assert any(m["name"] == "myrepo" for m in members)
1539
1540	def test_list_json_url_matches(self, tmp_path: pathlib.Path) -> None:
1541	"""url field in JSON matches the registered URL."""
1542	repo = _make_repo(tmp_path)
1543	url = "https://example.com/myrepo"
1544	add_workspace_member(repo, "myrepo", url)
1545	members = _parse_list(_cli(["workspace", "list", "-j"], repo)[0])
1546	assert members[0]["url"] == url
1547
1548	def test_list_json_branch_default_main(self, tmp_path: pathlib.Path) -> None:
1549	"""branch field defaults to 'main'."""
1550	repo = _make_repo(tmp_path)
1551	add_workspace_member(repo, "core", "https://example.com/core")
1552	members = _parse_list(_cli(["workspace", "list", "-j"], repo)[0])
1553	assert members[0]["branch"] == "main"
1554
1555	def test_list_json_present_false_when_not_cloned(self, tmp_path: pathlib.Path) -> None:
1556	"""present=false when the checkout directory does not exist."""
1557	repo = _make_repo(tmp_path)
1558	add_workspace_member(repo, "core", "https://example.com/core")
1559	members = _parse_list(_cli(["workspace", "list", "-j"], repo)[0])
1560	assert members[0]["present"] is False
1561
1562	def test_list_json_head_commit_null_when_not_cloned(self, tmp_path: pathlib.Path) -> None:
1563	"""head_commit is null when the member is not yet cloned."""
1564	repo = _make_repo(tmp_path)
1565	add_workspace_member(repo, "core", "https://example.com/core")
1566	members = _parse_list(_cli(["workspace", "list", "-j"], repo)[0])
1567	assert members[0]["head_commit"] is None
1568
1569	def test_list_json_count_matches_registered(self, tmp_path: pathlib.Path) -> None:
1570	"""Array length equals number of registered members."""
1571	repo = _make_repo(tmp_path)
1572	for i in range(5):
1573	add_workspace_member(repo, f"svc{i}", f"https://example.com/svc{i}")
1574	members = _parse_list(_cli(["workspace", "list", "-j"], repo)[0])
1575	assert len(members) == 5
1576
1577	def test_list_json_reflects_update(self, tmp_path: pathlib.Path) -> None:
1578	"""Updated branch appears in list JSON after update."""
1579	repo = _make_repo(tmp_path)
1580	add_workspace_member(repo, "core", "https://example.com/core")
1581	update_workspace_member(repo, "core", branch="release")
1582	members = _parse_list(_cli(["workspace", "list", "-j"], repo)[0])
1583	assert members[0]["branch"] == "release"
1584
1585	def test_list_json_member_removed_not_shown(self, tmp_path: pathlib.Path) -> None:
1586	"""Removed member no longer appears in list."""
1587	repo = _make_repo(tmp_path)
1588	add_workspace_member(repo, "core", "https://example.com/core")
1589	add_workspace_member(repo, "data", "https://example.com/data")
1590	remove_workspace_member(repo, "core")
1591	members = _parse_list(_cli(["workspace", "list", "-j"], repo)[0])
1592	assert all(m["name"] != "core" for m in members)
1593	assert any(m["name"] == "data" for m in members)
1594
1595	def test_list_default_is_text(self, tmp_path: pathlib.Path) -> None:
1596	"""Without --json output is human-readable text."""
1597	repo = _make_repo(tmp_path)
1598	add_workspace_member(repo, "core", "https://example.com/core")
1599	stdout, _, rc = _cli(["workspace", "list"], repo)
1600	assert rc == 0
1601	assert not stdout.strip().startswith("[")
1602
1603	def test_list_text_empty_message(self, tmp_path: pathlib.Path) -> None:
1604	"""Text output says 'No workspace members' when list is empty."""
1605	repo = _make_repo(tmp_path)
1606	stdout, _, rc = _cli(["workspace", "list"], repo)
1607	assert rc == 0
1608	assert "No workspace members" in stdout
1609
1610	def test_list_text_shows_member_name(self, tmp_path: pathlib.Path) -> None:
1611	"""Text output includes the member name."""
1612	repo = _make_repo(tmp_path)
1613	add_workspace_member(repo, "myrepo", "https://example.com/myrepo")
1614	stdout, _, rc = _cli(["workspace", "list"], repo)
1615	assert rc == 0
1616	assert "myrepo" in stdout
1617
1618	def test_list_outside_repo_succeeds_empty(self, tmp_path: pathlib.Path) -> None:
1619	"""Workspace list from a non-repo dir returns empty list — no muse repo required."""
1620	empty = tmp_path / "empty"
1621	empty.mkdir()
1622	stdout, _, rc = _cli(["workspace", "list", "--json"], empty)
1623	assert rc == 0
1624	assert stdout.strip() == "[]"
1625
1626	def test_list_help_has_description(self, tmp_path: pathlib.Path) -> None:
1627	"""--help includes the rich description."""
1628	repo = _make_repo(tmp_path)
1629	stdout, _, rc = _cli(["workspace", "list", "--help"], repo)
1630	assert rc == 0
1631	assert "Agent quickstart" in stdout or "JSON output schema" in stdout
1632
1633	def test_list_json_valid_types(self, tmp_path: pathlib.Path) -> None:
1634	"""All JSON field types are correct."""
1635	repo = _make_repo(tmp_path)
1636	add_workspace_member(repo, "core", "https://example.com/core")
1637	raw = json.loads(_json_blob(_cli(["workspace", "list", "-j"], repo)[0]))
1638	entry = raw[0]
1639	assert isinstance(entry["name"], str)
1640	assert isinstance(entry["url"], str)
1641	assert isinstance(entry["path"], str)
1642	assert isinstance(entry["branch"], str)
1643	assert isinstance(entry["present"], bool)
1644	assert entry["head_commit"] is None or isinstance(entry["head_commit"], str)
1645	assert isinstance(entry["dirty"], bool)
1646
1647
1648	class TestWorkspaceListSecurity:
1649	"""ANSI sanitization and output integrity."""
1650
1651	def test_list_json_ansi_in_name_sanitized(self, tmp_path: pathlib.Path) -> None:
1652	"""ANSI codes in stored member name are stripped from JSON output."""
1653	repo = _make_repo(tmp_path)
1654	# Inject ANSI directly into manifest via core (bypassing CLI validation)
1655	manifest_path = repo / ".muse" / "workspace.toml"
1656	manifest_path.parent.mkdir(parents=True, exist_ok=True)
1657	manifest_path.write_text(
1658	'[workspace]\n[[workspace.members]]\n'
1659	'name = "core\\u001b[31mred\\u001b[0m"\n'
1660	'url = "https://example.com/core"\n'
1661	'path = "repos/core"\n'
1662	'branch = "main"\n'
1663	)
1664	stdout, _, rc = _cli(["workspace", "list", "-j"], repo)
1665	assert rc == 0
1666	assert "\x1b" not in stdout
1667
1668	def test_list_json_ansi_in_url_sanitized(self, tmp_path: pathlib.Path) -> None:
1669	"""ANSI codes in stored URL are stripped from JSON output."""
1670	repo = _make_repo(tmp_path)
1671	manifest_path = repo / ".muse" / "workspace.toml"
1672	manifest_path.parent.mkdir(parents=True, exist_ok=True)
1673	manifest_path.write_text(
1674	'[workspace]\n[[workspace.members]]\n'
1675	'name = "core"\n'
1676	'url = "https://example.com/core\\u001b[31m"\n'
1677	'path = "repos/core"\n'
1678	'branch = "main"\n'
1679	)
1680	stdout, _, rc = _cli(["workspace", "list", "-j"], repo)
1681	assert rc == 0
1682	assert "\x1b" not in stdout
1683
1684	def test_list_text_no_ansi_in_output(self, tmp_path: pathlib.Path) -> None:
1685	"""Text output contains no ANSI escape sequences."""
1686	repo = _make_repo(tmp_path)
1687	add_workspace_member(repo, "core", "https://example.com/core")
1688	stdout, _, rc = _cli(["workspace", "list"], repo)
1689	assert rc == 0
1690	assert "\x1b" not in stdout
1691
1692	def test_list_json_is_valid_json(self, tmp_path: pathlib.Path) -> None:
1693	"""JSON output is well-formed even with multiple members."""
1694	repo = _make_repo(tmp_path)
1695	for i in range(3):
1696	add_workspace_member(repo, f"svc{i}", f"https://example.com/svc{i}")
1697	stdout, _, rc = _cli(["workspace", "list", "-j"], repo)
1698	assert rc == 0
1699	raw = json.loads(_json_blob(stdout))
1700	assert isinstance(raw, list)
1701	assert len(raw) == 3
1702
1703	def test_list_json_dirty_is_bool(self, tmp_path: pathlib.Path) -> None:
1704	"""dirty field is always a boolean, never a string or int."""
1705	repo = _make_repo(tmp_path)
1706	add_workspace_member(repo, "core", "https://example.com/core")
1707	raw = json.loads(_json_blob(_cli(["workspace", "list", "-j"], repo)[0]))
1708	assert isinstance(raw[0]["dirty"], bool)
1709
1710
1711	class TestWorkspaceListStress:
1712	"""Performance and scale tests for workspace list."""
1713
1714	def test_list_50_members(self, tmp_path: pathlib.Path) -> None:
1715	"""List returns all 50 members when 50 are registered."""
1716	repo = _make_repo(tmp_path)
1717	for i in range(50):
1718	add_workspace_member(repo, f"svc{i:03d}", f"https://example.com/svc{i}")
1719	members = _parse_list(_cli(["workspace", "list", "-j"], repo)[0])
1720	assert len(members) == 50
1721
1722	def test_list_performance_50_members(self, tmp_path: pathlib.Path) -> None:
1723	"""Listing 50 members completes within 5 seconds."""
1724	repo = _make_repo(tmp_path)
1725	for i in range(50):
1726	add_workspace_member(repo, f"svc{i:03d}", f"https://example.com/svc{i}")
1727	t0 = time.monotonic()
1728	stdout, _, rc = _cli(["workspace", "list", "-j"], repo)
1729	elapsed = time.monotonic() - t0
1730	assert rc == 0
1731	assert elapsed < 5.0, f"list of 50 took {elapsed:.2f}s"
1732
1733	def test_list_concurrent_reads_consistent(self, tmp_path: pathlib.Path) -> None:
1734	"""Concurrent list reads all return the same member count."""
1735	repo = _make_repo(tmp_path)
1736	for i in range(20):
1737	add_workspace_member(repo, f"svc{i}", f"https://example.com/svc{i}")
1738	counts: list[int] = []
1739	errors: list[str] = []
1740	lock = threading.Lock()
1741
1742	def _run() -> None:
1743	stdout, _, rc = _cli(["workspace", "list", "-j"], repo)
1744	with lock:
1745	if rc != 0:
1746	errors.append(stdout)
1747	else:
1748	counts.append(len(json.loads(_json_blob(stdout))))
1749
1750	threads = [threading.Thread(target=_run) for _ in range(8)]
1751	for t in threads:
1752	t.start()
1753	for t in threads:
1754	t.join()
1755	assert not errors, f"Concurrent list errors: {errors}"
1756	assert all(c == 20 for c in counts), f"Inconsistent counts: {counts}"
1757
1758
1759	# ---------------------------------------------------------------------------
1760	# workspace remove — Extended, Security, Stress
1761	# ---------------------------------------------------------------------------
1762
1763
1764	class TestWorkspaceRemoveExtended:
1765	"""Extended unit / integration / e2e tests for muse workspace remove."""
1766
1767	def test_remove_exits_0_on_success(self, tmp_path: pathlib.Path) -> None:
1768	"""Successful remove exits with code 0."""
1769	repo = _make_repo(tmp_path)
1770	add_workspace_member(repo, "core", "https://example.com/core")
1771	_, _, rc = _cli(["workspace", "remove", "core"], repo)
1772	assert rc == 0
1773
1774	def test_remove_j_alias_works(self, tmp_path: pathlib.Path) -> None:
1775	"""-j is an accepted alias for --json."""
1776	repo = _make_repo(tmp_path)
1777	add_workspace_member(repo, "core", "https://example.com/core")
1778	stdout, _, rc = _cli(["workspace", "remove", "core", "-j"], repo)
1779	assert rc == 0
1780	d = _parse_remove(stdout)
1781	assert d["removed"] is True
1782
1783	def test_remove_json_name_matches(self, tmp_path: pathlib.Path) -> None:
1784	"""JSON output name matches the removed member's name."""
1785	repo = _make_repo(tmp_path)
1786	add_workspace_member(repo, "sounds", "https://example.com/sounds")
1787	stdout, _, rc = _cli(["workspace", "remove", "sounds", "--json"], repo)
1788	assert rc == 0
1789	d = _parse_remove(stdout)
1790	assert d["name"] == "sounds"
1791
1792	def test_remove_json_removed_true(self, tmp_path: pathlib.Path) -> None:
1793	"""JSON output always has removed=true on success."""
1794	repo = _make_repo(tmp_path)
1795	add_workspace_member(repo, "core", "https://example.com/core")
1796	stdout, _, rc = _cli(["workspace", "remove", "core", "--json"], repo)
1797	assert rc == 0
1798	assert json.loads(_json_blob(stdout))["removed"] is True
1799
1800	def test_remove_member_no_longer_in_list(self, tmp_path: pathlib.Path) -> None:
1801	"""After remove, the member is absent from workspace list."""
1802	repo = _make_repo(tmp_path)
1803	add_workspace_member(repo, "core", "https://example.com/core")
1804	add_workspace_member(repo, "data", "https://example.com/data")
1805	_cli(["workspace", "remove", "core"], repo)
1806	members = _parse_list(_cli(["workspace", "list", "--json"], repo)[0])
1807	names = [m["name"] for m in members]
1808	assert "core" not in names
1809	assert "data" in names
1810
1811	def test_remove_only_named_member_removed(self, tmp_path: pathlib.Path) -> None:
1812	"""Remove deletes exactly one member; others are untouched."""
1813	repo = _make_repo(tmp_path)
1814	for n in ("alpha", "beta", "gamma"):
1815	add_workspace_member(repo, n, f"https://example.com/{n}")
1816	_cli(["workspace", "remove", "beta"], repo)
1817	members = _parse_list(_cli(["workspace", "list", "--json"], repo)[0])
1818	names = [m["name"] for m in members]
1819	assert names == ["alpha", "gamma"]
1820
1821	def test_remove_idempotent_error_on_second_call(self, tmp_path: pathlib.Path) -> None:
1822	"""Removing the same member twice returns an error on the second call."""
1823	repo = _make_repo(tmp_path)
1824	add_workspace_member(repo, "core", "https://example.com/core")
1825	_, _, rc1 = _cli(["workspace", "remove", "core"], repo)
1826	_, _, rc2 = _cli(["workspace", "remove", "core"], repo)
1827	assert rc1 == 0
1828	assert rc2 != 0
1829
1830	def test_remove_nonexistent_exits_1(self, tmp_path: pathlib.Path) -> None:
1831	"""Removing a non-existent member exits with code 1."""
1832	repo = _make_repo(tmp_path)
1833	add_workspace_member(repo, "core", "https://example.com/core")
1834	_, _, rc = _cli(["workspace", "remove", "ghost"], repo)
1835	assert rc == 1
1836
1837	def test_remove_outside_repo_exits_1_member_not_found(self, tmp_path: pathlib.Path) -> None:
1838	"""Workspace remove from a non-repo dir exits 1 (member not found), not 2."""
1839	empty = tmp_path / "not_a_repo"
1840	empty.mkdir()
1841	_, _, rc = _cli(["workspace", "remove", "core"], empty)
1842	assert rc == 1
1843
1844	def test_remove_text_output_contains_name(self, tmp_path: pathlib.Path) -> None:
1845	"""Text output mentions the removed member's name."""
1846	repo = _make_repo(tmp_path)
1847	add_workspace_member(repo, "sounds", "https://example.com/sounds")
1848	stdout, _, rc = _cli(["workspace", "remove", "sounds"], repo)
1849	assert rc == 0
1850	assert "sounds" in stdout
1851
1852	def test_remove_text_success_marker(self, tmp_path: pathlib.Path) -> None:
1853	"""Text output contains a success indicator."""
1854	repo = _make_repo(tmp_path)
1855	add_workspace_member(repo, "core", "https://example.com/core")
1856	stdout, _, _ = _cli(["workspace", "remove", "core"], repo)
1857	assert "Removed" in stdout or "✅" in stdout
1858
1859	def test_remove_error_to_stderr_not_stdout(self, tmp_path: pathlib.Path) -> None:
1860	"""Error messages go to stderr; stdout is empty on failure."""
1861	repo = _make_repo(tmp_path)
1862	add_workspace_member(repo, "core", "https://example.com/core")
1863	stdout, stderr, rc = _cli(["workspace", "remove", "ghost"], repo)
1864	assert rc != 0
1865	assert "not found" in stderr
1866	assert "not found" not in stdout
1867
1868	def test_remove_text_no_json_on_success(self, tmp_path: pathlib.Path) -> None:
1869	"""Without --json, stdout does not contain a JSON object."""
1870	repo = _make_repo(tmp_path)
1871	add_workspace_member(repo, "core", "https://example.com/core")
1872	stdout, _, rc = _cli(["workspace", "remove", "core"], repo)
1873	assert rc == 0
1874	assert not stdout.strip().startswith("{")
1875
1876	def test_remove_count_decreases(self, tmp_path: pathlib.Path) -> None:
1877	"""Member count decreases by exactly one after remove."""
1878	repo = _make_repo(tmp_path)
1879	for n in ("a", "b", "c"):
1880	add_workspace_member(repo, n, f"https://example.com/{n}")
1881	before = len(_parse_list(_cli(["workspace", "list", "--json"], repo)[0]))
1882	_cli(["workspace", "remove", "b"], repo)
1883	after = len(_parse_list(_cli(["workspace", "list", "--json"], repo)[0]))
1884	assert after == before - 1
1885
1886	def test_remove_last_member_leaves_empty_manifest(self, tmp_path: pathlib.Path) -> None:
1887	"""Removing the only member results in an empty list."""
1888	repo = _make_repo(tmp_path)
1889	add_workspace_member(repo, "only", "https://example.com/only")
1890	_cli(["workspace", "remove", "only"], repo)
1891	members = _parse_list(_cli(["workspace", "list", "--json"], repo)[0])
1892	assert members == []
1893
1894	def test_remove_help_description_present(self, tmp_path: pathlib.Path) -> None:
1895	"""--help output contains the agent-friendly description."""
1896	repo = _make_repo(tmp_path)
1897	stdout, _, _ = _cli(["workspace", "remove", "--help"], repo)
1898	assert "Unregister" in stdout or "manifest" in stdout
1899
1900	def test_remove_json_schema_keys(self, tmp_path: pathlib.Path) -> None:
1901	"""JSON output has exactly the keys: name, removed."""
1902	repo = _make_repo(tmp_path)
1903	add_workspace_member(repo, "core", "https://example.com/core")
1904	stdout, _, rc = _cli(["workspace", "remove", "core", "--json"], repo)
1905	assert rc == 0
1906	d = json.loads(_json_blob(stdout))
1907	assert set(d.keys()) == {"name", "removed"}
1908
1909	def test_remove_no_manifest_exits_1(self, tmp_path: pathlib.Path) -> None:
1910	"""Remove on a repo with no workspace manifest exits 1."""
1911	repo = _make_repo(tmp_path)
1912	# No workspace.toml — remove_workspace_member raises ValueError
1913	_, stderr, rc = _cli(["workspace", "remove", "core"], repo)
1914	assert rc == 1
1915	assert stderr.strip() != ""
1916
1917
1918	class TestWorkspaceRemoveSecurity:
1919	"""Security hardening tests for muse workspace remove."""
1920
1921	def test_remove_ansi_in_name_arg_sanitized_in_json(self, tmp_path: pathlib.Path) -> None:
1922	"""ANSI codes in a stored name are stripped from JSON output."""
1923	repo = _make_repo(tmp_path)
1924	# Write manifest directly with an ANSI-injected name so it bypasses validator
1925	manifest_path = repo / ".muse" / "workspace.toml"
1926	manifest_path.parent.mkdir(parents=True, exist_ok=True)
1927	manifest_path.write_text(
1928	'[workspace]\n[[workspace.members]]\n'
1929	'name = "evil\\u001b[31m"\n'
1930	'url = "https://example.com/evil"\n'
1931	'path = "repos/evil"\n'
1932	'branch = "main"\n'
1933	)
1934	# Use the raw stored name as the CLI arg to remove it
1935	stdout, _, rc = _cli(["workspace", "remove", "evil\x1b[31m", "--json"], repo)
1936	# Either succeeds (found) or fails (not found) — either way, no ANSI in stdout
1937	assert "\x1b" not in stdout
1938
1939	def test_remove_text_output_no_ansi(self, tmp_path: pathlib.Path) -> None:
1940	"""Text output contains no ANSI escape sequences."""
1941	repo = _make_repo(tmp_path)
1942	add_workspace_member(repo, "core", "https://example.com/core")
1943	stdout, _, _ = _cli(["workspace", "remove", "core"], repo)
1944	assert "\x1b" not in stdout
1945
1946	def test_remove_json_valid_on_success(self, tmp_path: pathlib.Path) -> None:
1947	"""JSON output is well-formed on success."""
1948	repo = _make_repo(tmp_path)
1949	add_workspace_member(repo, "core", "https://example.com/core")
1950	stdout, _, rc = _cli(["workspace", "remove", "core", "--json"], repo)
1951	assert rc == 0
1952	d = json.loads(_json_blob(stdout))
1953	assert isinstance(d["name"], str)
1954	assert d["removed"] is True
1955
1956	def test_remove_removed_field_is_bool(self, tmp_path: pathlib.Path) -> None:
1957	"""removed field is a boolean, never a string or int."""
1958	repo = _make_repo(tmp_path)
1959	add_workspace_member(repo, "core", "https://example.com/core")
1960	stdout, _, rc = _cli(["workspace", "remove", "core", "--json"], repo)
1961	assert rc == 0
1962	d = json.loads(_json_blob(stdout))
1963	assert isinstance(d["removed"], bool)
1964
1965	def test_remove_symlink_manifest_fails_gracefully(self, tmp_path: pathlib.Path) -> None:
1966	"""A symlinked manifest is refused — exits non-zero without crashing."""
1967	repo = _make_repo(tmp_path)
1968	add_workspace_member(repo, "core", "https://example.com/core")
1969	manifest_path = repo / ".muse" / "workspace.toml"
1970	real = tmp_path / "real_workspace.toml"
1971	real.write_text(manifest_path.read_text())
1972	manifest_path.unlink()
1973	manifest_path.symlink_to(real)
1974	_, _, rc = _cli(["workspace", "remove", "core"], repo)
1975	# The symlink guard in _load_manifest returns None → ValueError → rc 1
1976	assert rc != 0
1977
1978	def test_remove_null_byte_in_name_raises(self, tmp_path: pathlib.Path) -> None:
1979	"""Null byte in name is rejected by the core validator."""
1980	repo = _make_repo(tmp_path)
1981	add_workspace_member(repo, "core", "https://example.com/core")
1982	# remove_workspace_member does a name-equality match; null byte won't match
1983	# any valid stored name — raises ValueError (not found)
1984	with pytest.raises(ValueError):
1985	remove_workspace_member(repo, "core\x00evil")
1986
1987
1988	class TestWorkspaceRemoveStress:
1989	"""Performance and scale tests for muse workspace remove."""
1990
1991	def test_remove_from_50_member_manifest(self, tmp_path: pathlib.Path) -> None:
1992	"""Remove works correctly when the manifest has 50 members."""
1993	repo = _make_repo(tmp_path)
1994	for i in range(50):
1995	add_workspace_member(repo, f"svc{i:03d}", f"https://example.com/svc{i}")
1996	_, _, rc = _cli(["workspace", "remove", "svc025", "--json"], repo)
1997	assert rc == 0
1998	members = _parse_list(_cli(["workspace", "list", "--json"], repo)[0])
1999	assert len(members) == 49
2000	assert all(m["name"] != "svc025" for m in members)
2001
2002	def test_remove_performance_50_members(self, tmp_path: pathlib.Path) -> None:
2003	"""Removing from a 50-member manifest completes within 5 seconds."""
2004	repo = _make_repo(tmp_path)
2005	for i in range(50):
2006	add_workspace_member(repo, f"svc{i:03d}", f"https://example.com/svc{i}")
2007	t0 = time.monotonic()
2008	_, _, rc = _cli(["workspace", "remove", "svc000", "--json"], repo)
2009	elapsed = time.monotonic() - t0
2010	assert rc == 0
2011	assert elapsed < 5.0, f"remove from 50 took {elapsed:.2f}s"
2012
2013	def test_remove_sequential_removes_all(self, tmp_path: pathlib.Path) -> None:
2014	"""Removing all 20 members one-by-one leaves an empty list."""
2015	repo = _make_repo(tmp_path)
2016	names = [f"svc{i:02d}" for i in range(20)]
2017	for n in names:
2018	add_workspace_member(repo, n, f"https://example.com/{n}")
2019	for n in names:
2020	_, _, rc = _cli(["workspace", "remove", n], repo)
2021	assert rc == 0
2022	members = _parse_list(_cli(["workspace", "list", "--json"], repo)[0])
2023	assert members == []
2024
2025
2026	# ---------------------------------------------------------------------------
2027	# workspace status — Extended, Security, Stress
2028	# ---------------------------------------------------------------------------
2029
2030
2031	class TestWorkspaceStatusExtended:
2032	"""Extended unit / integration / e2e tests for muse workspace status."""
2033
2034	def test_status_exits_0_all_members(self, tmp_path: pathlib.Path) -> None:
2035	repo = _make_repo(tmp_path)
2036	add_workspace_member(repo, "core", "https://example.com/core")
2037	_, _, rc = _cli(["workspace", "status"], repo)
2038	assert rc == 0
2039
2040	def test_status_j_alias_works(self, tmp_path: pathlib.Path) -> None:
2041	repo = _make_repo(tmp_path)
2042	add_workspace_member(repo, "core", "https://example.com/core")
2043	stdout, _, rc = _cli(["workspace", "status", "-j"], repo)
2044	assert rc == 0
2045	members = _parse_list(stdout)
2046	assert len(members) == 1
2047
2048	def test_status_named_exits_0(self, tmp_path: pathlib.Path) -> None:
2049	repo = _make_repo(tmp_path)
2050	add_workspace_member(repo, "core", "https://example.com/core")
2051	_, _, rc = _cli(["workspace", "status", "core"], repo)
2052	assert rc == 0
2053
2054	def test_status_named_json_single_element(self, tmp_path: pathlib.Path) -> None:
2055	repo = _make_repo(tmp_path)
2056	add_workspace_member(repo, "core", "https://example.com/core")
2057	add_workspace_member(repo, "data", "https://example.com/data")
2058	stdout, _, rc = _cli(["workspace", "status", "core", "--json"], repo)
2059	assert rc == 0
2060	members = _parse_list(stdout)
2061	assert len(members) == 1
2062	assert members[0]["name"] == "core"
2063
2064	def test_status_all_json_all_members_returned(self, tmp_path: pathlib.Path) -> None:
2065	repo = _make_repo(tmp_path)
2066	for n in ("alpha", "beta", "gamma"):
2067	add_workspace_member(repo, n, f"https://example.com/{n}")
2068	stdout, _, rc = _cli(["workspace", "status", "--json"], repo)
2069	assert rc == 0
2070	members = _parse_list(stdout)
2071	assert {m["name"] for m in members} == {"alpha", "beta", "gamma"}
2072
2073	def test_status_json_ten_fields(self, tmp_path: pathlib.Path) -> None:
2074	repo = _make_repo(tmp_path)
2075	add_workspace_member(repo, "core", "https://example.com/core")
2076	stdout, _, rc = _cli(["workspace", "status", "--json"], repo)
2077	assert rc == 0
2078	d = json.loads(_json_blob(stdout))[0]
2079	assert set(d.keys()) == {
2080	"name", "url", "path", "branch", "present", "head_commit", "dirty",
2081	"actual_branch", "stash_count", "feature_branches",
2082	}
2083
2084	def test_status_json_present_false_when_not_cloned(self, tmp_path: pathlib.Path) -> None:
2085	repo = _make_repo(tmp_path)
2086	add_workspace_member(repo, "core", "https://example.com/core")
2087	stdout, _, rc = _cli(["workspace", "status", "--json"], repo)
2088	assert rc == 0
2089	assert _parse_list(stdout)[0]["present"] is False
2090
2091	def test_status_json_head_commit_null_when_not_cloned(self, tmp_path: pathlib.Path) -> None:
2092	repo = _make_repo(tmp_path)
2093	add_workspace_member(repo, "core", "https://example.com/core")
2094	stdout, _, rc = _cli(["workspace", "status", "--json"], repo)
2095	assert rc == 0
2096	assert _parse_list(stdout)[0]["head_commit"] is None
2097
2098	def test_status_json_dirty_false_when_not_cloned(self, tmp_path: pathlib.Path) -> None:
2099	repo = _make_repo(tmp_path)
2100	add_workspace_member(repo, "core", "https://example.com/core")
2101	stdout, _, rc = _cli(["workspace", "status", "--json"], repo)
2102	assert rc == 0
2103	assert _parse_list(stdout)[0]["dirty"] is False
2104
2105	def test_status_empty_exits_0(self, tmp_path: pathlib.Path) -> None:
2106	repo = _make_repo(tmp_path)
2107	_, _, rc = _cli(["workspace", "status", "--json"], repo)
2108	assert rc == 0
2109
2110	def test_status_empty_json_empty_array(self, tmp_path: pathlib.Path) -> None:
2111	repo = _make_repo(tmp_path)
2112	stdout, _, rc = _cli(["workspace", "status", "--json"], repo)
2113	assert rc == 0
2114	assert _parse_list(stdout) == []
2115
2116	def test_status_nonexistent_name_exits_1(self, tmp_path: pathlib.Path) -> None:
2117	repo = _make_repo(tmp_path)
2118	add_workspace_member(repo, "core", "https://example.com/core")
2119	_, _, rc = _cli(["workspace", "status", "ghost"], repo)
2120	assert rc == 1
2121
2122	def test_status_nonexistent_error_to_stderr(self, tmp_path: pathlib.Path) -> None:
2123	repo = _make_repo(tmp_path)
2124	add_workspace_member(repo, "core", "https://example.com/core")
2125	stdout, stderr, rc = _cli(["workspace", "status", "ghost"], repo)
2126	assert rc != 0
2127	assert "not found" in stderr
2128	assert "not found" not in stdout
2129
2130	def test_status_outside_repo_succeeds_empty(self, tmp_path: pathlib.Path) -> None:
2131	"""Workspace status from a non-repo dir returns empty — no muse repo required."""
2132	empty = tmp_path / "not_a_repo"
2133	empty.mkdir()
2134	stdout, _, rc = _cli(["workspace", "status", "--json"], empty)
2135	assert rc == 0
2136	assert stdout.strip() == "[]"
2137
2138	def test_status_text_contains_member_name(self, tmp_path: pathlib.Path) -> None:
2139	repo = _make_repo(tmp_path)
2140	add_workspace_member(repo, "sounds", "https://example.com/sounds")
2141	stdout, _, rc = _cli(["workspace", "status"], repo)
2142	assert rc == 0
2143	assert "sounds" in stdout
2144
2145	def test_status_text_empty_message(self, tmp_path: pathlib.Path) -> None:
2146	repo = _make_repo(tmp_path)
2147	stdout, _, rc = _cli(["workspace", "status"], repo)
2148	assert rc == 0
2149	assert "No workspace members" in stdout
2150
2151	def test_status_help_description_present(self, tmp_path: pathlib.Path) -> None:
2152	repo = _make_repo(tmp_path)
2153	stdout, _, _ = _cli(["workspace", "status", "--help"], repo)
2154	assert "Agent quickstart" in stdout or "present" in stdout
2155
2156	def test_status_json_url_matches_registered(self, tmp_path: pathlib.Path) -> None:
2157	repo = _make_repo(tmp_path)
2158	add_workspace_member(repo, "core", "https://example.com/core")
2159	stdout, _, rc = _cli(["workspace", "status", "--json"], repo)
2160	assert rc == 0
2161	assert _parse_list(stdout)[0]["url"] == "https://example.com/core"
2162
2163
2164	class TestWorkspaceStatusSecurity:
2165	"""Security hardening tests for muse workspace status."""
2166
2167	def test_status_json_no_ansi_in_name(self, tmp_path: pathlib.Path) -> None:
2168	repo = _make_repo(tmp_path)
2169	manifest_path = repo / ".muse" / "workspace.toml"
2170	manifest_path.parent.mkdir(parents=True, exist_ok=True)
2171	manifest_path.write_text(
2172	'[workspace]\n[[workspace.members]]\n'
2173	'name = "evil\\u001b[31m"\n'
2174	'url = "https://example.com/evil"\n'
2175	'path = "repos/evil"\n'
2176	'branch = "main"\n'
2177	)
2178	stdout, _, rc = _cli(["workspace", "status", "--json"], repo)
2179	assert rc == 0
2180	assert "\x1b" not in stdout
2181
2182	def test_status_json_no_ansi_in_url(self, tmp_path: pathlib.Path) -> None:
2183	repo = _make_repo(tmp_path)
2184	manifest_path = repo / ".muse" / "workspace.toml"
2185	manifest_path.parent.mkdir(parents=True, exist_ok=True)
2186	manifest_path.write_text(
2187	'[workspace]\n[[workspace.members]]\n'
2188	'name = "core"\n'
2189	'url = "https://example.com/\\u001b[31mcore"\n'
2190	'path = "repos/core"\n'
2191	'branch = "main"\n'
2192	)
2193	stdout, _, rc = _cli(["workspace", "status", "--json"], repo)
2194	assert rc == 0
2195	assert "\x1b" not in stdout
2196
2197	def test_status_text_no_ansi(self, tmp_path: pathlib.Path) -> None:
2198	repo = _make_repo(tmp_path)
2199	add_workspace_member(repo, "core", "https://example.com/core")
2200	stdout, _, _ = _cli(["workspace", "status"], repo)
2201	assert "\x1b" not in stdout
2202
2203	def test_status_json_valid_json(self, tmp_path: pathlib.Path) -> None:
2204	repo = _make_repo(tmp_path)
2205	for i in range(3):
2206	add_workspace_member(repo, f"svc{i}", f"https://example.com/svc{i}")
2207	stdout, _, rc = _cli(["workspace", "status", "--json"], repo)
2208	assert rc == 0
2209	raw = json.loads(_json_blob(stdout))
2210	assert isinstance(raw, list)
2211	assert len(raw) == 3
2212
2213	def test_status_json_bool_fields_are_bool(self, tmp_path: pathlib.Path) -> None:
2214	repo = _make_repo(tmp_path)
2215	add_workspace_member(repo, "core", "https://example.com/core")
2216	stdout, _, rc = _cli(["workspace", "status", "--json"], repo)
2217	assert rc == 0
2218	d = json.loads(_json_blob(stdout))[0]
2219	assert isinstance(d["present"], bool)
2220	assert isinstance(d["dirty"], bool)
2221
2222	def test_status_symlink_manifest_exits_nonzero(self, tmp_path: pathlib.Path) -> None:
2223	repo = _make_repo(tmp_path)
2224	add_workspace_member(repo, "core", "https://example.com/core")
2225	manifest_path = repo / ".muse" / "workspace.toml"
2226	real = tmp_path / "real.toml"
2227	real.write_text(manifest_path.read_text())
2228	manifest_path.unlink()
2229	manifest_path.symlink_to(real)
2230	# symlink guard returns None → empty list (exits 0, empty array)
2231	stdout, _, rc = _cli(["workspace", "status", "--json"], repo)
2232	assert rc == 0
2233	assert _parse_list(stdout) == []
2234
2235
2236	class TestWorkspaceStatusStress:
2237	"""Performance and scale tests for muse workspace status."""
2238
2239	def test_status_50_members_all_returned(self, tmp_path: pathlib.Path) -> None:
2240	repo = _make_repo(tmp_path)
2241	for i in range(50):
2242	add_workspace_member(repo, f"svc{i:03d}", f"https://example.com/svc{i}")
2243	stdout, _, rc = _cli(["workspace", "status", "--json"], repo)
2244	assert rc == 0
2245	assert len(_parse_list(stdout)) == 50
2246
2247	def test_status_performance_50_members(self, tmp_path: pathlib.Path) -> None:
2248	repo = _make_repo(tmp_path)
2249	for i in range(50):
2250	add_workspace_member(repo, f"svc{i:03d}", f"https://example.com/svc{i}")
2251	t0 = time.monotonic()
2252	stdout, _, rc = _cli(["workspace", "status", "--json"], repo)
2253	elapsed = time.monotonic() - t0
2254	assert rc == 0
2255	assert elapsed < 5.0, f"status of 50 took {elapsed:.2f}s"
2256
2257	def test_status_concurrent_reads_consistent(self, tmp_path: pathlib.Path) -> None:
2258	repo = _make_repo(tmp_path)
2259	for i in range(20):
2260	add_workspace_member(repo, f"svc{i}", f"https://example.com/svc{i}")
2261	counts: list[int] = []
2262	errors: list[str] = []
2263	lock = threading.Lock()
2264
2265	def _run() -> None:
2266	stdout, _, rc = _cli(["workspace", "status", "--json"], repo)
2267	with lock:
2268	if rc != 0:
2269	errors.append(stdout)
2270	else:
2271	counts.append(len(json.loads(_json_blob(stdout))))
2272
2273	threads = [threading.Thread(target=_run) for _ in range(8)]
2274	for t in threads:
2275	t.start()
2276	for t in threads:
2277	t.join()
2278	assert not errors
2279	assert all(c == 20 for c in counts)

File History 5 commits

sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6 chore(tests): add docstring to tests/__init__.py so rc14 tr… Human 26 days ago

sha256:d8316ffae901be06347e16ab55be11868eb519dd16ade3e8aa16a99e662f7e62 baseline: rc14 re-baseline after rc3 store corruption recovery Human patch 26 days ago

sha256:50d413a635f57761c3fce1f70251b957b6423821525bf330747ef4007339222e Merge branch 'dev' into main Human 29 days ago

sha256:fb67fed5a4d3e40de84bdd163de94ef1386570bef1dd1a020a732c8a038962ce Merge branch 'dev' into main Human 48 days ago

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

function _make_repo

function _cli

function _json_blob

function _parse_add

function _parse_update

function _parse_list

function _parse_remove

function _parse_sync

function test_toml_escape_plain_string

function test_toml_escape_backslash

function test_toml_escape_double_quote

function test_toml_escape_injection_attempt

function test_toml_escape_newline

function test_toml_escape_carriage_return

function test_toml_escape_tab

function test_toml_escape_roundtrip

function test_validate_name_ok

function test_validate_name_empty_raises

function test_validate_name_too_long_raises

function test_validate_name_slash_raises

function test_validate_name_null_byte_raises

function test_validate_name_space_raises

function test_validate_url_https_ok

function test_validate_url_http_ok

function test_validate_url_local_path_ok

function test_validate_url_null_byte_raises

function test_validate_url_file_scheme_raises

function test_validate_url_ftp_scheme_raises

function test_validate_url_ssh_scheme_raises

function test_validate_path_ok

function test_validate_path_traversal_raises

function test_validate_path_null_byte_raises

function test_load_manifest_symlink_ignored

function test_load_manifest_oversized_ignored

function test_load_manifest_corrupt_toml_ignored

function test_load_manifest_missing_returns_none

function test_save_manifest_rejects_symlink_file

function test_update_url

function test_update_branch

function test_update_path

function test_update_nonexistent_raises

function test_update_invalid_url_raises

function test_get_workspace_member_found

function test_get_workspace_member_not_found_raises

function test_get_workspace_member_no_manifest_raises

function test_member_status_has_dirty_field

function test_sync_dry_run_returns_skipped

function test_sync_dry_run_no_subprocess

function test_sync_empty_manifest_returns_empty

function test_sync_dry_run_pull_action

function test_sync_named_member_only

function test_toml_injection_in_url_is_escaped

function test_ansi_in_name_sanitized_in_output

function test_path_traversal_in_member_path_rejected

function test_file_url_scheme_rejected

function test_null_byte_in_url_rejected

function test_add_duplicate_error_to_stderr

function test_remove_nonexistent_error_to_stderr

function test_update_no_flags_error_to_stderr

function test_status_nonexistent_error_to_stderr

class _AddJson

function test_add_json_schema

class _UpdateJson

function test_update_json_schema

class _ListMemberJson

function test_list_json_schema

function test_list_json_empty_list

class _RemoveJson

function test_remove_json_schema

function test_status_json_all

function test_status_json_named

function test_status_json_empty

class _SyncResultItemJson

class _SyncJson

function test_sync_json_dry_run

function test_sync_json_empty_manifest

function test_lifecycle_add_update_remove

function test_lifecycle_multiple_members

function test_add_custom_branch_and_path

function test_e2e_add_text_output

function test_e2e_list_text_output

function test_e2e_status_text_output

function test_e2e_remove_text_output

function test_e2e_sync_dry_run_text_output

function test_e2e_list_no_members

function test_e2e_update_text_output

function test_e2e_shorthand_branch_flag

function test_stress_50_members_add_list

function test_stress_add_remove_cycle

function test_stress_concurrent_list_reads

function _read

function test_stress_sync_parallel_dry_run

function test_stress_json_list_50_members

function test_stress_update_10_members

class TestWorkspaceAddExtended

function test_add_j_alias

function test_add_json_name_field

function test_add_json_url_field

function test_add_json_default_branch_is_main

function test_add_json_default_path_is_repos_name

function test_add_json_custom_branch

function test_add_json_custom_path

function test_add_json_all_fields_present

function test_add_default_is_text

function test_add_text_contains_name

function test_add_text_hints_sync

function test_add_duplicate_exits_1

function test_add_outside_repo_succeeds

function test_add_appears_in_list_after

function test_add_help_has_description

function test_add_local_path_url_accepted

function test_add_shorthand_branch_flag

function test_add_json_is_valid_json

class TestWorkspaceAddSecurity

function test_add_invalid_url_scheme_rejected

function test_add_ftp_scheme_rejected

function test_add_null_byte_in_url_rejected

function test_add_path_traversal_rejected

function test_add_invalid_name_rejected

function test_add_empty_name_rejected

function test_add_error_goes_to_stderr

function test_add_json_no_ansi_in_output

function test_add_ssh_scheme_rejected

class TestWorkspaceAddStress

function test_add_20_sequential

function test_add_performance

function test_add_remove_add_cycle

class TestWorkspaceUpdateExtended

function test_update_j_alias

function test_update_branch_reflected_in_json

function test_update_url_reflected_in_json

function test_update_path_reflected_in_json

function test_update_json_all_fields_present

function test_update_json_name_unchanged

function test_update_omitted_fields_preserved

function test_update_no_flags_exits_1

function test_update_nonexistent_exits_1

function test_update_outside_repo_exits_1_member_not_found

function test_update_default_is_text

function test_update_text_contains_name

function test_update_help_has_description

function test_update_multiple_flags_at_once

function test_update_reflected_in_list

function test_update_shorthand_branch_flag

function test_update_json_is_valid_json

class TestWorkspaceUpdateSecurity

function test_update_invalid_url_scheme_rejected

function test_update_path_traversal_rejected

function test_update_null_byte_in_path_rejected

function test_update_error_goes_to_stderr

function test_update_json_no_ansi_in_output

function test_update_ftp_url_rejected

function test_update_no_flags_error_to_stderr

class TestWorkspaceUpdateStress

function test_update_10_members_sequential

function test_update_performance

function test_update_repeated_same_member

class TestWorkspaceListExtended

function test_list_j_alias

function test_list_empty_exits_0

function test_list_json_is_array

function test_list_json_all_fields_present

function test_list_json_name_matches

function test_list_json_url_matches

function test_list_json_branch_default_main

function test_list_json_present_false_when_not_cloned

function test_list_json_head_commit_null_when_not_cloned

function test_list_json_count_matches_registered

function test_list_json_reflects_update

function test_list_json_member_removed_not_shown

function test_list_default_is_text

function test_list_text_empty_message

function test_list_text_shows_member_name

function test_list_outside_repo_succeeds_empty

function test_list_help_has_description

function test_list_json_valid_types

class TestWorkspaceListSecurity

function test_list_json_ansi_in_name_sanitized

function test_list_json_ansi_in_url_sanitized

function test_list_text_no_ansi_in_output

function test_list_json_is_valid_json

function test_list_json_dirty_is_bool

class TestWorkspaceListStress

function test_list_50_members

function test_list_performance_50_members

function test_list_concurrent_reads_consistent

function _run

class TestWorkspaceRemoveExtended

function test_remove_exits_0_on_success

function test_remove_j_alias_works

function test_remove_json_name_matches

function test_remove_json_removed_true

function test_remove_member_no_longer_in_list

function test_remove_only_named_member_removed

function test_remove_idempotent_error_on_second_call

function test_remove_nonexistent_exits_1

function test_remove_outside_repo_exits_1_member_not_found

function test_remove_text_output_contains_name

function test_remove_text_success_marker

function test_remove_error_to_stderr_not_stdout

function test_remove_text_no_json_on_success

function test_remove_count_decreases

function test_remove_last_member_leaves_empty_manifest

function test_remove_help_description_present

function test_remove_json_schema_keys

function test_remove_no_manifest_exits_1

class TestWorkspaceRemoveSecurity

function test_remove_ansi_in_name_arg_sanitized_in_json

function test_remove_text_output_no_ansi

function test_remove_json_valid_on_success

function test_remove_removed_field_is_bool

function test_remove_symlink_manifest_fails_gracefully

function test_remove_null_byte_in_name_raises

class TestWorkspaceRemoveStress

function test_remove_from_50_member_manifest

function test_remove_performance_50_members

function test_remove_sequential_removes_all

class TestWorkspaceStatusExtended

function test_status_exits_0_all_members

function test_status_j_alias_works

function test_status_named_exits_0

function test_status_named_json_single_element

function test_status_all_json_all_members_returned

function test_status_json_ten_fields

function test_status_json_present_false_when_not_cloned

function test_status_json_head_commit_null_when_not_cloned

function test_status_json_dirty_false_when_not_cloned

function test_status_empty_exits_0

function test_status_empty_json_empty_array

function test_status_nonexistent_name_exits_1

function test_status_nonexistent_error_to_stderr

function test_status_outside_repo_succeeds_empty

function test_status_text_contains_member_name

function test_status_text_empty_message

function test_status_help_description_present

function test_status_json_url_matches_registered

class TestWorkspaceStatusSecurity

function test_status_json_no_ansi_in_name

function test_status_json_no_ansi_in_url

function test_status_text_no_ansi

function test_status_json_valid_json

function test_status_json_bool_fields_are_bool

function test_status_symlink_manifest_exits_nonzero

class TestWorkspaceStatusStress

function test_status_50_members_all_returned

function test_status_performance_50_members

function test_status_concurrent_reads_consistent

function _run

Pathtests/test_cmd_workspace_hardening.py

Lines2,279

Size93.4 KB

LangPython

Refsha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6

Object ID

sha256:9c89bf95a4aff84732b4bba74a2f68597fe94b7812973e23f55d5ef784edf4b4…

Last commit

sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6

chore(tests): add docstring to tests/__init__.py …

26 days ago

Quick links

Blame History