feat(scheduler-core): build xceadmin scheduler recording client functions (task 10.3)

Adds record_start()/record_stop()/record_status() to new
src/xceadmin/src/xceadmin/scheduler.py, following push_config()'s exact
pipeline shape (load_schema -> follow_redirects -> decode_reply). Deliberately
diverges from push_config()'s "return failure as data" convention because
this task's own Observable text requires raising a typed error on an error
reply, not just connection/decode failure -- a new SchedAdminError
(xcecommon/errors.py) carries the full reply state (error/recording/file/
event_count/started_at) when SchedAdminReply.ok is false.

test_scheduler.py mirrors test_config_push_mock.py's real mock-TLS-server
technique, covering all three actions, redirect-following (standby-to-
primary), and the error-reply-raises-SchedAdminError path across the
error codes reference table.

188 passed, 3 skipped via the real xceadmin test runner (meson compile
-C xce.build xceadmin).

src/xceadmin/tests/test_scheduler.py

@@ -0,0 +1,410 @@
+"""Mock-controller tests for ``xceadmin.scheduler``'s record-control client
+functions (design.md, ``xceadmin.scheduler``; Requirements 5.10).
+
+Mirrors ``test_config_push_mock.py``'s mock TLS server pattern exactly
+(same ``mock_controller``/``wrap_reply``/``tls_certs`` fixtures from
+``conftest.py``) -- ``record_start``/``record_stop``/``record_status``
+have no live controller counterpart to test against yet (task 10.1's
+controller-side dispatch exists, but there is no running cluster in this
+test environment), so these are tested against a real local TLS mock
+server built from the actual bundled schema, exactly as
+``test_config_push_mock.py`` does for ``push_config``/``probe_config``.
+
+Only ``record_start``/``record_stop``/``record_status`` (task 10.3) are
+covered here. ``dump()`` (task 10.4, not yet built) will get its own
+golden-file tests appended to this file later (design.md's File Structure
+Plan places both in one ``test_scheduler.py``) -- not duplicated or
+anticipated here.
+"""
+
+from __future__ import annotations
+
+import ssl
+import socket
+import threading
+from pathlib import Path
+
+import pytest
+
+from xcecommon import wire
+from xcecommon.errors import NoPrimaryError, ReplyDecodeError, SchedAdminError
+from xcecommon.schema import load_schema
+from xceadmin import scheduler
+
+
+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. Identical to
+    ``test_config_push_mock.py``'s own ``_MockController`` -- not shared
+    across test files since pytest fixtures/helpers are conventionally
+    module-local in this test suite (each of ``test_config_push_mock.py``,
+    this file, etc. is self-contained).
+    """
+
+    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()
+
+
+# ---------------------------------------------------------------------------
+# record_start(): happy path, redirect-following
+# ---------------------------------------------------------------------------
+
+
+def test_record_start_auto_generated_filename(mock_controller, wrap_reply):
+    mod = load_schema()
+    received = {}
+
+    def _handler(conn):
+        request_bytes = wire.recv_frame(conn)
+        with mod.Request.from_bytes(request_bytes) as req:
+            received["opcode"] = req.opcode
+            received["action"] = req.body.schedadminmsg.action
+            received["recordFile"] = req.body.schedadminmsg.recordFile
+
+        reply = mod.RequestReply.new_message()
+        reply.status = 0
+        admin_reply = reply.data.init("schedadminreplymsg")
+        admin_reply.ok = True
+        admin_reply.error = ""
+        admin_reply.recording = True
+        admin_reply.file = "/var/log/xce/scheduler-replay-auto.replay"
+        admin_reply.eventCount = 0
+        admin_reply.startedAt = 1000
+        wire.send_frame(conn, wrap_reply(reply))
+
+    srv = mock_controller(_handler)
+
+    result = scheduler.record_start(f"127.0.0.1:{srv.port}")
+
+    assert received["opcode"] == "schedAdmin"
+    assert received["action"] == "recordStart"
+    assert received["recordFile"] == ""
+    assert result.recording is True
+    assert result.file == "/var/log/xce/scheduler-replay-auto.replay"
+    assert result.event_count == 0
+    assert result.started_at == 1000
+
+
+def test_record_start_explicit_filename_sent_on_wire(mock_controller, wrap_reply):
+    mod = load_schema()
+    received = {}
+
+    def _handler(conn):
+        request_bytes = wire.recv_frame(conn)
+        with mod.Request.from_bytes(request_bytes) as req:
+            received["recordFile"] = req.body.schedadminmsg.recordFile
+
+        reply = mod.RequestReply.new_message()
+        reply.status = 0
+        admin_reply = reply.data.init("schedadminreplymsg")
+        admin_reply.ok = True
+        admin_reply.error = ""
+        admin_reply.recording = True
+        admin_reply.file = "myfile.replay"
+        admin_reply.eventCount = 0
+        admin_reply.startedAt = 2000
+        wire.send_frame(conn, wrap_reply(reply))
+
+    srv = mock_controller(_handler)
+
+    result = scheduler.record_start(f"127.0.0.1:{srv.port}", filename="myfile.replay")
+
+    assert received["recordFile"] == "myfile.replay"
+    assert result.file == "myfile.replay"
+
+
+def test_record_start_follows_one_redirect_to_primary(mock_controller, 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"
+        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
+        admin_reply = reply.data.init("schedadminreplymsg")
+        admin_reply.ok = True
+        admin_reply.error = ""
+        admin_reply.recording = True
+        admin_reply.file = "on-primary.replay"
+        admin_reply.eventCount = 0
+        admin_reply.startedAt = 3000
+        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 = scheduler.record_start(f"127.0.0.1:{standby_srv.port}")
+
+    assert result.file == "on-primary.replay"
+
+
+# ---------------------------------------------------------------------------
+# record_stop() / record_status(): happy path
+# ---------------------------------------------------------------------------
+
+
+def test_record_stop_successful(mock_controller, wrap_reply):
+    mod = load_schema()
+    received = {}
+
+    def _handler(conn):
+        request_bytes = wire.recv_frame(conn)
+        with mod.Request.from_bytes(request_bytes) as req:
+            received["action"] = req.body.schedadminmsg.action
+
+        reply = mod.RequestReply.new_message()
+        reply.status = 0
+        admin_reply = reply.data.init("schedadminreplymsg")
+        admin_reply.ok = True
+        admin_reply.error = ""
+        admin_reply.recording = False
+        admin_reply.file = "final.replay"
+        admin_reply.eventCount = 14273
+        admin_reply.startedAt = 1000
+        wire.send_frame(conn, wrap_reply(reply))
+
+    srv = mock_controller(_handler)
+
+    result = scheduler.record_stop(f"127.0.0.1:{srv.port}")
+
+    assert received["action"] == "recordStop"
+    assert result.recording is False
+    assert result.event_count == 14273
+    assert result.file == "final.replay"
+
+
+def test_record_status_successful(mock_controller, wrap_reply):
+    mod = load_schema()
+    received = {}
+
+    def _handler(conn):
+        request_bytes = wire.recv_frame(conn)
+        with mod.Request.from_bytes(request_bytes) as req:
+            received["action"] = req.body.schedadminmsg.action
+
+        reply = mod.RequestReply.new_message()
+        reply.status = 0
+        admin_reply = reply.data.init("schedadminreplymsg")
+        admin_reply.ok = True
+        admin_reply.error = ""
+        admin_reply.recording = True
+        admin_reply.file = "current.replay"
+        admin_reply.eventCount = 500
+        admin_reply.startedAt = 4000
+        wire.send_frame(conn, wrap_reply(reply))
+
+    srv = mock_controller(_handler)
+
+    result = scheduler.record_status(f"127.0.0.1:{srv.port}")
+
+    assert received["action"] == "recordStatus"
+    assert result.recording is True
+    assert result.event_count == 500
+
+
+# ---------------------------------------------------------------------------
+# ok=false replies raise SchedAdminError (task 10.3's Observable: "raises a
+# typed error on ... an error reply")
+# ---------------------------------------------------------------------------
+
+
+def test_record_start_already_recording_raises_sched_admin_error(mock_controller, wrap_reply):
+    mod = load_schema()
+
+    def _handler(conn):
+        wire.recv_frame(conn)
+        reply = mod.RequestReply.new_message()
+        reply.status = 0
+        admin_reply = reply.data.init("schedadminreplymsg")
+        admin_reply.ok = False
+        admin_reply.error = "already recording"
+        admin_reply.recording = True
+        admin_reply.file = "current.replay"
+        admin_reply.eventCount = 7
+        admin_reply.startedAt = 5000
+        wire.send_frame(conn, wrap_reply(reply))
+
+    srv = mock_controller(_handler)
+
+    with pytest.raises(SchedAdminError) as exc_info:
+        scheduler.record_start(f"127.0.0.1:{srv.port}")
+
+    caught = exc_info.value
+    assert caught.error == "already recording"
+    assert caught.recording is True
+    assert caught.file == "current.replay"
+    assert caught.event_count == 7
+    assert caught.started_at == 5000
+
+
+def test_record_stop_not_recording_raises_sched_admin_error(mock_controller, wrap_reply):
+    mod = load_schema()
+
+    def _handler(conn):
+        wire.recv_frame(conn)
+        reply = mod.RequestReply.new_message()
+        reply.status = 0
+        admin_reply = reply.data.init("schedadminreplymsg")
+        admin_reply.ok = False
+        admin_reply.error = "not recording"
+        admin_reply.recording = False
+        admin_reply.file = ""
+        admin_reply.eventCount = 0
+        admin_reply.startedAt = 0
+        wire.send_frame(conn, wrap_reply(reply))
+
+    srv = mock_controller(_handler)
+
+    with pytest.raises(SchedAdminError) as exc_info:
+        scheduler.record_stop(f"127.0.0.1:{srv.port}")
+
+    assert exc_info.value.error == "not recording"
+    assert exc_info.value.recording is False
+
+
+@pytest.mark.parametrize(
+    "error_reason", ["scheduler not connected", "request timeout", "invalid record file name"]
+)
+def test_record_start_error_codes_raise_sched_admin_error(mock_controller, wrap_reply, error_reason):
+    """requirements.md's "Error codes reference" table -- every listed
+    ``SchedAdminReply{ok=false, error=...}`` reason raises the same typed
+    error, regardless of which specific reason string the reply carries.
+    """
+    mod = load_schema()
+
+    def _handler(conn):
+        wire.recv_frame(conn)
+        reply = mod.RequestReply.new_message()
+        reply.status = 0
+        admin_reply = reply.data.init("schedadminreplymsg")
+        admin_reply.ok = False
+        admin_reply.error = error_reason
+        admin_reply.recording = False
+        admin_reply.file = ""
+        admin_reply.eventCount = 0
+        admin_reply.startedAt = 0
+        wire.send_frame(conn, wrap_reply(reply))
+
+    srv = mock_controller(_handler)
+
+    with pytest.raises(SchedAdminError) as exc_info:
+        scheduler.record_start(f"127.0.0.1:{srv.port}")
+
+    assert exc_info.value.error == error_reason
+
+
+# ---------------------------------------------------------------------------
+# Decode failure (wrong reply variant) raises ReplyDecodeError
+# ---------------------------------------------------------------------------
+
+
+def test_record_status_wrong_reply_variant_raises_reply_decode_error(mock_controller, 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 schedadminreplymsg
+        wire.send_frame(conn, wrap_reply(reply))
+
+    srv = mock_controller(_handler)
+
+    with pytest.raises(ReplyDecodeError):
+        scheduler.record_status(f"127.0.0.1:{srv.port}")
+
+
+# ---------------------------------------------------------------------------
+# Connection failure raises NoPrimaryError (end-to-end composition with
+# xceadmin.discovery, mirrors test_config_push_mock.py's own equivalent
+# test for push_config)
+# ---------------------------------------------------------------------------
+
+
+def test_record_start_propagates_no_primary_error_on_connection_timeout(monkeypatch):
+    from xcecommon.errors import ConnectionTimeoutError
+
+    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):
+        scheduler.record_start("127.0.0.1:9991")
+
+
+# ---------------------------------------------------------------------------
+# _split_controller(): same default-port convention as push.py
+# ---------------------------------------------------------------------------
+
+
+def test_record_default_port_used_when_omitted():
+    host, port = scheduler._split_controller("somehost")
+    assert host == "somehost"
+    assert port == 9991
+
+
+def test_record_explicit_port_parsed():
+    host, port = scheduler._split_controller("somehost:1234")
+    assert host == "somehost"
+    assert port == 1234