Added pydantic to create a Track object so it becomes easier to manage. Wrote a module.tf parser.

Author: MOBergeron

ctf/deploy.py

@@ -11,9 +11,11 @@
 from ctf.destroy import destroy
 from ctf.generate import generate
 from ctf.logger import LOG
+from ctf.models import Track
 from ctf.utils import (
     add_tracks_to_terraform_modules,
     check_git_lfs,
+    does_track_require_build_container,
     find_ctf_root_directory,
     get_all_available_tracks,
     get_terraform_tracks_from_modules,
@@ -61,6 +63,18 @@ def deploy(
             if validate_track_can_be_deployed(track=track) and track in tracks
         )
 
+        tmp_tracks: set[Track] = set()
+        for track in distinct_tracks:
+            tmp_tracks.add(
+                Track(
+                    name=track.name,
+                    remote=track.remote,
+                    production=track.production,
+                    require_build_container=does_track_require_build_container(track),
+                )
+            )
+        distinct_tracks = tmp_tracks
+
         add_tracks_to_terraform_modules(
             tracks=distinct_tracks - get_terraform_tracks_from_modules(),
             remote=remote,
@@ -84,7 +98,7 @@ def deploy(
             "git",
             "lfs",
             "pull",
-            f"--include={','.join([os.path.join('challenges', track, 'ansible', '*') for track in distinct_tracks])}",
+            f"--include={','.join([os.path.join('challenges', track.name, 'ansible', '*') for track in distinct_tracks])}",
         ],
         check=True,
     )
@@ -106,6 +120,8 @@ def deploy(
         force = True
         destroy(tracks=tracks, production=production, remote=remote, force=force)
 
+        distinct_tracks = generate(tracks=tracks, production=production, remote=remote)
+
         subprocess.run(
             args=[terraform_binary(), "apply", "-auto-approve"],
             cwd=os.path.join(find_ctf_root_directory(), ".deploy"),
@@ -120,17 +136,23 @@ def deploy(
         exit(code=0)
 
     for track in distinct_tracks:
+        if track.require_build_container:
+            # TODO: Ansible build containers here
+            
+            # TODO: Set the "build_container" OpenTofu variable to false in module.tf
+            pass
+
         if not os.path.exists(
             path=(
                 path := os.path.join(
-                    find_ctf_root_directory(), "challenges", track, "ansible"
+                    find_ctf_root_directory(), "challenges", track.name, "ansible"
                 )
             )
         ):
             continue
 
         run_ansible_playbook(
-            remote=remote, production=production, track=track, path=path
+            remote=remote, production=production, track=track.name, path=path
         )
 
         if not production:
@@ -154,7 +176,7 @@ def deploy(
 
             if remote == "local":
                 LOG.debug(msg=f"Parsing track.yaml for track {track}")
-                track_yaml = parse_track_yaml(track_name=track)
+                track_yaml = parse_track_yaml(track_name=track.name)
 
                 for service in track_yaml["services"]:
                     if service.get("dev_port_mapping"):
@@ -175,12 +197,12 @@ def deploy(
                                 "device",
                                 "add",
                                 machine_name,
-                                f"proxy-{track}-{service['dev_port_mapping']}-to-{service['port']}",
+                                f"proxy-{track.name}-{service['dev_port_mapping']}-to-{service['port']}",
                                 "proxy",
                                 f"listen=tcp:0.0.0.0:{service['dev_port_mapping']}",
                                 f"connect=tcp:127.0.0.1:{service['port']}",
                                 "--project",
-                                track,
+                                track.name,
                             ],
                             cwd=path,
                             check=True,
@@ -212,7 +234,7 @@ def deploy(
                 msg=f"Running `incus project switch {tracks_list[track_index - 1]}`"
             )
             subprocess.run(
-                args=["incus", "project", "switch", tracks_list[track_index - 1]],
+                args=["incus", "project", "switch", tracks_list[track_index - 1].name],
                 check=True,
                 env=ENV,
             )

ctf/destroy.py

@@ -7,6 +7,7 @@
 
 from ctf import ENV
 from ctf.logger import LOG
+from ctf.models import Track
 from ctf.utils import (
     find_ctf_root_directory,
     get_terraform_tracks_from_modules,
@@ -60,8 +61,8 @@ def destroy(
 
     total_deployed_tracks = len(terraform_tracks)
 
-    r = (
-        subprocess.run(
+    current_project = Track(
+        name=subprocess.run(
             args=["incus", "project", "get-current"],
             check=True,
             capture_output=True,
@@ -71,16 +72,16 @@ def destroy(
         .strip()
     )
 
-    tmp_tracks = set(tracks)
+    tmp_tracks: set[Track] = set(Track(name=x) for x in tracks)
     if tmp_tracks and tmp_tracks != terraform_tracks:
         terraform_tracks &= tmp_tracks
         if not terraform_tracks:
             LOG.warning("No track to destroy.")
             return
 
-    if r in terraform_tracks:
-        projects = {
-            project["name"]
+    if current_project in terraform_tracks:
+        projects: set[Track] = {
+            Track(name=project["name"])
             for project in json.loads(
                 s=subprocess.run(
                     args=["incus", "project", "list", "--format=json"],
@@ -91,8 +92,8 @@ def destroy(
             )
         }
 
-        projects = list((projects - terraform_tracks))
-        if len(projects) == 0:
+        project_list = list((projects - terraform_tracks))
+        if len(project_list) == 0:
             LOG.critical(
                 msg="No project to switch to. This should never happen as the default should always exists."
             )
@@ -102,7 +103,7 @@ def destroy(
             "incus",
             "project",
             "switch",
-            "default" if "default" in projects else projects[0],
+            "default" if "default" in project_list else project_list[0].name,
         ]
 
         LOG.info(msg=f"Running `{' '.join(cmd)}`")
@@ -116,15 +117,15 @@ def destroy(
             *(
                 []  # If every track needs to be destroyed, destroy everything including the network zone as well.
                 if total_deployed_tracks == len(terraform_tracks)
-                else [f"-target=module.track-{track}" for track in terraform_tracks]
+                else [f"-target=module.track-{track.name}" for track in terraform_tracks]
             ),
         ],
         cwd=os.path.join(find_ctf_root_directory(), ".deploy"),
         check=False,
     )
 
-    projects = [
-        project["name"]
+    projects = {
+        Track(name=project["name"])
         for project in json.loads(
             s=subprocess.run(
                 args=["incus", "project", "list", "--format=json"],
@@ -133,10 +134,10 @@ def destroy(
                 env=ENV,
             ).stdout.decode()
         )
-    ]
+    }
 
-    networks = [
-        network["name"]
+    networks = {
+        Track(name=network["name"])
         for network in json.loads(
             s=subprocess.run(
                 args=["incus", "network", "list", "--format=json"],
@@ -145,10 +146,10 @@ def destroy(
                 env=ENV,
             ).stdout.decode()
         )
-    ]
+    }
 
-    network_acls = [
-        network_acl["name"]
+    network_acls = {
+        Track(name=network_acl["name"])
         for network_acl in json.loads(
             s=subprocess.run(
                 args=["incus", "network", "acl", "list", "--format=json"],
@@ -157,46 +158,46 @@ def destroy(
                 env=ENV,
             ).stdout.decode()
         )
-    ]
+    }
 
     for module in terraform_tracks:
         if module in projects:
-            LOG.warning(msg=f"The project {module} was not destroyed properly.")
+            LOG.warning(msg=f"The project {module.name} was not destroyed properly.")
             if (
                 force
                 or (input("Do you want to destroy it? [Y/n] ").lower() or "y") == "y"
             ):
                 subprocess.run(
-                    args=["incus", "project", "delete", module, "--force"],
+                    args=["incus", "project", "delete", module.name, "--force"],
                     check=False,
                     capture_output=True,
                     input=b"yes\n",
                     env=ENV,
                 )
 
-        if (tmp_module := module[0:15]) in networks:
-            LOG.warning(msg=f"The network {tmp_module} was not destroyed properly.")
+        if (tmp_module_name := module.name[0:15]) in networks:
+            LOG.warning(msg=f"The network {tmp_module_name} was not destroyed properly.")
             if (
                 force
                 or (input("Do you want to destroy it? [Y/n] ").lower() or "y") == "y"
             ):
                 subprocess.run(
-                    args=["incus", "network", "delete", tmp_module],
+                    args=["incus", "network", "delete", tmp_module_name],
                     check=False,
                     capture_output=True,
                     env=ENV,
                 )
 
         if (tmp_module := module) in network_acls or (
-            tmp_module := f"{module}-default"
+            tmp_module := f"{module.name}-default"
         ) in network_acls:
-            LOG.warning(msg=f"The network ACL {tmp_module} was not destroyed properly.")
+            LOG.warning(msg=f"The network ACL {tmp_module.name} was not destroyed properly.")
             if (
                 force
                 or (input("Do you want to destroy it? [Y/n] ").lower() or "y") == "y"
             ):
                 subprocess.run(
-                    args=["incus", "network", "acl", "delete", tmp_module],
+                    args=["incus", "network", "acl", "delete", tmp_module.name],
                     check=False,
                     capture_output=True,
                     env=ENV,

ctf/flags.py

@@ -2,26 +2,19 @@
 import io
 import json
 import os
-from enum import StrEnum, unique
 
 import rich
 import typer
 import yaml
 from typing_extensions import Annotated
 
 from ctf.logger import LOG
+from ctf.models import OutputFormat, Track
 from ctf.utils import find_ctf_root_directory, parse_track_yaml
 
 app = typer.Typer()
 
 
-@unique
-class OutputFormat(StrEnum):
-    JSON = "json"
-    CSV = "csv"
-    YAML = "yaml"
-
-
 @app.command(help="Get flags from tracks")
 def flags(
     tracks: Annotated[
@@ -37,7 +30,7 @@ def flags(
         typer.Option("--format", help="Output format", prompt="Output format"),
     ] = OutputFormat.JSON,
 ) -> None:
-    distinct_tracks: set[str] = set()
+    distinct_tracks: set[Track] = set()
 
     for entry in os.listdir(
         path=(
@@ -50,17 +43,17 @@ def flags(
             s=(track_directory := os.path.join(challenges_directory, entry))
         ) and os.path.exists(path=os.path.join(track_directory, "track.yaml")):
             if not tracks:
-                distinct_tracks.add(entry)
+                distinct_tracks.add(Track(name=entry))
             elif entry in tracks:
-                distinct_tracks.add(entry)
+                distinct_tracks.add(Track(name=entry))
 
     flags = []
     for track in distinct_tracks:
-        LOG.debug(msg=f"Parsing track.yaml for track {track}")
-        track_yaml = parse_track_yaml(track_name=track)
+        LOG.debug(msg=f"Parsing track.yaml for track {track.name}")
+        track_yaml = parse_track_yaml(track_name=track.name)
 
         if len(track_yaml["flags"]) == 0:
-            LOG.debug(msg=f"No flag in track {track}. Skipping...")
+            LOG.debug(msg=f"No flag in track {track.name}. Skipping...")
             continue
 
         flags.extend(track_yaml["flags"])

ctf/generate.py

@@ -6,12 +6,13 @@
 
 from ctf import ENV
 from ctf.logger import LOG
+from ctf.models import Track
 from ctf.utils import (
     add_tracks_to_terraform_modules,
     create_terraform_modules_file,
+    does_track_require_build_container,
     find_ctf_root_directory,
     get_all_available_tracks,
-    load_yaml_file,
     terraform_binary,
     validate_track_can_be_deployed,
 )
@@ -41,10 +42,10 @@ def generate(
     remote: Annotated[
         str, typer.Option("--remote", help="Incus remote to deploy to")
     ] = "local",
-) -> set[str]:
+) -> set[Track]:
     ENV["INCUS_REMOTE"] = remote
     # Get the list of tracks.
-    distinct_tracks = set(
+    distinct_tracks: set[Track] = set(
         track
         for track in get_all_available_tracks()
         if validate_track_can_be_deployed(track=track)
@@ -56,18 +57,17 @@ def generate(
         # Generate the Terraform modules file.
         create_terraform_modules_file(remote=remote, production=production)
 
+        tmp_tracks: set[Track] = set()
         for track in distinct_tracks:
-            if os.path.isfile(
-                build_yaml_file_path := os.path.join(
-                    find_ctf_root_directory(),
-                    "challenges",
-                    track,
-                    "ansible",
-                    "build.yaml",
+            tmp_tracks.add(
+                Track(
+                    name=track.name,
+                    remote=track.remote,
+                    production=track.production,
+                    require_build_container=does_track_require_build_container(track),
                 )
-            ) and (build_yaml_file := load_yaml_file(build_yaml_file_path)) and build_yaml_file:
-                #TODO: set build for track
-                pass
+            )
+        distinct_tracks = tmp_tracks
 
         add_tracks_to_terraform_modules(
             tracks=distinct_tracks,
@@ -80,7 +80,7 @@ def generate(
                 os.path.join(find_ctf_root_directory(), ".deploy", "common"),
                 (
                     terraform_directory := os.path.join(
-                        find_ctf_root_directory(), "challenges", track, "terraform"
+                        find_ctf_root_directory(), "challenges", track.name, "terraform"
                     )
                 ),
             )