193 lines
11 KiB
Python
193 lines
11 KiB
Python
from pathlib import Path
|
||
from types import SimpleNamespace
|
||
|
||
import pytest
|
||
|
||
from hockey_data.auth_bridge import HockeyUser
|
||
from hockey_data.mapping_context import MappingDataService
|
||
from hockey_data.models import Game, Player, PlayerTournament, Tournament
|
||
from tests.support import LocalTestDatabase
|
||
|
||
|
||
def _seed(database: LocalTestDatabase) -> None:
|
||
with database.session() as session:
|
||
session.add(Tournament(external_id="1437", level="khl", name_ru="Предсезонные матчи", season="2026/27"))
|
||
session.add(Game(
|
||
external_id="902918", tournament_external_id="1437", home_team_external_id="23", away_team_external_id="17",
|
||
home_team_name_ru="СКА", home_team_short_ru="СКА", home_team_city_ru="Санкт-Петербург", home_score=2,
|
||
away_team_name_ru="Динамо М", away_team_short_ru="ДИН", away_team_city_ru="Москва", away_score=1,
|
||
status="live", arena_ru="Ледовый дворец", start_time="19:30",
|
||
))
|
||
player = Player(external_id="18451", first_name_ru="Иван", last_name_ru="Иванов", full_name_ru="Иванов Иван", position_ru="Нападающий", photo_url="ivanov.png")
|
||
session.add(player); session.flush()
|
||
session.add(PlayerTournament(player_id=player.id, tournament_external_id="1437", club_external_id="23", team_name_ru="СКА", jersey_number="17"))
|
||
|
||
|
||
def test_context_variables_and_sql_catalog_are_dynamic(tmp_path: Path) -> None:
|
||
database = LocalTestDatabase(tmp_path / "mapping-context.sqlite3")
|
||
database.create_all(); _seed(database)
|
||
service = MappingDataService(database) # type: ignore[arg-type]
|
||
admin = HockeyUser(id="1", login="admin", display_name="Admin", is_admin=True, role="admin")
|
||
|
||
variables = service.list_variables()["variables"]
|
||
assert any(row["key"] == "game_id" and row["is_system"] for row in variables)
|
||
assert any(row["key"] == "highlighted_game_id" for row in variables)
|
||
|
||
custom = service.create_variable(SimpleNamespace(
|
||
key="my_special_player_id", label="Мой выделенный игрок", category="Пользовательские", description="",
|
||
value_type="id", entity_type="player", scope="match", source_type="manual", default_value="", enabled=True, sort_order=1200,
|
||
), admin)
|
||
assert custom["key"] == "my_special_player_id"
|
||
set_value = service.set_context_value("selected_player_id", "18451", admin, {"game_id": "902918", "session_token": "S1"})
|
||
assert set_value["value"] == "18451"
|
||
|
||
catalog = service.data_catalog(admin, {"game_id": "902918", "session_token": "S1"})
|
||
values = {item["key"]: item["value"] for item in catalog["items"]}
|
||
assert values["game.home.name"] == "СКА"
|
||
assert values["game.away.score"] == 1
|
||
assert values["selected_player.full_name"] == "Иванов Иван"
|
||
assert catalog["context"]["team1_id"] == "23"
|
||
assert catalog["context"]["team2_id"] == "17"
|
||
|
||
source = service.create_source(SimpleNamespace(
|
||
code="custom_demo", name="Свои данные", category="Тест", description="", sql_text='SELECT :game_id AS "game_ref", :my_special_player_id AS "special_player"',
|
||
field_metadata={"game_ref": {"label": "Матч"}, "special_player": {"label": "Игрок"}}, enabled=True, sort_order=1500,
|
||
), admin)
|
||
assert source["code"] == "custom_demo"
|
||
service.set_context_value("my_special_player_id", "18451", admin, {"game_id": "902918", "session_token": "S1"})
|
||
catalog2 = service.data_catalog(admin, {"game_id": "902918", "session_token": "S1"})
|
||
values2 = {item["key"]: item["value"] for item in catalog2["items"]}
|
||
assert values2["custom_demo.game_ref"] == "902918"
|
||
assert values2["custom_demo.special_player"] == "18451"
|
||
|
||
|
||
def test_sql_sources_reject_writes(tmp_path: Path) -> None:
|
||
database = LocalTestDatabase(tmp_path / "mapping-context-danger.sqlite3")
|
||
database.create_all()
|
||
service = MappingDataService(database) # type: ignore[arg-type]
|
||
admin = HockeyUser(id="1", login="admin", display_name="Admin", is_admin=True, role="admin")
|
||
with pytest.raises(Exception):
|
||
service.create_source(SimpleNamespace(
|
||
code="bad", name="bad", category="", description="", sql_text="DELETE FROM hockey_games", field_metadata={}, enabled=True, sort_order=1,
|
||
), admin)
|
||
|
||
|
||
def test_sql_table_source_exposes_every_cell_without_changing_query(tmp_path: Path) -> None:
|
||
database = LocalTestDatabase(tmp_path / "mapping-table.sqlite3")
|
||
database.create_all(); _seed(database)
|
||
service = MappingDataService(database) # type: ignore[arg-type]
|
||
admin = HockeyUser(id="1", login="admin", display_name="Admin", is_admin=True, role="admin")
|
||
|
||
service.create_source(SimpleNamespace(
|
||
code="score_rows", name="Таблица строк", category="Таблица", description="",
|
||
sql_text='SELECT 1 AS "place", \'СКА\' AS "team", 10 AS "points" UNION ALL SELECT 2, \'Динамо\', 8 UNION ALL SELECT 3, \'Локомотив\', 7',
|
||
field_metadata={"place": {"label": "Место"}, "team": {"label": "Команда"}, "points": {"label": "Очки"}}, enabled=True, sort_order=1600,
|
||
), admin)
|
||
catalog = service.data_catalog(admin, {"game_id": "902918", "session_token": "S1"})
|
||
values = {item["key"]: item["value"] for item in catalog["items"]}
|
||
# Backwards-compatible source.column means the first row.
|
||
assert values["score_rows.team"] == "СКА"
|
||
# New addressable cell keys let Mapping pick any row/column from one SQL table.
|
||
assert values["score_rows.row.2.team"] == "Динамо"
|
||
assert values["score_rows.row.3.points"] == 7
|
||
table = next(item for item in catalog["tables"] if item["code"] == "score_rows")
|
||
assert table["row_count"] == 3
|
||
assert [column["key"] for column in table["columns"]] == ["place", "team", "points"]
|
||
assert table["rows"][1]["values"]["team"] == "Динамо"
|
||
|
||
|
||
def test_language_suffix_pairs_create_virtual_auto_key(tmp_path: Path) -> None:
|
||
database = LocalTestDatabase(tmp_path / "mapping-language.sqlite3")
|
||
database.create_all(); _seed(database)
|
||
service = MappingDataService(database) # type: ignore[arg-type]
|
||
admin = HockeyUser(id="1", login="admin", display_name="Admin", is_admin=True, role="admin")
|
||
|
||
service.create_source(SimpleNamespace(
|
||
code="bilingual", name="Два языка", category="Тест", description="",
|
||
sql_text='SELECT \'17 августа 2026\' AS "date_RUS", \'August 17, 2026\' AS "date_ENG", \'СКА\' AS "team_RUS", \'SKA\' AS "team_ENG"',
|
||
field_metadata={}, enabled=True, sort_order=1700,
|
||
), admin)
|
||
|
||
ru = service.data_catalog(admin, {"game_id": "902918", "session_token": "S1", "ui_language": "ru"})
|
||
ru_values = {item["key"]: item["value"] for item in ru["items"]}
|
||
assert ru_values["bilingual.date"] == "17 августа 2026"
|
||
assert ru_values["bilingual.team"] == "СКА"
|
||
ru_item = next(item for item in ru["items"] if item["key"] == "bilingual.date")
|
||
assert ru_item["localized"] is True
|
||
assert ru_item["resolved_column"] == "date_RUS"
|
||
|
||
en = service.data_catalog(admin, {"game_id": "902918", "session_token": "S1", "ui_language": "en"})
|
||
en_values = {item["key"]: item["value"] for item in en["items"]}
|
||
assert en_values["bilingual.date"] == "August 17, 2026"
|
||
assert en_values["bilingual.team"] == "SKA"
|
||
en_table = next(item for item in en["tables"] if item["code"] == "bilingual")
|
||
assert [column["key"] for column in en_table["columns"]] == ["date", "team"]
|
||
assert all(column["localized"] for column in en_table["columns"])
|
||
|
||
|
||
def test_language_suffix_pairs_work_for_specific_table_rows(tmp_path: Path) -> None:
|
||
database = LocalTestDatabase(tmp_path / "mapping-language-table.sqlite3")
|
||
database.create_all(); _seed(database)
|
||
service = MappingDataService(database) # type: ignore[arg-type]
|
||
admin = HockeyUser(id="1", login="admin", display_name="Admin", is_admin=True, role="admin")
|
||
|
||
service.create_source(SimpleNamespace(
|
||
code="clubs_lang", name="Клубы", category="Тест", description="",
|
||
sql_text='SELECT 1 AS "id", \'СКА\' AS "name_RUS", \'SKA\' AS "name_ENG" UNION ALL SELECT 2, \'Динамо\', \'Dynamo\'',
|
||
field_metadata={}, enabled=True, sort_order=1701,
|
||
), admin)
|
||
catalog = service.data_catalog(admin, {"game_id": "902918", "session_token": "S1", "ui_language": "en"})
|
||
values = {item["key"]: item["value"] for item in catalog["items"]}
|
||
assert values["clubs_lang.row.1.name"] == "SKA"
|
||
assert values["clubs_lang.row.2.name"] == "Dynamo"
|
||
|
||
|
||
def test_sql_allows_comments_and_one_trailing_semicolon_but_not_two_statements(tmp_path: Path) -> None:
|
||
database = LocalTestDatabase(tmp_path / "mapping-sql-normalize.sqlite3")
|
||
database.create_all(); _seed(database)
|
||
service = MappingDataService(database) # type: ignore[arg-type]
|
||
admin = HockeyUser(id="1", login="admin", display_name="Admin", is_admin=True, role="admin")
|
||
|
||
source = service.create_source(SimpleNamespace(
|
||
code="commented", name="SQL comments", category="Тест", description="",
|
||
sql_text='-- дата текущего матча\nSELECT :game_id AS "game_id" /* безопасный комментарий */;',
|
||
field_metadata={}, enabled=True, sort_order=1800,
|
||
), admin)
|
||
assert source["code"] == "commented"
|
||
preview = service.preview_source(source["id"], admin, {"game_id": "902918"})
|
||
assert preview["rows"][0]["game_id"] == "902918"
|
||
|
||
with pytest.raises(Exception):
|
||
service.create_source(SimpleNamespace(
|
||
code="two_queries", name="bad", category="", description="",
|
||
sql_text='SELECT 1; SELECT 2', field_metadata={}, enabled=True, sort_order=1,
|
||
), admin)
|
||
|
||
|
||
def test_sql_context_parser_ignores_postgresql_time_formats_and_casts(tmp_path: Path) -> None:
|
||
database = LocalTestDatabase(tmp_path / "mapping-sql-params.sqlite3")
|
||
database.create_all()
|
||
service = MappingDataService(database) # type: ignore[arg-type]
|
||
params = service.validate_sql(
|
||
"SELECT TO_CHAR(NOW(), 'HH24:MI:SS') AS time_text, value::text AS value_text "
|
||
"FROM hockey_games WHERE external_id = :game_id"
|
||
)
|
||
assert params == {"game_id"}
|
||
|
||
|
||
def test_sql_source_stores_auto_refresh_settings(tmp_path: Path) -> None:
|
||
database = LocalTestDatabase(tmp_path / "mapping-sql-auto.sqlite3")
|
||
database.create_all()
|
||
service = MappingDataService(database) # type: ignore[arg-type]
|
||
admin = HockeyUser(id="1", login="admin", display_name="Admin", is_admin=True, role="admin")
|
||
source = service.create_source(SimpleNamespace(
|
||
code="clock", name="Часы", category="Система", description="",
|
||
sql_text="SELECT 1 AS clock_value", field_metadata={}, enabled=True,
|
||
auto_refresh_enabled=True, refresh_interval_ms=1000, sort_order=10,
|
||
), admin)
|
||
assert source["auto_refresh_enabled"] is True
|
||
assert source["refresh_interval_ms"] == 1000
|
||
listed = next(row for row in service.list_sources()["sources"] if row["code"] == "clock")
|
||
assert listed["auto_refresh_enabled"] is True
|
||
assert listed["refresh_interval_ms"] == 1000
|