feat(admin-config-tool): task 1.2 - shared exception taxonomy
XceAdminError base plus 8 subclasses (schema-not-found, validation, reserved-name, duplicate-entry, connection-timeout, reply-decode, too-many-redirects, no-primary) giving the future CLI layer one dispatch surface instead of catching unrelated types per module. Requirements: 7.1, 7.2, 7.3, 7.4
This commit is contained in:
parent
b30a856a2d
commit
2e4ef4e209
3 changed files with 327 additions and 1 deletions
197
src/xceadmin/tests/test_errors.py
Normal file
197
src/xceadmin/tests/test_errors.py
Normal file
|
|
@ -0,0 +1,197 @@
|
|||
"""Tests for the xcecommon.errors exception hierarchy (task 1.2).
|
||||
|
||||
Covers Requirements 7.1, 7.2, 7.3, 7.4: every specific failure mode gets its
|
||||
own XceAdminError subclass, and catching the common base type recovers a
|
||||
human-readable message regardless of which subclass was actually raised —
|
||||
the single CLI dispatch surface described in design.md's "xceadmin.cli —
|
||||
Implementation Note" (`except XceAdminError as e: raise
|
||||
click.ClickException(e.message)`).
|
||||
|
||||
This task only builds the exception hierarchy itself — the code that would
|
||||
actually raise these exceptions (schema loading, YAML validation, wire
|
||||
protocol) lands in later tasks, so these tests raise each exception
|
||||
directly rather than driving it through real application logic.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
|
||||
from xcecommon.errors import (
|
||||
ConnectionTimeoutError,
|
||||
DuplicateEntryError,
|
||||
NoPrimaryError,
|
||||
ReplyDecodeError,
|
||||
ReservedNameError,
|
||||
SchemaNotFoundError,
|
||||
TooManyRedirectsError,
|
||||
ValidationError,
|
||||
XceAdminError,
|
||||
)
|
||||
|
||||
ALL_SUBCLASSES = [
|
||||
SchemaNotFoundError,
|
||||
ValidationError,
|
||||
ReservedNameError,
|
||||
DuplicateEntryError,
|
||||
ConnectionTimeoutError,
|
||||
ReplyDecodeError,
|
||||
TooManyRedirectsError,
|
||||
NoPrimaryError,
|
||||
]
|
||||
|
||||
|
||||
def test_base_exception_carries_message() -> None:
|
||||
"""XceAdminError itself carries a .message attribute, recoverable via
|
||||
both .message and str(e)."""
|
||||
with pytest.raises(XceAdminError) as exc_info:
|
||||
raise XceAdminError("something went wrong")
|
||||
|
||||
assert exc_info.value.message == "something went wrong"
|
||||
assert str(exc_info.value) == "something went wrong"
|
||||
|
||||
|
||||
@pytest.mark.parametrize("exc_type", ALL_SUBCLASSES)
|
||||
def test_every_subclass_is_a_xceadminerror(exc_type: type[XceAdminError]) -> None:
|
||||
"""Every specific failure mode subclasses the common base — this is what
|
||||
lets xceadmin.cli use one dispatch surface instead of catching 8
|
||||
unrelated types (design.md, xceadmin.cli Implementation Note)."""
|
||||
assert issubclass(exc_type, XceAdminError)
|
||||
assert issubclass(exc_type, Exception)
|
||||
|
||||
|
||||
def test_schema_not_found_error_carries_path_and_message() -> None:
|
||||
"""Requirement 7.4: missing controller.capnp reports the path so the
|
||||
operator can check the installation."""
|
||||
path = "/opt/xceadmin/schema_files/controller.capnp"
|
||||
with pytest.raises(XceAdminError) as exc_info:
|
||||
raise SchemaNotFoundError(f"schema not found: {path}", path=path)
|
||||
|
||||
caught = exc_info.value
|
||||
assert isinstance(caught, SchemaNotFoundError)
|
||||
assert caught.message == f"schema not found: {path}"
|
||||
assert caught.path == path
|
||||
|
||||
|
||||
def test_validation_error_carries_file_json_pointer_and_constraint() -> None:
|
||||
"""Requirement 4.1 / design.md xcecommon.validation: schema validation
|
||||
failures report file, JSON-pointer field path, and violated constraint."""
|
||||
with pytest.raises(XceAdminError) as exc_info:
|
||||
raise ValidationError(
|
||||
"queues.yaml: /queues/0/name does not match pattern",
|
||||
file="queues.yaml",
|
||||
json_pointer="/queues/0/name",
|
||||
constraint="pattern",
|
||||
)
|
||||
|
||||
caught = exc_info.value
|
||||
assert isinstance(caught, ValidationError)
|
||||
assert caught.message == "queues.yaml: /queues/0/name does not match pattern"
|
||||
assert caught.file == "queues.yaml"
|
||||
assert caught.json_pointer == "/queues/0/name"
|
||||
assert caught.constraint == "pattern"
|
||||
|
||||
|
||||
def test_reserved_name_error_recovers_message_via_base_type() -> None:
|
||||
"""Requirement 4.2: the reserved name "lost_and_found" is rejected with
|
||||
a clear message explaining the reservation."""
|
||||
message = '"lost_and_found" is reserved and cannot be used as a queue or host name'
|
||||
with pytest.raises(XceAdminError) as exc_info:
|
||||
raise ReservedNameError(message)
|
||||
|
||||
assert isinstance(exc_info.value, ReservedNameError)
|
||||
assert exc_info.value.message == message
|
||||
|
||||
|
||||
def test_duplicate_entry_error_recovers_message_via_base_type() -> None:
|
||||
"""Requirement 4.3: duplicate entries within a file/section are
|
||||
reported by name."""
|
||||
message = "duplicate queue name 'batch' in queues.yaml"
|
||||
with pytest.raises(XceAdminError) as exc_info:
|
||||
raise DuplicateEntryError(message)
|
||||
|
||||
assert isinstance(exc_info.value, DuplicateEntryError)
|
||||
assert exc_info.value.message == message
|
||||
|
||||
|
||||
def test_connection_timeout_error_carries_host_and_port() -> None:
|
||||
"""Requirement 7.6 (declared here, raised by a later task): "timeout
|
||||
connecting to <host:port>"."""
|
||||
with pytest.raises(XceAdminError) as exc_info:
|
||||
raise ConnectionTimeoutError(
|
||||
"timeout connecting to localhost:9991", host="localhost", port=9991
|
||||
)
|
||||
|
||||
caught = exc_info.value
|
||||
assert isinstance(caught, ConnectionTimeoutError)
|
||||
assert caught.message == "timeout connecting to localhost:9991"
|
||||
assert caught.host == "localhost"
|
||||
assert caught.port == 9991
|
||||
|
||||
|
||||
def test_reply_decode_error_carries_reason() -> None:
|
||||
"""Requirement 7.3: a corrupt/truncated capnp reply or cluster.bin
|
||||
reports "invalid cluster.bin: <reason>"."""
|
||||
with pytest.raises(XceAdminError) as exc_info:
|
||||
raise ReplyDecodeError(
|
||||
"invalid cluster.bin: truncated capnp message",
|
||||
reason="truncated capnp message",
|
||||
)
|
||||
|
||||
caught = exc_info.value
|
||||
assert isinstance(caught, ReplyDecodeError)
|
||||
assert caught.message == "invalid cluster.bin: truncated capnp message"
|
||||
assert caught.reason == "truncated capnp message"
|
||||
|
||||
|
||||
def test_too_many_redirects_error_carries_hop_count() -> None:
|
||||
"""Requirement 3.2 (declared here, raised by a later task): more than 3
|
||||
redirect hops is an error."""
|
||||
with pytest.raises(XceAdminError) as exc_info:
|
||||
raise TooManyRedirectsError(
|
||||
"followed 3 redirects without reaching a primary", hops=3
|
||||
)
|
||||
|
||||
caught = exc_info.value
|
||||
assert isinstance(caught, TooManyRedirectsError)
|
||||
assert caught.message == "followed 3 redirects without reaching a primary"
|
||||
assert caught.hops == 3
|
||||
|
||||
|
||||
def test_no_primary_error_carries_host_and_port() -> None:
|
||||
"""Requirement 3.4 (declared here, raised by a later task): no primary
|
||||
elected within the discovery timeout."""
|
||||
with pytest.raises(XceAdminError) as exc_info:
|
||||
raise NoPrimaryError(
|
||||
"no primary available at localhost:9991", host="localhost", port=9991
|
||||
)
|
||||
|
||||
caught = exc_info.value
|
||||
assert isinstance(caught, NoPrimaryError)
|
||||
assert caught.message == "no primary available at localhost:9991"
|
||||
assert caught.host == "localhost"
|
||||
assert caught.port == 9991
|
||||
|
||||
|
||||
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`
|
||||
clause recovers e.message no matter which of the 8 subclasses was
|
||||
actually raised — this is what lets the future CLI layer (task 4.1) use
|
||||
one dispatch surface instead of 8 separate except clauses."""
|
||||
cases: list[tuple[XceAdminError, str]] = [
|
||||
(SchemaNotFoundError("msg-a", path="/x"), "msg-a"),
|
||||
(ValidationError("msg-b", file="f", json_pointer="/p", constraint="c"), "msg-b"),
|
||||
(ReservedNameError("msg-c"), "msg-c"),
|
||||
(DuplicateEntryError("msg-d"), "msg-d"),
|
||||
(ConnectionTimeoutError("msg-e", host="h", port=1), "msg-e"),
|
||||
(ReplyDecodeError("msg-f", reason="r"), "msg-f"),
|
||||
(TooManyRedirectsError("msg-g", hops=3), "msg-g"),
|
||||
(NoPrimaryError("msg-h", host="h", port=1), "msg-h"),
|
||||
]
|
||||
|
||||
for exc, expected_message in cases:
|
||||
try:
|
||||
raise exc
|
||||
except XceAdminError as caught:
|
||||
assert caught.message == expected_message
|
||||
else:
|
||||
pytest.fail("expected XceAdminError to be raised and caught")
|
||||
Loading…
Add table
Add a link
Reference in a new issue