hockey_new/tests/test_build97_mapping_chunked_transport.py

from __future__ import annotations

import asyncio
from pathlib import Path

from fastapi import HTTPException

from hockey_data.agent_bridge import MAPPING_BATCH_MAX_COMMANDS, VmixAgentHub
from tests.support import LocalTestDatabase


def _entry(index: int, input_ref: str, *, value_size: int = 4) -> dict:
    return {
        "data_key": f"source.row.{index}",
        "command": {
            "Function": "SetText",
            "Input": input_ref,
            "SelectedName": f"Field{index}.Text",
            "Value": "X" * value_size,
        },
    }


def test_mapping_chunk_planner_groups_by_input_and_caps_packets(tmp_path: Path) -> None:
    database = LocalTestDatabase(tmp_path / "mapping-chunks.sqlite3")
    database.create_all()
    hub = VmixAgentHub(database)  # type: ignore[arg-type]

    entries = [*[_entry(i, "INPUT-A") for i in range(95)], *[_entry(100 + i, "INPUT-B") for i in range(17)]]
    chunks = hub._mapping_batch_chunks(entries)

    assert len(chunks) == 4  # A: 40 + 40 + 15, B: 17
    assert all(len(chunk) <= MAPPING_BATCH_MAX_COMMANDS for chunk in chunks)
    assert all(len({item[1]["command"]["Input"] for item in chunk}) == 1 for chunk in chunks)
    # ACK positions can still be restored to original Mapping order.
    assert [index for chunk in chunks for index, _entry_item in chunk] == list(range(len(entries)))


def test_mapping_chunk_transport_sends_300_plus_links_in_small_batches(tmp_path: Path) -> None:
    database = LocalTestDatabase(tmp_path / "mapping-300.sqlite3")
    database.create_all()
    hub = VmixAgentHub(database)  # type: ignore[arg-type]
    packets: list[list[dict]] = []

    async def fake_batch(device_id: str, *, assignment_id: str, match_id: str, commands: list[dict], timeout: float = 8.0) -> dict:
        assert device_id == "GFX-MAP-300"
        packets.append(commands)
        return {"ok": True, "results": [{"ok": True} for _ in commands]}

    async def scenario() -> None:
        hub.send_vmix_batch = fake_batch  # type: ignore[method-assign]
        entries: list[dict] = []
        # 12 Inputs x 31 commands = 372 Mapping commands.
        for input_index in range(12):
            entries.extend(_entry(input_index * 1000 + row, f"INPUT-{input_index:02d}") for row in range(31))
        result = await hub._send_mapping_entries_resilient(
            "GFX-MAP-300",
            assignment_id="assignment-300",
            match_id="game-300",
            entries=entries,
            use_batch=True,
        )
        assert all(item["ok"] for item in result["results"])
        assert result["input_groups"] == 12
        assert result["chunks_total"] == 12
        assert result["chunks_ok"] == 12
        assert result["chunks_failed"] == 0
        assert result["fallback_commands"] == 0
        assert sum(len(packet) for packet in packets) == 372
        assert max(len(packet) for packet in packets) <= MAPPING_BATCH_MAX_COMMANDS
        assert all(len({command["Input"] for command in packet}) == 1 for packet in packets)

    asyncio.run(scenario())


def test_mapping_transport_splits_timeout_batch_and_keeps_going(tmp_path: Path) -> None:
    database = LocalTestDatabase(tmp_path / "mapping-retry.sqlite3")
    database.create_all()
    hub = VmixAgentHub(database)  # type: ignore[arg-type]
    packet_sizes: list[int] = []

    async def flaky_batch(device_id: str, *, assignment_id: str, match_id: str, commands: list[dict], timeout: float = 8.0) -> dict:
        packet_sizes.append(len(commands))
        if len(commands) > 10:
            raise HTTPException(status_code=504, detail="simulated batch timeout")
        return {"ok": True, "results": [{"ok": True} for _ in commands]}

    async def scenario() -> None:
        hub.send_vmix_batch = flaky_batch  # type: ignore[method-assign]
        entries = [_entry(i, "ONE-BIG-INPUT") for i in range(35)]
        result = await hub._send_mapping_entries_resilient(
            "GFX-MAP-RETRY",
            assignment_id="assignment-retry",
            match_id="game-retry",
            entries=entries,
            use_batch=True,
        )
        assert all(item["ok"] for item in result["results"])
        assert result["chunks_total"] == 1
        assert result["chunks_ok"] == 1
        assert result["chunks_failed"] == 0
        assert result["retries"] > 0
        assert packet_sizes[0] == 35
        assert any(size <= 10 for size in packet_sizes[1:])

    asyncio.run(scenario())


def test_mapping_transport_retries_isolated_failed_command(tmp_path: Path) -> None:
    database = LocalTestDatabase(tmp_path / "mapping-single-fallback.sqlite3")
    database.create_all()
    hub = VmixAgentHub(database)  # type: ignore[arg-type]
    singles: list[dict] = []

    async def partial_batch(device_id: str, *, assignment_id: str, match_id: str, commands: list[dict], timeout: float = 8.0) -> dict:
        results = [{"ok": True} for _ in commands]
        results[3] = {"ok": False, "reason": "temporary failure"}
        return {"ok": False, "results": results}

    async def single_ok(device_id: str, *, assignment_id: str, match_id: str, command: dict, timeout: float = 5.0) -> dict:
        singles.append(command)
        return {"ok": True}

    async def scenario() -> None:
        hub.send_vmix_batch = partial_batch  # type: ignore[method-assign]
        hub.send_vmix_command = single_ok  # type: ignore[method-assign]
        entries = [_entry(i, "INPUT-FALLBACK") for i in range(8)]
        result = await hub._send_mapping_entries_resilient(
            "GFX-MAP-FALLBACK",
            assignment_id="assignment-fallback",
            match_id="game-fallback",
            entries=entries,
            use_batch=True,
        )
        assert all(item["ok"] for item in result["results"])
        assert result["fallback_commands"] == 1
        assert result["retries"] == 1
        assert len(singles) == 1
        assert singles[0]["SelectedName"] == "Field3.Text"

    asyncio.run(scenario())


def test_build97_runtime_version_and_mapping_diagnostics() -> None:
    root = Path(__file__).resolve().parents[1]
    app = (root / "app.py").read_text(encoding="utf-8")
    bridge = (root / "hockey_data" / "agent_bridge.py").read_text(encoding="utf-8")
    admin = (root / "hockey_data" / "static" / "admin-directories.js").read_text(encoding="utf-8")

    assert 'BUILD_VERSION = "2026.08.21.1"' in app
    assert 'result["transport"] = "chunked_batch" if use_batch else "legacy"' in bridge
    assert 'MAPPING_BATCH_MAX_COMMANDS = 40' in bridge
    assert 'mappingApplyTransportSummary' in admin
    assert 'пакетов ${chunksApplied}/${chunksTotal}' in admin