PolyWeather/tests/test_scan_terminal_modules.py

from web.scan_terminal_filters import normalize_scan_terminal_filters
from web import scan_terminal_cache
from web import scan_terminal_service
from web.scan_terminal_metar_gate import _apply_metar_gate_to_row
from web.scan_terminal_payloads import (
    build_scan_terminal_incremental_payload,
    build_failed_scan_terminal_payload,
    build_scan_terminal_snapshot_id,
    build_stale_scan_terminal_payload,
    compact_ranked_scan_rows_for_payload,
    SCAN_PAYLOAD_DEFERRED_RUNWAY_POINTS,
    SCAN_PAYLOAD_FULL_RUNWAY_HISTORY_ROWS,
)
from web.scan_terminal_ranker import build_ranked_scan_terminal_result
from web import scan_terminal_city_row
from web.scan_terminal_city_row import _build_quick_row
from web.routers.scan import router as scan_router
from web.scan_terminal_service import _scan_terminal_prewarm_filters


class _FakeRedis:
    def __init__(self):
        self.data = {}

    def get(self, key):
        return self.data.get(key)

    def setex(self, key, _ttl, value):
        self.data[key] = value


def test_scan_terminal_cache_hydrates_success_payload_from_redis(monkeypatch):
    fake_redis = _FakeRedis()
    monkeypatch.setenv("POLYWEATHER_SCAN_TERMINAL_REDIS_CACHE_ENABLED", "true")
    monkeypatch.setattr(scan_terminal_cache, "_get_redis_client", lambda: fake_redis)
    scan_terminal_cache._SCAN_TERMINAL_CACHE.clear()

    filters = {"scan_mode": "tradable", "limit": 9}
    payload = {
        "generated_at": "2026-06-01T00:00:00Z",
        "rows": [{"id": "row-1"}],
        "summary": {"candidate_total": 1},
    }

    scan_terminal_cache.set_cached_scan_terminal_payload(filters, payload)
    scan_terminal_cache._SCAN_TERMINAL_CACHE.clear()

    entry = scan_terminal_cache.get_scan_terminal_cache_entry(filters)
    cached = scan_terminal_cache.get_cached_scan_terminal_payload(filters, ttl_sec=3600)

    assert entry["success_payload"]["rows"] == [{"id": "row-1"}]
    assert cached["summary"]["candidate_total"] == 1


def test_scan_terminal_failure_state_preserves_redis_success_payload(monkeypatch):
    fake_redis = _FakeRedis()
    monkeypatch.setenv("POLYWEATHER_SCAN_TERMINAL_REDIS_CACHE_ENABLED", "true")
    monkeypatch.setattr(scan_terminal_cache, "_get_redis_client", lambda: fake_redis)
    scan_terminal_cache._SCAN_TERMINAL_CACHE.clear()

    filters = {"scan_mode": "tradable", "limit": 9}
    scan_terminal_cache.set_cached_scan_terminal_payload(
        filters,
        {
            "generated_at": "2026-06-01T00:00:00Z",
            "rows": [{"id": "row-1"}],
        },
    )
    scan_terminal_cache._SCAN_TERMINAL_CACHE.clear()

    scan_terminal_cache.set_scan_terminal_failure_state(filters, error_message="timeout")
    scan_terminal_cache._SCAN_TERMINAL_CACHE.clear()

    entry = scan_terminal_cache.get_scan_terminal_cache_entry(filters)

    assert entry["success_payload"]["rows"] == [{"id": "row-1"}]
    assert entry["last_error"] == "timeout"


def test_scan_terminal_cache_keeps_previous_success_snapshot_for_diffs(monkeypatch):
    fake_redis = _FakeRedis()
    monkeypatch.setenv("POLYWEATHER_SCAN_TERMINAL_REDIS_CACHE_ENABLED", "true")
    monkeypatch.setattr(scan_terminal_cache, "_get_redis_client", lambda: fake_redis)
    scan_terminal_cache._SCAN_TERMINAL_CACHE.clear()

    filters = {"scan_mode": "tradable", "limit": 9}
    scan_terminal_cache.set_cached_scan_terminal_payload(
        filters,
        {
            "snapshot_id": "scan-old",
            "generated_at": "2026-06-01T00:00:00Z",
            "rows": [{"id": "row-1", "edge_percent": 3}],
        },
    )
    scan_terminal_cache.set_cached_scan_terminal_payload(
        filters,
        {
            "snapshot_id": "scan-new",
            "generated_at": "2026-06-01T00:01:00Z",
            "rows": [{"id": "row-1", "edge_percent": 4}],
        },
    )

    scan_terminal_cache._SCAN_TERMINAL_CACHE.clear()
    entry = scan_terminal_cache.get_scan_terminal_cache_entry(filters)

    assert entry["success_payload"]["snapshot_id"] == "scan-new"
    assert entry["previous_success_payload"]["snapshot_id"] == "scan-old"


