diff --git a/.doc_gen/metadata/scheduler_metadata.yaml b/.doc_gen/metadata/scheduler_metadata.yaml index ffd4a2b459a..3d53c627d2a 100644 --- a/.doc_gen/metadata/scheduler_metadata.yaml +++ b/.doc_gen/metadata/scheduler_metadata.yaml @@ -21,6 +21,15 @@ scheduler_hello: genai: some snippet_tags: - Scheduler.dotnetv3.HelloScheduler + Python: + versions: + - sdk_version: 3 + github: python/example_code/scheduler + sdkguide: + excerpts: + - description: + snippet_tags: + - python.example_code.scheduler.Hello services: scheduler: {ListSchedules} scheduler_CreateSchedule: @@ -42,6 +51,16 @@ scheduler_CreateSchedule: genai: most snippet_tags: - Scheduler.dotnetv3.CreateSchedule + Python: + versions: + - sdk_version: 3 + github: python/example_code/scheduler + sdkguide: + excerpts: + - description: + snippet_tags: + - python.example_code.scheduler.EventSchedulerWrapper.decl + - python.example_code.scheduler.CreateSchedule services: scheduler: {CreateSchedule} scheduler_CreateScheduleGroup: @@ -63,6 +82,16 @@ scheduler_CreateScheduleGroup: genai: most snippet_tags: - Scheduler.dotnetv3.CreateScheduleGroup + Python: + versions: + - sdk_version: 3 + github: python/example_code/scheduler + sdkguide: + excerpts: + - description: + snippet_tags: + - python.example_code.scheduler.EventSchedulerWrapper.decl + - python.example_code.scheduler.CreateScheduleGroup services: scheduler: {CreateScheduleGroup} scheduler_DeleteSchedule: @@ -76,6 +105,16 @@ scheduler_DeleteSchedule: genai: most snippet_tags: - Scheduler.dotnetv3.DeleteSchedule + Python: + versions: + - sdk_version: 3 + github: python/example_code/scheduler + sdkguide: + excerpts: + - description: + snippet_tags: + - python.example_code.scheduler.EventSchedulerWrapper.decl + - python.example_code.scheduler.DeleteSchedule Java: versions: - sdk_version: 2 @@ -105,6 +144,16 @@ scheduler_DeleteScheduleGroup: genai: most snippet_tags: - Scheduler.dotnetv3.DeleteScheduleGroup + Python: + versions: + - sdk_version: 3 + github: python/example_code/scheduler + sdkguide: + excerpts: + - description: + snippet_tags: + - python.example_code.scheduler.EventSchedulerWrapper.decl + - python.example_code.scheduler.DeleteScheduleGroup services: scheduler: {DeleteScheduleGroup} scheduler_ScheduledEventsWorkflow: @@ -145,5 +194,19 @@ scheduler_ScheduledEventsWorkflow: genai: most snippet_tags: - Scheduler.dotnetv3.SchedulerWrapper + Python: + versions: + - sdk_version: 3 + github: python/example_code/scheduler + sdkguide: + excerpts: + - description: Run an interactive scenario at a command prompt. + genai: some + snippet_tags: + - python.example_code.scheduler.FeatureScenario + - description: SchedulerWrapper class that wraps &EVlong; Scheduler actions. + genai: some + snippet_tags: + - python.example_code.scheduler.EventSchedulerWrapper.class services: scheduler: {CreateSchedule, CreateScheduleGroup, DeleteSchedule, DeleteScheduleGroups} diff --git a/python/example_code/scheduler/README.md b/python/example_code/scheduler/README.md new file mode 100644 index 00000000000..b449fffc42f --- /dev/null +++ b/python/example_code/scheduler/README.md @@ -0,0 +1,129 @@ +# EventBridge Scheduler code examples for the SDK for Python + +## Overview + +Shows how to use the AWS SDK for Python (Boto3) to work with Amazon EventBridge Scheduler. + + + + +_EventBridge Scheduler allows you to create, run, and manage tasks on a schedule from one central, managed service._ + +## ⚠ Important + +* Running this code might result in charges to your AWS account. For more details, see [AWS Pricing](https://aws.amazon.com/pricing/) and [Free Tier](https://aws.amazon.com/free/). +* Running the tests might result in charges to your AWS account. +* We recommend that you grant your code least privilege. At most, grant only the minimum permissions required to perform the task. For more information, see [Grant least privilege](https://docs.aws.amazon.com/IAM/latest/UserGuide/best-practices.html#grant-least-privilege). +* This code is not tested in every AWS Region. For more information, see [AWS Regional Services](https://aws.amazon.com/about-aws/global-infrastructure/regional-product-services). + + + + +## Code examples + +### Prerequisites + +For prerequisites, see the [README](../../README.md#Prerequisites) in the `python` folder. + +Install the packages required by these examples by running the following in a virtual environment: + +``` +python -m pip install -r requirements.txt +``` + + + + +### Get started + +- [Hello EventBridge Scheduler](hello/hello_scheduler.py#L4) (`ListSchedules`) + + +### Single actions + +Code excerpts that show you how to call individual service functions. + +- [CreateSchedule](scheduler_wrapper.py#L38) +- [CreateScheduleGroup](scheduler_wrapper.py#L131) +- [DeleteSchedule](scheduler_wrapper.py#L104) +- [DeleteScheduleGroup](scheduler_wrapper.py#L160) + +### Scenarios + +Code examples that show you how to accomplish a specific task by calling multiple +functions within the same service. + +- [Scheduled Events workflow](scenario/scheduler_scenario.py) + + + + + +## Run the examples + +### Instructions + + + + + +#### Hello EventBridge Scheduler + +This example shows you how to get started using EventBridge Scheduler. + +``` +python hello/hello_scheduler.py +``` + + +#### Scheduled Events workflow + +This example shows you how to do the following: + +- Deploy a CloudFormation stack with required resources. +- Create a EventBridge Scheduler schedule group. +- Create a one-time EventBridge Scheduler schedule with a flexible time window. +- Create a recurring EventBridge Scheduler schedule with a specified rate. +- Delete EventBridge Scheduler the schedule and schedule group. +- Clean up resources and delete the stack. + + + + +Start the example by running the following at a command prompt: + +``` +python scenario/scheduler_scenario.py +``` + + + + + +### Tests + +⚠ Running tests might result in charges to your AWS account. + + +To find instructions for running these tests, see the [README](../../README.md#Tests) +in the `python` folder. + + + + + + +## Additional resources + +- [EventBridge Scheduler User Guide](https://docs.aws.amazon.com/scheduler/latest/userguide/intro.html) +- [EventBridge Scheduler API Reference](https://docs.aws.amazon.com/scheduler/latest/apireference/Welcome.html) +- [SDK for Python EventBridge Scheduler reference](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/scheduler.html) + + + + +--- + +Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. + +SPDX-License-Identifier: Apache-2.0 \ No newline at end of file diff --git a/python/example_code/scheduler/hello/hello_scheduler.py b/python/example_code/scheduler/hello/hello_scheduler.py new file mode 100644 index 00000000000..fe6f70e2bf4 --- /dev/null +++ b/python/example_code/scheduler/hello/hello_scheduler.py @@ -0,0 +1,34 @@ +# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +# SPDX-License-Identifier: Apache-2.0 + +# snippet-start:[python.example_code.scheduler.Hello] +import boto3 + + +def hello_scheduler(scheduler_client): + """ + Use the AWS SDK for Python (Boto3) to create an Amazon EventBridge Scheduler + client and list the schedules in your account. + This example uses the default settings specified in your shared credentials + and config files. + + :param scheduler_client: A Boto3 Amazon EventBridge Scheduler Client object. This object wraps + the low-level Amazon EventBridge Scheduler service API. + """ + print("Hello, Amazon EventBridge Scheduler! Let's list some of your schedules:\n") + paginator = scheduler_client.get_paginator("list_schedules") + page_iterator = paginator.paginate(PaginationConfig={"MaxItems": 10}) + + schedule_names: [str] = [] + for page in page_iterator: + for schedule in page["Schedules"]: + schedule_names.append(schedule["Name"]) + + print(f"{len(schedule_names)} schedule(s) retrieved.") + for schedule_name in schedule_names: + print(f"\t{schedule_name}") + + +if __name__ == "__main__": + hello_scheduler(boto3.client("scheduler")) +# snippet-end:[python.example_code.scheduler.Hello] diff --git a/python/example_code/scheduler/hello/requirements.txt b/python/example_code/scheduler/hello/requirements.txt new file mode 100644 index 00000000000..e32a2fb0fa9 --- /dev/null +++ b/python/example_code/scheduler/hello/requirements.txt @@ -0,0 +1 @@ +boto3>=1.35.38 \ No newline at end of file diff --git a/python/example_code/scheduler/requirements.txt b/python/example_code/scheduler/requirements.txt new file mode 100644 index 00000000000..b2f85e5bc24 --- /dev/null +++ b/python/example_code/scheduler/requirements.txt @@ -0,0 +1,3 @@ +boto3>=1.35.38 +pytest>=8.3.3 +botocore>=1.35.38 \ No newline at end of file diff --git a/python/example_code/scheduler/scenario/README.md b/python/example_code/scheduler/scenario/README.md new file mode 100644 index 00000000000..9a78906596e --- /dev/null +++ b/python/example_code/scheduler/scenario/README.md @@ -0,0 +1,79 @@ +# Amazon EventBridge Scheduler Workflow + +## Overview +This example shows how to use AWS SDK for Python (Boto3) to work with Amazon EventBridge Scheduler with schedules and schedule groups. The workflow demonstrates how to create and delete one-time and recurring schedules within a schedule group to generate events on a specified target, such as an Amazon Simple Notification Service (Amazon SNS) Topic. + +The target SNS topic and the AWS Identity and Access Management (IAM) role used with the schedules are created as part of an AWS CloudFormation stack that is deployed at the start of the workflow, and deleted when the workflow is complete. + +![Scheduler scenario diagram](resources/scheduler-workflow.png) + +This workflow demonstrates the following steps and tasks: + +1. **Prepare the Application** + + - Prompts the user for an email address to use for the subscription for the SNS topic. + - Prompts the user for a name for the Cloud Formation stack. + - The user must confirm the email subscription to receive event emails. + - Deploys the Cloud Formation template in resources/cfn_template.yaml for resource creation. + - Stores the outputs of the stack into variables for use in the workflow. + - Creates a schedule group for all workflow schedules. + +2. **Create a one-time Schedule** + + - Creates a one-time schedule to send an initial event. + - Prompts the user for a name for the one-time schedule. + - The user must confirm the email subscription to receive an event email. + - The content of the email should include the name of the newly created schedule. + - Use a Flexible Time Window of 10 minutes and set the schedule to delete after completion. + +3. **Create a time-based schedule** + + - Prompts the user for a rate per minutes (example: every 2 minutes) for a scheduled recurring event. + - Creates the scheduled event for X times per hour for 1 hour. + - Deletes the schedule when the user is finished. + - Prompts the user to confirm when they are ready to delete the schedule. + +4. **Clean up** + + - Prompts the user to confirm they want to destroy the stack and clean up all resources. + - Deletes the schedule group. + - Destroys the Cloud Formation stack and wait until the stack has been removed. + +## Prerequisites + +Before running this workflow, ensure you have: + +- An AWS account with proper permissions to use Amazon EventBridge Scheduler and Amazon EventBridge. + +## AWS Services Used + +This workflow uses the following AWS services: + +- Amazon EventBridge Scheduler +- Amazon EventBridge +- Amazon Simple Notification Service (SNS) +- AWS CloudFormation + +### Resources + +The workflow scenario deploys the AWS CloudFormation stack with the required resources. + +## Amazon EventBridge Scheduler Actions + +The workflow covers the following EventBridge Scheduler API actions: + +- [`CreateSchedule`](https://docs.aws.amazon.com/scheduler/latest/APIReference/API_CreateSchedule.html) +- [`CreateScheduleGroup`](https://docs.aws.amazon.com/scheduler/latest/APIReference/API_CreateScheduleGroup.html) +- [`DeleteSchedule`](https://docs.aws.amazon.com/scheduler/latest/APIReference/API_DeleteSchedule.html) +- [`DeleteScheduleGroup`](https://docs.aws.amazon.com/scheduler/latest/APIReference/API_DeleteScheduleGroup.html) + + +## Additional resources + +* [EventBridge Scheduler User Guide](https://docs.aws.amazon.com/scheduler/latest/UserGuide/what-is-scheduler.html) +* [EventBridge Scheduler API Reference](https://docs.aws.amazon.com/scheduler/latest/APIReference/Welcome.html) + +--- + +Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +SPDX-License-Identifier: Apache-2.0 diff --git a/python/example_code/scheduler/scenario/cfn_template.yaml b/python/example_code/scheduler/scenario/cfn_template.yaml new file mode 100644 index 00000000000..7cd86232891 --- /dev/null +++ b/python/example_code/scheduler/scenario/cfn_template.yaml @@ -0,0 +1,50 @@ +Parameters: + email: + Type: String + Default: 'scheduler_test@example.com' + +Resources: + SchedulerSnsTopic: + Type: AWS::SNS::Topic + Properties: + KmsMasterKeyId: alias/aws/sns + + MySubscription: + Type: AWS::SNS::Subscription + Properties: + Endpoint: !Ref email + Protocol: email + TopicArn: !Ref SchedulerSnsTopic + + SchedulerRole: + Type: AWS::IAM::Role + Properties: + RoleName: example_scheduler_role + Path: / + AssumeRolePolicyDocument: + Version: '2012-10-17' + Statement: + - Effect: 'Allow' + Principal: + Service: 'scheduler.amazonaws.com' + Action: + - 'sts:AssumeRole' + Policies: + - PolicyName: 'Scheduler_SNS_policy' + PolicyDocument: + Version: '2012-10-17' + Statement: + - Effect: 'Allow' + Action: + - 'sns:Publish' + Resource: !Ref SchedulerSnsTopic + +Outputs: + + SNStopicARN: + Description: SNS topic ARN + Value: !Ref SchedulerSnsTopic + + RoleARN: + Description: Scheduler role ARN + Value: !GetAtt SchedulerRole.Arn diff --git a/python/example_code/scheduler/scenario/requirements.txt b/python/example_code/scheduler/scenario/requirements.txt new file mode 100644 index 00000000000..b2f85e5bc24 --- /dev/null +++ b/python/example_code/scheduler/scenario/requirements.txt @@ -0,0 +1,3 @@ +boto3>=1.35.38 +pytest>=8.3.3 +botocore>=1.35.38 \ No newline at end of file diff --git a/python/example_code/scheduler/scenario/resources/scheduler-workflow.png b/python/example_code/scheduler/scenario/resources/scheduler-workflow.png new file mode 100644 index 00000000000..53fa99b5cd6 Binary files /dev/null and b/python/example_code/scheduler/scenario/resources/scheduler-workflow.png differ diff --git a/python/example_code/scheduler/scenario/scheduler_scenario.py b/python/example_code/scheduler/scenario/scheduler_scenario.py new file mode 100644 index 00000000000..30831a6ab0d --- /dev/null +++ b/python/example_code/scheduler/scenario/scheduler_scenario.py @@ -0,0 +1,288 @@ +# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +# SPDX-License-Identifier: Apache-2.0 + +""" +Purpose + +Shows how to use the AWS SDK for Python (Boto3) with the Amazon EventBridge Scheduler to schedule +and receive events. +""" + +import logging +import sys +from datetime import datetime, timedelta, timezone +import os +from boto3.resources.base import ServiceResource +from boto3 import resource + +script_dir = os.path.dirname(os.path.abspath(__file__)) + +# Add relative path to include SchedulerWrapper. +sys.path.append(os.path.dirname(script_dir)) +from scheduler_wrapper import SchedulerWrapper + +# Add relative path to include demo_tools in this code example without need for setup. +sys.path.append(os.path.join(script_dir, "../../..")) +import demo_tools.question as q + + +DASHES = "-" * 80 + +sys.path + +logger = logging.getLogger(__name__) + + +# snippet-start:[python.example_code.scheduler.FeatureScenario] +class SchedulerScenario: + """ + A scenario that demonstrates how to use Boto3 to schedule and receive events using + the Amazon EventBridge Scheduler. + """ + + def __init__( + self, + scheduler_wrapper: SchedulerWrapper, + cloud_formation_resource: ServiceResource, + ): + self.eventbridge_scheduler = scheduler_wrapper + self.cloud_formation_resource = cloud_formation_resource + self.stack: ServiceResource = None + self.schedule_group_name = None + self.sns_topic_arn = None + self.role_arn = None + + def run(self) -> None: + """ + Runs the scenario. + """ + + print(DASHES) + print("Welcome to the Amazon EventBridge Scheduler Workflow.") + print(DASHES) + + print(DASHES) + self.prepare_application() + print(DASHES) + + print(DASHES) + self.create_one_time_schedule() + print(DASHES) + + print(DASHES) + self.create_recurring_schedule() + print(DASHES) + + print(DASHES) + if q.ask( + "Do you want to delete all resources created by this workflow? (y/n) ", + q.is_yesno, + ): + self.cleanup() + print(DASHES) + + print("Amazon EventBridge Scheduler workflow completed.") + + def prepare_application(self) -> None: + """ + Prepares the application by prompting the user setup information, deploying a CloudFormation stack and + creating a schedule group. + """ + print("Preparing the application...") + print( + "\nThis example creates resources in a CloudFormation stack, including an SNS topic" + + "\nthat will be subscribed to the EventBridge Scheduler events. " + + "\n\nYou will need to confirm the subscription in order to receive event emails. " + ) + + email_address = q.ask("Enter an email address to use for event subscriptions: ") + stack_name = q.ask("Enter a name for the AWS Cloud Formation Stack: ") + + template_file = SchedulerScenario.get_template_as_string() + + parameters = [{"ParameterKey": "email", "ParameterValue": email_address}] + + self.stack = self.deploy_cloudformation_stack( + stack_name, template_file, parameters + ) + outputs = self.stack.outputs + for output in outputs: + if output.get("OutputKey") == "RoleARN": + self.role_arn = output.get("OutputValue") + elif output.get("OutputKey") == "SNStopicARN": + self.sns_topic_arn = output.get("OutputValue") + + if not self.sns_topic_arn or not self.role_arn: + error_string = f""" + Failed to retrieve required outputs from CloudFormation stack. + 'sns_topic_arn'={self.sns_topic_arn}, 'role_arn'={self.role_arn} + """ + logger.error(error_string) + raise ValueError(error_string) + + print(f"Stack output RoleARN: {self.role_arn}") + print(f"Stack output SNStopicARN: a") + schedule_group_name = "workflow-schedules-group" + schedule_group_arn = self.eventbridge_scheduler.create_schedule_group( + schedule_group_name + ) + print( + f"Successfully created schedule group '{self.schedule_group_name}': {schedule_group_arn}." + ) + self.schedule_group_name = schedule_group_name + print("Application preparation complete.") + + def create_one_time_schedule(self) -> None: + """ + Creates a one-time schedule to send an initial event. + """ + schedule_name = q.ask("Enter a name for the one-time schedule:") + + scheduled_time = datetime.now(timezone.utc) + timedelta(minutes=1) + formatted_scheduled_time = scheduled_time.strftime("%Y-%m-%dT%H:%M:%S") + + print( + f"Creating a one-time schedule named '{schedule_name}' " + + f"\nto send an initial event in 1 minute with a flexible time window..." + ) + + schedule_arn = self.eventbridge_scheduler.create_schedule( + schedule_name, + f"at({formatted_scheduled_time})", + self.schedule_group_name, + self.sns_topic_arn, + self.role_arn, + f"One time scheduled event test from schedule {schedule_name}.", + delete_after_completion=True, + use_flexible_time_window=True, + ) + print( + f"Successfully created schedule '{schedule_name}' in schedule group 'workflow-schedules-group': {schedule_arn}." + ) + print(f"Subscription email will receive an email from this event.") + print(f"You must confirm your subscription to receive event emails.") + print(f"One-time schedule '{schedule_name}' created successfully.") + + def create_recurring_schedule(self) -> None: + """ + Create a recurring schedule to send events at a specified rate in minutes. + """ + + print("Creating a recurring schedule to send events for one hour...") + schedule_name = q.ask("Enter a name for the recurring schedule: ") + schedule_rate_in_minutes = q.ask( + "Enter the desired schedule rate (in minutes): ", q.is_int + ) + + schedule_arn = self.eventbridge_scheduler.create_schedule( + schedule_name, + f"rate({schedule_rate_in_minutes} minutes)", + self.schedule_group_name, + self.sns_topic_arn, + self.role_arn, + f"Recurrent event test from schedule {schedule_name}.", + ) + + print( + f"Successfully created schedule '{schedule_name}' in schedule group 'workflow-schedules-group': {schedule_arn}." + ) + print(f"Subscription email will receive an email from this event.") + print(f"You must confirm your subscription to receive event emails.") + + if q.ask( + f"Are you ready to delete the '{schedule_name}' schedule? (y/n)", q.is_yesno + ): + self.eventbridge_scheduler.delete_schedule( + schedule_name, self.schedule_group_name + ) + + def deploy_cloudformation_stack( + self, stack_name: str, cfn_template: str, parameters: [dict[str, str]] + ) -> ServiceResource: + """ + Deploys prerequisite resources used by the scenario. The resources are + defined in the associated `cfn_template.yaml` AWS CloudFormation script and are deployed + as a CloudFormation stack, so they can be easily managed and destroyed. + + :param stack_name: The name of the CloudFormation stack. + :param cfn_template: The CloudFormation template as a string. + :param parameters: The parameters for the CloudFormation stack. + :return: The CloudFormation stack resource. + """ + print(f"Deploying CloudFormation stack: {stack_name}.") + stack = self.cloud_formation_resource.create_stack( + StackName=stack_name, + TemplateBody=cfn_template, + Capabilities=["CAPABILITY_NAMED_IAM"], + Parameters=parameters, + ) + print(f"CloudFormation stack creation started: {stack_name}") + print("Waiting for CloudFormation stack creation to complete...") + waiter = self.cloud_formation_resource.meta.client.get_waiter( + "stack_create_complete" + ) + waiter.wait(StackName=stack.name) + stack.load() + print("CloudFormation stack creation complete.") + + return stack + + def destroy_cloudformation_stack(self, stack: ServiceResource) -> None: + """ + Destroys the resources managed by the CloudFormation stack, and the CloudFormation + stack itself. + + :param stack: The CloudFormation stack that manages the example resources. + """ + print( + f"CloudFormation stack '{stack.name}' is being deleted. This may take a few minutes." + ) + stack.delete() + waiter = self.cloud_formation_resource.meta.client.get_waiter( + "stack_delete_complete" + ) + waiter.wait(StackName=stack.name) + print(f"CloudFormation stack '{stack.name}' has been deleted.") + + def cleanup(self) -> None: + """ + Deletes the CloudFormation stack and the resources created for the demo. + """ + + if self.schedule_group_name: + schedule_group_name = self.schedule_group_name + self.schedule_group_name = None + self.eventbridge_scheduler.delete_schedule_group(schedule_group_name) + print(f"Successfully deleted schedule group '{schedule_group_name}'.") + + if self.stack is not None: + stack = self.stack + self.stack = None + self.destroy_cloudformation_stack(stack) + print("Stack deleted, demo complete.") + + @staticmethod + def get_template_as_string() -> str: + """ + Returns a string containing this scenario's CloudFormation template. + """ + script_directory = os.path.dirname(os.path.abspath(__file__)) + template_file_path = os.path.join(script_directory, "cfn_template.yaml") + file = open(template_file_path, "r") + return file.read() + + +if __name__ == "__main__": + demo: SchedulerScenario = None + try: + scheduler_wrapper = SchedulerWrapper.from_client() + cloud_formation_resource = resource("cloudformation") + demo = SchedulerScenario(scheduler_wrapper, cloud_formation_resource) + demo.run() + + except Exception as exception: + logging.exception("Something went wrong with the demo!") + if demo is not None: + demo.cleanup() + +# snippet-end:[python.example_code.scheduler.FeatureScenario] diff --git a/python/example_code/scheduler/scenario/test/conftest.py b/python/example_code/scheduler/scenario/test/conftest.py new file mode 100644 index 00000000000..84c4303f625 --- /dev/null +++ b/python/example_code/scheduler/scenario/test/conftest.py @@ -0,0 +1,63 @@ +# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +# SPDX-License-Identifier: Apache-2.0 + +""" +Contains common test fixtures used to run unit tests. +""" + +import sys + +import boto3 +import pytest +import os + + +script_dir = os.path.dirname(os.path.abspath(__file__)) + +# Add relative path to include SchedulerWrapper. +sys.path.append(script_dir) +sys.path.append(os.path.dirname(script_dir)) +import scheduler_scenario +from scheduler_wrapper import SchedulerWrapper + +# Add relative path to include demo_tools in this code example without need for setup. +sys.path.append(os.path.join(script_dir, "../../..")) + +from test_tools.fixtures.common import * + + +class ScenarioData: + def __init__( + self, + scheduler_client, + cloud_formation_resource, + scheduler_stubber, + cloud_formation_stubber, + ): + self.scheduler_client = scheduler_client + self.cloud_formation_resource = cloud_formation_resource + self.scheduler_stubber = scheduler_stubber + self.cloud_formation_stubber = cloud_formation_stubber + self.scenario = scheduler_scenario.SchedulerScenario( + scheduler_wrapper=SchedulerWrapper(self.scheduler_client), + cloud_formation_resource=self.cloud_formation_resource, + ) + + +@pytest.fixture +def scenario_data(make_stubber): + scheduler_client = boto3.client("scheduler") + scheduler_stubber = make_stubber(scheduler_client) + cloud_formation_resource = boto3.resource("cloudformation") + cloud_formation_stubber = make_stubber(cloud_formation_resource.meta.client) + return ScenarioData( + scheduler_client, + cloud_formation_resource, + scheduler_stubber, + cloud_formation_stubber, + ) + + +@pytest.fixture +def mock_wait(monkeypatch): + return diff --git a/python/example_code/scheduler/scenario/test/test_scenario_cleanup.py b/python/example_code/scheduler/scenario/test/test_scenario_cleanup.py new file mode 100644 index 00000000000..bfc40f5c2fb --- /dev/null +++ b/python/example_code/scheduler/scenario/test/test_scenario_cleanup.py @@ -0,0 +1,58 @@ +# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +# SPDX-License-Identifier: Apache-2.0 + +""" +Unit tests for cleanup in scheduler_scenario.py. +""" + +import pytest +from botocore.exceptions import ClientError +from botocore import waiter + +class MockManager: + def __init__(self, stub_runner, scenario_data, input_mocker): + self.scenario_data = scenario_data + self.stack_name = "python-tests" + self.schedule_group_name = "workflow-schedules-group" + scenario_data.scenario.schedule_group_name = self.schedule_group_name + scenario_data.scenario.stack = scenario_data.cloud_formation_resource.Stack(self.stack_name) + self.stub_runner = stub_runner + + def setup_stubs(self, error, stop_on, scheduler_stubber, cloud_formation_stubber, monkeypatch): + with self.stub_runner(error, stop_on) as runner: + runner.add(scheduler_stubber.stub_delete_schedule_group, self.schedule_group_name) + runner.add(cloud_formation_stubber.stub_delete_stack, self.stack_name) + + def mock_wait(self, **kwargs): + return + + monkeypatch.setattr(waiter.Waiter, "wait", mock_wait) + + + +@pytest.fixture +def mock_mgr(stub_runner, scenario_data, input_mocker): + return MockManager(stub_runner, scenario_data, input_mocker) + +@pytest.mark.integ +def test_scenario_cleanup(mock_mgr, capsys, monkeypatch): + mock_mgr.setup_stubs(None, None, mock_mgr.scenario_data.scheduler_stubber, + mock_mgr.scenario_data.cloud_formation_stubber, monkeypatch) + + mock_mgr.scenario_data.scenario.cleanup() + + +@pytest.mark.parametrize( + "error, stop_on_index", + [ + ("TESTERROR-stub_delete_schedule_group", 0), + ("TESTERROR-stub_delete_stack", 1), + ], +) + +@pytest.mark.integ +def test_scenario_cleanup_error(mock_mgr, caplog, error, stop_on_index, monkeypatch): + mock_mgr.setup_stubs(error, stop_on_index, mock_mgr.scenario_data.scheduler_stubber, mock_mgr.scenario_data.cloud_formation_stubber, monkeypatch) + + with pytest.raises(ClientError) as exc_info: + mock_mgr.scenario_data.scenario.cleanup() diff --git a/python/example_code/scheduler/scenario/test/test_scenario_create_one_time_schedule.py b/python/example_code/scheduler/scenario/test/test_scenario_create_one_time_schedule.py new file mode 100644 index 00000000000..823a96b6428 --- /dev/null +++ b/python/example_code/scheduler/scenario/test/test_scenario_create_one_time_schedule.py @@ -0,0 +1,64 @@ +# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +# SPDX-License-Identifier: Apache-2.0 + +""" +Unit tests for create_one_time_schedule in scheduler_scenario.py. +""" + +import pytest +from botocore.exceptions import ClientError +from scheduler_scenario import SchedulerScenario +from botocore import waiter + +class MockManager: + def __init__(self, stub_runner, scenario_data, input_mocker): + self.scenario_data = scenario_data + self.schedule_name = "python-test" + self.schedule_group_name = "workflow-schedules-group" + self.role_arn = "arn:aws:iam::123456789012:role/Test-Role" + self.sns_topic_arn = "arn:aws:sns:us-west-2:123456789012:my-topic" + self.schedule_expression = "at('2024-10-16T15:03:00')" + self.schedule_input = f"One time scheduled event test from schedule {self.schedule_name}." + self.schedule_arn = f"arn:aws:scheduler:us-east-1:123456789012:schedule/{self.schedule_group_name}/{self.schedule_name}" + scenario_data.scenario.sns_topic_arn = self.sns_topic_arn + scenario_data.scenario.role_arn = self.role_arn + scenario_data.scenario.schedule_group_name = "workflow-schedules-group" + answers = [ + self.schedule_name, + ] + input_mocker.mock_answers(answers) + self.stub_runner = stub_runner + + def setup_stubs(self, error, stop_on, scheduler_stubber): + with self.stub_runner(error, stop_on) as runner: + runner.add(scheduler_stubber.stub_create_schedule, self.schedule_arn, self.schedule_name, + self.schedule_expression, self.schedule_group_name, self.sns_topic_arn, self.role_arn, + self.schedule_input, + delete_after_completion=True, + use_flexible_time_window=True,) + + +@pytest.fixture +def mock_mgr(stub_runner, scenario_data, input_mocker): + return MockManager(stub_runner, scenario_data, input_mocker) + + +@pytest.mark.integ +def test_scenario_create_one_time_schedule(mock_mgr, capsys): + mock_mgr.setup_stubs(None, None, mock_mgr.scenario_data.scheduler_stubber) + + mock_mgr.scenario_data.scenario.create_one_time_schedule() + + +@pytest.mark.parametrize( + "error, stop_on_index", + [ + ("TESTERROR-stub_create_schedule", 0), + ], +) +@pytest.mark.integ +def test_scenario_create_one_time_schedule_error(mock_mgr, caplog, error, stop_on_index): + mock_mgr.setup_stubs(error, stop_on_index, mock_mgr.scenario_data.scheduler_stubber) + + with pytest.raises(ClientError) as exc_info: + mock_mgr.scenario_data.scenario.create_one_time_schedule() diff --git a/python/example_code/scheduler/scenario/test/test_scenario_create_recurring_schedule.py b/python/example_code/scheduler/scenario/test/test_scenario_create_recurring_schedule.py new file mode 100644 index 00000000000..9945dbdf749 --- /dev/null +++ b/python/example_code/scheduler/scenario/test/test_scenario_create_recurring_schedule.py @@ -0,0 +1,66 @@ +# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +# SPDX-License-Identifier: Apache-2.0 + +""" +Unit tests for create_recurring_schedule in scheduler_scenario.py. +""" + +import pytest +from botocore.exceptions import ClientError +from scheduler_scenario import SchedulerScenario +from botocore import waiter + +class MockManager: + def __init__(self, stub_runner, scenario_data, input_mocker): + self.scenario_data = scenario_data + self.schedule_name = "python-test2" + self.schedule_group_name = "workflow-schedules-group" + self.role_arn = "arn:aws:iam::123456789012:role/Test-Role" + self.sns_topic_arn = "arn:aws:sns:us-west-2:123456789012:my-topic" + self.schedule_rate_in_minutes = 5 + self.schedule_expression = f"rate({self.schedule_rate_in_minutes} minutes)" + self.schedule_input = f"Recurrent event test from schedule {self.schedule_name}." + self.schedule_arn = f"arn:aws:scheduler:us-east-1:123456789012:schedule/{self.schedule_group_name}/{self.schedule_name}" + scenario_data.scenario.sns_topic_arn = self.sns_topic_arn + scenario_data.scenario.role_arn = self.role_arn + scenario_data.scenario.schedule_group_name = "workflow-schedules-group" + answers = [ + self.schedule_name, + str(self.schedule_rate_in_minutes), + "y" + ] + input_mocker.mock_answers(answers) + self.stub_runner = stub_runner + + def setup_stubs(self, error, stop_on, scheduler_stubber): + with self.stub_runner(error, stop_on) as runner: + runner.add(scheduler_stubber.stub_create_schedule, self.schedule_arn, self.schedule_name, + self.schedule_expression, self.schedule_group_name, self.sns_topic_arn, self.role_arn, + self.schedule_input) + runner.add(scheduler_stubber.stub_delete_schedule, self.schedule_name, self.schedule_group_name) + + +@pytest.fixture +def mock_mgr(stub_runner, scenario_data, input_mocker): + return MockManager(stub_runner, scenario_data, input_mocker) + +@pytest.mark.integ +def test_scenario_create_recurring_schedule(mock_mgr, capsys): + mock_mgr.setup_stubs(None, None, mock_mgr.scenario_data.scheduler_stubber) + + mock_mgr.scenario_data.scenario.create_recurring_schedule() + + +@pytest.mark.parametrize( + "error, stop_on_index", + [ + ("TESTERROR-stub_create_schedule", 0), +("TESTERROR-stub_delete_schedule", 1), + ], +) +@pytest.mark.integ +def test_scenario_create_recurring_schedule_error(mock_mgr, caplog, error, stop_on_index): + mock_mgr.setup_stubs(error, stop_on_index, mock_mgr.scenario_data.scheduler_stubber) + + with pytest.raises(ClientError) as exc_info: + mock_mgr.scenario_data.scenario.create_recurring_schedule() diff --git a/python/example_code/scheduler/scenario/test/test_scenario_prepare_application.py b/python/example_code/scheduler/scenario/test/test_scenario_prepare_application.py new file mode 100644 index 00000000000..0040a289eae --- /dev/null +++ b/python/example_code/scheduler/scenario/test/test_scenario_prepare_application.py @@ -0,0 +1,80 @@ +# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +# SPDX-License-Identifier: Apache-2.0 + +""" +Unit tests for prepare_application in scheduler_scenario.py. +""" + +import pytest +from botocore.exceptions import ClientError +from scheduler_scenario import SchedulerScenario +from botocore import waiter + +class MockManager: + def __init__(self, stub_runner, scenario_data, input_mocker): + self.scenario_data = scenario_data + self.email_address = "carlos@example.com" + self.stack_name = "python-tests" + self.parameters = [{"ParameterKey": "email", "ParameterValue": self.email_address}] + self.capabilities = ['CAPABILITY_NAMED_IAM'] + self.cfn_template = SchedulerScenario.get_template_as_string() + self.stack_id = "arn:aws:cloudformation:us-east-1:123456789012:stack/myteststack/466df9e0-0dff-08e3-8e2f-5088487c4896" + self.outputs = [ + { + 'OutputKey': 'RoleARN', + 'OutputValue': 'arn:aws:iam::123456789012:role/Test-Role' + }, + { + 'OutputKey': 'SNStopicARN', + 'OutputValue': 'arn:aws:sns:us-west-2:123456789012:my-topic' + } + ] + self.schedule_group_name = "workflow-schedules-group" + self.schedule_group_arn = "arn:aws:scheduler:us-east-1:123456789012:schedule-group/tests" + answers = [ + self.email_address, + self.stack_name + ] + input_mocker.mock_answers(answers) + self.stub_runner = stub_runner + + def setup_stubs(self, error, stop_on, scheduler_stubber, cloud_formation_stubber, monkeypatch): + with self.stub_runner(error, stop_on) as runner: + runner.add(cloud_formation_stubber.stub_create_stack, self.stack_name, self.cfn_template, + self.capabilities, self.stack_id, self.parameters) + runner.add(cloud_formation_stubber.stub_describe_stacks, self.stack_name, "CREATE_COMPLETE", self.outputs) + runner.add(scheduler_stubber.stub_create_schedule_group, self.schedule_group_name, self.schedule_group_arn) + + def mock_wait(self, **kwargs): + return + + monkeypatch.setattr(waiter.Waiter, "wait", mock_wait) + + + +@pytest.fixture +def mock_mgr(stub_runner, scenario_data, input_mocker): + return MockManager(stub_runner, scenario_data, input_mocker) + +@pytest.mark.integ +def test_scenario_prepare_application(mock_mgr, capsys, monkeypatch): + mock_mgr.setup_stubs(None, None, mock_mgr.scenario_data.scheduler_stubber, + mock_mgr.scenario_data.cloud_formation_stubber, monkeypatch) + + mock_mgr.scenario_data.scenario.prepare_application() + + +@pytest.mark.parametrize( + "error, stop_on_index", + [ + ("TESTERROR-stub_create_stack", 0), + ("TESTERROR-stub_describe_stacks", 1), + ("TESTERROR-stub_create_schedule_group", 2), + ], +) +@pytest.mark.integ +def test_scenario_prepare_application_error(mock_mgr, caplog, error, stop_on_index, monkeypatch): + mock_mgr.setup_stubs(error, stop_on_index, mock_mgr.scenario_data.scheduler_stubber, mock_mgr.scenario_data.cloud_formation_stubber, monkeypatch) + + with pytest.raises(ClientError) as exc_info: + mock_mgr.scenario_data.scenario.prepare_application() diff --git a/python/example_code/scheduler/scheduler_wrapper.py b/python/example_code/scheduler/scheduler_wrapper.py new file mode 100644 index 00000000000..e7cced989c8 --- /dev/null +++ b/python/example_code/scheduler/scheduler_wrapper.py @@ -0,0 +1,192 @@ +# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +# SPDX-License-Identifier: Apache-2.0 + +""" +Purpose + +Shows how to use the AWS SDK for Python (Boto3) with Amazon EventBridge Scheduler API to schedule +and receive events. +""" + +import logging +from datetime import datetime, timedelta, timezone +import boto3 +from boto3 import client +from botocore.exceptions import ClientError + +logger = logging.getLogger(__name__) + + +# snippet-start:[python.example_code.scheduler.EventSchedulerWrapper.class] +# snippet-start:[python.example_code.scheduler.EventSchedulerWrapper.decl] +class SchedulerWrapper: + def __init__(self, eventbridge_scheduler_client: client): + self.scheduler_client = eventbridge_scheduler_client + + @classmethod + def from_client(cls) -> "SchedulerWrapper": + """ + Creates a SchedulerWrapper instance with a default EventBridge Scheduler client. + + :return: An instance of SchedulerWrapper initialized with the default EventBridge Scheduler client. + """ + eventbridge_scheduler_client = boto3.client("scheduler") + return cls(eventbridge_scheduler_client) + + # snippet-end:[python.example_code.scheduler.EventSchedulerWrapper.decl] + + # snippet-start:[python.example_code.scheduler.CreateSchedule] + def create_schedule( + self, + name: str, + schedule_expression: str, + schedule_group_name: str, + target_arn: str, + role_arn: str, + input: str, + delete_after_completion: bool = False, + use_flexible_time_window: bool = False, + ) -> str: + """ + Creates a new schedule with the specified parameters. + + :param name: The name of the schedule. + :param schedule_expression: The expression that defines when the schedule runs. + :param schedule_group_name: The name of the schedule group. + :param target_arn: The Amazon Resource Name (ARN) of the target. + :param role_arn: The Amazon Resource Name (ARN) of the execution IAM role. + :param input: The input for the target. + :param delete_after_completion: Whether to delete the schedule after it completes. + :param use_flexible_time_window: Whether to use a flexible time window. + + :return The ARN of the created schedule. + """ + try: + hours_to_run = 1 + flexible_time_window_minutes = 10 + parameters = { + "Name": name, + "ScheduleExpression": schedule_expression, + "GroupName": schedule_group_name, + "Target": {"Arn": target_arn, "RoleArn": role_arn, "Input": input}, + "StartDate": datetime.now(timezone.utc), + "EndDate": datetime.now(timezone.utc) + timedelta(hours=hours_to_run), + } + + if delete_after_completion: + parameters["ActionAfterCompletion"] = "DELETE" + + if use_flexible_time_window: + parameters["FlexibleTimeWindow"] = { + "Mode": "FLEXIBLE", + "MaximumWindowInMinutes": flexible_time_window_minutes, + } + else: + parameters["FlexibleTimeWindow"] = {"Mode": "OFF"} + + response = self.scheduler_client.create_schedule(**parameters) + return response["ScheduleArn"] + except ClientError as err: + if err.response["Error"]["Code"] == "ConflictException": + logger.error( + "Failed to create schedule '%s' due to a conflict. %s", + name, + err.response["Error"]["Message"], + ) + else: + logger.error( + "Error creating schedule: %s", err.response["Error"]["Message"] + ) + raise + + # snippet-end:[python.example_code.scheduler.CreateSchedule] + + # snippet-start:[python.example_code.scheduler.DeleteSchedule] + def delete_schedule(self, name: str, schedule_group_name: str) -> None: + """ + Deletes the schedule with the specified name and schedule group. + + :param name: The name of the schedule. + :param schedule_group_name: The name of the schedule group. + """ + try: + self.scheduler_client.delete_schedule( + Name=name, GroupName=schedule_group_name + ) + except ClientError as err: + if err.response["Error"]["Code"] == "ResourceNotFoundException": + logger.error( + "Failed to delete schedule with ID '%s' because the resource was not found: %s", + name, + err.response["Error"]["Message"], + ) + else: + logger.error( + "Error deleting schedule: %s", err.response["Error"]["Message"] + ) + raise + + # snippet-end:[python.example_code.scheduler.DeleteSchedule] + + # snippet-start:[python.example_code.scheduler.CreateScheduleGroup] + def create_schedule_group(self, name: str) -> str: + """ + Creates a new schedule group with the specified name and description. + + :param name: The name of the schedule group. + :param description: The description of the schedule group. + + :return: The ARN of the created schedule group. + """ + try: + response = self.scheduler_client.create_schedule_group(Name=name) + return response["ScheduleGroupArn"] + except ClientError as err: + if err.response["Error"]["Code"] == "ConflictException": + logger.error( + "Failed to create schedule group '%s' due to a conflict. %s", + name, + err.response["Error"]["Message"], + ) + else: + logger.error( + "Error creating schedule group: %s", + err.response["Error"]["Message"], + ) + raise + + # snippet-end:[python.example_code.scheduler.CreateScheduleGroup] + + # snippet-start:[python.example_code.scheduler.DeleteScheduleGroup] + def delete_schedule_group(self, name: str) -> None: + """ + Deletes the schedule group with the specified name. + + :param name: The name of the schedule group. + """ + try: + self.scheduler_client.delete_schedule_group(Name=name) + logger.info("Schedule group %s deleted successfully.", name) + except ClientError as err: + if err.response["Error"]["Code"] == "ResourceNotFoundException": + logger.error( + "Failed to delete schedule group with ID '%s' because the resource was not found: %s", + name, + err.response["Error"]["Message"], + ) + else: + logger.error( + "Error deleting schedule group: %s", + err.response["Error"]["Message"], + ) + raise + # snippet-end:[python.example_code.scheduler.DeleteScheduleGroup] + + +# snippet-end:[python.example_code.scheduler.EventSchedulerWrapper.class] + +if __name__ == "__main__": + try: + eventbridge = SchedulerWrapper.from_client() + except Exception: + logging.exception("Something went wrong with the demo!") diff --git a/python/test_tools/cloudformation_stubber.py b/python/test_tools/cloudformation_stubber.py index 35f76dabdcb..d179aee760e 100644 --- a/python/test_tools/cloudformation_stubber.py +++ b/python/test_tools/cloudformation_stubber.py @@ -25,13 +25,15 @@ def __init__(self, client, use_stubs=True): super().__init__(client, use_stubs) def stub_create_stack( - self, stack_name, setup_template, capabilities, stack_id, error_code=None + self, stack_name, setup_template, capabilities, stack_id, parameters= None, error_code=None ): expected_params = { "StackName": stack_name, "TemplateBody": setup_template, "Capabilities": capabilities, } + if parameters is not None: + expected_params["Parameters"] = parameters response = {"StackId": stack_id} self._stub_bifurcator( "create_stack", expected_params, response, error_code=error_code diff --git a/python/test_tools/scheduler_stubber.py b/python/test_tools/scheduler_stubber.py new file mode 100644 index 00000000000..8784835f75a --- /dev/null +++ b/python/test_tools/scheduler_stubber.py @@ -0,0 +1,137 @@ +# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +# SPDX-License-Identifier: Apache-2.0 + +""" +Stub functions that are used by the Amazon EventBridge Scheduler unit tests. + +When tests are run against an actual AWS account, the stubber class does not +set up stubs and passes all calls through to the Boto 3 client. +""" + +import io +import json +from botocore.stub import ANY +from boto3 import client + +from test_tools.example_stubber import ExampleStubber + +from datetime import timedelta, timezone, datetime + + + +class SchedulerStubber(ExampleStubber): + """ + A class that implements a variety of stub functions that are used by the + Amazon EventBridge Scheduler unit tests. + + The stubbed functions all expect certain parameters to be passed to them as + part of the tests, and will raise errors when the actual parameters differ from + the expected. + """ + + def __init__(self, scheduler_client: client, use_stubs=True) -> None: + """ + Initializes the object with a specific client and configures it for + stubbing or AWS passthrough. + + :param scheduler_client: A Boto 3 Amazon EventBridge Scheduler client. + :param use_stubs: When True, use stubs to intercept requests. Otherwise, + pass requests through to AWS. + """ + super().__init__(scheduler_client, use_stubs) + + def stub_create_schedule(self, + schedule_arn: str, + name: str, + schedule_expression: str, + schedule_group_name: str, + target_arn: str, + role_arn: str, + input: str, + delete_after_completion: bool = False, + use_flexible_time_window: bool = False, + error_code: str=None) -> None: + """ + Stub the create_schedule function. + + :param schedule_arn: The ARN of the created schedule. + :param name: The name of the schedule. + :param schedule_expression: The expression that defines when the schedule runs. + :param schedule_group_name: The name of the schedule group. + :param target_arn: The Amazon Resource Name (ARN) of the target. + :param role_arn: The Amazon Resource Name (ARN) of the execution IAM role. + :param input: The input for the target. + :param delete_after_completion: Whether to delete the schedule after it completes. + :param use_flexible_time_window: Whether to use a flexible time window. + :param error_code: Simulated error code to raise. + :return: None + """ + flexible_time_window_minutes = 10 + expected_params = {"Name": name, + "ScheduleExpression": ANY, + "GroupName": schedule_group_name, + "Target": {"Arn": target_arn, "RoleArn": role_arn, "Input": input}, + "StartDate":ANY, + "EndDate": ANY, + } + if delete_after_completion: + expected_params["ActionAfterCompletion"] = "DELETE" + + if use_flexible_time_window: + expected_params["FlexibleTimeWindow"] = { + "Mode": "FLEXIBLE", + "MaximumWindowInMinutes": flexible_time_window_minutes, + } + else: + expected_params["FlexibleTimeWindow"] = {"Mode": "OFF"} + + response = {"ScheduleArn": schedule_arn} + self._stub_bifurcator( + "create_schedule", expected_params, response, error_code=error_code + ) + + def stub_create_schedule_group(self, group_name: str, schedule_group_arn: str, error_code: str =None) -> None: + """ + Stub the create_schedule_group function. + + :param group_name: The name of the schedule group. + :param schedule_group_arn: The ARN of the created schedule group. + :param error_code: Simulated error code to raise. + :return: None + """ + expected_params = {"Name": group_name} + response = {"ScheduleGroupArn": schedule_group_arn} + self._stub_bifurcator( + "create_schedule_group", expected_params, response, error_code=error_code + ) + + def stub_delete_schedule(self, name: str, schedule_group_name: str, error_code: str =None) -> None: + """ + Stub the delete_schedule function. + + :param name: The name of the schedule. + :param schedule_group_name: The name of the schedule group. + :param error_code: Simulated error code to raise. + :return: None + """ + expected_params = {"Name": name, + "GroupName": schedule_group_name} + response = {} + self._stub_bifurcator( + "delete_schedule", expected_params, response, error_code=error_code + ) + + def stub_delete_schedule_group(self, schedule_group_name: str, error_code: str =None) -> None: + """ + Stub the delete_schedule_group function. + + :param schedule_group_name: The name of the schedule group. + :param error_code: Simulated error code to raise. + :return: None + """ + expected_params = {"Name": schedule_group_name} + response = {} + self._stub_bifurcator( + "delete_schedule_group", expected_params, response, error_code=error_code + ) + diff --git a/python/test_tools/stubber_factory.py b/python/test_tools/stubber_factory.py index 5283ac6a240..e0164656a80 100644 --- a/python/test_tools/stubber_factory.py +++ b/python/test_tools/stubber_factory.py @@ -63,6 +63,7 @@ from test_tools.medical_imaging_stubber import MedicalImagingStubber from test_tools.redshift_stubber import RedshiftStubber from test_tools.redshift_data_stubber import RedshiftDataStubber +from test_tools.scheduler_stubber import SchedulerStubber class StubberFactoryNotImplemented(Exception): @@ -160,6 +161,8 @@ def stubber_factory(service_name): return S3Stubber elif service_name == "s3control": return S3ControlStubber + elif service_name == "scheduler": + return SchedulerStubber elif service_name == "secretsmanager": return SecretsManagerStubber elif service_name == "ses":