Scheduler: create_scheduler() now errors on duplicate names (#6641)

moto/scheduler/exceptions.py

@@ -2,6 +2,11 @@
 from moto.core.exceptions import JsonRESTError
 
 
+class ScheduleExists(JsonRESTError):
+    def __init__(self, name: str) -> None:
+        super().__init__("ConflictException", f"Schedule {name} already exists.")
+
+
 class ScheduleNotFound(JsonRESTError):
     def __init__(self) -> None:
         super().__init__("ResourceNotFoundException", "Schedule not found")

moto/scheduler/models.py

@@ -5,6 +5,7 @@ from moto.core import BaseBackend, BackendDict, BaseModel
 from moto.core.utils import unix_time
 from moto.utilities.tagging_service import TaggingService
 
+from .exceptions import ScheduleExists
 from .exceptions import ScheduleNotFound, ScheduleGroupNotFound
 
 
@@ -155,6 +156,8 @@ class EventBridgeSchedulerBackend(BaseBackend):
         The ClientToken parameter is not yet implemented
         """
         group = self.schedule_groups[group_name or "default"]
+        if name in group.schedules:
+            raise ScheduleExists(name)
         schedule = Schedule(
             region=self.region_name,
             account_id=self.account_id,

tests/test_scheduler/test_scheduler.py

@@ -144,6 +144,26 @@ def test_update_schedule(extra_kwargs):
     assert schedule["CreationDate"] != schedule["LastModificationDate"]
 
 
+@mock_scheduler
+def test_create_duplicate_schedule():
+    client = boto3.client("scheduler", region_name="us-east-1")
+    params = {
+        "ScheduleExpression": "at(2022-12-12T00:00:00)",
+        "FlexibleTimeWindow": {
+            "MaximumWindowInMinutes": 4,
+            "Mode": "FLEXIBLE",
+        },
+        "Target": {"Arn": "arn1", "RoleArn": "arn2"},
+    }
+
+    client.create_schedule(Name="schedule1", **params)
+    with pytest.raises(ClientError) as exc:
+        client.create_schedule(Name="schedule1", **params)
+    err = exc.value.response["Error"]
+    assert err["Code"] == "ConflictException"
+    assert err["Message"] == "Schedule schedule1 already exists."
+
+
 @mock_scheduler
 def test_get_schedule_for_unknown_group():
     client = boto3.client("scheduler", region_name="eu-west-1")