def test_build_scan_terminal_incremental_payload_returns_not_modified():
    filters = {"scan_mode": "tradable", "limit": 2}
    current = {
        "generated_at": "2026-06-01T00:01:00Z",
        "snapshot_id": "scan-current",
        "status": "ready",
        "stale": False,
        "filters": filters,
        "summary": {"candidate_total": 2},
        "top_signal": {"id": "row-1"},
        "rows": [{"id": "row-1"}, {"id": "row-2"}],
    }

    payload = build_scan_terminal_incremental_payload(
        filters=filters,
        current_payload=current,
        since_snapshot_id="scan-current",
        base_payload=current,
    )

    assert payload["status"] == "not_modified"
    assert payload["rows"] == []
    assert payload["summary"] == current["summary"]
    assert payload["top_signal"] == current["top_signal"]
    assert payload["diff"] == {
        "mode": "not_modified",
        "base_snapshot_id": "scan-current",
        "snapshot_id": "scan-current",
        "rows_changed": [],
        "removed_row_ids": [],
    }


def test_build_scan_terminal_incremental_payload_returns_changed_row_delta():
    filters = {"scan_mode": "tradable", "limit": 3}
    base = {
        "generated_at": "2026-06-01T00:00:00Z",
        "snapshot_id": "scan-old",
        "status": "ready",
        "stale": False,
        "filters": filters,
        "summary": {"candidate_total": 2},
        "top_signal": {"id": "row-1"},
        "rows": [
            {"id": "row-1", "rank": 1, "edge_percent": 3},
            {"id": "row-removed", "rank": 2, "edge_percent": 2},
        ],
    }
    current = {
        "generated_at": "2026-06-01T00:01:00Z",
        "snapshot_id": "scan-new",
        "status": "ready",
        "stale": False,
        "filters": filters,
        "summary": {"candidate_total": 2},
        "top_signal": {"id": "row-added"},
        "rows": [
            {"id": "row-1", "rank": 1, "edge_percent": 4},
            {"id": "row-added", "rank": 2, "edge_percent": 5},
        ],
    }

    payload = build_scan_terminal_incremental_payload(
        filters=filters,
        current_payload=current,
        since_snapshot_id="scan-old",
        base_payload=base,
    )

    assert payload["status"] == "ready"
    assert payload["rows"] == []
    assert payload["snapshot_id"] == "scan-new"
    assert payload["diff"]["mode"] == "row_delta"
    assert payload["diff"]["base_snapshot_id"] == "scan-old"
    assert payload["diff"]["snapshot_id"] == "scan-new"
    assert {row["id"] for row in payload["diff"]["rows_changed"]} == {
        "row-1",
        "row-added",
    }
    assert payload["diff"]["removed_row_ids"] == ["row-removed"]


def test_build_scan_terminal_incremental_payload_falls_back_to_full_without_base():
    filters = {"scan_mode": "tradable", "limit": 2}
    current = {
        "generated_at": "2026-06-01T00:01:00Z",
        "snapshot_id": "scan-new",
        "status": "ready",
        "stale": False,
        "filters": filters,
        "summary": {"candidate_total": 1},
        "top_signal": None,
        "rows": [{"id": "row-1", "edge_percent": 4}],
    }

    payload = build_scan_terminal_incremental_payload(
        filters=filters,
        current_payload=current,
        since_snapshot_id="scan-old",
        base_payload=None,
    )

    assert payload["status"] == "ready"
    assert payload["rows"] == current["rows"]
    assert payload["diff"]["mode"] == "full"
    assert payload["diff"]["base_snapshot_id"] == "scan-old"
    assert payload["diff"]["snapshot_id"] == "scan-new"


def test_scan_terminal_prewarm_covers_default_api_limit():
    limits = {filters["limit"] for filters in _scan_terminal_prewarm_filters()}

    assert 25 in limits
    assert 180 in limits


