1import hashlib 2import json 3from datetime import datetime 4 5import pytest 6from freezegun import freeze_time 7import os 8from random import random 9 10import re 11import sure # noqa # pylint: disable=unused-import 12 13import boto3 14from botocore.exceptions import ClientError 15from dateutil.tz import tzlocal 16 17from moto import mock_ecr 18from unittest import SkipTest 19 20from moto.core import ACCOUNT_ID 21 22 23def _create_image_digest(contents=None): 24 if not contents: 25 contents = "docker_image{0}".format(int(random() * 10 ** 6)) 26 return "sha256:%s" % hashlib.sha256(contents.encode("utf-8")).hexdigest() 27 28 29def _create_image_manifest(): 30 return { 31 "schemaVersion": 2, 32 "mediaType": "application/vnd.docker.distribution.manifest.v2+json", 33 "config": { 34 "mediaType": "application/vnd.docker.container.image.v1+json", 35 "size": 7023, 36 "digest": _create_image_digest("config"), 37 }, 38 "layers": [ 39 { 40 "mediaType": "application/vnd.docker.image.rootfs.diff.tar.gzip", 41 "size": 32654, 42 "digest": _create_image_digest("layer1"), 43 }, 44 { 45 "mediaType": "application/vnd.docker.image.rootfs.diff.tar.gzip", 46 "size": 16724, 47 "digest": _create_image_digest("layer2"), 48 }, 49 { 50 "mediaType": "application/vnd.docker.image.rootfs.diff.tar.gzip", 51 "size": 73109, 52 # randomize image digest 53 "digest": _create_image_digest(), 54 }, 55 ], 56 } 57 58 59@mock_ecr 60def test_create_repository(): 61 # given 62 client = boto3.client("ecr", region_name="us-east-1") 63 repo_name = "test-repo" 64 65 # when 66 response = client.create_repository(repositoryName=repo_name) 67 68 # then 69 repo = response["repository"] 70 repo["repositoryName"].should.equal(repo_name) 71 repo["repositoryArn"].should.equal( 72 f"arn:aws:ecr:us-east-1:{ACCOUNT_ID}:repository/{repo_name}" 73 ) 74 repo["registryId"].should.equal(ACCOUNT_ID) 75 repo["repositoryUri"].should.equal( 76 f"{ACCOUNT_ID}.dkr.ecr.us-east-1.amazonaws.com/{repo_name}" 77 ) 78 repo["createdAt"].should.be.a(datetime) 79 repo["imageTagMutability"].should.equal("MUTABLE") 80 repo["imageScanningConfiguration"].should.equal({"scanOnPush": False}) 81 repo["encryptionConfiguration"].should.equal({"encryptionType": "AES256"}) 82 83 84@mock_ecr 85def test_create_repository_with_non_default_config(): 86 # given 87 region_name = "eu-central-1" 88 client = boto3.client("ecr", region_name=region_name) 89 repo_name = "test-repo" 90 kms_key = f"arn:aws:kms:{region_name}:{ACCOUNT_ID}:key/51d81fab-b138-4bd2-8a09-07fd6d37224d" 91 92 # when 93 response = client.create_repository( 94 repositoryName=repo_name, 95 imageTagMutability="IMMUTABLE", 96 imageScanningConfiguration={"scanOnPush": True}, 97 encryptionConfiguration={"encryptionType": "KMS", "kmsKey": kms_key}, 98 tags=[{"Key": "key-1", "Value": "value-1"}], 99 ) 100 101 # then 102 repo = response["repository"] 103 repo["repositoryName"].should.equal(repo_name) 104 repo["repositoryArn"].should.equal( 105 f"arn:aws:ecr:{region_name}:{ACCOUNT_ID}:repository/{repo_name}" 106 ) 107 repo["registryId"].should.equal(ACCOUNT_ID) 108 repo["repositoryUri"].should.equal( 109 f"{ACCOUNT_ID}.dkr.ecr.{region_name}.amazonaws.com/{repo_name}" 110 ) 111 repo["createdAt"].should.be.a(datetime) 112 repo["imageTagMutability"].should.equal("IMMUTABLE") 113 repo["imageScanningConfiguration"].should.equal({"scanOnPush": True}) 114 repo["encryptionConfiguration"].should.equal( 115 {"encryptionType": "KMS", "kmsKey": kms_key} 116 ) 117 118 119@mock_ecr 120def test_create_repository_with_aws_managed_kms(): 121 # given 122 region_name = "eu-central-1" 123 client = boto3.client("ecr", region_name=region_name) 124 repo_name = "test-repo" 125 126 # when 127 repo = client.create_repository( 128 repositoryName=repo_name, encryptionConfiguration={"encryptionType": "KMS"} 129 )["repository"] 130 131 # then 132 repo["repositoryName"].should.equal(repo_name) 133 repo["encryptionConfiguration"]["encryptionType"].should.equal("KMS") 134 repo["encryptionConfiguration"]["kmsKey"].should.match( 135 r"arn:aws:kms:eu-central-1:[0-9]{12}:key/[a-f0-9]{8}-[a-f0-9]{4}-[1-5][a-f0-9]{3}-[ab89][a-f0-9]{3}-[a-f0-9]{12}$" 136 ) 137 138 139@mock_ecr 140def test_create_repository_error_already_exists(): 141 # given 142 client = boto3.client("ecr", region_name="eu-central-1") 143 repo_name = "test-repo" 144 client.create_repository(repositoryName=repo_name) 145 146 # when 147 with pytest.raises(ClientError) as e: 148 client.create_repository(repositoryName=repo_name) 149 150 # then 151 ex = e.value 152 ex.operation_name.should.equal("CreateRepository") 153 ex.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400) 154 ex.response["Error"]["Code"].should.contain("RepositoryAlreadyExistsException") 155 ex.response["Error"]["Message"].should.equal( 156 f"The repository with name '{repo_name}' already exists " 157 f"in the registry with id '{ACCOUNT_ID}'" 158 ) 159 160 161@mock_ecr 162def test_describe_repositories(): 163 client = boto3.client("ecr", region_name="us-east-1") 164 _ = client.create_repository(repositoryName="test_repository1") 165 _ = client.create_repository(repositoryName="test_repository0") 166 response = client.describe_repositories() 167 len(response["repositories"]).should.equal(2) 168 169 repository_arns = [ 170 f"arn:aws:ecr:us-east-1:{ACCOUNT_ID}:repository/test_repository1", 171 f"arn:aws:ecr:us-east-1:{ACCOUNT_ID}:repository/test_repository0", 172 ] 173 sorted( 174 [ 175 response["repositories"][0]["repositoryArn"], 176 response["repositories"][1]["repositoryArn"], 177 ] 178 ).should.equal(sorted(repository_arns)) 179 180 repository_uris = [ 181 f"{ACCOUNT_ID}.dkr.ecr.us-east-1.amazonaws.com/test_repository1", 182 f"{ACCOUNT_ID}.dkr.ecr.us-east-1.amazonaws.com/test_repository0", 183 ] 184 sorted( 185 [ 186 response["repositories"][0]["repositoryUri"], 187 response["repositories"][1]["repositoryUri"], 188 ] 189 ).should.equal(sorted(repository_uris)) 190 191 192@mock_ecr 193def test_describe_repositories_1(): 194 client = boto3.client("ecr", region_name="us-east-1") 195 _ = client.create_repository(repositoryName="test_repository1") 196 _ = client.create_repository(repositoryName="test_repository0") 197 response = client.describe_repositories(registryId=ACCOUNT_ID) 198 len(response["repositories"]).should.equal(2) 199 200 repository_arns = [ 201 f"arn:aws:ecr:us-east-1:{ACCOUNT_ID}:repository/test_repository1", 202 f"arn:aws:ecr:us-east-1:{ACCOUNT_ID}:repository/test_repository0", 203 ] 204 sorted( 205 [ 206 response["repositories"][0]["repositoryArn"], 207 response["repositories"][1]["repositoryArn"], 208 ] 209 ).should.equal(sorted(repository_arns)) 210 211 repository_uris = [ 212 f"{ACCOUNT_ID}.dkr.ecr.us-east-1.amazonaws.com/test_repository1", 213 f"{ACCOUNT_ID}.dkr.ecr.us-east-1.amazonaws.com/test_repository0", 214 ] 215 sorted( 216 [ 217 response["repositories"][0]["repositoryUri"], 218 response["repositories"][1]["repositoryUri"], 219 ] 220 ).should.equal(sorted(repository_uris)) 221 222 223@mock_ecr 224def test_describe_repositories_2(): 225 client = boto3.client("ecr", region_name="us-east-1") 226 _ = client.create_repository(repositoryName="test_repository1") 227 _ = client.create_repository(repositoryName="test_repository0") 228 response = client.describe_repositories(registryId="109876543210") 229 len(response["repositories"]).should.equal(0) 230 231 232@mock_ecr 233def test_describe_repositories_3(): 234 client = boto3.client("ecr", region_name="us-east-1") 235 _ = client.create_repository(repositoryName="test_repository1") 236 _ = client.create_repository(repositoryName="test_repository0") 237 response = client.describe_repositories(repositoryNames=["test_repository1"]) 238 len(response["repositories"]).should.equal(1) 239 repository_arn = f"arn:aws:ecr:us-east-1:{ACCOUNT_ID}:repository/test_repository1" 240 response["repositories"][0]["repositoryArn"].should.equal(repository_arn) 241 242 repository_uri = f"{ACCOUNT_ID}.dkr.ecr.us-east-1.amazonaws.com/test_repository1" 243 response["repositories"][0]["repositoryUri"].should.equal(repository_uri) 244 245 246@mock_ecr 247def test_describe_repositories_with_image(): 248 # given 249 client = boto3.client("ecr", region_name="us-east-1") 250 repo_name = "test-repo" 251 client.create_repository(repositoryName=repo_name) 252 client.put_image( 253 repositoryName=repo_name, 254 imageManifest=json.dumps(_create_image_manifest()), 255 imageTag="latest", 256 ) 257 258 # when 259 response = client.describe_repositories(repositoryNames=[repo_name]) 260 261 # then 262 response["repositories"].should.have.length_of(1) 263 264 repo = response["repositories"][0] 265 repo["registryId"].should.equal(ACCOUNT_ID) 266 repo["repositoryArn"].should.equal( 267 f"arn:aws:ecr:us-east-1:{ACCOUNT_ID}:repository/{repo_name}" 268 ) 269 repo["repositoryName"].should.equal(repo_name) 270 repo["repositoryUri"].should.equal( 271 f"{ACCOUNT_ID}.dkr.ecr.us-east-1.amazonaws.com/{repo_name}" 272 ) 273 repo["createdAt"].should.be.a(datetime) 274 repo["imageScanningConfiguration"].should.equal({"scanOnPush": False}) 275 repo["imageTagMutability"].should.equal("MUTABLE") 276 repo["encryptionConfiguration"].should.equal({"encryptionType": "AES256"}) 277 278 279@mock_ecr 280def test_delete_repository(): 281 # given 282 client = boto3.client("ecr", region_name="us-east-1") 283 repo_name = "test-repo" 284 client.create_repository(repositoryName=repo_name) 285 286 # when 287 response = client.delete_repository(repositoryName=repo_name) 288 289 # then 290 repo = response["repository"] 291 repo["repositoryName"].should.equal(repo_name) 292 repo["repositoryArn"].should.equal( 293 f"arn:aws:ecr:us-east-1:{ACCOUNT_ID}:repository/{repo_name}" 294 ) 295 repo["registryId"].should.equal(ACCOUNT_ID) 296 repo["repositoryUri"].should.equal( 297 f"{ACCOUNT_ID}.dkr.ecr.us-east-1.amazonaws.com/{repo_name}" 298 ) 299 repo["createdAt"].should.be.a(datetime) 300 repo["imageTagMutability"].should.equal("MUTABLE") 301 302 response = client.describe_repositories() 303 response["repositories"].should.have.length_of(0) 304 305 306@mock_ecr 307def test_delete_repository_with_force(): 308 client = boto3.client("ecr", region_name="us-east-1") 309 repo_name = "test-repo" 310 client.create_repository(repositoryName=repo_name) 311 client.put_image( 312 repositoryName=repo_name, 313 imageManifest=json.dumps(_create_image_manifest()), 314 imageTag="latest", 315 ) 316 317 # when 318 # when 319 response = client.delete_repository(repositoryName=repo_name, force=True) 320 321 # then 322 repo = response["repository"] 323 repo["repositoryName"].should.equal(repo_name) 324 repo["repositoryArn"].should.equal( 325 f"arn:aws:ecr:us-east-1:{ACCOUNT_ID}:repository/{repo_name}" 326 ) 327 repo["registryId"].should.equal(ACCOUNT_ID) 328 repo["repositoryUri"].should.equal( 329 f"{ACCOUNT_ID}.dkr.ecr.us-east-1.amazonaws.com/{repo_name}" 330 ) 331 repo["createdAt"].should.be.a(datetime) 332 repo["imageTagMutability"].should.equal("MUTABLE") 333 334 response = client.describe_repositories() 335 response["repositories"].should.have.length_of(0) 336 337 338@mock_ecr 339def test_put_image(): 340 client = boto3.client("ecr", region_name="us-east-1") 341 _ = client.create_repository(repositoryName="test_repository") 342 343 response = client.put_image( 344 repositoryName="test_repository", 345 imageManifest=json.dumps(_create_image_manifest()), 346 imageTag="latest", 347 ) 348 349 response["image"]["imageId"]["imageTag"].should.equal("latest") 350 response["image"]["imageId"]["imageDigest"].should.contain("sha") 351 response["image"]["repositoryName"].should.equal("test_repository") 352 response["image"]["registryId"].should.equal(ACCOUNT_ID) 353 354 355@mock_ecr 356def test_put_image_with_push_date(): 357 if os.environ.get("TEST_SERVER_MODE", "false").lower() == "true": 358 raise SkipTest("Cant manipulate time in server mode") 359 360 client = boto3.client("ecr", region_name="us-east-1") 361 _ = client.create_repository(repositoryName="test_repository") 362 363 with freeze_time("2018-08-28 00:00:00"): 364 image1_date = datetime.now(tzlocal()) 365 _ = client.put_image( 366 repositoryName="test_repository", 367 imageManifest=json.dumps(_create_image_manifest()), 368 imageTag="latest", 369 ) 370 371 with freeze_time("2019-05-31 00:00:00"): 372 image2_date = datetime.now(tzlocal()) 373 _ = client.put_image( 374 repositoryName="test_repository", 375 imageManifest=json.dumps(_create_image_manifest()), 376 imageTag="latest", 377 ) 378 379 describe_response = client.describe_images(repositoryName="test_repository") 380 381 type(describe_response["imageDetails"]).should.be(list) 382 len(describe_response["imageDetails"]).should.be(2) 383 384 set( 385 [ 386 describe_response["imageDetails"][0]["imagePushedAt"], 387 describe_response["imageDetails"][1]["imagePushedAt"], 388 ] 389 ).should.equal(set([image1_date, image2_date])) 390 391 392@mock_ecr 393def test_put_image_with_multiple_tags(): 394 client = boto3.client("ecr", region_name="us-east-1") 395 _ = client.create_repository(repositoryName="test_repository") 396 manifest = _create_image_manifest() 397 response = client.put_image( 398 repositoryName="test_repository", 399 imageManifest=json.dumps(manifest), 400 imageTag="v1", 401 ) 402 403 response["image"]["imageId"]["imageTag"].should.equal("v1") 404 response["image"]["imageId"]["imageDigest"].should.contain("sha") 405 response["image"]["repositoryName"].should.equal("test_repository") 406 response["image"]["registryId"].should.equal(ACCOUNT_ID) 407 408 response1 = client.put_image( 409 repositoryName="test_repository", 410 imageManifest=json.dumps(manifest), 411 imageTag="latest", 412 ) 413 414 response1["image"]["imageId"]["imageTag"].should.equal("latest") 415 response1["image"]["imageId"]["imageDigest"].should.contain("sha") 416 response1["image"]["repositoryName"].should.equal("test_repository") 417 response1["image"]["registryId"].should.equal(ACCOUNT_ID) 418 419 response2 = client.describe_images(repositoryName="test_repository") 420 type(response2["imageDetails"]).should.be(list) 421 len(response2["imageDetails"]).should.be(1) 422 423 response2["imageDetails"][0]["imageDigest"].should.contain("sha") 424 425 response2["imageDetails"][0]["registryId"].should.equal(ACCOUNT_ID) 426 427 response2["imageDetails"][0]["repositoryName"].should.equal("test_repository") 428 429 len(response2["imageDetails"][0]["imageTags"]).should.be(2) 430 response2["imageDetails"][0]["imageTags"].should.be.equal(["v1", "latest"]) 431 432 433@mock_ecr 434def test_list_images(): 435 client = boto3.client("ecr", region_name="us-east-1") 436 _ = client.create_repository(repositoryName="test_repository_1") 437 438 _ = client.create_repository(repositoryName="test_repository_2") 439 440 _ = client.put_image( 441 repositoryName="test_repository_1", 442 imageManifest=json.dumps(_create_image_manifest()), 443 imageTag="latest", 444 ) 445 446 _ = client.put_image( 447 repositoryName="test_repository_1", 448 imageManifest=json.dumps(_create_image_manifest()), 449 imageTag="v1", 450 ) 451 452 _ = client.put_image( 453 repositoryName="test_repository_1", 454 imageManifest=json.dumps(_create_image_manifest()), 455 imageTag="v2", 456 ) 457 458 _ = client.put_image( 459 repositoryName="test_repository_2", 460 imageManifest=json.dumps(_create_image_manifest()), 461 imageTag="oldest", 462 ) 463 464 response = client.list_images(repositoryName="test_repository_1") 465 type(response["imageIds"]).should.be(list) 466 len(response["imageIds"]).should.be(3) 467 468 for image in response["imageIds"]: 469 image["imageDigest"].should.contain("sha") 470 471 image_tags = ["latest", "v1", "v2"] 472 set( 473 [ 474 response["imageIds"][0]["imageTag"], 475 response["imageIds"][1]["imageTag"], 476 response["imageIds"][2]["imageTag"], 477 ] 478 ).should.equal(set(image_tags)) 479 480 response = client.list_images(repositoryName="test_repository_2") 481 type(response["imageIds"]).should.be(list) 482 len(response["imageIds"]).should.be(1) 483 response["imageIds"][0]["imageTag"].should.equal("oldest") 484 response["imageIds"][0]["imageDigest"].should.contain("sha") 485 486 487@mock_ecr 488def test_list_images_from_repository_that_doesnt_exist(): 489 client = boto3.client("ecr", region_name="us-east-1") 490 _ = client.create_repository(repositoryName="test_repository_1") 491 492 # non existing repo 493 error_msg = re.compile( 494 r".*The repository with name 'repo-that-doesnt-exist' does not exist in the registry with id '123'.*", 495 re.MULTILINE, 496 ) 497 client.list_images.when.called_with( 498 repositoryName="repo-that-doesnt-exist", registryId="123" 499 ).should.throw(Exception, error_msg) 500 501 # repo does not exist in specified registry 502 error_msg = re.compile( 503 r".*The repository with name 'test_repository_1' does not exist in the registry with id '222'.*", 504 re.MULTILINE, 505 ) 506 client.list_images.when.called_with( 507 repositoryName="test_repository_1", registryId="222" 508 ).should.throw(Exception, error_msg) 509 510 511@mock_ecr 512def test_describe_images(): 513 client = boto3.client("ecr", region_name="us-east-1") 514 _ = client.create_repository(repositoryName="test_repository") 515 516 _ = client.put_image( 517 repositoryName="test_repository", 518 imageManifest=json.dumps(_create_image_manifest()), 519 ) 520 521 _ = client.put_image( 522 repositoryName="test_repository", 523 imageManifest=json.dumps(_create_image_manifest()), 524 imageTag="latest", 525 ) 526 527 _ = client.put_image( 528 repositoryName="test_repository", 529 imageManifest=json.dumps(_create_image_manifest()), 530 imageTag="v1", 531 ) 532 533 _ = client.put_image( 534 repositoryName="test_repository", 535 imageManifest=json.dumps(_create_image_manifest()), 536 imageTag="v2", 537 ) 538 539 response = client.describe_images(repositoryName="test_repository") 540 type(response["imageDetails"]).should.be(list) 541 len(response["imageDetails"]).should.be(4) 542 543 response["imageDetails"][0]["imageDigest"].should.contain("sha") 544 response["imageDetails"][1]["imageDigest"].should.contain("sha") 545 response["imageDetails"][2]["imageDigest"].should.contain("sha") 546 response["imageDetails"][3]["imageDigest"].should.contain("sha") 547 548 response["imageDetails"][0]["registryId"].should.equal(ACCOUNT_ID) 549 response["imageDetails"][1]["registryId"].should.equal(ACCOUNT_ID) 550 response["imageDetails"][2]["registryId"].should.equal(ACCOUNT_ID) 551 response["imageDetails"][3]["registryId"].should.equal(ACCOUNT_ID) 552 553 response["imageDetails"][0]["repositoryName"].should.equal("test_repository") 554 response["imageDetails"][1]["repositoryName"].should.equal("test_repository") 555 response["imageDetails"][2]["repositoryName"].should.equal("test_repository") 556 response["imageDetails"][3]["repositoryName"].should.equal("test_repository") 557 558 response["imageDetails"][0].should_not.have.key("imageTags") 559 len(response["imageDetails"][1]["imageTags"]).should.be(1) 560 len(response["imageDetails"][2]["imageTags"]).should.be(1) 561 len(response["imageDetails"][3]["imageTags"]).should.be(1) 562 563 image_tags = ["latest", "v1", "v2"] 564 set( 565 [ 566 response["imageDetails"][1]["imageTags"][0], 567 response["imageDetails"][2]["imageTags"][0], 568 response["imageDetails"][3]["imageTags"][0], 569 ] 570 ).should.equal(set(image_tags)) 571 572 response["imageDetails"][0]["imageSizeInBytes"].should.equal(52428800) 573 response["imageDetails"][1]["imageSizeInBytes"].should.equal(52428800) 574 response["imageDetails"][2]["imageSizeInBytes"].should.equal(52428800) 575 response["imageDetails"][3]["imageSizeInBytes"].should.equal(52428800) 576 577 578@mock_ecr 579def test_describe_images_by_tag(): 580 client = boto3.client("ecr", region_name="us-east-1") 581 _ = client.create_repository(repositoryName="test_repository") 582 583 tag_map = {} 584 for tag in ["latest", "v1", "v2"]: 585 put_response = client.put_image( 586 repositoryName="test_repository", 587 imageManifest=json.dumps(_create_image_manifest()), 588 imageTag=tag, 589 ) 590 tag_map[tag] = put_response["image"] 591 592 for tag, put_response in tag_map.items(): 593 response = client.describe_images( 594 repositoryName="test_repository", imageIds=[{"imageTag": tag}] 595 ) 596 len(response["imageDetails"]).should.be(1) 597 image_detail = response["imageDetails"][0] 598 image_detail["registryId"].should.equal(ACCOUNT_ID) 599 image_detail["repositoryName"].should.equal("test_repository") 600 image_detail["imageTags"].should.equal([put_response["imageId"]["imageTag"]]) 601 image_detail["imageDigest"].should.equal(put_response["imageId"]["imageDigest"]) 602 603 604@mock_ecr 605def test_describe_images_tags_should_not_contain_empty_tag1(): 606 client = boto3.client("ecr", region_name="us-east-1") 607 _ = client.create_repository(repositoryName="test_repository") 608 609 manifest = _create_image_manifest() 610 client.put_image( 611 repositoryName="test_repository", imageManifest=json.dumps(manifest) 612 ) 613 614 tags = ["v1", "v2", "latest"] 615 for tag in tags: 616 client.put_image( 617 repositoryName="test_repository", 618 imageManifest=json.dumps(manifest), 619 imageTag=tag, 620 ) 621 622 response = client.describe_images( 623 repositoryName="test_repository", imageIds=[{"imageTag": tag}] 624 ) 625 len(response["imageDetails"]).should.be(1) 626 image_detail = response["imageDetails"][0] 627 len(image_detail["imageTags"]).should.equal(3) 628 image_detail["imageTags"].should.be.equal(tags) 629 630 631@mock_ecr 632def test_describe_images_tags_should_not_contain_empty_tag2(): 633 client = boto3.client("ecr", region_name="us-east-1") 634 _ = client.create_repository(repositoryName="test_repository") 635 636 manifest = _create_image_manifest() 637 tags = ["v1", "v2"] 638 for tag in tags: 639 client.put_image( 640 repositoryName="test_repository", 641 imageManifest=json.dumps(manifest), 642 imageTag=tag, 643 ) 644 645 client.put_image( 646 repositoryName="test_repository", imageManifest=json.dumps(manifest) 647 ) 648 649 client.put_image( 650 repositoryName="test_repository", 651 imageManifest=json.dumps(manifest), 652 imageTag="latest", 653 ) 654 655 response = client.describe_images( 656 repositoryName="test_repository", imageIds=[{"imageTag": tag}] 657 ) 658 len(response["imageDetails"]).should.be(1) 659 image_detail = response["imageDetails"][0] 660 len(image_detail["imageTags"]).should.equal(3) 661 image_detail["imageTags"].should.be.equal(["v1", "v2", "latest"]) 662 663 664@mock_ecr 665def test_describe_repository_that_doesnt_exist(): 666 client = boto3.client("ecr", region_name="us-east-1") 667 668 error_msg = re.compile( 669 r".*The repository with name 'repo-that-doesnt-exist' does not exist in the registry with id '123'.*", 670 re.MULTILINE, 671 ) 672 client.describe_repositories.when.called_with( 673 repositoryNames=["repo-that-doesnt-exist"], registryId="123" 674 ).should.throw(ClientError, error_msg) 675 676 677@mock_ecr 678def test_describe_image_that_doesnt_exist(): 679 client = boto3.client("ecr", region_name="us-east-1") 680 client.create_repository(repositoryName="test_repository") 681 682 error_msg1 = re.compile( 683 r".*The image with imageId {imageDigest:'null', imageTag:'testtag'} does not exist within " 684 r"the repository with name 'test_repository' in the registry with id '123456789012'.*", 685 re.MULTILINE, 686 ) 687 688 client.describe_images.when.called_with( 689 repositoryName="test_repository", 690 imageIds=[{"imageTag": "testtag"}], 691 registryId=ACCOUNT_ID, 692 ).should.throw(client.exceptions.ImageNotFoundException, error_msg1) 693 694 error_msg2 = re.compile( 695 r".*The repository with name 'repo-that-doesnt-exist' does not exist in the registry with id '123456789012'.*", 696 re.MULTILINE, 697 ) 698 client.describe_images.when.called_with( 699 repositoryName="repo-that-doesnt-exist", 700 imageIds=[{"imageTag": "testtag"}], 701 registryId=ACCOUNT_ID, 702 ).should.throw(ClientError, error_msg2) 703 704 705@mock_ecr 706def test_delete_repository_that_doesnt_exist(): 707 client = boto3.client("ecr", region_name="us-east-1") 708 repo_name = "repo-that-doesnt-exist" 709 710 # when 711 with pytest.raises(ClientError) as e: 712 client.delete_repository(repositoryName=repo_name) 713 714 # then 715 ex = e.value 716 ex.operation_name.should.equal("DeleteRepository") 717 ex.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400) 718 ex.response["Error"]["Code"].should.contain("RepositoryNotFoundException") 719 ex.response["Error"]["Message"].should.equal( 720 f"The repository with name '{repo_name}' does not exist " 721 f"in the registry with id '{ACCOUNT_ID}'" 722 ) 723 724 725@mock_ecr 726def test_delete_repository_error_not_empty(): 727 client = boto3.client("ecr", region_name="us-east-1") 728 repo_name = "test-repo" 729 client.create_repository(repositoryName=repo_name) 730 client.put_image( 731 repositoryName=repo_name, 732 imageManifest=json.dumps(_create_image_manifest()), 733 imageTag="latest", 734 ) 735 736 # when 737 with pytest.raises(ClientError) as e: 738 client.delete_repository(repositoryName=repo_name) 739 740 # then 741 ex = e.value 742 ex.operation_name.should.equal("DeleteRepository") 743 ex.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400) 744 ex.response["Error"]["Code"].should.contain("RepositoryNotEmptyException") 745 ex.response["Error"]["Message"].should.equal( 746 f"The repository with name '{repo_name}' " 747 f"in registry with id '{ACCOUNT_ID}' " 748 "cannot be deleted because it still contains images" 749 ) 750 751 752@mock_ecr 753def test_describe_images_by_digest(): 754 client = boto3.client("ecr", region_name="us-east-1") 755 _ = client.create_repository(repositoryName="test_repository") 756 757 tags = ["latest", "v1", "v2"] 758 digest_map = {} 759 for tag in tags: 760 put_response = client.put_image( 761 repositoryName="test_repository", 762 imageManifest=json.dumps(_create_image_manifest()), 763 imageTag=tag, 764 ) 765 digest_map[put_response["image"]["imageId"]["imageDigest"]] = put_response[ 766 "image" 767 ] 768 769 for digest, put_response in digest_map.items(): 770 response = client.describe_images( 771 repositoryName="test_repository", imageIds=[{"imageDigest": digest}] 772 ) 773 len(response["imageDetails"]).should.be(1) 774 image_detail = response["imageDetails"][0] 775 image_detail["registryId"].should.equal(ACCOUNT_ID) 776 image_detail["repositoryName"].should.equal("test_repository") 777 image_detail["imageTags"].should.equal([put_response["imageId"]["imageTag"]]) 778 image_detail["imageDigest"].should.equal(digest) 779 780 781@mock_ecr 782def test_get_authorization_token_assume_region(): 783 client = boto3.client("ecr", region_name="us-east-1") 784 auth_token_response = client.get_authorization_token() 785 786 auth_token_response.should.contain("authorizationData") 787 auth_token_response.should.contain("ResponseMetadata") 788 auth_token_response["authorizationData"].should.equal( 789 [ 790 { 791 "authorizationToken": "QVdTOjEyMzQ1Njc4OTAxMi1hdXRoLXRva2Vu", 792 "proxyEndpoint": f"https://{ACCOUNT_ID}.dkr.ecr.us-east-1.amazonaws.com", 793 "expiresAt": datetime(2015, 1, 1, tzinfo=tzlocal()), 794 } 795 ] 796 ) 797 798 799@mock_ecr 800def test_get_authorization_token_explicit_regions(): 801 client = boto3.client("ecr", region_name="us-east-1") 802 auth_token_response = client.get_authorization_token( 803 registryIds=["10987654321", "878787878787"] 804 ) 805 806 auth_token_response.should.contain("authorizationData") 807 auth_token_response.should.contain("ResponseMetadata") 808 auth_token_response["authorizationData"].should.equal( 809 [ 810 { 811 "authorizationToken": "QVdTOjEwOTg3NjU0MzIxLWF1dGgtdG9rZW4=", 812 "proxyEndpoint": "https://10987654321.dkr.ecr.us-east-1.amazonaws.com", 813 "expiresAt": datetime(2015, 1, 1, tzinfo=tzlocal()), 814 }, 815 { 816 "authorizationToken": "QVdTOjg3ODc4Nzg3ODc4Ny1hdXRoLXRva2Vu", 817 "proxyEndpoint": "https://878787878787.dkr.ecr.us-east-1.amazonaws.com", 818 "expiresAt": datetime(2015, 1, 1, tzinfo=tzlocal()), 819 }, 820 ] 821 ) 822 823 824@mock_ecr 825def test_batch_get_image(): 826 client = boto3.client("ecr", region_name="us-east-1") 827 _ = client.create_repository(repositoryName="test_repository") 828 829 _ = client.put_image( 830 repositoryName="test_repository", 831 imageManifest=json.dumps(_create_image_manifest()), 832 imageTag="latest", 833 ) 834 835 _ = client.put_image( 836 repositoryName="test_repository", 837 imageManifest=json.dumps(_create_image_manifest()), 838 imageTag="v1", 839 ) 840 841 _ = client.put_image( 842 repositoryName="test_repository", 843 imageManifest=json.dumps(_create_image_manifest()), 844 imageTag="v2", 845 ) 846 847 response = client.batch_get_image( 848 repositoryName="test_repository", imageIds=[{"imageTag": "v2"}] 849 ) 850 851 type(response["images"]).should.be(list) 852 len(response["images"]).should.be(1) 853 854 response["images"][0]["imageManifest"].should.contain( 855 "vnd.docker.distribution.manifest.v2+json" 856 ) 857 response["images"][0]["registryId"].should.equal(ACCOUNT_ID) 858 response["images"][0]["repositoryName"].should.equal("test_repository") 859 860 response["images"][0]["imageId"]["imageTag"].should.equal("v2") 861 response["images"][0]["imageId"]["imageDigest"].should.contain("sha") 862 863 type(response["failures"]).should.be(list) 864 len(response["failures"]).should.be(0) 865 866 867@mock_ecr 868def test_batch_get_image_that_doesnt_exist(): 869 client = boto3.client("ecr", region_name="us-east-1") 870 _ = client.create_repository(repositoryName="test_repository") 871 872 _ = client.put_image( 873 repositoryName="test_repository", 874 imageManifest=json.dumps(_create_image_manifest()), 875 imageTag="latest", 876 ) 877 878 _ = client.put_image( 879 repositoryName="test_repository", 880 imageManifest=json.dumps(_create_image_manifest()), 881 imageTag="v1", 882 ) 883 884 _ = client.put_image( 885 repositoryName="test_repository", 886 imageManifest=json.dumps(_create_image_manifest()), 887 imageTag="v2", 888 ) 889 890 response = client.batch_get_image( 891 repositoryName="test_repository", imageIds=[{"imageTag": "v5"}] 892 ) 893 894 type(response["images"]).should.be(list) 895 len(response["images"]).should.be(0) 896 897 type(response["failures"]).should.be(list) 898 len(response["failures"]).should.be(1) 899 response["failures"][0]["failureReason"].should.equal("Requested image not found") 900 response["failures"][0]["failureCode"].should.equal("ImageNotFound") 901 response["failures"][0]["imageId"]["imageTag"].should.equal("v5") 902 903 904@mock_ecr 905def test_batch_delete_image_by_tag(): 906 client = boto3.client("ecr", region_name="us-east-1") 907 client.create_repository(repositoryName="test_repository") 908 909 manifest = _create_image_manifest() 910 911 tags = ["v1", "v1.0", "latest"] 912 for tag in tags: 913 client.put_image( 914 repositoryName="test_repository", 915 imageManifest=json.dumps(manifest), 916 imageTag=tag, 917 ) 918 919 describe_response1 = client.describe_images(repositoryName="test_repository") 920 921 batch_delete_response = client.batch_delete_image( 922 registryId="012345678910", 923 repositoryName="test_repository", 924 imageIds=[{"imageTag": "latest"}], 925 ) 926 927 describe_response2 = client.describe_images(repositoryName="test_repository") 928 929 type(describe_response1["imageDetails"][0]["imageTags"]).should.be(list) 930 len(describe_response1["imageDetails"][0]["imageTags"]).should.be(3) 931 932 type(describe_response2["imageDetails"][0]["imageTags"]).should.be(list) 933 len(describe_response2["imageDetails"][0]["imageTags"]).should.be(2) 934 935 type(batch_delete_response["imageIds"]).should.be(list) 936 len(batch_delete_response["imageIds"]).should.be(1) 937 938 batch_delete_response["imageIds"][0]["imageTag"].should.equal("latest") 939 940 type(batch_delete_response["failures"]).should.be(list) 941 len(batch_delete_response["failures"]).should.be(0) 942 943 944@mock_ecr 945def test_batch_delete_image_delete_last_tag(): 946 client = boto3.client("ecr", region_name="us-east-1") 947 client.create_repository(repositoryName="test_repository") 948 949 client.put_image( 950 repositoryName="test_repository", 951 imageManifest=json.dumps(_create_image_manifest()), 952 imageTag="v1", 953 ) 954 955 describe_response1 = client.describe_images(repositoryName="test_repository") 956 957 batch_delete_response = client.batch_delete_image( 958 registryId="012345678910", 959 repositoryName="test_repository", 960 imageIds=[{"imageTag": "v1"}], 961 ) 962 963 describe_response2 = client.describe_images(repositoryName="test_repository") 964 965 type(describe_response1["imageDetails"][0]["imageTags"]).should.be(list) 966 len(describe_response1["imageDetails"][0]["imageTags"]).should.be(1) 967 968 type(describe_response2["imageDetails"]).should.be(list) 969 len(describe_response2["imageDetails"]).should.be(0) 970 971 type(batch_delete_response["imageIds"]).should.be(list) 972 len(batch_delete_response["imageIds"]).should.be(1) 973 974 batch_delete_response["imageIds"][0]["imageTag"].should.equal("v1") 975 976 type(batch_delete_response["failures"]).should.be(list) 977 len(batch_delete_response["failures"]).should.be(0) 978 979 980@mock_ecr 981def test_batch_delete_image_with_nonexistent_tag(): 982 client = boto3.client("ecr", region_name="us-east-1") 983 client.create_repository(repositoryName="test_repository") 984 985 manifest = _create_image_manifest() 986 987 tags = ["v1", "v1.0", "latest"] 988 for tag in tags: 989 client.put_image( 990 repositoryName="test_repository", 991 imageManifest=json.dumps(manifest), 992 imageTag=tag, 993 ) 994 995 describe_response = client.describe_images(repositoryName="test_repository") 996 997 missing_tag = "missing-tag" 998 batch_delete_response = client.batch_delete_image( 999 registryId="012345678910", 1000 repositoryName="test_repository", 1001 imageIds=[{"imageTag": missing_tag}], 1002 ) 1003 1004 type(describe_response["imageDetails"][0]["imageTags"]).should.be(list) 1005 len(describe_response["imageDetails"][0]["imageTags"]).should.be(3) 1006 1007 type(batch_delete_response["imageIds"]).should.be(list) 1008 len(batch_delete_response["imageIds"]).should.be(0) 1009 1010 batch_delete_response["failures"][0]["imageId"]["imageTag"].should.equal( 1011 missing_tag 1012 ) 1013 batch_delete_response["failures"][0]["failureCode"].should.equal("ImageNotFound") 1014 batch_delete_response["failures"][0]["failureReason"].should.equal( 1015 "Requested image not found" 1016 ) 1017 1018 type(batch_delete_response["failures"]).should.be(list) 1019 len(batch_delete_response["failures"]).should.be(1) 1020 1021 1022@mock_ecr 1023def test_batch_delete_image_by_digest(): 1024 client = boto3.client("ecr", region_name="us-east-1") 1025 client.create_repository(repositoryName="test_repository") 1026 1027 manifest = _create_image_manifest() 1028 1029 tags = ["v1", "v2", "latest"] 1030 for tag in tags: 1031 client.put_image( 1032 repositoryName="test_repository", 1033 imageManifest=json.dumps(manifest), 1034 imageTag=tag, 1035 ) 1036 1037 describe_response = client.describe_images(repositoryName="test_repository") 1038 image_digest = describe_response["imageDetails"][0]["imageDigest"] 1039 1040 batch_delete_response = client.batch_delete_image( 1041 registryId="012345678910", 1042 repositoryName="test_repository", 1043 imageIds=[{"imageDigest": image_digest}], 1044 ) 1045 1046 describe_response = client.describe_images(repositoryName="test_repository") 1047 1048 type(describe_response["imageDetails"]).should.be(list) 1049 len(describe_response["imageDetails"]).should.be(0) 1050 1051 type(batch_delete_response["imageIds"]).should.be(list) 1052 len(batch_delete_response["imageIds"]).should.be(3) 1053 1054 batch_delete_response["imageIds"][0]["imageDigest"].should.equal(image_digest) 1055 batch_delete_response["imageIds"][1]["imageDigest"].should.equal(image_digest) 1056 batch_delete_response["imageIds"][2]["imageDigest"].should.equal(image_digest) 1057 1058 set( 1059 [ 1060 batch_delete_response["imageIds"][0]["imageTag"], 1061 batch_delete_response["imageIds"][1]["imageTag"], 1062 batch_delete_response["imageIds"][2]["imageTag"], 1063 ] 1064 ).should.equal(set(tags)) 1065 1066 type(batch_delete_response["failures"]).should.be(list) 1067 len(batch_delete_response["failures"]).should.be(0) 1068 1069 1070@mock_ecr 1071def test_batch_delete_image_with_invalid_digest(): 1072 client = boto3.client("ecr", region_name="us-east-1") 1073 client.create_repository(repositoryName="test_repository") 1074 1075 manifest = _create_image_manifest() 1076 1077 tags = ["v1", "v2", "latest"] 1078 for tag in tags: 1079 client.put_image( 1080 repositoryName="test_repository", 1081 imageManifest=json.dumps(manifest), 1082 imageTag=tag, 1083 ) 1084 1085 invalid_image_digest = "sha256:invalid-digest" 1086 1087 batch_delete_response = client.batch_delete_image( 1088 registryId="012345678910", 1089 repositoryName="test_repository", 1090 imageIds=[{"imageDigest": invalid_image_digest}], 1091 ) 1092 1093 type(batch_delete_response["imageIds"]).should.be(list) 1094 len(batch_delete_response["imageIds"]).should.be(0) 1095 1096 type(batch_delete_response["failures"]).should.be(list) 1097 len(batch_delete_response["failures"]).should.be(1) 1098 1099 batch_delete_response["failures"][0]["imageId"]["imageDigest"].should.equal( 1100 invalid_image_digest 1101 ) 1102 batch_delete_response["failures"][0]["failureCode"].should.equal( 1103 "InvalidImageDigest" 1104 ) 1105 batch_delete_response["failures"][0]["failureReason"].should.equal( 1106 "Invalid request parameters: image digest should satisfy the regex '[a-zA-Z0-9-_+.]+:[a-fA-F0-9]+'" 1107 ) 1108 1109 1110@mock_ecr 1111def test_batch_delete_image_with_missing_parameters(): 1112 client = boto3.client("ecr", region_name="us-east-1") 1113 client.create_repository(repositoryName="test_repository") 1114 1115 batch_delete_response = client.batch_delete_image( 1116 registryId="012345678910", repositoryName="test_repository", imageIds=[{}] 1117 ) 1118 1119 type(batch_delete_response["imageIds"]).should.be(list) 1120 len(batch_delete_response["imageIds"]).should.be(0) 1121 1122 type(batch_delete_response["failures"]).should.be(list) 1123 len(batch_delete_response["failures"]).should.be(1) 1124 1125 batch_delete_response["failures"][0]["failureCode"].should.equal( 1126 "MissingDigestAndTag" 1127 ) 1128 batch_delete_response["failures"][0]["failureReason"].should.equal( 1129 "Invalid request parameters: both tag and digest cannot be null" 1130 ) 1131 1132 1133@mock_ecr 1134def test_batch_delete_image_with_matching_digest_and_tag(): 1135 client = boto3.client("ecr", region_name="us-east-1") 1136 client.create_repository(repositoryName="test_repository") 1137 1138 manifest = _create_image_manifest() 1139 1140 tags = ["v1", "v1.0", "latest"] 1141 for tag in tags: 1142 client.put_image( 1143 repositoryName="test_repository", 1144 imageManifest=json.dumps(manifest), 1145 imageTag=tag, 1146 ) 1147 1148 describe_response = client.describe_images(repositoryName="test_repository") 1149 image_digest = describe_response["imageDetails"][0]["imageDigest"] 1150 1151 batch_delete_response = client.batch_delete_image( 1152 registryId="012345678910", 1153 repositoryName="test_repository", 1154 imageIds=[{"imageDigest": image_digest, "imageTag": "v1"}], 1155 ) 1156 1157 describe_response = client.describe_images(repositoryName="test_repository") 1158 1159 type(describe_response["imageDetails"]).should.be(list) 1160 len(describe_response["imageDetails"]).should.be(0) 1161 1162 type(batch_delete_response["imageIds"]).should.be(list) 1163 len(batch_delete_response["imageIds"]).should.be(3) 1164 1165 batch_delete_response["imageIds"][0]["imageDigest"].should.equal(image_digest) 1166 batch_delete_response["imageIds"][1]["imageDigest"].should.equal(image_digest) 1167 batch_delete_response["imageIds"][2]["imageDigest"].should.equal(image_digest) 1168 1169 set( 1170 [ 1171 batch_delete_response["imageIds"][0]["imageTag"], 1172 batch_delete_response["imageIds"][1]["imageTag"], 1173 batch_delete_response["imageIds"][2]["imageTag"], 1174 ] 1175 ).should.equal(set(tags)) 1176 1177 type(batch_delete_response["failures"]).should.be(list) 1178 len(batch_delete_response["failures"]).should.be(0) 1179 1180 1181@mock_ecr 1182def test_batch_delete_image_with_mismatched_digest_and_tag(): 1183 client = boto3.client("ecr", region_name="us-east-1") 1184 client.create_repository(repositoryName="test_repository") 1185 1186 manifest = _create_image_manifest() 1187 1188 tags = ["v1", "latest"] 1189 for tag in tags: 1190 client.put_image( 1191 repositoryName="test_repository", 1192 imageManifest=json.dumps(manifest), 1193 imageTag=tag, 1194 ) 1195 1196 describe_response = client.describe_images(repositoryName="test_repository") 1197 image_digest = describe_response["imageDetails"][0]["imageDigest"] 1198 1199 batch_delete_response = client.batch_delete_image( 1200 registryId="012345678910", 1201 repositoryName="test_repository", 1202 imageIds=[{"imageDigest": image_digest, "imageTag": "v2"}], 1203 ) 1204 1205 type(batch_delete_response["imageIds"]).should.be(list) 1206 len(batch_delete_response["imageIds"]).should.be(0) 1207 1208 type(batch_delete_response["failures"]).should.be(list) 1209 len(batch_delete_response["failures"]).should.be(1) 1210 1211 batch_delete_response["failures"][0]["imageId"]["imageDigest"].should.equal( 1212 image_digest 1213 ) 1214 batch_delete_response["failures"][0]["imageId"]["imageTag"].should.equal("v2") 1215 batch_delete_response["failures"][0]["failureCode"].should.equal("ImageNotFound") 1216 batch_delete_response["failures"][0]["failureReason"].should.equal( 1217 "Requested image not found" 1218 ) 1219 1220 1221@mock_ecr 1222def test_list_tags_for_resource(): 1223 # given 1224 client = boto3.client("ecr", region_name="eu-central-1") 1225 repo_name = "test-repo" 1226 arn = client.create_repository( 1227 repositoryName=repo_name, tags=[{"Key": "key-1", "Value": "value-1"}], 1228 )["repository"]["repositoryArn"] 1229 1230 # when 1231 tags = client.list_tags_for_resource(resourceArn=arn)["tags"] 1232 1233 # then 1234 tags.should.equal([{"Key": "key-1", "Value": "value-1"}]) 1235 1236 1237@mock_ecr 1238def test_list_tags_for_resource_error_not_exists(): 1239 # given 1240 region_name = "eu-central-1" 1241 client = boto3.client("ecr", region_name=region_name) 1242 repo_name = "not-exists" 1243 1244 # when 1245 with pytest.raises(ClientError) as e: 1246 client.list_tags_for_resource( 1247 resourceArn=f"arn:aws:ecr:{region_name}:{ACCOUNT_ID}:repository/{repo_name}" 1248 ) 1249 1250 # then 1251 ex = e.value 1252 ex.operation_name.should.equal("ListTagsForResource") 1253 ex.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400) 1254 ex.response["Error"]["Code"].should.contain("RepositoryNotFoundException") 1255 ex.response["Error"]["Message"].should.equal( 1256 f"The repository with name '{repo_name}' does not exist " 1257 f"in the registry with id '{ACCOUNT_ID}'" 1258 ) 1259 1260 1261@mock_ecr 1262def test_list_tags_for_resource_error_invalid_param(): 1263 # given 1264 region_name = "eu-central-1" 1265 client = boto3.client("ecr", region_name=region_name) 1266 1267 # when 1268 with pytest.raises(ClientError) as e: 1269 client.list_tags_for_resource(resourceArn="invalid",) 1270 1271 # then 1272 ex = e.value 1273 ex.operation_name.should.equal("ListTagsForResource") 1274 ex.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400) 1275 ex.response["Error"]["Code"].should.contain("InvalidParameterException") 1276 ex.response["Error"]["Message"].should.equal( 1277 "Invalid parameter at 'resourceArn' failed to satisfy constraint: " 1278 "'Invalid ARN'" 1279 ) 1280 1281 1282@mock_ecr 1283def test_tag_resource(): 1284 # given 1285 client = boto3.client("ecr", region_name="eu-central-1") 1286 repo_name = "test-repo" 1287 arn = client.create_repository( 1288 repositoryName=repo_name, tags=[{"Key": "key-1", "Value": "value-1"}], 1289 )["repository"]["repositoryArn"] 1290 1291 # when 1292 client.tag_resource(resourceArn=arn, tags=[{"Key": "key-2", "Value": "value-2"}]) 1293 1294 # then 1295 tags = client.list_tags_for_resource(resourceArn=arn)["tags"] 1296 sorted(tags, key=lambda i: i["Key"]).should.equal( 1297 sorted( 1298 [ 1299 {"Key": "key-1", "Value": "value-1"}, 1300 {"Key": "key-2", "Value": "value-2"}, 1301 ], 1302 key=lambda i: i["Key"], 1303 ) 1304 ) 1305 1306 1307@mock_ecr 1308def test_tag_resource_error_not_exists(): 1309 # given 1310 region_name = "eu-central-1" 1311 client = boto3.client("ecr", region_name=region_name) 1312 repo_name = "not-exists" 1313 1314 # when 1315 with pytest.raises(ClientError) as e: 1316 client.tag_resource( 1317 resourceArn=f"arn:aws:ecr:{region_name}:{ACCOUNT_ID}:repository/{repo_name}", 1318 tags=[{"Key": "key-1", "Value": "value-2"}], 1319 ) 1320 1321 # then 1322 ex = e.value 1323 ex.operation_name.should.equal("TagResource") 1324 ex.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400) 1325 ex.response["Error"]["Code"].should.contain("RepositoryNotFoundException") 1326 ex.response["Error"]["Message"].should.equal( 1327 f"The repository with name '{repo_name}' does not exist " 1328 f"in the registry with id '{ACCOUNT_ID}'" 1329 ) 1330 1331 1332@mock_ecr 1333def test_untag_resource(): 1334 # given 1335 client = boto3.client("ecr", region_name="eu-central-1") 1336 repo_name = "test-repo" 1337 arn = client.create_repository( 1338 repositoryName=repo_name, 1339 tags=[ 1340 {"Key": "key-1", "Value": "value-1"}, 1341 {"Key": "key-2", "Value": "value-2"}, 1342 ], 1343 )["repository"]["repositoryArn"] 1344 1345 # when 1346 client.untag_resource(resourceArn=arn, tagKeys=["key-1"]) 1347 1348 # then 1349 tags = client.list_tags_for_resource(resourceArn=arn)["tags"] 1350 tags.should.equal([{"Key": "key-2", "Value": "value-2"}]) 1351 1352 1353@mock_ecr 1354def test_untag_resource_error_not_exists(): 1355 # given 1356 region_name = "eu-central-1" 1357 client = boto3.client("ecr", region_name=region_name) 1358 repo_name = "not-exists" 1359 1360 # when 1361 with pytest.raises(ClientError) as e: 1362 client.untag_resource( 1363 resourceArn=f"arn:aws:ecr:{region_name}:{ACCOUNT_ID}:repository/{repo_name}", 1364 tagKeys=["key-1"], 1365 ) 1366 1367 # then 1368 ex = e.value 1369 ex.operation_name.should.equal("UntagResource") 1370 ex.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400) 1371 ex.response["Error"]["Code"].should.contain("RepositoryNotFoundException") 1372 ex.response["Error"]["Message"].should.equal( 1373 f"The repository with name '{repo_name}' does not exist " 1374 f"in the registry with id '{ACCOUNT_ID}'" 1375 ) 1376 1377 1378@mock_ecr 1379def test_put_image_tag_mutability(): 1380 # given 1381 client = boto3.client("ecr", region_name="eu-central-1") 1382 repo_name = "test-repo" 1383 client.create_repository(repositoryName=repo_name) 1384 1385 response = client.describe_repositories(repositoryNames=[repo_name]) 1386 response["repositories"][0]["imageTagMutability"].should.equal("MUTABLE") 1387 1388 # when 1389 response = client.put_image_tag_mutability( 1390 repositoryName=repo_name, imageTagMutability="IMMUTABLE", 1391 ) 1392 1393 # then 1394 response["imageTagMutability"].should.equal("IMMUTABLE") 1395 response["registryId"].should.equal(ACCOUNT_ID) 1396 response["repositoryName"].should.equal(repo_name) 1397 1398 response = client.describe_repositories(repositoryNames=[repo_name]) 1399 response["repositories"][0]["imageTagMutability"].should.equal("IMMUTABLE") 1400 1401 1402@mock_ecr 1403def test_put_image_tag_mutability_error_not_exists(): 1404 # given 1405 region_name = "eu-central-1" 1406 client = boto3.client("ecr", region_name=region_name) 1407 repo_name = "not-exists" 1408 1409 # when 1410 with pytest.raises(ClientError) as e: 1411 client.put_image_tag_mutability( 1412 repositoryName=repo_name, imageTagMutability="IMMUTABLE", 1413 ) 1414 1415 # then 1416 ex = e.value 1417 ex.operation_name.should.equal("PutImageTagMutability") 1418 ex.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400) 1419 ex.response["Error"]["Code"].should.contain("RepositoryNotFoundException") 1420 ex.response["Error"]["Message"].should.equal( 1421 f"The repository with name '{repo_name}' does not exist " 1422 f"in the registry with id '{ACCOUNT_ID}'" 1423 ) 1424 1425 1426@mock_ecr 1427def test_put_image_tag_mutability_error_invalid_param(): 1428 # given 1429 region_name = "eu-central-1" 1430 client = boto3.client("ecr", region_name=region_name) 1431 repo_name = "test-repo" 1432 client.create_repository(repositoryName=repo_name) 1433 1434 # when 1435 with pytest.raises(ClientError) as e: 1436 client.put_image_tag_mutability( 1437 repositoryName=repo_name, imageTagMutability="invalid", 1438 ) 1439 1440 # then 1441 ex = e.value 1442 ex.operation_name.should.equal("PutImageTagMutability") 1443 ex.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400) 1444 ex.response["Error"]["Code"].should.contain("InvalidParameterException") 1445 ex.response["Error"]["Message"].should.equal( 1446 "Invalid parameter at 'imageTagMutability' failed to satisfy constraint: " 1447 "'Member must satisfy enum value set: [IMMUTABLE, MUTABLE]'" 1448 ) 1449 1450 1451@mock_ecr 1452def test_put_image_scanning_configuration(): 1453 # given 1454 client = boto3.client("ecr", region_name="eu-central-1") 1455 repo_name = "test-repo" 1456 client.create_repository(repositoryName=repo_name) 1457 1458 response = client.describe_repositories(repositoryNames=[repo_name]) 1459 response["repositories"][0]["imageScanningConfiguration"].should.equal( 1460 {"scanOnPush": False} 1461 ) 1462 1463 # when 1464 response = client.put_image_scanning_configuration( 1465 repositoryName=repo_name, imageScanningConfiguration={"scanOnPush": True} 1466 ) 1467 1468 # then 1469 response["imageScanningConfiguration"].should.equal({"scanOnPush": True}) 1470 response["registryId"].should.equal(ACCOUNT_ID) 1471 response["repositoryName"].should.equal(repo_name) 1472 1473 response = client.describe_repositories(repositoryNames=[repo_name]) 1474 response["repositories"][0]["imageScanningConfiguration"].should.equal( 1475 {"scanOnPush": True} 1476 ) 1477 1478 1479@mock_ecr 1480def test_put_image_scanning_configuration_error_not_exists(): 1481 # given 1482 region_name = "eu-central-1" 1483 client = boto3.client("ecr", region_name=region_name) 1484 repo_name = "not-exists" 1485 1486 # when 1487 with pytest.raises(ClientError) as e: 1488 client.put_image_scanning_configuration( 1489 repositoryName=repo_name, imageScanningConfiguration={"scanOnPush": True}, 1490 ) 1491 1492 # then 1493 ex = e.value 1494 ex.operation_name.should.equal("PutImageScanningConfiguration") 1495 ex.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400) 1496 ex.response["Error"]["Code"].should.contain("RepositoryNotFoundException") 1497 ex.response["Error"]["Message"].should.equal( 1498 f"The repository with name '{repo_name}' does not exist " 1499 f"in the registry with id '{ACCOUNT_ID}'" 1500 ) 1501 1502 1503@mock_ecr 1504def test_set_repository_policy(): 1505 # given 1506 client = boto3.client("ecr", region_name="eu-central-1") 1507 repo_name = "test-repo" 1508 client.create_repository(repositoryName=repo_name) 1509 policy = { 1510 "Version": "2012-10-17", 1511 "Statement": [ 1512 { 1513 "Sid": "root", 1514 "Effect": "Allow", 1515 "Principal": {"AWS": f"arn:aws:iam::{ACCOUNT_ID}:root"}, 1516 "Action": ["ecr:DescribeImages"], 1517 } 1518 ], 1519 } 1520 1521 # when 1522 response = client.set_repository_policy( 1523 repositoryName=repo_name, policyText=json.dumps(policy), 1524 ) 1525 1526 # then 1527 response["registryId"].should.equal(ACCOUNT_ID) 1528 response["repositoryName"].should.equal(repo_name) 1529 json.loads(response["policyText"]).should.equal(policy) 1530 1531 1532@mock_ecr 1533def test_set_repository_policy_error_not_exists(): 1534 # given 1535 region_name = "eu-central-1" 1536 client = boto3.client("ecr", region_name=region_name) 1537 repo_name = "not-exists" 1538 policy = { 1539 "Version": "2012-10-17", 1540 "Statement": [ 1541 { 1542 "Sid": "root", 1543 "Effect": "Allow", 1544 "Principal": {"AWS": f"arn:aws:iam::{ACCOUNT_ID}:root"}, 1545 "Action": ["ecr:DescribeImages"], 1546 } 1547 ], 1548 } 1549 1550 # when 1551 with pytest.raises(ClientError) as e: 1552 client.set_repository_policy( 1553 repositoryName=repo_name, policyText=json.dumps(policy), 1554 ) 1555 1556 # then 1557 ex = e.value 1558 ex.operation_name.should.equal("SetRepositoryPolicy") 1559 ex.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400) 1560 ex.response["Error"]["Code"].should.contain("RepositoryNotFoundException") 1561 ex.response["Error"]["Message"].should.equal( 1562 f"The repository with name '{repo_name}' does not exist " 1563 f"in the registry with id '{ACCOUNT_ID}'" 1564 ) 1565 1566 1567@mock_ecr 1568def test_set_repository_policy_error_invalid_param(): 1569 # given 1570 region_name = "eu-central-1" 1571 client = boto3.client("ecr", region_name=region_name) 1572 repo_name = "test-repo" 1573 client.create_repository(repositoryName=repo_name) 1574 policy = { 1575 "Version": "2012-10-17", 1576 "Statement": [{"Effect": "Allow"}], 1577 } 1578 1579 # when 1580 with pytest.raises(ClientError) as e: 1581 client.set_repository_policy( 1582 repositoryName=repo_name, policyText=json.dumps(policy), 1583 ) 1584 1585 # then 1586 ex = e.value 1587 ex.operation_name.should.equal("SetRepositoryPolicy") 1588 ex.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400) 1589 ex.response["Error"]["Code"].should.contain("InvalidParameterException") 1590 ex.response["Error"]["Message"].should.equal( 1591 "Invalid parameter at 'PolicyText' failed to satisfy constraint: " 1592 "'Invalid repository policy provided'" 1593 ) 1594 1595 1596@mock_ecr 1597def test_get_repository_policy(): 1598 # given 1599 client = boto3.client("ecr", region_name="eu-central-1") 1600 repo_name = "test-repo" 1601 client.create_repository(repositoryName=repo_name) 1602 policy = { 1603 "Version": "2012-10-17", 1604 "Statement": [ 1605 { 1606 "Sid": "root", 1607 "Effect": "Allow", 1608 "Principal": {"AWS": f"arn:aws:iam::{ACCOUNT_ID}:root"}, 1609 "Action": ["ecr:DescribeImages"], 1610 } 1611 ], 1612 } 1613 client.set_repository_policy( 1614 repositoryName=repo_name, policyText=json.dumps(policy), 1615 ) 1616 1617 # when 1618 response = client.get_repository_policy(repositoryName=repo_name) 1619 1620 # then 1621 response["registryId"].should.equal(ACCOUNT_ID) 1622 response["repositoryName"].should.equal(repo_name) 1623 json.loads(response["policyText"]).should.equal(policy) 1624 1625 1626@mock_ecr 1627def test_get_repository_policy_error_repo_not_exists(): 1628 # given 1629 region_name = "eu-central-1" 1630 client = boto3.client("ecr", region_name=region_name) 1631 repo_name = "not-exists" 1632 1633 # when 1634 with pytest.raises(ClientError) as e: 1635 client.get_repository_policy(repositoryName=repo_name) 1636 1637 # then 1638 ex = e.value 1639 ex.operation_name.should.equal("GetRepositoryPolicy") 1640 ex.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400) 1641 ex.response["Error"]["Code"].should.contain("RepositoryNotFoundException") 1642 ex.response["Error"]["Message"].should.equal( 1643 f"The repository with name '{repo_name}' does not exist " 1644 f"in the registry with id '{ACCOUNT_ID}'" 1645 ) 1646 1647 1648@mock_ecr 1649def test_get_repository_policy_error_policy_not_exists(): 1650 # given 1651 region_name = "eu-central-1" 1652 client = boto3.client("ecr", region_name=region_name) 1653 repo_name = "test-repo" 1654 client.create_repository(repositoryName=repo_name) 1655 1656 # when 1657 with pytest.raises(ClientError) as e: 1658 client.get_repository_policy(repositoryName=repo_name) 1659 1660 # then 1661 ex = e.value 1662 ex.operation_name.should.equal("GetRepositoryPolicy") 1663 ex.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400) 1664 ex.response["Error"]["Code"].should.contain("RepositoryPolicyNotFoundException") 1665 ex.response["Error"]["Message"].should.equal( 1666 "Repository policy does not exist " 1667 f"for the repository with name '{repo_name}' " 1668 f"in the registry with id '{ACCOUNT_ID}'" 1669 ) 1670 1671 1672@mock_ecr 1673def test_delete_repository_policy(): 1674 # given 1675 client = boto3.client("ecr", region_name="eu-central-1") 1676 repo_name = "test-repo" 1677 client.create_repository(repositoryName=repo_name) 1678 policy = { 1679 "Version": "2012-10-17", 1680 "Statement": [ 1681 { 1682 "Sid": "root", 1683 "Effect": "Allow", 1684 "Principal": {"AWS": f"arn:aws:iam::{ACCOUNT_ID}:root"}, 1685 "Action": ["ecr:DescribeImages"], 1686 } 1687 ], 1688 } 1689 client.set_repository_policy( 1690 repositoryName=repo_name, policyText=json.dumps(policy), 1691 ) 1692 1693 # when 1694 response = client.delete_repository_policy(repositoryName=repo_name) 1695 1696 # then 1697 response["registryId"].should.equal(ACCOUNT_ID) 1698 response["repositoryName"].should.equal(repo_name) 1699 json.loads(response["policyText"]).should.equal(policy) 1700 1701 with pytest.raises(ClientError) as e: 1702 client.get_repository_policy(repositoryName=repo_name) 1703 1704 e.value.response["Error"]["Code"].should.contain( 1705 "RepositoryPolicyNotFoundException" 1706 ) 1707 1708 1709@mock_ecr 1710def test_delete_repository_policy_error_repo_not_exists(): 1711 # given 1712 region_name = "eu-central-1" 1713 client = boto3.client("ecr", region_name=region_name) 1714 repo_name = "not-exists" 1715 1716 # when 1717 with pytest.raises(ClientError) as e: 1718 client.delete_repository_policy(repositoryName=repo_name) 1719 1720 # then 1721 ex = e.value 1722 ex.operation_name.should.equal("DeleteRepositoryPolicy") 1723 ex.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400) 1724 ex.response["Error"]["Code"].should.contain("RepositoryNotFoundException") 1725 ex.response["Error"]["Message"].should.equal( 1726 f"The repository with name '{repo_name}' does not exist " 1727 f"in the registry with id '{ACCOUNT_ID}'" 1728 ) 1729 1730 1731@mock_ecr 1732def test_delete_repository_policy_error_policy_not_exists(): 1733 # given 1734 region_name = "eu-central-1" 1735 client = boto3.client("ecr", region_name=region_name) 1736 repo_name = "test-repo" 1737 client.create_repository(repositoryName=repo_name) 1738 1739 # when 1740 with pytest.raises(ClientError) as e: 1741 client.delete_repository_policy(repositoryName=repo_name) 1742 1743 # then 1744 ex = e.value 1745 ex.operation_name.should.equal("DeleteRepositoryPolicy") 1746 ex.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400) 1747 ex.response["Error"]["Code"].should.contain("RepositoryPolicyNotFoundException") 1748 ex.response["Error"]["Message"].should.equal( 1749 "Repository policy does not exist " 1750 f"for the repository with name '{repo_name}' " 1751 f"in the registry with id '{ACCOUNT_ID}'" 1752 ) 1753 1754 1755@mock_ecr 1756def test_put_lifecycle_policy(): 1757 # given 1758 client = boto3.client("ecr", region_name="eu-central-1") 1759 repo_name = "test-repo" 1760 client.create_repository(repositoryName=repo_name) 1761 policy = { 1762 "rules": [ 1763 { 1764 "rulePriority": 1, 1765 "description": "test policy", 1766 "selection": { 1767 "tagStatus": "untagged", 1768 "countType": "imageCountMoreThan", 1769 "countNumber": 30, 1770 }, 1771 "action": {"type": "expire"}, 1772 } 1773 ] 1774 } 1775 1776 # when 1777 response = client.put_lifecycle_policy( 1778 repositoryName=repo_name, lifecyclePolicyText=json.dumps(policy), 1779 ) 1780 1781 # then 1782 response["registryId"].should.equal(ACCOUNT_ID) 1783 response["repositoryName"].should.equal(repo_name) 1784 json.loads(response["lifecyclePolicyText"]).should.equal(policy) 1785 1786 1787@mock_ecr 1788def test_put_lifecycle_policy_error_repo_not_exists(): 1789 # given 1790 region_name = "eu-central-1" 1791 client = boto3.client("ecr", region_name=region_name) 1792 repo_name = "not-exists" 1793 policy = { 1794 "rules": [ 1795 { 1796 "rulePriority": 1, 1797 "description": "test policy", 1798 "selection": { 1799 "tagStatus": "untagged", 1800 "countType": "imageCountMoreThan", 1801 "countNumber": 30, 1802 }, 1803 "action": {"type": "expire"}, 1804 } 1805 ] 1806 } 1807 1808 # when 1809 with pytest.raises(ClientError) as e: 1810 client.put_lifecycle_policy( 1811 repositoryName=repo_name, lifecyclePolicyText=json.dumps(policy) 1812 ) 1813 1814 # then 1815 ex = e.value 1816 ex.operation_name.should.equal("PutLifecyclePolicy") 1817 ex.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400) 1818 ex.response["Error"]["Code"].should.contain("RepositoryNotFoundException") 1819 ex.response["Error"]["Message"].should.equal( 1820 f"The repository with name '{repo_name}' does not exist " 1821 f"in the registry with id '{ACCOUNT_ID}'" 1822 ) 1823 1824 1825@mock_ecr 1826def test_get_lifecycle_policy(): 1827 # given 1828 client = boto3.client("ecr", region_name="eu-central-1") 1829 repo_name = "test-repo" 1830 client.create_repository(repositoryName=repo_name) 1831 policy = { 1832 "rules": [ 1833 { 1834 "rulePriority": 1, 1835 "description": "test policy", 1836 "selection": { 1837 "tagStatus": "untagged", 1838 "countType": "imageCountMoreThan", 1839 "countNumber": 30, 1840 }, 1841 "action": {"type": "expire"}, 1842 } 1843 ] 1844 } 1845 client.put_lifecycle_policy( 1846 repositoryName=repo_name, lifecyclePolicyText=json.dumps(policy), 1847 ) 1848 1849 # when 1850 response = client.get_lifecycle_policy(repositoryName=repo_name) 1851 1852 # then 1853 response["registryId"].should.equal(ACCOUNT_ID) 1854 response["repositoryName"].should.equal(repo_name) 1855 json.loads(response["lifecyclePolicyText"]).should.equal(policy) 1856 response["lastEvaluatedAt"].should.be.a(datetime) 1857 1858 1859@mock_ecr 1860def test_get_lifecycle_policy_error_repo_not_exists(): 1861 # given 1862 region_name = "eu-central-1" 1863 client = boto3.client("ecr", region_name=region_name) 1864 repo_name = "not-exists" 1865 1866 # when 1867 with pytest.raises(ClientError) as e: 1868 client.get_lifecycle_policy(repositoryName=repo_name) 1869 1870 # then 1871 ex = e.value 1872 ex.operation_name.should.equal("GetLifecyclePolicy") 1873 ex.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400) 1874 ex.response["Error"]["Code"].should.contain("RepositoryNotFoundException") 1875 ex.response["Error"]["Message"].should.equal( 1876 f"The repository with name '{repo_name}' does not exist " 1877 f"in the registry with id '{ACCOUNT_ID}'" 1878 ) 1879 1880 1881@mock_ecr 1882def test_get_lifecycle_policy_error_policy_not_exists(): 1883 # given 1884 region_name = "eu-central-1" 1885 client = boto3.client("ecr", region_name=region_name) 1886 repo_name = "test-repo" 1887 client.create_repository(repositoryName=repo_name) 1888 1889 # when 1890 with pytest.raises(ClientError) as e: 1891 client.get_lifecycle_policy(repositoryName=repo_name) 1892 1893 # then 1894 ex = e.value 1895 ex.operation_name.should.equal("GetLifecyclePolicy") 1896 ex.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400) 1897 ex.response["Error"]["Code"].should.contain("LifecyclePolicyNotFoundException") 1898 ex.response["Error"]["Message"].should.equal( 1899 "Lifecycle policy does not exist " 1900 f"for the repository with name '{repo_name}' " 1901 f"in the registry with id '{ACCOUNT_ID}'" 1902 ) 1903 1904 1905@mock_ecr 1906def test_delete_lifecycle_policy(): 1907 # given 1908 client = boto3.client("ecr", region_name="eu-central-1") 1909 repo_name = "test-repo" 1910 client.create_repository(repositoryName=repo_name) 1911 policy = { 1912 "rules": [ 1913 { 1914 "rulePriority": 1, 1915 "description": "test policy", 1916 "selection": { 1917 "tagStatus": "untagged", 1918 "countType": "imageCountMoreThan", 1919 "countNumber": 30, 1920 }, 1921 "action": {"type": "expire"}, 1922 } 1923 ] 1924 } 1925 client.put_lifecycle_policy( 1926 repositoryName=repo_name, lifecyclePolicyText=json.dumps(policy), 1927 ) 1928 1929 # when 1930 response = client.delete_lifecycle_policy(repositoryName=repo_name) 1931 1932 # then 1933 response["registryId"].should.equal(ACCOUNT_ID) 1934 response["repositoryName"].should.equal(repo_name) 1935 json.loads(response["lifecyclePolicyText"]).should.equal(policy) 1936 response["lastEvaluatedAt"].should.be.a(datetime) 1937 1938 with pytest.raises(ClientError) as e: 1939 client.get_lifecycle_policy(repositoryName=repo_name) 1940 1941 e.value.response["Error"]["Code"].should.contain("LifecyclePolicyNotFoundException") 1942 1943 1944@mock_ecr 1945def test_delete_lifecycle_policy_error_repo_not_exists(): 1946 # given 1947 region_name = "eu-central-1" 1948 client = boto3.client("ecr", region_name=region_name) 1949 repo_name = "not-exists" 1950 1951 # when 1952 with pytest.raises(ClientError) as e: 1953 client.delete_lifecycle_policy(repositoryName=repo_name) 1954 1955 # then 1956 ex = e.value 1957 ex.operation_name.should.equal("DeleteLifecyclePolicy") 1958 ex.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400) 1959 ex.response["Error"]["Code"].should.contain("RepositoryNotFoundException") 1960 ex.response["Error"]["Message"].should.equal( 1961 f"The repository with name '{repo_name}' does not exist " 1962 f"in the registry with id '{ACCOUNT_ID}'" 1963 ) 1964 1965 1966@mock_ecr 1967def test_delete_lifecycle_policy_error_policy_not_exists(): 1968 # given 1969 region_name = "eu-central-1" 1970 client = boto3.client("ecr", region_name=region_name) 1971 repo_name = "test-repo" 1972 client.create_repository(repositoryName=repo_name) 1973 1974 # when 1975 with pytest.raises(ClientError) as e: 1976 client.delete_lifecycle_policy(repositoryName=repo_name) 1977 1978 # then 1979 ex = e.value 1980 ex.operation_name.should.equal("DeleteLifecyclePolicy") 1981 ex.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400) 1982 ex.response["Error"]["Code"].should.contain("LifecyclePolicyNotFoundException") 1983 ex.response["Error"]["Message"].should.equal( 1984 "Lifecycle policy does not exist " 1985 f"for the repository with name '{repo_name}' " 1986 f"in the registry with id '{ACCOUNT_ID}'" 1987 ) 1988 1989 1990@mock_ecr 1991def test_put_registry_policy(): 1992 # given 1993 client = boto3.client("ecr", region_name="eu-central-1") 1994 policy = { 1995 "Version": "2012-10-17", 1996 "Statement": [ 1997 { 1998 "Effect": "Allow", 1999 "Principal": { 2000 "AWS": ["arn:aws:iam::111111111111:root", "222222222222"] 2001 }, 2002 "Action": ["ecr:CreateRepository", "ecr:ReplicateImage"], 2003 "Resource": "*", 2004 } 2005 ], 2006 } 2007 2008 # when 2009 response = client.put_registry_policy(policyText=json.dumps(policy)) 2010 2011 # then 2012 response["registryId"].should.equal(ACCOUNT_ID) 2013 json.loads(response["policyText"]).should.equal(policy) 2014 2015 2016@mock_ecr 2017def test_put_registry_policy_error_invalid_action(): 2018 # given 2019 client = boto3.client("ecr", region_name="eu-central-1") 2020 policy = { 2021 "Version": "2012-10-17", 2022 "Statement": [ 2023 { 2024 "Effect": "Allow", 2025 "Principal": {"AWS": "arn:aws:iam::111111111111:root"}, 2026 "Action": [ 2027 "ecr:CreateRepository", 2028 "ecr:ReplicateImage", 2029 "ecr:DescribeRepositories", 2030 ], 2031 "Resource": "*", 2032 } 2033 ], 2034 } 2035 2036 # when 2037 with pytest.raises(ClientError) as e: 2038 client.put_registry_policy(policyText=json.dumps(policy)) 2039 2040 # then 2041 ex = e.value 2042 ex.operation_name.should.equal("PutRegistryPolicy") 2043 ex.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400) 2044 ex.response["Error"]["Code"].should.contain("InvalidParameterException") 2045 ex.response["Error"]["Message"].should.equal( 2046 "Invalid parameter at 'PolicyText' failed to satisfy constraint: " 2047 "'Invalid registry policy provided'" 2048 ) 2049 2050 2051@mock_ecr 2052def test_get_registry_policy(): 2053 # given 2054 client = boto3.client("ecr", region_name="eu-central-1") 2055 policy = { 2056 "Version": "2012-10-17", 2057 "Statement": [ 2058 { 2059 "Effect": "Allow", 2060 "Principal": { 2061 "AWS": ["arn:aws:iam::111111111111:root", "222222222222"] 2062 }, 2063 "Action": ["ecr:CreateRepository", "ecr:ReplicateImage"], 2064 "Resource": "*", 2065 } 2066 ], 2067 } 2068 client.put_registry_policy(policyText=json.dumps(policy)) 2069 2070 # when 2071 response = client.get_registry_policy() 2072 2073 # then 2074 response["registryId"].should.equal(ACCOUNT_ID) 2075 json.loads(response["policyText"]).should.equal(policy) 2076 2077 2078@mock_ecr 2079def test_get_registry_policy_error_policy_not_exists(): 2080 # given 2081 client = boto3.client("ecr", region_name="eu-central-1") 2082 2083 # when 2084 with pytest.raises(ClientError) as e: 2085 client.get_registry_policy() 2086 2087 # then 2088 ex = e.value 2089 ex.operation_name.should.equal("GetRegistryPolicy") 2090 ex.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400) 2091 ex.response["Error"]["Code"].should.contain("RegistryPolicyNotFoundException") 2092 ex.response["Error"]["Message"].should.equal( 2093 f"Registry policy does not exist in the registry with id '{ACCOUNT_ID}'" 2094 ) 2095 2096 2097@mock_ecr 2098def test_delete_registry_policy(): 2099 # given 2100 client = boto3.client("ecr", region_name="eu-central-1") 2101 policy = { 2102 "Version": "2012-10-17", 2103 "Statement": [ 2104 { 2105 "Effect": "Allow", 2106 "Principal": { 2107 "AWS": ["arn:aws:iam::111111111111:root", "222222222222"] 2108 }, 2109 "Action": ["ecr:CreateRepository", "ecr:ReplicateImage"], 2110 "Resource": "*", 2111 } 2112 ], 2113 } 2114 client.put_registry_policy(policyText=json.dumps(policy)) 2115 2116 # when 2117 response = client.delete_registry_policy() 2118 2119 # then 2120 response["registryId"].should.equal(ACCOUNT_ID) 2121 json.loads(response["policyText"]).should.equal(policy) 2122 2123 with pytest.raises(ClientError) as e: 2124 client.get_registry_policy() 2125 2126 e.value.response["Error"]["Code"].should.contain("RegistryPolicyNotFoundException") 2127 2128 2129@mock_ecr 2130def test_delete_registry_policy_error_policy_not_exists(): 2131 # given 2132 client = boto3.client("ecr", region_name="eu-central-1") 2133 2134 # when 2135 with pytest.raises(ClientError) as e: 2136 client.delete_registry_policy() 2137 2138 # then 2139 ex = e.value 2140 ex.operation_name.should.equal("DeleteRegistryPolicy") 2141 ex.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400) 2142 ex.response["Error"]["Code"].should.contain("RegistryPolicyNotFoundException") 2143 ex.response["Error"]["Message"].should.equal( 2144 f"Registry policy does not exist in the registry with id '{ACCOUNT_ID}'" 2145 ) 2146 2147 2148@mock_ecr 2149def test_start_image_scan(): 2150 # given 2151 client = boto3.client("ecr", region_name="eu-central-1") 2152 repo_name = "test-repo" 2153 client.create_repository(repositoryName=repo_name) 2154 image_tag = "latest" 2155 image_digest = client.put_image( 2156 repositoryName=repo_name, 2157 imageManifest=json.dumps(_create_image_manifest()), 2158 imageTag="latest", 2159 )["image"]["imageId"]["imageDigest"] 2160 2161 # when 2162 response = client.start_image_scan( 2163 repositoryName=repo_name, imageId={"imageTag": image_tag} 2164 ) 2165 2166 # then 2167 response["registryId"].should.equal(ACCOUNT_ID) 2168 response["repositoryName"].should.equal(repo_name) 2169 response["imageId"].should.equal( 2170 {"imageDigest": image_digest, "imageTag": image_tag} 2171 ) 2172 response["imageScanStatus"].should.equal({"status": "IN_PROGRESS"}) 2173 2174 2175@mock_ecr 2176def test_start_image_scan_error_repo_not_exists(): 2177 # given 2178 region_name = "eu-central-1" 2179 client = boto3.client("ecr", region_name=region_name) 2180 repo_name = "not-exists" 2181 2182 # when 2183 with pytest.raises(ClientError) as e: 2184 client.start_image_scan( 2185 repositoryName=repo_name, imageId={"imageTag": "latest"} 2186 ) 2187 2188 # then 2189 ex = e.value 2190 ex.operation_name.should.equal("StartImageScan") 2191 ex.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400) 2192 ex.response["Error"]["Code"].should.contain("RepositoryNotFoundException") 2193 ex.response["Error"]["Message"].should.equal( 2194 f"The repository with name '{repo_name}' does not exist " 2195 f"in the registry with id '{ACCOUNT_ID}'" 2196 ) 2197 2198 2199@mock_ecr 2200def test_start_image_scan_error_image_not_exists(): 2201 # given 2202 client = boto3.client("ecr", region_name="eu-central-1") 2203 repo_name = "test-repo" 2204 client.create_repository(repositoryName=repo_name) 2205 image_tag = "not-exists" 2206 2207 # when 2208 with pytest.raises(ClientError) as e: 2209 client.start_image_scan( 2210 repositoryName=repo_name, imageId={"imageTag": image_tag} 2211 ) 2212 2213 # then 2214 ex = e.value 2215 ex.operation_name.should.equal("StartImageScan") 2216 ex.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400) 2217 ex.response["Error"]["Code"].should.contain("ImageNotFoundException") 2218 ex.response["Error"]["Message"].should.equal( 2219 f"The image with imageId {{imageDigest:'null', imageTag:'{image_tag}'}} does not exist " 2220 f"within the repository with name '{repo_name}' " 2221 f"in the registry with id '{ACCOUNT_ID}'" 2222 ) 2223 2224 2225@mock_ecr 2226def test_start_image_scan_error_image_tag_digest_mismatch(): 2227 # given 2228 client = boto3.client("ecr", region_name="eu-central-1") 2229 repo_name = "test-repo" 2230 client.create_repository(repositoryName=repo_name) 2231 image_digest = client.put_image( 2232 repositoryName=repo_name, 2233 imageManifest=json.dumps(_create_image_manifest()), 2234 imageTag="latest", 2235 )["image"]["imageId"]["imageDigest"] 2236 image_tag = "not-latest" 2237 2238 # when 2239 with pytest.raises(ClientError) as e: 2240 client.start_image_scan( 2241 repositoryName=repo_name, 2242 imageId={"imageTag": image_tag, "imageDigest": image_digest}, 2243 ) 2244 2245 # then 2246 ex = e.value 2247 ex.operation_name.should.equal("StartImageScan") 2248 ex.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400) 2249 ex.response["Error"]["Code"].should.contain("ImageNotFoundException") 2250 ex.response["Error"]["Message"].should.equal( 2251 f"The image with imageId {{imageDigest:'{image_digest}', imageTag:'{image_tag}'}} does not exist " 2252 f"within the repository with name '{repo_name}' " 2253 f"in the registry with id '{ACCOUNT_ID}'" 2254 ) 2255 2256 2257@mock_ecr 2258def test_start_image_scan_error_daily_limit(): 2259 # given 2260 client = boto3.client("ecr", region_name="eu-central-1") 2261 repo_name = "test-repo" 2262 client.create_repository(repositoryName=repo_name) 2263 image_tag = "latest" 2264 client.put_image( 2265 repositoryName=repo_name, 2266 imageManifest=json.dumps(_create_image_manifest()), 2267 imageTag="latest", 2268 ) 2269 client.start_image_scan(repositoryName=repo_name, imageId={"imageTag": image_tag}) 2270 2271 # when 2272 with pytest.raises(ClientError) as e: 2273 client.start_image_scan( 2274 repositoryName=repo_name, imageId={"imageTag": image_tag} 2275 ) 2276 2277 # then 2278 ex = e.value 2279 ex.operation_name.should.equal("StartImageScan") 2280 ex.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400) 2281 ex.response["Error"]["Code"].should.contain("LimitExceededException") 2282 ex.response["Error"]["Message"].should.equal( 2283 "The scan quota per image has been exceeded. Wait and try again." 2284 ) 2285 2286 2287@mock_ecr 2288def test_describe_image_scan_findings(): 2289 # given 2290 client = boto3.client("ecr", region_name="eu-central-1") 2291 repo_name = "test-repo" 2292 client.create_repository(repositoryName=repo_name) 2293 image_tag = "latest" 2294 image_digest = client.put_image( 2295 repositoryName=repo_name, 2296 imageManifest=json.dumps(_create_image_manifest()), 2297 imageTag="latest", 2298 )["image"]["imageId"]["imageDigest"] 2299 client.start_image_scan(repositoryName=repo_name, imageId={"imageTag": image_tag}) 2300 2301 # when 2302 response = client.describe_image_scan_findings( 2303 repositoryName=repo_name, imageId={"imageTag": image_tag} 2304 ) 2305 2306 # then 2307 response["registryId"].should.equal(ACCOUNT_ID) 2308 response["repositoryName"].should.equal(repo_name) 2309 response["imageId"].should.equal( 2310 {"imageDigest": image_digest, "imageTag": image_tag} 2311 ) 2312 response["imageScanStatus"].should.equal( 2313 {"status": "COMPLETE", "description": "The scan was completed successfully."} 2314 ) 2315 scan_findings = response["imageScanFindings"] 2316 scan_findings["imageScanCompletedAt"].should.be.a(datetime) 2317 scan_findings["vulnerabilitySourceUpdatedAt"].should.be.a(datetime) 2318 scan_findings["findings"].should.equal( 2319 [ 2320 { 2321 "name": "CVE-9999-9999", 2322 "uri": "https://cve.mitre.org/cgi-bin/cvename.cgi?name=CVE-9999-9999", 2323 "severity": "HIGH", 2324 "attributes": [ 2325 {"key": "package_version", "value": "9.9.9"}, 2326 {"key": "package_name", "value": "moto_fake"}, 2327 {"key": "CVSS2_VECTOR", "value": "AV:N/AC:L/Au:N/C:P/I:P/A:P",}, 2328 {"key": "CVSS2_SCORE", "value": "7.5"}, 2329 ], 2330 } 2331 ] 2332 ) 2333 scan_findings["findingSeverityCounts"].should.equal({"HIGH": 1}) 2334 2335 2336@mock_ecr 2337def test_describe_image_scan_findings_error_repo_not_exists(): 2338 # given 2339 region_name = "eu-central-1" 2340 client = boto3.client("ecr", region_name=region_name) 2341 repo_name = "not-exists" 2342 2343 # when 2344 with pytest.raises(ClientError) as e: 2345 client.describe_image_scan_findings( 2346 repositoryName=repo_name, imageId={"imageTag": "latest"} 2347 ) 2348 2349 # then 2350 ex = e.value 2351 ex.operation_name.should.equal("DescribeImageScanFindings") 2352 ex.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400) 2353 ex.response["Error"]["Code"].should.contain("RepositoryNotFoundException") 2354 ex.response["Error"]["Message"].should.equal( 2355 f"The repository with name '{repo_name}' does not exist " 2356 f"in the registry with id '{ACCOUNT_ID}'" 2357 ) 2358 2359 2360@mock_ecr 2361def test_describe_image_scan_findings_error_image_not_exists(): 2362 # given 2363 client = boto3.client("ecr", region_name="eu-central-1") 2364 repo_name = "test-repo" 2365 client.create_repository(repositoryName=repo_name) 2366 image_tag = "not-exists" 2367 2368 # when 2369 with pytest.raises(ClientError) as e: 2370 client.describe_image_scan_findings( 2371 repositoryName=repo_name, imageId={"imageTag": image_tag} 2372 ) 2373 2374 # then 2375 ex = e.value 2376 ex.operation_name.should.equal("DescribeImageScanFindings") 2377 ex.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400) 2378 ex.response["Error"]["Code"].should.contain("ImageNotFoundException") 2379 ex.response["Error"]["Message"].should.equal( 2380 f"The image with imageId {{imageDigest:'null', imageTag:'{image_tag}'}} does not exist " 2381 f"within the repository with name '{repo_name}' " 2382 f"in the registry with id '{ACCOUNT_ID}'" 2383 ) 2384 2385 2386@mock_ecr 2387def test_describe_image_scan_findings_error_scan_not_exists(): 2388 # given 2389 client = boto3.client("ecr", region_name="eu-central-1") 2390 repo_name = "test-repo" 2391 client.create_repository(repositoryName=repo_name) 2392 image_tag = "latest" 2393 client.put_image( 2394 repositoryName=repo_name, 2395 imageManifest=json.dumps(_create_image_manifest()), 2396 imageTag=image_tag, 2397 ) 2398 2399 # when 2400 with pytest.raises(ClientError) as e: 2401 client.describe_image_scan_findings( 2402 repositoryName=repo_name, imageId={"imageTag": image_tag} 2403 ) 2404 2405 # then 2406 ex = e.value 2407 ex.operation_name.should.equal("DescribeImageScanFindings") 2408 ex.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400) 2409 ex.response["Error"]["Code"].should.contain("ScanNotFoundException") 2410 ex.response["Error"]["Message"].should.equal( 2411 f"Image scan does not exist for the image with '{{imageDigest:'null', imageTag:'{image_tag}'}}' " 2412 f"in the repository with name '{repo_name}' " 2413 f"in the registry with id '{ACCOUNT_ID}'" 2414 ) 2415 2416 2417@mock_ecr 2418def test_put_replication_configuration(): 2419 # given 2420 client = boto3.client("ecr", region_name="eu-central-1") 2421 config = { 2422 "rules": [ 2423 {"destinations": [{"region": "eu-west-1", "registryId": ACCOUNT_ID},]}, 2424 ] 2425 } 2426 2427 # when 2428 response = client.put_replication_configuration(replicationConfiguration=config) 2429 2430 # then 2431 response["replicationConfiguration"].should.equal(config) 2432 2433 2434@mock_ecr 2435def test_put_replication_configuration_error_feature_disabled(): 2436 # given 2437 client = boto3.client("ecr", region_name="eu-central-1") 2438 config = { 2439 "rules": [ 2440 { 2441 "destinations": [ 2442 {"region": "eu-central-1", "registryId": "111111111111"}, 2443 ] 2444 }, 2445 { 2446 "destinations": [ 2447 {"region": "eu-central-1", "registryId": "222222222222"}, 2448 ] 2449 }, 2450 ] 2451 } 2452 2453 # when 2454 with pytest.raises(ClientError) as e: 2455 client.put_replication_configuration(replicationConfiguration=config) 2456 2457 # then 2458 ex = e.value 2459 ex.operation_name.should.equal("PutReplicationConfiguration") 2460 ex.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400) 2461 ex.response["Error"]["Code"].should.contain("ValidationException") 2462 ex.response["Error"]["Message"].should.equal("This feature is disabled") 2463 2464 2465@mock_ecr 2466def test_put_replication_configuration_error_same_source(): 2467 # given 2468 region_name = "eu-central-1" 2469 client = boto3.client("ecr", region_name=region_name) 2470 config = { 2471 "rules": [ 2472 {"destinations": [{"region": region_name, "registryId": ACCOUNT_ID}]}, 2473 ] 2474 } 2475 2476 # when 2477 with pytest.raises(ClientError) as e: 2478 client.put_replication_configuration(replicationConfiguration=config) 2479 2480 # then 2481 ex = e.value 2482 ex.operation_name.should.equal("PutReplicationConfiguration") 2483 ex.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400) 2484 ex.response["Error"]["Code"].should.contain("InvalidParameterException") 2485 ex.response["Error"]["Message"].should.equal( 2486 "Invalid parameter at 'replicationConfiguration' failed to satisfy constraint: " 2487 "'Replication destination cannot be the same as the source registry'" 2488 ) 2489 2490 2491@mock_ecr 2492def test_describe_registry(): 2493 # given 2494 client = boto3.client("ecr", region_name="eu-central-1") 2495 2496 # when 2497 response = client.describe_registry() 2498 2499 # then 2500 response["registryId"].should.equal(ACCOUNT_ID) 2501 response["replicationConfiguration"].should.equal({"rules": []}) 2502 2503 2504@mock_ecr 2505def test_describe_registry_after_update(): 2506 # given 2507 client = boto3.client("ecr", region_name="eu-central-1") 2508 config = { 2509 "rules": [ 2510 {"destinations": [{"region": "eu-west-1", "registryId": ACCOUNT_ID}]}, 2511 ] 2512 } 2513 client.put_replication_configuration(replicationConfiguration=config) 2514 2515 # when 2516 response = client.describe_registry() 2517 2518 # then 2519 response["replicationConfiguration"].should.equal(config) 2520