from __future__ import annotations import json import os import socket import statistics import subprocess import sys import time import unittest from datetime import datetime, timezone from urllib.request import Request, urlopen RESULTS_PATH = os.path.join(os.path.dirname(__file__), "benchmark-results.json") def _git_sha(): try: return subprocess.run( ["git", "log", "-1", "--format=%H"], capture_output=True, text=True, timeout=5, ).stdout.strip() except Exception: return "" def _save_results(project, queries): existing = {"results": []} if os.path.isfile(RESULTS_PATH): try: with open(RESULTS_PATH) as f: existing = json.load(f) except (json.JSONDecodeError, OSError): pass existing.setdefault("results", []).append({ "timestamp": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), "project": project, "git_sha": _git_sha(), "queries": queries, }) existing["results"] = existing["results"][-50:] with open(RESULTS_PATH, "w") as f: json.dump(existing, f, indent=2, ensure_ascii=False) def _print_and_collect(results_list, label, iterations, times): median = statistics.median(times) mn = min(times) mx = max(times) print(f" {label:<46s} median={median:>8.3f}ms min={mn:>8.3f}ms max={mx:>8.3f}ms (n={iterations})") results_list.append({ "label": label.strip(), "iterations": iterations, "median_ms": round(median, 3), "min_ms": round(mn, 3), "max_ms": round(mx, 3), }) def free_port(): with socket.socket() as sock: sock.bind(("127.0.0.1", 0)) return sock.getsockname()[1] # ---- shared server for API & browser tests ---- _server_proc = None _server_port = None def start_server(): global _server_proc, _server_port if _server_proc is not None: return _server_port = free_port() _server_proc = subprocess.Popen( [sys.executable, "-m", "uvicorn", "app:app", "--host", "127.0.0.1", "--port", str(_server_port), "--log-level", "error"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, ) for _ in range(120): try: with urlopen(f"http://127.0.0.1:{_server_port}/api/sources", timeout=2) as r: if r.status == 200: return except Exception: time.sleep(0.5) stop_server() raise RuntimeError("CCRD server did not start") def stop_server(): global _server_proc, _server_port if _server_proc: try: _server_proc.terminate() _server_proc.wait(timeout=10) except Exception: _server_proc.kill() _server_proc = None _server_port = None # ---- local benchmarks (pure function) ---- import app # noqa: E402 class LocalSearchBenchmarkTests(unittest.TestCase): _results: list = [] @classmethod def setUpClass(cls): if not app.source_counts: app.load_data() def _bench(self, label, fn, iterations=10): times = [] for _ in range(iterations): start = time.perf_counter() fn() elapsed = (time.perf_counter() - start) * 1000 times.append(elapsed) _print_and_collect(self._results, label, iterations, times) def test_benchmarks(self): self._bench('"手机" normal', lambda: app.search("手机")) self._bench('"手机" exact', lambda: app.search("手机", exact=True)) self._bench('"文化革命" exact', lambda: app.search("文化革命", exact=True)) self._bench('"文化 革命" normal', lambda: app.search("文化 革命")) self._bench('"手。机" exact', lambda: app.search("手。机", exact=True)) self._bench('"手*机" exact', lambda: app.search("手*机", exact=True)) self._bench('"手?机" exact', lambda: app.search("手?机", exact=True)) self._bench('"文" exact', lambda: app.search("文", exact=True)) self._bench('"ABC" normal', lambda: app.search("ABC")) self._bench('"文化革命" exact search_paths=False', lambda: app.search("文化革命", exact=True, search_paths=False)) self._bench('"手机" exact min_size=100', lambda: app.search("手机", exact=True, min_size=100)) self._bench('sort by name', lambda: app.search(page=1, page_size=100, sort="name")) self._bench('sort by size', lambda: app.search(page=1, page_size=100, sort="size")) @classmethod def tearDownClass(cls): if cls._results: _save_results("CCRD-Search-local", cls._results) # ---- API benchmarks (HTTP to local server) ---- class ApiSearchBenchmarkTests(unittest.TestCase): _results: list = [] def _bench(self, label, path, payload=None, iterations=5): body = json.dumps(payload).encode() if payload else None req = Request(f"http://127.0.0.1:{_server_port}{path}", data=body, headers={"Content-Type": "application/json"} if body else {}) times = [] for i in range(iterations + 1): start = time.perf_counter() with urlopen(req, timeout=60) as r: r.read() elapsed = (time.perf_counter() - start) * 1000 if i > 0: times.append(elapsed) _print_and_collect(self._results, label, iterations, times) def test_search_api_benchmarks(self): self._bench('"手机" normal', "/api/search", {"q": "手机"}) self._bench('"手机" exact', "/api/search", {"q": "手机", "exact": True}) self._bench('"文化革命" exact', "/api/search", {"q": "文化革命", "exact": True}) self._bench('"文化 革命" normal', "/api/search", {"q": "文化 革命"}) self._bench('"手。机" exact', "/api/search", {"q": "手。机", "exact": True}) self._bench('"手*机" exact', "/api/search", {"q": "手*机", "exact": True}) self._bench('"手?机" exact', "/api/search", {"q": "手?机", "exact": True}) self._bench('"文" exact', "/api/search", {"q": "文", "exact": True}) self._bench('"ABC" normal', "/api/search", {"q": "ABC"}) self._bench('sources', "/api/sources") @classmethod def tearDownClass(cls): if cls._results: _save_results("CCRD-Search-api", cls._results) # ---- browser benchmarks (Playwright) ---- try: from playwright.sync_api import sync_playwright except ImportError: sync_playwright = None @unittest.skipUnless(sync_playwright, "install requirements-test.txt, run: playwright install chromium") class BrowserSearchBenchmarkTests(unittest.TestCase): _results: list = [] @classmethod def setUpClass(cls): cls.playwright = sync_playwright().start() cls.browser = cls.playwright.chromium.launch(headless=True, args=["--no-sandbox", "--disable-gpu", "--disable-dev-shm-usage"]) @classmethod def tearDownClass(cls): if cls._results: _save_results("CCRD-Search-browser", cls._results) cls.browser.close() cls.playwright.stop() def _bench_search(self, label, query, iterations=3): ctx = self.browser.new_context(viewport={"width": 1440, "height": 900}) page = ctx.new_page() times = [] for i in range(iterations + 1): page.goto(f"http://127.0.0.1:{_server_port}/") page.wait_for_selector("#results-list .result-item", timeout=30000) before_hash = page.evaluate("""() => { const items = document.querySelectorAll('.result-item'); return Array.from(items).slice(0, 3).map(el => el.textContent.substring(0, 50)).join('|'); }""") inp = page.locator("#search-input") start = time.perf_counter() inp.fill(query) page.wait_for_function(f""" () => {{ const items = document.querySelectorAll('.result-item'); if (items.length > 0) {{ const hash = Array.from(items).slice(0, 3).map(el => el.textContent.substring(0, 50)).join('|'); if (hash !== {json.dumps(before_hash)}) return true; }} const empty = document.querySelector('#empty-state'); return !!(empty && empty.style.display !== 'none'); }} """, timeout=15000) elapsed = (time.perf_counter() - start) * 1000 if i > 0: times.append(elapsed) ctx.close() _print_and_collect(self._results, label, iterations, times) def test_browser_search_benchmarks(self): self._bench_search('"手机" normal', "手机") self._bench_search('"文化革命" exact', "文化革命") self._bench_search('"手*机" exact', "手*机") def setUpModule(): start_server() def tearDownModule(): stop_server()