gabriel / musehub public

test_install_script.py file-level

at sha256:3 · View file ↗ · Intel ↗

History
1 files
1 commits
0 hotspots
0 🧊 dead
0 πŸ’₯ blast risk
sha256:0 fix: fall back to any indexed mpack in read_object_bytes when push mpac… · gabriel · Jun 17, 2026
1 """Section 43 β€” Install Script: 7-layer test suite.
2
3 Covers:
4 - musehub/api/routes/musehub/install.py::sanitize_musehub_url,
5 build_sdist_url, generate_install_script, generate_uninstall_script,
6 GET /install.sh, GET /uninstall.sh
7 """
8 from __future__ import annotations
9
10 import re
11 import time
12
13 import pytest
14
15 from musehub.api.routes.musehub.install import (
16 _SAFE_URL_RE,
17 build_sdist_url,
18 generate_install_script,
19 generate_uninstall_script,
20 sanitize_musehub_url,
21 )
22 from musehub.protocol.version import MUSE_VERSION
23
24
25 # ─────────────────────────────────────────────────────────────────────────────
26 # LAYER 1 β€” UNIT
27 # ─────────────────────────────────────────────────────────────────────────────
28
29
30 class TestSanitizeMusehubUrlUnit:
31 """Unit: sanitize_musehub_url rejects unsafe URLs."""
32
33 def test_accepts_http_url(self) -> None:
34 assert sanitize_musehub_url("https://localhost:1337") == "https://localhost:1337"
35
36 def test_accepts_https_url(self) -> None:
37 assert sanitize_musehub_url("https://musehub.example.com") == "https://musehub.example.com"
38
39 def test_accepts_url_with_path(self) -> None:
40 result = sanitize_musehub_url("https://hub.example.com/muse")
41 assert result == "https://hub.example.com/muse"
42
43 def test_strips_trailing_slash(self) -> None:
44 result = sanitize_musehub_url("https://hub.example.com/")
45 assert not result.endswith("/")
46
47 def test_rejects_url_with_double_quote(self) -> None:
48 with pytest.raises(ValueError):
49 sanitize_musehub_url('https://example.com"evil')
50
51 def test_rejects_url_with_single_quote(self) -> None:
52 with pytest.raises(ValueError):
53 sanitize_musehub_url("https://example.com'evil")
54
55 def test_rejects_url_with_backtick(self) -> None:
56 with pytest.raises(ValueError):
57 sanitize_musehub_url("https://example.com`id`")
58
59 def test_rejects_url_with_dollar(self) -> None:
60 with pytest.raises(ValueError):
61 sanitize_musehub_url("https://example.com$HOME")
62
63 def test_rejects_url_with_semicolon(self) -> None:
64 with pytest.raises(ValueError):
65 sanitize_musehub_url("https://example.com;rm -rf /")
66
67 def test_rejects_url_with_newline(self) -> None:
68 with pytest.raises(ValueError):
69 sanitize_musehub_url("https://example.com\nevil")
70
71 def test_rejects_non_http_scheme(self) -> None:
72 with pytest.raises(ValueError):
73 sanitize_musehub_url("ftp://example.com")
74
75 def test_rejects_empty_string(self) -> None:
76 with pytest.raises(ValueError):
77 sanitize_musehub_url("")
78
79
80 class TestBuildSdistUrlUnit:
81 """Unit: build_sdist_url produces correct URLs β€” no platform/arch needed."""
82
83 def test_basic_url(self) -> None:
84 url = build_sdist_url("https://hub.example.com", "1.2.3")
85 assert url == "https://hub.example.com/releases/muse-1.2.3.tar.gz"
86
87 def test_ends_with_tar_gz(self) -> None:
88 url = build_sdist_url("https://localhost:1337", "0.1.0")
89 assert url.endswith(".tar.gz")
90
91 def test_version_embedded(self) -> None:
92 url = build_sdist_url("http://localhost", "9.9.9")
93 assert "9.9.9" in url
94
95 def test_under_releases_path(self) -> None:
96 url = build_sdist_url("https://hub.example.com", "1.0.0")
97 assert "/releases/" in url
98
99 def test_no_platform_in_url(self) -> None:
100 url = build_sdist_url("https://hub.example.com", "1.0.0")
101 for platform in ("linux", "darwin", "arm64", "x86_64"):
102 assert platform not in url
103
104
105 class TestGenerateInstallScriptUnit:
106 """Unit: generate_install_script content and structure."""
107
108 def test_starts_with_shebang(self) -> None:
109 script = generate_install_script("https://localhost:1337")
110 assert script.startswith("#!/usr/bin/env sh")
111
112 def test_contains_musehub_url(self) -> None:
113 url = "https://hub.example.com"
114 script = generate_install_script(url)
115 assert url in script
116
117 def test_url_is_double_quoted_in_assignment(self) -> None:
118 url = "https://hub.example.com"
119 script = generate_install_script(url)
120 assert f'MUSEHUB_URL="{url}"' in script
121
122 def test_contains_muse_version(self) -> None:
123 script = generate_install_script("https://localhost:1337")
124 assert MUSE_VERSION in script
125
126 def test_contains_set_euf(self) -> None:
127 script = generate_install_script("https://localhost:1337")
128 assert "set -euf" in script
129
130 def test_contains_tar_gz_archive_reference(self) -> None:
131 script = generate_install_script("https://localhost:1337")
132 assert ".tar.gz" in script
133
134 def test_requires_python_314(self) -> None:
135 script = generate_install_script("https://localhost:1337")
136 assert "3.14" in script
137
138 def test_creates_venv(self) -> None:
139 script = generate_install_script("https://localhost:1337")
140 assert "venv" in script
141
142 def test_uses_pip_install(self) -> None:
143 script = generate_install_script("https://localhost:1337")
144 assert "pip" in script and "install" in script
145
146 def test_creates_symlink(self) -> None:
147 script = generate_install_script("https://localhost:1337")
148 assert "ln -sf" in script
149
150 def test_no_eval_in_script(self) -> None:
151 script = generate_install_script("https://localhost:1337")
152 assert "\neval " not in script and " eval " not in script
153
154 def test_raises_for_unsafe_url(self) -> None:
155 with pytest.raises(ValueError):
156 generate_install_script('http://evil.com";rm -rf /')
157
158 def test_custom_version_embedded(self) -> None:
159 script = generate_install_script("https://localhost:1337", version="99.0.0")
160 assert "99.0.0" in script
161
162
163 class TestGenerateUninstallScriptUnit:
164 """Unit: generate_uninstall_script never removes ~/.muse data."""
165
166 def test_starts_with_shebang(self) -> None:
167 script = generate_uninstall_script()
168 assert script.startswith("#!/usr/bin/env sh")
169
170 def test_contains_set_euf(self) -> None:
171 script = generate_uninstall_script()
172 assert "set -euf" in script
173
174 def test_mentions_muse_dir_preserved(self) -> None:
175 script = generate_uninstall_script()
176 assert "~/.muse" in script
177 assert "preserved" in script.lower() or "never" in script.lower()
178
179 def test_does_not_rm_rf_muse_dir(self) -> None:
180 script = generate_uninstall_script()
181 for line in script.splitlines():
182 stripped = line.strip()
183 if stripped.startswith("#"):
184 continue
185 if re.match(r"^rm\s", stripped) and "~/.muse" in stripped:
186 pytest.fail(f"Active rm targeting .muse: {stripped!r}")
187
188 def test_removes_symlink_and_venv(self) -> None:
189 script = generate_uninstall_script()
190 assert "rm" in script
191 assert "venv" in script.lower() or "VENV" in script
192
193
194 # ─────────────────────────────────────────────────────────────────────────────
195 # LAYER 2 β€” INTEGRATION
196 # ─────────────────────────────────────────────────────────────────────────────
197
198
199 class TestInstallScriptIntegration:
200 """Integration: script functions work together coherently."""
201
202 def test_install_script_contains_sdist_url_pattern(self) -> None:
203 base = "https://hub.example.com"
204 script = generate_install_script(base)
205 sdist_url = build_sdist_url(base, MUSE_VERSION)
206 # The sdist URL or its components must be constructable from the script's vars
207 assert base in script
208 assert MUSE_VERSION in script
209 assert sdist_url.endswith(".tar.gz")
210
211 def test_safe_url_re_pattern_matches_valid_urls(self) -> None:
212 valid = [
213 "https://localhost:1337",
214 "https://musehub.example.com",
215 "https://hub.example.com/muse",
216 "http://192.168.1.1:8080",
217 ]
218 for url in valid:
219 assert _SAFE_URL_RE.match(url), f"Expected {url!r} to be safe"
220
221 def test_safe_url_re_pattern_rejects_metacharacters(self) -> None:
222 dangerous = [
223 "http://evil.com\";rm -rf /",
224 "http://evil.com`id`",
225 "http://evil.com$HOME",
226 "http://evil.com;evil",
227 "http://evil.com'x",
228 ]
229 for url in dangerous:
230 assert not _SAFE_URL_RE.match(url), f"Expected {url!r} to be rejected"
231
232 def test_generate_install_script_version_matches_muse_version_default(self) -> None:
233 script = generate_install_script("http://localhost")
234 assert MUSE_VERSION in script
235
236 def test_uninstall_does_not_reference_musehub_url(self) -> None:
237 """Uninstall script is URL-independent β€” no MUSEHUB_URL substitution needed."""
238 script = generate_uninstall_script()
239 assert "MUSEHUB_URL" not in script
240
241 def test_build_sdist_url_with_current_version(self) -> None:
242 url = build_sdist_url("https://staging.musehub.ai", MUSE_VERSION)
243 assert MUSE_VERSION in url
244 assert url.endswith(".tar.gz")
245 assert "linux" not in url
246 assert "darwin" not in url
247
248
249 # ─────────────────────────────────────────────────────────────────────────────
250 # LAYER 3 β€” E2E
251 # ─────────────────────────────────────────────────────────────────────────────
252
253
254 class TestInstallScriptE2E:
255 """E2E: /install.sh and /uninstall.sh endpoints via async test client."""
256
257 async def test_get_install_sh_returns_200(self, client: AsyncClient) -> None:
258 r = await client.get("/install.sh")
259 assert r.status_code == 200
260
261 async def test_get_install_sh_content_type_is_shell(self, client: AsyncClient) -> None:
262 r = await client.get("/install.sh")
263 ct = r.headers.get("content-type", "")
264 assert "sh" in ct or "plain" in ct
265
266 async def test_get_install_sh_starts_with_shebang(self, client: AsyncClient) -> None:
267 r = await client.get("/install.sh")
268 assert r.text.startswith("#!/usr/bin/env sh")
269
270 async def test_get_install_sh_contains_set_euf(self, client: AsyncClient) -> None:
271 r = await client.get("/install.sh")
272 assert "set -euf" in r.text
273
274 async def test_get_install_sh_contains_muse_version(self, client: AsyncClient) -> None:
275 r = await client.get("/install.sh")
276 assert MUSE_VERSION in r.text
277
278 async def test_get_install_sh_contains_tar_gz(self, client: AsyncClient) -> None:
279 r = await client.get("/install.sh")
280 assert ".tar.gz" in r.text
281
282 async def test_get_install_sh_contains_venv(self, client: AsyncClient) -> None:
283 r = await client.get("/install.sh")
284 assert "venv" in r.text
285
286 async def test_get_install_sh_no_auth_required(self, client: AsyncClient) -> None:
287 r = await client.get("/install.sh")
288 assert r.status_code == 200
289
290 async def test_get_uninstall_sh_returns_200(self, client: AsyncClient) -> None:
291 r = await client.get("/uninstall.sh")
292 assert r.status_code == 200
293
294 async def test_get_uninstall_sh_content_type_is_shell(self, client: AsyncClient) -> None:
295 r = await client.get("/uninstall.sh")
296 ct = r.headers.get("content-type", "")
297 assert "sh" in ct or "plain" in ct
298
299 async def test_get_uninstall_sh_starts_with_shebang(self, client: AsyncClient) -> None:
300 r = await client.get("/uninstall.sh")
301 assert r.text.startswith("#!/usr/bin/env sh")
302
303 async def test_get_uninstall_sh_no_auth_required(self, client: AsyncClient) -> None:
304 r = await client.get("/uninstall.sh")
305 assert r.status_code == 200
306
307 async def test_get_uninstall_sh_does_not_rm_muse_dir(self, client: AsyncClient) -> None:
308 r = await client.get("/uninstall.sh")
309 for line in r.text.splitlines():
310 stripped = line.strip()
311 if stripped.startswith("#"):
312 continue
313 if re.match(r"^rm\s", stripped) and "~/.muse" in stripped:
314 pytest.fail(f"Uninstall script removes .muse: {stripped!r}")
315
316 async def test_get_uninstall_sh_mentions_muse_dir_preserved(self, client: AsyncClient) -> None:
317 r = await client.get("/uninstall.sh")
318 text = r.text.lower()
319 assert "~/.muse" in text
320 assert "preserved" in text or "never" in text
321
322
323 # ─────────────────────────────────────────────────────────────────────────────
324 # LAYER 4 β€” STRESS
325 # ─────────────────────────────────────────────────────────────────────────────
326
327
328 class TestInstallScriptStress:
329 """Stress: repeated generation and URL validation at volume."""
330
331 def test_generate_install_script_1000_times_identical(self) -> None:
332 url = "https://hub.example.com"
333 first = generate_install_script(url)
334 for _ in range(1000):
335 assert generate_install_script(url) == first
336
337 def test_generate_uninstall_script_1000_times_identical(self) -> None:
338 first = generate_uninstall_script()
339 for _ in range(1000):
340 assert generate_uninstall_script() == first
341
342 def test_sanitize_url_10000_valid_urls(self) -> None:
343 for i in range(10_000):
344 url = f"https://hub{i}.example.com"
345 result = sanitize_musehub_url(url)
346 assert result == url
347
348 def test_build_sdist_url_1000_times(self) -> None:
349 for _ in range(1000):
350 url = build_sdist_url("https://hub.example.com", MUSE_VERSION)
351 assert url.endswith(".tar.gz")
352
353 async def test_get_install_sh_20_sequential_requests_identical(self, client: AsyncClient) -> None:
354 first = None
355 for _ in range(20):
356 r = await client.get("/install.sh")
357 assert r.status_code == 200
358 if first is None:
359 first = r.text
360 assert r.text == first
361
362
363 # ─────────────────────────────────────────────────────────────────────────────
364 # LAYER 5 β€” DATA INTEGRITY
365 # ─────────────────────────────────────────────────────────────────────────────
366
367
368 class TestInstallScriptDataIntegrity:
369 """Data Integrity: script content correctness and determinism."""
370
371 def test_install_script_is_valid_utf8(self) -> None:
372 script = generate_install_script("https://localhost:1337")
373 encoded = script.encode("utf-8")
374 assert encoded.decode("utf-8") == script
375
376 def test_uninstall_script_is_valid_utf8(self) -> None:
377 script = generate_uninstall_script()
378 encoded = script.encode("utf-8")
379 assert encoded.decode("utf-8") == script
380
381 def test_install_script_deterministic_same_url(self) -> None:
382 url = "https://hub.example.com"
383 assert generate_install_script(url) == generate_install_script(url)
384
385 def test_install_script_different_for_different_urls(self) -> None:
386 s1 = generate_install_script("https://hub1.example.com")
387 s2 = generate_install_script("https://hub2.example.com")
388 assert s1 != s2
389
390 def test_sdist_url_unique_per_version(self) -> None:
391 urls = [build_sdist_url("https://hub.example.com", v) for v in ("1.0.0", "2.0.0", "3.0.0")]
392 assert len(urls) == len(set(urls))
393
394 def test_uninstall_script_does_not_change_between_calls(self) -> None:
395 assert generate_uninstall_script() == generate_uninstall_script()
396
397 def test_install_script_uses_mktemp_for_tmp_dir(self) -> None:
398 script = generate_install_script("http://localhost")
399 assert "mktemp" in script
400
401 def test_install_script_cleans_up_tmp_on_exit(self) -> None:
402 script = generate_install_script("http://localhost")
403 assert "trap" in script or "cleanup" in script
404
405 def test_install_script_creates_bin_dir_with_mkdir_p(self) -> None:
406 script = generate_install_script("http://localhost")
407 assert "mkdir -p" in script
408
409
410 # ─────────────────────────────────────────────────────────────────────────────
411 # LAYER 6 β€” SECURITY
412 # ─────────────────────────────────────────────────────────────────────────────
413
414
415 class TestInstallScriptSecurity:
416 """Security: shell injection prevention and uninstall data-preservation guarantee."""
417
418 @pytest.mark.parametrize("injection", [
419 'https://evil.com"; rm -rf / ; echo "',
420 "https://evil.com`id`",
421 "https://evil.com$(id)",
422 "https://evil.com$HOME",
423 "https://evil.com; cat /etc/passwd",
424 "https://evil.com\nrm -rf /",
425 "https://evil.com' && evil",
426 ])
427 def test_sanitize_rejects_injection_url(self, injection: str) -> None:
428 with pytest.raises(ValueError):
429 sanitize_musehub_url(injection)
430
431 @pytest.mark.parametrize("injection", [
432 'https://evil.com"; rm -rf / ; echo "',
433 "https://evil.com`id`",
434 ])
435 def test_generate_install_raises_for_injection(self, injection: str) -> None:
436 with pytest.raises(ValueError):
437 generate_install_script(injection)
438
439 def test_uninstall_script_never_removes_muse_dir(self) -> None:
440 script = generate_uninstall_script()
441 for line in script.splitlines():
442 stripped = line.strip()
443 if stripped.startswith("#"):
444 continue
445 if re.match(r"^rm\s", stripped) and "~/.muse" in stripped:
446 pytest.fail(f"Active rm targeting .muse: {stripped!r}")
447
448 def test_install_script_no_eval(self) -> None:
449 script = generate_install_script("http://localhost")
450 for line in script.splitlines():
451 if line.strip().startswith("#"):
452 continue
453 assert not re.match(r"^\s*eval\s", line), f"eval found: {line!r}"
454
455 def test_install_script_no_source_untrusted(self) -> None:
456 script = generate_install_script("http://localhost")
457 for line in script.splitlines():
458 if line.strip().startswith("#"):
459 continue
460 assert not re.search(r"\bsource\s+https?://", line)
461 assert not re.search(r"^\s*\.\s+https?://", line)
462
463 async def test_endpoints_return_200_not_500(self, client: AsyncClient) -> None:
464 for path in ("/install.sh", "/uninstall.sh"):
465 r = await client.get(path)
466 assert r.status_code == 200, f"{path} returned {r.status_code}"
467
468 async def test_endpoints_no_stack_trace(self, client: AsyncClient) -> None:
469 for path in ("/install.sh", "/uninstall.sh"):
470 r = await client.get(path)
471 assert "Traceback" not in r.text
472
473 def test_install_script_musehub_url_quoted(self) -> None:
474 url = "https://hub.example.com"
475 script = generate_install_script(url)
476 assert f'MUSEHUB_URL="{url}"' in script
477
478
479 # ─────────────────────────────────────────────────────────────────────────────
480 # LAYER 7 β€” PERFORMANCE
481 # ─────────────────────────────────────────────────────────────────────────────
482
483
484 class TestInstallScriptPerformance:
485 """Performance: script generation and endpoint latency budgets."""
486
487 def test_generate_install_script_under_5ms(self) -> None:
488 url = "https://localhost:1337"
489 t0 = time.perf_counter()
490 generate_install_script(url)
491 elapsed = time.perf_counter() - t0
492 assert elapsed < 0.005, f"generate_install_script took {elapsed*1000:.1f}ms"
493
494 def test_generate_uninstall_script_under_1ms(self) -> None:
495 t0 = time.perf_counter()
496 generate_uninstall_script()
497 elapsed = time.perf_counter() - t0
498 assert elapsed < 0.001, f"generate_uninstall_script took {elapsed*1000:.1f}ms"
499
500 def test_sanitize_url_under_1ms_per_call(self) -> None:
501 url = "https://hub.example.com"
502 t0 = time.perf_counter()
503 for _ in range(1000):
504 sanitize_musehub_url(url)
505 elapsed = time.perf_counter() - t0
506 assert elapsed < 0.1, f"1K sanitize_url calls took {elapsed*1000:.1f}ms"
507
508 async def test_get_install_sh_under_100ms(self, client: AsyncClient) -> None:
509 t0 = time.perf_counter()
510 r = await client.get("/install.sh")
511 elapsed = time.perf_counter() - t0
512 assert r.status_code == 200
513 assert elapsed < 0.1, f"GET /install.sh took {elapsed*1000:.1f}ms"
514
515 async def test_get_uninstall_sh_under_100ms(self, client: AsyncClient) -> None:
516 t0 = time.perf_counter()
517 r = await client.get("/uninstall.sh")
518 elapsed = time.perf_counter() - t0
519 assert r.status_code == 200
520 assert elapsed < 0.1, f"GET /uninstall.sh took {elapsed*1000:.1f}ms"