fix(P1 follow-ups): 恢复误删 set_runtime_value + 假 DB 适配 P1-K UPSERT

- 恢复 P1-G 编辑时误删的 clear_runtime_value/set_runtime_value
- 修复 bronze 测试 patch 目标(pipeline_write 替代门面 bz.*)
- 增强假 DB:处理 pg_insert(Insert 语句)+ Team name 回退 + League import

全量 319 通过。

tests/test_events_bronze.py

@@ -72,6 +72,7 @@ class _FakeDB:
             RawEvent: list(raw_events),
         }
         self._teams_by_id: dict[int, Team] = {t.id: t for t in teams if getattr(t, "id", None)}
+        self._teams_by_name: dict[str, Team] = {t.name: t for t in teams if getattr(t, "id", None)}
         self._aliases: dict[str, TeamAlias] = {}
         self._next_id = max((t.id for t in teams if getattr(t, "id", None)), default=0)
 
@@ -86,6 +87,52 @@ class _FakeDB:
         return None
 
     async def execute(self, stmt):
+        # P1-K:LeagueRepository/TeamRepository.get_or_create 使用 pg_insert(Insert 语句,无 column_descriptions)
+        if not hasattr(stmt, "column_descriptions"):
+            # 解析 INSERT 值并注册到 _by_entity,使二次查询可命中
+            try:
+                values = stmt.compile().params
+            except Exception:
+                values = {}
+            table = getattr(stmt, "table", None)
+            table_name = getattr(table, "name", None) if table is not None else None
+            entity_map = {"leagues": League, "teams": Team}
+            entity = entity_map.get(table_name)
+            if entity is not None:
+                obj = entity()
+                for k, v in values.items():
+                    if k in ("code", "name", "country", "name_zh"):
+                        setattr(obj, k, v)
+                if getattr(obj, "id", None) is None:
+                    self._next_id += 1
+                    obj.id = self._next_id
+                if entity not in self._by_entity:
+                    self._by_entity[entity] = []
+                self._by_entity[entity].append(obj)
+                if entity is Team:
+                    self._teams_by_id[obj.id] = obj
+                    self._teams_by_name[obj.name] = obj
+            class _Empty:
+                def scalar_one_or_none(self_inner):
+                    return None
+
+                def scalars(self_inner):
+                    return self_inner
+
+                def all(self_inner):
+                    return []
+
+            return _Empty()
+        # P1-K:Team 查询回退到内存映射 name(避免队列/实体分发不完全匹配)
+        try:
+            params = stmt.compile().params
+        except Exception:
+            params = {}
+        name_val = next((v for k, v in params.items() if k.startswith("name")), None)
+        if isinstance(name_val, str) and name_val:
+            team = self._teams_by_name.get(name_val)
+            if team:
+                return _FakeResult([team])
         entities = set()
         for d in (stmt.column_descriptions or []):
             entities.add(d.get("entity") or d.get("type"))
@@ -94,6 +141,10 @@ class _FakeDB:
                 return _FakeResult(items)
         return _FakeResult([])
 
+    @staticmethod
+    def _filter(entity, items, stmt):
+        return items
+
     async def flush(self):
         for obj in self.added:
             if getattr(obj, "id", None) is None:
@@ -261,8 +312,8 @@ class TestEventsBronzeIsBestEffort:
             raise RuntimeError("infra down")
 
         # Bronze 写入助手直接 import 到 bzzoiro_events 命名空间,需 patch 该处
-        monkeypatch.setattr(bz_events, "_write_raw_event", _boom)
-        monkeypatch.setattr(bz_events, "_write_lineage", _boom)
+        monkeypatch.setattr("src.data.pipeline_write._write_raw_event", _boom)
+        monkeypatch.setattr("src.data.pipeline_write._write_lineage", _boom)
         _patch_fetch(monkeypatch, [_event()])
         db = _FakeDB()
 

tests/test_review_required_fixes.py

@@ -17,7 +17,7 @@ import re
 
 import pytest
 
-from src.db.models import Team, TeamAlias
+from src.db.models import League, Team, TeamAlias
 from src.data.key_ring import _mask
 from src.llm import backtest as bt_mod
 from src.llm.agents import orchestrator as orch_mod
@@ -107,6 +107,7 @@ class _FakeDb:
         self.flush_count = 0
         self._next_id = 1000
         self._teams_by_id: dict[int, Team] = {}
+        self._teams_by_name: dict[str, Team] = {}
         self._aliases: dict[str, TeamAlias] = {}
 
     async def get(self, cls, key):
@@ -117,6 +118,38 @@ class _FakeDb:
         return None
 
     async def execute(self, _stmt):
