iomgaa
484900d300
Issue #10 Task 2: head-and-tail rather than a head-only cut, because the code and request_id that let you chase the provider sit at the very end of a JSON error body. Cap 2048 follows k8s client-go for the same job.
90 lines
3.3 KiB
Python
90 lines
3.3 KiB
Python
"""HTTP 错误响应体摘要口径(issue #10 设计 §3.2/§3.4)。
|
|
|
|
摘要是 message 与 `body_text` 共用的**同一份串**,故它的边界行为直接决定
|
|
遥测里看到的与下游 catch 到的是否一致——本组用例把规则钉成算术。
|
|
"""
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from polygateway.transports._http_errors import (
|
|
_ERROR_BODY_CAP,
|
|
_HEAD_CHARS,
|
|
_TAIL_CHARS,
|
|
compose_message,
|
|
response_body,
|
|
summarize_body,
|
|
)
|
|
|
|
# issue #10 原文给出的真实响应体(一字不改),关键在于 code 收尾
|
|
_REAL_SAMPLE = (
|
|
'{"error":{"message":"<400> ***.***.InvalidParameter: The image format is illegal '
|
|
'and cannot be opened","type":"invalid_request_error","param":"",'
|
|
'"code":"invalid_parameter_error"}}'
|
|
)
|
|
|
|
|
|
class TestSummarizeBody:
|
|
def test_short_body_passes_through(self):
|
|
assert summarize_body(_REAL_SAMPLE) == _REAL_SAMPLE
|
|
|
|
def test_whitespace_collapsed(self):
|
|
"""错误体常是缩进 JSON: 不折叠会把一行日志炸成多行、遥测列不可读。"""
|
|
assert (
|
|
summarize_body('{\n "error": {\n "code": "x"\n }\n}')
|
|
== '{ "error": { "code": "x" } }'
|
|
)
|
|
|
|
@pytest.mark.parametrize("raw", ("", " ", "\n\t \n"))
|
|
def test_blank_yields_empty(self, raw):
|
|
assert summarize_body(raw) == ""
|
|
|
|
def test_exactly_at_cap_is_untouched(self):
|
|
body = "x" * _ERROR_BODY_CAP
|
|
assert summarize_body(body) == body
|
|
|
|
def test_one_over_cap_is_summarized(self):
|
|
summary = summarize_body("x" * (_ERROR_BODY_CAP + 1))
|
|
assert summary != "x" * (_ERROR_BODY_CAP + 1)
|
|
assert "略" in summary
|
|
|
|
def test_head_and_tail_both_survive(self):
|
|
"""头部硬切会丢掉尾部,而 JSON 错误体的 code/request_id 正在尾部。"""
|
|
body = "H" * 5000 + "T" * 5000
|
|
summary = summarize_body(body)
|
|
assert summary[:_HEAD_CHARS] == body[:_HEAD_CHARS]
|
|
assert summary[-_TAIL_CHARS:] == body[-_TAIL_CHARS:]
|
|
assert f"…(略 {10000 - _HEAD_CHARS - _TAIL_CHARS} 字)…" in summary
|
|
|
|
def test_real_sample_tail_visible_in_oversized_body(self):
|
|
"""设计 §7 用例 3c: 超长体里,追查网关方所需的 code 仍须可见。"""
|
|
summary = summarize_body("PADDING" * 1000 + _REAL_SAMPLE)
|
|
assert '"code":"invalid_parameter_error"}}' in summary
|
|
|
|
def test_idempotent(self):
|
|
"""再摘要一次不得嵌套标记,否则重复经手的串会层层套娃。"""
|
|
once = summarize_body("y" * 9999)
|
|
assert summarize_body(once) == once
|
|
|
|
|
|
class TestComposeMessage:
|
|
def test_empty_summary_leaves_message_intact(self):
|
|
assert compose_message("qwen_1 请求被拒: 400", "") == "qwen_1 请求被拒: 400"
|
|
|
|
def test_non_empty_summary_is_appended(self):
|
|
assert compose_message("qwen_1 请求被拒: 400", "{}") == "qwen_1 请求被拒: 400 | {}"
|
|
|
|
|
|
class TestResponseBody:
|
|
def test_reads_buffered_text(self):
|
|
assert response_body(httpx.Response(400, content=b'{"e":1}')) == '{"e":1}'
|
|
|
|
def test_unread_stream_degrades_to_empty(self):
|
|
"""取不到诊断信息绝不能升级为崩溃: 未读缓冲返回空串,且不触发网络读。"""
|
|
|
|
class _Unread(httpx.SyncByteStream):
|
|
def __iter__(self):
|
|
yield b"body"
|
|
|
|
assert response_body(httpx.Response(400, stream=_Unread())) == ""
|