feat(admin): add jmp admin apply for Jumpstarter manifests

python/packages/jumpstarter-cli-admin/jumpstarter_cli_admin/__init__.py

@@ -3,6 +3,7 @@
 from jumpstarter_cli_common.opt import opt_log_level
 from jumpstarter_cli_common.version import version
 
+from .apply import apply
 from .create import create
 from .delete import delete
 from .get import get
@@ -17,6 +18,7 @@ def admin():
 
 
 admin.add_command(get)
+admin.add_command(apply)
 admin.add_command(create)
 admin.add_command(delete)
 admin.add_command(import_res)

python/packages/jumpstarter-cli-admin/jumpstarter_cli_admin/apply.py

@@ -0,0 +1,95 @@
+from http import HTTPStatus
+from typing import IO, Optional
+
+import click
+from jumpstarter_cli_common.blocking import blocking
+from jumpstarter_cli_common.opt import (
+    OutputType,
+    opt_context,
+    opt_kubeconfig,
+    opt_namespace,
+    opt_output_all,
+)
+from jumpstarter_cli_common.print import model_print
+from jumpstarter_kubernetes import ApplyV1Alpha1Api, ManifestError, load_manifests
+from kubernetes_asyncio.client.exceptions import ApiException
+from kubernetes_asyncio.config.config_exception import ConfigException
+
+from .k8s import (
+    handle_k8s_api_exception,
+    handle_k8s_config_exception,
+)
+
+
+@click.command("apply")
+@click.option(
+    "-f",
+    "--filename",
+    "filenames",
+    type=click.File("r"),
+    multiple=True,
+    required=True,
+    help="Manifest to apply, or '-' to read from stdin. Can be set multiple times.",
+)
+@click.option(
+    "--force-conflicts",
+    is_flag=True,
+    default=False,
+    help="Take ownership of fields another manager holds, instead of reporting the conflict.",
+)
+@click.option(
+    "--dry-run",
+    is_flag=True,
+    default=False,
+    help="Send the manifests to the server for validation without persisting them.",
+)
+@opt_namespace
+@opt_kubeconfig
+@opt_context
+@opt_output_all
+@blocking
+async def apply(
+    filenames: tuple[IO, ...],
+    force_conflicts: bool,
+    dry_run: bool,
+    namespace: str,
+    kubeconfig: Optional[str],
+    context: Optional[str],
+    output: OutputType,
+):
+    """Apply Jumpstarter manifests to a Kubernetes cluster
+
+    Creates the resources in a manifest, or updates them in place when they
+    already exist. Only Jumpstarter resources (clients, exporters, exporter
+    sets, virtual target classes) can be applied.
+    """
+    try:
+        manifests = []
+        for file in filenames:
+            manifests.extend(load_manifests(file.read(), file.name))
+    except ManifestError as e:
+        raise click.ClickException(str(e)) from e
+
+    try:
+        async with ApplyV1Alpha1Api(namespace, kubeconfig, context) as api:
+            applied = await api.apply_all(manifests, dry_run=dry_run, force_conflicts=force_conflicts)
+    except ManifestError as e:
+        raise click.ClickException(str(e)) from e
+    except ApiException as e:
+        try:
+            handle_k8s_api_exception(e)
+        except click.ClickException as error:
+            if e.status == HTTPStatus.CONFLICT and not force_conflicts:
+                # The conflict names the fields; say how to win them on purpose.
+                error.message += "\nRe-run with --force-conflicts to take ownership of those fields."
+            raise
+    except ConfigException as e:
+        handle_k8s_config_exception(e)
+
+    if output is None:
+        # Mirror kubectl: one line per resource saying what happened to it.
+        suffix = " (dry run)" if dry_run else ""
+        for resource in applied.items:
+            click.echo(f"{resource.qualified_name} {resource.action}{suffix}")
+    else:
+        model_print(applied, output, namespace=namespace)

python/packages/jumpstarter-cli-admin/jumpstarter_cli_admin/apply_test.py

