aaronrene / overseer-kit public
test_q1_app_stress.py python
42 lines 1.4 KB
Raw
sha256:a78e7e5a8740e03315f325d19edeb3aa1b306b3337d04abbaa9a9e0f3bbeb7a1 docs: MuseHub-first before ISR #74 — staging solidify NEXT Human 1 day ago
1 """Stress tests for Track Q / Q1 overseer app (§Q0.12)."""
2
3 from __future__ import annotations
4
5 from concurrent.futures import ThreadPoolExecutor, as_completed
6 from pathlib import Path
7
8 from tests.fixtures.app import seed_app_repo, start_test_app
9
10
11 def test_concurrent_status_reads(tmp_path: Path) -> None:
12 seed_app_repo(tmp_path)
13 large = "# Handover\n" + ("line\n" * 5000)
14 (tmp_path / "docs" / "OVERSEER-HANDOVER.md").write_text(large, encoding="utf-8")
15
16 handle, client = start_test_app(tmp_path)
17 try:
18
19 def read_status() -> int:
20 status, payload = client.get("/api/status")
21 assert payload["result"]["initialized"] is True
22 return status
23
24 with ThreadPoolExecutor(max_workers=20) as pool:
25 futures = [pool.submit(read_status) for _ in range(24)]
26 codes = [future.result() for future in as_completed(futures)]
27 assert all(code == 200 for code in codes)
28 finally:
29 handle.shutdown()
30
31
32 def test_large_doc_read_bounded(tmp_path: Path) -> None:
33 seed_app_repo(tmp_path)
34 text = "x" * 200_000
35 (tmp_path / "docs" / "ROADMAP.md").write_text(text, encoding="utf-8")
36 handle, client = start_test_app(tmp_path)
37 try:
38 status, payload = client.get("/api/docs/roadmap")
39 assert status == 200
40 assert len(payload["result"]["text"]) == 200_000
41 finally:
42 handle.shutdown()
File History 1 commit
sha256:a78e7e5a8740e03315f325d19edeb3aa1b306b3337d04abbaa9a9e0f3bbeb7a1 docs: MuseHub-first before ISR #74 — staging solidify NEXT Human 1 day ago