Tests done

Author: meyertst-aws

python/example_code/ecr/ecr_getting_started.py

@@ -7,7 +7,8 @@
 Shows how to use the AWS SDK for Python (Boto3) with Amazon Elastic Container Registry (Amazon ECR) to perform
 basic operations.
 
-This Boto3 code example requires an IAM Role that has permissions to interact with the Amazon ECR service.
+To demonstrate granting permissions with a policy, an AWS Identity and Access Management (IAM) role Amazon Resource Name (ARN)
+can be passed as a script argument.
 
 To create an IAM role, see:
 
@@ -27,7 +28,7 @@
 
 script_dir = os.path.dirname(os.path.abspath(__file__))
 
-# Add relative path to include ecrWrapper.
+# Add relative path to include ecr_wrapper.
 sys.path.append(os.path.dirname(script_dir))
 from ecr_wrapper import ECRWrapper
 
@@ -71,7 +72,7 @@ def __init__(
         self.ecr_wrapper = ecr_wrapper
         self.docker_client = docker_client
         self.tag = "echo-text"
-        self.repo_name = "ecr-basics"
+        self.repository_name = "ecr-basics"
         self.docker_image = None
         self.full_tag_name = None
         self.repository = None
@@ -110,8 +111,8 @@ def run(self, role_arn: str) -> None:
 to store, manage, and deploy Docker container images.
         """
         )
-        print(f"Creating a repository named {self.repo_name}")
-        self.repository = ecr_wrapper.create_repository(self.repo_name)
+        print(f"Creating a repository named {self.repository_name}")
+        self.repository = self.ecr_wrapper.create_repository(self.repository_name)
         print(f"The ARN of the ECR repository is {self.repository['repositoryArn']}")
         repository_uri = self.repository["repositoryUri"]
         press_enter_to_continue()
@@ -128,7 +129,7 @@ def run(self, role_arn: str) -> None:
         )
         print(f"Building a docker image from 'docker_files/Dockerfile'")
         self.full_tag_name = f"{repository_uri}:{self.tag}"
-        self.docker_image = docker_client.images.build(
+        self.docker_image = self.docker_client.images.build(
             path="docker_files", tag=self.full_tag_name
         )[0]
         self.docker_image.tag(repository=repository_uri, tag=self.tag)
@@ -153,6 +154,7 @@ def run(self, role_arn: str) -> None:
 repository.
         """
             )
+
             self.grant_role_download_access(role_arn)
             print(f"Download access granted to the IAM role ARN {role_arn}")
             press_enter_to_continue()
@@ -165,7 +167,8 @@ def run(self, role_arn: str) -> None:
 Now we will retrieve the ECR policy to ensure it was successfully set.
             """
             )
-            policy_text = ecr_wrapper.get_repository_policy(self.repo_name)
+
+            policy_text = self.ecr_wrapper.get_repository_policy(self.repository_name)
             print("Policy Text:")
             print(f"{policy_text}")
             press_enter_to_continue()
@@ -184,7 +187,8 @@ def run(self, role_arn: str) -> None:
 ECR repository, such as pushing, pulling, or managing your Docker images.
         """
         )
-        authorization_token = ecr_wrapper.get_authorization_token()
+
+        authorization_token = self.ecr_wrapper.get_authorization_token()
         print("Authorization token retrieved.")
         press_enter_to_continue()
         print_dashes()
@@ -199,7 +203,9 @@ def run(self, role_arn: str) -> None:
 correct container image from the ECR repository.
         """
         )
-        repository_descriptions = ecr_wrapper.describe_repositories([self.repo_name])
+        repository_descriptions = self.ecr_wrapper.describe_repositories(
+            [self.repository_name]
+        )
         repository_uri = repository_descriptions[0]["repositoryUri"]
         print(f"Repository URI found: {repository_uri}")
         press_enter_to_continue()
@@ -229,14 +235,12 @@ def run(self, role_arn: str) -> None:
 
 The method uses the authorization token to create an `AuthConfig` object, which is used to authenticate
 the Docker client when pushing the image. Finally, the method tags the Docker image with the specified
-repository name and image tag, and then pushes the image to the ECR repository using the Docker client.
+repository name and image image_tag, and then pushes the image to the ECR repository using the Docker client.
         """
         )
         decoded_authorization = base64.b64decode(authorization_token).decode("utf-8")
         username, password = decoded_authorization.split(":")
 
