"""The Codex quota rides the existing 2 s server snapshot.

Guards the two constraints that make this feature cheap: no new periodic
device request and no new periodic quota message. Only the already-scheduled
server_snapshot frame gains a few small fields.
"""
import asyncio
import inspect

import pytest

import device_gateway.app as gateway
from device_gateway.collectors.system import SystemCollector
from device_gateway.control.codex_quota import CodexQuota


class Native:
    generation = 1

    def __init__(self, payload):
        self.payload = payload
        self.quota_reads = 0
        self.calls = []

    async def start(self):
        pass

    async def write(self, value):
        pass

    async def close(self):
        pass

    async def call(self, method, params):
        self.calls.append((method, params))
        if method == "initialize":
            return {}
        assert method == "account/rateLimits/read"
        self.quota_reads += 1
        return self.payload


def quota_snapshot(used):
    return {"rateLimitsByLimitId": {"codex": {"primary": {
        "usedPercent": used, "resetsAt": 1700000000, "windowDurationMins": 300}}}}


def test_payload_carries_quota_on_existing_snapshot_type(monkeypatch, tmp_path):
    native = Native(quota_snapshot(28))
    adapter = gateway.CodexAdapter(gateway.project_registry, native)
    monkeypatch.setitem(gateway.adapters, "codex", adapter)
    asyncio.run(adapter.ready())

    collector = SystemCollector(hostname="test-host", disk_path=str(tmp_path))
    snapshot = collector.collect()
    payload = gateway.device_snapshot_payload(snapshot)

    # Same message type and same metric fields as before.
    assert payload["type"] == "server_snapshot"
    assert payload["hostname"] == "test-host"
    assert payload["cpu_percent"] == snapshot.cpu_percent
    # Plus the small derived additions.
    assert payload["codex_quota_valid"] is True
    assert payload["codex_remaining_percent"] == 72
    assert payload["codex_resets_at"] == 1700000000
    assert payload["codex_window_minutes"] == 300
    assert payload["host_state"] in {"Idle", "Active", "Busy", "Offline"}
    # No account identity or plan leaks onto the wire.
    blob = repr(payload)
    for secret in ("accountId", "planType", "credits", "limitName"):
        assert secret not in blob


def test_missing_codex_adapter_reports_invalid_quota(monkeypatch, tmp_path):
    monkeypatch.delitem(gateway.adapters, "codex", raising=False)
    collector = SystemCollector(hostname="test-host", disk_path=str(tmp_path))
    payload = gateway.device_snapshot_payload(collector.collect())
    assert payload["type"] == "server_snapshot"
    assert payload["codex_quota_valid"] is False
    assert payload["codex_remaining_percent"] == 0
    assert payload["codex_resets_at"] is None


def test_snapshot_interval_is_unchanged():
    # The broadcast cadence is still the single configured 2 s loop; the quota
    # adds no timer of its own.
    assert gateway.snapshot_interval_seconds == 2.0
    source = inspect.getsource(gateway.monitor_system)
    assert source.count("asyncio.sleep(snapshot_interval_seconds)") == 1
    assert "codex" not in source.replace("device_snapshot_payload(snapshot)", "")


def test_no_quota_poll_message_or_endpoint_exists():
    source = inspect.getsource(gateway)
    for forbidden in ("codex_quota_poll", "quota_heartbeat", "quota_request"):
        assert forbidden not in source


def test_broadcast_payloads_do_not_add_quota_frames(monkeypatch, tmp_path):
    """One collect() must produce exactly one snapshot frame, not two."""
    native = Native(quota_snapshot(10))
    adapter = gateway.CodexAdapter(gateway.project_registry, native)
    monkeypatch.setitem(gateway.adapters, "codex", adapter)
    asyncio.run(adapter.ready())
    before = native.quota_reads

    frames = []

    async def capture(payload):
        frames.append(payload)

    monkeypatch.setattr(gateway, "broadcast_server_snapshot", capture)

    async def one_tick():
        snapshot = gateway.system_collector.collect()
        await gateway.broadcast_server_snapshot(gateway.device_snapshot_payload(snapshot))

    asyncio.run(one_tick())
    assert len(frames) == 1
    assert frames[0]["type"] == "server_snapshot"
    # Composing a frame never triggers a native quota read.
    assert native.quota_reads == before

def test_quota_starts_and_reconnects_without_selecting_codex(tmp_path):
    """A Pi-only watch still receives account quota; no thread action is sent."""
    native = Native(quota_snapshot(28))
    adapter = gateway.CodexAdapter(gateway.project_registry, native)
    adapter.quota_periodic_seconds = 0.02

    async def scenario():
        task = asyncio.create_task(gateway.maintain_codex_quota(adapter, retry_seconds=0.01))
        try:
            async def wait_remaining(value):
                async with asyncio.timeout(1):
                    while not adapter.quota.valid or adapter.quota.remaining_percent != value:
                        await asyncio.sleep(0.001)
            await wait_remaining(72)
            native.generation += 1
            native.payload = quota_snapshot(44)
            adapter.disconnected()
            assert not adapter.quota.valid
            await wait_remaining(56)
            assert adapter.quota_periodic_task is not None
            assert all(method in ('initialize', 'account/rateLimits/read')
                       for method, _ in native.calls)
        finally:
            task.cancel()
            with pytest.raises(asyncio.CancelledError): await task
            await adapter.close()
    asyncio.run(scenario())