minos.aggregate.testing.snapshots.repositories.testcases

Classes

SnapshotRepositoryTestCase

class SnapshotRepositoryTestCase[source]

Bases: MinosTestCase, ABC

snapshot_repository: SnapshotRepository
class Owner[source]

Bases: RootEntity

For testing purposes

name: str
surname: str
age: Optional[int]
__init__(*args, uuid=NULL_UUID, version=0, created_at=NULL_DATETIME, updated_at=NULL_DATETIME, _event_repository, _snapshot_repository, **kwargs)

Class constructor.

Parameters
apply_diff(event)

Apply the differences over the instance.

Parameters

event (Event) – The FieldDiffContainer containing the values to be set.

Returns

This method does not return anything.

Return type

None

property avro_bytes: bytes

Generate bytes representation of the current instance.

Returns

A bytes object.

property avro_data: dict[str, Any]

Compute the avro data of the model.

Returns

A dictionary object.

property avro_str: str

Generate bytes representation of the current instance.

Returns

A bytes object.

classname = 'minos.aggregate.testing.snapshots.repositories.testcases.SnapshotRepositoryTestCase.Owner'
async classmethod create(*args, **kwargs)

Create a new RootEntity instance.

Parameters

  • args – Additional positional arguments.

  • kwargs – Additional named arguments.

Returns

A new RootEntity instance.

Return type

T

static decode_data(decoder, target, type_, **kwargs)

Decode data with the given decoder.

Parameters

  • decoder (DataDecoder) – The decoder instance.

  • target (Any) – The data to be decoded.

  • type – The data type.

  • kwargs – Additional named arguments.

  • type_ (ModelType) –

Returns

A decoded instance.

Return type

Any

static decode_schema(decoder, target, **kwargs)

Decode schema with the given encoder.

Parameters

  • decoder (SchemaDecoder) – The decoder instance.

  • target (Any) – The schema to be decoded.

  • kwargs – Additional named arguments.

Returns

The decoded schema as a type.

Return type

Any

async delete()

Delete the given root entity instance.

Returns

This method does not return anything.

Return type

None

diff(another)

Compute the difference with another instance.

Both RootEntity instances (self and another) must share the same uuid value.

Parameters

another (RootEntity) – Another RootEntity instance.

Returns

An FieldDiffContainer instance.

Return type

Event

static encode_data(encoder, target, **kwargs)

Encode data with the given encoder.

Parameters

  • encoder (DataEncoder) – The encoder instance.

  • target (Any) – An optional pre-encoded data.

  • kwargs – Additional named arguments.

Returns

The encoded data of the instance.

Return type

Any

static encode_schema(encoder, target, **kwargs)

Encode schema with the given encoder.

Parameters

  • encoder (SchemaEncoder) – The encoder instance.

  • target (Any) – An optional pre-encoded schema.

  • kwargs – Additional named arguments.

Returns

The encoded schema of the instance.

Return type

Any

property fields: dict[str, minos.common.model.fields.Field]

Fields getter

classmethod find(condition, ordering=None, limit=None, *, _snapshot_repository, **kwargs)

Find a collection of instances based on a given Condition.

Parameters

  • condition (_Condition) – The Condition that must be satisfied by all the instances.

  • ordering (Optional[_Ordering]) – Optional argument to return the instance with specific ordering strategy. The default behaviour is to retrieve them without any order pattern.

  • limit (Optional[int]) – Optional argument to return only a subset of instances. The default behaviour is to return all the instances that meet the given condition.

  • _snapshot_repository (SnapshotRepository) – Snapshot to be set to the instances.

Returns

An asynchronous iterator of RootEntity instances.

Return type

AsyncIterator[T]

classmethod from_avro(schema, data)

Build a new instance from the avro schema and data.

Parameters

  • schema (Any) – The avro schema of the model.

  • data (Any) – The avro data of the model.

Returns

A new DynamicModel instance.

Return type

T

classmethod from_avro_bytes(raw, batch_mode=False, **kwargs)

Build a single instance or a sequence of instances from bytes.