def test_scan_terminal_prewarm_queues_city_refresh_without_analyze(monkeypatch):
    enqueued = []

    class _DB:
        @staticmethod
        def enqueue_observation_refresh_request(**kwargs):
            enqueued.append(kwargs)
            return True

    monkeypatch.setattr(scan_terminal_service, "_SCAN_PREWARM_DB", _DB(), raising=False)
    monkeypatch.setattr(
        scan_terminal_service,
        "_analyze",
        lambda *_args, **_kwargs: (_ for _ in ()).throw(
            AssertionError("scan terminal prewarm must not fetch external sources")
        ),
        raising=False,
    )

    assert scan_terminal_service._queue_scan_terminal_city_prewarm("shenzhen") == "shenzhen"
    assert enqueued == [
        {
            "city": "shenzhen",
            "kind": "panel",
            "priority": "normal",
            "reason": "scan_terminal_prewarm",
        }
    ]


def test_scan_city_terminal_rows_reuses_persisted_panel_cache(monkeypatch):
    payload = {
        "display_name": "Paris",
        "local_date": "2026-06-10",
        "local_time": "12:00",
        "current": {"temp": 18.0},
        "risk": {},
        "deb": {"prediction": 20.0},
        "probabilities": {},
        "multi_model": {},
    }

    class _Cache:
        @staticmethod
        def get_city_cache(kind, city):
            assert (kind, city) == ("panel", "paris")
            return {"payload": payload}

    monkeypatch.setattr(scan_terminal_city_row, "_PANEL_CACHE_DB", _Cache())
    monkeypatch.setattr(
        scan_terminal_city_row,
        "_analyze",
        lambda *_args, **_kwargs: (_ for _ in ()).throw(
            AssertionError("non-force scan must not fetch external sources")
        ),
    )

    result = scan_terminal_city_row._scan_city_terminal_rows(
        "paris",
        {"market_type": "maxtemp"},
        force_refresh=False,
    )

    assert result["city"] == "paris"
    assert result["rows"][0]["current_temp"] == 18.0


def test_scan_city_terminal_rows_uses_canonical_without_analyze(monkeypatch):
    enqueued = []

    class _Cache:
        @staticmethod
        def get_city_cache(kind, city):
            assert (kind, city) == ("panel", "qingdao")
            return None

        @staticmethod
        def get_canonical_temperature(city):
            assert city == "qingdao"
            return {
                "payload": {
                    "city": "qingdao",
                    "value": 22.5,
                    "temp_symbol": "°C",
                    "source": "amsc_awos",
                    "source_label": "AMSC AWOS",
                    "source_role": "settlement_proxy",
                    "observed_at": "2026-06-01T08:00:00Z",
                    "fetched_at": "2026-06-01T08:00:30Z",
                    "freshness_sec": 30,
                    "freshness_status": "fresh",
                    "confidence": 0.92,
                }
            }

        @staticmethod
        def enqueue_observation_refresh_request(**kwargs):
            enqueued.append(kwargs)
            return True

    monkeypatch.setattr(scan_terminal_city_row, "_PANEL_CACHE_DB", _Cache())
    monkeypatch.setattr(
        scan_terminal_city_row,
        "_analyze",
        lambda *_args, **_kwargs: (_ for _ in ()).throw(
            AssertionError("scan terminal must not fetch external sources")
        ),
    )

    result = scan_terminal_city_row._scan_city_terminal_rows(
        "qingdao",
        {"market_type": "maxtemp"},
        force_refresh=False,
    )

    assert result["city"] == "qingdao"
    assert result["candidate_total"] == 1
    assert result["rows"][0]["current_temp"] == 22.5
    assert result["rows"][0]["temp_symbol"] == "°C"
    assert enqueued == [
        {
            "city": "qingdao",
            "kind": "panel",
            "priority": "high",
            "reason": "scan_terminal_canonical_fallback",
        }
    ]


def test_scan_city_terminal_rows_cold_start_only_enqueues(monkeypatch):
    enqueued = []

    class _Cache:
        @staticmethod
        def get_city_cache(kind, city):
            assert (kind, city) == ("panel", "seoul")
            return None

        @staticmethod
        def get_canonical_temperature(city):
            assert city == "seoul"
            return None

        @staticmethod
        def enqueue_observation_refresh_request(**kwargs):
            enqueued.append(kwargs)
            return True

    monkeypatch.setattr(scan_terminal_city_row, "_PANEL_CACHE_DB", _Cache())
    monkeypatch.setattr(
        scan_terminal_city_row,
        "_analyze",
        lambda *_args, **_kwargs: (_ for _ in ()).throw(
            AssertionError("scan terminal must not fetch external sources")
        ),
    )

    result = scan_terminal_city_row._scan_city_terminal_rows(
        "seoul",
        {"market_type": "maxtemp"},
        force_refresh=False,
    )

    assert result == {
        "city": "seoul",
        "rows": [],
        "candidate_total": 0,
        "primary_scores": [],
    }
    assert enqueued == [
        {
            "city": "seoul",
            "kind": "panel",
            "priority": "high",
            "reason": "scan_terminal_cold_start",
        }
    ]


