angzarr_client.testing¶
Testing utilities for angzarr applications.
This module provides helpers for writing tests against angzarr aggregates, sagas, and other components. It includes:
UUID generation: Deterministic UUID creation for reproducible tests
Proto builders: Simplified constructors for EventBook, CommandBook, etc.
Scenario context: BDD-style context for tracking test state
- Example usage:
- from angzarr_client.testing import (
ScenarioContext, uuid_for, make_cover, make_event_book, pack_event,
)
- def test_player_registration():
ctx = ScenarioContext() ctx.domain = “player” ctx.root = uuid_for(“player-alice”)
# Build event book with prior events book = ctx.event_book()
# Execute command and verify result = handler.handle(cmd, book) assert result.player_id == “player_alice@test.com”
Submodules¶
Attributes¶
Classes¶
Shared context for BDD test scenarios. |
Functions¶
|
Create a CommandBook with a single command. |
|
Create a CommandPage. |
|
Create a Cover from domain and root bytes. |
|
Create an EventBook. |
|
Create an EventPage. |
|
Create a timestamp for now. |
|
Pack a protobuf message into an Any with the canonical type URL. |
|
Generate a deterministic 16-byte UUID from a name. |
|
Generate a deterministic UUID object from a name. |
|
Generate a deterministic UUID string from a name. |
Package Contents¶
- angzarr_client.testing.make_command_book(cover: angzarr_client.proto.angzarr.Cover, command: google.protobuf.any_pb2.Any, sequence: int = 0) angzarr_client.proto.angzarr.CommandBook¶
Create a CommandBook with a single command.
- Parameters:
cover – The Cover identifying the target aggregate
command – The packed command (ProtoAny)
sequence – The command sequence number (defaults to 0)
- Returns:
CommandBook proto with one page
- angzarr_client.testing.make_command_page(sequence: int, command: google.protobuf.any_pb2.Any) angzarr_client.proto.angzarr.CommandPage¶
Create a CommandPage.
- Parameters:
sequence – The command sequence number
command – The packed command (ProtoAny)
- Returns:
CommandPage proto
- angzarr_client.testing.make_cover(domain: str, root: bytes, correlation_id: str = '') angzarr_client.proto.angzarr.Cover¶
Create a Cover from domain and root bytes.
- Parameters:
domain – The aggregate domain name
root – The aggregate root as 16 bytes
correlation_id – Optional correlation ID for cross-domain tracking
- Returns:
Cover proto with domain and root set
- angzarr_client.testing.make_event_book(cover: angzarr_client.proto.angzarr.Cover, pages: list[angzarr_client.proto.angzarr.EventPage] = None, next_sequence: int = None) angzarr_client.proto.angzarr.EventBook¶
Create an EventBook.
- Parameters:
cover – The Cover identifying the aggregate
pages – List of EventPages (defaults to empty)
next_sequence – Next sequence number (defaults to len(pages))
- Returns:
EventBook proto
- angzarr_client.testing.make_event_page(sequence: int, event: google.protobuf.any_pb2.Any) angzarr_client.proto.angzarr.EventPage¶
Create an EventPage.
- Parameters:
sequence – The event sequence number
event – The packed event (ProtoAny)
- Returns:
EventPage proto
- angzarr_client.testing.make_timestamp() google.protobuf.timestamp_pb2.Timestamp¶
Create a timestamp for now.
Alias for angzarr_client.helpers.now() for backwards compatibility.
- Returns:
Current time as protobuf Timestamp
- angzarr_client.testing.pack_event(msg: google.protobuf.message.Message) google.protobuf.any_pb2.Any¶
Pack a protobuf message into an Any with the canonical type URL.
The type URL is derived from the message’s descriptor (
msg.DESCRIPTOR.full_name) prefixed with the standardtype.googleapis.com/— per the google.protobuf.Any spec.Audit finding #47 (Option C — drop the second arg, derive name from the message): mirrors Rust’s
testing::builders::pack_event<M>(msg). Removes the previoustype_url_prefixparameter (which was a typo-prone footgun and diverged in semantics from the Rust 2nd-arg convention).- Returns:
ProtoAny containing the packed message.
- class angzarr_client.testing.ScenarioContext¶
Shared context for BDD test scenarios.
Tracks the current aggregate, event history, command results, and rebuilt state across Given/When/Then steps.
- domain¶
Current aggregate domain being tested
- root¶
Current aggregate root as bytes
- events¶
List of packed events (ProtoAny) in history
- result¶
Last command handler result (event or tuple of events)
- error¶
Last CommandRejectedError if command was rejected
- state¶
Rebuilt aggregate state after applying events
Example
ctx = ScenarioContext() ctx.domain = “player” ctx.root = uuid_for(“player-alice”)
# Given player registered ctx.add_event(PlayerRegistered(email=”alice@test.com”))
# When deposit funds try:
ctx.result = handler.handle(deposit_cmd, ctx.event_book())
- except CommandRejectedError as e:
ctx.error = e
# Then balance updated assert ctx.result.new_balance == 100
- result: Any = None¶
- error: angzarr_client.errors.CommandRejectedError | None = None¶
- state: Any = None¶
- event_book() angzarr_client.proto.angzarr.EventBook¶
Build EventBook from accumulated events.
Creates an EventBook with proper sequencing from the events added via add_event().
- Returns:
EventBook with cover, pages, and next_sequence set
- add_event(event_msg)¶
Add an event to history.
Packs the event message (deriving the type URL from
event_msg.DESCRIPTOR.full_name) and appends to the event list.Audit finding #47: the previous
type_url_prefixarg is dropped — the canonical URL is derived from the message descriptor.- Parameters:
event_msg – The protobuf event message to add
- clear_events()¶
Clear all events from history.
- clear_result()¶
Clear the last result and error.
- reset()¶
Reset context to initial state.
- angzarr_client.testing.DEFAULT_TEST_NAMESPACE¶
- angzarr_client.testing.uuid_for(name: str, namespace: uuid.UUID = DEFAULT_TEST_NAMESPACE) bytes¶
Generate a deterministic 16-byte UUID from a name.
The same name always generates the same UUID within a namespace. Returns bytes suitable for use as aggregate root IDs.
- Parameters:
name – A string identifier (e.g., “player-alice”, “table-1”)
namespace – UUID namespace for generation (defaults to test namespace)
- Returns:
16-byte UUID as bytes
Example
root = uuid_for(“player-alice”) assert len(root) == 16 assert uuid_for(“player-alice”) == root # deterministic
- angzarr_client.testing.uuid_obj_for(name: str, namespace: uuid.UUID = DEFAULT_TEST_NAMESPACE) uuid.UUID¶
Generate a deterministic UUID object from a name.
- Parameters:
name – A string identifier
namespace – UUID namespace for generation
- Returns:
UUID object
Example
id_obj = uuid_obj_for(“player-alice”) assert id_obj.bytes == uuid_for(“player-alice”)
- angzarr_client.testing.uuid_str_for(name: str, namespace: uuid.UUID = DEFAULT_TEST_NAMESPACE) str¶
Generate a deterministic UUID string from a name.
- Parameters:
name – A string identifier
namespace – UUID namespace for generation
- Returns:
UUID as standard string format (8-4-4-4-12)
Example
id_str = uuid_str_for(“player-alice”) assert “-” in id_str # standard UUID format