@@ -0,0 +1,203 @@
+import json
+from unittest.mock import AsyncMock, patch
+
+from click.testing import CliRunner
+from jumpstarter_kubernetes import (
+    ApplyV1Alpha1Api,
+    V1Alpha1AppliedResource,
+    V1Alpha1AppliedResourceList,
+)
+from kubernetes_asyncio.client.exceptions import ApiException
+
+from .apply import apply
+
+CLIENT_MANIFEST = """\
+apiVersion: jumpstarter.dev/v1alpha1
+kind: Client
+metadata:
+  name: hello
+"""
+
+EXPORTER_SET_MANIFEST = """\
+apiVersion: virtualtarget.jumpstarter.dev/v1alpha1
+kind: ExporterSet
+metadata:
+  name: pool
+"""
+
+
+def applied(kind: str, name: str, action: str, api_version: str = "jumpstarter.dev/v1alpha1"):
+    return V1Alpha1AppliedResource(
+        apiVersion=api_version,
+        kind=kind,
+        name=name,
+        namespace="default",
+        action=action,
+        resource={"apiVersion": api_version, "kind": kind, "metadata": {"name": name}},
+    )
+
+
+def run(args, files: dict[str, str], apply_all: AsyncMock, input: str | None = None):
+    runner = CliRunner()
+    with runner.isolated_filesystem():
+        for filename, contents in files.items():
+            with open(filename, "w") as f:
+                f.write(contents)
+        with (
+            patch.object(ApplyV1Alpha1Api, "_load_kube_config", AsyncMock()),
+            patch.object(ApplyV1Alpha1Api, "apply_all", apply_all),
+        ):
+            return runner.invoke(apply, args, input=input)
+
+
+def test_apply_reports_what_happened_to_each_resource():
+    apply_all = AsyncMock(
+        return_value=V1Alpha1AppliedResourceList(
+            items=[
+                applied("Client", "hello", "created"),
+                applied("ExporterSet", "pool", "configured", "virtualtarget.jumpstarter.dev/v1alpha1"),
+            ]
+        )
+    )
+
+    result = run(
+        ["-f", "client.yaml", "-f", "set.yaml"],
+        {"client.yaml": CLIENT_MANIFEST, "set.yaml": EXPORTER_SET_MANIFEST},
+        apply_all,
+    )
+
+    assert result.exit_code == 0
+    assert "client.jumpstarter.dev/hello created" in result.output
+    assert "exporterset.virtualtarget.jumpstarter.dev/pool configured" in result.output
+    # Both files reach the cluster in the order they were given.
+    assert [m["kind"] for m in apply_all.call_args.args[0]] == ["Client", "ExporterSet"]
+    assert apply_all.call_args.kwargs["dry_run"] is False
+
+
+def test_apply_reads_a_multi_document_manifest():
+    apply_all = AsyncMock(
+        return_value=V1Alpha1AppliedResourceList(
+            items=[
+                applied("Client", "hello", "created"),
+                applied("ExporterSet", "pool", "created", "virtualtarget.jumpstarter.dev/v1alpha1"),
+            ]
+        )
+    )
+
+    result = run(["-f", "both.yaml"], {"both.yaml": f"{CLIENT_MANIFEST}---\n{EXPORTER_SET_MANIFEST}"}, apply_all)
+
+    assert result.exit_code == 0
+    assert [m["kind"] for m in apply_all.call_args.args[0]] == ["Client", "ExporterSet"]
+
+
+def test_apply_says_when_nothing_was_persisted():
+    apply_all = AsyncMock(return_value=V1Alpha1AppliedResourceList(items=[applied("Client", "hello", "created")]))
+
+    result = run(["-f", "client.yaml", "--dry-run"], {"client.yaml": CLIENT_MANIFEST}, apply_all)
+
+    assert result.exit_code == 0
+    assert "client.jumpstarter.dev/hello created (dry run)" in result.output
+    assert apply_all.call_args.kwargs["dry_run"] is True
+
+
+def test_apply_prints_names_only():
+    apply_all = AsyncMock(return_value=V1Alpha1AppliedResourceList(items=[applied("Client", "hello", "created")]))
+
+    result = run(["-f", "client.yaml", "-o", "name"], {"client.yaml": CLIENT_MANIFEST}, apply_all)
+
+    assert result.exit_code == 0
+    assert result.output.strip() == "client.jumpstarter.dev/hello"
+
+
+def test_apply_prints_the_stored_resources_as_json():
+    apply_all = AsyncMock(return_value=V1Alpha1AppliedResourceList(items=[applied("Client", "hello", "created")]))
+
+    result = run(["-f", "client.yaml", "-o", "json"], {"client.yaml": CLIENT_MANIFEST}, apply_all)
+
+    assert result.exit_code == 0
+    parsed = json.loads(result.output)
+    assert parsed["items"][0]["action"] == "created"
+    assert parsed["items"][0]["resource"]["metadata"]["name"] == "hello"
+
+
+def test_apply_reads_a_manifest_from_stdin():
+    apply_all = AsyncMock(return_value=V1Alpha1AppliedResourceList(items=[applied("Client", "hello", "created")]))
+
+    result = run(["-f", "-"], {}, apply_all, input=CLIENT_MANIFEST)
+
+    assert result.exit_code == 0
+    assert [m["kind"] for m in apply_all.call_args.args[0]] == ["Client"]
+
+
+def test_apply_rejects_a_manifest_that_is_not_jumpstarters():
+    apply_all = AsyncMock()
+
+    result = run(
+        ["-f", "secret.yaml"],
+        {"secret.yaml": "apiVersion: v1\nkind: Secret\nmetadata:\n  name: creds\n"},
+        apply_all,
+    )
+
+    assert result.exit_code != 0
+    assert "core Kubernetes resource" in result.output
+    # Nothing reaches the cluster when the manifest is refused.
+    apply_all.assert_not_awaited()
+
+
+def test_apply_requires_a_manifest():
+    result = run([], {}, AsyncMock())
+
+    assert result.exit_code != 0
+    assert "Missing option" in result.output
+
+
+CONFLICT_BODY = json.dumps(
+    {
+        "kind": "Status",
+        "reason": "Conflict",
+        "message": 'Apply failed with 1 conflict: conflict with "other-controller": .metadata.labels.owner',
+    }
+)
+
+
+def conflict() -> ApiException:
+    error = ApiException(status=409, reason="Conflict")
+    error.body = CONFLICT_BODY
+    return error
+
+
+def test_apply_says_how_to_resolve_a_field_ownership_conflict():
+    apply_all = AsyncMock(side_effect=conflict())
+
+    result = run(["-f", "client.yaml"], {"client.yaml": CLIENT_MANIFEST}, apply_all)
+
+    assert result.exit_code != 0
+    assert ".metadata.labels.owner" in result.output
+    assert "--force-conflicts" in result.output
+
+
+def test_apply_does_not_suggest_a_flag_that_is_already_set():
+    apply_all = AsyncMock(side_effect=conflict())
+
+    result = run(["--force-conflicts", "-f", "client.yaml"], {"client.yaml": CLIENT_MANIFEST}, apply_all)
+
+    assert result.exit_code != 0
+    assert "--force-conflicts" not in result.output
+
+
+def test_apply_asks_the_cluster_to_take_ownership_when_told_to():
+    apply_all = AsyncMock(return_value=V1Alpha1AppliedResourceList(items=[applied("Client", "hello", "configured")]))
+
+    result = run(["--force-conflicts", "-f", "client.yaml"], {"client.yaml": CLIENT_MANIFEST}, apply_all)
+
+    assert result.exit_code == 0
+    assert apply_all.call_args.kwargs["force_conflicts"] is True
+
+
+def test_apply_leaves_conflicting_fields_alone_by_default():
+    apply_all = AsyncMock(return_value=V1Alpha1AppliedResourceList(items=[applied("Client", "hello", "configured")]))
+
+    result = run(["-f", "client.yaml"], {"client.yaml": CLIENT_MANIFEST}, apply_all)
+
+    assert result.exit_code == 0
+    assert apply_all.call_args.kwargs["force_conflicts"] is False

