tests/test_cmd_workspace_hardening.py · gabriel/muse

test_cmd_workspace_hardening.py python

2,255 lines 92.4 KB

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

File History 1 commit

sha256:1c4b3e3a9a1f300774c3ee662b572a698d5fd405bf765a71e6011a2e9c3eaaaa feat: Muse — version control for the agent era Human 152 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_seven_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,255

Size92.4 KB

LangPython

Refsha256:1c4b3e3a9a1f300774c3ee662b572a698d5fd405bf765a71e6011a2e9c3eaaaa

Object ID

sha256:8dcd3e412c5af79614b10211b7df20ab608a94ccee2b227a7264633efa43bfef…

Last commit

sha256:1c4b3e3a9a1f300774c3ee662b572a698d5fd405bf765a71e6011a2e9c3eaaaa

feat: Muse — version control for the agent era

152 days ago

Quick links

Blame History