"""Mock-controller tests for ``xceadmin.push`` (design.md, ``xceadmin.push``;
Requirements 2.1, 2.2, 2.3, 2.4, 2.5, 2.6, 2.7, 3.1, 3.2, 3.3, 3.4, 6.1,
6.2, 6.3, 7.1, 7.2, 7.3).

``push_config``'s real controller counterpart is blocked on a separate,
already-documented controller-side dispatch-gate defect
(``ha_dispatch_gate()``/``cluster.c``'s inverted primary check --
spec.json ``blocked_on``), and ``probe_config``'s ``OPCode_probeConfigTag``
has no server dispatch case at all (``research.md``, "ProbeConfigTag
liveness"). Both are therefore tested against a real local TLS mock
server built from the actual bundled schema -- not a live cluster --
exactly as design.md's ``xceadmin.push`` Boundary Commitments anticipate.
"""

from __future__ import annotations

import ssl
import socket
import threading
from pathlib import Path

import pytest

from xcecommon import wire
from xcecommon.errors import ConnectionTimeoutError, NoPrimaryError, ReplyDecodeError, TooManyRedirectsError
from xcecommon.schema import load_schema
from xceadmin import discovery, push


def _server_context(cert_file: Path, key_file: Path) -> ssl.SSLContext:
    context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
    context.load_cert_chain(certfile=str(cert_file), keyfile=str(key_file))
    return context


class _MockController:
    """A one-shot TLS server: accepts one connection, hands the raw
    accepted TLS socket to ``handler``, then closes.
    """

    def __init__(self, handler, cert_file: Path, key_file: Path):
        self._handler = handler
        self._cert_file = cert_file
        self._key_file = key_file
        self._listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self._listener.bind(("127.0.0.1", 0))
        self._listener.listen(1)
        self.port = self._listener.getsockname()[1]
        self._thread = threading.Thread(target=self._run, daemon=True)
        self._thread.start()

    def _run(self) -> None:
        try:
            raw_conn, _ = self._listener.accept()
        except OSError:
            return
        try:
            tls_conn = _server_context(self._cert_file, self._key_file).wrap_socket(
                raw_conn, server_side=True
            )
        except (ssl.SSLError, OSError):
            raw_conn.close()
            return
        try:
            self._handler(tls_conn)
        except OSError:
            pass
        finally:
            tls_conn.close()

    def close(self) -> None:
        self._listener.close()
        self._thread.join(timeout=5)


@pytest.fixture
def mock_controller(tls_certs):
    servers: list[_MockController] = []

    def _make(handler) -> _MockController:
        srv = _MockController(handler, tls_certs.cert_file, tls_certs.key_file)
        servers.append(srv)
        return srv

    yield _make
    for srv in servers:
        srv.close()


@pytest.fixture
def cluster_bin(tmp_path):
    path = tmp_path / "cluster.bin"
    path.write_bytes(b"fake cluster config binary contents")
    return str(path)


# ---------------------------------------------------------------------------
# push_config(): happy path, redirect-following, error mapping
# ---------------------------------------------------------------------------


def test_push_config_successful_commit(mock_controller, cluster_bin, wrap_reply):
    mod = load_schema()
    received = {}

    def _handler(conn):
        request_bytes = wire.recv_frame(conn)
        received["request_bytes"] = request_bytes
        with mod.Request.from_bytes(request_bytes) as req:
            received["opcode"] = req.opcode
            received["sha256"] = bytes(req.body.upgradeconfigrequestmsg.sha256)
            received["bytes"] = bytes(req.body.upgradeconfigrequestmsg.bytes)

        reply = mod.RequestReply.new_message()
        reply.status = 0
        upgrade_reply = reply.data.init("upgradeconfigreplymsg")
        upgrade_reply.versionTag = 7
        upgrade_reply.ackedCount = 3
        upgrade_reply.error = ""
        wire.send_frame(conn, wrap_reply(reply))

    srv = mock_controller(_handler)

    result = push.push_config(cluster_bin, f"127.0.0.1:{srv.port}")

    assert result.version_tag == 7
    assert result.acked_count == 3
    assert result.error == ""
    assert received["opcode"] == "upgradeConfig"
    assert received["bytes"] == Path(cluster_bin).read_bytes()
    import hashlib

    assert received["sha256"] == hashlib.sha256(Path(cluster_bin).read_bytes()).digest()