-        print(f"username {username}\n{password}")
-
         resp = self.docker_client.api.push(
             repository=repository_uri,
             auth_config={"username": username, "password": password},
@@ -250,7 +254,9 @@ def run(self, role_arn: str) -> None:
         print_dashes()
 
         print("* Verify if the image is in the ECR Repository.")
-        image_descriptions = ecr_wrapper.describe_images(self.repo_name, [self.tag])
+        image_descriptions = self.ecr_wrapper.describe_images(
+            self.repository_name, [self.tag]
+        )
         if len(image_descriptions) > 0:
             print("Image found in ECR Repository.")
         else:
@@ -273,7 +279,7 @@ def run(self, role_arn: str) -> None:
 
 2. Describe the image using this command:
 
-   aws ecr describe-images --repository-name {self.repo_name} --image-ids imageTag={self.tag}
+   aws ecr describe-images --repository-name {self.repository_name} --image-ids imageTag={self.tag}
 
 3. Run the Docker container and view the output using this command:
 
@@ -291,11 +297,11 @@ def cleanup(self, ask: bool):
         if self.repository is not None and (
             not ask
             or q.ask(
-                f"Would you like to delete the ECR repository '{self.repo_name}? (y/n) "
+                f"Would you like to delete the ECR repository '{self.repository_name}? (y/n) "
             )
         ):
-            print(f"Deleting the ECR repository '{self.repo_name}'.")
-            self.ecr_wrapper.delete_repository(self.repo_name)
+            print(f"Deleting the ECR repository '{self.repository_name}'.")
+            self.ecr_wrapper.delete_repository(self.repository_name)
 
         if self.full_tag_name is not None and (
             not ask
@@ -325,7 +331,9 @@ def grant_role_download_access(self, role_arn: str):
             ],
         }
 
-        self.ecr_wrapper.set_repository_policy(self.repo_name, json.dumps(policy_json))
+        self.ecr_wrapper.set_repository_policy(
+            self.repository_name, json.dumps(policy_json)
+        )
 
     # snippet-end:[python.example_code.ecr.grant_role_download_access]
 
@@ -350,7 +358,10 @@ def put_expiration_policy(self):
             ]
         }
 
-        self.ecr_wrapper.put_lifecycle_policy(self.repo_name, json.dumps(policy_json))
+        self.ecr_wrapper.put_lifecycle_policy(
+            self.repository_name, json.dumps(policy_json)
+        )
+
     # snippet-end:[python.example_code.ecr.put_expiration_policy]
 
 
@@ -374,15 +385,15 @@ def put_expiration_policy(self):
     no_art = args.no_art
     iam_role_arn = args.iam_role_arn
     demo = None
-    docker_client = None
+    a_docker_client = None
     try:
