lei.chen/myxce
1
0
Fork 1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

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

Browse source 
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).
This commit is contained in:
Rongsong Shen 2026-08-06 04:43:40 +00:00
parent 30ff995cac
commit b64875f4c3
Signed by: rshen
SSH key fingerprint: SHA256:8Vf2pZ3h6rJa7EhGK+WaAQ9Mta/apZvteW6g9U/WKKk
5 changed files with 677 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

3
.kiro/specs/scheduler-core/tasks.md
View file

@ -273,12 +273,13 @@
- _Depends: 10.1, 9.4_
- _Verified: 2026-08-05 — audited task 10.1's own `test_sched_admin_dispatch.c` first: standby rejection and disconnected-scheduler rejection were already fully covered (confirmed unchanged). Two genuine gaps closed: (1) forward timeout — no test previously drove the real 5s `pthread_cond_timedwait` deadline to expiry. Added an `#ifdef XCE_TESTING`-only timeout-override seam in `cluster.c` (`SCHEDADMIN_TIMEOUT_MS` macro + `schedadmin_set_timeout_ms_for_test()`), verified genuinely inert in production builds: `-DXCE_TESTING` is set on exactly the 3 unit-test meson targets, absent from `xce_controller`'s own target and from every project-wide arg; the non-test `#else` branch is a bare `5000L` literal with the test-only global/setter entirely absent from that translation unit; deadline math confirmed correct for both the 5000ms default (behaviorally identical to the original `deadline.tv_sec += 5`) and the test's 200ms override (correct carry logic). `test_schedadmin_forward_timeout` sets a 200ms override, forwards to a real unconnected `sched.ep` with nothing ever replying, brackets the call with real `CLOCK_MONOTONIC` timestamps proving the wait genuinely blocked, and asserts the exact `{status=0, ok=false, error="request timeout"}` shape from requirements.md AC 10 — with the override reset to 5000 strictly before any `TEST_ASSERT` call (verified line-by-line, no `longjmp`-induced test-pollution risk). (2) "Round trip against the real scheduler-side handler" — investigated a literal two-binary integration (controller test invoking the scheduler's real `scheduler_req_handler()` in-process) and judged it architecturally awkward and out of scope (separate meson targets, incompatible fixture types, real symbol-collision risk) with no net correctness gain, since task 9.4's own `test_admin_request.c` already white-box-drives the real scheduler handler rigorously. Instead added two more round-trip outcome-branch tests using the same real `xce_encode`/`xce_decode` wire functions production uses: `test_schedadmin_round_trip_rpc_status_only` (a genuinely new C branch — reply with no `SchedAdminReply` payload) and `test_schedadmin_round_trip_business_error` (same branch as 10.1's existing success test, but proving `ok=false`/error-string field-passthrough fidelity — accurately self-described as payload coverage, not overclaimed as a new branch). Review independently verified every claim against source (confirmed `-DXCE_TESTING`'s exact meson scoping, confirmed sequential single-threaded Unity test execution rules out any cross-test race on the test-only global, confirmed `git diff` exactly matches `git diff -w` for `cluster.c` — zero reformat noise, learning correctly applied from task 10.1's cleanup). Controller suite 3/3 (1093/1093 Unity assertions, up from 1090), scheduler suite 26/26 unaffected, clang-format clean — independently re-verified by me via fresh `meson compile`/`meson test --suite controller`/`meson test --suite scheduler` runs before marking complete._
- [ ] 10.3 Build the xceadmin scheduler recording commands
- [x] 10.3 Build the xceadmin scheduler recording commands
- Implement record-start, record-stop, and record-status client functions that connect to a controller, follow redirects to the primary if needed, send the corresponding admin request, and decode the reply
- Observable: each function returns a decoded reply struct on success and raises a typed error on connection failure, decode failure, or an error reply
- _Requirements: 5.10_
- _Depends: 10.1_
- _Boundary: XceAdminSchedulerCLI — not (P): 10.3/10.4 both add functions to the same new scheduler.py module; run sequentially to avoid concurrent edits to one file_
- _Verified: 2026-08-05 — new `src/xceadmin/src/xceadmin/scheduler.py` (design.md's exact target file/function names) implements `record_start(controller, filename=None)`/`record_stop(controller)`/`record_status(controller)`, all funneled through one shared `_send_admin_request()`: build `SchedAdminRequest` via the real bundled schema (`request.opcode = "schedAdmin"`, `request.body.init("schedadminmsg")`) → `xceadmin.discovery.follow_redirects` (redirect-to-primary, mirrors `push_config`) → `xceadmin.discovery.decode_reply` → extract `reply.data.schedadminreplymsg`. Deliberate, documented divergence from `push_config()`'s "return failure as data" convention: this task's own Observable text explicitly says "raises a typed error on ... an error reply", so a `SchedAdminReply{ok=false}` raises a new `SchedAdminError` (`xcecommon/errors.py`, subclasses `XceAdminError`, carries `error`/`recording`/`file`/`event_count`/`started_at` — the full reply state, not just the reason string, since e.g. "already recording" still reports meaningful current-file state per requirements.md's Reply field semantics table); connection failures (`ConnectionTimeoutError`/`NoPrimaryError`/`TooManyRedirectsError`) and decode failures (`ReplyDecodeError`) reuse the existing typed errors `follow_redirects`/`decode_reply` already raise, unchanged. On success returns a new frozen `RecordResult` dataclass (`recording`/`file`/`event_count`/`started_at`; `ok`/`error` omitted as trivially-true/empty post-raise). `test_errors.py` extended with `SchedAdminError` in `ALL_SUBCLASSES` plus its own carries-state test (matching the existing exhaustive per-subclass pattern). New `test_scheduler.py` (mirrors `test_config_push_mock.py`'s mock-TLS-server pattern): success for all 3 actions (incl. explicit vs. auto-generated filename), one redirect-following case (`record_start` standby→primary), `ok=false` raising `SchedAdminError` for both `record_start`("already recording") and `record_stop`("not recording") plus a parametrized sweep of the Error codes reference table's other reasons, wrong-reply-variant → `ReplyDecodeError`, and a monkeypatched connection-timeout → `NoPrimaryError` (composition test, same shape as `push_config`'s own). 188 passed / 3 skipped (pre-existing live-cluster E2E tests, unaffected) via `meson compile -C xce.build xceadmin` — the real xceadmin test-running command (custom_target wrapping `meson-build.sh`'s pytest run; confirmed no `meson test --suite xceadmin` exists, `test()` is never used for this Python-only subdir). Noted the bundled schema copy under the live checkout (`src/xceadmin/src/xcecommon/schema_files/controller.capnp`) is an untracked, pre-existing-stale build artifact re-synced from `src/schema/controller.capnp` on every `meson compile` run — not touched, not a bug introduced by this task (see cerebrum.md). Independent adversarial review confirmed every claim above against source and a fresh test run (188/188+3 skipped, exact match), including the raise-vs-return divergence being a defensible, well-justified reading of this task's own binding Observable text (not a misreading of `push_config()`'s convention) — APPROVED, no remediation required. One non-blocking coverage gap noted for a future pass: `_send_admin_request()`'s `reply.status != 0` defensive branch has no dedicated test, unlike `push_config()`'s analogous coverage. This entry was originally self-authored by the implementer rather than the orchestrator (a process deviation, flagged and reviewed accordingly) — content re-verified accurate and left in place rather than rewritten._
- [ ] 10.4 Build the xceadmin scheduler dump command
- Implement a local-file decoder that reads a replay file's message stream and formats it to human-readable text — no network connection involved

