All tests implemented. Formatt, lint & typecheck

src/node_deployer/__main__.py

@@ -1,17 +1,20 @@
 #!/usr/bin/env python
 
+
 def main() -> None:
-    """Entry point for the CLI
+    """Entry point for the CLI"""
-    """
     from .config import config
+
     config.update_config("cli")
     from .node_deployer import app
+
     app()
 
+
 def debug() -> None:
-    """Entry point for the debug CLI
+    """Entry point for the debug CLI"""
-    """
     from .config import config
+
     config.update_config("debug")
     from .node_deployer import app
 
@@ -31,5 +34,6 @@ def debug() -> None:
 
     app()
 
+
 if __name__ == "__main__":
     main()

src/node_deployer/autoignition.py

@@ -91,7 +91,7 @@ def convert_json_via_fuelignition(
         f.write(container_image.read())
 
 
-def build_fuelignition() -> docker.models.images.Image: # type: ignore
+def build_fuelignition() -> docker.models.images.Image:  # type: ignore
     """Builds the fuel-ignition docker image
 
     Returns:
@@ -148,7 +148,8 @@ def json_to_img(
     json_path: Annotated[
         Path,
         typer.Option(
-            "--json-path", "-i",
+            "--json-path",
+            "-i",
             help="The fuel-ignition json for configuring the disk image",
             prompt=True,
             exists=True,
@@ -158,7 +159,8 @@ def json_to_img(
     img_path: Annotated[
         Path,
         typer.Option(
-            "--img-path", "-o",
+            "--img-path",
+            "-o",
             help="The file to output the disk image to",
             prompt=True,
             dir_okay=False,
@@ -176,7 +178,7 @@ def json_to_img(
             flag_value=True,
             expose_value=config.DEBUG,
             hidden=not config.DEBUG,
-        )
+        ),
     ] = False,
 ) -> None:
     """Converts a fuel-ignition json file to an ignition disk image file

src/node_deployer/config.py

@@ -9,15 +9,17 @@ import tomllib
 CLIENT = docker.from_env(version="auto")
 MAX_PORT: int = 65535
 
+
 def __get_project_root():
     r = Path(__file__)
     while r.name != "src":
         r = r.parent
     return r.parent
 
+
 PROJECT_ROOT: Path = __get_project_root()
 
-ConfigLabel = Union[str, list[str]] # After PEP695 support: type ConfigLabel = str | list[str]
+ConfigLabel = Union[str, list[str]]  # After PEP695 support: type ConfigLabel = str | list[str]
 
 
 class Config(SimpleNamespace):
@@ -82,7 +84,8 @@ class Config(SimpleNamespace):
             "FUELIGNITION_BUILD_DIR", self.FUELIGNITION_BUILD_DIR
         )
         config["DOCKERFILE_DIR"] = src_dir / config.get("DOCKERFILE_DIR", self.DOCKERFILE_DIR)
-        config["CWD_MOUNT"] = docker.types.Mount(  # type: ignore <- I really wish docker-py had typeshed stubs
+        # I really wish docker-py had typeshed stubs
+        config["CWD_MOUNT"] = docker.types.Mount(  # type: ignore
             target=str(cwd_mountdir),
             source=str(PROJECT_ROOT),
             type="bind",

src/node_deployer/create_disk.py

@@ -168,7 +168,7 @@ def create_ignition_disk(
             is_flag=True,
             flag_value=True,
             hidden=not config.DEBUG,
-        )
+        ),
     ] = False,
 ) -> None:
     """Creates an ignition image and writes it to the specified disk
@@ -204,13 +204,13 @@ def create_ignition_disk(
         raise typer.BadParameter("No disk specified")
 
     create_img(
-        hostname = hostname,
+        hostname=hostname,
-        password = password,
+        password=password,
-        switch_ip = switch_ip,
+        switch_ip=switch_ip,
-        switch_port = switch_port,
+        switch_port=switch_port,
-        swarm_token = swarm_token,
+        swarm_token=swarm_token,
-        img_path = config.BUILD_DIR / "ignition.img",
+        img_path=config.BUILD_DIR / "ignition.img",
-        debug = debug,
+        debug=debug,
     )
     valid, response = validate()
     if not valid:

src/node_deployer/create_img.py

@@ -163,7 +163,7 @@ def create_img(
             flag_value=True,
             hidden=not config.DEBUG,
         ),
-     ] = False,
+    ] = False,
 ) -> None:
     """Creates an ignition image for a node that will automatically join a swarm
 

