release: cut v2.1
This commit is contained in:
Hermes Agent
@@ -5,6 +5,7 @@
|
||||
|
||||
import sys
|
||||
import os
|
||||
from types import ModuleType
|
||||
|
||||
# 添加scripts到路径
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'scripts'))
|
||||
@@ -15,8 +16,10 @@ spec = importlib.util.spec_from_file_location(
|
||||
"spec_checker_v2",
|
||||
os.path.join(os.path.dirname(__file__), '..', 'skills', 'gaokao-spec-checker', 'scripts', 'spec_checker_v2.py')
|
||||
)
|
||||
assert spec is not None and spec.loader is not None
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(module)
|
||||
assert isinstance(module, ModuleType)
|
||||
|
||||
GaokaoSpecCheckerV2 = module.GaokaoSpecCheckerV2
|
||||
|
||||
|
||||
96
tests/test_audit_integration.py
Normal file
96
tests/test_audit_integration.py
Normal file
@@ -0,0 +1,96 @@
|
||||
"""gaokao-audit 端到端集成测试。"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib
|
||||
import json
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import Any, cast
|
||||
|
||||
import pytest
|
||||
|
||||
_REPO_ROOT = Path(__file__).resolve().parents[1]
|
||||
if str(_REPO_ROOT) not in sys.path:
|
||||
sys.path.insert(0, str(_REPO_ROOT))
|
||||
|
||||
_AUDIT_CLI = importlib.import_module("skills.gaokao-audit.scripts.audit_cli")
|
||||
_REPORT_GENERATOR = importlib.import_module(
|
||||
"skills.gaokao-audit.scripts.report_generator"
|
||||
)
|
||||
_ReportGeneratorBase = cast(type[Any], _REPORT_GENERATOR.ReportGenerator)
|
||||
|
||||
SAMPLE_PLAN = (
|
||||
_REPO_ROOT / "skills" / "gaokao-audit" / "tests" / "fixtures" / "sample_xianyu.txt"
|
||||
)
|
||||
|
||||
|
||||
class _CaptureReportGenerator(_ReportGeneratorBase): # type: ignore[valid-type, misc]
|
||||
last_html: str = ""
|
||||
|
||||
def __init__(self, **kwargs: object) -> None:
|
||||
super().__init__(
|
||||
now_text=lambda: "2026-06-12 23:40",
|
||||
report_id_factory=lambda: "AUDIT-E2E-001",
|
||||
**kwargs,
|
||||
)
|
||||
|
||||
def generate_pdf(self, result, output_path: str, **kwargs: object) -> str:
|
||||
html = self.render_html(result, **kwargs)
|
||||
type(self).last_html = html
|
||||
target = Path(output_path)
|
||||
target.parent.mkdir(parents=True, exist_ok=True)
|
||||
target.write_bytes(b"%PDF-1.4\ne2e fake pdf\n")
|
||||
return str(target)
|
||||
|
||||
|
||||
def _extract_json(stdout: str) -> dict:
|
||||
lines = stdout.splitlines()
|
||||
for index, line in enumerate(lines):
|
||||
if line.startswith("{"):
|
||||
return json.loads("\n".join(lines[index:]))
|
||||
raise AssertionError(f"stdout 中未找到 JSON 输出: {stdout}")
|
||||
|
||||
|
||||
def test_audit_cli_end_to_end_generates_pdf_and_report_content(
|
||||
tmp_path: Path,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
capsys: pytest.CaptureFixture[str],
|
||||
) -> None:
|
||||
output_path = tmp_path / "audit-e2e.pdf"
|
||||
monkeypatch.setattr(_AUDIT_CLI, "ReportGenerator", _CaptureReportGenerator)
|
||||
|
||||
exit_code = _AUDIT_CLI.main([
|
||||
str(SAMPLE_PLAN),
|
||||
"--output",
|
||||
str(output_path),
|
||||
"--json",
|
||||
])
|
||||
captured = capsys.readouterr()
|
||||
payload = _extract_json(captured.out)
|
||||
rendered_html = _CaptureReportGenerator.last_html
|
||||
|
||||
assert exit_code == 0
|
||||
assert output_path.exists()
|
||||
assert output_path.read_bytes().startswith(b"%PDF-1.4")
|
||||
|
||||
assert "输入文件" in captured.out
|
||||
assert str(output_path) in captured.out
|
||||
assert "综合评分" in captured.out
|
||||
|
||||
assert payload["province"] == "湖南"
|
||||
assert payload["candidate_score"] == 578
|
||||
assert payload["source"] == "百度AI"
|
||||
assert len(payload["volunteers"]) == 6
|
||||
assert payload["policy_errors"] == []
|
||||
assert payload["policy_serious_warnings"]
|
||||
assert payload["crowd_risks"]
|
||||
risk_schools = {item["school"] for item in payload["crowd_risks"]}
|
||||
assert "湖南师范大学" in risk_schools
|
||||
assert payload["overall_score"] < 100
|
||||
|
||||
assert "AUDIT-E2E-001" in rendered_html
|
||||
assert "湖南 578分 物理+化学+生物" in rendered_html
|
||||
assert any(school in rendered_html for school in risk_schools)
|
||||
assert "免责声明" in rendered_html
|
||||
assert "本审核仅供建议，最终填报以官方公布为准" in rendered_html
|
||||
97
tests/test_sync_remotes_cli.py
Normal file
97
tests/test_sync_remotes_cli.py
Normal file
@@ -0,0 +1,97 @@
|
||||
"""gaokao-sync-remotes CLI tests (T10.3)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
PROJECT_ROOT = Path(__file__).resolve().parents[1]
|
||||
SCRIPT_PATH = PROJECT_ROOT / "scripts" / "gaokao-sync-remotes"
|
||||
DEFAULT_REMOTES = ("gitea", "origin", "tksea")
|
||||
|
||||
|
||||
def _git(repo: Path, *args: str) -> subprocess.CompletedProcess[str]:
|
||||
return subprocess.run(
|
||||
["git", *args],
|
||||
cwd=repo,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
)
|
||||
|
||||
|
||||
def _run_cli(repo: Path, *args: str) -> subprocess.CompletedProcess[str]:
|
||||
return subprocess.run(
|
||||
[sys.executable, str(SCRIPT_PATH), *args],
|
||||
cwd=repo,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
)
|
||||
|
||||
|
||||
def _init_repo(tmp_path: Path) -> tuple[Path, dict[str, Path]]:
|
||||
repo = tmp_path / "repo"
|
||||
repo.mkdir()
|
||||
assert _git(repo, "init", "-b", "main").returncode == 0
|
||||
assert _git(repo, "config", "user.name", "T10 Tester").returncode == 0
|
||||
assert _git(repo, "config", "user.email", "t10@example.com").returncode == 0
|
||||
|
||||
readme = repo / "README.md"
|
||||
readme.write_text("hello\n", encoding="utf-8")
|
||||
assert _git(repo, "add", "README.md").returncode == 0
|
||||
assert _git(repo, "commit", "-m", "init").returncode == 0
|
||||
|
||||
remotes: dict[str, Path] = {}
|
||||
for name in DEFAULT_REMOTES:
|
||||
bare = tmp_path / f"{name}.git"
|
||||
assert _git(tmp_path, "init", "--bare", str(bare)).returncode == 0
|
||||
assert _git(repo, "remote", "add", name, str(bare)).returncode == 0
|
||||
remotes[name] = bare
|
||||
return repo, remotes
|
||||
|
||||
|
||||
def _remote_head(remote_path: Path, branch: str = "main") -> str:
|
||||
result = subprocess.run(
|
||||
["git", "--git-dir", str(remote_path), "rev-parse", branch],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
)
|
||||
assert result.returncode == 0, result.stderr
|
||||
return result.stdout.strip()
|
||||
|
||||
|
||||
def test_dry_run_lists_all_three_push_commands(tmp_path: Path) -> None:
|
||||
repo, remotes = _init_repo(tmp_path)
|
||||
|
||||
result = _run_cli(repo, "--dry-run")
|
||||
|
||||
assert result.returncode == 0, result.stderr
|
||||
for remote_name in remotes:
|
||||
assert f"[DRY-RUN] git push {remote_name} main" in result.stdout
|
||||
for remote_path in remotes.values():
|
||||
assert not (remote_path / "refs" / "heads" / "main").exists()
|
||||
|
||||
|
||||
def test_pushes_main_to_all_three_remotes_and_verifies_heads(tmp_path: Path) -> None:
|
||||
repo, remotes = _init_repo(tmp_path)
|
||||
|
||||
result = _run_cli(repo)
|
||||
|
||||
assert result.returncode == 0, result.stderr
|
||||
local_head = _git(repo, "rev-parse", "HEAD").stdout.strip()
|
||||
for remote_name, remote_path in remotes.items():
|
||||
assert f"OK {remote_name}: main @ {local_head}" in result.stdout
|
||||
assert _remote_head(remote_path) == local_head
|
||||
|
||||
|
||||
def test_missing_remote_fails_before_push(tmp_path: Path) -> None:
|
||||
repo, _ = _init_repo(tmp_path)
|
||||
assert _git(repo, "remote", "remove", "tksea").returncode == 0
|
||||
|
||||
result = _run_cli(repo)
|
||||
|
||||
assert result.returncode == 2
|
||||
assert "missing remotes: tksea" in result.stderr
|
||||
327
tests/test_t5_e2e_workflows.py
Normal file
327
tests/test_t5_e2e_workflows.py
Normal file
@@ -0,0 +1,327 @@
|
||||
"""T5.1 端到端业务场景测试。
|
||||
|
||||
覆盖 5 条主链路：
|
||||
1. 咨询 -> 方案生成
|
||||
2. 审核 -> 报告
|
||||
3. 订单 -> 交付
|
||||
4. 升级流程
|
||||
5. 数据溯源展示
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib
|
||||
import importlib.util
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import Any, cast
|
||||
|
||||
import pytest
|
||||
|
||||
PROJECT_ROOT = Path(__file__).resolve().parents[1]
|
||||
if str(PROJECT_ROOT) not in sys.path:
|
||||
sys.path.insert(0, str(PROJECT_ROOT))
|
||||
|
||||
_AUDIT_CLI = importlib.import_module("skills.gaokao-audit.scripts.audit_cli")
|
||||
_REPORT_GENERATOR = importlib.import_module(
|
||||
"skills.gaokao-audit.scripts.report_generator"
|
||||
)
|
||||
_ReportGeneratorBase = cast(type[Any], _REPORT_GENERATOR.ReportGenerator)
|
||||
_TRACE_CLI = importlib.import_module("data.crowd_db.cli")
|
||||
|
||||
SAMPLE_PLAN = (
|
||||
PROJECT_ROOT
|
||||
/ "skills"
|
||||
/ "gaokao-audit"
|
||||
/ "tests"
|
||||
/ "fixtures"
|
||||
/ "sample_xianyu.txt"
|
||||
)
|
||||
QUICK_SCRIPT = PROJECT_ROOT / "scripts" / "gaokao-quick-3min.py"
|
||||
ORDER_SCRIPT = PROJECT_ROOT / "scripts" / "gaokao-order-manager"
|
||||
TRACE_SCRIPT = PROJECT_ROOT / "scripts" / "gaokao-data-trace"
|
||||
|
||||
os.environ.setdefault("GAOKAO_ORDERS_FERNET_KEY", "test-secret-for-t5-e2e")
|
||||
|
||||
|
||||
class _CaptureReportGenerator(_ReportGeneratorBase): # type: ignore[valid-type, misc]
|
||||
last_html: str = ""
|
||||
|
||||
def __init__(self, **kwargs: object) -> None:
|
||||
super().__init__(
|
||||
now_text=lambda: "2026-06-13 10:00",
|
||||
report_id_factory=lambda: "AUDIT-T5-E2E-001",
|
||||
**kwargs,
|
||||
)
|
||||
|
||||
def generate_pdf(self, result, output_path: str, **kwargs: object) -> str:
|
||||
html = self.render_html(result, **kwargs)
|
||||
type(self).last_html = html
|
||||
target = Path(output_path)
|
||||
target.parent.mkdir(parents=True, exist_ok=True)
|
||||
target.write_bytes(b"%PDF-1.4\nt5 fake pdf\n")
|
||||
return str(target)
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def quick_module():
|
||||
spec = importlib.util.spec_from_file_location("gaokao_quick_3min", QUICK_SCRIPT)
|
||||
assert spec is not None and spec.loader is not None
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(module)
|
||||
return module
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def tmp_orders_db(tmp_path: Path) -> Path:
|
||||
return tmp_path / "orders.db"
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def order_env() -> dict[str, str]:
|
||||
return {"GAOKAO_ORDERS_FERNET_KEY": os.environ["GAOKAO_ORDERS_FERNET_KEY"]}
|
||||
|
||||
|
||||
def _run_script(
|
||||
script: Path, *args: str, env: dict[str, str] | None = None
|
||||
) -> subprocess.CompletedProcess[str]:
|
||||
merged_env = os.environ.copy()
|
||||
if env:
|
||||
merged_env.update(env)
|
||||
return subprocess.run(
|
||||
[sys.executable, str(script), *args],
|
||||
cwd=PROJECT_ROOT,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
env=merged_env,
|
||||
)
|
||||
|
||||
|
||||
def _extract_json(stdout: str) -> dict:
|
||||
lines = stdout.splitlines()
|
||||
for index, line in enumerate(lines):
|
||||
if line.startswith("{"):
|
||||
return json.loads("\n".join(lines[index:]))
|
||||
raise AssertionError(f"stdout 中未找到 JSON 输出: {stdout}")
|
||||
|
||||
|
||||
def test_consultation_to_plan_generation_flow(quick_module) -> None:
|
||||
reply = """1. 李明
|
||||
2. 浙江
|
||||
3. 612
|
||||
4. 15230
|
||||
5. R
|
||||
6. 物理、数学
|
||||
7. C
|
||||
8. ③
|
||||
9. ①
|
||||
10. ②
|
||||
"""
|
||||
|
||||
info = quick_module.parse_quick_response(reply)
|
||||
summary = quick_module.generate_quick_summary(info)
|
||||
recommendation = quick_module.generate_quick_recommendation(info)
|
||||
|
||||
assert info["basic"]["name"] == "李明"
|
||||
assert info["basic"]["province"] == "浙江"
|
||||
assert info["exam"]["score"] == 612
|
||||
assert info["exam"]["rank"] == 15230
|
||||
assert info["profile"]["type_code"] == "R"
|
||||
assert "✅ 核心信息完整！可以开始推荐" in summary
|
||||
assert "📊 高考：612分" in summary
|
||||
assert "📊 位次：15230名" in summary
|
||||
assert "计算机科学与技术" in recommendation
|
||||
assert "物理数学强 → 计算机、电子信息、自动化" in recommendation
|
||||
|
||||
|
||||
def test_audit_to_report_flow(
|
||||
tmp_path: Path,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
capsys: pytest.CaptureFixture[str],
|
||||
) -> None:
|
||||
output_path = tmp_path / "audit-report.pdf"
|
||||
monkeypatch.setattr(_AUDIT_CLI, "ReportGenerator", _CaptureReportGenerator)
|
||||
|
||||
exit_code = _AUDIT_CLI.main([
|
||||
str(SAMPLE_PLAN),
|
||||
"--output",
|
||||
str(output_path),
|
||||
"--json",
|
||||
])
|
||||
captured = capsys.readouterr()
|
||||
payload = _extract_json(captured.out)
|
||||
|
||||
assert exit_code == 0
|
||||
assert output_path.exists()
|
||||
assert output_path.read_bytes().startswith(b"%PDF-1.4")
|
||||
assert payload["province"] == "湖南"
|
||||
assert payload["candidate_score"] == 578
|
||||
assert payload["crowd_risks"]
|
||||
assert "AUDIT-T5-E2E-001" in _CaptureReportGenerator.last_html
|
||||
assert "免责声明" in _CaptureReportGenerator.last_html
|
||||
|
||||
|
||||
def test_order_to_delivery_flow_records_artifacts(
|
||||
tmp_orders_db: Path,
|
||||
tmp_path: Path,
|
||||
order_env: dict[str, str],
|
||||
) -> None:
|
||||
plan_path = tmp_path / "plan.md"
|
||||
report_path = tmp_path / "audit.json"
|
||||
pdf_path = tmp_path / "report.pdf"
|
||||
plan_path.write_text("咨询后生成的志愿方案", encoding="utf-8")
|
||||
report_path.write_text('{"overall_score": 88}', encoding="utf-8")
|
||||
pdf_path.write_bytes(b"%PDF-1.4\nreport\n")
|
||||
|
||||
created = _run_script(
|
||||
ORDER_SCRIPT,
|
||||
"--db",
|
||||
str(tmp_orders_db),
|
||||
"create",
|
||||
"--source",
|
||||
"xianyu",
|
||||
"--service-version",
|
||||
"audit",
|
||||
"--amount-cents",
|
||||
"4900",
|
||||
"--customer-name",
|
||||
"王家长",
|
||||
"--customer-phone",
|
||||
"13800001234",
|
||||
"--candidate-name",
|
||||
"李明",
|
||||
"--candidate-province",
|
||||
"湖南",
|
||||
"--candidate-score",
|
||||
"578",
|
||||
"--candidate-rank",
|
||||
"26800",
|
||||
env=order_env,
|
||||
)
|
||||
assert created.returncode == 0, created.stderr
|
||||
order_id = json.loads(created.stdout)["order"]["id"]
|
||||
|
||||
updated = _run_script(
|
||||
ORDER_SCRIPT,
|
||||
"--db",
|
||||
str(tmp_orders_db),
|
||||
"update",
|
||||
order_id,
|
||||
"--assigned-consultant",
|
||||
"long-teacher",
|
||||
"--plan-file",
|
||||
str(plan_path),
|
||||
"--audit-report",
|
||||
str(report_path),
|
||||
"--pdf-path",
|
||||
str(pdf_path),
|
||||
"--note",
|
||||
"方案与审核报告已归档",
|
||||
env=order_env,
|
||||
)
|
||||
assert updated.returncode == 0, updated.stderr
|
||||
updated_payload = json.loads(updated.stdout)
|
||||
assert updated_payload["order"]["plan_file"] == str(plan_path)
|
||||
assert updated_payload["order"]["audit_report"] == str(report_path)
|
||||
assert updated_payload["order"]["pdf_path"] == str(pdf_path)
|
||||
|
||||
paid = _run_script(
|
||||
ORDER_SCRIPT,
|
||||
"--db",
|
||||
str(tmp_orders_db),
|
||||
"pay",
|
||||
order_id,
|
||||
"--reason",
|
||||
"xianyu-paid",
|
||||
env=order_env,
|
||||
)
|
||||
assert paid.returncode == 0, paid.stderr
|
||||
|
||||
delivered = _run_script(
|
||||
ORDER_SCRIPT,
|
||||
"--db",
|
||||
str(tmp_orders_db),
|
||||
"deliver",
|
||||
order_id,
|
||||
"--reason",
|
||||
"pdf-delivered",
|
||||
env=order_env,
|
||||
)
|
||||
assert delivered.returncode == 0, delivered.stderr
|
||||
delivered_payload = json.loads(delivered.stdout)
|
||||
assert delivered_payload["order"]["status"] == "delivered"
|
||||
assert delivered_payload["order"]["delivered_at"] is not None
|
||||
assert delivered_payload["order"]["plan_file"] == str(plan_path)
|
||||
assert delivered_payload["order"]["pdf_path"] == str(pdf_path)
|
||||
|
||||
|
||||
def test_upgrade_flow_creates_delta_order(
|
||||
tmp_orders_db: Path,
|
||||
order_env: dict[str, str],
|
||||
) -> None:
|
||||
created = _run_script(
|
||||
ORDER_SCRIPT,
|
||||
"--db",
|
||||
str(tmp_orders_db),
|
||||
"create",
|
||||
"--source",
|
||||
"wechat",
|
||||
"--service-version",
|
||||
"audit",
|
||||
"--amount-cents",
|
||||
"4900",
|
||||
"--customer-name",
|
||||
"王家长",
|
||||
"--customer-phone",
|
||||
"13900001234",
|
||||
env=order_env,
|
||||
)
|
||||
assert created.returncode == 0, created.stderr
|
||||
source_order_id = json.loads(created.stdout)["order"]["id"]
|
||||
|
||||
upgraded = _run_script(
|
||||
ORDER_SCRIPT,
|
||||
"--db",
|
||||
str(tmp_orders_db),
|
||||
"upgrade",
|
||||
source_order_id,
|
||||
"--service-version",
|
||||
"standard",
|
||||
"--target-amount-cents",
|
||||
"9900",
|
||||
"--reason",
|
||||
"upgrade_to_standard",
|
||||
env=order_env,
|
||||
)
|
||||
assert upgraded.returncode == 0, upgraded.stderr
|
||||
payload = json.loads(upgraded.stdout)
|
||||
|
||||
assert payload["order"]["upgrade_from"] == source_order_id
|
||||
assert payload["order"]["service_version"] == "standard"
|
||||
assert payload["order"]["amount_cents"] == 5000
|
||||
assert payload["source_order"]["id"] == source_order_id
|
||||
assert "upgraded" in payload["source_order"]["tags"]
|
||||
|
||||
|
||||
def test_traceability_display_flow(capsys: pytest.CaptureFixture[str]) -> None:
|
||||
exit_code = _TRACE_CLI.main(["--human", "长沙理工大学"])
|
||||
captured = capsys.readouterr()
|
||||
|
||||
assert exit_code == 0
|
||||
assert "query: 长沙理工大学" in captured.out
|
||||
assert "湖南 / 2025年数据 / 长沙理工大学 / 会计学" in captured.out
|
||||
assert "source_type: report (⚠️报告)" in captured.out
|
||||
assert "source_url: https://" in captured.out
|
||||
assert "confidence: 0.85" in captured.out
|
||||
|
||||
|
||||
def test_traceability_json_entrypoint_matches_cli_contract() -> None:
|
||||
result = _run_script(TRACE_SCRIPT, "长沙理工大学")
|
||||
assert result.returncode == 0, result.stderr
|
||||
payload = json.loads(result.stdout)
|
||||
assert payload["query"] == "长沙理工大学"
|
||||
assert payload["match_count"] >= 1
|
||||
assert any(match["province"] == "湖南" for match in payload["matches"])
|
||||
168
tests/test_t5_performance.py
Normal file
168
tests/test_t5_performance.py
Normal file
@@ -0,0 +1,168 @@
|
||||
"""T5.2 性能与并发测试。"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import csv
|
||||
import importlib.util
|
||||
import os
|
||||
import socket
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
from pathlib import Path
|
||||
from typing import Iterator
|
||||
|
||||
import pytest
|
||||
|
||||
PROJECT_ROOT = Path(__file__).resolve().parents[1]
|
||||
if str(PROJECT_ROOT) not in sys.path:
|
||||
sys.path.insert(0, str(PROJECT_ROOT))
|
||||
|
||||
QUICK_SCRIPT = PROJECT_ROOT / "scripts" / "gaokao-quick-3min.py"
|
||||
LOCUST_FILE = PROJECT_ROOT / "locustfile.py"
|
||||
|
||||
_SAMPLE_REPLY = """1. 李明
|
||||
2. 浙江
|
||||
3. 612
|
||||
4. 15230
|
||||
5. R
|
||||
6. 物理、数学
|
||||
7. C
|
||||
8. ③
|
||||
9. ①
|
||||
10. ②
|
||||
"""
|
||||
|
||||
|
||||
def _load_quick_module():
|
||||
spec = importlib.util.spec_from_file_location("gaokao_quick_3min", QUICK_SCRIPT)
|
||||
assert spec is not None and spec.loader is not None
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(module)
|
||||
return module
|
||||
|
||||
|
||||
def _generate_100_plans() -> float:
|
||||
module = _load_quick_module()
|
||||
start = time.perf_counter()
|
||||
for _ in range(100):
|
||||
info = module.parse_quick_response(_SAMPLE_REPLY)
|
||||
module.generate_quick_summary(info)
|
||||
module.generate_quick_recommendation(info)
|
||||
return time.perf_counter() - start
|
||||
|
||||
|
||||
def _find_free_port() -> int:
|
||||
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
|
||||
sock.bind(("127.0.0.1", 0))
|
||||
return int(sock.getsockname()[1])
|
||||
|
||||
|
||||
def _wait_for_health(base_url: str, timeout: float = 20.0) -> None:
|
||||
deadline = time.time() + timeout
|
||||
while time.time() < deadline:
|
||||
try:
|
||||
with urllib.request.urlopen(f"{base_url}/health", timeout=1.0) as resp:
|
||||
if resp.status == 200:
|
||||
return
|
||||
except (urllib.error.URLError, TimeoutError):
|
||||
time.sleep(0.2)
|
||||
raise AssertionError(f"admin app 未在 {timeout}s 内就绪: {base_url}")
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def admin_server(tmp_path: Path) -> Iterator[str]:
|
||||
port = _find_free_port()
|
||||
env = os.environ.copy()
|
||||
env.update(
|
||||
{
|
||||
"GAOKAO_ENV": "dev",
|
||||
"GAOKAO_DB_PATH": str(tmp_path / "admin.db"),
|
||||
"GAOKAO_ORDERS_DB_PATH": str(tmp_path / "orders.db"),
|
||||
"GAOKAO_JWT_SECRET": "x" * 64,
|
||||
"GAOKAO_ADMIN_USER": "admin",
|
||||
"GAOKAO_ADMIN_PASS": "admin123",
|
||||
}
|
||||
)
|
||||
process = subprocess.Popen(
|
||||
[
|
||||
sys.executable,
|
||||
"-m",
|
||||
"admin.app",
|
||||
"--host",
|
||||
"127.0.0.1",
|
||||
"--port",
|
||||
str(port),
|
||||
"--log-format",
|
||||
"plain",
|
||||
],
|
||||
cwd=PROJECT_ROOT,
|
||||
env=env,
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.STDOUT,
|
||||
text=True,
|
||||
)
|
||||
try:
|
||||
base_url = f"http://127.0.0.1:{port}"
|
||||
_wait_for_health(base_url)
|
||||
yield base_url
|
||||
finally:
|
||||
process.terminate()
|
||||
try:
|
||||
process.wait(timeout=10)
|
||||
except subprocess.TimeoutExpired:
|
||||
process.kill()
|
||||
process.wait(timeout=10)
|
||||
|
||||
|
||||
@pytest.mark.timeout(30)
|
||||
def test_plan_generation_100_runs_under_5_seconds(benchmark) -> None:
|
||||
elapsed = benchmark(_generate_100_plans)
|
||||
assert elapsed < 5.0
|
||||
|
||||
|
||||
@pytest.mark.timeout(90)
|
||||
def test_admin_locust_10_concurrency_success_rate_above_95(
|
||||
admin_server: str, tmp_path: Path
|
||||
) -> None:
|
||||
report_prefix = tmp_path / "t5_2"
|
||||
command = [
|
||||
"locust",
|
||||
"-f",
|
||||
str(LOCUST_FILE),
|
||||
"--host",
|
||||
admin_server,
|
||||
"--headless",
|
||||
"-u",
|
||||
"10",
|
||||
"-r",
|
||||
"2",
|
||||
"-t",
|
||||
"15s",
|
||||
"--csv",
|
||||
str(report_prefix),
|
||||
]
|
||||
result = subprocess.run(
|
||||
command,
|
||||
cwd=PROJECT_ROOT,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
)
|
||||
assert result.returncode == 0, result.stdout + "\n" + result.stderr
|
||||
|
||||
stats_path = report_prefix.with_name(report_prefix.name + "_stats.csv")
|
||||
with stats_path.open(newline="", encoding="utf-8") as handle:
|
||||
rows = list(csv.DictReader(handle))
|
||||
|
||||
aggregate = next(
|
||||
row for row in rows if row.get("Name") == "Aggregated" and not row.get("Type")
|
||||
)
|
||||
request_count = int(aggregate["Request Count"])
|
||||
failure_count = int(aggregate["Failure Count"])
|
||||
success_rate = ((request_count - failure_count) / request_count) * 100
|
||||
|
||||
assert request_count > 0
|
||||
assert success_rate > 95.0
|
||||
Reference in New Issue
Block a user