78 lines
3.5 KiB
Python
78 lines
3.5 KiB
Python
"""Client contract checks: exact signed bytes, gzip decoding, idempotency and report identity."""
|
|
|
|
import base64
|
|
import gzip
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
from urllib.parse import urlsplit
|
|
|
|
import httpx
|
|
import pytest
|
|
from scripts.test_plm_risk_report import DEFAULT_PAYLOAD, PLMReportTest
|
|
|
|
SECRET = "client-test-secret-at-least-32-characters"
|
|
URL = "http://test/api/v1/integrations/risk-assessments/report"
|
|
DIMENSIONS = ("供应商表现", "物料品类", "产品品类", "加工厂质量", "标准符合性", "节点阶段", "缺陷等级")
|
|
REPORT = (
|
|
"# 新品风险评估报告\n\n" + "\n".join(f"| {name} | 50 | 10% | default | 默认分 |" for name in DIMENSIONS)
|
|
).encode()
|
|
|
|
|
|
def test_client_sends_signed_snapshot_and_handles_compression(tmp_path):
|
|
payload = json.loads(DEFAULT_PAYLOAD.read_text(encoding="utf-8"))
|
|
methods, nonces, post_bodies = [], [], []
|
|
|
|
def handler(request):
|
|
methods.append(request.method)
|
|
timestamp, nonce = request.headers["X-Timestamp"], request.headers["X-Nonce"]
|
|
nonces.append(nonce)
|
|
prefix = timestamp.encode() + b"\n" + nonce.encode() + b"\n"
|
|
if request.method == "GET":
|
|
target = urlsplit(str(request.url))
|
|
message = b"GET\n" + target.path.encode() + b"\n" + target.query.encode() + b"\n" + prefix
|
|
assert request.content == b""
|
|
assert "Idempotency-Key" not in request.headers and "Content-Type" not in request.headers
|
|
else:
|
|
message = prefix + request.content
|
|
assert request.headers["Idempotency-Key"] == payload["requestId"]
|
|
post_bodies.append(request.content)
|
|
expected = base64.b64encode(hmac.new(SECRET.encode(), message, hashlib.sha256).digest()).decode()
|
|
assert request.headers["X-Signature"] == expected
|
|
if len(methods) == 1:
|
|
return httpx.Response(404, json={"code": 404, "message": "not found", "data": None})
|
|
return httpx.Response(
|
|
200,
|
|
headers={
|
|
"Content-Type": "text/markdown; charset=utf-8",
|
|
"Content-Encoding": "gzip",
|
|
"X-Request-Id": payload["requestId"],
|
|
"X-Assessment-Id": "2",
|
|
"X-Risk-Score": "44.00",
|
|
"X-Risk-Level": "medium",
|
|
"X-Model-Version": "database-seven-dimension-v1",
|
|
},
|
|
content=gzip.compress(REPORT),
|
|
)
|
|
|
|
with httpx.Client(transport=httpx.MockTransport(handler)) as client:
|
|
result = PLMReportTest(client, URL, SECRET, payload, tmp_path).run()
|
|
assert result["success"] and result["score"] == 44 and result["dimensionCount"] == 7
|
|
assert methods == ["GET", "POST", "POST", "GET"]
|
|
assert len(set(nonces)) == 4
|
|
assert post_bodies[0] == post_bodies[1] == (tmp_path / "request.json").read_bytes()
|
|
assert (tmp_path / "report.md").read_bytes() == REPORT
|
|
assert SECRET not in json.dumps(result) and "x-signature" not in json.dumps(result).lower()
|
|
|
|
|
|
def test_client_rejects_report_for_different_request(tmp_path):
|
|
payload = json.loads(DEFAULT_PAYLOAD.read_text(encoding="utf-8"))
|
|
|
|
def handler(request):
|
|
return httpx.Response(200, headers={"Content-Type": "text/markdown", "X-Request-Id": "wrong"}, content=REPORT)
|
|
|
|
with httpx.Client(transport=httpx.MockTransport(handler)) as client:
|
|
with pytest.raises(ValueError, match="X-Request-Id"):
|
|
PLMReportTest(client, URL, SECRET, payload, tmp_path).run()
|
|
assert not (tmp_path / "report.md").exists()
|