src/node_deployer/debug.py

@@ -8,7 +8,8 @@ from .config import config
 
 
 def get_debug_f(f: Callable) -> Callable:
-    import snoop # type: ignore
+    import snoop  # type: ignore
+
     return wraps(f)(snoop.snoop(**config.snoop["snoop"])(f))
 
 
@@ -43,7 +44,7 @@ def debug_guard(f: Callable) -> Callable:
         debug_f = get_debug_f(f)
         if kwargs.get("debug", False):
             # Snoop depth is set to compensate for wrapper stack frames
-            return debug_f(*args, **kwargs) # noqa: F821 #* ss is installed in debug_mode
+            return debug_f(*args, **kwargs)  # noqa: F821 #* ss is installed in debug_mode
         else:
             return f(*args, **kwargs)
 

src/node_deployer/node_deployer.py

@@ -20,18 +20,11 @@ app = typer.Typer(
 )
 
 # Register commands
-app.command(
+app.command(help=str(create_ignition_disk.__doc__).split("Args:")[0].strip(), **cmd_params)(
-    help = str(create_ignition_disk.__doc__).split("Args:")[0].strip(),
+    create_ignition_disk
-    **cmd_params
+)
-)(create_ignition_disk)
+app.command(help=str(create_img.__doc__).split("Args:")[0].strip(), **cmd_params)(create_img)
-app.command(
+app.command(help=str(json_to_img.__doc__).split("Args:")[0].strip(), **cmd_params)(json_to_img)
-    help = str(create_img.__doc__).split("Args:")[0].strip(),
-    **cmd_params
-)(create_img)
-app.command(
-    help = str(json_to_img.__doc__).split("Args:")[0].strip(),
-    **cmd_params
-)(json_to_img)
 
 if __name__ == "__main__":
     config.update_config("cli")

src/node_deployer/utils.py

@@ -1,6 +1,6 @@
 from functools import wraps
 from pathlib import Path
-from typing import Callable
+from typing import Callable, List
 import docker
 
 from .config import config