193
src/xceadmin/src/xceadmin/scheduler.py Normal file
View file

@ -0,0 +1,193 @@
"""``xceadmin scheduler`` client functions (design.md, ``xceadmin.scheduler``;
Requirements 5.10, 5.20).
This module is the shared home for the scheduler recording-control client
functions (``record_start``/``record_stop``/``record_status``, task 10.3)
AND the local ``.replay`` file decoder (``dump()``, task 10.4, NOT built
here) -- design.md's File Structure Plan places both in this one file,
mirroring ``push.py``'s split (all business logic here, ``cli.py`` stays a
thin ``click`` layer -- task 10.5, also not built here).
``record_start``/``record_stop``/``record_status`` mirror ``push.py``'s
``push_config()`` pipeline exactly: build request bytes with the real
``controller.capnp`` schema (``xcecommon.schema.load_schema()``) -> connect
+ follow redirects to the primary (``xceadmin.discovery.follow_redirects``,
Requirement 3.1-3.4 -- the scheduler is a singleton attached to the
primary controller, so these commands must reach it the same way ``config
push`` reaches the primary) -> decode the ``Request{opcode: reply,
body.replymsg: RequestReply}`` envelope (``xceadmin.discovery.decode_reply``)
-> extract the ``SchedAdminReply`` payload. All three actions share one
``SchedAdminRequest``/``SchedAdminReply`` wire shape
(``controller.capnp``'s ``schedadminmsg``/``schedadminreplymsg`` union
members, added in task 7.4) with only ``action`` (and, for ``recordStart``
only, ``recordFile``) varying -- so all three go through one shared
``_send_admin_request()`` helper rather than three near-identical
functions.
**Error-vs-return convention -- a deliberate divergence from ``push.py``**:
``push_config()`` returns an application-level failure (``status != 0`` or
a non-empty ``UpgradeConfigReply.error``) as data on ``PushResult`` for the
CLI layer to interpret (design.md's `xceadmin.push` Boundary Commitments;
`cli.py` checks ``result.error or result.acked_count <= 0`` itself). Task
10.3's Observable is written differently and explicitly: "raises a typed
error on connection failure, decode failure, or **an error reply**" --
naming the business-level ``ok=false`` case as a THIRD raising failure
mode, on equal footing with connection/decode failures, not a fourth
"return it and let the caller decide" case. This module follows that
literal instruction rather than ``push_config``'s precedent: a
``SchedAdminReply{ok=false, error=...}`` raises
:class:`~xcecommon.errors.SchedAdminError` (new, ``xcecommon.errors``)
carrying the full reply state (``error``, ``recording``, ``file``,
``event_count``, ``started_at``) so a caller that wants to report it (e.g.
"already recording -- current file: X") does not need a second
``record_status()`` round-trip. Connection failures
(:class:`~xcecommon.errors.ConnectionTimeoutError` /
:class:`~xcecommon.errors.NoPrimaryError` /
:class:`~xcecommon.errors.TooManyRedirectsError`) and decode failures
(:class:`~xcecommon.errors.ReplyDecodeError`) already raise by construction
of ``follow_redirects``/``decode_reply`` -- reused as-is, not reimplemented.
On success, the decoded ``SchedAdminReply`` is returned as
:class:`RecordResult` -- ``recording``/``file``/``event_count``/
``started_at`` only (``ok``/``error`` are omitted: since a
``ok=false`` reply now always raises, they would be trivially
``True``/``""`` on every successful return, matching ``DryRunResult``'s
precedent of re-deriving semantic fields rather than mirroring the raw
wire struct verbatim).
"""
from __future__ import annotations
from dataclasses import dataclass
from xcecommon import wire
from xcecommon.errors import ReplyDecodeError, SchedAdminError
from xcecommon.schema import load_schema
from xceadmin.discovery import decode_reply, follow_redirects
_DEFAULT_PORT = 9991
@dataclass(frozen=True)
class RecordResult:
recording: bool
file: str
event_count: int
started_at: int
def _split_controller(controller: str) -> tuple[str, int]:
"""``"host:port"`` -> ``(host, port)``; a bare host uses the default
port 9991 -- same convention as ``push.py``'s own ``_split_controller``
(each resource module owns this trivial parsing rather than sharing a
private helper across modules, matching this package's existing
per-module-self-contained precedent).
"""
if ":" in controller:
host, port_str = controller.rsplit(":", 1)
return host, int(port_str)
return controller, _DEFAULT_PORT
def _send_via(request_bytes: bytes):
def _send(host: str, port: int) -> bytes:
sock = wire.connect(host, port)
try:
return wire.send_request(sock, request_bytes)
finally:
sock.close()
return _send
def _build_request(action: str, record_file: str = "") -> bytes:
mod = load_schema()
request = mod.Request.new_message()
request.opcode = "schedAdmin"
request.seq = 1
admin = request.body.init("schedadminmsg")
admin.action = action
admin.recordFile = record_file
return request.to_bytes()
def _send_admin_request(controller: str, action: str, record_file: str = "") -> RecordResult:
request_bytes = _build_request(action, record_file)
host, port = _split_controller(controller)
_, _, reply_bytes = follow_redirects(host, port, _send_via(request_bytes))
with decode_reply(reply_bytes) as reply:
if reply.status != 0:
# Defensive, mirrors push_config()'s own status-first check:
# a non-zero RequestReply.status means reply.data may not even
# be the schedadminreplymsg variant (e.g. a bare debugmsg), so
# this is reported before ever inspecting reply.data.which().
reason = f"controller returned status {reply.status}"
raise SchedAdminError(
f"scheduler admin request failed: {reason}",
error=reason,
recording=False,
file="",
event_count=0,
started_at=0,
)
if reply.data.which() != "schedadminreplymsg":
reason = f"unexpected reply variant {reply.data.which()!r}, expected schedadminreplymsg"
raise ReplyDecodeError(f"invalid reply: {reason}", reason)
admin_reply = reply.data.schedadminreplymsg
if not admin_reply.ok:
raise SchedAdminError(
f"scheduler admin request failed: {admin_reply.error}",
error=admin_reply.error,
recording=admin_reply.recording,
file=admin_reply.file,
event_count=admin_reply.eventCount,
started_at=admin_reply.startedAt,
)
return RecordResult(
recording=admin_reply.recording,
file=admin_reply.file,
event_count=admin_reply.eventCount,
started_at=admin_reply.startedAt,
)
def record_start(controller: str, filename: str | None = None) -> RecordResult:
"""Start scheduler recording (``xceadmin scheduler record start
[FILE]``): ``SchedAdminRequest{action=recordStart, recordFile=filename
or ""}`` (requirements.md AC 10 -- empty ``recordFile`` means the
scheduler auto-generates a name).
Raises :class:`~xcecommon.errors.SchedAdminError` if already recording
(module docstring); connection/decode failures raise their own
existing typed errors (``xceadmin.discovery``).
"""
return _send_admin_request(controller, "recordStart", filename or "")
def record_stop(controller: str) -> RecordResult:
"""Stop scheduler recording (``xceadmin scheduler record stop``):
``SchedAdminRequest{action=recordStop}``.
Raises :class:`~xcecommon.errors.SchedAdminError` if not currently
recording (module docstring).
"""
return _send_admin_request(controller, "recordStop")
def record_status(controller: str) -> RecordResult:
"""Query scheduler recording status (``xceadmin scheduler record
status``): ``SchedAdminRequest{action=recordStatus}``.
``recordStatus`` always replies ``ok=true`` (requirements.md's Reply
field semantics table has no error row for this action), so this only
raises for connection/decode failures in practice -- kept behind the
same shared helper as ``record_start``/``record_stop`` rather than a
special-cased fourth code path, matching the "Reply field semantics by
action" table's own framing of all three actions as one contract.
"""
return _send_admin_request(controller, "recordStatus")

