feat(admin-config-tool): task 3.3 - xceadmin.push

push_config()/probe_config(): build/send Request{opcode:upgradeConfig}
and the direct-struct ProbeConfigTagRequest respectively, following
redirects to the primary for the former (ProbeConfigTagReply has no
redirect fields, so probe_config connects directly). Maps controller
status/reply-variant failures to typed results/errors. Tested against
a real local TLS mock controller built from the bundled schema, since
neither opcode has a live controller-side handler today (separately
documented, unrelated to this task).

Unblocked mid-task by bug-1461 (commit e53c086): controller.capnp had
a struct-name collision that miscompiled UpgradeConfigRequest.sha256/
.bytes, making this task's core request impossible to construct via
pycapnp until fixed at the schema root.

src/xceadmin/src/xceadmin/push.py

@@ -0,0 +1,162 @@
+"""``config push`` orchestration for xceadmin (design.md, ``xceadmin.push``;
+Requirements 2.1, 2.2, 2.3, 2.4, 2.5, 2.6, 2.7, 6.1, 6.2, 6.3, 7.1, 7.2,
+7.3, 7.4, 7.5, 7.6).
+
+Two functions, not one boolean-flagged function (Design Review
+correction, ``research.md``): ``push_config()`` sends ``Request{opcode:
+upgradeConfig, upgradeconfigrequestmsg}`` (``Request``/``RequestReply``
+wrapped) and ``probe_config()`` sends ``ProbeConfigTagRequest`` (a direct
+struct, no wrapper) -- their replies share no fields, so a shared return
+type would misrepresent one of the two.
+
+Both delegate connection + redirect-following to
+``xceadmin.discovery.follow_redirects`` before decoding a reply, and both
+build/decode using the real ``controller.capnp`` schema
+(``xcecommon.schema.load_schema()``) -- never touching the socket
+directly except through ``xcecommon.wire``.
+
+``probe_config()`` has no live controller counterpart today
+(``research.md``, "ProbeConfigTag liveness": no ``OPCode_probeConfigTag``
+dispatch case exists in ``cluster.c``) -- implemented against the
+documented wire contract, exercised only by a mock TLS server until the
+controller gains a handler. The real C reference client
+(``config_repl.c:1907``, ``xce_probe_config_tag``) always sends an empty
+``name`` ("anonymous probe; peer logs remote addr") and returns only a
+bare config tag with no accept/reject signal -- ``ProbeConfigTagReply``
+has no field to carry one. ``probe_config()`` matches this exactly:
+``accepted`` is a client-side inference (true on any successful
+round-trip), not a value read off the wire, since the wire contract has
+nothing else to say. Real SHA/version-compatibility validation
+(Requirement 6.1's aspirational description) is future controller-side
+work, not something this client can check today.
+
+``probe_config()`` connects directly and does NOT go through
+``xceadmin.discovery.follow_redirects``: ``ProbeConfigTagReply`` is a
+direct struct with only a ``configTag`` field -- no ``redirectHost``/
+``redirectPort`` exist on it at all, so there is no redirect signal for
+``follow_redirects`` (which decodes every reply as ``RequestReply``) to
+read. Feeding ``ProbeConfigTagReply`` bytes through it would misinterpret
+them as a different struct shape. Only ``push_config()``'s
+``Request``/``RequestReply``-wrapped exchange supports redirects.
+"""
+
+from __future__ import annotations
+
+import hashlib
+from dataclasses import dataclass
+from pathlib import Path
+
+from xcecommon import wire
+from xcecommon.errors import ReplyDecodeError
+from xcecommon.schema import load_schema
+from xceadmin.discovery import follow_redirects
+
+_DEFAULT_PORT = 9991
+
+
+@dataclass(frozen=True)
+class PushResult:
+    version_tag: int
+    acked_count: int
+    error: str
+
+
+@dataclass(frozen=True)
+class DryRunResult:
+    current_generation: int
+    would_be_generation: int
+    accepted: bool
+    reason: str
+
+
+def _split_controller(controller: str) -> tuple[str, int]:
+    """``"host:port"`` -> ``(host, port)``; a bare host uses the default
+    port 9991 (Requirement 2.2's default).
+    """
+    if ":" in controller:
+        host, port_str = controller.rsplit(":", 1)
+        return host, int(port_str)
+    return controller, _DEFAULT_PORT
+
+
+def _send_via(request_bytes: bytes, cert_dir: str):
+    def _send(host: str, port: int) -> bytes:
+        sock = wire.connect(host, port, cert_dir)
+        try:
+            return wire.send_request(sock, request_bytes)
+        finally:
+            sock.close()
+
+    return _send
+
+
+def push_config(cluster_bin_path: str, controller: str, cert_dir: str) -> PushResult:
+    """Push ``cluster_bin_path`` to the controller cluster (2.1). Follows
+    redirects to the primary (3.1-3.4) before sending.
+
+    Postconditions: exit 0 iff ``acked_count > 0`` and ``error == ""``
+    (2.6); otherwise the caller exits non-zero with ``error`` displayed
+    (2.7) -- this function only reports the result, it does not exit.
+    """
+    data = Path(cluster_bin_path).read_bytes()
+    sha256 = hashlib.sha256(data).digest()
+
+    mod = load_schema()
+    request = mod.Request.new_message()
+    request.opcode = "upgradeConfig"
+    request.seq = 1
+    upgrade = request.body.init("upgradeconfigrequestmsg")
+    upgrade.sha256 = sha256
+    upgrade.bytes = data
+    request_bytes = request.to_bytes()
+
+    host, port = _split_controller(controller)
+    _, _, reply_bytes = follow_redirects(host, port, cert_dir, _send_via(request_bytes, cert_dir))
+
+    with mod.RequestReply.from_bytes(reply_bytes) as reply:
+        if reply.status != 0:
+            return PushResult(
+                version_tag=0, acked_count=0, error=f"controller returned status {reply.status}"
+            )
+        if reply.data.which() != "upgradeconfigreplymsg":
+            reason = f"unexpected reply variant {reply.data.which()!r}, expected upgradeconfigreplymsg"
+            raise ReplyDecodeError(f"invalid reply: {reason}", reason)
+        upgrade_reply = reply.data.upgradeconfigreplymsg
+        return PushResult(
+            version_tag=upgrade_reply.versionTag,
+            acked_count=upgrade_reply.ackedCount,
+            error=upgrade_reply.error,
+        )
+
+
+def probe_config(cluster_bin_path: str, controller: str, cert_dir: str) -> DryRunResult:
+    """Probe the controller's current config generation without pushing
+    anything (6.1). ``cluster_bin_path`` is read only to confirm it
+    exists and is readable before spending a network round-trip -- its
+    content plays no role in ``ProbeConfigTagRequest``, which carries no
+    config data (see module docstring).
+
+    Postconditions: exit 0 and "OK: controller will accept this config
+    (generation N->M)" iff ``accepted`` (6.2); otherwise the caller exits
+    non-zero with ``reason`` displayed (6.3) -- this function only
+    reports the result, it does not exit.
+    """
+    Path(cluster_bin_path).read_bytes()
+
+    mod = load_schema()
+    request = mod.ProbeConfigTagRequest.new_message()
+    request.name = ""  # matches config_repl.c's xce_probe_config_tag exactly
+    request_bytes = request.to_bytes()
+
+    host, port = _split_controller(controller)
+    reply_bytes = _send_via(request_bytes, cert_dir)(host, port)
+
+    with mod.ProbeConfigTagReply.from_bytes(reply_bytes) as reply:
+        current_generation = reply.configTag
+
+    return DryRunResult(
+        current_generation=current_generation,
+        would_be_generation=current_generation + 1,
+        accepted=True,
+        reason="",
+    )

