shangfangjian
41cb2edd47
旧代码读 r.get("inserted")(顶层无此 key)导致联赛级计数总为 0;
改为读 r["leagues"][code].inserted/updated,与顶层 total_* 分离。
抽 _accumulate_ingest_result 纯函数 + 4 合约测试(全绿)。
68 lines
3.0 KiB
Python
68 lines
3.0 KiB
Python
"""P1-A 回归测试: ingest 联赛级 inserted/updated 必须读 leagues[code],而非顶层 r.get("inserted")。
|
|
|
|
运行: pytest tests/test_p1_a_ingest_league_counts.py -v
|
|
(纯函数测试,无 DB/网络依赖。)
|
|
"""
|
|
from src.api.routes.ingest import _accumulate_ingest_result
|
|
|
|
|
|
class TestAccumulateIngestResult:
|
|
"""P1-A: _accumulate_ingest_result 联赛级计数必须来自 r["leagues"][code]。"""
|
|
|
|
def _merged(self):
|
|
return {"leagues": {}, "total_inserted": 0, "total_updated": 0, "errors": []}
|
|
|
|
def test_league_counts_read_from_leagues_key(self):
|
|
"""核心: 联赛级 inserted/updated 应来自 leagues[code],而非顶层 inserted/updated。"""
|
|
merged = self._merged()
|
|
r = {
|
|
# 顶层无 inserted/updated 键(只有 total_*)
|
|
"total_inserted": 5,
|
|
"total_updated": 2,
|
|
"errors": [],
|
|
"leagues": {"E0": {"inserted": 3, "updated": 1, "rows": 4, "errors": []}},
|
|
}
|
|
_accumulate_ingest_result(merged, "E0", r)
|
|
|
|
# 顶层总计
|
|
assert merged["total_inserted"] == 5
|
|
assert merged["total_updated"] == 2
|
|
# 联赛级计数来自 leagues["E0"],而非顶层
|
|
assert merged["leagues"]["E0"]["inserted"] == 3, "联赛 inserted 必须来自 leagues[code]"
|
|
assert merged["leagues"]["E0"]["updated"] == 1, "联赛 updated 必须来自 leagues[code]"
|
|
|
|
def test_does_not_read_top_level_inserted(self):
|
|
"""防御: 若 r 误含顶层 inserted 键,不得影响联赛级计数。"""
|
|
merged = self._merged()
|
|
r = {
|
|
"total_inserted": 5,
|
|
"total_updated": 2,
|
|
"inserted": 999, # 错误的顶层键(旧代码可能读这个)
|
|
"updated": 999,
|
|
"errors": [],
|
|
"leagues": {"E0": {"inserted": 3, "updated": 1}},
|
|
}
|
|
_accumulate_ingest_result(merged, "E0", r)
|
|
# 必须忽略顶层 inserted/updated,使用 leagues["E0"]
|
|
assert merged["leagues"]["E0"]["inserted"] == 3
|
|
assert merged["leagues"]["E0"]["updated"] == 1
|
|
|
|
def test_missing_league_key_defaults_to_zero(self):
|
|
"""r["leagues"] 无该 code 时,默认 0 不抛错。"""
|
|
merged = self._merged()
|
|
r = {"total_inserted": 1, "total_updated": 0, "errors": [], "leagues": {}}
|
|
_accumulate_ingest_result(merged, "E0", r)
|
|
assert merged["leagues"]["E0"]["inserted"] == 0
|
|
assert merged["total_inserted"] == 1
|
|
|
|
def test_multiple_calls_accumulate(self):
|
|
"""多次调用应累加到同一联赛。"""
|
|
merged = self._merged()
|
|
r1 = {"total_inserted": 3, "total_updated": 1, "errors": [], "leagues": {"E0": {"inserted": 3, "updated": 1}}}
|
|
r2 = {"total_inserted": 2, "total_updated": 0, "errors": [], "leagues": {"E0": {"inserted": 2, "updated": 0}}}
|
|
_accumulate_ingest_result(merged, "E0", r1)
|
|
_accumulate_ingest_result(merged, "E0", r2)
|
|
assert merged["leagues"]["E0"]["inserted"] == 5
|
|
assert merged["leagues"]["E0"]["updated"] == 1
|
|
assert merged["total_inserted"] == 5
|