Support ArgoCD repository credentials when cloning Git repositories

src/nyl/generator/dispatch.py

@@ -79,6 +79,7 @@ def default(
             working_dir=working_dir,
             kube_version=kube_version,
             api_versions=kube_api_versions,
+            client=client,
         )
 
         return DispatchingGenerator(

src/nyl/generator/helmchart.py

@@ -9,10 +9,16 @@
 from urllib.parse import parse_qs, urlparse
 
 from loguru import logger
+from kubernetes.client.api_client import ApiClient
 
 from nyl.generator import Generator
 from nyl.resources.helmchart import ChartRef, HelmChart, ReleaseMetadata
 from nyl.tools import yaml
+from nyl.tools.argocd_repo_credentials import (
+    apply_credential_to_git_url,
+    find_matching_credential,
+    query_argocd_repository_credentials,
+)
 from nyl.tools.kubernetes import populate_namespace_to_resources
 from nyl.tools.shell import pretty_cmd
 from nyl.tools.types import ResourceList
@@ -47,6 +53,9 @@ class HelmChartGenerator(Generator[HelmChart], resource_type=HelmChart):
     `--validate` flag.
     """
 
+    client: ApiClient | None = None
+    """ Kubernetes API client for querying ArgoCD repository credentials. """
+
     def _materialize_chart(self, chart_ref: ChartRef) -> ChartRepositoryVersion:
         repository: str | None = None
         chart: str | None = None
@@ -104,6 +113,28 @@ def _materialize_chart(self, chart_ref: ChartRef) -> ChartRepositoryVersion:
             # Clone the repository and find the chart in the repository.
             parsed = urlparse(chart_ref.git)
             without_query_params = parsed._replace(query="").geturl()
+
+            # Try to get ArgoCD repository credentials for this Git URL
+            git_url_with_auth = without_query_params
+            if self.client:
+                try:
+                    credentials = query_argocd_repository_credentials(self.client)
+                    matching_credential = find_matching_credential(without_query_params, credentials)
+                    if matching_credential:
+                        if matching_credential.is_https:
+                            logger.debug("Using ArgoCD HTTPS repository credential for {}", without_query_params)
+                            git_url_with_auth = apply_credential_to_git_url(without_query_params, matching_credential)
+                        elif matching_credential.is_ssh:
+                            logger.info("ArgoCD SSH repository credential found for {} but SSH key authentication "
+                                      "is not fully implemented. Using original URL.", without_query_params)
+                        else:
+                            logger.debug("ArgoCD repository credential found for {} but no authentication method available", 
+                                       without_query_params)
+                    else:
+                        logger.debug("No ArgoCD repository credential found for {}", without_query_params)
+                except Exception as e:
+                    logger.warning("Failed to query ArgoCD repository credentials: {}", e)
+
             hashed = hashlib.md5(without_query_params.encode()).hexdigest()
             clone_dir = self.git_repo_cache_dir / f"{hashed}-{PosixPath(parsed.path).name}"
             if clone_dir.exists():
@@ -112,7 +143,7 @@ def _materialize_chart(self, chart_ref: ChartRef) -> ChartRepositoryVersion:
                 cwd = clone_dir
             else:
                 logger.debug("Cloning {} to {}", without_query_params, clone_dir)
-                command = ["git", "clone", without_query_params, str(clone_dir)]
+                command = ["git", "clone", git_url_with_auth, str(clone_dir)]
                 cwd = None
             subprocess.check_call(command, cwd=cwd, stdout=sys.stderr)
 

src/nyl/generator/helmchart_test.py

@@ -1,12 +1,14 @@
 from pathlib import Path
 from tempfile import TemporaryDirectory
 from typing import Iterator
+from unittest.mock import Mock, patch
 
 import pytest
 
 from nyl.generator.helmchart import HelmChartGenerator
 from nyl.resources import ObjectMetadata
 from nyl.resources.helmchart import ChartRef, HelmChart, HelmChartSpec
+from nyl.tools.argocd_repo_credentials import RepoCredential
 from nyl.tools.testing import TESTDATA_PATH
 
 CHART_PATH = TESTDATA_PATH / "helm-test-chart"
@@ -19,6 +21,15 @@ def generator() -> Iterator[HelmChartGenerator]:
         yield HelmChartGenerator(tmp, tmp, [], tmp, "1.31", set())
 
 
+@pytest.fixture
+def generator_with_client() -> Iterator[HelmChartGenerator]:
+    """Generator with a mock Kubernetes client for testing ArgoCD credentials."""
+    with TemporaryDirectory() as _tmp:
+        tmp = Path(_tmp)
+        mock_client = Mock()
+        yield HelmChartGenerator(tmp, tmp, [], tmp, "1.31", set(), client=mock_client)
+
+
 def test_helmchart_resource_overrides_labels(generator: HelmChartGenerator) -> None:
     """
     This test verifies that resources generated by a Helm chart component have their labels updated and overwritten
@@ -43,3 +54,96 @@ def test_helmchart_resource_overrides_labels(generator: HelmChartGenerator) -> N
         "app.kubernetes.io/version": "1.0.0",
         "team": "testing",
     }