src/xceadmin/tests/test_config_push_mock.py

@@ -0,0 +1,270 @@
+"""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 ReplyDecodeError
+from xcecommon.schema import load_schema
+from xceadmin import push
+
+REPO_HA_DATA_DIR = Path(__file__).resolve().parents[3] / "ha" / "data"
+CERT_FILE = REPO_HA_DATA_DIR / "xce.crt"
+KEY_FILE = REPO_HA_DATA_DIR / "xce.key"
+CERT_DIR = str(REPO_HA_DATA_DIR)
+
+pytestmark = pytest.mark.skipif(
+    not (CERT_FILE.is_file() and KEY_FILE.is_file()),
+    reason="ha/data/xce.crt or ha/data/xce.key not present",
+)
+
+
+def _server_context() -> 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):
+        self._handler = handler
+        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().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():
+    servers: list[_MockController] = []
+
+    def _make(handler) -> _MockController:
+        srv = _MockController(handler)
+        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):
+    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, reply.to_bytes())
+
+    srv = mock_controller(_handler)
+
+    result = push.push_config(cluster_bin, f"127.0.0.1:{srv.port}", CERT_DIR)
+
+    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):
+    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, reply.to_bytes())
+
+    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, reply.to_bytes())
+
+    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}", CERT_DIR)
+
+    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):
+    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, reply.to_bytes())
+
+    srv = mock_controller(_handler)
+
+    result = push.push_config(cluster_bin, f"127.0.0.1:{srv.port}", CERT_DIR)
+
+    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):
+    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, reply.to_bytes())
+
+    srv = mock_controller(_handler)
+
+    with pytest.raises(ReplyDecodeError):
+        push.push_config(cluster_bin, f"127.0.0.1:{srv.port}", CERT_DIR)
+
+
+def test_push_config_default_port_used_when_omitted():
+    host, port = push._split_controller("somehost")
+    assert host == "somehost"
+    assert port == 9991
+
+
+def test_push_config_explicit_port_parsed():
+    host, port = push._split_controller("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):
+    mod = load_schema()
+    received = {}
+
+    def _handler(conn):
+        request_bytes = wire.recv_frame(conn)
+        received["request_bytes"] = request_bytes
+        with mod.ProbeConfigTagRequest.from_bytes(request_bytes) as req:
+            received["name"] = req.name
+
+        reply = mod.ProbeConfigTagReply.new_message()
+        reply.configTag = 41
+        wire.send_frame(conn, reply.to_bytes())
+
+    srv = mock_controller(_handler)
+
+    result = push.probe_config(cluster_bin, f"127.0.0.1:{srv.port}", CERT_DIR)
+
+    assert result.current_generation == 41
+    assert result.would_be_generation == 42
+    assert result.accepted is True
+    assert result.reason == ""
+    assert received["name"] == ""  # matches config_repl.c's xce_probe_config_tag exactly
+
+
+def test_probe_config_zero_generation_still_accepted(mock_controller, cluster_bin):
+    """A fresh cluster with no config committed yet (configTag == 0) is
+    still a successful probe -- there is no rejection signal on this
+    wire shape (module docstring); the caller decides significance.
+    """
+    mod = load_schema()
+
+    def _handler(conn):
+        wire.recv_frame(conn)
+        reply = mod.ProbeConfigTagReply.new_message()
+        reply.configTag = 0
+        wire.send_frame(conn, reply.to_bytes())
+
+    srv = mock_controller(_handler)
+
+    result = push.probe_config(cluster_bin, f"127.0.0.1:{srv.port}", CERT_DIR)
+
+    assert result.current_generation == 0
+    assert result.would_be_generation == 1
+    assert result.accepted is True