diff --git a/tests/test_scrape_telemetry.py b/tests/test_scrape_telemetry.py new file mode 100644 index 0000000..0395956 --- /dev/null +++ b/tests/test_scrape_telemetry.py @@ -0,0 +1,203 @@ +"""抓取客户端失败分支的埋点测试：失败原因必须能从 span 上读出来 + +这里钉住的核心是一个真实存在过的 bug：失败分支里写的是无参 +`span.record_exception()`，而该方法的 `exception` 是必填位置参数——**每一次抓取 +失败都会在记录异常时抛 TypeError**，把真正的失败原因（反爬阻断 / 解析失败 / +404）整个替换掉。链路上只剩「调用发生过」，异常怎么来的完全看不到，失败页面 +快照也永远落不下来（抛错发生在 snapshot 之前）。 + +用独立的 InMemory provider 断言 span，不碰全局 provider——OTel 的全局 provider +只允许设置一次，测试间共享会互相污染（与 test_telemetry.py 同一套思路）。 +""" +from __future__ import annotations + +import json + +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 StatusCode + +from app.scraping.models.scrape import ( + ItemDetailRequest, + RakumaSearchRequest, + SearchRequest, + ShopItemsRequest, +) +from app.scraping.services import rakuma_client as rakuma_module +from app.scraping.services import rakuten_client as rakuten_module +from app.scraping.services.rakuma_client import RakumaClient +from app.scraping.services.rakuten_client import RakutenClient +from app.shared.config import Settings +from app.shared.errors import AppError, ScrapeParseError, UpstreamBlockedError + + +@pytest.fixture +def spans(monkeypatch) -> InMemorySpanExporter: + """把两个抓取客户端模块级 tracer 换成写内存的，用于断言 span""" + exporter = InMemorySpanExporter() + provider = TracerProvider() + provider.add_span_processor(SimpleSpanProcessor(exporter)) + tracer = provider.get_tracer("test") + monkeypatch.setattr(rakuten_module, "tracer", tracer) + monkeypatch.setattr(rakuma_module, "tracer", tracer) + return exporter + + +def _settings() -> Settings: + return Settings(_env_file=None, otel_snapshot_max_bytes=2_000_000) + + +class _StubSession: + """按需返回 HTML 或抛异常的会话桩；两站的 fetch_html 签名差异在这里吸收""" + + def __init__(self, *, html: str = "", error: Exception | None = None): + self._html = html + self._error = error + + async def fetch_html(self, url: str, *, mobile: bool = False) -> str: + if self._error is not None: + raise self._error + return self._html + + async def fetch(self, url: str, *, mobile: bool = False, validator=None): + if self._error is not None: + raise self._error + raise AssertionError("本测试不该走到 fetch 的成功分支") + + +def _span(exporter: InMemorySpanExporter, name: str): + return next(s for s in exporter.get_finished_spans() if s.name == name) + + +async def test_parse_failure_preserves_original_exception(spans): + """回归：解析失败要原样抛出业务异常，不能被埋点自身的 TypeError 顶替 + + 这是原 bug 最直接的后果——上游拿到的不再是 4001（解析失败），而是一个 + TypeError 兜底成的 500，错误码表整个失效。 + """ + client = RakutenClient(_settings(), _StubSession(html="no state")) + + with pytest.raises(ScrapeParseError) as excinfo: + await client.search(SearchRequest(keyword="switch")) + # 是业务异常而不是埋点炸出来的 TypeError + assert isinstance(excinfo.value, AppError) + assert excinfo.value.err_code == 4001 + + +async def test_parse_failure_records_reason_and_page_snapshot(spans): + """页面取回来了但解析不出：stage=parse，且失败页面 HTML 落成 event + + 「站点改版了」只能靠当时那份 HTML 判断，所以快照必须真的落下来——原 bug 里 + 记异常那步先抛了，snapshot 这行永远执行不到。 + """ + html = "changed layout" + client = RakutenClient(_settings(), _StubSession(html=html)) + + with pytest.raises(ScrapeParseError): + await client.search(SearchRequest(keyword="switch")) + + span = _span(spans, "parse.rakuten.search") + assert span.status.status_code is StatusCode.ERROR + assert span.attributes["parse.stage"] == "parse" + # 真实异常类名，不是一律写死的 "parse_error" + assert span.attributes["parse.fail_reason"] == "ScrapeParseError" + assert span.attributes["error.code"] == 4001 + assert span.attributes["error.retryable"] is False + + snapshot = next(e for e in span.events if e.name == "parse.failed_html") + assert snapshot.attributes["snapshot.html"] == html + # 快照要能对上是哪个地址的页面 + assert "search.rakuten.co.jp" in snapshot.attributes["parse.url"] + + +async def test_fetch_failure_marks_fetch_stage_without_snapshot(spans): + """页面根本没取回来：stage=fetch，没有快照可落 + + 与 parse 阶段的排查方向相反（通道/反爬/上游 5xx，而不是站点改版）， + 所以要能直接按 parse.stage 分流，而不是对着有没有 HTML 猜。 + """ + blocked = UpstreamBlockedError("Blocked while fetching: challenge page detected") + client = RakutenClient(_settings(), _StubSession(error=blocked)) + + with pytest.raises(UpstreamBlockedError): + await client.search(SearchRequest(keyword="switch")) + + span = _span(spans, "parse.rakuten.search") + assert span.attributes["parse.stage"] == "fetch" + assert span.attributes["parse.fail_reason"] == "UpstreamBlockedError" + # 反爬阻断是可重试的，这一位直接决定上游要不要重来 + assert span.attributes["error.retryable"] is True + assert not [e for e in span.events if e.name == "parse.failed_html"] + + +async def test_item_detail_failure_records_reason(spans): + """详情接口走的是 fetch()（带校验器）而非 fetch_html，失败分支同样要记全""" + blocked = UpstreamBlockedError("challenge page detected") + client = RakutenClient(_settings(), _StubSession(error=blocked)) + + with pytest.raises(UpstreamBlockedError): + await client.item_detail( + ItemDetailRequest(shop_code="someshop", item_code="10000001") + ) + + span = _span(spans, "parse.rakuten.item_detail") + assert span.status.status_code is StatusCode.ERROR + assert span.attributes["parse.fail_reason"] == "UpstreamBlockedError" + + +async def test_shop_items_marks_delegate_stage(spans): + """shop_items 自己不抓页面：失败在转调的 shop_detail / search 里 + + 按 html 推断会得出「fetch 失败」的错误结论（它手里从来没有 html）， + 所以这里显式标 delegate，且不落快照——真正的现场在被转调那个 span 上。 + """ + blocked = UpstreamBlockedError("challenge page detected") + client = RakutenClient(_settings(), _StubSession(error=blocked)) + + with pytest.raises(UpstreamBlockedError): + await client.shop_items(ShopItemsRequest(shop_code="someshop")) + + span = _span(spans, "parse.rakuten.shop_items") + assert span.attributes["parse.stage"] == "delegate" + assert not [e for e in span.events if e.name == "parse.failed_html"] + # 被转调的那一步才是现场所在，它自己落了快照 + inner = _span(spans, "parse.rakuten.shop_detail") + assert inner.attributes["parse.fail_reason"] == "UpstreamBlockedError" + + +async def test_rakuma_parse_failure_records_reason_and_snapshot(spans): + """ラクマ 侧五个接口是同一套失败分支，同样要能读出原因与现场""" + html = "rakuma changed" + client = RakumaClient(_settings(), _StubSession(html=html)) + + with pytest.raises(AppError): + await client.search(RakumaSearchRequest(keyword="switch")) + + span = _span(spans, "parse.rakuma.search") + assert span.status.status_code is StatusCode.ERROR + assert span.attributes["parse.stage"] == "parse" + assert "error.code" in span.attributes + snapshot = next(e for e in span.events if e.name == "parse.failed_html") + assert snapshot.attributes["snapshot.html"] == html + + +async def test_successful_scrape_leaves_span_ok(spans, search_state): + """成功路径不被误标 ERROR，不落失败快照，且照常记结果指标 + + 作为上面那些失败断言的对照：证明 ERROR 状态与 parse.fail_reason 是真的由失败 + 触发的，不是每条 span 都长这样。 + """ + html = f"" + client = RakutenClient(_settings(), _StubSession(html=html)) + + result = await client.search(SearchRequest(keyword="switch")) + + span = _span(spans, "parse.rakuten.search") + assert span.status.status_code is not StatusCode.ERROR + assert "parse.fail_reason" not in span.attributes + assert "error.type" not in span.attributes + assert not [e for e in span.events if e.name == "parse.failed_html"] + # 成功时记的是结果指标，与失败侧属性互不重叠 + assert span.attributes["parse.items"] == len(result.items) diff --git a/tests/test_worker_client_telemetry.py b/tests/test_worker_client_telemetry.py new file mode 100644 index 0000000..ba5dd4b --- /dev/null +++ b/tests/test_worker_client_telemetry.py @@ -0,0 +1,158 @@ +"""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="502 Bad Gateway")) + 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