+
+
+def test_helmchart_git_with_argocd_credentials(generator_with_client: HelmChartGenerator) -> None:
+    """
+    Test that HelmChart with git source uses ArgoCD repository credentials when available.
+    """
+    # Mock the credential query to return a test credential
+    test_credential = RepoCredential(
+        url="https://github.com/myorg/",
+        username="testuser",
+        password="testtoken"
+    )
+
+    with patch('nyl.generator.helmchart.query_argocd_repository_credentials') as mock_query, \
+         patch('nyl.generator.helmchart.find_matching_credential') as mock_find, \
+         patch('nyl.generator.helmchart.apply_credential_to_git_url') as mock_apply, \
+         patch('nyl.generator.helmchart.subprocess.check_call') as mock_subprocess:
+
+        # Setup mocks
+        mock_query.return_value = [test_credential]
+        mock_find.return_value = test_credential
+        mock_apply.return_value = "https://testuser:testtoken@github.com/myorg/test-repo.git"
+
+        # Create chart reference with Git URL
+        chart_ref = ChartRef(git="https://github.com/myorg/test-repo.git")
+
+        # Call the _materialize_chart method
+        generator_with_client._materialize_chart(chart_ref)
+
+        # Verify that ArgoCD credentials were queried
+        mock_query.assert_called_once_with(generator_with_client.client)
+
+        # Verify that credential matching was attempted
+        mock_find.assert_called_once_with("https://github.com/myorg/test-repo.git", [test_credential])
+
+        # Verify that credentials were applied to the URL
+        mock_apply.assert_called_once_with("https://github.com/myorg/test-repo.git", test_credential)
+
+        # Verify that git clone was called with the authenticated URL
+        mock_subprocess.assert_called()
+        clone_call = mock_subprocess.call_args_list[0]
+        assert "https://testuser:testtoken@github.com/myorg/test-repo.git" in clone_call[0][0]
+
+
+def test_helmchart_git_without_argocd_credentials(generator_with_client: HelmChartGenerator) -> None:
+    """
+    Test that HelmChart with git source falls back to original URL when no ArgoCD credentials are found.
+    """
+    with patch('nyl.generator.helmchart.query_argocd_repository_credentials') as mock_query, \
+         patch('nyl.generator.helmchart.find_matching_credential') as mock_find, \
+         patch('nyl.generator.helmchart.subprocess.check_call') as mock_subprocess:
+
+        # Setup mocks - no credentials found
+        mock_query.return_value = []
+        mock_find.return_value = None
+
+        # Create chart reference with Git URL
+        chart_ref = ChartRef(git="https://github.com/myorg/test-repo.git")
+
+        # Call the _materialize_chart method
+        generator_with_client._materialize_chart(chart_ref)
+
+        # Verify that ArgoCD credentials were queried
+        mock_query.assert_called_once_with(generator_with_client.client)
+
+        # Verify that credential matching was attempted
+        mock_find.assert_called_once_with("https://github.com/myorg/test-repo.git", [])
+
+        # Verify that git clone was called with the original URL
+        mock_subprocess.assert_called()
+        clone_call = mock_subprocess.call_args_list[0]
+        assert "https://github.com/myorg/test-repo.git" in clone_call[0][0]
+
+
+def test_helmchart_git_without_client() -> None:
+    """
+    Test that HelmChart with git source works without Kubernetes client (no ArgoCD credential lookup).
+    """
+    with TemporaryDirectory() as _tmp:
+        tmp = Path(_tmp)
+        generator_no_client = HelmChartGenerator(tmp, tmp, [], tmp, "1.31", set(), client=None)
+
+        with patch('nyl.generator.helmchart.subprocess.check_call') as mock_subprocess:
+            # Create chart reference with Git URL
+            chart_ref = ChartRef(git="https://github.com/myorg/test-repo.git")
+
+            # Call the _materialize_chart method
+            generator_no_client._materialize_chart(chart_ref)
+
+            # Verify that git clone was called with the original URL
+            mock_subprocess.assert_called()
+            clone_call = mock_subprocess.call_args_list[0]
+            assert "https://github.com/myorg/test-repo.git" in clone_call[0][0]