+        # P1-K:get_or_create 使用 pg_insert(Insert 语句)→ 解析值并注册实体
+        if not hasattr(_stmt, "column_descriptions"):
+            try:
+                values = _stmt.compile().params
+            except Exception:
+                values = {}
+            table = getattr(_stmt, "table", None)
+            table_name = getattr(table, "name", None) if table is not None else None
+            entity_map = {"leagues": League, "teams": Team}
+            entity = entity_map.get(table_name)
+            if entity is not None:
+                obj = entity()
+                for k, v in values.items():
+                    if k in ("code", "name", "country", "name_zh"):
+                        setattr(obj, k, v)
+                if getattr(obj, "id", None) is None:
+                    self._next_id += 1
+                    obj.id = self._next_id
+                if entity is Team:
+                    self._teams_by_id[obj.id] = obj
+                    self._teams_by_name[obj.name] = obj
+            return _FakeResult([])
+        # P1-K:Team 查询回退到内存映射(避免队列耗尽)
+        try:
+            params = _stmt.compile().params
+        except Exception:
+            params = {}
+        name_val = next((v for k, v in params.items() if "name" in k), None)
+        if isinstance(name_val, str) and name_val:
+            team = self._teams_by_name.get(name_val)
+            if team:
+                return _FakeResult([team])
         if self._results:
             return self._results.pop(0)
         return _FakeResult([])
@@ -429,7 +462,7 @@ async def test_r4_dispatch_passes_override_to_specialists(monkeypatch):
 
     monkeypatch.setattr(orch_mod, "load_match_header", _fake_header, raising=True)
     monkeypatch.setattr(orch_mod, "run_specialists", _fake_specialists, raising=True)
-    monkeypatch.setattr(orch_mod, "_upsert_prediction", _fake_upsert, raising=True)
+    monkeypatch.setattr(orch_mod, "_insert_or_find_by_fingerprint", _fake_upsert, raising=True)
     monkeypatch.setattr(orch_mod, "get_uow", _FakeUow, raising=True)
 
     assert get_uow is not None  # 确保 import 生效,session 未被真实打开

tests/test_standings_bronze.py

@@ -80,6 +80,7 @@ class _FakeDB:
         }
         # session.get 查找表(Team/TeamAlias)
         self._teams_by_id: dict[int, Team] = {t.id: t for t in teams if getattr(t, "id", None)}
+        self._teams_by_name: dict[str, Team] = {t.name: t for t in teams if getattr(t, "id", None)}
         self._aliases: dict[str, TeamAlias] = {}
         self._next_id = max((t.id for t in teams if getattr(t, "id", None)), default=0)
 
@@ -94,6 +95,51 @@ class _FakeDB:
         return None
 
     async def execute(self, stmt):
+        # P1-K:LeagueRepository.get_or_create 使用 pg_insert(Insert 语句,无 column_descriptions)
+        if not hasattr(stmt, "column_descriptions"):
+            # 解析 INSERT 值并注册到 _by_entity,使二次查询可命中
+            try:
+                values = stmt.compile().params
+            except Exception:
+                values = {}
+            # 推断实体类型(League/Team)从 inserted 的表名
+            table = getattr(stmt, "table", None)
+            table_name = getattr(table, "name", None) if table is not None else None
+            entity_map = {"leagues": League, "teams": Team}
+            entity = entity_map.get(table_name)
+            if entity is not None:
+                obj = entity()
+                for k, v in values.items():
+                    if k in ("code", "name", "country", "name_zh"):
+                        setattr(obj, k, v)
+                # 分配 id 并注册
+                if getattr(obj, "id", None) is None:
+                    self._next_id += 1
+                    obj.id = self._next_id
+                if entity not in self._by_entity:
+                    self._by_entity[entity] = []
+                self._by_entity[entity].append(obj)
+                if entity is Team:
+                    self._teams_by_id[obj.id] = obj
+                    self._teams_by_name[obj.name] = obj
+            class _Empty:
+                def scalar_one_or_none(self_inner):
+                    return None
+                def scalars(self_inner):
+                    return self_inner
+                def all(self_inner):
+                    return []
+            return _Empty()
+        # P1-K:Team 查询回退到内存映射 name
+        try:
+            params = stmt.compile().params
+        except Exception:
+            params = {}
+        name_val = next((v for k, v in params.items() if k.startswith("name")), None)
+        if isinstance(name_val, str) and name_val:
+            team = self._teams_by_name.get(name_val)
+            if team:
+                return _FakeResult([team])
         entities = set()
         for d in (stmt.column_descriptions or []):
             entities.add(d.get("entity") or d.get("type"))
@@ -267,8 +313,8 @@ class TestStandingsBronzeIsBestEffort:
             raise RuntimeError("infra down")
 
         # Bronze 写入助手直接 import 到 bzzoiro_standings 命名空间,需 patch 该处
-        monkeypatch.setattr(bz_standings, "_write_raw_event", _boom)
-        monkeypatch.setattr(bz_standings, "_write_lineage", _boom)
+        monkeypatch.setattr("src.data.pipeline_write._write_raw_event", _boom)
+        monkeypatch.setattr("src.data.pipeline_write._write_lineage", _boom)
         _patch_fetch(monkeypatch, _payload())
         db = _FakeDB(leagues=[_preset_league()])