def test_push_config_follows_one_redirect_to_primary(mock_controller, cluster_bin, wrap_reply):
    mod = load_schema()

    def _standby_handler(conn):
        wire.recv_frame(conn)
        reply = mod.RequestReply.new_message()
        reply.status = 0
        reply.redirectHost = "127.0.0.1"
        # port filled in once the primary server exists, see below
        reply.redirectPort = _standby_handler.primary_port
        reply.data.nodata = None
        wire.send_frame(conn, wrap_reply(reply))

    def _primary_handler(conn):
        wire.recv_frame(conn)
        reply = mod.RequestReply.new_message()
        reply.status = 0
        upgrade_reply = reply.data.init("upgradeconfigreplymsg")
        upgrade_reply.versionTag = 9
        upgrade_reply.ackedCount = 2
        upgrade_reply.error = ""
        wire.send_frame(conn, wrap_reply(reply))

    primary_srv = mock_controller(_primary_handler)
    _standby_handler.primary_port = primary_srv.port
    standby_srv = mock_controller(_standby_handler)

    result = push.push_config(cluster_bin, f"127.0.0.1:{standby_srv.port}")

    assert result.version_tag == 9
    assert result.acked_count == 2


def test_push_config_non_zero_status_reports_error_not_exception(mock_controller, cluster_bin, wrap_reply):
    mod = load_schema()

    def _handler(conn):
        wire.recv_frame(conn)
        reply = mod.RequestReply.new_message()
        reply.status = 1
        reply.data.debugmsg = "simulated failure"
        wire.send_frame(conn, wrap_reply(reply))

    srv = mock_controller(_handler)

    result = push.push_config(cluster_bin, f"127.0.0.1:{srv.port}")

    assert result.acked_count == 0
    assert "status 1" in result.error


def test_push_config_wrong_reply_variant_raises_reply_decode_error(mock_controller, cluster_bin, wrap_reply):
    mod = load_schema()

    def _handler(conn):
        wire.recv_frame(conn)
        reply = mod.RequestReply.new_message()
        reply.status = 0
        reply.data.nodata = None  # not upgradeconfigreplymsg
        wire.send_frame(conn, wrap_reply(reply))

    srv = mock_controller(_handler)

    with pytest.raises(ReplyDecodeError):
        push.push_config(cluster_bin, f"127.0.0.1:{srv.port}")


def test_push_config_rejects_bare_requestreply_bytes_old_wire_shape_regression(
    mock_controller, cluster_bin
):
    """Regression test (task 6.1): a mock server that still sends the OLD,
    wrong shape -- a bare RequestReply with no outer Request envelope --
    must be cleanly rejected as ReplyDecodeError, not silently accepted
    and not an unhandled capnp.KjException. This locks in that the fixed
    client only accepts the real controller wire shape.
    """
    mod = load_schema()

    def _handler(conn):
        wire.recv_frame(conn)
        reply = mod.RequestReply.new_message()
        reply.status = 0
        upgrade_reply = reply.data.init("upgradeconfigreplymsg")
        upgrade_reply.versionTag = 1
        upgrade_reply.ackedCount = 1
        upgrade_reply.error = ""
        wire.send_frame(conn, reply.to_bytes())  # NOT wrapped -- old shape

    srv = mock_controller(_handler)

    with pytest.raises(ReplyDecodeError):
        push.push_config(cluster_bin, f"127.0.0.1:{srv.port}")


def test_push_config_default_port_used_when_omitted():
    # push.py delegates --controller parsing to xceadmin.discovery.
    # resolve_controllers (shared with scheduler.py) -- see test_discovery.py
    # for that function's own dedicated coverage; this test just confirms
    # push_config's pipeline still reaches it for the single-host case.
    (host, port), = discovery.resolve_controllers("somehost")
    assert host == "somehost"
    assert port == 9991


def test_push_config_explicit_port_parsed():
    (host, port), = discovery.resolve_controllers("somehost:1234")
    assert host == "somehost"
    assert port == 1234


# ---------------------------------------------------------------------------
# probe_config(): dry-run against the mock controller
# ---------------------------------------------------------------------------


def test_probe_config_reports_current_and_would_be_generation(mock_controller, cluster_bin, wrap_reply):
    """bug-1507: probe_config now sends a Request envelope and decodes
    through follow_redirects → decode_reply, matching push_config's path.
    The mock server must unwrap the Request envelope on receive and wrap
    its reply the same way a real controller does.  """
    mod = load_schema()
    received = {}

    def _handler(conn):
        request_bytes = wire.recv_frame(conn)
        received["request_bytes"] = request_bytes
        # Request{opcode: probeConfigTag, body.probeconfigtagrequestmsg: ...}
        with mod.Request.from_bytes(request_bytes) as req:
            assert req.opcode == "probeConfigTag"
            assert req.body.which() == "probeconfigtagrequestmsg"
            received["name"] = req.body.probeconfigtagrequestmsg.name

        # Build reply: RequestReply{status:0, data.probeconfigtagreplymsg: {configTag:41}}
        inner = mod.RequestReply.new_message()
        inner.status = 0
        probe_reply = inner.data.init("probeconfigtagreplymsg")
        probe_reply.configTag = 41
        wire.send_frame(conn, wrap_reply(inner))

    srv = mock_controller(_handler)

    result = push.probe_config(cluster_bin, f"127.0.0.1:{srv.port}")

    assert result.current_generation == 41
    assert result.would_be_generation == 42
    assert result.accepted is True
    assert result.reason == ""
    assert received["name"] == ""


