pytest_moto_fixtures.services package

Submodules

pytest_moto_fixtures.services.eventbridge module

Access Event Bridge service.

class pytest_moto_fixtures.services.eventbridge.EventBridgeBus(*, client: EventBridgeClient, name: str, arn: str, queue: SQSQueue)

Bases: object

Bus in Event Bridge service.

An SQS queue is used to receive messages sent to the bus.

arn: str

Bus ARN.

client: EventBridgeClient

Event Bridge Client.

name: str

Bus name.

purge_bus_events() → None

Purge events in queue of topic.

put_event(*, source: str, detail_type: str, detail: str | dict[Any, Any], resources: list[str] | NoArgs = NoArgs.NO_ARG, time: datetime | NoArgs = NoArgs.NO_ARG) → None

Put event to bus.

Parameters:

  • source – Source of event.

  • detail_type – Event detail type.

  • detail – Event details. Receives a string in JSON format or a dict.

  • resources – List of resources associated with the event.

  • time – Date and time of the event. If not provided, the current time will be used.

queue: SQSQueue

Queu to bus messages.

receive_event() → EventTypeDef | None

Receive event from the queue of bus and removes them.

Returns:

Event received, or None if the queue has no events.

pytest_moto_fixtures.services.eventbridge.eventbridge_create_bus(*, eventbridge_client: EventBridgeClient, sqs_client: SQSClient, name: str | None = None, tags: Sequence[TagTypeDef] | NoArgs = NoArgs.NO_ARG) → Iterator[EventBridgeBus]

Context for creating an Event Bridge bus with SQS queue targeted and removing it on exit.

Parameters:

  • eventbridge_client – Event Bridge client where the bus will be created.

  • sqs_client – SQS client where the queue will be created.

  • name – Name of bus and queue to be created. If it is None a random name will be used.

  • tags – Tags of bus to be created.

Returns:

Bus created in Event Bridge service.

pytest_moto_fixtures.services.s3 module

Access S3 service.

class pytest_moto_fixtures.services.s3.S3Bucket(*, client: S3Client, name: str)

Bases: object

Bucket in S3 service.

client: S3Client

S3 Client.

name: str

Bucket name.

prune() → None

Prune objects in bucket.

pytest_moto_fixtures.services.s3.s3_create_bucket(*, s3_client: S3Client, name: str | NoArgs = NoArgs.NO_ARG) → Iterator[S3Bucket]

Context for creating an S3 bucket and removing it on exit.

Parameters:

  • s3_client – S3 client where bucket will be created.

  • name – Name of bucket to be created. If it is None a random name will be used.

Returns:

Bucket created in S3 service.

pytest_moto_fixtures.services.sns module

Access SNS service.

class pytest_moto_fixtures.services.sns.SNSTopic(*, client: SNSClient, name: str, arn: str, queue: SQSQueue)

Bases: object

Topic in SNS service.

An SQS queue is used to receive messages sent to the topic.

arn: str

Topic ARN.

client: SNSClient

SNS Client.

name: str

Topic name.

publish_message(*, message: str, attributes: Mapping[str, MessageAttributeValueTypeDef] | NoArgs = NoArgs.NO_ARG, deduplication_id: str | NoArgs = NoArgs.NO_ARG, group_id: str | NoArgs = NoArgs.NO_ARG) → None

Send message to topic.

Parameters:

  • message – Message body.

  • attributes – Attributes of message.

  • deduplication_id – Identifier to check for duplicate messages.

  • group_id – Identifier to group messages that should be delivered sequentially.

purge_topic_messages() → None

Purge messages in queue of topic.

queue: SQSQueue

Queue to topic messages.

receive_message() → MessageTypeDef | None

Receive message from the queue of topic and removes them.

Returns:

Message received, or None if the queue has no messages.

