docker_runner tests

0xricksanchez · 0xricksanchez · Oct 25, 2022 · Oct 17, 2022 · Oct 17, 2022 · Oct 17, 2022
commit bf7dfd00c297a8d5313109e4a9bc5e73bffb1cf0
diff --git a/src/docker_runner.py b/src/docker_runner.py
@@ -6,6  6,7 @@
 from pathlib import Path
 
 import docker
 from docker.models.images import Image
 from fabric import Connection
 from loguru import logger
 
@@ -67,24  68,27 @@ def build_image(self, dockerfile=None, buildargs=None, image_tag=None):
         buildargs = buildargs if buildargs else self.buildargs
         tag = image_tag if image_tag else self.tag
         nocache = True if self.update_containers else False
-        for log_entry in self.cli.build(
-            path=str(self.dockerfile_ctx), dockerfile=dockerfile, tag=tag, decode=True, buildargs=buildargs, nocache=nocache, rm=True
-        ):
-            v = next(iter(log_entry.values()))
-            if isinstance(v, str):
-                v = " ".join(v.strip().split())
-                if v and not self.update_containers:
-                    logger.debug(v)
-                elif v and self.update_containers:
-                    logger.info(v)
-        if self.update_containers:
-            self.cli.prune_images(filters={"dangling": True})
-
-    def get_image(self, tag=None):
         try:
             for log_entry in self.cli.build(
                 path=str(self.dockerfile_ctx), dockerfile=dockerfile, tag=tag, decode=True, buildargs=buildargs, nocache=nocache, rm=True
             ):
                 v = next(iter(log_entry.values()))
                 if isinstance(v, str):
                     v = " ".join(v.strip().split())
                     if v and not self.update_containers:
                         logger.debug(v)
                     elif v and self.update_containers:
                         logger.info(v)
             if self.update_containers:
                 self.cli.prune_images(filters={"dangling": True})
             return 0
         except docker.errors.APIError:
             return 1
 
     def get_image(self, tag=None) -> Image:
         to_check = tag if tag else self.tag
         try:
-            image = self.client.images.get(to_check)
-            return image
             return self.client.images.get(to_check)
         except docker.errors.ImageNotFound:
             return None
 
@@ -94,13  98,13 @@ def is_base_image(self) -> bool:
         else:
             return False
 
-    def build_base_img(self) -> None:
-        self.build_image(dockerfile=self.dockerfile_base_img, image_tag=self.tag_base_image)
     def build_base_img(self) -> int:
         return self.build_image(dockerfile=self.dockerfile_base_img, image_tag=self.tag_base_image)
 
-    def run(self, check_existing: bool = False) -> None:
     def run(self, check_existing: bool = False) -> int:
         if self.update_containers:
             self.build_image()
-            return
             return 1
         if check_existing:
             self.check_existing()
         if not self.image:
@@ -111,6  115,7 @@ def run(self, check_existing: bool = False) -> None:
             self.build_image()
             self.image = self.get_image()
         self.run_container()
         return 0
 
     def run_container(self) -> None:
         pass
@@ -128,15  133,20 @@ def wait_for_container(self) -> dict:
             exit(-1)
         return ret
 
-    def check_existing(self) -> None:
     def pull_image(self, repo: str, tag: None) -> Image:
         tag = tag if tag else self.tag
         return self.client.images.pull(repo, tag=tag)
 
     def check_existing(self) -> Image:
         if self.update_containers:
-            return
             return None
         if self.force_rebuild:
             logger.info(f"Force-rebuilding {type(self).__name__}")
             self.image = None
-        else:
-            self.image = self.get_image()
-            if self.image and self.skip_prompts:
-                return
-            elif self.image and not is_reuse(self.image.tags[0]):
-                self.image = None
             return self.image
         self.image = self.get_image()
         if self.image and self.skip_prompts:
             return self.image
         elif self.image and not is_reuse(self.image.tags[0]):
             self.image = None
             return self.image
diff --git a/src/tests/files/.dockerfile_test b/src/tests/files/.dockerfile_test
@@ -0,0  1,3 @@
 FROM busybox:latest
 LABEL maintainer="Christopher Krah <[email protected]>"
 
diff --git a/src/tests/test_docker_runner.py b/src/tests/test_docker_runner.py
@@ -1,5  1,6 @@
 from docker.models.containers import Container
 from ..docker_runner import DockerRunner
 from pathlib import Path
 
 GENERIC_ARGS = {
     "skip_prompts": False,
@@ -14,29  15,37 @@
     "status": "unpack",
 }
 
 DOCKERFILE = Path("src/tests/files/.dockerfile_test")
 
 
 class MockDockerRunner(DockerRunner):
     def __init__(self, **kwargs) -> None:
         super().__init__(**kwargs)
         self.force_rebuild = False
 
-    def run_container(self, entrypoint_cmd: str) -> Container:
-        self.container = self.client.containers.run("ubuntu:latest", entrypoint_cmd, detach=True)
     def run_container(self, repo: str, tag: str, entrypoint_cmd: str) -> Container:
         self.container = self.client.containers.run(f"{repo}:{tag}", entrypoint_cmd, detach=True)
         return self.container
 
 
 def test_pull_image() -> None:
     dr = DockerRunner(**GENERIC_ARGS | MOCK_UNPACKER_RES)
     assert dr.pull_image("busybox", "latest") is not None
 
 
 def test_get_image_exists() -> None:
     dr = DockerRunner(**GENERIC_ARGS | MOCK_UNPACKER_RES)
