import asyncio
from contextlib import asynccontextmanager
from datetime import datetime, timedelta, timezone
from types import SimpleNamespace
from unittest.mock import AsyncMock
from uuid import uuid4

import pytest
from fastapi import HTTPException

from app.mcp import query_service, scada_readings
from app.mcp.catalog import TOOLS
from app.mcp.models import ScadaInput
from app.domain.domain_errors import DomainError
from app.services.operational_scada_coverage import require_complete_scada_evidence


START = datetime(2026, 7, 16, tzinfo=timezone.utc)


def _args(**extra):
    return ScadaInput(context_handle="a" * 43, start=START, end=START + timedelta(hours=1), **extra)


def test_all_scada_reads_every_catalogue_and_observation_page_with_full_attributes(monkeypatch):
    devices = [{"device_id": f"D{i}", "device_type": ["pressure", "pipe_flow", "quality"][i % 3],
                "measurement_unit": ["m", "m3/h", "mg/L"][i % 3], "node_id": f"J{i}",
                "longitude": 121.5, "latitude": 31.0, "reliability": 1} for i in range(503)]
    rows = [{"device_id": f"D{i % 502}", "time": START.isoformat(),
             "monitored_value": float(i), "cleaned_value": None} for i in range(1205)]

    async def catalog(connection, *, limit, offset):
        return devices[offset:offset + limit]

    async def series(connection, **args):
        assert args["ids"] == [device["device_id"] for device in devices]
        assert args["field"] is None
        return rows[args["offset"]:args["offset"] + args["limit"]]

    catalog_reader, series_reader = AsyncMock(side_effect=catalog), AsyncMock(side_effect=series)
    monkeypatch.setattr(scada_readings.reads, "scada_devices", catalog_reader)
    monkeypatch.setattr(scada_readings.reads, "series_rows", series_reader)
    monkeypatch.setattr(query_service, "authorize_query", AsyncMock())

    @asynccontextmanager
    async def timescale(): yield object()

    scope = SimpleNamespace(connection=object(), timescale=timescale)
    result = asyncio.run(query_service.execute_query(scope, SimpleNamespace(replay_context=None),
                                                    TOOLS["query_scada_readings"], _args()))
    assert result.data["devices"] == devices and result.data["readings"] == rows
    assert [call.kwargs["offset"] for call in catalog_reader.await_args_list] == [0, 500]
    assert [call.kwargs["offset"] for call in series_reader.await_args_list] == [0, 1000]
    coverage = result.data["coverage"]
    assert coverage["query_complete"] is True and coverage["queried_device_count"] == 503
    assert coverage["devices_with_readings"] == 502 and coverage["missing_device_ids"] == ["D502"]
    assert coverage["reading_count"] == 1205 and len(result.units) == 503
    assert result.origin == "system_data" and result.page is None and not result.truncated


def test_all_scada_replay_reads_only_frozen_devices_batches_and_window(monkeypatch):
    devices = [{"device_id": key, "device_type": kind, "measurement_unit": unit, "node_id": key}
               for key, kind, unit in [("P1", "pressure", "m"), ("Q1", "quality", "mg/L"), ("NEW", "pressure", "m")]]
    rows = [{"device_id": "P1", "time": START.isoformat(), "monitored_value": 20.,
             "cleaned_value": 20., "quality": "valid", "reason": "冻结观测", "source_hash": "internal-hash"}]
    context = {"replay_id": str(uuid4()), "knowledge_cutoff": (START + timedelta(hours=1)).isoformat(),
               "batch_sequence": 9}
    repository = SimpleNamespace(
        get=AsyncMock(return_value={"start_time": START.isoformat(), "config": {"device_ids": ["P1", "Q1"]},
                                   "devices": devices[:2]}),
        detected_outputs=AsyncMock(return_value=rows),
    )
    monkeypatch.setattr(scada_readings.reads, "scada_devices", AsyncMock(return_value=devices))
    monkeypatch.setattr(scada_readings, "ScadaReplayRepository", lambda *args: repository)
    timescale = AsyncMock(side_effect=AssertionError("Frozen reads must never use live TSDB"))
    scope = SimpleNamespace(connection=object(), project=SimpleNamespace(project_id=uuid4()), timescale=timescale)
    data, units, gaps, origin = asyncio.run(scada_readings.read_complete_scada(
        scope, SimpleNamespace(replay_context=context), _args(),
    ))
    assert data["devices"] == devices
    assert data["readings"] == [{key: value for key, value in rows[0].items() if key != "source_hash"}]
    assert data["coverage"]["queried_device_ids"] == ["P1", "Q1"]
    assert data["coverage"]["missing_device_ids"] == ["Q1"]
    assert data["coverage"]["outside_frozen_scope_device_ids"] == ["NEW"]
    repository.detected_outputs.assert_awaited_once_with(
        context["replay_id"], ["P1", "Q1"], START, START + timedelta(hours=1), 9, limit=1001, offset=0,
    )
    assert origin == "synthetic" and len(gaps) == 2 and units == {"P1": "m", "Q1": "mg/L"}
    timescale.assert_not_called()


