Scheduler: get_schedule() now returns the CreationDate and LastModificationDate attributes (#6510)

This commit is contained in:
Bert Blommers 2023-07-11 09:32:22 +00:00 committed by GitHub
parent 2dbbdc3123
commit 6668310884
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 45 additions and 9 deletions

View File

@ -39,6 +39,7 @@ class Schedule(BaseModel):
self.kms_key_arn = kms_key_arn self.kms_key_arn = kms_key_arn
self.start_date = start_date self.start_date = start_date
self.end_date = end_date self.end_date = end_date
self.creation_date = self.last_modified_date = unix_time()
@staticmethod @staticmethod
def validate_target(target: Dict[str, Any]) -> Dict[str, Any]: # type: ignore[misc] def validate_target(target: Dict[str, Any]) -> Dict[str, Any]: # type: ignore[misc]
@ -63,11 +64,36 @@ class Schedule(BaseModel):
"KmsKeyArn": self.kms_key_arn, "KmsKeyArn": self.kms_key_arn,
"StartDate": self.start_date, "StartDate": self.start_date,
"EndDate": self.end_date, "EndDate": self.end_date,
"CreationDate": self.creation_date,
"LastModificationDate": self.last_modified_date,
} }
if short: if short:
dct["Target"] = {"Arn": dct["Target"]["Arn"]} dct["Target"] = {"Arn": dct["Target"]["Arn"]}
return dct return dct
def update(
self,
description: str,
end_date: str,
flexible_time_window: Dict[str, Any],
kms_key_arn: str,
schedule_expression: str,
schedule_expression_timezone: str,
start_date: str,
state: str,
target: Dict[str, Any],
) -> None:
self.schedule_expression = schedule_expression
self.schedule_expression_timezone = schedule_expression_timezone
self.flexible_time_window = flexible_time_window
self.target = Schedule.validate_target(target)
self.description = description
self.state = state
self.kms_key_arn = kms_key_arn
self.start_date = start_date
self.end_date = end_date
self.last_modified_date = unix_time()
class ScheduleGroup(BaseModel): class ScheduleGroup(BaseModel):
def __init__(self, region: str, account_id: str, name: str): def __init__(self, region: str, account_id: str, name: str):
@ -173,15 +199,17 @@ class EventBridgeSchedulerBackend(BaseBackend):
The ClientToken is not yet implemented The ClientToken is not yet implemented
""" """
schedule = self.get_schedule(group_name=group_name, name=name) schedule = self.get_schedule(group_name=group_name, name=name)
schedule.schedule_expression = schedule_expression schedule.update(
schedule.schedule_expression_timezone = schedule_expression_timezone description=description,
schedule.flexible_time_window = flexible_time_window end_date=end_date,
schedule.target = Schedule.validate_target(target) flexible_time_window=flexible_time_window,
schedule.description = description kms_key_arn=kms_key_arn,
schedule.state = state schedule_expression=schedule_expression,
schedule.kms_key_arn = kms_key_arn schedule_expression_timezone=schedule_expression_timezone,
schedule.start_date = start_date start_date=start_date,
schedule.end_date = end_date state=state,
target=target,
)
return schedule return schedule
def list_schedules( def list_schedules(

View File

@ -3,6 +3,7 @@ import boto3
import pytest import pytest
from botocore.client import ClientError from botocore.client import ClientError
from datetime import datetime
from moto import mock_scheduler from moto import mock_scheduler
from moto.core import DEFAULT_ACCOUNT_ID from moto.core import DEFAULT_ACCOUNT_ID
@ -44,6 +45,9 @@ def test_create_get_schedule():
"RoleArn": "n/a", "RoleArn": "n/a",
"RetryPolicy": {"MaximumEventAgeInSeconds": 86400, "MaximumRetryAttempts": 185}, "RetryPolicy": {"MaximumEventAgeInSeconds": 86400, "MaximumRetryAttempts": 185},
} }
assert isinstance(resp["CreationDate"], datetime)
assert isinstance(resp["LastModificationDate"], datetime)
assert resp["CreationDate"] == resp["LastModificationDate"]
@mock_scheduler @mock_scheduler
@ -135,6 +139,10 @@ def test_update_schedule(extra_kwargs):
"RetryPolicy": {"MaximumEventAgeInSeconds": 86400, "MaximumRetryAttempts": 185}, "RetryPolicy": {"MaximumEventAgeInSeconds": 86400, "MaximumRetryAttempts": 185},
} }
assert isinstance(schedule["CreationDate"], datetime)
assert isinstance(schedule["LastModificationDate"], datetime)
assert schedule["CreationDate"] != schedule["LastModificationDate"]
@mock_scheduler @mock_scheduler
def test_get_schedule_for_unknown_group(): def test_get_schedule_for_unknown_group():