@@ -52,7 +52,7 @@ def next_free_tcp_port(port: int) -> int:
     Returns:
         int: The next free port
     """
-    ports = []
+    ports: List[int] = []
     try:
         containers = config.CLIENT.containers.list(all=True)
         ports = []
@@ -63,15 +63,14 @@ def next_free_tcp_port(port: int) -> int:
             for x in list(container.ports.values())[0]:
                 ports.append(int(x["HostPort"]))
     except docker.errors.NotFound:  # type: ignore
-        #* This error is raised if container list changes between getting the list and
+        # * This error is raised if container list changes between getting the list and
-        #* getting the ports. If this happens, just try again
+        # * getting the ports. If this happens, just try again
         return next_free_tcp_port(port)
     if not ports:
         return port
-    ports = set(ports)
+    unique_ports = set(ports)
-    while port in ports:
+    while port in unique_ports:
         port += 1
         if port > 65535:
             raise ValueError("No free ports")
     return port
-    

tests/data/node_deployer/config.ign

@@ -0,0 +1,75 @@
+{
+  "ignition": {
+    "version": "3.2.0"
+  },
+  "passwd": {
+    "users": [
+      {
+        "name": "root"
+      }
+    ]
+  },
+  "storage": {
+    "files": [
+      {
+        "path": "/root/join_swarm.json",
+        "mode": 420,
+        "overwrite": true,
+        "contents": {
+          "source": "data:text/plain;charset=utf-8;base64,eyJTV0lUQ0hfSVBfQUREUkVTUyI6ICIxOTIuMTY4LjEuMSIsICJTV0lUQ0hfUE9SVCI6IDQyLCAiU1dBUk1fVE9LRU4iOiAiU1dNVEtOLTEtVEhJU0lTQVRFU1RTV0FSTVRPS0VORk9SVEVTVElOR1BVUlBPU0VTQU5EVEhBVE1FQU5TSVRORUVEU1RPQkVRVUlURUxPTkcifQ=="
+        }
+      },
+      {
+        "path": "/root/join_swarm.sh",
+        "mode": 420,
+        "overwrite": true,
+        "contents": {
+          "source": "data:text/plain;charset=utf-8;base64,IyEvYmluL2Jhc2gKCmlmIFtbICRFVUlEIC1uZSAwIF1dOyB0aGVuCiAgIGVjaG8gIlRoaXMgc2NyaXB0IG11c3QgYmUgcnVuIGFzIHJvb3QiIAogICBleGl0IDEKZmkKCiMgTG9hZCB0aGUgY29uZmlnIGZpbGUgaW50byB2YXJpYWJsZXMKZXZhbCAiJChqcSAtciAndG9fZW50cmllc1tdIHwgImV4cG9ydCBcKC5rZXkpPVwoLnZhbHVlIHwgQHNoKSInIC9yb290L2pvaW5fc3dhcm0uanNvbikiCgppZiBbWyAkKGRvY2tlciBpbmZvIHwgZ3JlcCBTd2FybSB8IGF3ayAne3ByaW50ICQyfScpID09ICJpbmFjdGl2ZSIgXV07IHRoZW4KICAgIGRvY2tlciBzd2FybSBqb2luIC0tdG9rZW4gJFNXQVJNX1RPS0VOIFskU1dJVENIX0lQX0FERFJFU1NdOiRTV0lUQ0hfUE9SVAplbHNlCiAgICBlY2hvICJUaGlzIG5vZGUgaXMgYWxyZWFkeSBwYXJ0IG9mIGEgc3dhcm0iCiAgICBkb2NrZXIgaW5mbyAtZiBqc29uIHwganEgLlN3YXJtCmZpCg=="
+        }
+      },
+      {
+        "path": "/etc/hostname",
+        "mode": 420,
+        "overwrite": true,
+        "contents": {
+          "source": "data:,test_hostname"
+        }
+      },
+      {
+        "path": "/etc/NetworkManager/system-connections/eth0.nmconnection",
+        "mode": 384,
+        "overwrite": true,
+        "contents": {
+          "source": "data:text/plain;charset=utf-8;base64,Cltjb25uZWN0aW9uXQppZD1ldGgwCnR5cGU9ZXRoZXJuZXQKaW50ZXJmYWNlLW5hbWU9ZXRoMAoKW2lwdjRdCmRucy1zZWFyY2g9Cm1ldGhvZD1hdXRvCgpbaXB2Nl0KZG5zLXNlYXJjaD0KYWRkci1nZW4tbW9kZT1ldWk2NAptZXRob2Q9YXV0bwo=",
+          "human_read": "\n[connection]\nid=eth0\ntype=ethernet\ninterface-name=eth0\n\n[ipv4]\ndns-search=\nmethod=auto\n\n[ipv6]\ndns-search=\naddr-gen-mode=eui64\nmethod=auto\n"
+        }
+      },
+      {
+        "path": "/etc/NetworkManager/conf.d/noauto.conf",
+        "mode": 420,
+        "overwrite": true,
+        "contents": {
+          "source": "data:text/plain;charset=utf-8;base64,W21haW5dCiMgRG8gbm90IGRvIGF1dG9tYXRpYyAoREhDUC9TTEFBQykgY29uZmlndXJhdGlvbiBvbiBldGhlcm5ldCBkZXZpY2VzCiMgd2l0aCBubyBvdGhlciBtYXRjaGluZyBjb25uZWN0aW9ucy4Kbm8tYXV0by1kZWZhdWx0PSoK",
+          "human_read": "[main]\n# Do not do automatic (DHCP/SLAAC) configuration on ethernet devices\n# with no other matching connections.\nno-auto-default=*\n"
+        }
+      }
+    ]
+  },
+  "systemd": {
+    "units": [
+      {
+        "name": "cockpit.socket.service",
+        "enabled": true
+      },
+      {
+        "name": "docker.service",
+        "enabled": true
+      },
+      {
+        "name": "join_swarm.service",
+        "enabled": false,
+        "contents": "[Unit]\nDescription=Ensure that node joins a swarm on startup\n\n[Service]\nExecStart=/root/join_swarm.sh\n\n[Install]\nWantedBy=multi-user.target"
+      }
+    ]
+  }
+}

tests/test_node_deployer.py

@@ -1,5 +1,6 @@
 import atexit
 import filecmp
+import os
 from pathlib import Path
 import pickle
 import shutil
@@ -9,25 +10,47 @@ import tomllib
 
 
 config.update_config("test")
-config.BUILD_DIR = config.BUILD_DIR / "tests"
+config.BUILD_DIR = config.BUILD_DIR / f"tests/{os.getpid()}"
-atexit.register(lambda: shutil.rmtree(config.BUILD_DIR, ignore_errors=True))
 
-from node_deployer import autoignition, create_img  # noqa: E402
+
+def remove_pid_build_dir():
+    shutil.rmtree(config.BUILD_DIR, ignore_errors=True)
+
+
+def remove_test_build_dir():
+    test_build_dir = config.BUILD_DIR.parent
+    if any(test_build_dir.iterdir()):
+        try:
+            test_build_dir.rmdir()
+        except OSError:
+            pass
+
+
+def cleanup():
+    remove_pid_build_dir()
+    remove_test_build_dir()
+
+
+atexit.register(cleanup)
+
+from node_deployer import autoignition, create_disk, create_img  # noqa: E402
 
 
 with open(config.PROJECT_ROOT / "tests/data/node_deployer/test_args.toml", "rb") as f:
     TEST_PARAMS = tomllib.load(f)
 
+TEST_DATA_DIR = config.PROJECT_ROOT / "tests/data/node_deployer"
+
 
 class TestAutoignition:
     def test_json_to_img(self, tmp_path: Path):
         tmp_path.mkdir(parents=True, exist_ok=True)
         autoignition.json_to_img(
-            config.PROJECT_ROOT / "tests/data/node_deployer/fuelignition.json",
+            TEST_DATA_DIR / "fuelignition.json",
             tmp_path / "ignition.img",
         )
         assert filecmp.cmp(
-            config.PROJECT_ROOT / "tests/data/node_deployer/ignition.img",
+            TEST_DATA_DIR / "ignition.img",
             tmp_path / "ignition.img",
         )
 
@@ -35,14 +58,12 @@ class TestAutoignition:
 class TestCreateImg:
     def test_load_template(self):
         template = create_img.load_template()
-        with open(
+        with open(TEST_DATA_DIR / "create_img/load_template.pkl", "rb") as f:
-            config.PROJECT_ROOT / "tests/data/node_deployer/create_img/load_template.pkl", "rb"
-        ) as f:
             assert pickle.load(f) == template
 
     def test_apply_ignition_settings(self):
         with open(
-            config.PROJECT_ROOT / "tests/data/node_deployer/create_img/load_template.pkl",
+            TEST_DATA_DIR / "create_img/load_template.pkl",
             mode="rb",
         ) as f:
             template = pickle.load(f)
@@ -51,7 +72,7 @@ class TestCreateImg:
             **TEST_PARAMS["create_img"]["apply_ignition_settings"],
         )
         with open(
-            config.PROJECT_ROOT / "tests/data/node_deployer/create_img/apply_ignition_settings.pkl",
+            TEST_DATA_DIR / "create_img/apply_ignition_settings.pkl",
             mode="rb",
         ) as f:
             expected = pickle.load(f)
@@ -66,23 +87,32 @@ class TestCreateImg:
         )
         assert filecmp.cmp(
             tmp_path / "ignition.img",
-            config.PROJECT_ROOT / "tests/data/node_deployer/ignition.img",
+            TEST_DATA_DIR / "ignition.img",
         )
 
 
-# class TestWriteDisk:
+class TestCreateDisk:
-#     def init(self):
+    def init_buildfile(self, filename: str):
-#         test_target = config.BUILD_DIR / "ignition.img"
+        config.BUILD_DIR.mkdir(parents=True, exist_ok=True)
-#         if not test_target.exists():
+        test_target = config.BUILD_DIR / filename
-#             test_target.write_bytes(
+        if not test_target.exists():
-#                 Path(config.PROJECT_ROOT / "tests/data/node_deployer/ignition.img").read_bytes()
+            test_target.write_bytes(Path(TEST_DATA_DIR / filename).read_bytes())
-#             )
 
-#     def test_validation_result(self):
+    def test_filter_validation_response(self):
-#         raise NotImplementedError
+        self.init_buildfile("config.ign")
+        with open(TEST_DATA_DIR / "create_disk/validation_result.pkl", "rb") as f:
+            input = pickle.load(f)
+        test_result = create_disk.filter_validation_response(input)
+        assert test_result == ""
 
-#     def test_filter_validation_result(self):
+    def test_validation_result(self):
-#         raise NotImplementedError
+        self.init_buildfile("config.ign")
+        test_result = create_disk.validation_result()
+        with open(TEST_DATA_DIR / "create_disk/validation_result.pkl", "rb") as f:
+            expected = pickle.load(f)
+        assert test_result == expected
 
-#     def test_validate(self):
+    def test_validate(self):
-#         raise NotImplementedError
+        self.init_buildfile("config.ign")
+        test_result = create_disk.validate()
+        assert test_result == (True, "")