250 lines
8.1 KiB
Python
250 lines
8.1 KiB
Python
import json
|
|
import sys
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
if str(ROOT) not in sys.path:
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
import match_odds
|
|
from match_odds import MatchOddsRepository, OddsGameListUnavailable, parse_game_list_xml, run
|
|
|
|
|
|
SAMPLE_GAME_LIST_XML = """<?xml version="1.0" encoding="UTF-8"?>
|
|
<serverresponse sip="14.39">
|
|
<dataCount>1</dataCount>
|
|
<totalDataCount>1</totalDataCount>
|
|
<pageCount>1</pageCount>
|
|
<ec id="ec11290788" hasEC="Y" myGame="ft">
|
|
<game id="8892295">
|
|
<GID>8892295</GID>
|
|
<DATETIME>07-09 04:00p</DATETIME>
|
|
<LID>104036</LID>
|
|
<LEAGUE>世界杯2026(美加墨)</LEAGUE>
|
|
<GIDM>7972917</GIDM>
|
|
<GNUM_H>402008</GNUM_H>
|
|
<GNUM_C>402007</GNUM_C>
|
|
<TEAM_H_ID>122821</TEAM_H_ID>
|
|
<TEAM_C_ID>122841</TEAM_C_ID>
|
|
<TEAM_H>法国</TEAM_H>
|
|
<TEAM_C>摩洛哥</TEAM_C>
|
|
<STRONG>H</STRONG>
|
|
<MORE>162</MORE>
|
|
<ECID>11290788</ECID>
|
|
<RATIO_R>1</RATIO_R>
|
|
<IOR_RH>1.030</IOR_RH>
|
|
<IOR_RC>0.860</IOR_RC>
|
|
<RATIO_OUO>2.5</RATIO_OUO>
|
|
<RATIO_OUU>2.5</RATIO_OUU>
|
|
<IOR_OUH>0.850</IOR_OUH>
|
|
<IOR_OUC>1.030</IOR_OUC>
|
|
<IOR_MH>1.57</IOR_MH>
|
|
<IOR_MC>6.60</IOR_MC>
|
|
<IOR_MN>3.90</IOR_MN>
|
|
<RATIO_HR>0 / 0.5</RATIO_HR>
|
|
<IOR_HRH>0.810</IOR_HRH>
|
|
<IOR_HRC>1.070</IOR_HRC>
|
|
<IOR_HMH>2.28</IOR_HMH>
|
|
<IOR_HMC>5.90</IOR_HMC>
|
|
<IOR_HMN>2.11</IOR_HMN>
|
|
<SYSTIME>2026-07-09 10:29:39</SYSTIME>
|
|
<IS_RB>N</IS_RB>
|
|
</game>
|
|
</ec>
|
|
</serverresponse>
|
|
"""
|
|
|
|
|
|
class FakeCursor:
|
|
def __init__(self):
|
|
self.executed = []
|
|
self.rowcount = 1
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc, tb):
|
|
return False
|
|
|
|
def execute(self, sql, params=None):
|
|
self.executed.append((sql, params))
|
|
|
|
|
|
class FakeConnection:
|
|
def __init__(self):
|
|
self.cursor_instance = FakeCursor()
|
|
self.commit_count = 0
|
|
|
|
def cursor(self):
|
|
return self.cursor_instance
|
|
|
|
def commit(self):
|
|
self.commit_count += 1
|
|
|
|
|
|
class MatchOddsTest(unittest.TestCase):
|
|
def _build_repo(self, prefix="la_"):
|
|
repo = object.__new__(MatchOddsRepository)
|
|
repo.prefix = prefix
|
|
repo.conn = FakeConnection()
|
|
return repo
|
|
|
|
def test_parse_game_list_xml_extracts_main_markets_and_raw_data(self):
|
|
items = parse_game_list_xml(
|
|
SAMPLE_GAME_LIST_XML,
|
|
source_site="hg",
|
|
source_url="https://mos022.com/",
|
|
sport_type="FT",
|
|
show_type="today",
|
|
odds_type="H",
|
|
)
|
|
|
|
self.assertEqual(len(items), 1)
|
|
item = items[0]
|
|
self.assertEqual(item["source_match_id"], "8892295")
|
|
self.assertEqual(item["league_name"], "世界杯2026(美加墨)")
|
|
self.assertEqual(item["home_team"], "法国")
|
|
self.assertEqual(item["away_team"], "摩洛哥")
|
|
self.assertEqual(item["match_time_text"], "07-09 04:00p")
|
|
self.assertEqual(item["handicap_line"], "1")
|
|
self.assertEqual(item["handicap_home_odds"], "1.030")
|
|
self.assertEqual(item["handicap_away_odds"], "0.860")
|
|
self.assertEqual(item["total_line"], "2.5")
|
|
self.assertEqual(item["total_over_odds"], "1.030")
|
|
self.assertEqual(item["total_under_odds"], "0.850")
|
|
self.assertEqual(item["home_win_odds"], "1.57")
|
|
self.assertEqual(item["away_win_odds"], "6.60")
|
|
self.assertEqual(item["draw_odds"], "3.90")
|
|
self.assertEqual(item["raw_data"]["GID"], "8892295")
|
|
self.assertEqual(item["markets"]["first_half"]["home_win"], "2.28")
|
|
|
|
def test_repository_upserts_latest_odds_by_source_match_id(self):
|
|
items = parse_game_list_xml(
|
|
SAMPLE_GAME_LIST_XML,
|
|
source_site="hg",
|
|
source_url="https://mos022.com/",
|
|
sport_type="FT",
|
|
show_type="today",
|
|
odds_type="H",
|
|
)
|
|
repo = self._build_repo()
|
|
|
|
saved = repo.upsert_latest(items, now_ts=1783607379)
|
|
|
|
self.assertEqual(saved, 1)
|
|
self.assertEqual(repo.conn.commit_count, 1)
|
|
sql, params = repo.conn.cursor_instance.executed[0]
|
|
self.assertIn("INSERT INTO `la_match_odds`", sql)
|
|
self.assertIn("ON DUPLICATE KEY UPDATE", sql)
|
|
self.assertIn("`source_match_id`", sql)
|
|
self.assertEqual(params[0], "hg")
|
|
self.assertEqual(params[2], "8892295")
|
|
self.assertEqual(json.loads(params[-3])["GID"], "8892295")
|
|
self.assertEqual(json.loads(params[-2])["moneyline"]["draw"], "3.90")
|
|
|
|
def test_run_skips_without_credentials(self):
|
|
result = run({"prefix": "la_"}, env={"TITAN_ODDS_ENABLED": "0"})
|
|
|
|
self.assertFalse(result["success"])
|
|
self.assertEqual(result["candidate_count"], 0)
|
|
self.assertIn("ODDS_USERNAME", result["error"])
|
|
|
|
def test_run_returns_structured_failure_when_database_is_unavailable(self):
|
|
class FakeClient:
|
|
base_url = "https://mos022.com/"
|
|
|
|
def __init__(self, **kwargs):
|
|
pass
|
|
|
|
def login(self):
|
|
return None
|
|
|
|
def fetch_game_list(self, sport_type, show_type):
|
|
return parse_game_list_xml(
|
|
SAMPLE_GAME_LIST_XML,
|
|
source_site="hg",
|
|
source_url="https://mos022.com/",
|
|
sport_type=sport_type,
|
|
show_type=show_type,
|
|
odds_type="H",
|
|
)
|
|
|
|
with mock.patch.object(match_odds, "HgOddsClient", FakeClient):
|
|
with mock.patch.object(match_odds, "MatchOddsRepository", side_effect=RuntimeError("db down")):
|
|
result = run(
|
|
{"prefix": "la_"},
|
|
env={
|
|
"ODDS_USERNAME": "u",
|
|
"ODDS_PASSWORD": "p",
|
|
"ODDS_SPORT_TYPES": "FT",
|
|
"ODDS_SHOW_TYPES": "today",
|
|
"ODDS_REQUEST_DELAY_SECONDS": "0",
|
|
"TITAN_ODDS_ENABLED": "0",
|
|
},
|
|
)
|
|
|
|
self.assertFalse(result["success"])
|
|
self.assertEqual(result["candidate_count"], 1)
|
|
self.assertIn("db down", result["error"])
|
|
|
|
def test_run_skips_unavailable_game_list_combinations_without_failing(self):
|
|
class FakeClient:
|
|
base_url = "https://mos022.com/"
|
|
|
|
def __init__(self, **kwargs):
|
|
pass
|
|
|
|
def login(self):
|
|
return None
|
|
|
|
def fetch_game_list(self, sport_type, show_type):
|
|
if show_type == "early":
|
|
raise OddsGameListUnavailable("早盘日期不可用")
|
|
return parse_game_list_xml(
|
|
SAMPLE_GAME_LIST_XML,
|
|
source_site="hg",
|
|
source_url="https://mos022.com/",
|
|
sport_type=sport_type,
|
|
show_type=show_type,
|
|
odds_type="H",
|
|
)
|
|
|
|
class FakeRepo:
|
|
def __init__(self, db_config):
|
|
pass
|
|
|
|
def ensure_table(self):
|
|
pass
|
|
|
|
def upsert_latest(self, items, now_ts=None):
|
|
return len(items)
|
|
|
|
def close(self):
|
|
pass
|
|
|
|
with mock.patch.object(match_odds, "HgOddsClient", FakeClient):
|
|
with mock.patch.object(match_odds, "MatchOddsRepository", FakeRepo):
|
|
result = run(
|
|
{"prefix": "la_"},
|
|
env={
|
|
"ODDS_USERNAME": "u",
|
|
"ODDS_PASSWORD": "p",
|
|
"ODDS_SPORT_TYPES": "FT",
|
|
"ODDS_SHOW_TYPES": "today,early",
|
|
"ODDS_REQUEST_DELAY_SECONDS": "0",
|
|
"TITAN_ODDS_ENABLED": "0",
|
|
},
|
|
)
|
|
|
|
self.assertTrue(result["success"])
|
|
self.assertEqual(result["candidate_count"], 1)
|
|
self.assertEqual(result["saved_count"], 1)
|
|
self.assertEqual(result["skipped_count"], 1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|