def test_all_scada_empty_catalogue_does_not_query_unfiltered_timeseries(monkeypatch):
    monkeypatch.setattr(scada_readings.reads, "scada_devices", AsyncMock(return_value=[]))
    timescale = AsyncMock(side_effect=AssertionError("An empty catalogue must not select every TSDB row"))
    data, units, _, _ = asyncio.run(scada_readings.read_complete_scada(
        SimpleNamespace(connection=object(), timescale=timescale), SimpleNamespace(replay_context=None), _args(),
    ))
    assert data["coverage"]["queried_device_count"] == 0 and data["readings"] == [] and units == {}
    timescale.assert_not_called()


def test_all_scada_rejects_offset_and_empty_explicit_selection():
    for values in ({"offset": 1}, {"device_ids": []}):
        with pytest.raises(ValueError):
            _args(**values)
    assert _args(device_ids=["P1"], offset=100).offset == 100


@pytest.mark.parametrize("captured,approved,explicit,direct,allowed", [
    ({"scada_readings", "scada_devices"}, {"scada_readings"}, False, False, False),
    ({"scada_readings"}, {"scada_readings", "scada_devices"}, False, False, False),
    ({"scada_readings", "scada_devices"}, {"scada_devices"}, False, False, False),
    ({"scada_readings", "scada_devices"}, {"scada_readings", "scada_devices"}, False, False, True),
    ({"scada_readings"}, {"scada_readings"}, True, False, True),
    (set(), set(), False, True, True),
])
def test_complete_scada_requires_both_captured_and_current_source_grants(
    monkeypatch, captured, approved, explicit, direct, allowed,
):
    @asynccontextmanager
    async def transaction():
        yield

    can_read = AsyncMock(side_effect=lambda task_id, source: source in approved)
    monkeypatch.setattr(query_service, "permissions_for_context", lambda _: {"scada.view"} if direct else set())
    monkeypatch.setattr(query_service, "task_service", lambda _: SimpleNamespace(can_read_source=can_read))
    binding = SimpleNamespace(
        replay_context=None, tool_cap=["query_scada_readings"],
        permission_cap=["scada.view"] if direct else [], task_id=uuid4(), task_scope_cap=captured,
    )
    scope = SimpleNamespace(project=object(), connection=SimpleNamespace(transaction=transaction))
    args = _args(**({"device_ids": ["P1"]} if explicit else {}))
    call = query_service.authorize_query(scope, binding, TOOLS["query_scada_readings"], args)
    if allowed:
        asyncio.run(call)
    else:
        with pytest.raises(HTTPException) as error:
            asyncio.run(call)
        assert error.value.status_code == 403
    if direct:
        can_read.assert_not_awaited()
    elif allowed:
        expected = {"scada_readings"} if explicit else {"scada_readings", "scada_devices"}
        assert {call.args[1] for call in can_read.await_args_list} == expected


def test_all_scada_reports_oversize_instead_of_successfully_returning_a_partial_scan(monkeypatch):
    monkeypatch.setattr(scada_readings.reads, "scada_devices", AsyncMock(return_value=[
        {"device_id": "P1", "description": "x" * 200},
    ]))
    monkeypatch.setattr(scada_readings, "MAX_COMPLETE_SCADA_BYTES", 100)
    with pytest.raises(HTTPException) as error:
        asyncio.run(scada_readings.read_complete_scada(
            SimpleNamespace(connection=object()), SimpleNamespace(replay_context=None), _args(),
        ))
    assert error.value.status_code == 413


@pytest.mark.parametrize("defect", [None, "old_batch", "other_replay", "old_window", "unreferenced", "truncated"])
def test_final_scada_evidence_must_match_the_current_frozen_scope(defect):
    replay_id = str(uuid4())
    replay = {"replay_id": replay_id, "knowledge_cutoff": START.isoformat(), "batch_sequence": 9}
    task = {"workflow": "burst_report", "source": {"replay_context": replay}}
    coverage = {"mode": "all_scada", "query_complete": True, "replay_id": replay_id, "batch_sequence": 9}
    window = {"start": (START - timedelta(hours=1)).isoformat(), "end": START.isoformat()}
    if defect == "old_batch": coverage["batch_sequence"] = 8
    if defect == "other_replay": coverage["replay_id"] = str(uuid4())
    if defect == "old_window": window["end"] = (START - timedelta(minutes=15)).isoformat()
    evidence = [{"evidence_id": "mcp:all", "data_snapshot": {
        "source": "query_scada_readings", "truncated": defect == "truncated",
        "data": {"data": {"coverage": coverage}, "window": window},
    }}]
    refs = [] if defect == "unreferenced" else ["mcp:all"]
    if defect is None:
        require_complete_scada_evidence(task, evidence, refs)
    else:
        with pytest.raises(DomainError) as error:
            require_complete_scada_evidence(task, evidence, refs)
        assert error.value.code == "SCADA_COVERAGE_REQUIRED"