"""JsonUnixSocket must reconnect after the daemon it talks to goes away.

A socket object survives its peer. If ``start()`` treats "self.socket is not
None" as "connected", a restarted Codex daemon leaves the gateway pinned to a
dead connection: every later call fails instantly, ``generation`` never moves,
and the adapter never re-reads quota. These tests pin the reconnect contract so
that regression cannot return silently.
"""
import asyncio

import pytest

from device_gateway.control.rpc import JsonUnixSocket


class FakeSocket:
    """Minimal stand-in for a websockets client connection."""

    def __init__(self, state=1):
        self.state = state
        self.sent = []
        self.closed = False

    def __aiter__(self):
        return self

    async def __anext__(self):
        # Block until the test closes the socket, like an idle daemon.
        while self.state != 3:
            await asyncio.sleep(0.01)
        raise StopAsyncIteration

    async def send(self, payload):
        self.sent.append(payload)

    async def close(self):
        self.closed = True
        self.state = 3


class FakeConnect:
    """Records every connection attempt and hands back scripted sockets."""

    def __init__(self):
        self.calls = []
        self.created = []

    def __call__(self, path, **kwargs):
        self.calls.append((path, kwargs))
        socket = FakeSocket()
        self.created.append(socket)

        async def attempt():
            return socket

        return attempt()


def socket_transport(monkeypatch, connect):
    monkeypatch.setattr("device_gateway.control.rpc.unix_connect", connect)
    return JsonUnixSocket("/tmp/fake-codex.sock")


def test_start_connects_once_and_bumps_generation(monkeypatch):
    connect = FakeConnect()
    rpc = socket_transport(monkeypatch, connect)

    async def scenario():
        await rpc.start()
        assert len(connect.calls) == 1
        assert rpc.generation == 1
        assert rpc.socket is connect.created[0]
        await rpc.close()

    asyncio.run(scenario())


def test_start_reuses_a_live_socket(monkeypatch):
    connect = FakeConnect()
    rpc = socket_transport(monkeypatch, connect)

    async def scenario():
        await rpc.start()
        await rpc.start()
        assert len(connect.calls) == 1, "an open socket must not be replaced"
        assert rpc.generation == 1
        await rpc.close()

    asyncio.run(scenario())


def test_start_replaces_a_socket_closed_by_the_peer(monkeypatch):
    """The daemon restarting closes the socket; the next start must reconnect."""
    connect = FakeConnect()
    rpc = socket_transport(monkeypatch, connect)

    async def scenario():
        await rpc.start()
        connect.created[0].state = 3  # peer went away, object is still referenced

        await rpc.start()
        assert len(connect.calls) == 2, "a closed socket must be replaced"
        assert rpc.socket is connect.created[1]
        assert rpc.generation == 2, "reconnect must bump generation to refresh quota"
        await rpc.close()

    asyncio.run(scenario())


def test_reader_cleanup_does_not_clear_a_newer_socket(monkeypatch):
    """A slow reader for an old socket must not strand the replacement."""
    connect = FakeConnect()
    rpc = socket_transport(monkeypatch, connect)

    async def scenario():
        await rpc.start()
        old_socket = connect.created[0]
        old_reader = rpc.reader

        # The peer drops, start() reconnects, and only then does the old
        # reader task get to run its finally block.
        old_socket.state = 3
        await rpc.start()
        new_socket = rpc.socket
        assert new_socket is connect.created[1]

        await old_reader

        assert rpc.socket is new_socket, "old reader must not clear the new socket"
        await rpc.close()

    asyncio.run(scenario())


def test_reconnect_preserves_pending_call_failures(monkeypatch):
    """Pending work fails loudly instead of hanging when the daemon drops."""
    connect = FakeConnect()
    rpc = socket_transport(monkeypatch, connect)

    async def scenario():
        await rpc.start()
        pending = asyncio.get_running_loop().create_future()
        rpc.pending[7] = pending

        # The peer goes away, so the reader task ends on its own and must
        # fail the in-flight call instead of leaving it hanging forever.
        connect.created[0].state = 3
        await rpc.reader

        assert pending.done()
        assert isinstance(pending.exception(), Exception)
        assert rpc.socket is None
        await rpc.close()

    asyncio.run(scenario())