2
2
from datetime import datetime , timedelta , timezone
3
3
4
4
import pytest
5
+ from botocore .exceptions import ClientError
5
6
6
7
from localstack .testing .pytest import markers
7
8
from localstack .utils .strings import short_uid
@@ -272,7 +273,7 @@ def test_create_archive_error_duplicate(
272
273
EventPattern = json .dumps (TEST_EVENT_PATTERN ),
273
274
RetentionDays = 1 ,
274
275
)
275
- with pytest .raises (Exception ) as error :
276
+ with pytest .raises (ClientError ) as error :
276
277
aws_client .events .create_archive (
277
278
ArchiveName = archive_name ,
278
279
EventSourceArn = event_bus_arn ,
@@ -282,7 +283,7 @@ def test_create_archive_error_duplicate(
282
283
)
283
284
284
285
snapshot .add_transformer ([snapshot .transform .regex (archive_name , "<archive-name>" )])
285
- snapshot .match ("create-archive-duplicate-error" , error )
286
+ snapshot .match ("create-archive-duplicate-error" , error . value . response )
286
287
287
288
@markers .aws .validated
288
289
@pytest .mark .skipif (is_old_provider (), reason = "not supported by the old provider" )
@@ -291,7 +292,7 @@ def test_create_archive_error_unknown_event_bus(self, aws_client, snapshot):
291
292
non_existing_event_bus_arn = (
292
293
f"arn:aws:events:us-east-1:123456789012:event-bus/{ not_existing_event_bus_name } "
293
294
)
294
- with pytest .raises (Exception ) as error :
295
+ with pytest .raises (ClientError ) as error :
295
296
aws_client .events .create_archive (
296
297
ArchiveName = "test-archive" ,
297
298
EventSourceArn = non_existing_event_bus_arn ,
@@ -303,19 +304,19 @@ def test_create_archive_error_unknown_event_bus(self, aws_client, snapshot):
303
304
snapshot .add_transformer (
304
305
[snapshot .transform .regex (not_existing_event_bus_name , "<event-bus-name>" )]
305
306
)
306
- snapshot .match ("create-archive-unknown-event-bus-error" , error )
307
+ snapshot .match ("create-archive-unknown-event-bus-error" , error . value . response )
307
308
308
309
@markers .aws .validated
309
310
@pytest .mark .skipif (is_old_provider (), reason = "not supported by the old provider" )
310
311
def test_describe_archive_error_unknown_archive (self , aws_client , snapshot ):
311
312
not_existing_archive_name = f"doesnotexist-{ short_uid ()} "
312
- with pytest .raises (Exception ) as error :
313
+ with pytest .raises (ClientError ) as error :
313
314
aws_client .events .describe_archive (ArchiveName = not_existing_archive_name )
314
315
315
316
snapshot .add_transformer (
316
317
[snapshot .transform .regex (not_existing_archive_name , "<archive-name>" )]
317
318
)
318
- snapshot .match ("describe-archive-unknown-archive-error" , error )
319
+ snapshot .match ("describe-archive-unknown-archive-error" , error . value . response )
319
320
320
321
@markers .aws .validated
321
322
@pytest .mark .skipif (is_old_provider (), reason = "not supported by the old provider" )
@@ -326,37 +327,37 @@ def test_list_archive_error_unknown_source_arn(
326
327
non_existing_event_bus_arn = (
327
328
f"arn:aws:events:{ region_name } :{ account_id } :event-bus/{ not_existing_event_bus_name } "
328
329
)
329
- with pytest .raises (Exception ) as error :
330
+ with pytest .raises (ClientError ) as error :
330
331
aws_client .events .list_archives (EventSourceArn = non_existing_event_bus_arn )
331
332
332
333
snapshot .add_transformer (
333
334
[snapshot .transform .regex (not_existing_event_bus_name , "<event-bus-name>" )]
334
335
)
335
- snapshot .match ("list-archives-unknown-event-bus-error" , error )
336
+ snapshot .match ("list-archives-unknown-event-bus-error" , error . value . response )
336
337
337
338
@markers .aws .validated
338
339
@pytest .mark .skip (reason = "not possible to test with localstack" )
339
340
def test_update_archive_error_unknown_archive (self , aws_client , snapshot ):
340
341
not_existing_archive_name = f"doesnotexist-{ short_uid ()} "
341
- with pytest .raises (Exception ) as error :
342
+ with pytest .raises (ClientError ) as error :
342
343
aws_client .events .update_archive (ArchiveName = not_existing_archive_name )
343
344
344
345
snapshot .add_transformer (
345
346
[snapshot .transform .regex (not_existing_archive_name , "<archive-name>" )]
346
347
)
347
- snapshot .match ("update-archive-unknown-archive-error" , error )
348
+ snapshot .match ("update-archive-unknown-archive-error" , error . value . response )
348
349
349
350
@markers .aws .validated
350
351
@pytest .mark .skipif (is_old_provider (), reason = "not supported by the old provider" )
351
352
def test_delete_archive_error_unknown_archive (self , aws_client , snapshot ):
352
353
not_existing_archive_name = f"doesnotexist-{ short_uid ()} "
353
- with pytest .raises (Exception ) as error :
354
+ with pytest .raises (ClientError ) as error :
354
355
aws_client .events .delete_archive (ArchiveName = not_existing_archive_name )
355
356
356
357
snapshot .add_transformer (
357
358
[snapshot .transform .regex (not_existing_archive_name , "<archive-name>" )]
358
359
)
359
- snapshot .match ("delete-archive-unknown-archive-error" , error )
360
+ snapshot .match ("delete-archive-unknown-archive-error" , error . value . response )
360
361
361
362
362
363
class TestReplay :
@@ -669,7 +670,7 @@ def test_start_replay_error_unknown_event_bus(
669
670
end_time = datetime .now (timezone .utc )
670
671
671
672
replay_name = f"test-replay-{ short_uid ()} "
672
- with pytest .raises (Exception ) as error :
673
+ with pytest .raises (ClientError ) as error :
673
674
aws_client .events .start_replay (
674
675
ReplayName = replay_name ,
675
676
Description = "description of the replay" ,
@@ -684,11 +685,11 @@ def test_start_replay_error_unknown_event_bus(
684
685
snapshot .add_transformer (
685
686
[snapshot .transform .regex (not_existing_event_bus_name , "<event-bus-name>" )]
686
687
)
687
- snapshot .match ("start-replay-unknown-event-bus-error" , error )
688
+ snapshot .match ("start-replay-unknown-event-bus-error" , error . value . response )
688
689
689
690
event_bus_arn = events_create_event_bus (Name = not_existing_event_bus_name )["EventBusArn" ]
690
691
691
- with pytest .raises (Exception ) as error :
692
+ with pytest .raises (ClientError ) as error :
692
693
aws_client .events .start_replay (
693
694
ReplayName = replay_name ,
694
695
Description = "description of the replay" ,
@@ -700,7 +701,7 @@ def test_start_replay_error_unknown_event_bus(
700
701
}, # the destination must be the exact same event bus the archive is created for
701
702
)
702
703
703
- snapshot .match ("start-replay-wrong-event-bus-error" , error )
704
+ snapshot .match ("start-replay-wrong-event-bus-error" , error . value . response )
704
705
705
706
@markers .aws .validated
706
707
def test_start_replay_error_unknown_archive (
@@ -709,7 +710,7 @@ def test_start_replay_error_unknown_archive(
709
710
not_existing_archive_name = f"doesnotexist-{ short_uid ()} "
710
711
start_time = datetime .now (timezone .utc ) - timedelta (minutes = 1 )
711
712
end_time = datetime .now (timezone .utc )
712
- with pytest .raises (Exception ) as error :
713
+ with pytest .raises (ClientError ) as error :
713
714
aws_client .events .start_replay (
714
715
ReplayName = "test-replay" ,
715
716
Description = "description of the replay" ,
@@ -724,7 +725,7 @@ def test_start_replay_error_unknown_archive(
724
725
snapshot .add_transformer (
725
726
[snapshot .transform .regex (not_existing_archive_name , "<archive-name>" )]
726
727
)
727
- snapshot .match ("start-replay-unknown-archive-error" , error )
728
+ snapshot .match ("start-replay-unknown-archive-error" , error . value . response )
728
729
729
730
@markers .aws .validated
730
731
def test_start_replay_error_duplicate_name_same_archive (
@@ -751,7 +752,7 @@ def test_start_replay_error_duplicate_name_same_archive(
751
752
},
752
753
)
753
754
754
- with pytest .raises (Exception ) as error :
755
+ with pytest .raises (ClientError ) as error :
755
756
aws_client .events .start_replay (
756
757
ReplayName = replay_name ,
757
758
Description = "description of the replay" ,
@@ -764,7 +765,7 @@ def test_start_replay_error_duplicate_name_same_archive(
764
765
)
765
766
766
767
snapshot .add_transformer ([snapshot .transform .regex (replay_name , "<replay-name>" )])
767
- snapshot .match ("start-replay-duplicate-error" , error )
768
+ snapshot .match ("start-replay-duplicate-error" , error . value . response )
768
769
769
770
@markers .aws .validated
770
771
def test_start_replay_error_duplicate_different_archive (
@@ -800,7 +801,7 @@ def test_start_replay_error_duplicate_different_archive(
800
801
},
801
802
)
802
803
803
- with pytest .raises (Exception ) as error :
804
+ with pytest .raises (ClientError ) as error :
804
805
aws_client .events .start_replay (
805
806
ReplayName = replay_name ,
806
807
Description = "description of the replay" ,
@@ -813,7 +814,7 @@ def test_start_replay_error_duplicate_different_archive(
813
814
)
814
815
815
816
snapshot .add_transformer ([snapshot .transform .regex (replay_name , "<replay-name>" )])
816
- snapshot .match ("start-replay-duplicate-error" , error )
817
+ snapshot .match ("start-replay-duplicate-error" , error . value . response )
817
818
818
819
@markers .aws .validated
819
820
@pytest .mark .skipif (is_old_provider (), reason = "not supported by the old provider" )
@@ -833,7 +834,7 @@ def test_start_replay_error_invalid_end_time(
833
834
)
834
835
835
836
replay_name = f"test-replay-{ short_uid ()} "
836
- with pytest .raises (Exception ) as error :
837
+ with pytest .raises (ClientError ) as error :
837
838
aws_client .events .start_replay (
838
839
ReplayName = replay_name ,
839
840
Description = "description of the replay" ,
@@ -845,7 +846,7 @@ def test_start_replay_error_invalid_end_time(
845
846
},
846
847
)
847
848
848
- snapshot .match ("start-replay-invalid-end-time-error" , error )
849
+ snapshot .match ("start-replay-invalid-end-time-error" , error . value . response )
849
850
850
851
@markers .aws .validated
851
852
@pytest .mark .skip (reason = "currently no concurrency for replays in localstack" )
@@ -881,7 +882,7 @@ def tests_concurrency_error_too_many_active_replays(
881
882
)
882
883
883
884
# only 10 replays are allowed to be in state STARTING or RUNNING at the same time
884
- with pytest .raises (Exception ) as error :
885
+ with pytest .raises (ClientError ) as error :
885
886
replay_name = f"{ replay_name_prefix } -test-replay-{ num_replays } "
886
887
aws_client .events .start_replay (
887
888
ReplayName = replay_name ,
@@ -901,15 +902,15 @@ def tests_concurrency_error_too_many_active_replays(
901
902
snapshot .transform .jsonpath ("$..NextToken" , "next_token" ),
902
903
]
903
904
)
904
- snapshot .match ("list-replays-with-limit" , error )
905
+ snapshot .match ("list-replays-with-limit" , error . value . response )
905
906
906
907
@markers .aws .validated
907
908
def test_describe_replay_error_unknown_replay (self , aws_client , snapshot ):
908
909
not_existing_replay_name = f"doesnotexist-{ short_uid ()} "
909
- with pytest .raises (Exception ) as error :
910
+ with pytest .raises (ClientError ) as error :
910
911
aws_client .events .describe_replay (ReplayName = not_existing_replay_name )
911
912
912
913
snapshot .add_transformer (
913
914
[snapshot .transform .regex (not_existing_replay_name , "<replay-name>" )]
914
915
)
915
- snapshot .match ("describe-replay-unknown-replay-error" , error )
916
+ snapshot .match ("describe-replay-unknown-replay-error" , error . value . response )
0 commit comments