aaronrene / overseer-kit public
test_lac_stress.py python
59 lines 2.4 KB
Raw
sha256:a78e7e5a8740e03315f325d19edeb3aa1b306b3337d04abbaa9a9e0f3bbeb7a1 docs: MuseHub-first before ISR #74 — staging solidify NEXT Human 1 day ago
1 """Stress tests — concurrent landing SVG + api/health GETs (§LAC.12)."""
2
3 from __future__ import annotations
4
5 from concurrent.futures import ThreadPoolExecutor, as_completed
6 from pathlib import Path
7
8 import urllib.request
9
10 from tests.fixtures.app import seed_app_repo, start_test_app
11
12 KIT_ROOT = Path(__file__).resolve().parents[2]
13 LANDING_SVGS = (
14 KIT_ROOT / "docs" / "landing" / "assets" / "diagrams" / "lanes.svg",
15 KIT_ROOT / "docs" / "landing" / "assets" / "diagrams" / "regimes.svg",
16 KIT_ROOT / "docs" / "landing" / "assets" / "diagrams" / "layers.svg",
17 KIT_ROOT / "docs" / "landing" / "assets" / "diagrams" / "kit-consumer.svg",
18 )
19
20
21 def test_concurrent_landing_svgs_and_health_do_not_leak_credentials(tmp_path: Path) -> None:
22 seed_app_repo(tmp_path)
23 handle, client = start_test_app(tmp_path)
24 try:
25 def one_round(i: int) -> tuple[int, str]:
26 svg = LANDING_SVGS[i % len(LANDING_SVGS)]
27 text = svg.read_text(encoding="utf-8")
28 status, health = client.get("/api/health")
29 blob = text + "\n" + str(health)
30 return status, blob
31
32 with ThreadPoolExecutor(max_workers=8) as pool:
33 futures = [pool.submit(one_round, i) for i in range(24)]
34 for fut in as_completed(futures):
35 status, blob = fut.result()
36 assert status == 200
37 assert client.session not in blob
38 assert client.csrf not in blob
39 assert "test-session-credential" not in blob
40 assert "test-csrf-token" not in blob
41
42 # Static path also survives concurrent GETs of Path B diagrams.
43 def fetch_svg(name: str) -> int:
44 req = urllib.request.Request(
45 f"{client.base_url}/assets/diagrams/{name}",
46 headers={"Authorization": f"Bearer {client.session}"},
47 )
48 with urllib.request.urlopen(req, timeout=5) as response:
49 body = response.read()
50 assert client.session.encode() not in body
51 return response.status
52
53 names = ["lanes.svg", "regimes.svg", "layers.svg", "kit-consumer.svg"]
54 with ThreadPoolExecutor(max_workers=8) as pool:
55 futs = [pool.submit(fetch_svg, names[i % 4]) for i in range(20)]
56 for fut in as_completed(futs):
57 assert fut.result() == 200
58 finally:
59 handle.shutdown()
File History 1 commit
sha256:a78e7e5a8740e03315f325d19edeb3aa1b306b3337d04abbaa9a9e0f3bbeb7a1 docs: MuseHub-first before ISR #74 — staging solidify NEXT Human 1 day ago