-    assert dr.get_image(tag="like_dbg_base") != None
     assert dr.get_image(tag="busybox") is not None
 
 
 def test_get_image_not_exists() -> None:
     dr = DockerRunner(**GENERIC_ARGS | MOCK_UNPACKER_RES)
-    assert dr.get_image(tag="like_dbg_not_exist") == None
     assert dr.get_image(tag="like_dbg_not_exist") is None
 
 
 def test_is_base_image() -> None:
     dr = DockerRunner(**GENERIC_ARGS | MOCK_UNPACKER_RES)
-    dr.tag_base_image = "like_dbg_base"
     dr.tag_base_image = "busybox"
     assert dr.is_base_image() is True
 
 
@@ -46,34  55,99 @@ def test_is_not_base_image() -> None:
     assert dr.is_base_image() is False
 
 
 def test_build_base_image() -> None:
     dr = DockerRunner(**GENERIC_ARGS | MOCK_UNPACKER_RES)
     dr.dockerfile_base_image = str(DOCKERFILE)
     dr.tag_base_image = "a_base_img"
     assert dr.build_base_img() == 0
 
 
 def test_list_running_containers_is_empty() -> None:
     dr = DockerRunner(**GENERIC_ARGS | MOCK_UNPACKER_RES)
     assert dr.list_running_containers() == []
 
 
 def test_run_container_sucess() -> None:
     mdr = MockDockerRunner(**GENERIC_ARGS | MOCK_UNPACKER_RES)
-    assert mdr.run_container("/bin/true") != None
     assert mdr.run_container("busybox", "latest", "/bin/true") is not None
 
 
 def test_list_running_containers_is_one() -> None:
     mdr = MockDockerRunner(**GENERIC_ARGS | MOCK_UNPACKER_RES)
-    mdr.run_container("tail -f /dev/null")
     mdr.run_container("ubuntu", "latest", "tail -f /dev/null")
     assert len(mdr.list_running_containers()) > 0
     mdr.stop_container()
 
 
 def test_stop_container() -> None:
     mdr = MockDockerRunner(**GENERIC_ARGS | MOCK_UNPACKER_RES)
     old_len = len(mdr.list_running_containers())
-    mdr.run_container("tail -f /dev/null")
     mdr.run_container("ubuntu", "latest", "tail -f /dev/null")
     mdr.stop_container()
     new_len = len(mdr.list_running_containers()) > 0
     assert old_len == new_len
 
 
 def test_wait_for_container() -> None:
     mdr = MockDockerRunner(**GENERIC_ARGS | MOCK_UNPACKER_RES)
-    mdr.run_container("/bin/true")
     mdr.run_container("busybox", "latest", "/bin/true")
     ret = mdr.wait_for_container()
     assert ret["StatusCode"] == 0
 
 
 def test_check_existing_ok() -> None:
     mdr = MockDockerRunner(**GENERIC_ARGS | MOCK_UNPACKER_RES)
     mdr.skip_prompts = True
     mdr.tag = "busybox"
     assert mdr.check_existing().tags == ["busybox:latest"]
 
 
 def test_check_existing_update_containers() -> None:
     mdr = MockDockerRunner(**GENERIC_ARGS | MOCK_UNPACKER_RES)
     mdr.update_containers = True
     assert mdr.check_existing() is None
 
 
 def test_check_existing_force_rebuild() -> None:
     mdr = MockDockerRunner(**GENERIC_ARGS | MOCK_UNPACKER_RES)
     mdr.update_containers = False
     mdr.force_rebuild = True
     assert mdr.check_existing() is None
 
 
 def test_build_image() -> None:
     dr = DockerRunner(**GENERIC_ARGS | MOCK_UNPACKER_RES)
     dr.dockerfile = str(DOCKERFILE)
     assert dr.build_image() == 0
 
 
 def test_build_image_update() -> None:
     dr = DockerRunner(**GENERIC_ARGS | MOCK_UNPACKER_RES)
     dr.dockerfile = str(DOCKERFILE)
     dr.update_containers = True
     assert dr.build_image() == 0
 
 
 def test_run_update() -> None:
     dr = DockerRunner(**GENERIC_ARGS | MOCK_UNPACKER_RES)
     dr.dockerfile = str(DOCKERFILE)
     dr.update_containers = True
     assert dr.run() == 1
 
 
 def test_run_check_existing() -> None:
     dr = DockerRunner(**GENERIC_ARGS | MOCK_UNPACKER_RES)
     dr.dockerfile = str(DOCKERFILE)
     dr.update_containers = False
     dr.skip_prompts = True
     dr.force_rebuild = False
     dr.tag = "busybox"
     assert dr.run(check_existing=True) == 0
 
 
 def test_run_no_image() -> None:
     dr = DockerRunner(**GENERIC_ARGS | MOCK_UNPACKER_RES)
     dr.dockerfile = str(DOCKERFILE)
     dr.tag = "busybox"
     dr.update_containers = False
     assert dr.run(check_existing=False) == 0
Original file line number	Diff line number	Diff line change
		@@ -0,0 1,3 @@
		FROM busybox:latest
		LABEL maintainer="Christopher Krah <[email protected]>"