def test_scan_router_does_not_expose_terminal_ai_endpoint():
    routes = {
        getattr(route, "path", None): getattr(route, "methods", set())
        for route in scan_router.routes
    }

    assert "/api/scan/terminal/ai" not in routes


def test_normalize_scan_terminal_filters_clamps_and_swaps_bounds():
    filters = normalize_scan_terminal_filters(
        {
            "min_price": 1.2,
            "max_price": -0.2,
            "limit": 999,
            "high_liquidity_only": True,
            "min_liquidity": 100,
            "timezone_offset_seconds": "28800",
        }
    )

    assert filters["min_price"] == 0.0
    assert filters["max_price"] == 1.0
    assert filters["limit"] == 200
    assert filters["min_liquidity"] == 5000.0
    assert filters["timezone_offset_seconds"] == 28800


def test_ranked_scan_terminal_result_sorts_and_summarizes_unique_markets():
    result = build_ranked_scan_terminal_result(
        city_results=[
            {
                "candidate_total": 2,
                "primary_scores": [80.0],
                "rows": [
                    {
                        "id": "low",
                        "market_key": "m1",
                        "final_score": 70.0,
                        "edge_percent": 4.0,
                        "volume": 100,
                    },
                    {
                        "id": "high",
                        "market_key": "m1",
                        "final_score": 90.0,
                        "edge_percent": 2.0,
                        "volume": 250,
                    },
                ],
            },
            {
                "candidate_total": 1,
                "primary_scores": [60.0],
                "rows": [
                    {
                        "id": "tie-break",
                        "market_key": "m2",
                        "final_score": 90.0,
                        "edge_percent": 5.0,
                        "volume": 300,
                    }
                ],
            },
        ],
        filters={"limit": 2},
        total_city_count=3,
        failed_city_count=1,
    )

    assert [row["id"] for row in result["ranked_rows"]] == ["tie-break", "high"]
    assert [row["rank"] for row in result["ranked_rows"]] == [1, 2]
    assert result["top_signal"]["id"] == "tie-break"
    assert result["summary"]["candidate_total"] == 3
    assert result["summary"]["visible_count"] == 2
    assert result["summary"]["tradable_market_count"] == 2
    assert result["summary"]["total_volume"] == 550
    assert result["summary"]["failed_city_count"] == 1


def test_scan_terminal_payload_helpers_preserve_stale_and_failed_shape():
    success_payload = {
        "generated_at": "2026-04-28T00:00:00Z",
        "snapshot_id": "scan-old",
        "filters": {"scan_mode": "tradable"},
        "rows": [{"id": "row-1"}],
    }

    stale = build_stale_scan_terminal_payload(
        filters={"scan_mode": "trend"},
        success_payload=success_payload,
        error_message="refresh failed",
        failed_at="2026-04-28T00:01:00Z",
    )
    failed = build_failed_scan_terminal_payload(
        filters={"scan_mode": "trend"},
        error_message="network down",
        failed_at="2026-04-28T00:02:00Z",
    )

    assert stale["status"] == "stale"
    assert stale["stale"] is True
    assert stale["rows"] == [{"id": "row-1"}]
    assert stale["filters"] == {"scan_mode": "trend"}
    assert stale["last_success_at"] == "2026-04-28T00:00:00Z"
    assert failed["status"] == "failed"
    assert failed["summary"]["candidate_total"] == 0
    assert failed["rows"] == []


def test_scan_terminal_snapshot_id_is_stable_for_same_ranked_inputs():
    summary = {
        "candidate_total": 2,
        "tradable_market_count": 2,
        "avg_edge_percent": 3.5,
    }
    rows = [
        {"id": "a", "edge_percent": 4.0, "final_score": 90.0},
        {"id": "b", "edge_percent": 3.0, "final_score": 80.0},
    ]

    first = build_scan_terminal_snapshot_id({"limit": 2}, rows, summary, rows[0])
    second = build_scan_terminal_snapshot_id({"limit": 2}, rows, summary, rows[0])

    assert first == second
    assert first.startswith("scan-")


