diff --git a/local-k8s/src/local_k8s/cli.py b/local-k8s/src/local_k8s/cli.py index 756f8d2..90c4736 100644 --- a/local-k8s/src/local_k8s/cli.py +++ b/local-k8s/src/local_k8s/cli.py @@ -3,8 +3,8 @@ import click +from local_k8s.groups.chart import chart from local_k8s.groups.cluster import cluster -from local_k8s.groups.ct import ct from local_k8s.groups.tools import tools from local_k8s.shared import get_bin_dir @@ -39,4 +39,4 @@ def configure_logging(verbose: int) -> None: cli.add_command(cluster) cli.add_command(tools) -cli.add_command(ct) +cli.add_command(chart) diff --git a/local-k8s/src/local_k8s/cluster.py b/local-k8s/src/local_k8s/cluster.py index 9abb747..0e6b97a 100644 --- a/local-k8s/src/local_k8s/cluster.py +++ b/local-k8s/src/local_k8s/cluster.py @@ -6,7 +6,7 @@ from tempfile import TemporaryDirectory from local_k8s.models import ClusterComponents -from local_k8s.shared import execute +from local_k8s.shared import CommandResult, execute CLUSTER_NAME = "test-cluster" @@ -40,7 +40,9 @@ def create_cluster(kind_config: Path, components: ClusterComponents) -> None: def _create_kind_cluster(kind_config: Path) -> None: LOG.info("Creating kind cluster") - for line in kind("get", "clusters", "--quiet").splitlines(): + for line in kind( + "get", "clusters", "--quiet", capture_stdout=True + ).stdout.splitlines(): if line == CLUSTER_NAME: LOG.info("Reusing existing cluster: %s", CLUSTER_NAME) break @@ -66,8 +68,8 @@ def _create_kind_cluster(kind_config: Path) -> None: def _add_helm_repos(components: ClusterComponents) -> None: if components.helm_repos: LOG.info("Adding helm repos") - repo_out = helm("repo", "list", "--no-headers") - existing_repos = [tuple(s.split()) for s in repo_out.splitlines()] + repo_out = helm("repo", "list", "--no-headers", capture_stdout=True) + existing_repos = [tuple(s.split()) for s in repo_out.stdout.splitlines()] for name, repo in components.helm_repos.items(): if (name, repo) in existing_repos: LOG.info("Not adding %s -> %s as already present", name, repo) @@ -77,8 +79,8 @@ def _add_helm_repos(components: ClusterComponents) -> None: def _install_helm_components(components: ClusterComponents) -> None: - list_out = helm("list", "--all-namespaces", "--deployed", "-q") - installed = list_out.splitlines() + list_out = helm("list", "--all-namespaces", "--deployed", "-q", capture_stdout=True) + installed = list_out.stdout.splitlines() LOG.info("Installing cluster components") for desired in components.cluster_components: if desired.name in installed: @@ -140,15 +142,15 @@ def dump_to_stdout(filter_namespaces: list[str], out_dir: Path) -> None: print(log_file.read_text()) -def kind(*args: str) -> str: - return execute("kind", *args) +def kind(*args: str, capture_stdout: bool = False) -> CommandResult: + return execute("kind", *args, capture_stdout=capture_stdout) -def helm(*args: str) -> str: +def helm(*args: str, capture_stdout: bool = False) -> CommandResult: with kube_guard(): - return execute("helm", *args) + return execute("helm", *args, capture_stdout=capture_stdout) -def kubectl(*args: str) -> str: +def kubectl(*args: str, capture_stdout: bool = False) -> CommandResult: with kube_guard(): - return execute("kubectl", *args) + return execute("kubectl", *args, capture_stdout=capture_stdout) diff --git a/local-k8s/src/local_k8s/commands/ct/__init__.py b/local-k8s/src/local_k8s/commands/chart/__init__.py similarity index 100% rename from local-k8s/src/local_k8s/commands/ct/__init__.py rename to local-k8s/src/local_k8s/commands/chart/__init__.py diff --git a/local-k8s/src/local_k8s/commands/chart/lint.py b/local-k8s/src/local_k8s/commands/chart/lint.py new file mode 100644 index 0000000..bbfc2a0 --- /dev/null +++ b/local-k8s/src/local_k8s/commands/chart/lint.py @@ -0,0 +1,52 @@ +from contextlib import chdir +from pathlib import Path +from subprocess import CalledProcessError + +import click +from click import ClickException + +from local_k8s.shared import execute, resolve_chart +from local_k8s.static import CT_YAML + + +@click.command("lint") +@click.option( + "--helm-dir", + type=click.Path( + exists=True, + file_okay=False, + dir_okay=True, + path_type=Path, + ), + required=True, +) +@click.option( + "--chart", + type=click.Path( + exists=False, + file_okay=False, + dir_okay=True, + path_type=Path, + ), + required=True, +) +def lint(helm_dir: Path, chart: Path) -> None: + helm_dir = helm_dir.resolve() + if not (helm_dir / CT_YAML).is_file(): + raise ClickException(f"{CT_YAML} is required in the root of --helm-dir") + + resolved_chart = resolve_chart(helm_dir, chart) + + with chdir(helm_dir): + try: + execute( + "ct", + "lint", + "--config", + str(CT_YAML), + "--charts", + str(resolved_chart.path_relative_to_helm_dir), + "--check-version-increment=true", + ) + except CalledProcessError as e: + raise ClickException(f"lint failed with rc={e.returncode}") from e diff --git a/local-k8s/src/local_k8s/commands/chart/list_changed.py b/local-k8s/src/local_k8s/commands/chart/list_changed.py new file mode 100644 index 0000000..ce450a4 --- /dev/null +++ b/local-k8s/src/local_k8s/commands/chart/list_changed.py @@ -0,0 +1,67 @@ +import json +from contextlib import chdir +from pathlib import Path +from subprocess import CalledProcessError + +import click +from click import ClickException + +from local_k8s.shared import execute +from local_k8s.static import CT_YAML + + +@click.command("list-changed") +@click.option( + "--helm-dir", + type=click.Path( + exists=True, + file_okay=False, + dir_okay=True, + path_type=Path, + ), + required=True, +) +@click.option("--target-branch", default="main", show_default=True) +@click.option("--since", default=None) +def list_changed(helm_dir: Path, target_branch: str, since: str | None) -> None: + helm_dir = helm_dir.resolve() + ct_path = helm_dir / CT_YAML + if not ct_path.is_file(): + raise ClickException(f"{CT_YAML} is required in the root of --helm-dir") + + # Get the root of our git repo. + repo_root = Path( + execute( + "git", + "-C", + str(helm_dir), + "rev-parse", + "--show-toplevel", + skip_resolve=True, + capture_stdout=True, + ).stdout.strip() + ).resolve() + + # ct list-changed operates a bit differently to the other CT commands. + # It must be executed from the root of the repo, and the paths to + # the chart-dir/config need to be relative not absolute. This is + # why we convert to relative below, if that is not done, it asplodes. + with chdir(repo_root): + args = [ + "ct", + "list-changed", + "--config", + str(ct_path.relative_to(repo_root)), + "--chart-dirs", + str((helm_dir / "charts").relative_to(repo_root)), + "--target-branch", + target_branch, + ] + if since is not None: + args.extend(["--since", since]) + try: + out = execute(*args, capture_stdout=True) + except CalledProcessError as e: + raise ClickException(f"list-changed failed with rc={e.returncode}") from e + charts = [line.strip() for line in out.stdout.splitlines() if line.strip()] + click.echo(json.dumps(charts)) diff --git a/local-k8s/src/local_k8s/commands/chart/metadata.py b/local-k8s/src/local_k8s/commands/chart/metadata.py new file mode 100644 index 0000000..9838865 --- /dev/null +++ b/local-k8s/src/local_k8s/commands/chart/metadata.py @@ -0,0 +1,27 @@ +import json +from pathlib import Path + +import click +from click import ClickException +from pydantic import ValidationError + +from local_k8s.models import ChartMetadata + + +@click.command("metadata") +@click.option( + "--chart", + type=click.Path( + exists=True, + file_okay=False, + dir_okay=True, + path_type=Path, + ), + required=True, +) +def metadata(chart: Path) -> None: + try: + meta = ChartMetadata.from_chart_dir(chart) + except (ValueError, ValidationError) as e: + raise ClickException(str(e)) from e + click.echo(json.dumps(meta.model_dump(mode="json"))) diff --git a/local-k8s/src/local_k8s/commands/chart/push.py b/local-k8s/src/local_k8s/commands/chart/push.py new file mode 100644 index 0000000..df6c28d --- /dev/null +++ b/local-k8s/src/local_k8s/commands/chart/push.py @@ -0,0 +1,146 @@ +import json +from pathlib import Path +from subprocess import CalledProcessError +from tempfile import TemporaryDirectory + +import click +from click import ClickException +from pydantic import ValidationError + +from local_k8s.models import ChartMetadata +from local_k8s.shared import execute + +REGISTRY_HOST = "ghcr.io" + + +def _detailed_failure_for(step: str, e: CalledProcessError) -> ClickException: + stdout = e.output if isinstance(e.output, str) else "" + msg = f"{step} failed with rc={e.returncode}" + if stdout.strip(): + msg = f"{msg}\n{stdout.strip()}" + return ClickException(msg) + + +@click.command("push") +@click.option( + "--chart", + type=click.Path( + exists=True, + file_okay=False, + dir_okay=True, + path_type=Path, + ), + required=True, +) +@click.option( + "--registry-config", + type=click.Path( + exists=True, + file_okay=True, + dir_okay=False, + path_type=Path, + ), + required=True, +) +@click.option("--oci-repo", required=True, help="OCI path after host, e.g. owner/repo") +@click.option("--beta", default=None) +@click.option("--fail-if-exists", is_flag=True, default=False) +def push( + chart: Path, + registry_config: Path, + oci_repo: str, + beta: str | None, + fail_if_exists: bool, +) -> None: + try: + meta = ChartMetadata.from_chart_dir(chart) + except (ValueError, ValidationError) as e: + raise ClickException(str(e)) from e + + version = f"{meta.version}-beta.{beta}" if beta is not None else meta.version + + oci_base = f"oci://{REGISTRY_HOST}/{oci_repo}" + oci_chart = f"{oci_base}/{meta.name}" + archive_name = f"{meta.name}-{version}.tgz" + registry_config_arg = str(registry_config) + + if fail_if_exists: + result = execute( + "helm", + "show", + "chart", + oci_chart, + "--version", + version, + "--registry-config", + registry_config_arg, + check=False, + capture_stdout=True, + capture_stderr=True, + ) + if result.returncode == 0: + raise ClickException(f"{meta.name}:{version} already exists in registry") + text = f"{result.stderr}\n{result.stdout}".lower() + if "not found" not in text and "manifest unknown" not in text: + raise ClickException( + f"exist check failed (rc={result.returncode}): {result.stderr.strip()}" + ) + + try: + execute( + "helm", + "dependency", + "build", + str(chart), + "--registry-config", + registry_config_arg, + capture_stdout=True, + ) + except CalledProcessError as e: + raise _detailed_failure_for("dependency build", e) from e + + with TemporaryDirectory() as tmp: + package_args = [ + "helm", + "package", + str(chart), + "--destination", + tmp, + ] + if beta is not None: + package_args.extend(["--version", version]) + try: + execute(*package_args, capture_stdout=True) + execute( + "helm", + "push", + str(Path(tmp) / archive_name), + oci_base, + "--registry-config", + registry_config_arg, + capture_stdout=True, + ) + execute( + "helm", + "show", + "chart", + oci_chart, + "--version", + version, + "--registry-config", + registry_config_arg, + capture_stdout=True, + ) + except CalledProcessError as e: + raise _detailed_failure_for("push", e) from e + + click.echo( + json.dumps( + { + "name": meta.name, + "version": version, + "oci_ref": f"{REGISTRY_HOST}/{oci_repo}/{meta.name}:{version}", + "archive": archive_name, + } + ) + ) diff --git a/local-k8s/src/local_k8s/commands/ct/lint_and_install.py b/local-k8s/src/local_k8s/commands/chart/test.py similarity index 66% rename from local-k8s/src/local_k8s/commands/ct/lint_and_install.py rename to local-k8s/src/local_k8s/commands/chart/test.py index 7eea50a..dac3fe4 100644 --- a/local-k8s/src/local_k8s/commands/ct/lint_and_install.py +++ b/local-k8s/src/local_k8s/commands/chart/test.py @@ -6,9 +6,10 @@ import click import yaml from click import ClickException +from pydantic import ValidationError -from local_k8s.models import CiSecrets -from local_k8s.shared import execute +from local_k8s.models import ChartMetadata, CiSecrets +from local_k8s.shared import ResolvedChart, execute, resolve_chart from local_k8s.static import CI_SECRETS_YAML, CT_YAML IMAGE_SECRET_PATHS = [ @@ -20,7 +21,7 @@ LOG = logging.getLogger(__name__) -@click.command("lint-and-install") +@click.command("test") @click.option( "--helm-dir", type=click.Path( @@ -31,14 +32,57 @@ ), required=True, ) -def lint_and_install(helm_dir: Path) -> None: - with chdir(helm_dir.absolute()): - if not CT_YAML.exists(): - raise ClickException(f"{CT_YAML} is required in the root of --helm-dir") +@click.option( + "--chart", + type=click.Path( + exists=False, + file_okay=False, + dir_okay=True, + path_type=Path, + ), + default=None, +) +def test(helm_dir: Path, chart: Path | None) -> None: + helm_dir = helm_dir.resolve() + if not (helm_dir / CT_YAML).is_file(): + raise ClickException(f"{CT_YAML} is required in the root of --helm-dir") + if chart is not None: + resolved_charts = [resolve_chart(helm_dir, chart)] + else: + resolved_charts = [ + ResolvedChart( + absolute_path=helm_dir / discovered_chart, + path_relative_to_helm_dir=discovered_chart, + ) + for discovered_chart in discover_charts(helm_dir) + ] + + with chdir(helm_dir): namespace = create_test_namespace(CT_YAML) create_secrets(namespace=namespace) - execute_lint_and_install(CT_YAML) + + for resolved_chart in resolved_charts: + test_chart(resolved_chart) + + +def discover_charts(helm_dir: Path) -> list[Path]: + return sorted( + p.parent.relative_to(helm_dir) for p in helm_dir.glob("charts/*/Chart.yaml") + ) + + +def test_chart(resolved_chart: ResolvedChart) -> None: + try: + meta = ChartMetadata.from_chart_dir(resolved_chart.absolute_path) + except (ValueError, ValidationError) as e: + raise ClickException(str(e)) from e + + if meta.type == "library": + click.echo(f"Skipping install for library chart: {meta.name}") + return + + execute_lint_and_install(CT_YAML, resolved_chart.path_relative_to_helm_dir) def create_test_namespace(ct_yaml_path: Path) -> str: @@ -46,8 +90,8 @@ def create_test_namespace(ct_yaml_path: Path) -> str: test_namespace: str | None = ct_yaml.get("namespace") if test_namespace is None: raise ClickException(f"namespace must be specified in {CT_YAML}") - namespaces = execute("kubectl", "get", "namespaces") - for line in namespaces.splitlines(): + namespaces = execute("kubectl", "get", "namespaces", capture_stdout=True) + for line in namespaces.stdout.splitlines(): ns, *_ = line.split() if ns == test_namespace: break @@ -110,7 +154,6 @@ def create_image_pull_secret(namespace: str) -> None: f"--namespace={namespace}", f"--from-file=.dockerconfigjson={auth_json_path}", "--type=kubernetes.io/dockerconfigjson", - capture_stdout=False, ) @@ -121,24 +164,27 @@ def secret_exists(namespace: str, secret_name: str) -> bool: "secrets", f"--namespace={namespace}", "--no-headers", + capture_stdout=True, ) - for line in existing_secrets.splitlines(): + for line in existing_secrets.stdout.splitlines(): existing_secret, *_ = line.split() if existing_secret == secret_name: return True return False -def execute_lint_and_install(ct_yaml_path: Path) -> None: +def execute_lint_and_install( + ct_yaml_path: Path, chart_path_relative_to_helm_dir: Path +) -> None: try: execute( "ct", "lint-and-install", "--config", str(ct_yaml_path), - "--all", - "--check-version-increment=false", - capture_stdout=False, + "--charts", + str(chart_path_relative_to_helm_dir), + "--check-version-increment=true", ) except CalledProcessError as e: raise ClickException(f"lint-and-install failed with rc={e.returncode}") from e diff --git a/local-k8s/src/local_k8s/commands/tools/commands.py b/local-k8s/src/local_k8s/commands/tools/commands.py index e506b2a..ee17dbd 100644 --- a/local-k8s/src/local_k8s/commands/tools/commands.py +++ b/local-k8s/src/local_k8s/commands/tools/commands.py @@ -68,9 +68,9 @@ def install_binary_tool(tool: RequiredTool, tools_dir: Path) -> Path: def install_helm_unit_tests() -> None: - plugins = execute("helm", "plugin", "list") + plugins = execute("helm", "plugin", "list", capture_stdout=True) LOG.info("Installing helm unit test") - for line in plugins.splitlines(): + for line in plugins.stdout.splitlines(): if line.startswith("unittest"): LOG.info("helm unittest already installed, skipping ") return @@ -109,6 +109,11 @@ def install_binary_tools() -> None: LOG.info(f"export PATH={bin_dir.resolve()}:$PATH") +@click.command("path") +def path() -> None: + click.echo(str(get_bin_dir().resolve())) + + @click.command("install") def install() -> None: system = platform.system() diff --git a/local-k8s/src/local_k8s/groups/chart.py b/local-k8s/src/local_k8s/groups/chart.py new file mode 100644 index 0000000..1a42b51 --- /dev/null +++ b/local-k8s/src/local_k8s/groups/chart.py @@ -0,0 +1,19 @@ +import click + +from local_k8s.commands.chart.lint import lint +from local_k8s.commands.chart.list_changed import list_changed +from local_k8s.commands.chart.metadata import metadata +from local_k8s.commands.chart.push import push +from local_k8s.commands.chart.test import test + + +@click.group("chart", help="Helm chart lint, test, and publish commands") +def chart() -> None: + pass + + +chart.add_command(metadata) +chart.add_command(list_changed) +chart.add_command(lint) +chart.add_command(push) +chart.add_command(test) diff --git a/local-k8s/src/local_k8s/groups/ct.py b/local-k8s/src/local_k8s/groups/ct.py deleted file mode 100644 index f740313..0000000 --- a/local-k8s/src/local_k8s/groups/ct.py +++ /dev/null @@ -1,11 +0,0 @@ -import click - -from local_k8s.commands.ct.lint_and_install import lint_and_install - - -@click.group("ct", help="Execute ct (chart testing) commands") -def ct() -> None: - pass - - -ct.add_command(lint_and_install) diff --git a/local-k8s/src/local_k8s/groups/tools.py b/local-k8s/src/local_k8s/groups/tools.py index a732879..c9dd374 100644 --- a/local-k8s/src/local_k8s/groups/tools.py +++ b/local-k8s/src/local_k8s/groups/tools.py @@ -1,6 +1,6 @@ import click -from local_k8s.commands.tools.commands import install, uninstall +from local_k8s.commands.tools.commands import install, path, uninstall @click.group("tools", help="Manage k8s tooling") @@ -9,4 +9,5 @@ def tools() -> None: tools.add_command(install) +tools.add_command(path) tools.add_command(uninstall) diff --git a/local-k8s/src/local_k8s/models.py b/local-k8s/src/local_k8s/models.py index b85e3a5..763eb50 100644 --- a/local-k8s/src/local_k8s/models.py +++ b/local-k8s/src/local_k8s/models.py @@ -63,3 +63,19 @@ def resolve_path(self) -> Path: class CiSecrets(BaseModel): model_config = ConfigDict(extra="forbid") secrets: list[CiSecret] + + +class ChartMetadata(BaseModel): + model_config = ConfigDict(extra="ignore") + + name: str = Field(min_length=1) + version: str = Field(min_length=1) + type: str = "application" + appVersion: str | None = None + + @classmethod + def from_chart_dir(cls, chart_dir: Path) -> Self: + chart_yaml = chart_dir / "Chart.yaml" + if not chart_yaml.is_file(): + raise ValueError(f"Chart.yaml not found at {chart_yaml}") + return cls.model_validate(yaml.safe_load(chart_yaml.read_text())) diff --git a/local-k8s/src/local_k8s/shared.py b/local-k8s/src/local_k8s/shared.py index b829fa4..b46e981 100644 --- a/local-k8s/src/local_k8s/shared.py +++ b/local-k8s/src/local_k8s/shared.py @@ -2,14 +2,48 @@ import os import subprocess import sys +from dataclasses import dataclass from pathlib import Path -from typing import Literal, overload + +from click import ClickException from local_k8s.static import TOOLS_BY_NAME LOG = logging.getLogger(__name__) +@dataclass(frozen=True) +class CommandResult: + returncode: int + stdout: str + stderr: str + + +@dataclass(frozen=True) +class ResolvedChart: + absolute_path: Path + path_relative_to_helm_dir: Path + + +def resolve_chart(helm_dir: Path, chart: Path) -> ResolvedChart: + """ + chart list-changed emits a relative path which we need to resolve for + other commands to accept. This ensures the CI workflow in .github can + operate without any funny string munging. + """ + absolute_path = chart.resolve() + try: + path_relative_to_helm_dir = absolute_path.relative_to(helm_dir) + except ValueError as e: + raise ClickException( + f"--chart {chart} is not inside --helm-dir {helm_dir}" + ) from e + return ResolvedChart( + absolute_path=absolute_path, + path_relative_to_helm_dir=path_relative_to_helm_dir, + ) + + def get_repo_name() -> str: if sys.prefix == sys.base_prefix: raise Exception("local-k8s must be run from an application virtualenv") @@ -44,18 +78,35 @@ def resolve(name: str) -> None: raise Exception(f"{name} not installed at {path}; run: local-k8s tools install") -@overload -def execute(*args: str, capture_stdout: Literal[True] = True) -> str: ... - - -@overload -def execute(*args: str, capture_stdout: Literal[False]) -> int: ... - - -def execute(*args: str, capture_stdout: bool = True) -> str | int: - resolve(args[0]) +def execute( + *args: str, + capture_stdout: bool = False, + capture_stderr: bool = False, + skip_resolve: bool = False, + check: bool = True, + input: str | None = None, +) -> CommandResult: + if not skip_resolve: + resolve(args[0]) LOG.debug("Executing: %s", list(args)) - if capture_stdout: - return subprocess.check_output(list(args), text=True) - else: - return subprocess.check_call(list(args)) + completed = subprocess.run( + list(args), + input=input, + text=True, + stdout=subprocess.PIPE if capture_stdout else None, + stderr=subprocess.PIPE if capture_stderr else None, + check=False, + ) + result = CommandResult( + returncode=completed.returncode, + stdout=completed.stdout or "", + stderr=completed.stderr or "", + ) + if check and result.returncode != 0: + raise subprocess.CalledProcessError( + result.returncode, + list(args), + output=result.stdout, + stderr=result.stderr, + ) + return result diff --git a/local-k8s/tests/_charts.py b/local-k8s/tests/_charts.py new file mode 100644 index 0000000..5037335 --- /dev/null +++ b/local-k8s/tests/_charts.py @@ -0,0 +1,9 @@ +from pathlib import Path + +import yaml + + +def write_chart(chart_dir: Path, chart_yaml: dict[str, object]) -> Path: + chart_dir.mkdir(parents=True, exist_ok=True) + (chart_dir / "Chart.yaml").write_text(yaml.dump(chart_yaml)) + return chart_dir diff --git a/local-k8s/tests/_fake_execute.py b/local-k8s/tests/_fake_execute.py new file mode 100644 index 0000000..c40cb23 --- /dev/null +++ b/local-k8s/tests/_fake_execute.py @@ -0,0 +1,63 @@ +from __future__ import annotations + +import subprocess +from dataclasses import dataclass, field +from typing import Any +from unittest.mock import call + +from local_k8s.shared import CommandResult + + +@dataclass +class _Rule: + prefix: tuple[str, ...] + result: CommandResult + raises: BaseException | None = None + + def matches(self, args: tuple[str, ...]) -> bool: + return args[: len(self.prefix)] == self.prefix + + +@dataclass +class FakeExecute: + _rules: list[_Rule] = field(default_factory=list) + calls: list[Any] = field(default_factory=list) # unittest.mock.call objects + + def on( + self, + *prefix: str, + stdout: str = "", + stderr: str = "", + returncode: int = 0, + raises: BaseException | None = None, + ) -> FakeExecute: + self._rules.append( + _Rule(prefix, CommandResult(returncode, stdout, stderr), raises) + ) + return self + + def calls_for(self, *prefix: str) -> list[Any]: + return [c for c in self.calls if c.args[: len(prefix)] == prefix] + + def __call__(self, *args: str, **kwargs: object) -> CommandResult: + self.calls.append(call(*args, **kwargs)) + for rule in self._rules: + if rule.matches(args): + result, raises = rule.result, rule.raises + break + else: + prefixes = [rule.prefix for rule in self._rules] + raise AssertionError( + f"No rule matched execute{args!r} with {kwargs!r}; " + f"configured prefixes: {prefixes!r}" + ) + if raises is not None: + raise raises + if kwargs.get("check", True) and result.returncode != 0: + raise subprocess.CalledProcessError( + result.returncode, + list(args), + output=result.stdout, + stderr=result.stderr, + ) + return result diff --git a/local-k8s/tests/conftest.py b/local-k8s/tests/conftest.py new file mode 100644 index 0000000..480a17e --- /dev/null +++ b/local-k8s/tests/conftest.py @@ -0,0 +1,27 @@ +from collections.abc import Callable +from pathlib import Path +from types import ModuleType + +import pytest + +from _fake_execute import FakeExecute + + +@pytest.fixture +def fake_execute( + monkeypatch: pytest.MonkeyPatch, +) -> Callable[[ModuleType], FakeExecute]: + def _install(module: ModuleType) -> FakeExecute: + fake = FakeExecute() + monkeypatch.setattr(module, "execute", fake) + return fake + + return _install + + +@pytest.fixture +def helm_dir(tmp_path: Path) -> Path: + d = tmp_path / "helm" + d.mkdir() + (d / "ct.yaml").write_text("namespace: test-ns\n") + return d diff --git a/local-k8s/tests/test_chart_list_changed.py b/local-k8s/tests/test_chart_list_changed.py new file mode 100644 index 0000000..b4d24df --- /dev/null +++ b/local-k8s/tests/test_chart_list_changed.py @@ -0,0 +1,48 @@ +from collections.abc import Callable +from pathlib import Path +from types import ModuleType +from unittest.mock import call + +from click.testing import CliRunner + +from _fake_execute import FakeExecute +from local_k8s.cli import cli +from local_k8s.commands.chart import list_changed as list_changed_module + + +def test_list_changed_passes_repo_relative_paths( + tmp_path: Path, + helm_dir: Path, + fake_execute: Callable[[ModuleType], FakeExecute], +) -> None: + fake = fake_execute(list_changed_module) + fake.on("git", stdout=f"{tmp_path.resolve()}\n") + fake.on("ct", "list-changed") + + result = CliRunner().invoke( + cli, ["chart", "list-changed", "--helm-dir", str(helm_dir)] + ) + + assert result.exit_code == 0 + assert fake.calls == [ + call( + "git", + "-C", + str(helm_dir.absolute()), + "rev-parse", + "--show-toplevel", + skip_resolve=True, + capture_stdout=True, + ), + call( + "ct", + "list-changed", + "--config", + "helm/ct.yaml", + "--chart-dirs", + "helm/charts", + "--target-branch", + "main", + capture_stdout=True, + ), + ] diff --git a/local-k8s/tests/test_chart_metadata.py b/local-k8s/tests/test_chart_metadata.py new file mode 100644 index 0000000..fd70726 --- /dev/null +++ b/local-k8s/tests/test_chart_metadata.py @@ -0,0 +1,30 @@ +import json +from pathlib import Path + +from click.testing import CliRunner + +from _charts import write_chart +from local_k8s.cli import cli + + +def test_metadata_full_fields(tmp_path: Path) -> None: + chart_yaml = { + "name": "mychart", + "version": "1.2.3", + "type": "library", + "appVersion": "9.0", + } + chart_dir = write_chart(tmp_path / "mychart", chart_yaml) + + result = CliRunner().invoke(cli, ["chart", "metadata", "--chart", str(chart_dir)]) + + assert result.exit_code == 0 + assert json.loads(result.stdout) == chart_yaml + + +def test_metadata_missing_version_fails(tmp_path: Path) -> None: + chart_dir = write_chart(tmp_path / "mychart", {"name": "mychart"}) + + result = CliRunner().invoke(cli, ["chart", "metadata", "--chart", str(chart_dir)]) + + assert result.exit_code != 0 diff --git a/local-k8s/tests/test_chart_push.py b/local-k8s/tests/test_chart_push.py new file mode 100644 index 0000000..6911bd9 --- /dev/null +++ b/local-k8s/tests/test_chart_push.py @@ -0,0 +1,58 @@ +from collections.abc import Callable +from pathlib import Path +from types import ModuleType +from unittest.mock import call + +from click.testing import CliRunner + +from _charts import write_chart +from _fake_execute import FakeExecute +from local_k8s.cli import cli +from local_k8s.commands.chart import push as push_module + + +def test_push_fail_if_exists_aborts_when_chart_present( + tmp_path: Path, fake_execute: Callable[[ModuleType], FakeExecute] +) -> None: + chart = write_chart( + tmp_path / "charts" / "ewb", {"name": "ewb", "version": "1.2.3"} + ) + + auth = tmp_path / "auth.json" + auth.write_text("{}\n") + + fake = fake_execute(push_module) + fake.on("helm", "show", "chart") + + result = CliRunner().invoke( + cli, + [ + "chart", + "push", + "--chart", + str(chart), + "--registry-config", + str(auth), + "--oci-repo", + "org/repo", + "--fail-if-exists", + ], + ) + + assert result.exit_code != 0 + assert "already exists" in result.output + assert fake.calls == [ + call( + "helm", + "show", + "chart", + "oci://ghcr.io/org/repo/ewb", + "--version", + "1.2.3", + "--registry-config", + str(auth), + check=False, + capture_stdout=True, + capture_stderr=True, + ), + ] diff --git a/local-k8s/tests/test_chart_test.py b/local-k8s/tests/test_chart_test.py new file mode 100644 index 0000000..1e93da5 --- /dev/null +++ b/local-k8s/tests/test_chart_test.py @@ -0,0 +1,177 @@ +from collections.abc import Callable +from pathlib import Path +from types import ModuleType +from unittest.mock import call + +import pytest +from click.testing import CliRunner + +from _charts import write_chart +from _fake_execute import FakeExecute +from local_k8s.cli import cli +from local_k8s.commands.chart import test as test_module + + +def _write_chart( + helm_dir: Path, chart_dir_name: str, chart_type: str | None = None +) -> Path: + chart_yaml: dict[str, object] = {"name": chart_dir_name, "version": "1.0.0"} + if chart_type is not None: + chart_yaml["type"] = chart_type + return write_chart(helm_dir / "charts" / chart_dir_name, chart_yaml) + + +@pytest.fixture +def patched_image_secret(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + auth_json = tmp_path / "auth.json" + auth_json.write_text("{}\n") + monkeypatch.setattr(test_module, "IMAGE_SECRET_PATHS", [auth_json]) + + +def _install_chart_execute( + fake_execute: Callable[[ModuleType], FakeExecute], +) -> FakeExecute: + return ( + fake_execute(test_module) + .on("kubectl", "get", "namespaces", stdout="test-ns\n") + .on("kubectl", "get", "secrets") + .on("kubectl", "create", "secret") + ) + + +def test_test_missing_ct_yaml_fails(tmp_path: Path) -> None: + helm_dir = tmp_path / "helm" + helm_dir.mkdir() + chart_dir = _write_chart(helm_dir, "mychart") + + result = CliRunner().invoke( + cli, + ["chart", "test", "--helm-dir", str(helm_dir), "--chart", str(chart_dir)], + ) + + assert result.exit_code != 0 + assert "ct.yaml" in result.output + + +def test_library_chart_skips_install( + helm_dir: Path, + patched_image_secret: None, + fake_execute: Callable[[ModuleType], FakeExecute], +) -> None: + chart_dir = _write_chart(helm_dir, "mylib", chart_type="library") + fake = _install_chart_execute(fake_execute) + + result = CliRunner().invoke( + cli, ["chart", "test", "--helm-dir", str(helm_dir), "--chart", str(chart_dir)] + ) + + assert result.exit_code == 0, result.output + assert "Skipping install" in result.output + assert fake.calls_for("ct") == [] + + +def test_application_chart_runs_lint_and_install( + tmp_path: Path, + helm_dir: Path, + patched_image_secret: None, + fake_execute: Callable[[ModuleType], FakeExecute], + monkeypatch: pytest.MonkeyPatch, +) -> None: + _write_chart(helm_dir, "myapp") + fake = _install_chart_execute(fake_execute) + fake.on("ct", "lint-and-install") + + # Mirrors the real workflow contract: invoked from repo root with the + # repo-root-relative path that `chart list-changed` would emit. + monkeypatch.chdir(tmp_path) + result = CliRunner().invoke( + cli, + ["chart", "test", "--helm-dir", "helm", "--chart", "helm/charts/myapp"], + ) + + assert result.exit_code == 0, result.output + assert fake.calls_for("ct")[-1] == call( + "ct", + "lint-and-install", + "--config", + "ct.yaml", + "--charts", + "charts/myapp", + "--check-version-increment=true", + ) + + +def test_application_chart_lint_and_install_failure_raises( + helm_dir: Path, + patched_image_secret: None, + fake_execute: Callable[[ModuleType], FakeExecute], +) -> None: + chart_dir = _write_chart(helm_dir, "myapp") + fake = _install_chart_execute(fake_execute) + fake.on("ct", "lint-and-install", returncode=3) + + result = CliRunner().invoke( + cli, ["chart", "test", "--helm-dir", str(helm_dir), "--chart", str(chart_dir)] + ) + + assert result.exit_code != 0 + assert "rc=3" in result.output + + +def test_chart_outside_helm_dir_fails( + tmp_path: Path, + helm_dir: Path, + patched_image_secret: None, + fake_execute: Callable[[ModuleType], FakeExecute], +) -> None: + outside_chart = _write_chart(tmp_path / "other", "myapp") + + result = CliRunner().invoke( + cli, + ["chart", "test", "--helm-dir", str(helm_dir), "--chart", str(outside_chart)], + ) + + assert result.exit_code != 0 + assert "not inside --helm-dir" in result.output + + +def test_discovery_mode_processes_all_charts_and_skips_libraries( + helm_dir: Path, + patched_image_secret: None, + fake_execute: Callable[[ModuleType], FakeExecute], +) -> None: + _write_chart(helm_dir, "app-a") + _write_chart(helm_dir, "app-b") + _write_chart(helm_dir, "lib-a", chart_type="library") + fake = _install_chart_execute(fake_execute) + fake.on("ct", "lint-and-install") + + result = CliRunner().invoke(cli, ["chart", "test", "--helm-dir", str(helm_dir)]) + + assert result.exit_code == 0, result.output + assert "Skipping install" in result.output + + assert len(fake.calls_for("kubectl", "get", "namespaces")) == 1, ( + "namespace/secret setup must run once, not per chart" + ) + + assert fake.calls_for("ct") == [ + call( + "ct", + "lint-and-install", + "--config", + "ct.yaml", + "--charts", + "charts/app-a", + "--check-version-increment=true", + ), + call( + "ct", + "lint-and-install", + "--config", + "ct.yaml", + "--charts", + "charts/app-b", + "--check-version-increment=true", + ), + ] diff --git a/local-k8s/tests/test_shared.py b/local-k8s/tests/test_shared.py index 842d720..4309027 100644 --- a/local-k8s/tests/test_shared.py +++ b/local-k8s/tests/test_shared.py @@ -3,7 +3,7 @@ import pytest import local_k8s.shared as shared -from local_k8s.shared import resolve +from local_k8s.shared import ResolvedChart, resolve, resolve_chart @pytest.fixture @@ -11,6 +11,7 @@ def bin_dir(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: d = tmp_path / "bin" d.mkdir() monkeypatch.setattr(shared, "get_bin_dir", lambda: d) + monkeypatch.setenv("PATH", f"{d}:{tmp_path}") return d @@ -36,3 +37,28 @@ def test_resolve_not_executable(bin_dir: Path) -> None: def test_resolve_unknown(bin_dir: Path) -> None: with pytest.raises(Exception, match="Failed to locate tool"): resolve("nosuch") + + +def test_resolve_chart_accepts_repo_root_relative_path( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + helm_dir = (tmp_path / "helm").resolve() + chart_dir = helm_dir / "charts" / "myapp" + chart_dir.mkdir(parents=True) + + monkeypatch.chdir(tmp_path) + result = resolve_chart(helm_dir, Path("helm/charts/myapp")) + + assert result == ResolvedChart( + absolute_path=chart_dir, + path_relative_to_helm_dir=Path("charts/myapp"), + ) + + +def test_resolve_chart_outside_helm_dir_raises(tmp_path: Path) -> None: + helm_dir = (tmp_path / "helm").resolve() + helm_dir.mkdir() + outside_chart = (tmp_path / "other" / "myapp").resolve() + + with pytest.raises(Exception, match="not inside --helm-dir"): + resolve_chart(helm_dir, outside_chart)