python/packages/jumpstarter-cli-admin/jumpstarter_cli_admin/k8s.py

@@ -10,10 +10,21 @@ def handle_k8s_api_exception(e: ApiException):
     # Try to parse the JSON response
     try:
         json_body = json.loads(e.body)
-        raise click.ClickException(f"Error from server ({json_body['reason']}): {json_body['message']}") from e
-    except json.decoder.JSONDecodeError:
+    except (json.decoder.JSONDecodeError, TypeError):
         raise click.ClickException(f"Server error: {e.body}") from e
 
+    # Valid JSON is not necessarily a Status: a proxy in front of the API server
+    # can answer with a bare string, a list, or null.
+    if not isinstance(json_body, dict):
+        raise click.ClickException(f"Server error: {e.body}") from e
+
+    # Not every Status carries a reason: a 500 from an admission or schema
+    # check often has only a message, and losing it leaves nothing to act on.
+    message = json_body.get("message") or e.reason or "unknown error"
+    reason = json_body.get("reason")
+    prefix = f"Error from server ({reason})" if reason else "Error from server"
+    raise click.ClickException(f"{prefix}: {message}") from e
+
 
 def handle_k8s_config_exception(e: ConfigException):
     """Handle a Kubernetes config exception"""

python/packages/jumpstarter-cli-admin/jumpstarter_cli_admin/k8s_test.py