def test_scan_terminal_payload_slims_deferred_runway_history_rows():
    runway_points = [
        {"time": f"2026-05-31T{hour:02d}:00:00+00:00", "temp": 20 + hour}
        for hour in range(24)
    ]
    rows = [
        {
            "id": f"row-{index}",
            "runway_plate_history": {"35R": list(runway_points)},
        }
        for index in range(SCAN_PAYLOAD_FULL_RUNWAY_HISTORY_ROWS + 2)
    ]

    compacted = compact_ranked_scan_rows_for_payload(rows)

    assert len(compacted[0]["runway_plate_history"]["35R"]) == len(runway_points)
    assert (
        len(
            compacted[SCAN_PAYLOAD_FULL_RUNWAY_HISTORY_ROWS - 1][
                "runway_plate_history"
            ]["35R"]
        )
        == len(runway_points)
    )
    assert (
        len(
            compacted[SCAN_PAYLOAD_FULL_RUNWAY_HISTORY_ROWS][
                "runway_plate_history"
            ]["35R"]
        )
        == SCAN_PAYLOAD_DEFERRED_RUNWAY_POINTS
    )
    assert len(rows[SCAN_PAYLOAD_FULL_RUNWAY_HISTORY_ROWS]["runway_plate_history"]["35R"]) == len(
        runway_points
    )


def test_scan_terminal_quick_row_compacts_runway_history_for_list_payload():
    raw_history = {
        "35R": [
            {"time": "2026-05-31T00:00:00+00:00", "temp": 22.11},
            {"time": "2026-05-31T00:01:00+00:00", "temp": 22.22},
            {"time": "2026-05-31T00:02:00+00:00", "temp": 22.33},
            {"time": "2026-05-31T00:10:00+00:00", "temp": 23.44},
            {"time": "2026-05-31T00:11:00+00:00", "temp": 23.55},
        ]
    }

    row = _build_quick_row(
        city="shanghai",
        data={
            "display_name": "Shanghai",
            "local_date": "2026-05-31",
            "local_time": "2026-05-31T08:11:00+08:00",
            "temp_symbol": "°C",
            "current": {"temp": 22.3, "max_so_far": 23.0},
            "risk": {"airport": "Shanghai Pudong", "level": "medium"},
            "deb": {"prediction": 24.0},
            "probabilities": {"distribution": []},
            "multi_model": {},
            "runway_plate_history": raw_history,
        },
    )

    compact_history = row["runway_plate_history"]["35R"]

    assert len(compact_history) == 2
    assert compact_history[0]["temp"] == 22.3
    assert compact_history[1]["temp"] == 23.6
    assert len(str(row["runway_plate_history"])) < len(str(raw_history))


def test_scan_terminal_quick_row_exposes_airport_primary_source_metadata():
    row = _build_quick_row(
        city="ankara",
        data={
            "display_name": "Ankara",
            "local_date": "2026-06-14",
            "local_time": "2026-06-14T15:10:00+03:00",
            "temp_symbol": "°C",
            "current": {"temp": 19.0, "max_so_far": 19.0},
            "risk": {"airport": "Esenboğa", "icao": "LTAC", "level": "medium"},
            "airport_primary": {
                "station_code": "LTAC",
                "station_label": "Esenboğa 机场",
                "source_code": "mgm",
                "source_label": "MGM",
            },
            "official_network_source": "turkey_mgm",
            "official_network_status": {
                "provider_code": "turkey_mgm",
                "provider_label": "MGM",
            },
            "deb": {"prediction": 20.0},
            "probabilities": {"distribution": []},
            "multi_model": {},
        },
    )

    assert row["icao"] == "LTAC"
    assert row["station_source_code"] == "mgm"
    assert row["station_source_label"] == "MGM"
    assert row["station_code"] == "LTAC"
    assert row["network_provider"] == "turkey_mgm"


def test_metar_gate_vetoes_yes_when_observed_breaks_above_bucket():
    row = {
        "id": "yes-row",
        "side": "yes",
        "target_lower": 32.0,
        "target_upper": 34.0,
        "target_unit": "°C",
        "metar_context": {
            "obs_count": 6,
            "max_temp": 35.0,
            "last_temp": 35.0,
            "trend_delta": 1.0,
            "stale_for_today": False,
        },
    }

    _apply_metar_gate_to_row(row)

    assert row["v4_metar_decision"] == "veto"
    assert row["ai_decision"] == "veto"
    assert "越过目标桶上沿" in row["ai_reason_zh"]