OSAC-1221: add InstanceType E2E tests

tests/core/grpc_client.py

@@ -293,3 +293,47 @@ def update_compute_instance_catalog_item(self, *, catalog_item_id: str, **fields
 
     def delete_compute_instance_catalog_item(self, *, catalog_item_id: str) -> None:
         self.call(service=f"{PRIVATE_API}.ComputeInstanceCatalogItems/Delete", data={"id": catalog_item_id})
+
+    # InstanceType operations (private API only)
+
+    def create_instance_type(
+        self,
+        *,
+        name: str,
+        cores: int,
+        memory_gib: int,
+        description: str = "",
+    ) -> str:
+        response: dict[str, Any] = self.call(
+            service=f"{PRIVATE_API}.InstanceTypes/Create",
+            data={
+                "object": {
+                    "metadata": {"name": name},
+                    "spec": {
+                        "cores": cores,
+                        "memory_gib": memory_gib,
+                        "description": description,
+                    },
+                }
+            },
+        )
+        return response["object"]["id"]
+
+    def get_instance_type(self, *, name: str) -> dict[str, Any]:
+        return self.call(service=f"{PRIVATE_API}.InstanceTypes/Get", data={"id": name})
+
+    def list_instance_type_names(self) -> list[str]:
+        response: dict[str, Any] = self.call(service=f"{PRIVATE_API}.InstanceTypes/List")
+        return [item["metadata"]["name"] for item in response.get("items", [])]
+
+    def update_instance_type(self, *, name: str, state: str) -> dict[str, Any]:
+        return self.call(
+            service=f"{PRIVATE_API}.InstanceTypes/Update",
+            data={
+                "object": {"id": name, "spec": {"state": state}},
+                "updateMask": {"paths": ["spec.state"]},
+            },
+        )
+
+    def delete_instance_type(self, *, name: str) -> None:
+        self.call(service=f"{PRIVATE_API}.InstanceTypes/Delete", data={"id": name})

tests/core/osac_cli.py

@@ -23,6 +23,10 @@ def __init__(self, *, binary: str, address: str, token_script: str, namespace: s
     def close(self) -> None:
         shutil.rmtree(self._config_dir, ignore_errors=True)
 
+    @property
+    def config_dir(self) -> str:
+        return self._config_dir
+
     def _run(self, *args: str, timeout: int = 300) -> str:
         return run(self.binary, "--config", self._config_dir, *args, timeout=timeout)
 
@@ -46,23 +50,20 @@ def create_compute_instance(
         *,
         template: str,
         network_attachments: list[dict[str, Any]] | None = None,
-        cores: int = 2,
-        memory_gib: int = 4,
+        cores: int | None = None,
+        memory_gib: int | None = None,
         boot_disk_size: int = 20,
         image: str = "quay.io/containerdisks/fedora:latest",
         image_source_type: str = "registry",
         run_strategy: str = "Always",
         user_data_secret_ref: str | None = None,
+        instance_type: str | None = None,
     ) -> str:
         args: list[str] = [
             "create",
             "computeinstance",
             "--template",
             template,
-            "--cores",
-            str(cores),
-            "--memory-gib",
-            str(memory_gib),
             "--boot-disk-size",
             str(boot_disk_size),
             "--image",
@@ -73,6 +74,13 @@ def create_compute_instance(
             run_strategy,
         ]
 
+        if instance_type is not None:
+            if cores is not None or memory_gib is not None:
+                raise ValueError("Cannot specify cores/memory_gib together with instance_type")
+            args.extend(["--instance-type", instance_type])
+        else:
+            args.extend(["--cores", str(2 if cores is None else cores), "--memory-gib", str(4 if memory_gib is None else memory_gib)])
+
         # Add network attachments
         if network_attachments is not None:
             for idx, attachment in enumerate(network_attachments):
@@ -107,6 +115,18 @@ def create_compute_instance(
     def delete_compute_instance(self, *, uuid: str) -> None:
         self._run("delete", "computeinstance", uuid)
 
+    def create_instance_type(self, *, name: str, cores: int, memory_gib: int, description: str = "") -> str:
+        args: list[str] = ["create", "instancetype", "--name", name, "--cores", str(cores), "--memory-gib", str(memory_gib)]
+        if description:
+            args.extend(["--description", description])
+        return self._parse_uuid(self._run(*args))
+
+    def describe_instance_type(self, *, name: str) -> str:
+        return self._run("describe", "instancetype", name)
+
+    def delete_instance_type(self, *, name: str) -> None:
+        self._run("delete", "instancetype", name)
+
     def create_cluster(
         self,
         *,

tests/vmaas/test_compute_instance_instance_type.py

@@ -0,0 +1,245 @@
+from __future__ import annotations
+
+import re
+import subprocess
+from typing import Any, Iterator
+from uuid import uuid4
+
+import pytest
+
+from tests.core.grpc_client import GRPCClient
+from tests.core.helpers import wait_for_cr, wait_for_deletion
+from tests.core.k8s_client import K8sClient
+from tests.core.osac_cli import OsacCLI
+from tests.core.runner import run_unchecked
+
+IT_CORES: int = 2
+IT_MEMORY_GIB: int = 4
+
+
+def _build_create_ci_args(
+    cli: OsacCLI,
+    vm_template: str,
+    default_subnet: str,
+    it_name: str,
+    ci_name: str | None = None,
+) -> list[str]:
+    args = [cli.binary, "--config", cli.config_dir, "create", "computeinstance"]
+    if ci_name is not None:
+        args += ["--name", ci_name]
+    args += [
+        "--template", vm_template,
+        "--network-attachment", f"subnet={default_subnet}",
+        "--instance-type", it_name,
+        "--boot-disk-size", "20",
+        "--image", "quay.io/containerdisks/fedora:latest",
+        "--image-source-type", "registry",
+        "--run-strategy", "Always",
+    ]
+    return args
+
+
+@pytest.fixture
+def active_instance_type(private_grpc: GRPCClient) -> Iterator[str]:
+    """Create an ACTIVE instance type for testing; clean up after."""
+    it_name = f"e2e-ci-it-{uuid4().hex[:8]}"
+    private_grpc.create_instance_type(
+        name=it_name,
+        cores=IT_CORES,
+        memory_gib=IT_MEMORY_GIB,
+        description="E2E compute instance test type",
+    )
+    yield it_name
+    try:
+        private_grpc.delete_instance_type(name=it_name)
+    except subprocess.CalledProcessError as e:
+        output = ((e.stdout or "") + (e.stderr or "")).lower()
+        if "not found" not in output:
+            raise
+
+
+def test_compute_instance_happy_path(
+    cli: OsacCLI,
+    grpc: GRPCClient,
+    k8s_hub_client: K8sClient,
+    default_subnet: str,
+    vm_template: str,
+    active_instance_type: str,
+) -> None:
+    ci_uuid: str | None = None
+    ci_name: str | None = None
+
+    try:
+        ci_uuid = cli.create_compute_instance(
+            template=vm_template,
+            network_attachments=[{"subnet": default_subnet}],
+            instance_type=active_instance_type,
+        )
+        assert ci_uuid in grpc.list_compute_instance_ids(), (
+            f"ComputeInstance {ci_uuid} not found in list after create"
+        )
+
+        # Wait for CR and verify reconciler expansion
+        ci_name = wait_for_cr(k8s=k8s_hub_client, uuid=ci_uuid)
+        ci_obj: dict[str, Any] = k8s_hub_client.get_json(resource="computeinstance", name=ci_name)
+        spec: dict[str, Any] = ci_obj["spec"]
+        assert spec["cores"] == IT_CORES, (
+            f"E2E-02: reconciler should expand cores from instance type: {spec['cores']} != {IT_CORES}"
+        )
+        assert spec["memoryGiB"] == IT_MEMORY_GIB, (
+            f"E2E-02: reconciler should expand memory from instance type: {spec['memoryGiB']} != {IT_MEMORY_GIB}"
+        )
+
+        # Verify osac.openshift.io/instance-type-name label (E2E-03)
+        labels: dict[str, str] = ci_obj["metadata"].get("labels", {})
+        assert labels.get("osac.openshift.io/instance-type-name") == active_instance_type, (
+            f"E2E-03: osac.openshift.io/instance-type-name label mismatch: "
+            f"{labels.get('osac.openshift.io/instance-type-name')!r} != {active_instance_type!r}"
+        )
+    finally:
+        if ci_uuid is not None:
+            cli.delete_compute_instance(uuid=ci_uuid)
+            if ci_name is not None:
+                wait_for_deletion(k8s=k8s_hub_client, name=ci_name)
+
+
+def test_compute_instance_deletion_protection(
+    cli: OsacCLI,
+    grpc: GRPCClient,
+    k8s_hub_client: K8sClient,
+    default_subnet: str,
+    vm_template: str,
+    active_instance_type: str,
+) -> None:
+    ci_uuid: str | None = None
+    ci_name: str | None = None
+
+    try:
+        ci_uuid = cli.create_compute_instance(
+            template=vm_template,
+            network_attachments=[{"subnet": default_subnet}],
+            instance_type=active_instance_type,
+        )
+        assert ci_uuid in grpc.list_compute_instance_ids(), (
+            f"ComputeInstance {ci_uuid} not found in list after create"
+        )
+        ci_name = wait_for_cr(k8s=k8s_hub_client, uuid=ci_uuid)
+
+        output, rc = run_unchecked(
+            cli.binary, "--config", cli.config_dir, "delete", "instancetype", active_instance_type,
+        )
+        assert rc != 0, "delete should be rejected when ComputeInstance references instance type"
+        error_lower = output.lower()
+        assert any(term in error_lower for term in [
+            "409", "conflict", "referenced", "failedprecondition", "in use",
+        ]), f"Expected conflict/reference error, got: {output}"
+    finally:
+        if ci_uuid is not None:
+            cli.delete_compute_instance(uuid=ci_uuid)
+            if ci_name is not None:
+                wait_for_deletion(k8s=k8s_hub_client, name=ci_name)
+
+
+def test_compute_instance_deprecated_warning(
+    cli: OsacCLI,
+    private_grpc: GRPCClient,
+    k8s_hub_client: K8sClient,
+    default_subnet: str,
+    vm_template: str,
+    active_instance_type: str,
+) -> None:
+    deprecated_ci_uuid: str | None = None
+    deprecated_ci_name: str | None = None
+
+    try:
+        private_grpc.update_instance_type(
+            name=active_instance_type, state="INSTANCE_TYPE_STATE_DEPRECATED",
+        )
+
+        dep_output, dep_rc = run_unchecked(
+            *_build_create_ci_args(cli, vm_template, default_subnet, active_instance_type),
+        )
+        assert dep_rc == 0, f"create with DEPRECATED type should succeed, got: {dep_output}"
+        assert "deprecat" in dep_output.lower() or "warning" in dep_output.lower(), (
+            f"Expected deprecation warning in output, got: {dep_output}"
+        )
+
+        uuid_match: re.Match[str] | None = re.search(r"'([^']+)'", dep_output)
+        assert uuid_match is not None, f"Failed to parse UUID from CLI output: {dep_output}"
+        deprecated_ci_uuid = uuid_match.group(1)
+        deprecated_ci_name = wait_for_cr(k8s=k8s_hub_client, uuid=deprecated_ci_uuid)
+    finally:
+        if deprecated_ci_uuid is not None:
+            cli.delete_compute_instance(uuid=deprecated_ci_uuid)
+            if deprecated_ci_name is not None:
+                wait_for_deletion(k8s=k8s_hub_client, name=deprecated_ci_name)
+
+
+def test_compute_instance_nonexistent_instance_type(
+    cli: OsacCLI,
+    k8s_hub_client: K8sClient,
+    default_subnet: str,
+    vm_template: str,
+) -> None:
+    missing_it_name = f"nonexistent-it-{uuid4().hex[:8]}"
+    ci_name = f"e2e-neg-{uuid4().hex[:8]}"
+    output, rc = run_unchecked(
+        *_build_create_ci_args(
+            cli, vm_template, default_subnet, missing_it_name, ci_name=ci_name,
+        ),
+    )
+    if rc == 0:
+        uuid_match = re.search(r"'([^']+)'", output)
+        if uuid_match:
+            ci_uuid = uuid_match.group(1)
+            cr_name = wait_for_cr(k8s=k8s_hub_client, uuid=ci_uuid)
+            cli.delete_compute_instance(uuid=ci_uuid)
+            wait_for_deletion(k8s=k8s_hub_client, name=cr_name)
+        elif k8s_hub_client.is_present(resource="computeinstance", name=ci_name):
+            pytest.fail(
+                f"ComputeInstance CR {ci_name} leaked but UUID could not be parsed from output: {output}"
+            )
+    assert rc != 0, f"create with nonexistent instance type should fail, got: {output}"
+    error_lower = output.lower()
+    assert any(term in error_lower for term in [
+        "not found", "404", "notfound",
+    ]), f"Expected not-found error, got: {output}"
+
+
+def test_compute_instance_obsolete_instance_type(
+    cli: OsacCLI,
+    private_grpc: GRPCClient,
+    k8s_hub_client: K8sClient,
+    default_subnet: str,
+    vm_template: str,
+    active_instance_type: str,
+) -> None:
+    private_grpc.update_instance_type(
+        name=active_instance_type, state="INSTANCE_TYPE_STATE_DEPRECATED",
+    )
+    private_grpc.update_instance_type(
+        name=active_instance_type, state="INSTANCE_TYPE_STATE_OBSOLETE",
+    )
+
+    ci_name = f"e2e-obs-{uuid4().hex[:8]}"
+    output, rc = run_unchecked(
+        *_build_create_ci_args(
+            cli, vm_template, default_subnet, active_instance_type, ci_name=ci_name,
+        ),
+    )
+    if rc == 0:
+        uuid_match = re.search(r"'([^']+)'", output)
+        if uuid_match:
+            ci_uuid = uuid_match.group(1)
+            cr_name = wait_for_cr(k8s=k8s_hub_client, uuid=ci_uuid)
+            cli.delete_compute_instance(uuid=ci_uuid)
+            wait_for_deletion(k8s=k8s_hub_client, name=cr_name)
+        elif k8s_hub_client.is_present(resource="computeinstance", name=ci_name):
+            pytest.fail(
+                f"ComputeInstance CR {ci_name} leaked but UUID could not be parsed from output: {output}"
+            )
+    assert rc != 0, f"create with OBSOLETE instance type should be rejected, got: {output}"
+    error_lower = output.lower()
+    assert any(term in error_lower for term in [
+        "obsolete", "rejected", "failedprecondition", "409", "conflict",
+    ]), f"Expected rejection error for obsolete instance type, got: {output}"