39
src/xceadmin/src/xcecommon/errors.py
View file

@ -127,3 +127,42 @@ class NoPrimaryError(XceAdminError):
super().__init__(message)
self.host = host
self.port = port
class SchedAdminError(XceAdminError):
"""A ``SchedAdminRequest`` (scheduler recording control) round-tripped
successfully but the reply itself reports failure --
``SchedAdminReply{ok=false, error=...}`` (Requirement 5.10; declared
here, raised by a later task -- see ``xceadmin.scheduler``).
Unlike ``config push``'s ``PushResult``/``DryRunResult`` (which return
an application-level failure as data for the caller to inspect), the
scheduler recording commands' task text is explicit: "raises a typed
error on connection failure, decode failure, or an error reply" --
three failure modes treated uniformly. This is a deliberate,
documented divergence from ``push_config``'s convention, not an
oversight (see ``xceadmin.scheduler`` module docstring).
Carries the full reply state alongside ``error``, not just the reason
string -- requirements.md's "Reply field semantics by action" table
shows a business-level failure (e.g. "already recording") still
reports meaningful ``recording``/``file`` state, so a caller that
wants to display it (e.g. "already recording -- current file: X")
does not have to re-query ``record status`` separately.
"""
def __init__(
self,
message: str,
error: str,
recording: bool,
file: str,
event_count: int,
started_at: int,
) -> None:
super().__init__(message)
self.error = error
self.recording = recording
self.file = file
self.event_count = event_count
self.started_at = started_at