Parameters

  • raw (bytes) – A bytes representation of the model.

  • batch_mode (bool) – If True the data is processed as a list of models, otherwise the data is processed as a single model.

  • kwargs – Additional named arguments.

Returns

A single instance or a sequence of instances.

Return type

Union[T, list[~T]]

classmethod from_avro_str(raw, **kwargs)

Build a single instance or a sequence of instances from bytes

Parameters

raw (str) – A str representation of the model.

Returns

A single instance or a sequence of instances.

Return type

Union[T, list[~T]]

classmethod from_diff(event, *args, **kwargs)

Build a new instance from an Event.

Parameters

  • event (Event) – The difference that contains the data.

  • args – Additional positional arguments.

  • kwargs – Additional named arguments.

Returns

A new RootEntity instance.

Return type

T

classmethod from_model_type(model_type, *args, **kwargs)

Build a DeclarativeModel from a ModelType.

Parameters

  • model_type (ModelType) – ModelType object containing the model structure.

  • args – Positional arguments to be passed to the model constructor.

  • kwargs – Named arguments to be passed to the model constructor.

Returns

A new DeclarativeModel instance.

Return type

T

classmethod from_typed_dict(typed_dict, *args, **kwargs)

Build a Model from a TypeDict and data.

Parameters

  • typed_dict (TypedDict) – TypeDict object containing the DTO’s structure

  • args – Positional arguments to be passed to the model constructor.

  • kwargs – Named arguments to be passed to the model constructor.

Returns

A new DataTransferObject instance.

Return type

T

async classmethod get(uuid, *, _snapshot_repository, **kwargs)

Get one instance from the database based on its identifier.

Parameters

  • uuid (UUID) – The identifier of the instance.

  • _snapshot_repository (SnapshotRepository) – Snapshot to be set to the root entity.

Returns

A RootEntity instance.

Return type

T

classmethod get_all(ordering=None, limit=None, *, _snapshot_repository, **kwargs)

Get all instance from the database.

Parameters

  • ordering (Optional[_Ordering]) – Optional argument to return the instance with specific ordering strategy. The default behaviour is to retrieve them without any order pattern.

  • limit (Optional[int]) – Optional argument to return only a subset of instances. The default behaviour is to return all the instances that meet the given condition.

  • _snapshot_repository (SnapshotRepository) – Snapshot to be set to the root entity.

Returns

A RootEntity instance.

Return type

AsyncIterator[T]

items() → a set-like object providing a view on D's items
keys() → a set-like object providing a view on D's keys
async refresh()

Refresh the state of the given instance.

Returns

This method does not return anything.

Return type

None

async save()

Store the current instance on the repository.

If didn’t exist previously creates a new one, otherwise updates the existing one.

Return type

None

classmethod to_avro_bytes(models)

Create a bytes representation of the given object instances.

Parameters

models (list[~T]) – A sequence of minos models.

Returns

A bytes object.

Return type

bytes

classmethod to_avro_str(models)

Build the avro string representation of the given object instances.

Parameters

models (list[~T]) – A sequence of minos models.

Returns

A bytes object.

Return type

str

type_hints_parameters = ()
async update(**kwargs)

Update an existing RootEntity instance.

Parameters

  • kwargs – Additional named arguments.

  • self (T) –

Returns

An updated RootEntity instance.

Return type

T

values() → an object providing a view on D's values
version: int
created_at: datetime
updated_at: datetime
uuid: UUID
class Car[source]

Bases: RootEntity

For testing purposes

doors: int
color: str
owner: Optional[Ref[SnapshotRepositoryTestCase.Owner]]
__init__(*args, uuid=NULL_UUID, version=0, created_at=NULL_DATETIME, updated_at=NULL_DATETIME, _event_repository, _snapshot_repository, **kwargs)

Class constructor.

Parameters
apply_diff(event)

Apply the differences over the instance.

Parameters

event (Event) – The FieldDiffContainer containing the values to be set.

Returns

This method does not return anything.

Return type

None

property avro_bytes: bytes

Generate bytes representation of the current instance.

Returns

A bytes object.

property avro_data: dict[str, Any]

Compute the avro data of the model.

