217 lines
8.8 KiB
Python
217 lines
8.8 KiB
Python
from __future__ import annotations
|
||
|
||
import tempfile
|
||
import unittest
|
||
from pathlib import Path
|
||
|
||
from hockey_data.client import Stat2TVNotFoundError, Stat2TVResponse
|
||
from hockey_data.config import HockeySettingsStore
|
||
from hockey_data.models import Tournament
|
||
from hockey_data.service import HockeyDataService
|
||
from support import LocalTestDatabase
|
||
from hockey_data.tournament_statistics_parser import parse_tournament_statistics_xml
|
||
|
||
|
||
POWERPLAY_XML = """<?xml version="1.0" encoding="UTF-8"?>
|
||
<PowerPlay idtournament="1370" genDate="2026-07-16 14:00:00">
|
||
<Teams name="Большинство" nameen="Power play">
|
||
<Team id="11" name="Команда А" nameen="Team A" rank="1" gp="10" ppg="8" ppo="30" pp_pct="26.7" />
|
||
<Team id="22" name="Команда Б" nameen="Team B" rank="2" gp="10" ppg="6" ppo="31" pp_pct="19.4" />
|
||
</Teams>
|
||
</PowerPlay>
|
||
"""
|
||
|
||
RANK_XML = """<?xml version="1.0" encoding="UTF-8"?>
|
||
<Rank idtournament="1370" genDate="2026-07-16 14:01:00">
|
||
<Category id="points" name="Бомбардиры" nameen="Points leaders">
|
||
<Player id="101" rank="1" number="9" name="Иванов Иван" nameen="Ivan Ivanov" team="Команда А" teamen="Team A" gp="10" g="5" a="7" pts="12" />
|
||
</Category>
|
||
<Category id="goals" name="Снайперы" nameen="Goal leaders">
|
||
<Player id="202" rank="1" number="17" name="Петров Пётр" nameen="Pyotr Petrov" team="Команда Б" teamen="Team B" gp="10" g="8" />
|
||
</Category>
|
||
</Rank>
|
||
"""
|
||
|
||
RANK_NESTED_XML = """<?xml version="1.0" encoding="UTF-8"?>
|
||
<Rankings tournamentId="1371">
|
||
<Category id="assists" title="Ассистенты" title_en="Assist leaders">
|
||
<Player id="303" rank="1" number="44">
|
||
<Name>Сидоров Алексей</Name>
|
||
<NameEn>Alexei Sidorov</NameEn>
|
||
<Team>Команда В</Team>
|
||
<TeamEn>Team C</TeamEn>
|
||
<Stat name="GP" value="5" />
|
||
<Stat name="A" value="9" />
|
||
<Stat name="PTS" value="10" />
|
||
</Player>
|
||
</Category>
|
||
</Rankings>
|
||
"""
|
||
|
||
EMPTY_POWERPLAY_XML = """<?xml version="1.0" encoding="UTF-8"?>
|
||
<PowerPlay idtournament="1372" genDate="2026-07-16 14:02:00"><Teams /></PowerPlay>
|
||
"""
|
||
|
||
RANK_WRAPPED_STATS_XML = """<?xml version="1.0" encoding="UTF-8"?>
|
||
<Rank idTournament="1373">
|
||
<Category id="goalkeepers" name="Вратари" nameEn="Goalkeepers">
|
||
<Player id="404" rank="1" name="Орлов Максим" nameEn="Maxim Orlov" team="Команда Г" teamEn="Team D">
|
||
<Statistics>
|
||
<GP>7</GP>
|
||
<Wins>6</Wins>
|
||
<SavePct>94.2</SavePct>
|
||
</Statistics>
|
||
</Player>
|
||
</Category>
|
||
</Rank>
|
||
"""
|
||
|
||
|
||
class TournamentStatisticsParserTests(unittest.TestCase):
|
||
def test_parses_powerplay_team_rows(self) -> None:
|
||
result = parse_tournament_statistics_xml(
|
||
POWERPLAY_XML, resource_type="powerplay", language="ru"
|
||
)
|
||
|
||
self.assertTrue(result["available"])
|
||
self.assertEqual(result["tournament_id"], "1370")
|
||
self.assertEqual(len(result["sections"]), 1)
|
||
first = result["sections"][0]["rows"][0]
|
||
self.assertEqual(first["name"], "Команда А")
|
||
self.assertEqual(first["rank"], "1")
|
||
self.assertEqual(first["values"]["ppg"], "8")
|
||
self.assertEqual(
|
||
[column["key"] for column in result["sections"][0]["columns"]],
|
||
["gp", "ppg", "ppo", "pp_pct"],
|
||
)
|
||
|
||
def test_parses_rank_categories_and_english_names(self) -> None:
|
||
result = parse_tournament_statistics_xml(
|
||
RANK_XML, resource_type="rank", language="en"
|
||
)
|
||
|
||
self.assertTrue(result["available"])
|
||
self.assertEqual([section["id"] for section in result["sections"]], ["points", "goals"])
|
||
self.assertEqual(result["sections"][0]["label"], "Points leaders")
|
||
self.assertEqual(result["sections"][0]["rows"][0]["name"], "Ivan Ivanov")
|
||
self.assertEqual(result["sections"][0]["rows"][0]["team"], "Team A")
|
||
|
||
def test_parses_nested_stat_fields_without_technical_duplicate_rows(self) -> None:
|
||
result = parse_tournament_statistics_xml(
|
||
RANK_NESTED_XML, resource_type="rank", language="ru"
|
||
)
|
||
|
||
self.assertTrue(result["available"])
|
||
self.assertEqual(len(result["sections"]), 1)
|
||
self.assertEqual(len(result["sections"][0]["rows"]), 1)
|
||
row = result["sections"][0]["rows"][0]
|
||
self.assertEqual(row["name"], "Сидоров Алексей")
|
||
self.assertEqual(row["values"], {"gp": "5", "a": "9", "pts": "10"})
|
||
self.assertEqual(row["source_tag"], "Player")
|
||
|
||
def test_parses_stats_wrapped_in_container(self) -> None:
|
||
result = parse_tournament_statistics_xml(
|
||
RANK_WRAPPED_STATS_XML, resource_type="rank", language="en"
|
||
)
|
||
|
||
self.assertTrue(result["available"])
|
||
self.assertEqual(result["tournament_id"], "1373")
|
||
row = result["sections"][0]["rows"][0]
|
||
self.assertEqual(row["name"], "Maxim Orlov")
|
||
self.assertEqual(row["values"]["gp"], "7")
|
||
self.assertEqual(row["values"]["wins"], "6")
|
||
self.assertEqual(row["values"]["save_pct"], "94.2")
|
||
|
||
def test_empty_valid_xml_is_marked_unavailable(self) -> None:
|
||
result = parse_tournament_statistics_xml(
|
||
EMPTY_POWERPLAY_XML, resource_type="powerplay", language="ru"
|
||
)
|
||
self.assertFalse(result["available"])
|
||
self.assertEqual(result["sections"], [])
|
||
|
||
|
||
class TournamentStatisticsCachingTests(unittest.IsolatedAsyncioTestCase):
|
||
async def test_mixed_optional_resources_are_cached_without_repeated_requests(self) -> None:
|
||
with tempfile.TemporaryDirectory() as temp_dir:
|
||
base_dir = Path(temp_dir)
|
||
database = LocalTestDatabase(base_dir / "test.sqlite3")
|
||
database.create_all()
|
||
settings = HockeySettingsStore(base_dir / "settings")
|
||
service = HockeyDataService(
|
||
base_dir=base_dir,
|
||
database=database,
|
||
settings=settings,
|
||
)
|
||
with database.session() as session:
|
||
session.add(Tournament(external_id="1370"))
|
||
|
||
calls: list[str] = []
|
||
|
||
async def fake_fetch(endpoint: str) -> Stat2TVResponse:
|
||
calls.append(endpoint)
|
||
if "powerplay-1370.xml" in endpoint:
|
||
return Stat2TVResponse(
|
||
content=POWERPLAY_XML.encode("utf-8"),
|
||
status_code=200,
|
||
content_type="application/xml",
|
||
auth_mode="test",
|
||
url=f"https://example.test/{endpoint}",
|
||
)
|
||
raise Stat2TVNotFoundError(
|
||
f"https://example.test/{endpoint}", status_code=404
|
||
)
|
||
|
||
service.client.fetch_xml = fake_fetch # type: ignore[method-assign]
|
||
|
||
first = await service.sync_tournament_statistics("1370")
|
||
second = await service.sync_tournament_statistics("1370")
|
||
|
||
self.assertTrue(first["statistics"]["powerplay"]["available"])
|
||
self.assertFalse(first["statistics"]["rank"]["available"])
|
||
self.assertEqual(first["statistics"]["available_tabs"], ["powerplay"])
|
||
self.assertEqual(len(calls), 2)
|
||
self.assertTrue(second["resources"]["powerplay"]["skipped"])
|
||
self.assertTrue(second["resources"]["rank"]["skipped"])
|
||
self.assertEqual(len(calls), 2)
|
||
database.engine.dispose()
|
||
|
||
async def test_valid_empty_xml_uses_negative_cache(self) -> None:
|
||
with tempfile.TemporaryDirectory() as temp_dir:
|
||
base_dir = Path(temp_dir)
|
||
database = LocalTestDatabase(base_dir / "test.sqlite3")
|
||
database.create_all()
|
||
settings = HockeySettingsStore(base_dir / "settings")
|
||
service = HockeyDataService(
|
||
base_dir=base_dir,
|
||
database=database,
|
||
settings=settings,
|
||
)
|
||
with database.session() as session:
|
||
session.add(Tournament(external_id="1372"))
|
||
|
||
calls: list[str] = []
|
||
|
||
async def fake_fetch(endpoint: str) -> Stat2TVResponse:
|
||
calls.append(endpoint)
|
||
return Stat2TVResponse(
|
||
content=EMPTY_POWERPLAY_XML.encode("utf-8"),
|
||
status_code=200,
|
||
content_type="application/xml",
|
||
auth_mode="test",
|
||
url=f"https://example.test/{endpoint}",
|
||
)
|
||
|
||
service.client.fetch_xml = fake_fetch # type: ignore[method-assign]
|
||
|
||
await service.sync_tournament_statistics("1372")
|
||
second = await service.sync_tournament_statistics("1372")
|
||
|
||
self.assertEqual(len(calls), 2) # powerplay + rank only on first sync
|
||
self.assertTrue(second["resources"]["powerplay"]["skipped"])
|
||
self.assertTrue(second["resources"]["rank"]["skipped"])
|
||
database.engine.dispose()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|