-        docker_client = docker.from_env()
-        if not docker_client.ping():
+        a_docker_client = docker.from_env()
+        if not a_docker_client.ping():
             raise docker.errors.DockerException("Docker is not running.")
     except docker.errors.DockerException as err:
         logging.error(
             """
-        The Python Docker could not be created. 
+        The Python Docker client could not be created. 
         Do you have Docker installed and running?
         Here is the error message:
         %s
@@ -391,8 +402,8 @@ def put_expiration_policy(self):
         )
         sys.exit("Error with Docker.")
     try:
-        ecr_wrapper = ECRWrapper.from_client()
-        demo = ECRGettingStarted(ecr_wrapper, docker_client)
+        an_ecr_wrapper = ECRWrapper.from_client()
+        demo = ECRGettingStarted(an_ecr_wrapper, a_docker_client)
         demo.run(iam_role_arn)
 
     except Exception as exception:

python/example_code/ecr/ecr_wrapper.py

@@ -52,7 +52,7 @@ def create_repository(self, repository_name: str) -> dict[str, any]:
                 response = self.ecr_client.describe_repositories(
                     repositoryNames=[repository_name]
                 )
-                return response["repository"]
+                return self.describe_repositories([repository_name])[0]
             else:
                 logger.error(
                     "Error creating repository %s. Here's why %s",

python/example_code/ecr/hello/hello_ecr.py

@@ -6,7 +6,8 @@
 import argparse
 from boto3 import client
 
-def hello_ecr(ecr_client : client, repository_name :str) -> None:
+
+def hello_ecr(ecr_client: client, repository_name: str) -> None:
     """
     Use the AWS SDK for Python (Boto3) to create an Amazon Elastic Container Registry (Amazon ECR)
     client and list the images in a repository.
@@ -17,9 +18,13 @@ def hello_ecr(ecr_client : client, repository_name :str) -> None:
                              the low-level Amazon ECR service API.
     :param repository_name: The name of an Amazon ECR repository in your account.
     """
-    print(f"Hello, Amazon ECR! Let's list some images in the repository '{repository_name}':\n")
+    print(
+        f"Hello, Amazon ECR! Let's list some images in the repository '{repository_name}':\n"
+    )
     paginator = ecr_client.get_paginator("list_images")
-    page_iterator = paginator.paginate(repositoryName=repository_name, PaginationConfig={"MaxItems": 10})
+    page_iterator = paginator.paginate(
+        repositoryName=repository_name, PaginationConfig={"MaxItems": 10}
+    )
 
     image_names: [str] = []
     for page in page_iterator:
@@ -32,17 +37,14 @@ def hello_ecr(ecr_client : client, repository_name :str) -> None:
 
 
 if __name__ == "__main__":
-    parser = argparse.ArgumentParser(
-        description="Run hello Amazon ECR."
-    )
+    parser = argparse.ArgumentParser(description="Run hello Amazon ECR.")
     parser.add_argument(
         "--repository-name",
         type=str,
         help="the name of an Amazon ECR repository in your account.",
-        required=True
+        required=True,
     )
     args = parser.parse_args()
 
-
     hello_ecr(boto3.client("ecr"), args.repository_name)
 # snippet-end:[python.example_code.ecr.Hello]

python/example_code/ecr/test/conftest.py

@@ -0,0 +1,111 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""
+Contains common test fixtures used to run unit tests.
+"""
+
+import sys
+
+import boto3
+import pytest
+import os
+
+
+script_dir = os.path.dirname(os.path.abspath(__file__))
+
+sys.path.append(script_dir)
+import ecr_getting_started
+from ecr_wrapper import ECRWrapper
+
+# Add relative path to include demo_tools in this code example without need for setup.
+sys.path.append(os.path.join(script_dir, "../.."))
+
+from test_tools.fixtures.common import *
+
+
+class MockDockerImage:
+    def __init__(self):
+        self.repository = None
+        self.image_tag = (None,)
+        self.path = None
+
+    def tag(self, repository: str, tag: str) -> None:
+        assert repository == self.repository, "MockDockerImage.tag"
+        assert tag == self.image_tag, "MockDockerImage.tag"
+
+    def build(self, path: str, tag: str) -> tuple["MockDockerImage", None]:
+        assert path == self.path, "MockDockerImage.build"
+        assert tag == f"{self.repository}:{self.image_tag}", "MockDockerImage.build"
+        return self, None
+
+    def remove(self, tag: str) -> None:
+        assert tag == f"{self.repository}:{self.image_tag}", "MockDockerImage.remove"
+        pass
+
+
+class MockDockerAPI:
+    def __init__(self):
+        self.repository = None
+        self.tag = (None,)
+        self.username = None
+        self.password = None
+
+    def push(
+        self,
+        repository: str,
+        auth_config: dict[str, str],
+        tag: str,
+        stream: bool,
+        decode: bool,
+    ) -> list[str]:
+        assert repository == self.repository, "MockDockerAPI.push"
+        assert auth_config["username"] == self.username, "MockDockerAPI.push"
+        assert auth_config["password"] == self.password, "MockDockerAPI.push"
+        assert tag == self.tag, "MockDockerAPI.push"
+        assert stream == True, "MockDockerAPI.push"
+        assert decode == True, "MockDockerAPI.push"
+        return ["pushing", "push complete"]
+
+
+class MockDockerClient:
+    def __init__(self):
+        self.images = MockDockerImage()
+        self.api = MockDockerAPI()
+
+    def set_mocking_values(self, repository, tag, path, username, password):
+        self.images.repository = repository
+        self.images.image_tag = tag
+        self.images.path = path
+        self.api.repository = repository
+        self.api.tag = tag
+        self.api.username = username
+        self.api.password = password
+
+
+class ScenarioData:
+    def __init__(
+        self,
+        ecr_client,
+        mock_docker_client,
+        ecr_stubber,
+    ):
+        self.ecr_client = ecr_client
+        self.mock_docker_client = mock_docker_client
+        self.ecr_stubber = ecr_stubber
+        self.scenario = ecr_getting_started.ECRGettingStarted(
+            ecr_wrapper=ECRWrapper(self.ecr_client),
+            docker_client=self.mock_docker_client,
+        )
+
+
+@pytest.fixture
+def scenario_data(make_stubber):
+    ecr_client = boto3.client("ecr")
+    mock_docker_client = MockDockerClient()
+    ecr_stubber = make_stubber(ecr_client)
+    return ScenarioData(
+        ecr_client,
+        mock_docker_client,
+        ecr_stubber,
+    )