Add unit test for docker_cli_wrapper.py

Fix generate only

Author: angela-ko

tesseract_core/sdk/docker_cli_wrapper.py

@@ -78,8 +78,6 @@ def remove(self, image_id: str) -> None:
                     capture_output=True,
                 )
 
-                self._update_images()
-
             except subprocess.CalledProcessError as ex:
                 raise CLIDockerClient.Errors.ImageNotFound(
                     f"Cannot remove image {image_id}: {ex}"
@@ -136,6 +134,7 @@ def buildx(
                 return None
 
             out_pipe = LogPipe(logging.DEBUG)
+
             with out_pipe as out_pipe_fd:
                 proc = subprocess.run(build_cmd, stdout=out_pipe_fd, stderr=out_pipe_fd)
 
@@ -174,6 +173,8 @@ def buildx(
         def _update_images(self) -> None:
             """Updates the list of images by querying Docker CLI."""
             try:
+                # Reset self.images completely
+                self.images = []
                 image_ids = subprocess.run(
                     ["docker", "images", "-q"],  # List only image IDs
                     capture_output=True,
@@ -182,20 +183,11 @@ def _update_images(self) -> None:
                 )
                 if not image_ids.stdout:
                     images = []
-                else:
-                    images = image_ids.stdout.strip().split("\n")
-
-                # Clean up deleted images.
-                for image in self.images:
-                    if image.id not in images:
-                        self.images.remove(image)
-
-                # Filter list to exclude image ids that are already in self.images.
-                images = [
-                    image_id
-                    for image_id in images
-                    if image_id not in self.images and image_id
-                ]
+                    return
+
+                images = image_ids.stdout.strip().split("\n")
+                # Filter list to exclude empty strings.
+                images = [image_id for image_id in images if image_id]
                 json_dicts = get_docker_metadata(images, is_image=True)
                 for _, json_dict in json_dicts.items():
                     image = self.Image(json_dict)
@@ -337,6 +329,11 @@ def get(self, container: str) -> Container:
                 for _, container_obj in self.list().items():
                     if container_obj.name == container:
                         return container_obj
+
+            if container_obj is None:
+                raise CLIDockerClient.Errors.ContainerError(
+                    f"Container {container} not found."
+                )
             return container_obj
 
         def run(
@@ -426,6 +423,8 @@ def run(
         def _update_containers(self) -> None:
             """Update self.containers."""
             try:
+                # Reset self.containers completely
+                self.containers = {}
                 result = subprocess.run(
                     ["docker", "ps", "-q"],  # List only container IDs
                     capture_output=True,
@@ -434,19 +433,13 @@ def _update_containers(self) -> None:
                 )
                 if not result.stdout:
                     container_ids = []
-                else:
-                    container_ids = result.stdout.strip().split("\n")
-
-                # Check if theres any cleaned up containers.
-                for container_id in list(self.containers.keys()):
-                    if container_id not in container_ids:
-                        del self.containers[container_id]
-                # Filter list to exclude container ids that are already in self.containers
-                # also exclude empty strings.
+                    return
+
+                container_ids = result.stdout.strip().split("\n")
+
+                # Filter list to  exclude empty strings.
                 container_ids = [
-                    container_id
-                    for container_id in container_ids
-                    if container_id not in self.containers and container_id
+                    container_id for container_id in container_ids if container_id
                 ]
                 json_dicts = get_docker_metadata(container_ids)
                 for container_id, json_dict in json_dicts.items():

tests/conftest.py

@@ -83,6 +83,22 @@ def unit_tesseract_path(request) -> Path:
     return UNIT_TESSERACT_PATH / request.param
 
 
+@pytest.fixture()
+def dummy_docker_file(tmpdir):
+    """Create a dummy Dockerfile for testing."""
+    dockerfile_path = tmpdir / "Dockerfile"
+    dockerfile_content = """
+        FROM alpine
+        CMD echo "Hello, Tesseract!" && sleep infinity
+
+        # Set environment variables
+        ENV TESSERACT_NAME="dummy-tesseract" \
+        """
+    with open(dockerfile_path, "w") as f:
+        f.write(dockerfile_content)
+    return dockerfile_path
+
+
 @pytest.fixture(scope="session")
 def dummy_tesseract_location():
     """Return the dummy tesseract location."""

tests/endtoend_tests/test_docker_cli_wrapper.py

@@ -0,0 +1,93 @@
+"""End to end tests for docker cli wrapper."""
+
+from common import image_exists
+
+from tesseract_core.sdk.docker_cli_wrapper import CLIDockerClient
+
+
+def test_create_image(docker_client, dummy_tesseract_location, dummy_docker_file):
+    """Test image building, getting, and removing."""
+    image, image1 = None, None
+    # Create an image
+    try:
+        image_name = "tesseract_create_image_test:dummy"
+
+        docker_client.images.buildx(
+            dummy_tesseract_location, image_name, dummy_docker_file
+        )
+        image = docker_client.images.get(image_name)
+        assert image is not None
+        assert image.name == image_name
+
+        # Create a second image
+        image1_name = "tesseract_core_test:dummy1"
+        docker_client.images.buildx(
+            dummy_tesseract_location, image1_name, dummy_docker_file
+        )
+        image1 = docker_client.images.get(image1_name)
+        assert image1 is not None
+        assert image1.name == image1_name
+
+        # Check that image and image1 both exist
+        assert image_exists(docker_client, image.name)
+        assert image_exists(docker_client, image1.name)
+
+    finally:
+        # Clean up the images
+        try:
+            if image:
+                docker_client.images.remove(image.name)
+            if image1:
+                docker_client.images.remove(image1.name)
+
+        except CLIDockerClient.Errors.ImageNotFound:
+            pass
+
+        # Check that images are removed
+        assert not image_exists(docker_client, image.name)
+        assert not image_exists(docker_client, image1.name)
+
+
+def test_create_container(docker_client, dummy_tesseract_location, dummy_docker_file):
+    """Test container creation, run, logs, and remove."""
+    # Create a container
+    image, container = None, None
+    try:
+        image_name = "tesseract_create_container_test:dummy"
+
+        docker_client.images.buildx(
+            dummy_tesseract_location, image_name, dummy_docker_file
+        )
+        image = docker_client.images.get(image_name)
+        assert image is not None
+        assert image.name == image_name
+
+        container = docker_client.containers.run(image_name, [], detach=True)
+        assert container is not None
+        container_id = container.id
+
+        container_get = docker_client.containers.get(container_id)
+        assert container_get is not None
+        assert container_get.id == container.id
+
+        logs = container.logs()
+        assert logs == ("Hello, Tesseract!\n", "")
+
+    finally:
+        try:
+            # Clean up the container
+            if container:
+                container.remove(v=True, force=True)
+
+            # Check that the container is removed
+            containers = docker_client.containers.list()
+            assert container.id not in containers.keys()
+        except CLIDockerClient.Errors.ContainerError:
+            pass
+
+        # Image has to be removed after container is removed
+        try:
+            if image:
+                docker_client.images.remove(image.id)
+        except CLIDockerClient.Errors.ImageNotFound:
+            pass

tests/sdk_tests/test_engine.py

@@ -50,8 +50,11 @@ def test_build_image(
     if generate_only:
         # Check stdout if it contains the correct docker build command
         assert got is None
-        command = f"docker buildx build --load --no-cache --tag {image_name} --file {dockerfile_path} {tmpdir}"
-        assert command in caplog.text
+        # Split into 2 parts because label is timestamp based
+        command1 = f"docker buildx build --load --tag {image_name} --label"
+        command2 = f"--file {dockerfile_path} {tmpdir}"
+        assert command1 in caplog.text
+        assert command2 in caplog.text
     else:
         assert got.attrs == mocked_docker.images.get(image_name).attrs