test(scheduler-core): xceadmin scheduler CLI end-to-end mock-server tests (task 10.6)

Add real CliRunner-through-real-mock-TLS-server coverage for scheduler
record start/stop/status (incl. standby->primary redirect handoff) and
dump, closing the gap task 10.5 left open (which only monkeypatched the
CLI's function bindings). 227/227 xceadmin tests green. Closes Group 10.

src/xceadmin/tests/test_cli_scheduler_mock.py

@@ -0,0 +1,377 @@
+"""Full-stack CLI + mock-TLS-server tests for the ``scheduler`` command
+group (design.md, ``XceAdminSchedulerCLI``; task 10.6, Requirement 5.10).
+
+Every ``scheduler``-related test in ``test_cli.py`` monkeypatches the
+imported ``record_start``/``record_stop``/``record_status``/``dump``
+bindings on ``xceadmin.cli`` directly (task 10.5's own coverage,
+deliberately left at the CLI-wiring layer only -- see that file's own
+comment and task 10.5's tasks.md ``Verified`` note: "full mock-server/
+redirect coverage is task 10.6's job"). This file closes that gap: it
+invokes the REAL ``click.testing.CliRunner`` entry points
+(``scheduler record start/stop/status``, ``scheduler dump``) with NO
+monkeypatching of ``xceadmin.scheduler``'s functions, backed by a real
+local TLS mock server (record commands) or a real fixture ``.replay``
+file on disk (dump). This proves CLI arg parsing -> ``scheduler.py`` ->
+``xceadmin.discovery.follow_redirects``/``decode_reply`` ->
+``xcecommon.wire`` actually compose end to end through the click entry
+points, including at least one real standby -> primary redirect handoff.
+
+Does not re-test ``scheduler.py``'s own request/reply/error-mapping logic
+in isolation (``test_scheduler.py``'s mock-server tests already cover
+that at the function level) or ``cli.py``'s argument-parsing/error-
+translation logic in isolation (``test_cli.py``'s monkeypatch section
+already covers that) -- only their real composition.
+
+Split into its own file (rather than appended to ``test_cli.py``) to keep
+both files under this project's 800-line file-size guideline -- mirrors
+this suite's existing precedent of ``test_config_push_mock.py`` being
+split out from ``test_cli.py``'s ``config`` tests, and each mock-server
+test file (``test_config_push_mock.py``, ``test_scheduler.py``, this
+file) being self-contained with its own local ``_MockController``/
+``mock_controller`` fixture rather than a shared one (``test_scheduler.py``
+module docstring: "not shared across test files since pytest fixtures/
+helpers are conventionally module-local in this test suite").
+"""
+
+from __future__ import annotations
+
+import ssl
+import socket
+import threading
+from pathlib import Path
+
+import pytest
+from click.testing import CliRunner
+
+from xcecommon import wire
+from xcecommon.schema import load_schema
+from xceadmin.cli import cli
+
+
+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``/``test_scheduler.py``'s own
+    ``_MockController`` -- kept module-local per this suite's existing
+    convention (see module docstring).
+    """
+
+    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()
+
+
+# ---------------------------------------------------------------------------
+# scheduler record start/stop/status: real CLI entry point, real mock
+# server -- happy path, redirect-following, and one error case.
+# ---------------------------------------------------------------------------
+
+
+def test_cli_scheduler_record_start_against_real_mock_server(mock_controller, wrap_reply) -> None:
+    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-cli-test.replay"
+        admin_reply.eventCount = 0
+        admin_reply.startedAt = 1000
+        wire.send_frame(conn, wrap_reply(reply))
+
+    srv = mock_controller(_handler)
+
+    runner = CliRunner()
+    result = runner.invoke(cli, ["scheduler", "record", "start", "--controller", f"127.0.0.1:{srv.port}"])
+
+    assert result.exit_code == 0, result.output
+    assert received["opcode"] == "schedAdmin"
+    assert received["action"] == "recordStart"
+    assert received["recordFile"] == ""
+    assert "recording:   yes" in result.output
+    assert "/var/log/xce/scheduler-replay-cli-test.replay" in result.output
+
+
+def test_cli_scheduler_record_start_follows_redirect_to_primary(mock_controller, wrap_reply) -> None:
+    """The Observable's required redirect-following case: a standby-first,
+    then-primary handoff exercised through the real CLI entry point end to
+    end. Not scheduler.py directly (test_scheduler.py's own
+    ``test_record_start_follows_one_redirect_to_primary`` already covers
+    that at the function level) and not monkeypatched (test_cli.py's own
+    section already covers plain CLI delegation) -- this is the
+    intersection: real CLI invocation, real redirect handshake, two real
+    mock servers.
+    """
+    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-via-cli.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)
+
+    runner = CliRunner()
+    result = runner.invoke(
+        cli, ["scheduler", "record", "start", "--controller", f"127.0.0.1:{standby_srv.port}"]
+    )
+
+    assert result.exit_code == 0, result.output
+    assert "on-primary-via-cli.replay" in result.output
+
+
+def test_cli_scheduler_record_stop_against_real_mock_server(mock_controller, wrap_reply) -> None:
+    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-via-cli.replay"
+        admin_reply.eventCount = 14273
+        admin_reply.startedAt = 1000
+        wire.send_frame(conn, wrap_reply(reply))
+
+    srv = mock_controller(_handler)
+
+    runner = CliRunner()
+    result = runner.invoke(cli, ["scheduler", "record", "stop", "--controller", f"127.0.0.1:{srv.port}"])
+
+    assert result.exit_code == 0, result.output
+    assert received["action"] == "recordStop"
+    assert "recording:   no" in result.output
+    assert "events:      14273" in result.output
+
+
+def test_cli_scheduler_record_status_against_real_mock_server(mock_controller, wrap_reply) -> None:
+    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-via-cli.replay"
+        admin_reply.eventCount = 500
+        admin_reply.startedAt = 4000
+        wire.send_frame(conn, wrap_reply(reply))
+
+    srv = mock_controller(_handler)
+
+    runner = CliRunner()
+    result = runner.invoke(cli, ["scheduler", "record", "status", "--controller", f"127.0.0.1:{srv.port}"])
+
+    assert result.exit_code == 0, result.output
+    assert received["action"] == "recordStatus"
+    assert "recording:   yes" in result.output
+    assert "events:      500" in result.output
+
+
+def test_cli_scheduler_record_start_already_recording_error_end_to_end(mock_controller, wrap_reply) -> None:
+    """Proves the error path also composes end to end through the real CLI
+    + real mock server (SchedAdminError enrichment via
+    ``_format_sched_admin_error``), not just the happy path -- complements
+    ``test_cli.py``'s
+    ``test_scheduler_record_start_already_recording_exits_nonzero_with_rich_message``
+    by proving the real ``scheduler.py`` raise site is what feeds the
+    CLI's formatter, not a monkeypatched stand-in for it.
+    """
+    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-via-cli.replay"
+        admin_reply.eventCount = 7
+        admin_reply.startedAt = 5000
+        wire.send_frame(conn, wrap_reply(reply))
+
+    srv = mock_controller(_handler)
+
+    runner = CliRunner()
+    result = runner.invoke(cli, ["scheduler", "record", "start", "--controller", f"127.0.0.1:{srv.port}"])
+
+    assert result.exit_code != 0
+    assert "already recording" in result.output
+    assert "current-via-cli.replay" in result.output
+
+
+# ---------------------------------------------------------------------------
+# scheduler dump: real CLI entry point, real fixture .replay file on disk
+# (no network, no mock server -- dump() has no network dependency).
+# ---------------------------------------------------------------------------
+
+
+def _build_minimal_replay_stream(mod) -> bytes:
+    """A minimal, real (not monkeypatched) ``.replay`` byte stream --
+    ``ReplayHeader`` + initial ``SnapshotEntry`` only, no trailing
+    events/sessions -- built with pycapnp's write-side API, the same
+    technique ``test_scheduler.py``'s own ``_build_full_replay_stream()``
+    uses. Deliberately NOT reused/imported from there: this test only
+    needs to prove the CLI reads a real file end to end through the real
+    ``dump()``, not re-exercise every message kind ``dump()`` can decode --
+    that is already covered exhaustively by ``test_scheduler.py``'s own
+    golden-file tests. List fields left unset (jobs/queues/workers/
+    resources/resvals/users/projects) read back as empty per Cap'n Proto's
+    own default-value semantics for an unset list pointer -- no explicit
+    ``.init(..., 0)`` call is needed to make that safe to decode.
+    """
+    parts: list[bytes] = []
+
+    hdr = mod.ReplayHeader.new_message()
+    hdr.recorderVersion = "1.0"
+    hdr.recordingStartTs = 1000
+    hdr.recordingStartWall = "2026-08-06T00:00:00Z"
+    hdr.schedPolicy = "builtin_fcfs"
+    hdr.cluster.name = "cli-test-cluster"
+    hdr.cluster.recordingHost = "ctrl-1"
+    hdr.cluster.init("controllers", 1)
+    hdr.cluster.controllers[0] = "127.0.0.1:9991"
+    parts.append(hdr.to_bytes())
+
+    snap = mod.SnapshotEntry.new_message()
+    snap.ts = 1000
+    snap.data.config.clustername = "cli-test-cluster"
+    snap.data.prngSeed = 7
+    snap.data.internalVersion = 1
+    parts.append(snap.to_bytes())
+
+    return b"".join(parts)
+
+
+def test_cli_scheduler_dump_against_real_fixture_replay_file(tmp_path) -> None:
+    """``scheduler dump`` invoked through the real CLI entry point against a
+    real ``.replay`` file on disk -- no monkeypatching of ``dump()``
+    (unlike ``test_cli.py``'s own
+    ``test_scheduler_dump_echoes_returned_text_exactly``), proving the CLI
+    actually calls the real decoder in ``scheduler.py`` and prints its
+    real output.
+    """
+    mod = load_schema()
+    fixture = tmp_path / "cli-fixture.replay"
+    fixture.write_bytes(_build_minimal_replay_stream(mod))
+
+    runner = CliRunner()
+    result = runner.invoke(cli, ["scheduler", "dump", str(fixture)])
+
+    assert result.exit_code == 0, result.output
+    assert "=== ReplayHeader ===" in result.output
+    assert "recorderVersion: 1.0" in result.output
+    assert "cluster.name: cli-test-cluster" in result.output
+    assert "=== SnapshotEntry (initial) ===" in result.output
+    assert "prngSeed: 7" in result.output
+
+
+def test_cli_scheduler_dump_missing_file_against_real_decoder_exits_nonzero(tmp_path) -> None:
+    """Same real-decoder, no-monkeypatch proof for the error path: a real
+    missing file through the real ``dump()`` raises ``ReplayFileError``,
+    which the CLI must still translate into a clean ``ClickException``
+    rather than a raw traceback.
+    """
+    missing = tmp_path / "does-not-exist.replay"
+
+    runner = CliRunner()
+    result = runner.invoke(cli, ["scheduler", "dump", str(missing)])
+
+    assert result.exit_code != 0
+    assert not isinstance(result.exception, KeyError | AttributeError | TypeError | FileNotFoundError)
+    assert str(missing) in result.output