33
src/xceadmin/tests/test_errors.py
View file

@ -21,6 +21,7 @@ from xcecommon.errors import (
NoPrimaryError,
ReplyDecodeError,
ReservedNameError,
SchedAdminError,
SchemaNotFoundError,
TooManyRedirectsError,
ValidationError,
@ -36,6 +37,7 @@ ALL_SUBCLASSES = [
ReplyDecodeError,
TooManyRedirectsError,
NoPrimaryError,
SchedAdminError,
]
@ -171,6 +173,31 @@ def test_no_primary_error_carries_host_and_port() -> None:
assert caught.port == 9991
def test_sched_admin_error_carries_error_and_reply_state() -> None:
"""Requirement 5.10 (declared here, raised by a later task): a
``SchedAdminReply{ok=false, error=...}`` is surfaced as a typed error
carrying the reason plus the reply's recording/file/event/started-at
state (e.g. "already recording" still reports the current file)."""
with pytest.raises(XceAdminError) as exc_info:
raise SchedAdminError(
"scheduler admin request failed: already recording",
error="already recording",
recording=True,
file="/var/log/xce/current.replay",
event_count=42,
started_at=1234567890,
)
caught = exc_info.value
assert isinstance(caught, SchedAdminError)
assert caught.message == "scheduler admin request failed: already recording"
assert caught.error == "already recording"
assert caught.recording is True
assert caught.file == "/var/log/xce/current.replay"
assert caught.event_count == 42
assert caught.started_at == 1234567890
def test_catching_base_type_recovers_message_for_every_subclass() -> None:
"""The entire point of the shared exception contract (design.md,
xceadmin.cli Implementation Note): a single `except XceAdminError as e`
@ -186,6 +213,12 @@ def test_catching_base_type_recovers_message_for_every_subclass() -> None:
(ReplyDecodeError("msg-f", reason="r"), "msg-f"),
(TooManyRedirectsError("msg-g", hops=3), "msg-g"),
(NoPrimaryError("msg-h", host="h", port=1), "msg-h"),
(
SchedAdminError(
"msg-i", error="e", recording=False, file="", event_count=0, started_at=0
),
"msg-i",
),
]
for exc, expected_message in cases:

410
src/xceadmin/tests/test_scheduler.py Normal file
View file

@ -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