pytest_moto_fixtures.services.sns.sns_create_fifo_topic(*, sns_client: SNSClient, sqs_client: SQSClient, name: str | None = None, attributes: Mapping[str, str] | NoArgs = NoArgs.NO_ARG, tags: Sequence[TagTypeDef] | NoArgs = NoArgs.NO_ARG) → Iterator[SNSTopic]

Context for creating an SNS fifo topic with SQS fifo queue subscribed and removing it on exit.

Parameters:

  • sns_client – SNS client where the topic will be created.

  • sqs_client – SQS client where the queue will be created.

  • name – Name of topic and queue to be created. If it is None a random name will be used, and if it does not end with '.fifo' it will be appended.

  • attributes – Attributes of topic to be created. If it does not have the 'FifoTopic' attribute it will be added.

  • tags – Tags of topic to be created.

Returns:

Topic created in SNS service.

pytest_moto_fixtures.services.sns.sns_create_topic(*, sns_client: SNSClient, sqs_client: SQSClient, name: str | None = None, attributes: Mapping[str, str] | NoArgs = NoArgs.NO_ARG, tags: Sequence[TagTypeDef] | NoArgs = NoArgs.NO_ARG) → Iterator[SNSTopic]

Context for creating an SNS topic with SQS queue subscribed and removing it on exit.

Parameters:

  • sns_client – SNS client where the topic will be created.

  • sqs_client – SQS client where the queue will be created.

  • name – Name of topic and queue to be created. If it is None a random name will be used.

  • attributes – Attributes of topic to be created.

  • tags – Tags of topic to be created.

Returns:

Topic created in SNS service.

pytest_moto_fixtures.services.sqs module

Access SQS service.

class pytest_moto_fixtures.services.sqs.SQSQueue(*, client: SQSClient, name: str, arn: str, url: str)

Bases: object

Queue in SQS service.

arn: str

Queue ARN.

client: SQSClient

SQS Client.

name: str

Queue name.

purge_queue() → None

Purge messages in queue.

receive_message() → MessageTypeDef | None

Receives messages from the queue and removes them.

Returns:

Messages received from the queue, or None if the queue has no messages.

send_message(*, body: str, delay_seconds: int | NoArgs = NoArgs.NO_ARG, deduplication_id: str | NoArgs = NoArgs.NO_ARG, group_id: str | NoArgs = NoArgs.NO_ARG) → None

Send message to queue.

Parameters:

  • body – Message body.

  • delay_seconds – Message delivery delay in seconds.

  • deduplication_id – Identifier to check for duplicate messages.

  • group_id – Identifier to group messages that should be delivered sequentially.

url: str

Queue URL.

pytest_moto_fixtures.services.sqs.sqs_create_fifo_queue(*, sqs_client: SQSClient, name: str | None = None, attributes: Mapping[QueueAttributeNameType, str] | NoArgs = NoArgs.NO_ARG, tags: Mapping[str, str] | NoArgs = NoArgs.NO_ARG) → Iterator[SQSQueue]

Context for creating an SQS fifo queue and removing it on exit.

Parameters:

  • sqs_client – SQS client where the queue will be created.

  • name – Name of queue to be created. If it is None a random name will be used, and if it does not end with '.fifo' it will be appended.

  • attributes – Attributes of queue to be created. If it does not have the 'FifoQueue' attribute it will be added.

  • tags – Tags of queue to be created.

Returns:

Queue created in SQS service.

pytest_moto_fixtures.services.sqs.sqs_create_queue(*, sqs_client: SQSClient, name: str | None = None, attributes: Mapping[QueueAttributeNameType, str] | NoArgs = NoArgs.NO_ARG, tags: Mapping[str, str] | NoArgs = NoArgs.NO_ARG) → Iterator[SQSQueue]

Context for creating an SQS queue and removing it on exit.

Parameters:

  • sqs_client – SQS client where the queue will be created.

  • name – Name of queue to be created. If it is None a random name will be used.

  • attributes – Attributes of queue to be created.

  • tags – Tags of queue to be created.

Returns:

Queue created in SQS service.

Module contents

Access to AWS services.