Files
Profeto/tests/test_bzzoirot_stats.py
T
Profeto Agent 6c24672e87 安全加固 + 数据管线修复 + 质量改进(第2批)
安全加固:
- /api/v1/predict 限流改用 get_client_ip 防 X-Forwarded-For 伪造
- 新增 TRUST_PROXY_HEADERS/REQUIRE_ADMIN_AUTH 配置,默认 fail-closed
- 生产环境管理接口未配置鉴权时拒绝(503),不再放行

数据管线修复:
- injuries IntegrityError 改用 begin_nested(SAVEPOINT)隔离批次
- injuries 空名单 vs 未配置语义区分(has_data 精确标记)
- bzzoiro 统计字段映射(shots/possession/cards) + 入库条件放宽
- available_at 语义收紧(回测防泄漏)
- team_names NFKD 去变音双重查找 + 补齐变体键

预测系统改进:
- multi-agent 全专家失败时 status=degraded 跳过终裁
- agent_weights 独立持久化到 predictions 表

新增迁移:
- 0014_predictions_agent_weights.py

新增测试(8个文件):
- test_ip_spoofing.py: 限流防伪造
- test_require_admin_fail_closed.py: 生产 fail-closed
- test_injuries_integrity_rollback.py: SAVEPOINT 隔离
- test_injuries_slice_semantic.py: 空名单 vs 未配置
- test_available_at.py: 回测防泄漏
- test_bzzoirot_stats.py: 统计字段映射
- test_team_names_normalize.py: NFKD 变体
- test_multi_agent_degraded.py: 全失败 degraded
- test_agent_weights_persist.py: 权重持久化
2026-09-19 08:02:57 +00:00

207 lines
6.7 KiB
Python

