new: added new --path optional argument (ENG-470)

dreadnode_cli/agent/cli.py

@@ -94,6 +94,14 @@ def init(
             help="Initialize the agent using a custom template from a github repository, ZIP archive URL or local folder",
         ),
     ] = None,
+    path: t.Annotated[
+        str | None,
+        typer.Option(
+            "--path",
+            "-p",
+            help="If --source has been provided, use --path to specify a subfolder to initialize from",
+        ),
+    ] = None,
 ) -> None:
     print(f":coffee: Fetching strike '{strike}' ...")
 
@@ -164,7 +172,12 @@ def init(
 
         try:
             # initialize from local folder, validation performed inside install_template_from_dir
-            install_template_from_dir(source_dir, directory, context)
+            #
+            # NOTE: source_dir and path are passed separately because github repos zip archives
+            # usually contain a single branch folder, the real source dir, and the path is not known
+            # beforehand. install_template_from_dir handles this case and combines the fixed
+            # source_dir with path accordingly.
+            install_template_from_dir(source_dir, path, directory, context)
         except Exception:
             if cleanup and source_dir.exists():
                 shutil.rmtree(source_dir)

dreadnode_cli/agent/templates/__init__.py

@@ -26,10 +26,12 @@ def template_description(template: Template) -> str:
 
 def install_template(template: Template, dest: pathlib.Path, context: dict[str, t.Any]) -> None:
     """Install a template into a directory."""
-    install_template_from_dir(TEMPLATES_DIR / template.value, dest, context)
+    install_template_from_dir(TEMPLATES_DIR, template.value, dest, context)
 
 
-def install_template_from_dir(src: pathlib.Path, dest: pathlib.Path, context: dict[str, t.Any]) -> None:
+def install_template_from_dir(
+    src: pathlib.Path, sub_path: str | None, dest: pathlib.Path, context: dict[str, t.Any]
+) -> None:
     """Install a template from a source directory into a destination directory."""
 
     if not src.exists():
@@ -46,9 +48,14 @@ def install_template_from_dir(src: pathlib.Path, dest: pathlib.Path, context: di
         if len(subdirs) == 1:
             src = subdirs[0]
 
-        # check again for Dockerfile in the subdirectory
-        if not (src / "Dockerfile").exists() and not (src / "Dockerfile.j2").exists():
-            raise Exception("Source directory does not contain a Dockerfile")
+    if sub_path is not None:
+        src = src / sub_path
+        if not src.exists():
+            raise Exception(f"Source path '{src}' does not exist.")
+
+    # check again for Dockerfile in the subdirectory
+    if not (src / "Dockerfile").exists() and not (src / "Dockerfile.j2").exists():
+        raise Exception(f"Source directory {src} does not contain a Dockerfile")
 
     env = Environment(loader=FileSystemLoader(src))
 

dreadnode_cli/agent/tests/test_templates.py

@@ -16,7 +16,7 @@ def test_templates_install(tmp_path: pathlib.Path) -> None:
 
 
 def test_templates_install_from_dir(tmp_path: pathlib.Path) -> None:
-    templates.install_template_from_dir(templates.TEMPLATES_DIR / "rigging_basic", tmp_path, {"name": "World"})
+    templates.install_template_from_dir(templates.TEMPLATES_DIR / "rigging_basic", None, tmp_path, {"name": "World"})
 
     assert (tmp_path / "requirements.txt").exists()
     assert (tmp_path / "Dockerfile").exists()
@@ -43,7 +43,7 @@ def test_templates_install_from_dir_with_dockerfile_template(tmp_path: pathlib.P
     dest_dir.mkdir()
 
     # install template
-    templates.install_template_from_dir(source_dir, dest_dir, {"name": "TestContainer"})
+    templates.install_template_from_dir(source_dir, None, dest_dir, {"name": "TestContainer"})
 
     # verify Dockerfile was rendered correctly
     expected_dockerfile = """
@@ -90,7 +90,7 @@ def test_templates_install_from_dir_nested_structure(tmp_path: pathlib.Path) ->
     dest_dir.mkdir()
 
     # install template
-    templates.install_template_from_dir(source_dir, dest_dir, {"name": "TestApp"})
+    templates.install_template_from_dir(source_dir, None, dest_dir, {"name": "TestApp"})
 
     # verify regular files were copied
     assert (dest_dir / "Dockerfile").exists()
@@ -109,24 +109,24 @@ def test_templates_install_from_dir_nested_structure(tmp_path: pathlib.Path) ->
 def test_templates_install_from_dir_missing_source(tmp_path: pathlib.Path) -> None:
     source_dir = tmp_path / "nonexistent"
     with pytest.raises(Exception, match="Source directory '.*' does not exist"):
-        templates.install_template_from_dir(source_dir, tmp_path, {"name": "World"})
+        templates.install_template_from_dir(source_dir, None, tmp_path, {"name": "World"})
 
 
 def test_templates_install_from_dir_source_is_file(tmp_path: pathlib.Path) -> None:
     source_file = tmp_path / "source.txt"
     source_file.touch()
 
     with pytest.raises(Exception, match="Source '.*' is not a directory"):
-        templates.install_template_from_dir(source_file, tmp_path, {"name": "World"})
+        templates.install_template_from_dir(source_file, None, tmp_path, {"name": "World"})
 
 
 def test_templates_install_from_dir_missing_dockerfile(tmp_path: pathlib.Path) -> None:
     source_dir = tmp_path / "source"
     source_dir.mkdir()
     (source_dir / "agent.py").touch()
 
-    with pytest.raises(Exception, match="Source directory does not contain a Dockerfile"):
-        templates.install_template_from_dir(source_dir, tmp_path, {"name": "World"})
+    with pytest.raises(Exception, match="Source directory .+ does not contain a Dockerfile"):
+        templates.install_template_from_dir(source_dir, None, tmp_path, {"name": "World"})
 
 
 def test_templates_install_from_dir_single_inner_folder(tmp_path: pathlib.Path) -> None:
@@ -144,8 +144,37 @@ def test_templates_install_from_dir_single_inner_folder(tmp_path: pathlib.Path)
     dest_dir.mkdir()
 
     # install from the outer directory - should detect and use inner directory
-    templates.install_template_from_dir(source_dir, dest_dir, {"name": "World"})
+    templates.install_template_from_dir(source_dir, None, dest_dir, {"name": "World"})
 
     # assert files were copied from inner directory
     assert (dest_dir / "Dockerfile").exists()
     assert (dest_dir / "agent.py").exists()
+
+
+def test_templates_install_from_dir_with_path(tmp_path: pathlib.Path) -> None:
+    # create source directory with subdirectories
+    source_dir = tmp_path / "source"
+    source_dir.mkdir()
+
+    # create files in subdirectory
+    (source_dir / "Dockerfile").touch()
+    (source_dir / "agent.py").touch()
+
+    dest_dir = tmp_path / "dest"
+    dest_dir.mkdir()
+
+    # install from subdirectory path
+    templates.install_template_from_dir(tmp_path, "source", dest_dir, {"name": "World"})
+
+    # assert files were copied from subdirectory
+    assert (dest_dir / "Dockerfile").exists()
+    assert (dest_dir / "agent.py").exists()
+
+
+def test_templates_install_from_dir_invalid_path(tmp_path: pathlib.Path) -> None:
+    source_dir = tmp_path / "source"
+    source_dir.mkdir()
+    (source_dir / "Dockerfile").touch()
+
+    with pytest.raises(Exception, match="Source path '.*' does not exist."):
+        templates.install_template_from_dir(source_dir, "nonexistent", tmp_path, {"name": "World"})

dreadnode_cli/api.py

@@ -236,11 +236,10 @@ class GithubTokenResponse(BaseModel):
         expires_at: datetime
         repos: list[str]
 
-    def get_github_access_token(self, repos: list[str]) -> GithubTokenResponse | None:
+    def get_github_access_token(self, repos: list[str]) -> GithubTokenResponse:
         """Try to get a GitHub access token for the given repositories."""
-
         response = self.request("POST", "/api/github/token", json_data={"repos": repos})
-        return self.GithubTokenResponse(**response.json()) if response.status_code == 200 else None
+        return self.GithubTokenResponse(**response.json())
 
     # Strikes
 

dreadnode_cli/utils.py

@@ -70,7 +70,7 @@ def parse_jwt_token_expiration(token: str) -> datetime:
 
 def repo_exists(repo: GithubRepo) -> bool:
     """Check if a repo exists (or is private) on GitHub."""
-    response = httpx.get(f"https://github.com/repos/{repo.namespace}/{repo.repo}")
+    response = httpx.get(f"https://github.com/{repo.namespace}/{repo.repo}")
     return response.status_code == 200