mirror of
https://github.com/hpd840321/starRiverProperty.git
synced 2026-06-09 16:30:29 +08:00
反编译工作区
038f846dad
- docs/elevator-api-parity: 计划/报告模板/示例 - tools/elevator_api_parity: 端点目录、fixtures、对拍 client/compare、报告生成 - scripts/run_elevator_parity: JDK8 构建 + 单测/对拍（无服务时跳过对拍用例） Made-with: Cursor Former-commit-id: 3d54a40e1a7ae0b1724261d4f18910a6f415f853
82 lines
2.2 KiB
Python
82 lines
2.2 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
import requests
|
|
|
|
from parity.client import can_reach_both, default_headers
|
|
|
|
_DIR = Path(__file__).resolve().parent
|
|
|
|
|
|
def pytest_configure(config):
|
|
config._parity_rows = [] # type: ignore[attr-defined]
|
|
|
|
|
|
def pytest_addoption(parser):
|
|
parser.addoption(
|
|
"--base-old",
|
|
default=os.environ.get("ELEVATOR_BASE_OLD", "http://127.0.0.1:18080"),
|
|
)
|
|
parser.addoption(
|
|
"--base-new",
|
|
default=os.environ.get("ELEVATOR_BASE_NEW", "http://127.0.0.1:18081"),
|
|
)
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def base_old(request):
|
|
return str(request.config.getoption("--base-old")).rstrip("/")
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def base_new(request):
|
|
return str(request.config.getoption("--base-new")).rstrip("/")
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def session_http():
|
|
s = requests.Session()
|
|
s.headers.update(default_headers())
|
|
yield s
|
|
s.close()
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def two_instances_ready(base_old, base_new, session_http, request):
|
|
ok, msg = can_reach_both(base_old, base_new, session_http)
|
|
require = os.environ.get("ELEVATOR_PARITY_REQUIRE_LIVE", "")
|
|
if not ok and not require:
|
|
pytest.skip(f"双端健康检查不通过(跳过用例): {msg}")
|
|
if not ok and require:
|
|
pytest.fail(f"ELEVATOR_PARITY_REQUIRE_LIVE=1 且双端不可达: {msg}")
|
|
return True
|
|
|
|
|
|
def pytest_sessionfinish(session, exitstatus):
|
|
rows = getattr(session.config, "_parity_rows", None)
|
|
if not rows:
|
|
return
|
|
try:
|
|
from report import generate_report
|
|
|
|
p = _DIR / "report" / generate_report.timestamped_name("parity")
|
|
p.parent.mkdir(parents=True, exist_ok=True)
|
|
generate_report.write_file(
|
|
p,
|
|
str(session.config.getoption("--base-old", default="")),
|
|
str(session.config.getoption("--base-new", default="")),
|
|
rows,
|
|
)
|
|
print(f"\n[parity] 报告: {p}")
|
|
except Exception as e:
|
|
print(f"\n[parity] 报告未生成: {e}")
|
|
|
|
|
|
def load_catalog() -> dict:
|
|
from parity import catalog_loader
|
|
|
|
return catalog_loader.load()
|