"""回归测试: bzzoiro 采集链路正确映射射门/控球/角球/xG。
验证:
1. normalize_bzzoiro 正确映射统计字段
2. 入库条件不再强制要求 xG(任一统计字段即可)
3. API 没有的字段保持 None,不伪造
"""
from __future__ import annotations
from datetime import datetime, timezone
from unittest.mock import MagicMock
import pytest
from src.data.normalize import NormalizedMatch, normalize_bzzoiro
class TestNormalizeBzzoirotStats:
"""normalize_bzzoiro 应正确映射统计字段。"""
def test_maps_shots(self):
"""API 提供 shots 字段时应正确映射。"""
raw = {
"event_date": "2026-01-15T15:00:00Z",
"status": "finished",
"home_team": "Man City",
"away_team": "Man United",
"home_score": 2,
"away_score": 1,
"home_shots": 15,
"away_shots": 8,
"home_shots_on_target": 6,
"away_shots_on_target": 3,
}
m = normalize_bzzoiro(raw, "E0")
assert m is not None
assert m.home_shots == 15
assert m.away_shots == 8
assert m.home_shots_on_target == 6
assert m.away_shots_on_target == 3
def test_maps_corners(self):
"""API 提供 corners 字段时应正确映射。"""
raw = {
"event_date": "2026-01-15T15:00:00Z",
"status": "finished",
"home_team": "Liverpool",
"away_team": "Chelsea",
"home_score": 1,
"away_score": 1,
"home_corners": 7,
"away_corners": 4,
}
m = normalize_bzzoiro(raw, "E0")
assert m is not None
assert m.home_corners == 7
assert m.away_corners == 4
def test_maps_possession(self):
"""API 提供 possession 字段时应正确映射。"""
raw = {
"event_date": "2026-01-15T15:00:00Z",
"status": "finished",
"home_team": "Barcelona",
"away_team": "Real Madrid",
"home_score": 2,
"away_score": 0,
"home_possession": 62.5,
}
m = normalize_bzzoiro(raw, "SP1")
assert m is not None
assert m.home_possession == 62.5
def test_maps_xg(self):
"""API 提供 xG 字段时应正确映射。"""
raw = {
"event_date": "2026-01-15T15:00:00Z",
"status": "finished",
"home_team": "Bayern",
"away_team": "Dortmund",
"home_score": 3,
"away_score": 1,
"home_xg": 2.5,
"away_xg": 0.8,
}
m = normalize_bzzoiro(raw, "D1")
assert m is not None
assert m.home_xg == 2.5
assert m.away_xg == 0.8
def test_maps_cards(self):
"""API 提供 cards 字段时应正确映射。"""
raw = {
"event_date": "2026-01-15T15:00:00Z",
"status": "finished",
"home_team": "Arsenal",
"away_team": "Tottenham",
"home_score": 1,
"away_score": 0,
"home_yellow_cards": 2,
"away_yellow_cards": 3,
"home_red_cards": 0,
"away_red_cards": 1,
}
m = normalize_bzzoiro(raw, "E0")
assert m is not None
assert m.home_yellow_cards == 2
assert m.away_yellow_cards == 3
assert m.home_red_cards == 0
assert m.away_red_cards == 1
def test_missing_stats_stay_none(self):
"""API 没有统计字段时应保持 None,不伪造。"""
raw = {
"event_date": "2026-01-15T15:00:00Z",
"status": "finished",
"home_team": "A",
"away_team": "B",
"home_score": 1,
"away_score": 0,
}
m = normalize_bzzoiro(raw, "E0")
assert m is not None
# 无比分数据时不应有统计字段
assert m.home_shots is None
assert m.away_shots is None
assert m.home_xg is None
assert m.away_xg is None
assert m.home_possession is None
assert m.home_corners is None
def test_alternative_field_names(self):
"""API 使用替代字段名时也应正确映射。"""
raw = {
"event_date": "2026-01-15T15:00:00Z",
"status": "finished",
"home_team": "A",
"away_team": "B",
"home_score": 1,
"away_score": 0,
"shots_home": 12,
"shots_away": 6,
"sot_home": 5,
"sot_away": 2,
"corners_home": 8,
"corners_away": 3,
"xg_home": 1.8,
"xg_away": 0.5,
}
m = normalize_bzzoiro(raw, "E0")
assert m is not None
assert m.home_shots == 12
assert m.away_shots == 6
assert m.home_shots_on_target == 5
assert m.away_shots_on_target == 2
assert m.home_corners == 8
assert m.away_corners == 3
assert m.home_xg == 1.8
assert m.away_xg == 0.5
class TestIngestionCondition:
"""入库条件应不再强制要求 xG。"""
def test_any_stat_field_triggers_stats_creation(self):
"""任一统计字段存在即可触发 MatchStats 创建。"""
# 模拟 normalize 后的结果
nm = NormalizedMatch(
league_type="E0",
date=datetime(2026, 1, 15, 15, 0, tzinfo=timezone.utc),
home_team="A",
away_team="B",
match_status="finished",
home_goals=1,
away_goals=0,
# 只有 shots,无 xG
home_shots=10,
away_shots=5,
)
# 验证:任一统计字段存在
has_stats = (
nm.home_xg is not None or nm.away_xg is not None
or nm.home_shots is not None or nm.away_shots is not None
or nm.home_corners is not None or nm.away_corners is not None
or nm.home_possession is not None
)
assert has_stats is True, "有 shots 时应视为有统计数据"
def test_no_stats_means_no_match_stats(self):
"""没有任何统计字段时不应创建 MatchStats。"""
nm = NormalizedMatch(
league_type="E0",
date=datetime(2026, 1, 15, 15, 0, tzinfo=timezone.utc),
home_team="A",
away_team="B",
match_status="finished",
home_goals=1,
away_goals=0,
)
has_stats = (
nm.home_xg is not None or nm.away_xg is not None
or nm.home_shots is not None or nm.away_shots is not None
or nm.home_corners is not None or nm.away_corners is not None
or nm.home_possession is not None
)
assert has_stats is False, "无统计字段时不应创建 MatchStats"