From 60fd1a5cf2eb1c1c7f8c7bbd0a81c13c5797f3ac Mon Sep 17 00:00:00 2001 From: Bert Blommers Date: Mon, 11 Sep 2023 22:23:44 +0000 Subject: [PATCH] Techdebt: Improve date handling across Moto (#6803) --- moto/acm/models.py | 17 ++++---- moto/acmpca/models.py | 6 +-- moto/autoscaling/responses.py | 8 ++-- moto/awslambda/models.py | 20 ++++----- moto/cloudformation/models.py | 7 ++-- moto/cloudfront/models.py | 3 +- moto/cloudtrail/models.py | 16 ++++--- moto/cloudwatch/models.py | 22 +++------- moto/codebuild/models.py | 6 +-- moto/codecommit/models.py | 3 +- moto/codepipeline/models.py | 9 ++-- moto/cognitoidentity/models.py | 6 +-- moto/cognitoidp/models.py | 13 +++--- moto/config/models.py | 25 +++++------ moto/core/utils.py | 31 ++++++++++---- moto/core/versions.py | 4 ++ moto/datapipeline/models.py | 3 +- moto/dms/models.py | 7 ++-- moto/dynamodb/models/table.py | 13 +++--- moto/dynamodb_v20111205/models.py | 5 +-- moto/ec2/models/instances.py | 11 ++--- moto/ec2/models/key_pairs.py | 5 +-- moto/ec2/models/nat_gateways.py | 5 +-- moto/ec2/models/transit_gateway.py | 5 +-- .../ec2/models/transit_gateway_attachments.py | 5 +-- .../models/transit_gateway_route_tables.py | 5 +-- moto/ec2/utils.py | 4 +- moto/ecr/models.py | 14 +++---- moto/elasticache/models.py | 6 +-- moto/elbv2/models.py | 3 +- moto/emr/utils.py | 3 +- moto/events/models.py | 12 +++--- moto/firehose/models.py | 3 +- moto/glue/models.py | 34 +++++++-------- moto/greengrass/models.py | 42 +++++++++---------- moto/greengrass/responses.py | 9 +--- moto/iam/models.py | 33 ++++++++------- moto/instance_metadata/responses.py | 3 +- moto/iot/models.py | 7 ++-- moto/kinesis/models.py | 4 +- moto/kinesisvideo/models.py | 4 +- moto/logs/models.py | 8 ++-- moto/managedblockchain/models.py | 13 +++--- moto/opsworks/models.py | 10 ++--- moto/organizations/models.py | 11 +++-- moto/ram/models.py | 11 +++-- moto/rds/models.py | 29 ++++--------- moto/redshift/models.py | 8 +--- moto/route53/responses.py | 5 +-- moto/s3/models.py | 11 ++--- moto/s3control/config.py | 5 +-- moto/secretsmanager/models.py | 5 ++- moto/ses/models.py | 3 +- moto/ses/responses.py | 6 +-- moto/sesv2/models.py | 9 ++-- moto/sns/models.py | 5 +-- moto/ssm/models.py | 3 +- moto/stepfunctions/models.py | 8 ++-- moto/sts/models.py | 6 +-- moto/swf/models/activity_task.py | 5 +-- moto/swf/models/decision_task.py | 5 +-- moto/wafv2/models.py | 3 +- tests/test_acmpca/test_acmpca.py | 5 ++- tests/test_config/test_config.py | 31 ++++---------- tests/test_events/test_events_integration.py | 4 +- tests/test_iam/test_iam.py | 21 +++------- tests/test_iam/test_iam_groups.py | 3 +- tests/test_iam/test_iam_password_last_used.py | 4 +- tests/test_kinesis/test_kinesis.py | 7 ++-- .../test_kinesisvideoarchivedmedia.py | 5 ++- tests/test_logs/test_integration.py | 17 ++++---- tests/test_logs/test_logs.py | 16 +++---- tests/test_logs/test_logs_filter.py | 12 +++--- tests/test_logs/test_logs_query/test_boto3.py | 18 ++++---- tests/test_s3/test_s3.py | 9 ++-- tests/test_s3/test_s3_lock.py | 9 ++-- tests/test_sqs/test_server.py | 6 +-- .../responses/test_workflow_executions.py | 8 ++-- 78 files changed, 362 insertions(+), 423 deletions(-) diff --git a/moto/acm/models.py b/moto/acm/models.py index 470d68227..f899c771f 100644 --- a/moto/acm/models.py +++ b/moto/acm/models.py @@ -2,6 +2,7 @@ import base64 import re import datetime from moto.core import BaseBackend, BackendDict, BaseModel +from moto.core.utils import utcnow from moto import settings from typing import Any, Dict, List, Iterable, Optional, Tuple, Set @@ -122,7 +123,7 @@ class CertBundle(BaseModel): cert_type: str = "IMPORTED", cert_status: str = "ISSUED", ): - self.created_at = datetime.datetime.utcnow() + self.created_at = utcnow() self.cert = certificate self.key = private_key # AWS always returns your chain + root CA @@ -192,8 +193,8 @@ class CertBundle(BaseModel): .issuer_name(issuer) .public_key(key.public_key()) .serial_number(cryptography.x509.random_serial_number()) - .not_valid_before(datetime.datetime.utcnow()) - .not_valid_after(datetime.datetime.utcnow() + datetime.timedelta(days=365)) + .not_valid_before(utcnow()) + .not_valid_after(utcnow() + datetime.timedelta(days=365)) .add_extension( cryptography.x509.SubjectAlternativeName(unique_dns_names), critical=False, @@ -235,7 +236,7 @@ class CertBundle(BaseModel): self.cert, default_backend() ) - now = datetime.datetime.utcnow() + now = utcnow() if _cert.not_valid_after < now: raise AWSValidationException( "The certificate has expired, is not valid." @@ -264,7 +265,7 @@ class CertBundle(BaseModel): cert_armored, default_backend() ) - now = datetime.datetime.utcnow() + now = utcnow() if self._cert.not_valid_after < now: raise AWSValidationException( "The certificate chain has expired, is not valid." @@ -286,7 +287,7 @@ class CertBundle(BaseModel): # Basically, if the certificate is pending, and then checked again after a # while, it will appear as if its been validated. The default wait time is 60 # seconds but you can set an environment to change it. - waited_seconds = (datetime.datetime.utcnow() - self.created_at).total_seconds() + waited_seconds = (utcnow() - self.created_at).total_seconds() if ( self.type == "AMAZON_ISSUED" and self.status == "PENDING_VALIDATION" @@ -414,7 +415,7 @@ class AWSCertificateManagerBackend(BaseBackend): :param token: String token :return: None or ARN """ - now = datetime.datetime.utcnow() + now = utcnow() if token in self._idempotency_tokens: if self._idempotency_tokens[token]["expires"] < now: # Token has expired, new request @@ -428,7 +429,7 @@ class AWSCertificateManagerBackend(BaseBackend): def _set_idempotency_token_arn(self, token: str, arn: str) -> None: self._idempotency_tokens[token] = { "arn": arn, - "expires": datetime.datetime.utcnow() + datetime.timedelta(hours=1), + "expires": utcnow() + datetime.timedelta(hours=1), } def import_cert( diff --git a/moto/acmpca/models.py b/moto/acmpca/models.py index 8349e98a4..c07ad0f19 100644 --- a/moto/acmpca/models.py +++ b/moto/acmpca/models.py @@ -2,7 +2,7 @@ import base64 from .exceptions import ResourceNotFoundException from moto.core import BaseBackend, BackendDict, BaseModel -from moto.core.utils import unix_time +from moto.core.utils import unix_time, utcnow from moto.moto_api._internal import mock_random from moto.utilities.tagging_service import TaggingService @@ -78,8 +78,8 @@ class CertificateAuthority(BaseModel): .issuer_name(issuer) .public_key(self.key.public_key()) .serial_number(cryptography.x509.random_serial_number()) - .not_valid_before(datetime.datetime.utcnow()) - .not_valid_after(datetime.datetime.utcnow() + datetime.timedelta(days=365)) + .not_valid_before(utcnow()) + .not_valid_after(utcnow() + datetime.timedelta(days=365)) .sign(self.key, hashes.SHA512(), default_backend()) ) diff --git a/moto/autoscaling/responses.py b/moto/autoscaling/responses.py index 1cc30d6e0..563bfe578 100644 --- a/moto/autoscaling/responses.py +++ b/moto/autoscaling/responses.py @@ -1,5 +1,3 @@ -import datetime - from moto.core.responses import BaseResponse from moto.core.utils import iso_8601_datetime_with_milliseconds from moto.utilities.aws_headers import amz_crc32, amzn_request_id @@ -399,7 +397,7 @@ class AutoScalingResponse(BaseResponse): should_decrement=should_decrement, original_size=original_size, desired_capacity=desired_capacity, - timestamp=iso_8601_datetime_with_milliseconds(datetime.datetime.utcnow()), + timestamp=iso_8601_datetime_with_milliseconds(), ) @amz_crc32 @@ -417,7 +415,7 @@ class AutoScalingResponse(BaseResponse): standby_instances=standby_instances, original_size=original_size, desired_capacity=desired_capacity, - timestamp=iso_8601_datetime_with_milliseconds(datetime.datetime.utcnow()), + timestamp=iso_8601_datetime_with_milliseconds(), ) def suspend_processes(self) -> str: @@ -468,7 +466,7 @@ class AutoScalingResponse(BaseResponse): should_decrement=should_decrement, original_size=original_size, desired_capacity=desired_capacity, - timestamp=iso_8601_datetime_with_milliseconds(datetime.datetime.utcnow()), + timestamp=iso_8601_datetime_with_milliseconds(), ) def describe_tags(self) -> str: diff --git a/moto/awslambda/models.py b/moto/awslambda/models.py index 27dd9dc30..e098ebc50 100644 --- a/moto/awslambda/models.py +++ b/moto/awslambda/models.py @@ -26,7 +26,7 @@ import requests.exceptions from moto.awslambda.policy import Policy from moto.core import BaseBackend, BackendDict, BaseModel, CloudFormationModel from moto.core.exceptions import RESTError -from moto.core.utils import unix_time_millis, iso_8601_datetime_with_nanoseconds +from moto.core.utils import unix_time_millis, iso_8601_datetime_with_nanoseconds, utcnow from moto.iam.models import iam_backends from moto.iam.exceptions import IAMNotFoundException from moto.ecr.exceptions import ImageNotFoundException @@ -65,7 +65,7 @@ logger = logging.getLogger(__name__) def zip2tar(zip_bytes: bytes) -> io.BytesIO: tarstream = io.BytesIO() - timeshift = int((datetime.now() - datetime.utcnow()).total_seconds()) + timeshift = int((datetime.now() - utcnow()).total_seconds()) tarf = tarfile.TarFile(fileobj=tarstream, mode="w") with zipfile.ZipFile(io.BytesIO(zip_bytes), "r") as zipf: for zipinfo in zipf.infolist(): @@ -334,7 +334,7 @@ class LayerVersion(CloudFormationModel): self.license_info = spec.get("LicenseInfo", "") # auto-generated - self.created_date = datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S") + self.created_date = utcnow().strftime("%Y-%m-%d %H:%M:%S") self.version: Optional[int] = None self._attached = False self._layer: Optional["Layer"] = None @@ -555,7 +555,7 @@ class LambdaFunction(CloudFormationModel, DockerModel): # auto-generated self.version = version - self.last_modified = iso_8601_datetime_with_nanoseconds(datetime.utcnow()) + self.last_modified = iso_8601_datetime_with_nanoseconds() self._set_function_code(self.code) @@ -577,7 +577,7 @@ class LambdaFunction(CloudFormationModel, DockerModel): self.region, self.account_id, self.function_name, version ) self.version = version - self.last_modified = iso_8601_datetime_with_nanoseconds(datetime.utcnow()) + self.last_modified = iso_8601_datetime_with_nanoseconds() @property def architectures(self) -> List[str]: @@ -964,7 +964,7 @@ class LambdaFunction(CloudFormationModel, DockerModel): def save_logs(self, output: str) -> None: # Send output to "logs" backend invoke_id = random.uuid4().hex - date = datetime.utcnow() + date = utcnow() log_stream_name = ( f"{date.year}/{date.month:02d}/{date.day:02d}/[{self.version}]{invoke_id}" ) @@ -1118,7 +1118,7 @@ class FunctionUrlConfig: self.function = function self.config = config self.url = f"https://{random.uuid4().hex}.lambda-url.{function.region}.on.aws" - self.created = datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%S.000+0000") + self.created = utcnow().strftime("%Y-%m-%dT%H:%M:%S.000+0000") self.last_modified = self.created def to_dict(self) -> Dict[str, Any]: @@ -1137,7 +1137,7 @@ class FunctionUrlConfig: self.config["Cors"] = new_config["Cors"] if new_config.get("AuthType"): self.config["AuthType"] = new_config["AuthType"] - self.last_modified = datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%S") + self.last_modified = utcnow().strftime("%Y-%m-%dT%H:%M:%S") class EventSourceMapping(CloudFormationModel): @@ -1154,7 +1154,7 @@ class EventSourceMapping(CloudFormationModel): self.function_arn = spec["FunctionArn"] self.uuid = str(random.uuid4()) - self.last_modified = time.mktime(datetime.utcnow().timetuple()) + self.last_modified = time.mktime(utcnow().timetuple()) def _get_service_source_from_arn(self, event_source_arn: str) -> str: return event_source_arn.split(":")[2].lower() @@ -1887,7 +1887,7 @@ class LambdaBackend(BaseBackend): elif key == "Enabled": esm.enabled = spec[key] - esm.last_modified = time.mktime(datetime.utcnow().timetuple()) + esm.last_modified = time.mktime(utcnow().timetuple()) return esm def list_event_source_mappings( diff --git a/moto/cloudformation/models.py b/moto/cloudformation/models.py index f704eec7a..7615e00e0 100644 --- a/moto/cloudformation/models.py +++ b/moto/cloudformation/models.py @@ -11,6 +11,7 @@ from moto.core import BaseBackend, BackendDict, BaseModel, CloudFormationModel from moto.core.utils import ( iso_8601_datetime_with_milliseconds, iso_8601_datetime_without_milliseconds, + utcnow, ) from moto.moto_api._internal import mock_random from moto.sns.models import sns_backends @@ -407,7 +408,7 @@ class FakeStack(CloudFormationModel): self.custom_resources: Dict[str, CustomModel] = dict() self.output_map = self._create_output_map() - self.creation_time = datetime.utcnow() + self.creation_time = utcnow() self.status = "CREATE_PENDING" def has_template(self, other_template: str) -> bool: @@ -637,7 +638,7 @@ class FakeChangeSet(BaseModel): self.parameters = parameters self._parse_template() - self.creation_time = datetime.utcnow() + self.creation_time = utcnow() self.changes = self.diff() self.status: Optional[str] = None @@ -695,7 +696,7 @@ class FakeEvent(BaseModel): self.resource_status = resource_status self.resource_status_reason = resource_status_reason self.resource_properties = resource_properties - self.timestamp = datetime.utcnow() + self.timestamp = utcnow() self.event_id = mock_random.uuid4() self.client_request_token = None diff --git a/moto/cloudfront/models.py b/moto/cloudfront/models.py index 361dc7858..370575425 100644 --- a/moto/cloudfront/models.py +++ b/moto/cloudfront/models.py @@ -1,6 +1,5 @@ import string -from datetime import datetime from typing import Any, Dict, Iterable, List, Tuple, Optional from moto.core import BaseBackend, BackendDict, BaseModel from moto.core.utils import iso_8601_datetime_with_milliseconds @@ -252,7 +251,7 @@ class Invalidation(BaseModel): self, distribution: Distribution, paths: Dict[str, Any], caller_ref: str ): self.invalidation_id = Invalidation.random_id() - self.create_time = iso_8601_datetime_with_milliseconds(datetime.now()) + self.create_time = iso_8601_datetime_with_milliseconds() self.distribution = distribution self.status = "COMPLETED" diff --git a/moto/cloudtrail/models.py b/moto/cloudtrail/models.py index 93747e8ed..da0b855a2 100644 --- a/moto/cloudtrail/models.py +++ b/moto/cloudtrail/models.py @@ -4,7 +4,7 @@ import time from datetime import datetime from typing import Any, Dict, List, Optional, Iterable, Tuple from moto.core import BaseBackend, BackendDict, BaseModel -from moto.core.utils import iso_8601_datetime_without_milliseconds +from moto.core.utils import iso_8601_datetime_without_milliseconds, utcnow from moto.utilities.tagging_service import TaggingService from .exceptions import ( S3BucketDoesNotExistException, @@ -32,21 +32,19 @@ class TrailStatus: def start_logging(self) -> None: self.is_logging = True - self.started = datetime.utcnow() - self.latest_delivery_time = datetime2int(datetime.utcnow()) - self.latest_delivery_attempt = iso_8601_datetime_without_milliseconds( - datetime.utcnow() - ) + self.started = utcnow() + self.latest_delivery_time = datetime2int(utcnow()) + self.latest_delivery_attempt = iso_8601_datetime_without_milliseconds(utcnow()) def stop_logging(self) -> None: self.is_logging = False - self.stopped = datetime.utcnow() + self.stopped = utcnow() def description(self) -> Dict[str, Any]: if self.is_logging: - self.latest_delivery_time = datetime2int(datetime.utcnow()) + self.latest_delivery_time = datetime2int(utcnow()) self.latest_delivery_attempt = iso_8601_datetime_without_milliseconds( - datetime.utcnow() + utcnow() ) desc: Dict[str, Any] = { "IsLogging": self.is_logging, diff --git a/moto/cloudwatch/models.py b/moto/cloudwatch/models.py index e94513c43..d42755ee4 100644 --- a/moto/cloudwatch/models.py +++ b/moto/cloudwatch/models.py @@ -4,6 +4,7 @@ from moto.core import BaseBackend, BackendDict, BaseModel, CloudWatchMetricProvi from moto.core.utils import ( iso_8601_datetime_without_milliseconds, iso_8601_datetime_with_nanoseconds, + utcnow, ) from moto.moto_api._internal import mock_random from datetime import datetime, timedelta @@ -158,9 +159,7 @@ class FakeAlarm(BaseModel): self.ok_actions = ok_actions self.insufficient_data_actions = insufficient_data_actions self.unit = unit - self.configuration_updated_timestamp = iso_8601_datetime_with_nanoseconds( - datetime.now(tz=tzutc()) - ) + self.configuration_updated_timestamp = iso_8601_datetime_with_nanoseconds() self.treat_missing_data = treat_missing_data self.evaluate_low_sample_count_percentile = evaluate_low_sample_count_percentile self.threshold_metric_id = threshold_metric_id @@ -170,9 +169,7 @@ class FakeAlarm(BaseModel): self.state_reason = "Unchecked: Initial alarm creation" self.state_reason_data = "{}" self.state_value = "OK" - self.state_updated_timestamp = iso_8601_datetime_with_nanoseconds( - datetime.now(tz=tzutc()) - ) + self.state_updated_timestamp = iso_8601_datetime_with_nanoseconds() # only used for composite alarms self.rule = rule @@ -192,9 +189,7 @@ class FakeAlarm(BaseModel): self.state_reason = reason self.state_reason_data = reason_data self.state_value = state_value - self.state_updated_timestamp = iso_8601_datetime_with_nanoseconds( - datetime.now(tz=tzutc()) - ) + self.state_updated_timestamp = iso_8601_datetime_with_nanoseconds() def are_dimensions_same( @@ -227,7 +222,7 @@ class MetricDatumBase(BaseModel): ): self.namespace = namespace self.name = name - self.timestamp = timestamp or datetime.utcnow().replace(tzinfo=tzutc()) + self.timestamp = timestamp or utcnow().replace(tzinfo=tzutc()) self.dimensions = [ Dimension(dimension["Name"], dimension["Value"]) for dimension in dimensions ] @@ -335,9 +330,7 @@ class Statistics: """ def __init__(self, stats: List[str], dt: datetime, unit: Optional[str] = None): - self.timestamp: str = ( - iso_8601_datetime_without_milliseconds(dt) or self.timestamp_iso_8601_now() - ) + self.timestamp: str = iso_8601_datetime_without_milliseconds(dt or utcnow()) self.metric_data: List[MetricDatumBase] = [] self.stats = stats self.unit = unit @@ -437,9 +430,6 @@ class Statistics: [s.sum for s in self.metric_aggregated_list] ) - def timestamp_iso_8601_now(self) -> str: - return iso_8601_datetime_without_milliseconds(datetime.now()) # type: ignore[return-value] - class CloudWatchBackend(BaseBackend): def __init__(self, region_name: str, account_id: str): diff --git a/moto/codebuild/models.py b/moto/codebuild/models.py index d25b61875..b4991372f 100644 --- a/moto/codebuild/models.py +++ b/moto/codebuild/models.py @@ -18,7 +18,7 @@ class CodeBuildProjectMetadata(BaseModel): build_id: str, service_role: str, ): - current_date = iso_8601_datetime_with_milliseconds(datetime.datetime.utcnow()) + current_date = iso_8601_datetime_with_milliseconds() self.build_metadata: Dict[str, Any] = dict() self.build_metadata["id"] = build_id @@ -99,7 +99,7 @@ class CodeBuild(BaseModel): environment: Dict[str, Any], serviceRole: str = "some_role", ): - current_date = iso_8601_datetime_with_milliseconds(datetime.datetime.utcnow()) + current_date = iso_8601_datetime_with_milliseconds() self.project_metadata: Dict[str, Any] = dict() self.project_metadata["name"] = project_name @@ -201,7 +201,7 @@ class CodeBuildBackend(BaseBackend): return self.build_metadata[project_name].build_metadata def _set_phases(self, phases: List[Dict[str, Any]]) -> List[Dict[str, Any]]: - current_date = iso_8601_datetime_with_milliseconds(datetime.datetime.utcnow()) + current_date = iso_8601_datetime_with_milliseconds() # No phaseStatus for QUEUED on first start for existing_phase in phases: if existing_phase["phaseType"] == "QUEUED": diff --git a/moto/codecommit/models.py b/moto/codecommit/models.py index e3d74a2aa..b3e027262 100644 --- a/moto/codecommit/models.py +++ b/moto/codecommit/models.py @@ -1,7 +1,6 @@ from moto.core import BaseBackend, BackendDict, BaseModel from moto.core.utils import iso_8601_datetime_with_milliseconds from moto.moto_api._internal import mock_random -from datetime import datetime from typing import Dict, List, Optional from .exceptions import RepositoryDoesNotExistException, RepositoryNameExistsException @@ -14,7 +13,7 @@ class CodeCommit(BaseModel): repository_description: str, repository_name: str, ): - current_date = iso_8601_datetime_with_milliseconds(datetime.utcnow()) + current_date = iso_8601_datetime_with_milliseconds() self.repository_metadata = dict() self.repository_metadata["repositoryName"] = repository_name self.repository_metadata[ diff --git a/moto/codepipeline/models.py b/moto/codepipeline/models.py index 6fe20562d..354aba641 100644 --- a/moto/codepipeline/models.py +++ b/moto/codepipeline/models.py @@ -1,7 +1,6 @@ import json -from datetime import datetime from typing import Any, Dict, List, Tuple -from moto.core.utils import iso_8601_datetime_with_milliseconds +from moto.core.utils import iso_8601_datetime_with_milliseconds, utcnow from moto.iam.exceptions import IAMNotFoundException from moto.iam.models import iam_backends, IAMBackend @@ -24,8 +23,8 @@ class CodePipeline(BaseModel): self.tags: Dict[str, str] = {} self._arn = f"arn:aws:codepipeline:{region}:{account_id}:{pipeline['name']}" - self._created = datetime.utcnow() - self._updated = datetime.utcnow() + self._created = utcnow() + self._updated = utcnow() @property def metadata(self) -> Dict[str, str]: @@ -143,7 +142,7 @@ class CodePipelineBackend(BaseBackend): # version number is auto incremented pipeline["version"] = codepipeline.pipeline["version"] + 1 - codepipeline._updated = datetime.utcnow() + codepipeline._updated = utcnow() codepipeline.pipeline = codepipeline.add_default_values(pipeline) return codepipeline.pipeline diff --git a/moto/cognitoidentity/models.py b/moto/cognitoidentity/models.py index f7c1920f7..d326b7a4a 100644 --- a/moto/cognitoidentity/models.py +++ b/moto/cognitoidentity/models.py @@ -5,7 +5,7 @@ import re from collections import OrderedDict from typing import Any, Dict, List, Optional from moto.core import BaseBackend, BackendDict, BaseModel -from moto.core.utils import iso_8601_datetime_with_milliseconds +from moto.core.utils import iso_8601_datetime_with_milliseconds, utcnow from .exceptions import InvalidNameException, ResourceNotFoundError from .utils import get_random_identity_id @@ -29,7 +29,7 @@ class CognitoIdentityPool(BaseModel): self.saml_provider_arns = kwargs.get("saml_provider_arns", []) self.identity_pool_id = get_random_identity_id(region) - self.creation_time = datetime.datetime.utcnow() + self.creation_time = utcnow() self.tags = kwargs.get("tags") or {} @@ -137,7 +137,7 @@ class CognitoIdentityBackend(BaseBackend): def get_credentials_for_identity(self, identity_id: str) -> str: duration = 90 - now = datetime.datetime.utcnow() + now = utcnow() expiration = now + datetime.timedelta(seconds=duration) expiration_str = str(iso_8601_datetime_with_milliseconds(expiration)) return json.dumps( diff --git a/moto/cognitoidp/models.py b/moto/cognitoidp/models.py index 0c8c05399..fcc943238 100644 --- a/moto/cognitoidp/models.py +++ b/moto/cognitoidp/models.py @@ -9,6 +9,7 @@ from jose import jws from collections import OrderedDict from typing import Any, Dict, List, Tuple, Optional, Set from moto.core import BaseBackend, BackendDict, BaseModel +from moto.core.utils import utcnow from moto.moto_api._internal import mock_random as random from .exceptions import ( AliasExistsException, @@ -415,8 +416,8 @@ class CognitoIdpUserPool(BaseModel): "EmailMessage" ) - self.creation_date = datetime.datetime.utcnow() - self.last_modified_date = datetime.datetime.utcnow() + self.creation_date = utcnow() + self.last_modified_date = utcnow() self.mfa_config = "OFF" self.sms_mfa_config: Optional[Dict[str, Any]] = None @@ -711,8 +712,8 @@ class CognitoIdpIdentityProvider(BaseModel): def __init__(self, name: str, extended_config: Optional[Dict[str, Any]]): self.name = name self.extended_config = extended_config or {} - self.creation_date = datetime.datetime.utcnow() - self.last_modified_date = datetime.datetime.utcnow() + self.creation_date = utcnow() + self.last_modified_date = utcnow() if "AttributeMapping" not in self.extended_config: self.extended_config["AttributeMapping"] = {"username": "sub"} @@ -799,8 +800,8 @@ class CognitoIdpUser(BaseModel): self.enabled = True self.attributes = attributes self.attribute_lookup = flatten_attrs(attributes) - self.create_date = datetime.datetime.utcnow() - self.last_modified_date = datetime.datetime.utcnow() + self.create_date = utcnow() + self.last_modified_date = utcnow() self.sms_mfa_enabled = False self.software_token_mfa_enabled = False self.token_verified = False diff --git a/moto/config/models.py b/moto/config/models.py index ae8be7427..83bbf4ed3 100644 --- a/moto/config/models.py +++ b/moto/config/models.py @@ -53,6 +53,7 @@ from moto.config.exceptions import ( from moto.core import BaseBackend, BackendDict, BaseModel from moto.core.common_models import ConfigQueryModel from moto.core.responses import AWSServiceSpec +from moto.core.utils import utcnow from moto.iam.config import role_config_query, policy_config_query from moto.moto_api._internal import mock_random as random from moto.s3.config import s3_config_query @@ -238,13 +239,13 @@ class ConfigRecorderStatus(ConfigEmptyDictable): def start(self) -> None: self.recording = True self.last_status = "PENDING" - self.last_start_time = datetime2int(datetime.utcnow()) - self.last_status_change_time = datetime2int(datetime.utcnow()) + self.last_start_time = datetime2int(utcnow()) + self.last_status_change_time = datetime2int(utcnow()) def stop(self) -> None: self.recording = False - self.last_stop_time = datetime2int(datetime.utcnow()) - self.last_status_change_time = datetime2int(datetime.utcnow()) + self.last_stop_time = datetime2int(utcnow()) + self.last_status_change_time = datetime2int(utcnow()) class ConfigDeliverySnapshotProperties(ConfigEmptyDictable): @@ -404,8 +405,8 @@ class ConfigAggregator(ConfigEmptyDictable): self.configuration_aggregator_arn = f"arn:aws:config:{region}:{account_id}:config-aggregator/config-aggregator-{random_string()}" self.account_aggregation_sources = account_sources self.organization_aggregation_source = org_source - self.creation_time = datetime2int(datetime.utcnow()) - self.last_updated_time = datetime2int(datetime.utcnow()) + self.creation_time = datetime2int(utcnow()) + self.last_updated_time = datetime2int(utcnow()) # Tags are listed in the list_tags_for_resource API call. self.tags = tags or {} @@ -442,7 +443,7 @@ class ConfigAggregationAuthorization(ConfigEmptyDictable): self.aggregation_authorization_arn = f"arn:aws:config:{current_region}:{account_id}:aggregation-authorization/{authorized_account_id}/{authorized_aws_region}" self.authorized_account_id = authorized_account_id self.authorized_aws_region = authorized_aws_region - self.creation_time = datetime2int(datetime.utcnow()) + self.creation_time = datetime2int(utcnow()) # Tags are listed in the list_tags_for_resource API call. self.tags = tags or {} @@ -468,7 +469,7 @@ class OrganizationConformancePack(ConfigEmptyDictable): self.delivery_s3_bucket = delivery_s3_bucket self.delivery_s3_key_prefix = delivery_s3_key_prefix self.excluded_accounts = excluded_accounts or [] - self.last_update_time = datetime2int(datetime.utcnow()) + self.last_update_time = datetime2int(utcnow()) self.organization_conformance_pack_arn = f"arn:aws:config:{region}:{account_id}:organization-conformance-pack/{self._unique_pack_name}" self.organization_conformance_pack_name = name @@ -485,7 +486,7 @@ class OrganizationConformancePack(ConfigEmptyDictable): self.delivery_s3_bucket = delivery_s3_bucket self.delivery_s3_key_prefix = delivery_s3_key_prefix self.excluded_accounts = excluded_accounts - self.last_update_time = datetime2int(datetime.utcnow()) + self.last_update_time = datetime2int(utcnow()) class Scope(ConfigEmptyDictable): @@ -839,7 +840,7 @@ class ConfigRule(ConfigEmptyDictable): "CreatedBy field" ) - self.last_updated_time = datetime2int(datetime.utcnow()) + self.last_updated_time = datetime2int(utcnow()) self.tags = tags def validate_managed_rule(self) -> None: @@ -1046,7 +1047,7 @@ class ConfigBackend(BaseBackend): aggregator.tags = tags aggregator.account_aggregation_sources = account_sources aggregator.organization_aggregation_source = org_source - aggregator.last_updated_time = datetime2int(datetime.utcnow()) + aggregator.last_updated_time = datetime2int(utcnow()) return aggregator.to_dict() @@ -1922,7 +1923,7 @@ class ConfigBackend(BaseBackend): "AccountId": self.account_id, "ConformancePackName": f"OrgConformsPack-{pack._unique_pack_name}", "Status": pack._status, - "LastUpdateTime": datetime2int(datetime.utcnow()), + "LastUpdateTime": datetime2int(utcnow()), } ] diff --git a/moto/core/utils.py b/moto/core/utils.py index 483ce1af7..efda9a392 100644 --- a/moto/core/utils.py +++ b/moto/core/utils.py @@ -7,7 +7,7 @@ from gzip import decompress from typing import Any, Optional, List, Callable, Dict, Tuple from urllib.parse import urlparse, unquote from .common_types import TYPE_RESPONSE -from .versions import is_werkzeug_2_3_x +from .versions import is_werkzeug_2_3_x, PYTHON_311 def camelcase_to_underscores(argument: str) -> str: @@ -148,19 +148,17 @@ class convert_flask_to_responses_response(object): def iso_8601_datetime_with_milliseconds( value: Optional[datetime.datetime] = None, ) -> str: - date_to_use = value or datetime.datetime.now() + date_to_use = value or utcnow() return date_to_use.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z" # Even Python does not support nanoseconds, other languages like Go do (needed for Terraform) -def iso_8601_datetime_with_nanoseconds(value: datetime.datetime) -> str: - return value.strftime("%Y-%m-%dT%H:%M:%S.%f000Z") +def iso_8601_datetime_with_nanoseconds() -> str: + return utcnow().strftime("%Y-%m-%dT%H:%M:%S.%f000Z") -def iso_8601_datetime_without_milliseconds( - value: Optional[datetime.datetime], -) -> Optional[str]: - return value.strftime("%Y-%m-%dT%H:%M:%SZ") if value else None +def iso_8601_datetime_without_milliseconds(value: datetime.datetime) -> str: + return value.strftime("%Y-%m-%dT%H:%M:%SZ") def iso_8601_datetime_without_milliseconds_s3( @@ -181,7 +179,7 @@ def str_to_rfc_1123_datetime(value: str) -> datetime.datetime: def unix_time(dt: Optional[datetime.datetime] = None) -> float: - dt = dt or datetime.datetime.utcnow() + dt = dt or utcnow() epoch = datetime.datetime.utcfromtimestamp(0) delta = dt - epoch return (delta.days * 86400) + (delta.seconds + (delta.microseconds / 1e6)) @@ -191,6 +189,21 @@ def unix_time_millis(dt: Optional[datetime.datetime] = None) -> float: return unix_time(dt) * 1000.0 +def utcnow() -> datetime.datetime: + # Python 3.12 starts throwing deprecation warnings for utcnow() + # The docs recommend to use now(UTC) instead + # + # now(UTC) creates an aware datetime - but utcnow() creates a naive datetime + # That's why we have to `replace(tzinfo=None)` to make now(UTC) naive. + if PYTHON_311: + # Only available in 3.11 + from datetime import UTC # type: ignore + + return datetime.datetime.now(UTC).replace(tzinfo=None) + else: + return datetime.datetime.utcnow() + + def path_url(url: str) -> str: parsed_url = urlparse(url) path = parsed_url.path diff --git a/moto/core/versions.py b/moto/core/versions.py index 16a1df52d..3c42a2b8c 100644 --- a/moto/core/versions.py +++ b/moto/core/versions.py @@ -1,3 +1,5 @@ +import sys + from moto.utilities.distutils_version import LooseVersion try: @@ -6,6 +8,8 @@ except ImportError: from importlib_metadata import version # type: ignore[no-redef] +PYTHON_VERSION_INFO = sys.version_info +PYTHON_311 = sys.version_info >= (3, 11) RESPONSES_VERSION = version("responses") WERKZEUG_VERSION = version("werkzeug") diff --git a/moto/datapipeline/models.py b/moto/datapipeline/models.py index ee03740f8..71727501c 100644 --- a/moto/datapipeline/models.py +++ b/moto/datapipeline/models.py @@ -2,6 +2,7 @@ import datetime from collections import OrderedDict from moto.core import BaseBackend, BackendDict, BaseModel, CloudFormationModel +from moto.core.utils import utcnow from .utils import get_random_pipeline_id, remove_capitalization_of_dict_keys from typing import Any, Dict, Iterable, List @@ -22,7 +23,7 @@ class Pipeline(CloudFormationModel): self.unique_id = unique_id self.description = kwargs.get("description", "") self.pipeline_id = get_random_pipeline_id() - self.creation_time = datetime.datetime.utcnow() + self.creation_time = utcnow() self.objects: List[Any] = [] self.status = "PENDING" self.tags = kwargs.get("tags", []) diff --git a/moto/dms/models.py b/moto/dms/models.py index 87fbf9017..ad8719a4f 100644 --- a/moto/dms/models.py +++ b/moto/dms/models.py @@ -1,6 +1,7 @@ from datetime import datetime from typing import Any, Dict, List, Iterable, Optional from moto.core import BaseBackend, BackendDict, BaseModel +from moto.core.utils import utcnow from .exceptions import ( InvalidResourceStateFault, @@ -126,7 +127,7 @@ class FakeReplicationTask(BaseModel): self.arn = f"arn:aws:dms:{region_name}:{account_id}:task:{self.id}" self.status = "creating" - self.creation_date = datetime.utcnow() + self.creation_date = utcnow() self.start_date: Optional[datetime] = None self.stop_date: Optional[datetime] = None @@ -167,7 +168,7 @@ class FakeReplicationTask(BaseModel): def start(self) -> "FakeReplicationTask": self.status = "starting" - self.start_date = datetime.utcnow() + self.start_date = utcnow() self.run() return self @@ -176,7 +177,7 @@ class FakeReplicationTask(BaseModel): raise InvalidResourceStateFault("Replication task is not running") self.status = "stopped" - self.stop_date = datetime.utcnow() + self.stop_date = utcnow() return self def delete(self) -> "FakeReplicationTask": diff --git a/moto/dynamodb/models/table.py b/moto/dynamodb/models/table.py index 3e56009bc..0e211fd35 100644 --- a/moto/dynamodb/models/table.py +++ b/moto/dynamodb/models/table.py @@ -1,10 +1,9 @@ from collections import defaultdict import copy -import datetime from typing import Any, Dict, Optional, List, Tuple, Iterator, Sequence from moto.core import BaseModel, CloudFormationModel -from moto.core.utils import unix_time, unix_time_millis +from moto.core.utils import unix_time, unix_time_millis, utcnow from moto.dynamodb.comparisons import get_filter_expression, get_expected from moto.dynamodb.exceptions import ( InvalidIndexNameError, @@ -153,7 +152,7 @@ class StreamRecord(BaseModel): "awsRegion": "us-east-1", "dynamodb": { "StreamViewType": stream_type, - "ApproximateCreationDateTime": datetime.datetime.utcnow().isoformat(), + "ApproximateCreationDateTime": utcnow().isoformat(), "SequenceNumber": str(seq), "SizeBytes": 1, "Keys": keys, @@ -181,7 +180,7 @@ class StreamShard(BaseModel): self.id = "shardId-00000001541626099285-f35f62ef" self.starting_sequence_number = 1100000000017454423009 self.items: List[StreamRecord] = [] - self.created_on = datetime.datetime.utcnow() + self.created_on = utcnow() def to_json(self) -> Dict[str, Any]: return { @@ -277,7 +276,7 @@ class Table(CloudFormationModel): GlobalSecondaryIndex.create(i, self.table_key_attrs) for i in (global_indexes if global_indexes else []) ] - self.created_at = datetime.datetime.utcnow() + self.created_at = utcnow() self.items = defaultdict(dict) # type: ignore # [hash: DynamoType] or [hash: [range: DynamoType]] self.table_arn = self._generate_arn(table_name) self.tags = tags or [] @@ -416,7 +415,7 @@ class Table(CloudFormationModel): and (streams.get("StreamEnabled") or streams.get("StreamViewType")) ): self.stream_specification["StreamEnabled"] = True - self.latest_stream_label = datetime.datetime.utcnow().isoformat() + self.latest_stream_label = utcnow().isoformat() self.stream_shard = StreamShard(self.account_id, self) else: self.stream_specification = {"StreamEnabled": False} @@ -928,7 +927,7 @@ class Backup: self.table = copy.deepcopy(table) self.status = status or "AVAILABLE" self.type = type_ or "USER" - self.creation_date_time = datetime.datetime.utcnow() + self.creation_date_time = utcnow() self.identifier = self._make_identifier() def _make_identifier(self) -> str: diff --git a/moto/dynamodb_v20111205/models.py b/moto/dynamodb_v20111205/models.py index c916cb3a6..89f5b294c 100644 --- a/moto/dynamodb_v20111205/models.py +++ b/moto/dynamodb_v20111205/models.py @@ -1,11 +1,10 @@ from collections import defaultdict from typing import Any, Dict, Optional, List, Union, Tuple, Iterable -import datetime import json from collections import OrderedDict from moto.core import BaseBackend, BackendDict, BaseModel -from moto.core.utils import unix_time +from moto.core.utils import unix_time, utcnow from .comparisons import get_comparison_func @@ -114,7 +113,7 @@ class Table(BaseModel): self.range_key_type = range_key_type self.read_capacity = read_capacity self.write_capacity = write_capacity - self.created_at = datetime.datetime.utcnow() + self.created_at = utcnow() self.items: Dict[DynamoType, Union[Item, Dict[DynamoType, Item]]] = defaultdict( dict ) diff --git a/moto/ec2/models/instances.py b/moto/ec2/models/instances.py index 8463ed9c3..530d8097d 100644 --- a/moto/ec2/models/instances.py +++ b/moto/ec2/models/instances.py @@ -2,12 +2,11 @@ import contextlib import copy import warnings from collections import OrderedDict -from datetime import datetime from typing import Any, Dict, ItemsView, List, Tuple, Optional, Set from moto import settings from moto.core import CloudFormationModel -from moto.core.utils import camelcase_to_underscores +from moto.core.utils import camelcase_to_underscores, utcnow from moto.ec2.models.fleets import Fleet from moto.ec2.models.elastic_network_interfaces import NetworkInterface from moto.ec2.models.launch_templates import LaunchTemplateVersion @@ -381,9 +380,7 @@ class Instance(TaggedEC2Resource, BotoInstance, CloudFormationModel): self._state.name = "stopped" self._state.code = 80 - self._reason = ( - f"User initiated ({datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S UTC')})" - ) + self._reason = f"User initiated ({utcnow().strftime('%Y-%m-%d %H:%M:%S UTC')})" self._state_reason = StateReason( "Client.UserInitiatedShutdown: User initiated shutdown", "Client.UserInitiatedShutdown", @@ -433,9 +430,7 @@ class Instance(TaggedEC2Resource, BotoInstance, CloudFormationModel): self._state.name = "terminated" self._state.code = 48 - self._reason = ( - f"User initiated ({datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S UTC')})" - ) + self._reason = f"User initiated ({utcnow().strftime('%Y-%m-%d %H:%M:%S UTC')})" self._state_reason = StateReason( "Client.UserInitiatedShutdown: User initiated shutdown", "Client.UserInitiatedShutdown", diff --git a/moto/ec2/models/key_pairs.py b/moto/ec2/models/key_pairs.py index b480b037d..7d29018ff 100644 --- a/moto/ec2/models/key_pairs.py +++ b/moto/ec2/models/key_pairs.py @@ -1,5 +1,4 @@ from typing import Any, Dict, List -from datetime import datetime from moto.core import BaseModel from ..exceptions import ( @@ -15,7 +14,7 @@ from ..utils import ( generic_filter, random_key_pair_id, ) -from moto.core.utils import iso_8601_datetime_with_milliseconds +from moto.core.utils import iso_8601_datetime_with_milliseconds, utcnow class KeyPair(BaseModel): @@ -24,7 +23,7 @@ class KeyPair(BaseModel): self.name = name self.fingerprint = fingerprint self.material = material - self.create_time = datetime.utcnow() + self.create_time = utcnow() @property def created_iso_8601(self) -> str: diff --git a/moto/ec2/models/nat_gateways.py b/moto/ec2/models/nat_gateways.py index 9cf97d309..7d6fb80d0 100644 --- a/moto/ec2/models/nat_gateways.py +++ b/moto/ec2/models/nat_gateways.py @@ -1,8 +1,7 @@ -from datetime import datetime from typing import Any, Dict, List, Optional from moto.core import CloudFormationModel -from moto.core.utils import iso_8601_datetime_with_milliseconds +from moto.core.utils import iso_8601_datetime_with_milliseconds, utcnow from .core import TaggedEC2Resource from ..utils import random_nat_gateway_id, random_private_ip @@ -25,7 +24,7 @@ class NatGateway(CloudFormationModel, TaggedEC2Resource): self.connectivity_type = connectivity_type # protected properties - self._created_at = datetime.utcnow() + self._created_at = utcnow() self.ec2_backend = backend # NOTE: this is the core of NAT Gateways creation self._eni = self.ec2_backend.create_network_interface( diff --git a/moto/ec2/models/transit_gateway.py b/moto/ec2/models/transit_gateway.py index eef85c7a5..f2a942134 100644 --- a/moto/ec2/models/transit_gateway.py +++ b/moto/ec2/models/transit_gateway.py @@ -1,7 +1,6 @@ -from datetime import datetime from typing import Any, Dict, List, Optional from moto.core import CloudFormationModel -from moto.core.utils import iso_8601_datetime_with_milliseconds +from moto.core.utils import iso_8601_datetime_with_milliseconds, utcnow from moto.utilities.utils import filter_resources, merge_multiple_dicts from .core import TaggedEC2Resource from ..utils import ( @@ -35,7 +34,7 @@ class TransitGateway(TaggedEC2Resource, CloudFormationModel): self.description = description self.state = "available" self.options = merge_multiple_dicts(self.DEFAULT_OPTIONS, options or {}) - self._created_at = datetime.utcnow() + self._created_at = utcnow() @property def physical_resource_id(self) -> str: diff --git a/moto/ec2/models/transit_gateway_attachments.py b/moto/ec2/models/transit_gateway_attachments.py index 60c3b1317..2ed80b05d 100644 --- a/moto/ec2/models/transit_gateway_attachments.py +++ b/moto/ec2/models/transit_gateway_attachments.py @@ -1,6 +1,5 @@ -from datetime import datetime from typing import Any, Dict, List, Optional -from moto.core.utils import iso_8601_datetime_with_milliseconds +from moto.core.utils import iso_8601_datetime_with_milliseconds, utcnow from moto.utilities.utils import merge_multiple_dicts, filter_resources from .core import TaggedEC2Resource from .vpc_peering_connections import PeeringConnectionStatus @@ -28,7 +27,7 @@ class TransitGatewayAttachment(TaggedEC2Resource): self.state = "available" self.add_tags(tags or {}) - self._created_at = datetime.utcnow() + self._created_at = utcnow() self.resource_owner_id = backend.account_id self.transit_gateway_owner_id = backend.account_id self.owner_id = backend.account_id diff --git a/moto/ec2/models/transit_gateway_route_tables.py b/moto/ec2/models/transit_gateway_route_tables.py index 824200311..603969c9a 100644 --- a/moto/ec2/models/transit_gateway_route_tables.py +++ b/moto/ec2/models/transit_gateway_route_tables.py @@ -1,6 +1,5 @@ -from datetime import datetime from typing import Any, Dict, List, Optional -from moto.core.utils import iso_8601_datetime_with_milliseconds +from moto.core.utils import iso_8601_datetime_with_milliseconds, utcnow from moto.utilities.utils import filter_resources from .core import TaggedEC2Resource from ..utils import random_transit_gateway_route_table_id @@ -19,7 +18,7 @@ class TransitGatewayRouteTable(TaggedEC2Resource): self.id = random_transit_gateway_route_table_id() self.transit_gateway_id = transit_gateway_id - self._created_at = datetime.utcnow() + self._created_at = utcnow() self.default_association_route_table = default_association_route_table self.default_propagation_route_table = default_propagation_route_table diff --git a/moto/ec2/utils.py b/moto/ec2/utils.py index 3e0e609bc..c768f026c 100644 --- a/moto/ec2/utils.py +++ b/moto/ec2/utils.py @@ -3,12 +3,12 @@ import fnmatch import re import ipaddress -from datetime import datetime from cryptography.hazmat.primitives import serialization from cryptography.hazmat.backends import default_backend from cryptography.hazmat.primitives.asymmetric import rsa from typing import Any, Dict, List, Set, TypeVar, Tuple, Optional, Union +from moto.core.utils import utcnow from moto.iam import iam_backends from moto.moto_api._internal import mock_random as random from moto.utilities.utils import md5_hash @@ -305,7 +305,7 @@ def create_dns_entries(service_name: str, vpc_endpoint_id: str) -> Dict[str, str def utc_date_and_time() -> str: - x = datetime.utcnow() + x = utcnow() # Better performing alternative to x.strftime("%Y-%m-%dT%H:%M:%S.000Z") return f"{x.year}-{x.month:02d}-{x.day:02d}T{x.hour:02d}:{x.minute:02d}:{x.second:02d}.000Z" diff --git a/moto/ecr/models.py b/moto/ecr/models.py index f2d2961bf..7ac4db9e0 100644 --- a/moto/ecr/models.py +++ b/moto/ecr/models.py @@ -8,7 +8,7 @@ from typing import Any, Dict, List, Iterable, Optional from botocore.exceptions import ParamValidationError from moto.core import BaseBackend, BackendDict, BaseModel, CloudFormationModel -from moto.core.utils import iso_8601_datetime_without_milliseconds +from moto.core.utils import iso_8601_datetime_without_milliseconds, utcnow from moto.ecr.exceptions import ( ImageNotFoundException, RepositoryNotFoundException, @@ -81,7 +81,7 @@ class Repository(BaseObject, CloudFormationModel): f"arn:aws:ecr:{region_name}:{self.registry_id}:repository/{repository_name}" ) self.name = repository_name - self.created_at = datetime.utcnow() + self.created_at = utcnow() self.uri = ( f"{self.registry_id}.dkr.ecr.{region_name}.amazonaws.com/{repository_name}" ) @@ -920,9 +920,7 @@ class ECRBackend(BaseBackend): "registryId": repo.registry_id, "repositoryName": repository_name, "lifecyclePolicyText": repo.lifecycle_policy, - "lastEvaluatedAt": iso_8601_datetime_without_milliseconds( - datetime.utcnow() - ), + "lastEvaluatedAt": iso_8601_datetime_without_milliseconds(utcnow()), } def delete_lifecycle_policy( @@ -940,9 +938,7 @@ class ECRBackend(BaseBackend): "registryId": repo.registry_id, "repositoryName": repository_name, "lifecyclePolicyText": policy, - "lastEvaluatedAt": iso_8601_datetime_without_milliseconds( - datetime.utcnow() - ), + "lastEvaluatedAt": iso_8601_datetime_without_milliseconds(utcnow()), } def _validate_registry_policy_action(self, policy_text: str) -> None: @@ -1053,7 +1049,7 @@ class ECRBackend(BaseBackend): image.last_scan ), "vulnerabilitySourceUpdatedAt": iso_8601_datetime_without_milliseconds( - datetime.utcnow() + utcnow() ), "findings": [ { diff --git a/moto/elasticache/models.py b/moto/elasticache/models.py index 25e37807c..4772463de 100644 --- a/moto/elasticache/models.py +++ b/moto/elasticache/models.py @@ -1,7 +1,7 @@ -from datetime import datetime from typing import List, Optional, Dict, Any, Tuple from moto.core import BaseBackend, BackendDict, BaseModel +from moto.core.utils import utcnow from .exceptions import ( UserAlreadyExists, @@ -117,8 +117,8 @@ class CacheCluster(BaseModel): self.cache_node_ids_to_remove = cache_node_ids_to_remove self.cache_node_ids_to_reboot = cache_node_ids_to_reboot - self.cache_cluster_create_time = datetime.utcnow() - self.auth_token_last_modified_date = datetime.utcnow() + self.cache_cluster_create_time = utcnow() + self.auth_token_last_modified_date = utcnow() self.cache_cluster_status = "available" self.arn = f"arn:aws:elasticache:{region_name}:{account_id}:{cache_cluster_id}" self.cache_node_id = str(mock_random.uuid4()) diff --git a/moto/elbv2/models.py b/moto/elbv2/models.py index 4d9526a23..a54f7726f 100644 --- a/moto/elbv2/models.py +++ b/moto/elbv2/models.py @@ -1,4 +1,3 @@ -import datetime import re from jinja2 import Template from botocore.exceptions import ParamValidationError @@ -587,7 +586,7 @@ class FakeLoadBalancer(CloudFormationModel): loadbalancer_type: Optional[str] = None, ): self.name = name - self.created_time = iso_8601_datetime_with_milliseconds(datetime.datetime.now()) + self.created_time = iso_8601_datetime_with_milliseconds() self.scheme = scheme self.security_groups = security_groups self.subnets = subnets or [] diff --git a/moto/emr/utils.py b/moto/emr/utils.py index 522cc041c..3423db462 100644 --- a/moto/emr/utils.py +++ b/moto/emr/utils.py @@ -1,5 +1,4 @@ import copy -import datetime import re import string from typing import Any, List, Dict, Tuple, Iterator @@ -242,7 +241,7 @@ class EmrManagedSecurityGroup: @classmethod def description(cls) -> str: - created = iso_8601_datetime_with_milliseconds(datetime.datetime.now()) + created = iso_8601_datetime_with_milliseconds() return cls.desc_fmt.format(short_name=cls.short_name, created=created) diff --git a/moto/events/models.py b/moto/events/models.py index 0f647f827..5ec5a318c 100644 --- a/moto/events/models.py +++ b/moto/events/models.py @@ -552,7 +552,7 @@ class Archive(CloudFormationModel): self.retention = retention if retention else 0 self.arn = f"arn:aws:events:{region_name}:{account_id}:archive/{name}" - self.creation_time = unix_time(datetime.utcnow()) + self.creation_time = unix_time() self.state = "ENABLED" self.uuid = str(random.uuid4()) @@ -701,7 +701,7 @@ class Replay(BaseModel): self.arn = f"arn:aws:events:{region_name}:{account_id}:replay/{name}" self.state = ReplayState.STARTING - self.start_time = unix_time(datetime.utcnow()) + self.start_time = unix_time() self.end_time: Optional[float] = None def describe_short(self) -> Dict[str, Any]: @@ -740,7 +740,7 @@ class Replay(BaseModel): ) self.state = ReplayState.COMPLETED - self.end_time = unix_time(datetime.utcnow()) + self.end_time = unix_time() class Connection(BaseModel): @@ -759,7 +759,7 @@ class Connection(BaseModel): self.description = description self.authorization_type = authorization_type self.auth_parameters = auth_parameters - self.creation_time = unix_time(datetime.utcnow()) + self.creation_time = unix_time() self.state = "AUTHORIZED" self.arn = f"arn:aws:events:{region_name}:{account_id}:connection/{self.name}/{self.uuid}" @@ -836,7 +836,7 @@ class Destination(BaseModel): self.connection_arn = connection_arn self.invocation_endpoint = invocation_endpoint self.invocation_rate_limit_per_second = invocation_rate_limit_per_second - self.creation_time = unix_time(datetime.utcnow()) + self.creation_time = unix_time() self.http_method = http_method self.state = "ACTIVE" self.arn = f"arn:aws:events:{region_name}:{account_id}:api-destination/{name}/{self.uuid}" @@ -1354,7 +1354,7 @@ class EventsBackend(BaseBackend): "detail-type": event["DetailType"], "source": event["Source"], "account": self.account_id, - "time": event.get("Time", unix_time(datetime.utcnow())), + "time": event.get("Time", unix_time()), "region": self.region_name, "resources": event.get("Resources", []), "detail": json.loads(event["Detail"]), diff --git a/moto/firehose/models.py b/moto/firehose/models.py index f47af5774..f253db770 100644 --- a/moto/firehose/models.py +++ b/moto/firehose/models.py @@ -23,6 +23,7 @@ import warnings import requests from moto.core import BaseBackend, BackendDict, BaseModel +from moto.core.utils import utcnow from moto.firehose.exceptions import ( ConcurrentModificationException, InvalidArgumentException, @@ -442,7 +443,7 @@ class FirehoseBackend(BaseBackend): # Object name pattern: # DeliveryStreamName-DeliveryStreamVersion-YYYY-MM-DD-HH-MM-SS-RandomString prefix = f"{prefix}{'' if prefix.endswith('/') else '/'}" - now = datetime.utcnow() + now = utcnow() return ( f"{prefix}{now.strftime('%Y/%m/%d/%H')}/" f"{delivery_stream_name}-{version_id}-" diff --git a/moto/glue/models.py b/moto/glue/models.py index a9742b1e2..838b70d6d 100644 --- a/moto/glue/models.py +++ b/moto/glue/models.py @@ -8,7 +8,7 @@ from typing import Any, Dict, List, Optional from moto.core import BaseBackend, BackendDict, BaseModel from moto.moto_api import state_manager from moto.moto_api._internal import mock_random -from moto.core.utils import unix_time +from moto.core.utils import unix_time, utcnow from moto.moto_api._internal.managed_state_model import ManagedState from .exceptions import ( JsonRESTError, @@ -1044,7 +1044,7 @@ class FakeDatabase(BaseModel): def __init__(self, database_name: str, database_input: Dict[str, Any]): self.name = database_name self.input = database_input - self.created_time = datetime.utcnow() + self.created_time = utcnow() self.tables: Dict[str, FakeTable] = OrderedDict() def as_dict(self) -> Dict[str, Any]: @@ -1069,7 +1069,7 @@ class FakeTable(BaseModel): self.database_name = database_name self.name = table_name self.partitions: Dict[str, FakePartition] = OrderedDict() - self.created_time = datetime.utcnow() + self.created_time = utcnow() self.updated_time: Optional[datetime] = None self._current_version = 1 self.versions: Dict[str, Dict[str, Any]] = { @@ -1079,7 +1079,7 @@ class FakeTable(BaseModel): def update(self, table_input: Dict[str, Any]) -> None: self.versions[str(self._current_version + 1)] = table_input self._current_version += 1 - self.updated_time = datetime.utcnow() + self.updated_time = utcnow() def get_version(self, ver: str) -> Dict[str, Any]: try: @@ -1202,7 +1202,7 @@ class FakeCrawler(BaseModel): self.configuration = configuration self.crawler_security_configuration = crawler_security_configuration self.state = "READY" - self.creation_time = datetime.utcnow() + self.creation_time = utcnow() self.last_updated = self.creation_time self.version = 1 self.crawl_elapsed_time = 0 @@ -1339,8 +1339,8 @@ class FakeJob: self.code_gen_configuration_nodes = code_gen_configuration_nodes self.execution_class = execution_class or "STANDARD" self.source_control_details = source_control_details - self.created_on = datetime.utcnow() - self.last_modified_on = datetime.utcnow() + self.created_on = utcnow() + self.last_modified_on = utcnow() self.arn = ( f"arn:aws:glue:{backend.region_name}:{backend.account_id}:job/{self.name}" ) @@ -1420,9 +1420,9 @@ class FakeJobRun(ManagedState): self.allocated_capacity = allocated_capacity self.timeout = timeout self.worker_type = worker_type - self.started_on = datetime.utcnow() - self.modified_on = datetime.utcnow() - self.completed_on = datetime.utcnow() + self.started_on = utcnow() + self.modified_on = utcnow() + self.completed_on = utcnow() def get_name(self) -> str: return self.job_name @@ -1467,8 +1467,8 @@ class FakeRegistry(BaseModel): self.name = registry_name self.description = description self.tags = tags - self.created_time = datetime.utcnow() - self.updated_time = datetime.utcnow() + self.created_time = utcnow() + self.updated_time = utcnow() self.status = "AVAILABLE" self.registry_arn = f"arn:aws:glue:{backend.region_name}:{backend.account_id}:registry/{self.name}" self.schemas: Dict[str, FakeSchema] = OrderedDict() @@ -1506,8 +1506,8 @@ class FakeSchema(BaseModel): self.schema_status = AVAILABLE_STATUS self.schema_version_id = schema_version_id self.schema_version_status = AVAILABLE_STATUS - self.created_time = datetime.utcnow() - self.updated_time = datetime.utcnow() + self.created_time = utcnow() + self.updated_time = utcnow() self.schema_versions: Dict[str, FakeSchemaVersion] = OrderedDict() def update_next_schema_version(self) -> None: @@ -1553,8 +1553,8 @@ class FakeSchemaVersion(BaseModel): self.schema_version_status = AVAILABLE_STATUS self.version_number = version_number self.schema_version_id = str(mock_random.uuid4()) - self.created_time = datetime.utcnow() - self.updated_time = datetime.utcnow() + self.created_time = utcnow() + self.updated_time = utcnow() self.metadata: Dict[str, Any] = {} def get_schema_version_id(self) -> str: @@ -1623,7 +1623,7 @@ class FakeSession(BaseModel): self.glue_version = glue_version self.tags = tags self.request_origin = request_origin - self.creation_time = datetime.utcnow() + self.creation_time = utcnow() self.last_updated = self.creation_time self.arn = f"arn:aws:glue:{backend.region_name}:{backend.account_id}:session/{self.session_id}" self.backend = backend diff --git a/moto/greengrass/models.py b/moto/greengrass/models.py index 3eefdfa7b..493058e15 100644 --- a/moto/greengrass/models.py +++ b/moto/greengrass/models.py @@ -5,7 +5,7 @@ from typing import Any, Dict, List, Iterable, Optional import re from moto.core import BaseBackend, BackendDict, BaseModel -from moto.core.utils import iso_8601_datetime_with_milliseconds +from moto.core.utils import iso_8601_datetime_with_milliseconds, utcnow from moto.moto_api._internal import mock_random from .exceptions import ( GreengrassClientError, @@ -24,7 +24,7 @@ class FakeCoreDefinition(BaseModel): self.name = name self.id = str(mock_random.uuid4()) self.arn = f"arn:aws:greengrass:{region_name}:{account_id}:greengrass/definition/cores/{self.id}" - self.created_at_datetime = datetime.utcnow() + self.created_at_datetime = utcnow() self.latest_version = "" self.latest_version_arn = "" @@ -57,7 +57,7 @@ class FakeCoreDefinitionVersion(BaseModel): self.definition = definition self.version = str(mock_random.uuid4()) self.arn = f"arn:aws:greengrass:{region_name}:{account_id}:greengrass/definition/cores/{self.core_definition_id}/versions/{self.version}" - self.created_at_datetime = datetime.utcnow() + self.created_at_datetime = utcnow() def to_dict(self, include_detail: bool = False) -> Dict[str, Any]: obj: Dict[str, Any] = { @@ -86,8 +86,8 @@ class FakeDeviceDefinition(BaseModel): self.region_name = region_name self.id = str(mock_random.uuid4()) self.arn = f"arn:aws:greengrass:{region_name}:{account_id}:greengrass/definition/devices/{self.id}" - self.created_at_datetime = datetime.utcnow() - self.update_at_datetime = datetime.utcnow() + self.created_at_datetime = utcnow() + self.update_at_datetime = utcnow() self.latest_version = "" self.latest_version_arn = "" self.name = name @@ -124,7 +124,7 @@ class FakeDeviceDefinitionVersion(BaseModel): self.devices = devices self.version = str(mock_random.uuid4()) self.arn = f"arn:aws:greengrass:{region_name}:{account_id}:greengrass/definition/devices/{self.device_definition_id}/versions/{self.version}" - self.created_at_datetime = datetime.utcnow() + self.created_at_datetime = utcnow() def to_dict(self, include_detail: bool = False) -> Dict[str, Any]: obj: Dict[str, Any] = { @@ -153,8 +153,8 @@ class FakeResourceDefinition(BaseModel): self.region_name = region_name self.id = str(mock_random.uuid4()) self.arn = f"arn:aws:greengrass:{region_name}:{account_id}:greengrass/definition/resources/{self.id}" - self.created_at_datetime = datetime.utcnow() - self.update_at_datetime = datetime.utcnow() + self.created_at_datetime = utcnow() + self.update_at_datetime = utcnow() self.latest_version = "" self.latest_version_arn = "" self.name = name @@ -189,7 +189,7 @@ class FakeResourceDefinitionVersion(BaseModel): self.resources = resources self.version = str(mock_random.uuid4()) self.arn = f"arn:aws:greengrass:{region_name}:{account_id}:greengrass/definition/resources/{self.resource_definition_id}/versions/{self.version}" - self.created_at_datetime = datetime.utcnow() + self.created_at_datetime = utcnow() def to_dict(self) -> Dict[str, Any]: return { @@ -214,8 +214,8 @@ class FakeFunctionDefinition(BaseModel): self.region_name = region_name self.id = str(mock_random.uuid4()) self.arn = f"arn:aws:greengrass:{self.region_name}:{account_id}:greengrass/definition/functions/{self.id}" - self.created_at_datetime = datetime.utcnow() - self.update_at_datetime = datetime.utcnow() + self.created_at_datetime = utcnow() + self.update_at_datetime = utcnow() self.latest_version = "" self.latest_version_arn = "" self.name = name @@ -254,7 +254,7 @@ class FakeFunctionDefinitionVersion(BaseModel): self.default_config = default_config self.version = str(mock_random.uuid4()) self.arn = f"arn:aws:greengrass:{self.region_name}:{account_id}:greengrass/definition/functions/{self.function_definition_id}/versions/{self.version}" - self.created_at_datetime = datetime.utcnow() + self.created_at_datetime = utcnow() def to_dict(self) -> Dict[str, Any]: return { @@ -279,8 +279,8 @@ class FakeSubscriptionDefinition(BaseModel): self.region_name = region_name self.id = str(mock_random.uuid4()) self.arn = f"arn:aws:greengrass:{self.region_name}:{account_id}:greengrass/definition/subscriptions/{self.id}" - self.created_at_datetime = datetime.utcnow() - self.update_at_datetime = datetime.utcnow() + self.created_at_datetime = utcnow() + self.update_at_datetime = utcnow() self.latest_version = "" self.latest_version_arn = "" self.name = name @@ -315,7 +315,7 @@ class FakeSubscriptionDefinitionVersion(BaseModel): self.subscriptions = subscriptions self.version = str(mock_random.uuid4()) self.arn = f"arn:aws:greengrass:{self.region_name}:{account_id}:greengrass/definition/subscriptions/{self.subscription_definition_id}/versions/{self.version}" - self.created_at_datetime = datetime.utcnow() + self.created_at_datetime = utcnow() def to_dict(self) -> Dict[str, Any]: return { @@ -335,8 +335,8 @@ class FakeGroup(BaseModel): self.group_id = str(mock_random.uuid4()) self.name = name self.arn = f"arn:aws:greengrass:{self.region_name}:{account_id}:greengrass/groups/{self.group_id}" - self.created_at_datetime = datetime.utcnow() - self.last_updated_datetime = datetime.utcnow() + self.created_at_datetime = utcnow() + self.last_updated_datetime = utcnow() self.latest_version = "" self.latest_version_arn = "" @@ -373,7 +373,7 @@ class FakeGroupVersion(BaseModel): self.group_id = group_id self.version = str(mock_random.uuid4()) self.arn = f"arn:aws:greengrass:{self.region_name}:{account_id}:greengrass/groups/{self.group_id}/versions/{self.version}" - self.created_at_datetime = datetime.utcnow() + self.created_at_datetime = utcnow() self.core_definition_version_arn = core_definition_version_arn self.device_definition_version_arn = device_definition_version_arn self.function_definition_version_arn = function_definition_version_arn @@ -434,8 +434,8 @@ class FakeDeployment(BaseModel): self.id = str(mock_random.uuid4()) self.group_id = group_id self.group_arn = group_arn - self.created_at_datetime = datetime.utcnow() - self.update_at_datetime = datetime.utcnow() + self.created_at_datetime = utcnow() + self.update_at_datetime = utcnow() self.deployment_status = "InProgress" self.deployment_type = deployment_type self.arn = f"arn:aws:greengrass:{self.region_name}:{account_id}:/greengrass/groups/{self.group_id}/deployments/{self.id}" @@ -456,7 +456,7 @@ class FakeDeployment(BaseModel): class FakeAssociatedRole(BaseModel): def __init__(self, role_arn: str): self.role_arn = role_arn - self.associated_at = datetime.utcnow() + self.associated_at = utcnow() def to_dict(self, include_detail: bool = False) -> Dict[str, Any]: diff --git a/moto/greengrass/responses.py b/moto/greengrass/responses.py index 8eb1d1c46..e01f2249f 100644 --- a/moto/greengrass/responses.py +++ b/moto/greengrass/responses.py @@ -1,4 +1,3 @@ -from datetime import datetime from typing import Any import json @@ -789,11 +788,5 @@ class GreengrassResponse(BaseResponse): return ( 200, {"status": 200}, - json.dumps( - { - "DisassociatedAt": iso_8601_datetime_with_milliseconds( - datetime.utcnow() - ) - } - ), + json.dumps({"DisassociatedAt": iso_8601_datetime_with_milliseconds()}), ) diff --git a/moto/iam/models.py b/moto/iam/models.py index 76c1842a7..5346d652a 100644 --- a/moto/iam/models.py +++ b/moto/iam/models.py @@ -25,6 +25,7 @@ from moto.core.utils import ( iso_8601_datetime_without_milliseconds, iso_8601_datetime_with_milliseconds, unix_time, + utcnow, ) from moto.iam.policy_validation import ( IAMPolicyDocumentValidator, @@ -84,7 +85,7 @@ def mark_account_as_visited( account = iam_backends[account_id] if access_key in account["global"].access_keys: account["global"].access_keys[access_key].last_used = AccessKeyLastUsed( - timestamp=datetime.utcnow(), service=service, region=region + timestamp=utcnow(), service=service, region=region ) else: # User provided access credentials unknown to us @@ -100,7 +101,7 @@ class MFADevice: def __init__( self, serial_number: str, authentication_code_1: str, authentication_code_2: str ): - self.enable_date = datetime.utcnow() + self.enable_date = utcnow() self.serial_number = serial_number self.authentication_code_1 = authentication_code_1 self.authentication_code_2 = authentication_code_2 @@ -130,7 +131,9 @@ class VirtualMfaDevice: @property def enabled_iso_8601(self) -> str: - return iso_8601_datetime_without_milliseconds(self.enable_date) # type: ignore[return-value] + if self.enable_date: + return iso_8601_datetime_without_milliseconds(self.enable_date) + return "" class Policy(CloudFormationModel): @@ -172,8 +175,8 @@ class Policy(CloudFormationModel): ) ] - self.create_date = create_date or datetime.utcnow() - self.update_date = update_date or datetime.utcnow() + self.create_date = create_date or utcnow() + self.update_date = update_date or utcnow() def update_default_version(self, new_default_version_id: str) -> None: for version in self.versions: @@ -225,7 +228,7 @@ class OpenIDConnectProvider(BaseModel): self.url = parsed_url.netloc + parsed_url.path self.thumbprint_list = thumbprint_list self.client_id_list = client_id_list - self.create_date = datetime.utcnow() + self.create_date = utcnow() self.tags = tags or {} @property @@ -316,7 +319,7 @@ class PolicyVersion: self.is_default = is_default self.version_id = version_id - self.create_date = create_date or datetime.utcnow() + self.create_date = create_date or utcnow() @property def created_iso_8601(self) -> str: @@ -662,7 +665,7 @@ class Role(CloudFormationModel): self.path = path or "/" self.policies: Dict[str, str] = {} self.managed_policies: Dict[str, ManagedPolicy] = {} - self.create_date = datetime.utcnow() + self.create_date = utcnow() self.tags = tags self.last_used = None self.last_used_region = None @@ -907,7 +910,7 @@ class InstanceProfile(CloudFormationModel): self.name = name self.path = path or "/" self.roles = roles if roles else [] - self.create_date = datetime.utcnow() + self.create_date = utcnow() self.tags = {tag["Key"]: tag["Value"] for tag in tags or []} @property @@ -1044,7 +1047,7 @@ class SigningCertificate(BaseModel): self.id = certificate_id self.user_name = user_name self.body = body - self.upload_date = datetime.utcnow() + self.upload_date = utcnow() self.status = "Active" @property @@ -1077,7 +1080,7 @@ class AccessKey(CloudFormationModel): ) self.secret_access_key = random_alphanumeric(40) self.status = status - self.create_date = datetime.utcnow() + self.create_date = utcnow() self.last_used: Optional[datetime] = None @property @@ -1182,7 +1185,7 @@ class SshPublicKey(BaseModel): self.ssh_public_key_id = "APKA" + random_access_key() self.fingerprint = md5_hash(ssh_public_key_body.encode()).hexdigest() self.status = "Active" - self.upload_date = datetime.utcnow() + self.upload_date = utcnow() @property def uploaded_iso_8601(self) -> str: @@ -1195,7 +1198,7 @@ class Group(BaseModel): self.name = name self.id = random_resource_id() self.path = path - self.create_date = datetime.utcnow() + self.create_date = utcnow() self.users: List[User] = [] self.managed_policies: Dict[str, str] = {} @@ -1255,7 +1258,7 @@ class User(CloudFormationModel): self.name = name self.id = random_resource_id() self.path = path if path else "/" - self.create_date = datetime.utcnow() + self.create_date = utcnow() self.mfa_devices: Dict[str, MFADevice] = {} self.policies: Dict[str, str] = {} self.managed_policies: Dict[str, Dict[str, str]] = {} @@ -2851,7 +2854,7 @@ class IAMBackend(BaseBackend): device = self.virtual_mfa_devices.get(serial_number, None) if device: - device.enable_date = datetime.utcnow() + device.enable_date = utcnow() device.user = user device.user_attribute = { "Path": user.path, diff --git a/moto/instance_metadata/responses.py b/moto/instance_metadata/responses.py index 91c74ce66..026de7382 100644 --- a/moto/instance_metadata/responses.py +++ b/moto/instance_metadata/responses.py @@ -5,6 +5,7 @@ from urllib.parse import urlparse from moto.core.common_types import TYPE_RESPONSE from moto.core.responses import BaseResponse +from moto.core.utils import utcnow class InstanceMetadataResponse(BaseResponse): @@ -27,7 +28,7 @@ class InstanceMetadataResponse(BaseResponse): """ parsed_url = urlparse(full_url) - tomorrow = datetime.datetime.utcnow() + datetime.timedelta(days=1) + tomorrow = utcnow() + datetime.timedelta(days=1) credentials = dict( AccessKeyId="test-key", SecretAccessKey="test-secret-key", diff --git a/moto/iot/models.py b/moto/iot/models.py index f03bbe104..14289a00d 100644 --- a/moto/iot/models.py +++ b/moto/iot/models.py @@ -13,6 +13,7 @@ from typing import Any, Dict, List, Tuple, Optional, Pattern, Iterable, TYPE_CHE from .utils import PAGINATION_MODEL from moto.core import BaseBackend, BackendDict, BaseModel +from moto.core.utils import utcnow from moto.moto_api._internal import mock_random as random from moto.utilities.paginator import paginate from .exceptions import ( @@ -80,7 +81,7 @@ class FakeThing(BaseModel): if include_connectivity: obj["connectivity"] = { "connected": True, - "timestamp": time.mktime(datetime.utcnow().timetuple()), + "timestamp": time.mktime(utcnow().timetuple()), } return obj @@ -673,8 +674,8 @@ class IoTBackend(BaseBackend): .issuer_name(issuer) .public_key(key.public_key()) .serial_number(x509.random_serial_number()) - .not_valid_before(datetime.utcnow()) - .not_valid_after(datetime.utcnow() + timedelta(days=365)) + .not_valid_before(utcnow()) + .not_valid_after(utcnow() + timedelta(days=365)) .add_extension(x509.SubjectAlternativeName(sans), critical=False) .sign(key, hashes.SHA512(), default_backend()) ) diff --git a/moto/kinesis/models.py b/moto/kinesis/models.py index bcffd6c29..adabb162c 100644 --- a/moto/kinesis/models.py +++ b/moto/kinesis/models.py @@ -11,7 +11,7 @@ from operator import attrgetter from typing import Any, Dict, List, Optional, Tuple, Iterable from moto.core import BaseBackend, BackendDict, BaseModel, CloudFormationModel -from moto.core.utils import unix_time +from moto.core.utils import unix_time, utcnow from moto.moto_api._internal import mock_random as random from moto.utilities.paginator import paginate from moto.utilities.utils import md5_hash @@ -73,7 +73,7 @@ class Record(BaseModel): self.data = data self.sequence_number = sequence_number self.explicit_hash_key = explicit_hash_key - self.created_at_datetime = datetime.datetime.utcnow() + self.created_at_datetime = utcnow() self.created_at = unix_time(self.created_at_datetime) def to_json(self) -> Dict[str, Any]: diff --git a/moto/kinesisvideo/models.py b/moto/kinesisvideo/models.py index 7edb15137..30f1a890e 100644 --- a/moto/kinesisvideo/models.py +++ b/moto/kinesisvideo/models.py @@ -1,6 +1,6 @@ -from datetime import datetime from typing import Any, Dict, List from moto.core import BaseBackend, BackendDict, BaseModel +from moto.core.utils import utcnow from .exceptions import ResourceNotFoundException, ResourceInUseException from moto.moto_api._internal import mock_random as random @@ -26,7 +26,7 @@ class Stream(BaseModel): self.tags = tags self.status = "ACTIVE" self.version = random.get_random_string(include_digits=False, lower_case=True) - self.creation_time = datetime.utcnow() + self.creation_time = utcnow() stream_arn = f"arn:aws:kinesisvideo:{region_name}:{account_id}:stream/{stream_name}/1598784211076" self.data_endpoint_number = random.get_random_hex() self.arn = stream_arn diff --git a/moto/logs/models.py b/moto/logs/models.py index 4fb14dddf..8e960651b 100644 --- a/moto/logs/models.py +++ b/moto/logs/models.py @@ -1,8 +1,8 @@ -from datetime import datetime, timedelta +from datetime import timedelta from typing import Any, Dict, Iterable, List, Tuple, Optional from moto.core import BaseBackend, BackendDict, BaseModel from moto.core import CloudFormationModel -from moto.core.utils import unix_time_millis +from moto.core.utils import unix_time_millis, utcnow from moto.logs.metric_filters import MetricFilters from moto.logs.exceptions import ( ResourceNotFoundException, @@ -914,8 +914,8 @@ class LogsBackend(BaseBackend): rejected_info = {} allowed_events = [] last_timestamp = None - oldest = int(unix_time_millis(datetime.utcnow() - timedelta(days=14))) - newest = int(unix_time_millis(datetime.utcnow() + timedelta(hours=2))) + oldest = int(unix_time_millis(utcnow() - timedelta(days=14))) + newest = int(unix_time_millis(utcnow() + timedelta(hours=2))) for idx, event in enumerate(log_events): if last_timestamp and last_timestamp > event["timestamp"]: raise InvalidParameterException( diff --git a/moto/managedblockchain/models.py b/moto/managedblockchain/models.py index 7d5c85dba..edb50d56c 100644 --- a/moto/managedblockchain/models.py +++ b/moto/managedblockchain/models.py @@ -3,6 +3,7 @@ import re from typing import Any, Dict, List, Optional from moto.core import BaseBackend, BackendDict, BaseModel +from moto.core.utils import utcnow from .exceptions import ( BadRequestException, @@ -62,7 +63,7 @@ class ManagedBlockchainNetwork(BaseModel): region: str, description: Optional[str] = None, ): - self.creationdate = datetime.datetime.utcnow() + self.creationdate = utcnow() self.id = network_id self.name = name self.description = description @@ -179,7 +180,7 @@ class ManagedBlockchainProposal(BaseModel): self.network_threshold_comp = network_threshold_comp self.description = description - self.creationdate = datetime.datetime.utcnow() + self.creationdate = utcnow() self.expirationdate = self.creationdate + datetime.timedelta( hours=network_expiration ) @@ -211,7 +212,7 @@ class ManagedBlockchainProposal(BaseModel): return [] def check_to_expire_proposal(self) -> None: - if datetime.datetime.utcnow() > self.expirationdate: + if utcnow() > self.expirationdate: self.status = "EXPIRED" def to_dict(self) -> Dict[str, Any]: @@ -303,7 +304,7 @@ class ManagedBlockchainInvitation(BaseModel): self.status = "PENDING" self.region = region - self.creationdate = datetime.datetime.utcnow() + self.creationdate = utcnow() self.expirationdate = self.creationdate + datetime.timedelta(days=7) @property @@ -351,7 +352,7 @@ class ManagedBlockchainMember(BaseModel): member_configuration: Dict[str, Any], region: str, ): - self.creationdate = datetime.datetime.utcnow() + self.creationdate = utcnow() self.id = member_id self.networkid = networkid self.member_configuration = member_configuration @@ -430,7 +431,7 @@ class ManagedBlockchainNode(BaseModel): logpublishingconfiguration: Dict[str, Any], region: str, ): - self.creationdate = datetime.datetime.utcnow() + self.creationdate = utcnow() self.id = node_id self.instancetype = instancetype self.networkid = networkid diff --git a/moto/opsworks/models.py b/moto/opsworks/models.py index 69796d2e3..221671ddf 100644 --- a/moto/opsworks/models.py +++ b/moto/opsworks/models.py @@ -1,7 +1,7 @@ from moto.core import BaseBackend, BackendDict, BaseModel +from moto.core.utils import utcnow from moto.ec2 import ec2_backends from moto.moto_api._internal import mock_random as random -import datetime from typing import Any, Dict, List, Optional from .exceptions import ResourceNotFoundException, ValidationException @@ -79,7 +79,7 @@ class OpsworkInstance(BaseModel): self.platform = "linux (fixed)" self.id = str(random.uuid4()) - self.created_at = datetime.datetime.utcnow() + self.created_at = utcnow() def start(self) -> None: """ @@ -257,7 +257,7 @@ class Layer(BaseModel): self.use_ebs_optimized_instances = use_ebs_optimized_instances self.id = str(random.uuid4()) - self.created_at = datetime.datetime.utcnow() + self.created_at = utcnow() def __eq__(self, other: Any) -> bool: return self.id == other.id @@ -351,7 +351,7 @@ class Stack(BaseModel): self.layers: List[Layer] = [] self.apps: List[App] = [] self.account_number = account_id - self.created_at = datetime.datetime.utcnow() + self.created_at = utcnow() def __eq__(self, other: Any) -> bool: return self.id == other.id @@ -432,7 +432,7 @@ class App(BaseModel): self.environment = environment or {} self.id = str(random.uuid4()) - self.created_at = datetime.datetime.utcnow() + self.created_at = utcnow() def __eq__(self, other: Any) -> bool: return self.id == other.id diff --git a/moto/organizations/models.py b/moto/organizations/models.py index 71c211910..21c09f5ae 100644 --- a/moto/organizations/models.py +++ b/moto/organizations/models.py @@ -1,11 +1,10 @@ -import datetime import re import json from typing import Any, Dict, List, Optional from moto.core import BaseBackend, BackendDict, BaseModel from moto.core.exceptions import RESTError -from moto.core.utils import unix_time +from moto.core.utils import unix_time, utcnow from moto.organizations import utils from moto.organizations.exceptions import ( InvalidInputException, @@ -70,7 +69,7 @@ class FakeAccount(BaseModel): self.id = utils.make_random_account_id() self.name = kwargs["AccountName"] self.email = kwargs["Email"] - self.create_time = datetime.datetime.utcnow() + self.create_time = utcnow() self.status = "ACTIVE" self.joined_method = "CREATED" self.parent_id = organization.root_id @@ -290,7 +289,7 @@ class FakeServiceAccess(BaseModel): ) self.service_principal = kwargs["ServicePrincipal"] - self.date_enabled = datetime.datetime.utcnow() + self.date_enabled = utcnow() def describe(self) -> Dict[str, Any]: return { @@ -317,7 +316,7 @@ class FakeDelegatedAdministrator(BaseModel): def __init__(self, account: FakeAccount): self.account = account - self.enabled_date = datetime.datetime.utcnow() + self.enabled_date = utcnow() self.services: Dict[str, Any] = {} def add_service_principal(self, service_principal: str) -> None: @@ -331,7 +330,7 @@ class FakeDelegatedAdministrator(BaseModel): self.services[service_principal] = { "ServicePrincipal": service_principal, - "DelegationEnabledDate": unix_time(datetime.datetime.utcnow()), + "DelegationEnabledDate": unix_time(), } def remove_service_principal(self, service_principal: str) -> None: diff --git a/moto/ram/models.py b/moto/ram/models.py index 379116f39..e4f12106e 100644 --- a/moto/ram/models.py +++ b/moto/ram/models.py @@ -1,10 +1,9 @@ import re import string -from datetime import datetime from typing import Any, Dict, List from moto.core import BaseBackend, BackendDict, BaseModel -from moto.core.utils import unix_time +from moto.core.utils import unix_time, utcnow from moto.moto_api._internal import mock_random as random from moto.organizations.models import organizations_backends, OrganizationsBackend from moto.ram.exceptions import ( @@ -46,9 +45,9 @@ class ResourceShare(BaseModel): self.arn = ( f"arn:aws:ram:{self.region}:{account_id}:resource-share/{random.uuid4()}" ) - self.creation_time = datetime.utcnow() + self.creation_time = utcnow() self.feature_set = "STANDARD" - self.last_updated_time = datetime.utcnow() + self.last_updated_time = utcnow() self.name = kwargs["name"] self.owning_account_id = account_id self.principals: List[str] = [] @@ -128,7 +127,7 @@ class ResourceShare(BaseModel): self.resource_arns.append(resource) def delete(self) -> None: - self.last_updated_time = datetime.utcnow() + self.last_updated_time = utcnow() self.status = "DELETED" def describe(self) -> Dict[str, Any]: @@ -147,7 +146,7 @@ class ResourceShare(BaseModel): self.allow_external_principals = kwargs.get( "allowExternalPrincipals", self.allow_external_principals ) - self.last_updated_time = datetime.utcnow() + self.last_updated_time = utcnow() self.name = kwargs.get("name", self.name) diff --git a/moto/rds/models.py b/moto/rds/models.py index 8405171e7..c430ab369 100644 --- a/moto/rds/models.py +++ b/moto/rds/models.py @@ -1,5 +1,4 @@ import copy -import datetime import os import re import string @@ -136,9 +135,7 @@ class Cluster: self.status = "active" self.account_id = kwargs.get("account_id") self.region_name = kwargs.get("region") - self.cluster_create_time = iso_8601_datetime_with_milliseconds( - datetime.datetime.utcnow() - ) + self.cluster_create_time = iso_8601_datetime_with_milliseconds() self.copy_tags_to_snapshot = kwargs.get("copy_tags_to_snapshot") if self.copy_tags_to_snapshot is None: self.copy_tags_to_snapshot = True @@ -194,9 +191,7 @@ class Cluster: kwargs.get("enable_cloudwatch_logs_exports") or [] ) self.enable_http_endpoint = kwargs.get("enable_http_endpoint") # type: ignore - self.earliest_restorable_time = iso_8601_datetime_with_milliseconds( - datetime.datetime.utcnow() - ) + self.earliest_restorable_time = iso_8601_datetime_with_milliseconds() self.scaling_configuration = kwargs.get("scaling_configuration") if not self.scaling_configuration and self.engine_mode == "serverless": # In AWS, this default configuration only shows up when the Cluster is in a ready state, so a few minutes after creation @@ -470,9 +465,7 @@ class ClusterSnapshot(BaseModel): self.snapshot_type = snapshot_type self.tags = tags self.status = "available" - self.created_at = iso_8601_datetime_with_milliseconds( - datetime.datetime.utcnow() - ) + self.created_at = iso_8601_datetime_with_milliseconds() @property def arn(self) -> str: @@ -592,9 +585,7 @@ class Database(CloudFormationModel): self.port = Database.default_port(self.engine) # type: ignore self.db_instance_identifier = kwargs.get("db_instance_identifier") self.db_name = kwargs.get("db_name") - self.instance_create_time = iso_8601_datetime_with_milliseconds( - datetime.datetime.utcnow() - ) + self.instance_create_time = iso_8601_datetime_with_milliseconds() self.publicly_accessible = kwargs.get("publicly_accessible") if self.publicly_accessible is None: self.publicly_accessible = True @@ -1108,9 +1099,7 @@ class DatabaseSnapshot(BaseModel): self.snapshot_type = snapshot_type self.tags = tags self.status = "available" - self.created_at = iso_8601_datetime_with_milliseconds( - datetime.datetime.utcnow() - ) + self.created_at = iso_8601_datetime_with_milliseconds() @property def arn(self) -> str: @@ -1191,9 +1180,7 @@ class ExportTask(BaseModel): self.export_only = kwargs.get("export_only", []) self.status = "complete" - self.created_at = iso_8601_datetime_with_milliseconds( - datetime.datetime.utcnow() - ) + self.created_at = iso_8601_datetime_with_milliseconds() self.source_type = ( "SNAPSHOT" if type(snapshot) is DatabaseSnapshot else "CLUSTER" ) @@ -1241,9 +1228,7 @@ class EventSubscription(BaseModel): self.region_name = "" self.customer_aws_id = kwargs["account_id"] self.status = "active" - self.created_at = iso_8601_datetime_with_milliseconds( - datetime.datetime.utcnow() - ) + self.created_at = iso_8601_datetime_with_milliseconds() @property def es_arn(self) -> str: diff --git a/moto/redshift/models.py b/moto/redshift/models.py index ff31fc59a..1f868e69f 100644 --- a/moto/redshift/models.py +++ b/moto/redshift/models.py @@ -99,9 +99,7 @@ class Cluster(TaggableResourceMixin, CloudFormationModel): super().__init__(redshift_backend.account_id, region_name, tags) self.redshift_backend = redshift_backend self.cluster_identifier = cluster_identifier - self.create_time = iso_8601_datetime_with_milliseconds( - datetime.datetime.now(tzutc()) - ) + self.create_time = iso_8601_datetime_with_milliseconds() self.status = "available" self.node_type = node_type self.master_username = master_username @@ -532,9 +530,7 @@ class Snapshot(TaggableResourceMixin, BaseModel): self.snapshot_identifier = snapshot_identifier self.snapshot_type = snapshot_type self.status = "available" - self.create_time = iso_8601_datetime_with_milliseconds( - datetime.datetime.now(tzutc()) - ) + self.create_time = iso_8601_datetime_with_milliseconds() self.iam_roles_arn = iam_roles_arn or [] @property diff --git a/moto/route53/responses.py b/moto/route53/responses.py index 7356cd6fc..f13cecb2a 100644 --- a/moto/route53/responses.py +++ b/moto/route53/responses.py @@ -1,5 +1,4 @@ """Handles Route53 API requests, invokes method and returns response.""" -import datetime import re from urllib.parse import parse_qs @@ -357,9 +356,7 @@ class Route53(BaseResponse): 200, headers, template.render( - timestamp=iso_8601_datetime_with_milliseconds( - datetime.datetime.utcnow() - ), + timestamp=iso_8601_datetime_with_milliseconds(), ), ) diff --git a/moto/s3/models.py b/moto/s3/models.py index ea70f7caa..b88cced02 100644 --- a/moto/s3/models.py +++ b/moto/s3/models.py @@ -21,6 +21,7 @@ from moto.core.utils import ( rfc_1123_datetime, unix_time, unix_time_millis, + utcnow, ) from moto.cloudwatch.models import MetricDatum from moto.moto_api import state_manager @@ -70,7 +71,7 @@ class FakeDeleteMarker(BaseModel): def __init__(self, key: "FakeKey"): self.key = key self.name = key.name - self.last_modified = datetime.datetime.utcnow() + self.last_modified = utcnow() self._version_id = str(random.uuid4()) @property @@ -113,7 +114,7 @@ class FakeKey(BaseModel, ManagedState): ) self.name = name self.account_id = account_id - self.last_modified = datetime.datetime.utcnow() + self.last_modified = utcnow() self.acl: Optional[FakeAcl] = get_canned_acl("private") self.website_redirect_location: Optional[str] = None self.checksum_algorithm = None @@ -197,7 +198,7 @@ class FakeKey(BaseModel, ManagedState): self.acl = acl def restore(self, days: int) -> None: - self._expiry = datetime.datetime.utcnow() + datetime.timedelta(days) + self._expiry = utcnow() + datetime.timedelta(days) @property def etag(self) -> str: @@ -324,7 +325,7 @@ class FakeKey(BaseModel, ManagedState): return True if self.lock_mode == "COMPLIANCE": - now = datetime.datetime.utcnow() + now = utcnow() try: until = datetime.datetime.strptime( self.lock_until, "%Y-%m-%dT%H:%M:%SZ" # type: ignore @@ -1532,7 +1533,7 @@ class FakeBucket(CloudFormationModel): return False def default_retention(self) -> str: - now = datetime.datetime.utcnow() + now = utcnow() now += datetime.timedelta(self.default_lock_days) # type: ignore now += datetime.timedelta(self.default_lock_years * 365) # type: ignore return now.strftime("%Y-%m-%dT%H:%M:%SZ") diff --git a/moto/s3control/config.py b/moto/s3control/config.py index 2fdaa2ad8..a6906d9e1 100644 --- a/moto/s3control/config.py +++ b/moto/s3control/config.py @@ -1,4 +1,3 @@ -import datetime import json from boto3 import Session @@ -6,7 +5,7 @@ from typing import Any, Dict, List, Optional, Tuple from moto.core.exceptions import InvalidNextTokenException from moto.core.common_models import ConfigQueryModel -from moto.core.utils import unix_time +from moto.core.utils import unix_time, utcnow from moto.s3control import s3control_backends @@ -130,7 +129,7 @@ class S3AccountPublicAccessBlockConfigQuery(ConfigQueryModel): return None # Format the PAB to the AWS Config format: - creation_time = datetime.datetime.utcnow() + creation_time = utcnow() config_data = { "version": "1.3", "accountId": account_id, diff --git a/moto/secretsmanager/models.py b/moto/secretsmanager/models.py index a60ba1cd4..36fd3e7f1 100644 --- a/moto/secretsmanager/models.py +++ b/moto/secretsmanager/models.py @@ -5,6 +5,7 @@ import datetime from typing import Any, Dict, List, Tuple, Optional from moto.core import BaseBackend, BackendDict, BaseModel +from moto.core.utils import utcnow from moto.moto_api._internal import mock_random from .exceptions import ( SecretNotFoundException, @@ -783,7 +784,7 @@ class SecretsManagerBackend(BaseBackend): else: arn = secret_arn(self.account_id, self.region_name, secret_id=secret_id) name = secret_id - deletion_date = datetime.datetime.utcnow() + deletion_date = utcnow() return arn, name, self._unix_time_secs(deletion_date) else: if self.secrets[secret_id].is_deleted(): @@ -792,7 +793,7 @@ class SecretsManagerBackend(BaseBackend): perform the operation on a secret that's currently marked deleted." ) - deletion_date = datetime.datetime.utcnow() + deletion_date = utcnow() if force_delete_without_recovery: secret = self.secrets.pop(secret_id) diff --git a/moto/ses/models.py b/moto/ses/models.py index 04d5bcd53..1303a7a6c 100644 --- a/moto/ses/models.py +++ b/moto/ses/models.py @@ -8,6 +8,7 @@ from email.encoders import encode_7or8bit from typing import Any, Dict, List, Optional from moto.core import BaseBackend, BackendDict, BaseModel +from moto.core.utils import utcnow from moto.sns.models import sns_backends from .exceptions import ( MessageRejectedError, @@ -431,7 +432,7 @@ class SESBackend(BaseBackend): "Rejects": self.rejected_messages_count, "Complaints": 0, "Bounces": 0, - "Timestamp": datetime.datetime.utcnow(), + "Timestamp": utcnow(), } def add_template(self, template_info: Dict[str, str]) -> None: diff --git a/moto/ses/responses.py b/moto/ses/responses.py index ac5b2f56c..186feddb8 100644 --- a/moto/ses/responses.py +++ b/moto/ses/responses.py @@ -2,8 +2,8 @@ import base64 from typing import Any, Dict, List from moto.core.responses import BaseResponse +from moto.core.utils import utcnow from .models import ses_backends, SESBackend -from datetime import datetime class EmailResponse(BaseResponse): @@ -247,7 +247,7 @@ class EmailResponse(BaseResponse): template_info["html_part"] = template_data.get("._html_part", "") template_info["template_name"] = template_data.get("._name", "") template_info["subject_part"] = template_data.get("._subject_part", "") - template_info["Timestamp"] = datetime.utcnow() + template_info["Timestamp"] = utcnow() self.backend.add_template(template_info=template_info) template = self.response_template(CREATE_TEMPLATE) return template.render() @@ -259,7 +259,7 @@ class EmailResponse(BaseResponse): template_info["html_part"] = template_data.get("._html_part", "") template_info["template_name"] = template_data.get("._name", "") template_info["subject_part"] = template_data.get("._subject_part", "") - template_info["Timestamp"] = datetime.utcnow() + template_info["Timestamp"] = utcnow() self.backend.update_template(template_info=template_info) template = self.response_template(UPDATE_TEMPLATE) return template.render() diff --git a/moto/sesv2/models.py b/moto/sesv2/models.py index 2ab67f96e..465199d64 100644 --- a/moto/sesv2/models.py +++ b/moto/sesv2/models.py @@ -1,6 +1,5 @@ """SESV2Backend class with methods for supported APIs.""" -from datetime import datetime as dt from moto.core import BackendDict, BaseBackend, BaseModel from ..ses.models import ses_backends, Message, RawMessage from typing import Dict, List, Any @@ -21,8 +20,8 @@ class Contact(BaseModel): self.topic_default_preferences: List[Dict[str, str]] = [] self.topic_preferences = topic_preferences self.unsubscribe_all = unsubscribe_all - self.created_timestamp = iso_8601_datetime_with_milliseconds(dt.utcnow()) - self.last_updated_timestamp = iso_8601_datetime_with_milliseconds(dt.utcnow()) + self.created_timestamp = iso_8601_datetime_with_milliseconds() + self.last_updated_timestamp = iso_8601_datetime_with_milliseconds() @property def response_object(self) -> Dict[str, Any]: # type: ignore[misc] @@ -47,8 +46,8 @@ class ContactList(BaseModel): self.contact_list_name = contact_list_name self.description = description self.topics = topics - self.created_timestamp = iso_8601_datetime_with_milliseconds(dt.utcnow()) - self.last_updated_timestamp = iso_8601_datetime_with_milliseconds(dt.utcnow()) + self.created_timestamp = iso_8601_datetime_with_milliseconds() + self.last_updated_timestamp = iso_8601_datetime_with_milliseconds() self.contacts: Dict[str, Contact] = {} def create_contact(self, contact_list_name: str, params: Dict[str, Any]) -> None: diff --git a/moto/sns/models.py b/moto/sns/models.py index c707b0500..37c521032 100644 --- a/moto/sns/models.py +++ b/moto/sns/models.py @@ -1,5 +1,4 @@ import contextlib -import datetime import json import requests import re @@ -299,9 +298,7 @@ class Subscription(BaseModel): "MessageId": message_id, "TopicArn": self.topic.arn, "Message": message, - "Timestamp": iso_8601_datetime_with_milliseconds( - datetime.datetime.utcnow() - ), + "Timestamp": iso_8601_datetime_with_milliseconds(), "SignatureVersion": "1", "Signature": "EXAMPLElDMXvB8r9R83tGoNn0ecwd5UjllzsvSvbItzfaMpN2nk5HVSw7XnOn/49IkxDKz8YrlH2qJXj2iZB0Zo2O71c4qQk1fMUDi3LGpij7RCW7AW9vYYsSqIKRnFS94ilu7NFhUzLiieYr4BKHpdTmdD6c0esKEYBpabxDSc=", "SigningCertURL": "https://sns.us-east-1.amazonaws.com/SimpleNotificationService-f3ecfb7224c7233fe7bb5f59f96de52f.pem", diff --git a/moto/ssm/models.py b/moto/ssm/models.py index f570edf32..280772bd3 100644 --- a/moto/ssm/models.py +++ b/moto/ssm/models.py @@ -7,6 +7,7 @@ from collections import defaultdict from moto.core import BaseBackend, BackendDict, BaseModel, CloudFormationModel from moto.core.exceptions import RESTError +from moto.core.utils import utcnow from moto.ec2 import ec2_backends from moto.secretsmanager import secretsmanager_backends from moto.secretsmanager.exceptions import SecretsManagerClientError @@ -589,7 +590,7 @@ class Document(BaseModel): self.status = "Active" self.document_version = document_version self.owner = account_id - self.created_date = datetime.datetime.utcnow() + self.created_date = utcnow() if document_format == "JSON": try: diff --git a/moto/stepfunctions/models.py b/moto/stepfunctions/models.py index 26c6965f6..bcabad663 100644 --- a/moto/stepfunctions/models.py +++ b/moto/stepfunctions/models.py @@ -31,7 +31,7 @@ class StateMachine(CloudFormationModel): roleArn: str, tags: Optional[List[Dict[str, str]]] = None, ): - self.creation_date = iso_8601_datetime_with_milliseconds(datetime.now()) + self.creation_date = iso_8601_datetime_with_milliseconds() self.update_date = self.creation_date self.arn = arn self.name = name @@ -92,7 +92,7 @@ class StateMachine(CloudFormationModel): for key, value in kwargs.items(): if value is not None: setattr(self, key, value) - self.update_date = iso_8601_datetime_with_milliseconds(datetime.now()) + self.update_date = iso_8601_datetime_with_milliseconds() def add_tags(self, tags: List[Dict[str, str]]) -> List[Dict[str, str]]: merged_tags = [] @@ -253,7 +253,7 @@ class Execution: ) self.execution_arn = execution_arn self.name = execution_name - self.start_date = iso_8601_datetime_with_milliseconds(datetime.now()) + self.start_date = iso_8601_datetime_with_milliseconds() self.state_machine_arn = state_machine_arn self.execution_input = execution_input self.status = ( @@ -364,7 +364,7 @@ class Execution: def stop(self) -> None: self.status = "ABORTED" - self.stop_date = iso_8601_datetime_with_milliseconds(datetime.now()) + self.stop_date = iso_8601_datetime_with_milliseconds() class StepFunctionBackend(BaseBackend): diff --git a/moto/sts/models.py b/moto/sts/models.py index c60525415..5423761c5 100644 --- a/moto/sts/models.py +++ b/moto/sts/models.py @@ -5,7 +5,7 @@ import re import xmltodict from moto.core import BaseBackend, BaseModel, BackendDict -from moto.core.utils import iso_8601_datetime_with_milliseconds +from moto.core.utils import iso_8601_datetime_with_milliseconds, utcnow from moto.iam.models import iam_backends, AccessKey from moto.sts.utils import ( random_session_token, @@ -16,7 +16,7 @@ from moto.sts.utils import ( class Token(BaseModel): def __init__(self, duration: int, name: Optional[str] = None): - now = datetime.datetime.utcnow() + now = utcnow() self.expiration = now + datetime.timedelta(seconds=duration) self.name = name self.policy = None @@ -41,7 +41,7 @@ class AssumedRole(BaseModel): self.session_name = role_session_name self.role_arn = role_arn self.policy = policy - now = datetime.datetime.utcnow() + now = utcnow() self.expiration = now + datetime.timedelta(seconds=duration) self.external_id = external_id self.access_key = access_key diff --git a/moto/swf/models/activity_task.py b/moto/swf/models/activity_task.py index 37b42f29c..2ef3b52d6 100644 --- a/moto/swf/models/activity_task.py +++ b/moto/swf/models/activity_task.py @@ -1,8 +1,7 @@ -from datetime import datetime from typing import Any, Dict, Optional, TYPE_CHECKING from moto.core import BaseModel -from moto.core.utils import unix_time +from moto.core.utils import unix_time, utcnow from moto.moto_api._internal import mock_random from ..exceptions import SWFWorkflowExecutionClosedError @@ -37,7 +36,7 @@ class ActivityTask(BaseModel): self.workflow_execution = workflow_execution # this is *not* necessarily coherent with workflow execution history, # but that shouldn't be a problem for tests - self.scheduled_at = datetime.utcnow() + self.scheduled_at = utcnow() def _check_workflow_execution_open(self) -> None: if not self.workflow_execution.open: diff --git a/moto/swf/models/decision_task.py b/moto/swf/models/decision_task.py index 34f7e67c1..d41a189cf 100644 --- a/moto/swf/models/decision_task.py +++ b/moto/swf/models/decision_task.py @@ -1,8 +1,7 @@ -from datetime import datetime from typing import Any, Dict, Optional, TYPE_CHECKING from moto.core import BaseModel -from moto.core.utils import unix_time +from moto.core.utils import unix_time, utcnow from moto.moto_api._internal import mock_random from ..exceptions import SWFWorkflowExecutionClosedError @@ -29,7 +28,7 @@ class DecisionTask(BaseModel): self.state = "SCHEDULED" # this is *not* necessarily coherent with workflow execution history, # but that shouldn't be a problem for tests - self.scheduled_at = datetime.utcnow() + self.scheduled_at = utcnow() self.timeout_type: Optional[str] = None @property diff --git a/moto/wafv2/models.py b/moto/wafv2/models.py index 3ff346f57..f420c7a80 100644 --- a/moto/wafv2/models.py +++ b/moto/wafv2/models.py @@ -1,4 +1,3 @@ -import datetime import re from typing import Any, Dict, List, Optional, TYPE_CHECKING from moto.core import BaseBackend, BackendDict, BaseModel @@ -38,7 +37,7 @@ class FakeWebACL(BaseModel): rules: List[Dict[str, Any]], ): self.name = name - self.created_time = iso_8601_datetime_with_milliseconds(datetime.datetime.now()) + self.created_time = iso_8601_datetime_with_milliseconds() self.id = wacl_id self.arn = arn self.description = description or "" diff --git a/tests/test_acmpca/test_acmpca.py b/tests/test_acmpca/test_acmpca.py index 9bd2f1155..443fe0a33 100644 --- a/tests/test_acmpca/test_acmpca.py +++ b/tests/test_acmpca/test_acmpca.py @@ -4,6 +4,7 @@ import pytest from botocore.exceptions import ClientError from moto import mock_acmpca from moto.core import DEFAULT_ACCOUNT_ID +from moto.core.utils import utcnow import datetime import cryptography.x509 @@ -377,8 +378,8 @@ def create_cert(): .issuer_name(issuer) .public_key(key.public_key()) .serial_number(serial_number) - .not_valid_before(datetime.datetime.utcnow() - datetime.timedelta(days=10)) - .not_valid_after(datetime.datetime.utcnow() + datetime.timedelta(days=10)) + .not_valid_before(utcnow() - datetime.timedelta(days=10)) + .not_valid_after(utcnow() + datetime.timedelta(days=10)) .sign(key, hashes.SHA512(), default_backend()) ) diff --git a/tests/test_config/test_config.py b/tests/test_config/test_config.py index b32f0f071..4b66d951a 100644 --- a/tests/test_config/test_config.py +++ b/tests/test_config/test_config.py @@ -12,6 +12,7 @@ import pytest from moto import mock_s3 from moto.config import mock_config from moto.core import DEFAULT_ACCOUNT_ID as ACCOUNT_ID +from moto.core.utils import utcnow @mock_config @@ -1213,18 +1214,12 @@ def test_start_configuration_recorder(): result = client.describe_configuration_recorder_status()[ "ConfigurationRecordersStatus" ] - lower_bound = datetime.utcnow() - timedelta(minutes=5) + lower_bound = utcnow() - timedelta(minutes=5) assert result[0]["recording"] assert result[0]["lastStatus"] == "PENDING" + assert lower_bound < result[0]["lastStartTime"].replace(tzinfo=None) <= utcnow() assert ( - lower_bound - < result[0]["lastStartTime"].replace(tzinfo=None) - <= datetime.utcnow() - ) - assert ( - lower_bound - < result[0]["lastStatusChangeTime"].replace(tzinfo=None) - <= datetime.utcnow() + lower_bound < result[0]["lastStatusChangeTime"].replace(tzinfo=None) <= utcnow() ) @@ -1263,23 +1258,13 @@ def test_stop_configuration_recorder(): result = client.describe_configuration_recorder_status()[ "ConfigurationRecordersStatus" ] - lower_bound = datetime.utcnow() - timedelta(minutes=5) + lower_bound = utcnow() - timedelta(minutes=5) assert not result[0]["recording"] assert result[0]["lastStatus"] == "PENDING" + assert lower_bound < result[0]["lastStartTime"].replace(tzinfo=None) <= utcnow() + assert lower_bound < result[0]["lastStopTime"].replace(tzinfo=None) <= utcnow() assert ( - lower_bound - < result[0]["lastStartTime"].replace(tzinfo=None) - <= datetime.utcnow() - ) - assert ( - lower_bound - < result[0]["lastStopTime"].replace(tzinfo=None) - <= datetime.utcnow() - ) - assert ( - lower_bound - < result[0]["lastStatusChangeTime"].replace(tzinfo=None) - <= datetime.utcnow() + lower_bound < result[0]["lastStatusChangeTime"].replace(tzinfo=None) <= utcnow() ) diff --git a/tests/test_events/test_events_integration.py b/tests/test_events/test_events_integration.py index bb950b8ac..543151548 100644 --- a/tests/test_events/test_events_integration.py +++ b/tests/test_events/test_events_integration.py @@ -7,7 +7,7 @@ import os from moto import mock_events, mock_sqs, mock_logs, settings from moto.core import DEFAULT_ACCOUNT_ID as ACCOUNT_ID -from moto.core.utils import iso_8601_datetime_without_milliseconds +from moto.core.utils import iso_8601_datetime_without_milliseconds, utcnow @mock_events @@ -35,7 +35,7 @@ def test_send_to_cw_log_group(): ) # when - event_time = datetime.utcnow() + event_time = utcnow() client_events.put_events( Entries=[ { diff --git a/tests/test_iam/test_iam.py b/tests/test_iam/test_iam.py index fc733d29f..9f3301de4 100644 --- a/tests/test_iam/test_iam.py +++ b/tests/test_iam/test_iam.py @@ -6,6 +6,7 @@ from botocore.exceptions import ClientError from moto import mock_config, mock_iam, settings from moto.core import DEFAULT_ACCOUNT_ID as ACCOUNT_ID +from moto.core.utils import utcnow from moto.iam import iam_backends from moto.backends import get_backend from tests import DEFAULT_ACCOUNT_ID @@ -1530,11 +1531,7 @@ def test_create_access_key(): conn.create_access_key(UserName="my-user") conn.create_user(UserName="my-user") access_key = conn.create_access_key(UserName="my-user")["AccessKey"] - assert ( - 0 - <= (datetime.utcnow() - access_key["CreateDate"].replace(tzinfo=None)).seconds - < 10 - ) + assert 0 <= (utcnow() - access_key["CreateDate"].replace(tzinfo=None)).seconds < 10 assert len(access_key["AccessKeyId"]) == 20 assert len(access_key["SecretAccessKey"]) == 40 assert access_key["AccessKeyId"].startswith("AKIA") @@ -1545,11 +1542,7 @@ def test_create_access_key(): aws_secret_access_key=access_key["SecretAccessKey"], ) access_key = conn.create_access_key()["AccessKey"] - assert ( - 0 - <= (datetime.utcnow() - access_key["CreateDate"].replace(tzinfo=None)).seconds - < 10 - ) + assert 0 <= (utcnow() - access_key["CreateDate"].replace(tzinfo=None)).seconds < 10 assert len(access_key["AccessKeyId"]) == 20 assert len(access_key["SecretAccessKey"]) == 40 assert access_key["AccessKeyId"].startswith("AKIA") @@ -1904,7 +1897,7 @@ def test_get_credential_report_content(): UserName=username, AccessKeyId=key1["AccessKeyId"], Status="Inactive" ) key1 = conn.create_access_key(UserName=username)["AccessKey"] - timestamp = datetime.utcnow() + timestamp = utcnow() if not settings.TEST_SERVER_MODE: iam_backend = get_backend("iam")[ACCOUNT_ID]["global"] iam_backend.users[username].access_keys[1].last_used = timestamp @@ -2275,11 +2268,7 @@ def test_upload_ssh_public_key(): assert pubkey["SSHPublicKeyId"].startswith("APKA") assert "Fingerprint" in pubkey assert pubkey["Status"] == "Active" - assert ( - 0 - <= ((datetime.utcnow() - pubkey["UploadDate"].replace(tzinfo=None)).seconds) - < 10 - ) + assert 0 <= ((utcnow() - pubkey["UploadDate"].replace(tzinfo=None)).seconds) < 10 @mock_iam diff --git a/tests/test_iam/test_iam_groups.py b/tests/test_iam/test_iam_groups.py index 4fd3b59b5..0ff9a0810 100644 --- a/tests/test_iam/test_iam_groups.py +++ b/tests/test_iam/test_iam_groups.py @@ -7,6 +7,7 @@ import pytest from botocore.exceptions import ClientError from moto import mock_iam, settings from moto.core import DEFAULT_ACCOUNT_ID as ACCOUNT_ID +from moto.core.utils import utcnow from moto.backends import get_backend from freezegun import freeze_time from dateutil.tz import tzlocal @@ -123,7 +124,7 @@ def test_add_user_to_group(): # use internal api to set password, doesn't work in servermode if not settings.TEST_SERVER_MODE: iam_backend = get_backend("iam")[ACCOUNT_ID]["global"] - iam_backend.users[user].password_last_used = datetime.utcnow() + iam_backend.users[user].password_last_used = utcnow() # Execute result = conn.get_group(GroupName=group) diff --git a/tests/test_iam/test_iam_password_last_used.py b/tests/test_iam/test_iam_password_last_used.py index 045cc7f27..c8408447f 100644 --- a/tests/test_iam/test_iam_password_last_used.py +++ b/tests/test_iam/test_iam_password_last_used.py @@ -1,8 +1,8 @@ import boto3 -from datetime import datetime from moto import mock_iam, settings from moto.backends import get_backend from moto.core import DEFAULT_ACCOUNT_ID as ACCOUNT_ID +from moto.core.utils import utcnow from unittest import SkipTest @@ -31,7 +31,7 @@ def test_password_last_used(): assert not as_new_user.CurrentUser().password_last_used iam_backend = get_backend("iam")[ACCOUNT_ID]["global"] - iam_backend.users[username].password_last_used = datetime.utcnow() + iam_backend.users[username].password_last_used = utcnow() # Password is returned now assert as_new_user.CurrentUser().password_last_used diff --git a/tests/test_kinesis/test_kinesis.py b/tests/test_kinesis/test_kinesis.py index da6cdab45..556fe1a08 100644 --- a/tests/test_kinesis/test_kinesis.py +++ b/tests/test_kinesis/test_kinesis.py @@ -9,6 +9,7 @@ from dateutil.tz import tzlocal from moto import mock_kinesis from moto.core import DEFAULT_ACCOUNT_ID as ACCOUNT_ID +from moto.core.utils import utcnow @mock_kinesis @@ -406,7 +407,7 @@ def test_get_records_at_timestamp(): # To test around this limitation we wait until we well into the next second # before capturing the time and storing the records we expect to retrieve. time.sleep(1.0) - timestamp = datetime.datetime.utcnow() + timestamp = utcnow() keys = [str(i) for i in range(5, 10)] for k in keys: @@ -523,7 +524,7 @@ def test_get_records_at_very_new_timestamp(): for k in keys: conn.put_record(StreamName=stream_name, Data=k, PartitionKey=k) - timestamp = datetime.datetime.utcnow() + datetime.timedelta(seconds=1) + timestamp = utcnow() + datetime.timedelta(seconds=1) # Get a shard iterator response = conn.describe_stream(StreamName=stream_name) @@ -548,7 +549,7 @@ def test_get_records_from_empty_stream_at_timestamp(): stream_name = "my_stream" conn.create_stream(StreamName=stream_name, ShardCount=1) - timestamp = datetime.datetime.utcnow() + timestamp = utcnow() # Get a shard iterator response = conn.describe_stream(StreamName=stream_name) diff --git a/tests/test_kinesisvideoarchivedmedia/test_kinesisvideoarchivedmedia.py b/tests/test_kinesisvideoarchivedmedia/test_kinesisvideoarchivedmedia.py index 0bfe1dc69..fc5a5ff32 100644 --- a/tests/test_kinesisvideoarchivedmedia/test_kinesisvideoarchivedmedia.py +++ b/tests/test_kinesisvideoarchivedmedia/test_kinesisvideoarchivedmedia.py @@ -1,7 +1,8 @@ import boto3 from moto import mock_kinesisvideoarchivedmedia from moto import mock_kinesisvideo -from datetime import datetime, timedelta +from moto.core.utils import utcnow +from datetime import timedelta @mock_kinesisvideo @@ -67,7 +68,7 @@ def test_get_clip(): region_name=region_name, endpoint_url=data_endpoint, ) - end_timestamp = datetime.utcnow() - timedelta(hours=1) + end_timestamp = utcnow() - timedelta(hours=1) start_timestamp = end_timestamp - timedelta(minutes=5) res = client.get_clip( StreamName=stream_name, diff --git a/tests/test_logs/test_integration.py b/tests/test_logs/test_integration.py index 58d719acf..4714b5946 100644 --- a/tests/test_logs/test_integration.py +++ b/tests/test_logs/test_integration.py @@ -7,7 +7,6 @@ import zlib import boto3 from botocore.exceptions import ClientError -from datetime import datetime from moto import mock_logs, mock_lambda, mock_iam, mock_firehose, mock_s3 from moto import mock_kinesis from moto.core.utils import unix_time_millis @@ -152,8 +151,8 @@ def test_put_subscription_filter_with_lambda(): sub_filter["filterPattern"] = "" # when - ts_0 = int(unix_time_millis(datetime.utcnow())) - ts_1 = int(unix_time_millis(datetime.utcnow())) + 10 + ts_0 = int(unix_time_millis()) + ts_1 = int(unix_time_millis()) + 10 client_logs.put_log_events( logGroupName=log_group_name, logStreamName=log_stream_name, @@ -224,8 +223,8 @@ def test_subscription_filter_applies_to_new_streams(): client_logs.create_log_stream( # create log stream after subscription filter applied logGroupName=log_group_name, logStreamName=log_stream_name ) - ts_0 = int(unix_time_millis(datetime.utcnow())) - ts_1 = int(unix_time_millis(datetime.utcnow())) + 10 + ts_0 = int(unix_time_millis()) + ts_1 = int(unix_time_millis()) + 10 client_logs.put_log_events( logGroupName=log_group_name, logStreamName=log_stream_name, @@ -318,8 +317,8 @@ def test_put_subscription_filter_with_firehose(): _filter["filterPattern"] = "" # when - ts_0 = int(unix_time_millis(datetime.utcnow())) - ts_1 = int(unix_time_millis(datetime.utcnow())) + ts_0 = int(unix_time_millis()) + ts_1 = int(unix_time_millis()) client_logs.put_log_events( logGroupName=log_group_name, logStreamName=log_stream_name, @@ -383,8 +382,8 @@ def test_put_subscription_filter_with_kinesis(): ) # Create new log events - ts_0 = int(unix_time_millis(datetime.utcnow())) - ts_1 = int(unix_time_millis(datetime.utcnow())) + ts_0 = int(unix_time_millis()) + ts_1 = int(unix_time_millis()) logs.put_log_events( logGroupName="lg1", logStreamName="ls1", diff --git a/tests/test_logs/test_logs.py b/tests/test_logs/test_logs.py index 350754ccb..7eac61199 100644 --- a/tests/test_logs/test_logs.py +++ b/tests/test_logs/test_logs.py @@ -1,5 +1,5 @@ import json -from datetime import timedelta, datetime +from datetime import timedelta from uuid import UUID import boto3 @@ -8,7 +8,7 @@ from botocore.exceptions import ClientError from freezegun import freeze_time from moto import mock_logs, mock_s3, settings -from moto.core.utils import unix_time_millis +from moto.core.utils import unix_time_millis, utcnow from moto.logs.models import MAX_RESOURCE_POLICIES_PER_REGION TEST_REGION = "us-east-1" if settings.TEST_SERVER_MODE else "us-west-2" @@ -518,8 +518,8 @@ def test_put_log_events_in_wrong_order(): conn.create_log_group(logGroupName=log_group_name) conn.create_log_stream(logGroupName=log_group_name, logStreamName=log_stream_name) - ts_1 = int(unix_time_millis(datetime.utcnow() - timedelta(days=2))) - ts_2 = int(unix_time_millis(datetime.utcnow() - timedelta(days=5))) + ts_1 = int(unix_time_millis(utcnow() - timedelta(days=2))) + ts_2 = int(unix_time_millis(utcnow() - timedelta(days=5))) messages = [ {"message": f"Message {idx}", "timestamp": ts} @@ -550,7 +550,7 @@ def test_put_log_events_in_the_past(days_ago): conn.create_log_group(logGroupName=log_group_name) conn.create_log_stream(logGroupName=log_group_name, logStreamName=log_stream_name) - timestamp = int(unix_time_millis(datetime.utcnow() - timedelta(days=days_ago))) + timestamp = int(unix_time_millis(utcnow() - timedelta(days=days_ago))) messages = [{"message": "Message number {}", "timestamp": timestamp}] @@ -569,7 +569,7 @@ def test_put_log_events_in_the_future(minutes): conn.create_log_group(logGroupName=log_group_name) conn.create_log_stream(logGroupName=log_group_name, logStreamName=log_stream_name) - timestamp = int(unix_time_millis(datetime.utcnow() + timedelta(minutes=minutes))) + timestamp = int(unix_time_millis(utcnow() + timedelta(minutes=minutes))) messages = [{"message": "Message number {}", "timestamp": timestamp}] @@ -769,7 +769,7 @@ def test_get_log_events(): client.create_log_stream(logGroupName=log_group_name, logStreamName=log_stream_name) data = [ - (int(unix_time_millis(datetime.utcnow() + timedelta(milliseconds=x))), str(x)) + (int(unix_time_millis(utcnow() + timedelta(milliseconds=x))), str(x)) for x in range(20) ] events = [{"timestamp": x, "message": y} for x, y in data] @@ -862,7 +862,7 @@ def test_get_log_events_with_start_from_head(): client.create_log_stream(logGroupName=log_group_name, logStreamName=log_stream_name) data = [ - (int(unix_time_millis(datetime.utcnow() + timedelta(milliseconds=x))), str(x)) + (int(unix_time_millis(utcnow() + timedelta(milliseconds=x))), str(x)) for x in range(20) ] events = [{"timestamp": x, "message": y} for x, y in data] diff --git a/tests/test_logs/test_logs_filter.py b/tests/test_logs/test_logs_filter.py index 92722781d..60a416abf 100644 --- a/tests/test_logs/test_logs_filter.py +++ b/tests/test_logs/test_logs_filter.py @@ -1,9 +1,9 @@ import boto3 from unittest import TestCase -from datetime import timedelta, datetime +from datetime import timedelta from moto import mock_logs -from moto.core.utils import unix_time_millis +from moto.core.utils import unix_time_millis, utcnow TEST_REGION = "eu-west-1" @@ -47,8 +47,8 @@ class TestLogFilterParameters(TestLogFilter): def test_put_log_events_now(self): ts_1 = int(unix_time_millis()) - ts_2 = int(unix_time_millis(datetime.utcnow() + timedelta(minutes=5))) - ts_3 = int(unix_time_millis(datetime.utcnow() + timedelta(days=1))) + ts_2 = int(unix_time_millis(utcnow() + timedelta(minutes=5))) + ts_3 = int(unix_time_millis(utcnow() + timedelta(days=1))) messages = [ {"message": f"Message {idx}", "timestamp": ts} @@ -76,7 +76,7 @@ class TestLogFilterParameters(TestLogFilter): assert "Message 2" not in messages def test_filter_logs_paging(self): - timestamp = int(unix_time_millis(datetime.utcnow())) + timestamp = int(unix_time_millis()) messages = [] for i in range(25): messages.append({"message": f"Message number {i}", "timestamp": timestamp}) @@ -135,7 +135,7 @@ class TestLogFilterParameters(TestLogFilter): class TestLogsFilterPattern(TestLogFilter): def setUp(self) -> None: super().setUp() - now = int(unix_time_millis(datetime.utcnow())) + now = int(unix_time_millis()) messages = [ {"timestamp": now, "message": "hello"}, {"timestamp": now, "message": "world"}, diff --git a/tests/test_logs/test_logs_query/test_boto3.py b/tests/test_logs/test_logs_query/test_boto3.py index 20ddfdd10..32e73cad3 100644 --- a/tests/test_logs/test_logs_query/test_boto3.py +++ b/tests/test_logs/test_logs_query/test_boto3.py @@ -1,12 +1,12 @@ import time -from datetime import timedelta, datetime +from datetime import timedelta import boto3 import pytest from botocore.exceptions import ClientError from moto import mock_logs -from moto.core.utils import unix_time, unix_time_millis +from moto.core.utils import unix_time, unix_time_millis, utcnow @mock_logs @@ -52,7 +52,7 @@ def test_get_query_results(): data = [ ( - int(unix_time_millis(datetime.utcnow() - timedelta(minutes=x))), + int(unix_time_millis(utcnow() - timedelta(minutes=x))), f"event nr {x}", ) for x in range(5) @@ -65,8 +65,8 @@ def test_get_query_results(): query_id = client.start_query( logGroupName="test", - startTime=int(unix_time(datetime.utcnow() - timedelta(minutes=10))), - endTime=int(unix_time(datetime.utcnow() + timedelta(minutes=10))), + startTime=int(unix_time(utcnow() - timedelta(minutes=10))), + endTime=int(unix_time(utcnow() + timedelta(minutes=10))), queryString="fields @message", )["queryId"] @@ -94,8 +94,8 @@ def test_get_query_results(): # Only find events from last 2 minutes query_id = client.start_query( logGroupName="test", - startTime=int(unix_time(datetime.utcnow() - timedelta(minutes=2, seconds=1))), - endTime=int(unix_time(datetime.utcnow() - timedelta(seconds=1))), + startTime=int(unix_time(utcnow() - timedelta(minutes=2, seconds=1))), + endTime=int(unix_time(utcnow() - timedelta(seconds=1))), queryString="fields @message", )["queryId"] @@ -119,8 +119,8 @@ def test_describe_completed_query(): query_id = client.start_query( logGroupName="test", - startTime=int(unix_time(datetime.utcnow() + timedelta(minutes=10))), - endTime=int(unix_time(datetime.utcnow() + timedelta(minutes=10))), + startTime=int(unix_time(utcnow() + timedelta(minutes=10))), + endTime=int(unix_time(utcnow() + timedelta(minutes=10))), queryString="fields @message", )["queryId"] diff --git a/tests/test_s3/test_s3.py b/tests/test_s3/test_s3.py index 766f14de1..dc47c26f7 100644 --- a/tests/test_s3/test_s3.py +++ b/tests/test_s3/test_s3.py @@ -20,6 +20,7 @@ import requests from moto import settings, mock_s3, mock_config from moto.moto_api import state_manager +from moto.core.utils import utcnow from moto.s3.responses import DEFAULT_REGION_NAME import moto.s3.models as s3model @@ -1766,7 +1767,7 @@ def test_get_object_if_modified_since(): s3_client.get_object( Bucket=bucket_name, Key=key, - IfModifiedSince=datetime.datetime.utcnow() + datetime.timedelta(hours=1), + IfModifiedSince=utcnow() + datetime.timedelta(hours=1), ) err_value = err.value assert err_value.response["Error"] == {"Code": "304", "Message": "Not Modified"} @@ -1786,7 +1787,7 @@ def test_get_object_if_unmodified_since(): s3_client.get_object( Bucket=bucket_name, Key=key, - IfUnmodifiedSince=datetime.datetime.utcnow() - datetime.timedelta(hours=1), + IfUnmodifiedSince=utcnow() - datetime.timedelta(hours=1), ) err_value = err.value assert err_value.response["Error"]["Code"] == "PreconditionFailed" @@ -1840,7 +1841,7 @@ def test_head_object_if_modified_since(): s3_client.head_object( Bucket=bucket_name, Key=key, - IfModifiedSince=datetime.datetime.utcnow() + datetime.timedelta(hours=1), + IfModifiedSince=utcnow() + datetime.timedelta(hours=1), ) err_value = err.value assert err_value.response["Error"] == {"Code": "304", "Message": "Not Modified"} @@ -1882,7 +1883,7 @@ def test_head_object_if_unmodified_since(): s3_client.head_object( Bucket=bucket_name, Key=key, - IfUnmodifiedSince=datetime.datetime.utcnow() - datetime.timedelta(hours=1), + IfUnmodifiedSince=utcnow() - datetime.timedelta(hours=1), ) err_value = err.value assert err_value.response["Error"] == { diff --git a/tests/test_s3/test_s3_lock.py b/tests/test_s3/test_s3_lock.py index c7d9094f1..ef3338159 100644 --- a/tests/test_s3/test_s3_lock.py +++ b/tests/test_s3/test_s3_lock.py @@ -7,6 +7,7 @@ from botocore.client import ClientError import pytest from moto import mock_s3 +from moto.core.utils import utcnow from moto.s3.responses import DEFAULT_REGION_NAME @@ -20,7 +21,7 @@ def test_locked_object(): s3_client.create_bucket(Bucket=bucket_name, ObjectLockEnabledForBucket=True) - until = datetime.datetime.utcnow() + datetime.timedelta(0, seconds_lock) + until = utcnow() + datetime.timedelta(0, seconds_lock) s3_client.put_object( Bucket=bucket_name, Body=b"test", @@ -56,7 +57,7 @@ def test_fail_locked_object(): s3_client = boto3.client("s3", config=Config(region_name=DEFAULT_REGION_NAME)) s3_client.create_bucket(Bucket=bucket_name, ObjectLockEnabledForBucket=False) - until = datetime.datetime.utcnow() + datetime.timedelta(0, seconds_lock) + until = utcnow() + datetime.timedelta(0, seconds_lock) failed = False try: s3_client.put_object( @@ -88,7 +89,7 @@ def test_put_object_lock(): versions_response = s3_client.list_object_versions(Bucket=bucket_name) version_id = versions_response["Versions"][0]["VersionId"] - until = datetime.datetime.utcnow() + datetime.timedelta(0, seconds_lock) + until = utcnow() + datetime.timedelta(0, seconds_lock) s3_client.put_object_retention( Bucket=bucket_name, @@ -275,7 +276,7 @@ def test_put_object_lock_with_versions(): put_obj_2 = s3_client.put_object(Bucket=bucket_name, Body=b"test", Key=key_name) version_id_2 = put_obj_2["VersionId"] - until = datetime.datetime.utcnow() + datetime.timedelta(seconds=seconds_lock) + until = utcnow() + datetime.timedelta(seconds=seconds_lock) s3_client.put_object_retention( Bucket=bucket_name, diff --git a/tests/test_sqs/test_server.py b/tests/test_sqs/test_server.py index ec9fe1487..10551b6fb 100644 --- a/tests/test_sqs/test_server.py +++ b/tests/test_sqs/test_server.py @@ -1,10 +1,10 @@ """Test different server responses.""" -import datetime import re import threading import time import moto.server as server +from moto.core.utils import utcnow def test_sqs_list_identities(): @@ -85,12 +85,12 @@ def test_no_messages_polling_timeout(): test_client = backend.test_client() test_client.put(f"/?Action=CreateQueue&QueueName={queue_name}") wait_seconds = 5 - start = datetime.datetime.utcnow() + start = utcnow() test_client.get( f"/123/{queue_name}?Action=ReceiveMessage&" f"MaxNumberOfMessages=1&WaitTimeSeconds={wait_seconds}" ) - end = datetime.datetime.utcnow() + end = utcnow() duration = end - start assert duration.seconds >= wait_seconds assert duration.seconds <= wait_seconds + (wait_seconds / 2) diff --git a/tests/test_swf/responses/test_workflow_executions.py b/tests/test_swf/responses/test_workflow_executions.py index 21c0fbaca..2212de32b 100644 --- a/tests/test_swf/responses/test_workflow_executions.py +++ b/tests/test_swf/responses/test_workflow_executions.py @@ -1,11 +1,11 @@ -from datetime import datetime, timedelta +from datetime import timedelta import boto3 from botocore.exceptions import ClientError import pytest from moto import mock_swf -from moto.core.utils import unix_time +from moto.core.utils import unix_time, utcnow def setup_swf_environment_boto3(): @@ -244,7 +244,7 @@ def test_list_open_workflow_executions_boto3(): runId=run_id, ) - yesterday = datetime.utcnow() - timedelta(days=1) + yesterday = utcnow() - timedelta(days=1) oldest_date = unix_time(yesterday) response = client.list_open_workflow_executions( domain="test-domain", @@ -286,7 +286,7 @@ def test_list_closed_workflow_executions_boto3(): runId=run_id, ) - yesterday = datetime.utcnow() - timedelta(days=1) + yesterday = utcnow() - timedelta(days=1) oldest_date = unix_time(yesterday) response = client.list_closed_workflow_executions( domain="test-domain",