def test_probe_config_zero_generation_still_accepted(mock_controller, cluster_bin, wrap_reply):
    """A fresh cluster with no config committed yet (configTag == 0) is
    still a successful probe -- configTag == 0 is a valid, real response
    from a real controller handler now (bug-1506).  """
    mod = load_schema()

    def _handler(conn):
        wire.recv_frame(conn)
        inner = mod.RequestReply.new_message()
        inner.status = 0
        probe_reply = inner.data.init("probeconfigtagreplymsg")
        probe_reply.configTag = 0
        wire.send_frame(conn, wrap_reply(inner))

    srv = mock_controller(_handler)

    result = push.probe_config(cluster_bin, f"127.0.0.1:{srv.port}")

    assert result.current_generation == 0
    assert result.would_be_generation == 1
    assert result.accepted is True


@pytest.mark.parametrize("tag_value", [0, 1, 999999])
def test_probe_config_has_no_wire_level_rejection_signal_by_design(
    mock_controller, cluster_bin, tag_value, wrap_reply
):
    """ProbeConfigTagReply (controller.capnp:909) has only a bare
    ``configTag`` field -- no separate accept/reject flag exists.  A
    real controller handler (bug-1506) always returns status==0 here;
    every successful round-trip, regardless of the returned configTag
    value, yields ``accepted=True``.  A real "would this SHA be accepted"
    check is future work (Requirement 6.1).  """
    mod = load_schema()

    def _handler(conn):
        wire.recv_frame(conn)
        inner = mod.RequestReply.new_message()
        inner.status = 0
        probe_reply = inner.data.init("probeconfigtagreplymsg")
        probe_reply.configTag = tag_value
        wire.send_frame(conn, wrap_reply(inner))

    srv = mock_controller(_handler)

    result = push.probe_config(cluster_bin, f"127.0.0.1:{srv.port}")

    assert result.accepted is True
    assert result.reason == ""


# ---------------------------------------------------------------------------
# End-to-end redirect/discovery failure propagation through push_config
# ---------------------------------------------------------------------------


def test_push_config_propagates_too_many_redirects_end_to_end(mock_controller, cluster_bin, wrap_reply):
    """A real chain of 4 redirecting mock servers -- push_config's own
    composition with xceadmin.discovery must surface TooManyRedirectsError,
    not silently follow forever or return a bogus result.
    """
    mod = load_schema()
    servers = {}

    def _make_redirect_handler(next_port_getter):
        def _handler(conn):
            wire.recv_frame(conn)
            reply = mod.RequestReply.new_message()
            reply.status = 0
            reply.redirectHost = "127.0.0.1"
            reply.redirectPort = next_port_getter()
            reply.data.nodata = None
            wire.send_frame(conn, wrap_reply(reply))

        return _handler

    # 5 servers: a -> b -> c -> d -> e (4 redirects, one too many).
    names = ["a", "b", "c", "d", "e"]
    for i, name in enumerate(names[:-1]):
        next_name = names[i + 1]
        servers[name] = mock_controller(_make_redirect_handler(lambda n=next_name: servers[n].port))
    # The last server ('e') also redirects -- this is hop #4, exceeding the
    # 3-hop limit before it would ever be contacted.
    servers["e"] = mock_controller(_make_redirect_handler(lambda: 1))

    with pytest.raises(TooManyRedirectsError):
        push.push_config(cluster_bin, f"127.0.0.1:{servers['a'].port}")


def test_push_config_propagates_no_primary_error_on_connection_timeout(cluster_bin, monkeypatch):
    """xceadmin.discovery already unit-tests this conversion directly
    (test_discovery.py); this proves push_config's actual composition
    with it surfaces the same NoPrimaryError, without waiting out a real
    30s socket timeout -- xcecommon.wire.connect is monkeypatched to
    raise ConnectionTimeoutError immediately, exactly what it would
    eventually raise on a genuine timeout.
    """

    def _fake_connect(host, port, timeout_s=30.0):
        raise ConnectionTimeoutError(f"timeout connecting to {host}:{port}", host, port)

    monkeypatch.setattr(wire, "connect", _fake_connect)

    with pytest.raises(NoPrimaryError):
        push.push_config(cluster_bin, "127.0.0.1:9991")