"""standings 成功路径 Bronze 层回归测试(RawEvent + DataLineage)。 背景: events/stats 管线成功后均已补写 Bronze 层,唯独 standings 采集 成功后既不留原始载荷,也不留血缘 —— 三条管线的溯源链条在积分榜一环 缺失。本测试守护(与 test_events_bronze.py 对称): 1. 联赛成功 upsert → RawEvent(幂等键=standings:{league}:{season}) + Lineage(target_table="standings", transform_name="standings_ingest") 2. 更新已有快照(非插入)同样写 Bronze —— 积分榜是快照,刷新即采集 3. RawEvent 幂等: 同 source_record_id 已存在则跳过,血缘照写 4. 基础设施写入失败 → 只 warning,不拖垮采集主流程 5. 抓取失败路径继续走 _safe_write_ingest_failure,且不写 Bronze 范式: 假 db(按查询实体分发预置数据 + 记录 add,flush 分配自增 id) + monkeypatch 抓取函数,不依赖真实数据库。 """ from __future__ import annotations import pytest import src.data.bzzoiro as bz from src.db.models import DataLineage, IngestFailure, League, RawEvent, Standing, Team def _payload(): """构造一份最小合法的 bzzoiro /leagues/{id}/standings/ 原始载荷。""" return { "season": {"start_date": "2025-08-01", "end_date": "2026-05-31"}, "standings": [ { "position": 1, "team_name": "Arsenal FC", "played": 10, "won": 8, "drawn": 1, "lost": 1, "gf": 22, "ga": 8, "gd": 14, "pts": 25, "zone": {"key": "champions_league", "label": "Champions League"}, }, { "position": 2, "team_name": "Chelsea FC", "played": 10, "won": 6, "drawn": 2, "lost": 2, "gf": 18, "ga": 12, "gd": 6, "pts": 20, }, ], } def _patch_fetch(monkeypatch, payload): async def _fetch(league_code, season=None): return payload monkeypatch.setattr(bz, "fetch_bzzoiro_standings", _fetch) class _FakeResult: """支持 .scalars().all() / .scalar_one_or_none() 的最小假结果集。""" def __init__(self, items): self._items = list(items) def scalars(self): return self def __iter__(self): return iter(self._items) def all(self): return self._items def scalar_one_or_none(self): return self._items[0] if self._items else None class _FakeDB: """按查询实体分发预置数据;记录 add();flush 为无 id 对象分配自增主键。""" def __init__(self, leagues=(), teams=(), standings=(), raw_events=()): self.added = [] self._by_entity = { League: list(leagues), Team: list(teams), Standing: list(standings), RawEvent: list(raw_events), } self._next_id = 0 def add(self, obj): self.added.append(obj) async def execute(self, stmt): entities = set() for d in (stmt.column_descriptions or []): entities.add(d.get("entity") or d.get("type")) for entity, items in self._by_entity.items(): if entity in entities: return _FakeResult(self._filter(entity, items, stmt)) return _FakeResult([]) @staticmethod def _filter(entity, items, stmt): """RawEvent 查询按 source_record_id 过滤 —— 幂等测试需区分不同键。""" if entity is RawEvent: try: params = stmt.compile().params except Exception: return items rid = next((v for k, v in params.items() if "source_record_id" in k), None) if rid is not None: return [i for i in items if i.source_record_id == rid] return items async def flush(self): for obj in self.added: if getattr(obj, "id", None) is None: self._next_id += 1 obj.id = self._next_id def _preset_league(): lg = League(code="EPL", name="Premier League", country="England") lg.id = 42 return lg def _raw_events(db): return [o for o in db.added if isinstance(o, RawEvent)] def _lineages(db): return [o for o in db.added if isinstance(o, DataLineage)] # ============================================================ # 1. 成功 upsert → RawEvent + DataLineage # ============================================================ class TestStandingsBronzeOnUpsert: async def test_upsert_writes_raw_event_and_lineage(self, monkeypatch): _patch_fetch(monkeypatch, _payload()) db = _FakeDB(leagues=[_preset_league()]) result = await bz.ingest_bzzoiro_standings(db, leagues=["EPL"]) assert result["errors"] == [] assert result["total_upserted"] == 2 raws = _raw_events(db) assert len(raws) == 1 raw = raws[0] assert raw.source_system == "bzzoiro" # 幂等键: 联赛 + 实际入库的赛季标签(由载荷日期推导,与 Standing.season 同口径) assert raw.source_record_id == "standings:EPL:2025-2026" assert raw.ingest_batch_id.startswith("bzzoiro-standings-EPL-") # 整份原始载荷完整保留 assert raw.raw_payload["standings"][0]["team_name"] == "Arsenal FC" lineages = _lineages(db) assert len(lineages) == 1 lin = lineages[0] assert lin.source_system == "bzzoiro" assert lin.source_record_id == "standings:EPL:2025-2026" assert lin.target_table == "standings" assert lin.target_id == 42 # 联赛 id assert lin.transform_name == "standings_ingest" assert lin.transform_detail == { "league": "EPL", "season": "2025-2026", "rows_upserted": 2, } # RawEvent 与 Lineage 同批次,便于按批追溯 assert lin.batch_id == raw.ingest_batch_id async def test_updated_snapshot_also_writes_bronze(self, monkeypatch): """已有快照就地更新(非插入)同样是成功采集,必须留 Bronze 记录。""" payload = _payload() payload["standings"] = payload["standings"][:1] # 单队,便于命中同一行 _patch_fetch(monkeypatch, payload) team = Team(name="Arsenal FC", name_zh="阿森纳") team.id = 7 existing = Standing(league_id=42, season="2025-2026", team_id=7, position=9) existing.points = 1 db = _FakeDB(leagues=[_preset_league()], teams=[team], standings=[existing]) result = await bz.ingest_bzzoiro_standings(db, leagues=["EPL"]) assert result["total_upserted"] == 1 assert result["leagues"]["EPL"]["teams_created"] == 0 # 快照刷新也要留痕: RawEvent(幂等) + 血缘 assert len(_raw_events(db)) == 1 lineages = _lineages(db) assert len(lineages) == 1 assert lineages[0].transform_name == "standings_ingest" assert lineages[0].transform_detail["rows_upserted"] == 1 async def test_empty_upsert_writes_no_bronze(self, monkeypatch): """载荷有行但全部队名为空 → 没有任何 upsert,不应产生 RawEvent/Lineage。""" payload = {"standings": [{"position": 1, "team_name": ""}]} _patch_fetch(monkeypatch, payload) db = _FakeDB(leagues=[_preset_league()]) result = await bz.ingest_bzzoiro_standings(db, leagues=["EPL"]) assert result["total_upserted"] == 0 assert _raw_events(db) == [] assert _lineages(db) == [] # ============================================================ # 2. RawEvent 幂等: 同 source_record_id 跳过 # ============================================================ class TestStandingsRawEventIdempotent: async def test_existing_raw_event_is_skipped(self, monkeypatch): existing = RawEvent( source_system="bzzoiro", source_record_id="standings:EPL:2025-2026", raw_payload={"old": True}, ) _patch_fetch(monkeypatch, _payload()) db = _FakeDB(leagues=[_preset_league()], raw_events=[existing]) await bz.ingest_bzzoiro_standings(db, leagues=["EPL"]) new_raws = [r for r in _raw_events(db) if r is not existing] assert new_raws == [] assert len(_lineages(db)) == 1 # 血缘仍然记录本次采集 async def test_different_season_writes_new_raw_event(self, monkeypatch): """幂等键含赛季: 同联赛不同赛季各留一条 RawEvent。""" payload = _payload() payload["season"] = {"start_date": "2024-08-01", "end_date": "2025-05-31"} existing = RawEvent( source_system="bzzoiro", source_record_id="standings:EPL:2025-2026", raw_payload={"old": True}, ) _patch_fetch(monkeypatch, payload) db = _FakeDB(leagues=[_preset_league()], raw_events=[existing]) await bz.ingest_bzzoiro_standings(db, leagues=["EPL"]) new_raws = [r for r in _raw_events(db) if r is not existing] assert len(new_raws) == 1 assert new_raws[0].source_record_id == "standings:EPL:2024-2025" # ============================================================ # 3. 基础设施写入失败: 尽力而为,不拖垮主流程 # ============================================================ class TestStandingsBronzeIsBestEffort: async def test_bronze_write_failure_does_not_break_ingest(self, monkeypatch): async def _boom(*args, **kwargs): raise RuntimeError("infra down") monkeypatch.setattr(bz, "_write_raw_event", _boom) monkeypatch.setattr(bz, "_write_lineage", _boom) _patch_fetch(monkeypatch, _payload()) db = _FakeDB(leagues=[_preset_league()]) # 不应抛异常:Bronze 写不进去只记 warning result = await bz.ingest_bzzoiro_standings(db, leagues=["EPL"]) assert result["total_upserted"] == 2 assert [o for o in db.added if isinstance(o, Standing)] # ============================================================ # 4. 抓取失败: 继续写死信,且不写 Bronze # ============================================================ class TestStandingsFailurePathKeepsDeadLetter: async def test_fetch_failure_writes_deadletter_and_no_bronze(self, monkeypatch): async def _boom(league_code, season=None): raise RuntimeError("upstream 500") monkeypatch.setattr(bz, "fetch_bzzoiro_standings", _boom) db = _FakeDB() result = await bz.ingest_bzzoiro_standings(db, leagues=["SP1"], season="2025-2026") assert result["errors"] failures = [o for o in db.added if isinstance(o, IngestFailure)] assert len(failures) == 1 assert failures[0].entity_type == "standings" assert failures[0].error_type == "fetch_error" # 失败路径绝不写 Bronze(没有任何成功 upsert) assert _raw_events(db) == [] assert _lineages(db) == []