Returns

A dictionary object.

property avro_str: str

Generate bytes representation of the current instance.

Returns

A bytes object.

classname = 'minos.aggregate.testing.snapshots.repositories.testcases.SnapshotRepositoryTestCase.Car'
async classmethod create(*args, **kwargs)

Create a new RootEntity instance.

Parameters

  • args – Additional positional arguments.

  • kwargs – Additional named arguments.

Returns

A new RootEntity instance.

Return type

T

static decode_data(decoder, target, type_, **kwargs)

Decode data with the given decoder.

Parameters

  • decoder (DataDecoder) – The decoder instance.

  • target (Any) – The data to be decoded.

  • type – The data type.

  • kwargs – Additional named arguments.

  • type_ (ModelType) –

Returns

A decoded instance.

Return type

Any

static decode_schema(decoder, target, **kwargs)

Decode schema with the given encoder.

Parameters

  • decoder (SchemaDecoder) – The decoder instance.

  • target (Any) – The schema to be decoded.

  • kwargs – Additional named arguments.

Returns

The decoded schema as a type.

Return type

Any

async delete()

Delete the given root entity instance.

Returns

This method does not return anything.

Return type

None

diff(another)

Compute the difference with another instance.

Both RootEntity instances (self and another) must share the same uuid value.

Parameters

another (RootEntity) – Another RootEntity instance.

Returns

An FieldDiffContainer instance.

Return type

Event

static encode_data(encoder, target, **kwargs)

Encode data with the given encoder.

Parameters

  • encoder (DataEncoder) – The encoder instance.

  • target (Any) – An optional pre-encoded data.

  • kwargs – Additional named arguments.

Returns

The encoded data of the instance.

Return type

Any

static encode_schema(encoder, target, **kwargs)

Encode schema with the given encoder.

Parameters

  • encoder (SchemaEncoder) – The encoder instance.

  • target (Any) – An optional pre-encoded schema.

  • kwargs – Additional named arguments.

Returns

The encoded schema of the instance.

Return type

Any

property fields: dict[str, minos.common.model.fields.Field]

Fields getter

classmethod find(condition, ordering=None, limit=None, *, _snapshot_repository, **kwargs)

Find a collection of instances based on a given Condition.

Parameters

  • condition (_Condition) – The Condition that must be satisfied by all the instances.

  • ordering (Optional[_Ordering]) – Optional argument to return the instance with specific ordering strategy. The default behaviour is to retrieve them without any order pattern.

  • limit (Optional[int]) – Optional argument to return only a subset of instances. The default behaviour is to return all the instances that meet the given condition.

  • _snapshot_repository (SnapshotRepository) – Snapshot to be set to the instances.

Returns

An asynchronous iterator of RootEntity instances.

Return type

AsyncIterator[T]

classmethod from_avro(schema, data)

Build a new instance from the avro schema and data.

Parameters

  • schema (Any) – The avro schema of the model.

  • data (Any) – The avro data of the model.

Returns

A new DynamicModel instance.

Return type

T

classmethod from_avro_bytes(raw, batch_mode=False, **kwargs)

Build a single instance or a sequence of instances from bytes.

Parameters

  • raw (bytes) – A bytes representation of the model.

  • batch_mode (bool) – If True the data is processed as a list of models, otherwise the data is processed as a single model.

  • kwargs – Additional named arguments.

Returns

A single instance or a sequence of instances.

Return type

Union[T, list[~T]]

classmethod from_avro_str(raw, **kwargs)

Build a single instance or a sequence of instances from bytes

Parameters

raw (str) – A str representation of the model.

Returns

A single instance or a sequence of instances.

Return type

Union[T, list[~T]]

classmethod from_diff(event, *args, **kwargs)

Build a new instance from an Event.

Parameters

  • event (Event) – The difference that contains the data.

  • args – Additional positional arguments.

  • kwargs – Additional named arguments.

Returns

A new RootEntity instance.

Return type

T

classmethod from_model_type(model_type, *args, **kwargs)

Build a DeclarativeModel from a ModelType.

