From db95b1d14e65c791e84d1f639d05f52e6a5ea562 Mon Sep 17 00:00:00 2001 From: rkishner Date: Mon, 27 Jul 2026 18:47:20 +0300 Subject: [PATCH] feat: Add process sub resource to VirtualMachineTemplate Signed-off-by: rkishner --- ocp_resources/virtual_machine_template.py | 35 ++++++ tests/test_virtual_machine_template.py | 147 ++++++++++++++++++++++ 2 files changed, 182 insertions(+) create mode 100644 tests/test_virtual_machine_template.py diff --git a/ocp_resources/virtual_machine_template.py b/ocp_resources/virtual_machine_template.py index 14078dbb7b..472b8aa963 100644 --- a/ocp_resources/virtual_machine_template.py +++ b/ocp_resources/virtual_machine_template.py @@ -64,3 +64,38 @@ def to_dict(self) -> None: _spec["parameters"] = self.parameters # End of generated code + + def process( + self, + parameters: dict[str, str] | None = None, + client: Any = None, + ) -> Any: + """ + Process the VirtualMachineTemplate using the KubeVirt subresources API. + + Sends a POST request to the process subresource endpoint, substituting the + provided parameter values into the template and returning the result. + + Args: + parameters (dict[str, str] | None): Key-value pairs of template parameters + to substitute, e.g. ``{"NAME": "my-vm", "INSTANCETYPE": "u1.large"}``. + Defaults to an empty dict (no substitutions). + client: Optional Kubernetes API client. Defaults to ``self.client``. + + Returns: + Any: The raw API server response object. + + """ + _client = client or self.client + _subresources_api = f"subresources.{self.api_version}" + body: dict[str, Any] = { + "apiVersion": _subresources_api, + "kind": "ProcessOptions", + "parameters": parameters or {}, + } + response = _client.request( + "POST", + f"/apis/{_subresources_api}/namespaces/{self.namespace}/virtualmachinetemplates/{self.name}/process", + body, + ) + return response diff --git a/tests/test_virtual_machine_template.py b/tests/test_virtual_machine_template.py new file mode 100644 index 0000000000..fa5fda8419 --- /dev/null +++ b/tests/test_virtual_machine_template.py @@ -0,0 +1,147 @@ +from typing import Any +from unittest.mock import MagicMock + +import pytest + +from ocp_resources.resource import get_client +from ocp_resources.virtual_machine_template import VirtualMachineTemplate + +# ── constants ──────────────────────────────────────────────────────────────── + +VMT_NAMESPACE = "test-ns" +VMT_NAME = "test-vmt" +_API_GROUP = "template.kubevirt.io" +_API_VERSION = "v1alpha1" +_FULL_API_VERSION = f"{_API_GROUP}/{_API_VERSION}" +# The process method builds: "subresources." +_SUBRESOURCES_API = f"subresources.{_FULL_API_VERSION}" +# Expected POST path used by process() +VMT_PROCESS_PATH = f"/apis/{_SUBRESOURCES_API}/namespaces/{VMT_NAMESPACE}/virtualmachinetemplates/{VMT_NAME}/process" + +# Resource definition required to register VirtualMachineTemplate in the fake client +_VMT_RESOURCE_DEF: dict[str, Any] = { + "kind": "VirtualMachineTemplate", + "api_version": _API_VERSION, + "group": _API_GROUP, + "version": _API_VERSION, + "group_version": _FULL_API_VERSION, + "plural": "virtualmachinetemplates", + "singular": "virtualmachinetemplate", + "namespaced": True, +} + + +# ── helpers ─────────────────────────────────────────────────────────────────── + + +def _make_fake_client(response_data: dict[str, Any] | None = None) -> Any: + """ + Return a fake client (get_client(fake=True)) with VirtualMachineTemplate registered + and request() replaced by a MagicMock so call args and return values are inspectable. + """ + client = get_client(fake=True) + client.register_resources(_VMT_RESOURCE_DEF) + mock_response = MagicMock() + mock_response.to_dict.return_value = response_data if response_data is not None else {} + # Intentionally replacing the stub method with a MagicMock for call-arg inspection + client.request = MagicMock(return_value=mock_response) # type: ignore[method-assign] + return client + + +# ── fixtures ────────────────────────────────────────────────────────────────── + + +@pytest.fixture() +def vmt() -> VirtualMachineTemplate: + return VirtualMachineTemplate( + client=_make_fake_client(), + name=VMT_NAME, + namespace=VMT_NAMESPACE, + virtual_machine={"spec": {}}, + ) + + +# ── tests ───────────────────────────────────────────────────────────────────── + + +class TestProcessPath: + """process() sends a POST to the correct subresource URL.""" + + def test_process_default_parameters_uses_correct_path(self, vmt: VirtualMachineTemplate) -> None: + process_client = _make_fake_client() + vmt.process(client=process_client) + + method, path, _ = process_client.request.call_args.args + assert method == "POST" + assert path == VMT_PROCESS_PATH + + def test_process_with_parameters_uses_correct_path(self, vmt: VirtualMachineTemplate) -> None: + process_client = _make_fake_client() + vmt.process(parameters={"NAME": "my-vm"}, client=process_client) + + method, path, _ = process_client.request.call_args.args + assert method == "POST" + assert path == VMT_PROCESS_PATH + + +class TestProcessOptionsBody: + """process() sends the correct ProcessOptions body.""" + + def test_body_kind_is_process_options(self, vmt: VirtualMachineTemplate) -> None: + process_client = _make_fake_client() + vmt.process(client=process_client) + + _, _, body = process_client.request.call_args.args + assert body["kind"] == "ProcessOptions" + + def test_body_api_version_matches_subresources_api(self, vmt: VirtualMachineTemplate) -> None: + process_client = _make_fake_client() + vmt.process(client=process_client) + + _, _, body = process_client.request.call_args.args + assert body["apiVersion"] == _SUBRESOURCES_API + + def test_body_parameters_empty_when_none_supplied(self, vmt: VirtualMachineTemplate) -> None: + """parameters=None should produce an empty dict in the request body.""" + process_client = _make_fake_client() + vmt.process(client=process_client) + + _, _, body = process_client.request.call_args.args + assert body["parameters"] == {} + + def test_body_parameters_passed_through_when_supplied(self, vmt: VirtualMachineTemplate) -> None: + """Supplied parameters dict should appear unchanged in the request body.""" + supplied = {"NAME": "my-vm", "INSTANCETYPE": "u1.large"} + process_client = _make_fake_client() + vmt.process(parameters=supplied, client=process_client) + + _, _, body = process_client.request.call_args.args + assert body["parameters"] == supplied + + +class TestProcessClientSelection: + """process() uses the injected client, falling back to self.client.""" + + def test_process_uses_injected_client(self, vmt: VirtualMachineTemplate) -> None: + injected = _make_fake_client() + vmt.process(client=injected) + + injected.request.assert_called_once() + vmt.client.request.assert_not_called() + + def test_process_falls_back_to_self_client_when_not_injected(self, vmt: VirtualMachineTemplate) -> None: + vmt.client.request.return_value.to_dict.return_value = {} + vmt.process() + + vmt.client.request.assert_called_once() + + +class TestProcessReturnValue: + """process() returns the raw response object from the API client.""" + + def test_process_returns_raw_response(self, vmt: VirtualMachineTemplate) -> None: + process_client = _make_fake_client() + + result = vmt.process(client=process_client) + + assert result is process_client.request.return_value