q792602257andClaude Opus 5
8403b9b586
上一提交（8381896）的配套测试，钉住两个失败可见性问题： - 抓取侧回归：失败分支原先写的是无参 `span.record_exception()`，而 exception 是必填位置参数——每次抓取失败都会在记异常时抛 TypeError，把真正的原因（反爬 阻断 / 解析失败）整个顶替掉，失败页面快照也永远落不下来（抛错在 snapshot 之前）。测试断言业务异常原样抛出、stage 能区分 fetch/parse/delegate、 parse 阶段确实落了快照且带得上 URL - 网关侧：success=false code=6002 在 HTTP 层是 200，断言 span 记下了信封结论； 以及空转长轮询在 suppressed() 里确实不产 span，抑制之外照常产（防止空断言 是因为埋点根本没生效） - 两个文件都各带一条成功路径对照，证明 ERROR 状态不是恒真 Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
159 lines
6.3 KiB
Python
159 lines
6.3 KiB
Python
"""GatewayClient 埋点测试：信封里的失败必须在链路上看得见
|
|
|
|
要钉住的问题：**网关的失败在响应信封里，不在 HTTP 状态码上**。一次
|
|
`success=false, code=6002`（租约无效）的回报，httpx 自动 instrumentation 只看到
|
|
一个完成了的请求——而且那个 CLIENT span 在 `request()` 返回时就结束了，此时信封
|
|
还没解，所以它永远不可能带上这次调用的结论。链路里于是只剩「调用发生过」。
|
|
|
|
另一半是抑制：worker 的空转长轮询在 `suppressed()` 里调 `_request`，而
|
|
`suppress_instrumentation` 只被 instrumentation 库尊重，手工建的 span 不看它。
|
|
手工埋点若不配合 `span_unless_suppressed`，孤立 trace 会从这个口子重新灌回来。
|
|
|
|
全部用 httpx.MockTransport 拦截，不触达真实网关。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import httpx
|
|
import pytest
|
|
from opentelemetry.sdk.trace import TracerProvider
|
|
from opentelemetry.sdk.trace.export import SimpleSpanProcessor
|
|
from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
|
|
from opentelemetry.trace import SpanKind, StatusCode
|
|
|
|
from app.shared.config import Settings
|
|
from app.shared.errors import AppError
|
|
from app.shared.task_state import OrderState
|
|
from app.shared.telemetry import suppressed
|
|
from app.trading.worker import client as worker_client
|
|
from app.trading.worker.client import GatewayClient
|
|
|
|
|
|
@pytest.fixture
|
|
def spans(monkeypatch) -> InMemorySpanExporter:
|
|
"""把 client 模块级 tracer 换成写内存的（不碰全局 provider）"""
|
|
exporter = InMemorySpanExporter()
|
|
provider = TracerProvider()
|
|
provider.add_span_processor(SimpleSpanProcessor(exporter))
|
|
monkeypatch.setattr(worker_client, "tracer", provider.get_tracer("test"))
|
|
return exporter
|
|
|
|
|
|
def _client(handler) -> GatewayClient:
|
|
"""构建 GatewayClient 并把它的 httpx 客户端换成 MockTransport 版本"""
|
|
client = GatewayClient(
|
|
"https://gateway.example", "token", settings=Settings(_env_file=None)
|
|
)
|
|
client._client = httpx.AsyncClient(
|
|
base_url="https://gateway.example",
|
|
transport=httpx.MockTransport(handler),
|
|
)
|
|
return client
|
|
|
|
|
|
def _envelope(*, success: bool, code: int, msg: str, data=None, status: int = 200):
|
|
return httpx.Response(
|
|
status,
|
|
json={"success": success, "msg": msg, "data": data, "code": code},
|
|
)
|
|
|
|
|
|
async def test_envelope_failure_is_recorded_on_span(spans):
|
|
"""success=false 要把 span 置 ERROR 并落错误码——HTTP 层看这次是 200"""
|
|
gateway = _client(
|
|
lambda _r: _envelope(
|
|
success=False, code=6002, msg="租约无效：不是持有者", status=200
|
|
)
|
|
)
|
|
try:
|
|
with pytest.raises(AppError) as excinfo:
|
|
await gateway.report(
|
|
"task-1", "worker-1", state=OrderState.IN_CART, terminal=False
|
|
)
|
|
assert excinfo.value.err_code == 6002
|
|
finally:
|
|
await gateway.aclose()
|
|
|
|
span = spans.get_finished_spans()[0]
|
|
assert span.kind is SpanKind.CLIENT
|
|
assert span.status.status_code is StatusCode.ERROR
|
|
# 这次调用的结论——HTTP 状态码是 200，光看它分不出成败
|
|
assert span.attributes["api.success"] is False
|
|
assert span.attributes["api.code"] == 6002
|
|
assert span.attributes["api.status_code"] == 200
|
|
assert "租约无效" in span.attributes["api.msg"]
|
|
assert span.attributes["gateway.path"] == "/api/orders/task-1/report"
|
|
assert span.attributes["gateway.method"] == "POST"
|
|
|
|
|
|
async def test_successful_envelope_leaves_span_ok(spans):
|
|
"""success=true 不置 ERROR；作为对照说明上面的断言不是恒真"""
|
|
gateway = _client(
|
|
lambda _r: _envelope(success=True, code=0, msg="success", data={"recorded": True})
|
|
)
|
|
try:
|
|
await gateway.report("task-1", "worker-1", state=OrderState.IN_CART)
|
|
finally:
|
|
await gateway.aclose()
|
|
|
|
span = spans.get_finished_spans()[0]
|
|
assert span.status.status_code is not StatusCode.ERROR
|
|
assert span.attributes["api.success"] is True
|
|
assert "api.code" not in span.attributes
|
|
|
|
|
|
async def test_non_json_body_is_recorded_as_error(spans):
|
|
"""网关回了非 JSON（网关挂了/被反代拦了）：记成可重试的 3001"""
|
|
gateway = _client(lambda _r: httpx.Response(502, text="<html>502 Bad Gateway</html>"))
|
|
try:
|
|
with pytest.raises(AppError) as excinfo:
|
|
await gateway.renew("task-1", "worker-1")
|
|
assert excinfo.value.err_code == 3001
|
|
finally:
|
|
await gateway.aclose()
|
|
|
|
span = spans.get_finished_spans()[0]
|
|
assert span.status.status_code is StatusCode.ERROR
|
|
assert span.attributes["error.code"] == 3001
|
|
# 这类失败是可重试的，与信封里的业务拒绝（retryable=False）要能区分开
|
|
assert span.attributes["error.retryable"] is True
|
|
assert span.attributes["gateway.status_code"] == 502
|
|
|
|
|
|
async def test_idle_long_poll_produces_no_span(spans):
|
|
"""回归：空转长轮询在 suppressed() 里不能产生 span
|
|
|
|
worker 每 30 秒 lease 一次、绝大多数返回空。手工埋点若不看抑制标记，这些
|
|
调用会各自成为一条孤立 trace 把观测后台刷满——正是 suppressed() 要解决的问题。
|
|
"""
|
|
gateway = _client(lambda _r: _envelope(success=True, code=0, msg="success", data=None))
|
|
try:
|
|
with suppressed():
|
|
assert await gateway.lease("worker-1", wait=0) is None
|
|
assert spans.get_finished_spans() == ()
|
|
|
|
# 抑制之外照常埋点，证明上面的空断言不是因为埋点根本没生效
|
|
assert await gateway.lease("worker-1", wait=0) is None
|
|
finally:
|
|
await gateway.aclose()
|
|
|
|
assert [s.attributes["gateway.path"] for s in spans.get_finished_spans()] == [
|
|
"/api/orders/lease"
|
|
]
|
|
|
|
|
|
async def test_leased_task_call_is_still_traced(spans):
|
|
"""领到任务后的调用不受抑制影响：它们挂在任务根 span 底下，是要看的那部分"""
|
|
gateway = _client(
|
|
lambda _r: _envelope(
|
|
success=True, code=0, msg="success",
|
|
data={"task_id": "task-1", "site": "rakuten", "lease_count": 1},
|
|
)
|
|
)
|
|
try:
|
|
task = await gateway.lease("worker-1", wait=0)
|
|
assert task is not None and task.task_id == "task-1"
|
|
finally:
|
|
await gateway.aclose()
|
|
|
|
assert len(spans.get_finished_spans()) == 1
|