面向协议形态代码（分片流、事件日志、schema转换等）的基…

设计思路

核心不变量

• 不丢消息：乱序分片重组后，成功解析的事件数 ≥ 原始事件数中可合法解析的子集；失败项必须进入死信/错误队列，不允许静默丢弃
• 不丢字段：每条重组后的事件，其字段集合与字段值与原始 JSON 深度相等
• 健壮性：畸形输入不抛未捕获异常，只产生明确的解析错误结果

代码骨架

```python

import json, random

from hypothesis import given, strategies as st

st_payload = st.dictionaries(

st.text(min_size=1, max_size=20),

st.one_of(st.integers(), st.text(), st.floats(allow_nan=False),

st.booleans(), st.none()),

max_size=10,

)

st_raw = st.one_of(st_payload.map(json.dumps), # 合法JSON

st.text(max_size=50)) # 畸形字节串

@st.composite

def event_shards(draw):

raw = draw(st_raw)

n = draw(st.integers(1, 5))

if not raw:

return raw, [b""]

cuts = sorted(draw(st.data() and st.integers(0, len(raw)) for _ in range(n-1)))

parts = [raw[a:b] for a, b in zip([0]+cuts, cuts+[len(raw)])]

random.shuffle(parts) # 模拟乱序

return raw, parts

@given(event_shards())

def test_reassembly_no_loss_no_field_drop(case):

raw, parts = case

events, dlq = parse_and_reassemble(parts) # 被测模块

try:

expected = json.loads(raw)

except (json.JSONDecodeError, UnicodeDecodeError):

assert events == [] and len(dlq) == 1 # 畸形→死信

return

merged = "".join(p.decode("utf-8", errors="replace") for p in parts)

actual = json.loads(merged)

assert isinstance(actual, dict) # 结构未破坏

for k, v in expected.items(): # 逐字段比对

assert k in actual and actual[k] == v

assert len(events) + len(dlq) >= 1 # 总数不减

```

畸形流策略要点

• 分层混合：合法 JSON 与任意文本按比例混合，畸形样本覆盖截断