Parameters

  • model_type (ModelType) – ModelType object containing the model structure.

  • args – Positional arguments to be passed to the model constructor.

  • kwargs – Named arguments to be passed to the model constructor.

Returns

A new DeclarativeModel instance.

Return type

T

classmethod from_typed_dict(typed_dict, *args, **kwargs)

Build a Model from a TypeDict and data.

Parameters

  • typed_dict (TypedDict) – TypeDict object containing the DTO’s structure

  • args – Positional arguments to be passed to the model constructor.

  • kwargs – Named arguments to be passed to the model constructor.

Returns

A new DataTransferObject instance.

Return type

T

async classmethod get(uuid, *, _snapshot_repository, **kwargs)

Get one instance from the database based on its identifier.

Parameters

  • uuid (UUID) – The identifier of the instance.

  • _snapshot_repository (SnapshotRepository) – Snapshot to be set to the root entity.

Returns

A RootEntity instance.

Return type

T

classmethod get_all(ordering=None, limit=None, *, _snapshot_repository, **kwargs)

Get all instance from the database.

Parameters

  • ordering (Optional[_Ordering]) – Optional argument to return the instance with specific ordering strategy. The default behaviour is to retrieve them without any order pattern.

  • limit (Optional[int]) – Optional argument to return only a subset of instances. The default behaviour is to return all the instances that meet the given condition.

  • _snapshot_repository (SnapshotRepository) – Snapshot to be set to the root entity.

Returns

A RootEntity instance.

Return type

AsyncIterator[T]

items() → a set-like object providing a view on D's items
keys() → a set-like object providing a view on D's keys
async refresh()

Refresh the state of the given instance.

Returns

This method does not return anything.

Return type

None

async save()

Store the current instance on the repository.

If didn’t exist previously creates a new one, otherwise updates the existing one.

Return type

None

classmethod to_avro_bytes(models)

Create a bytes representation of the given object instances.

Parameters

models (list[~T]) – A sequence of minos models.

Returns

A bytes object.

Return type

bytes

classmethod to_avro_str(models)

Build the avro string representation of the given object instances.

Parameters

models (list[~T]) – A sequence of minos models.

Returns

A bytes object.

Return type

str

type_hints_parameters = ()
async update(**kwargs)

Update an existing RootEntity instance.

Parameters

  • kwargs – Additional named arguments.

  • self (T) –

Returns

An updated RootEntity instance.

Return type

T

values() → an object providing a view on D's values
version: int
created_at: datetime
updated_at: datetime
uuid: UUID
setUp()[source]
Return type

None

abstract build_snapshot_repository()[source]
Return type

SnapshotRepository

async populate()[source]
Return type

None

async populate_and_synchronize()[source]
async synchronize()[source]
async asyncSetUp()[source]
async asyncTearDown()[source]
assert_equal_snapshot_entries(expected, observed)[source]
Parameters
test_type()[source]
async test_dispatch()[source]
async test_dispatch_first_transaction()[source]
async test_dispatch_second_transaction()[source]
async test_dispatch_third_transaction()[source]
async test_dispatch_ignore_previous_version()[source]
async test_dispatch_with_offset()[source]
async test_find_by_uuid()[source]
async test_find_with_transaction()[source]
async test_find_with_transaction_delete()[source]
async test_find_with_transaction_reverted()[source]
async test_find_streaming_true()[source]
async test_find_with_duplicates()[source]
async test_find_empty()[source]
async test_get()[source]
async test_get_with_transaction()[source]
async test_get_raises()[source]
async test_get_with_transaction_raises()[source]
async test_find()[source]
async test_find_all()[source]
__init__(*args, **kwargs)
Parameters

  • args (Any) –

  • kwargs (Any) –

Return type

None

static __new__(cls, *args, **kwargs)
Parameters

  • args (Any) –

  • kwargs (Any) –

Return type

Any

get_config()
Return type

Config

get_config_file_path()
Return type

Path

get_injections()
Return type

list[Union[minos.common.injections.mixins.InjectableMixin, type[minos.common.injections.mixins.InjectableMixin], str]]

tearDown()
Return type

None

CONFIG_FILE_PATH: Path