@@ -0,0 +1,57 @@
+import click
+import pytest
+from kubernetes_asyncio.client.exceptions import ApiException
+
+from .k8s import handle_k8s_api_exception
+
+
+def api_exception(status: int, body, reason: str | None = None) -> ApiException:
+    error = ApiException(status=status, reason=reason)
+    error.body = body
+    return error
+
+
+def test_reports_the_reason_and_message_from_the_server():
+    error = api_exception(404, '{"reason":"NotFound","message":"clients.jumpstarter.dev \\"nope\\" not found"}')
+
+    with pytest.raises(click.ClickException, match=r"Error from server \(NotFound\): .*not found"):
+        handle_k8s_api_exception(error)
+
+
+def test_reports_a_status_that_carries_no_reason():
+    # Schema and admission failures answer with a message and nothing else.
+    error = api_exception(500, '{"kind":"Status","message":".provisioner: field not declared in schema","code":500}')
+
+    with pytest.raises(click.ClickException, match="Error from server: .provisioner: field not declared in schema"):
+        handle_k8s_api_exception(error)
+
+
+def test_falls_back_to_the_http_reason_when_the_body_says_nothing():
+    error = api_exception(503, "{}", reason="Service Unavailable")
+
+    with pytest.raises(click.ClickException, match="Error from server: Service Unavailable"):
+        handle_k8s_api_exception(error)
+
+
+def test_passes_through_a_body_that_is_not_json():
+    error = api_exception(502, "<html>bad gateway</html>")
+
+    with pytest.raises(click.ClickException, match="Server error: <html>bad gateway</html>"):
+        handle_k8s_api_exception(error)
+
+
+@pytest.mark.parametrize("body", ["null", '"just a string"', "[1,2,3]"])
+def test_passes_through_json_that_is_not_a_status(body):
+    # A proxy in front of the API server can answer with valid JSON that is not
+    # a Status object; reading it as one used to raise AttributeError.
+    error = api_exception(502, body)
+
+    with pytest.raises(click.ClickException, match="Server error: "):
+        handle_k8s_api_exception(error)
+
+
+def test_passes_through_a_missing_body():
+    error = api_exception(500, None)
+
+    with pytest.raises(click.ClickException, match="Server error: None"):
+        handle_k8s_api_exception(error)