diff --git a/pyproject.toml b/pyproject.toml index 9f643636..f4217a05 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -28,6 +28,7 @@ dependencies = [ "python-dotenv>=1.0.0", "pulp-tool @ git+https://github.com/konflux-ci/pulp-tool.git", "jq", + "kubernetes>=28.0", ] [tool.black] diff --git a/src/helpers/internal_request/internal_request.py b/src/helpers/internal_request/internal_request.py index 36380d2f..65dad907 100644 --- a/src/helpers/internal_request/internal_request.py +++ b/src/helpers/internal_request/internal_request.py @@ -1,7 +1,8 @@ """Create and wait for InternalRequest resources in Kubernetes. This module creates an InternalRequest resource in a Kubernetes cluster using -`kubectl`. Parameters are passed as Python mappings rather than CLI flags. +the Kubernetes Python client. Parameters are passed as Python mappings rather +than CLI flags. Sync and async behavior @@ -86,7 +87,9 @@ Prerequisites ------------- -* `kubectl` must be installed and configured to communicate with the cluster. +* The ``kubernetes`` Python package must be installed. +* When running in-cluster, a service account with permissions to manage + ``InternalRequest`` resources must be mounted. Note: @@ -99,17 +102,17 @@ from __future__ import annotations import json -import logging import re -import subprocess import time from collections.abc import Mapping, Sequence from pathlib import Path from typing import Any +from kubernetes import client as k8s_client +from kubernetes import config as k8s_config + from release_service_utils.helpers import retry from release_service_utils.helpers.logger import logger -from release_service_utils.helpers.subprocess_cmd import run_cmd PIPELINE_NAME_LABEL = "internal-services.appstudio.openshift.io/pipeline-name" PIPELINERUN_UID_LABEL = "internal-services.appstudio.openshift.io/pipelinerun-uid" @@ -120,6 +123,11 @@ EXIT_TIMEOUT = 124 _DURATION_RE = re.compile(r"^(\d+)h(\d+)m(\d+)s$") +_IR_GROUP = "appstudio.redhat.com" +_IR_VERSION = "v1alpha1" +_IR_PLURAL = "internalrequests" +_NAMESPACE_FILE = Path("/var/run/secrets/kubernetes.io/serviceaccount/namespace") + class InternalRequestWaitError(RuntimeError): """Raised when waiting for InternalRequests fails or times out.""" @@ -134,6 +142,29 @@ class _InternalRequestNotComplete(Exception): """Raised when InternalRequests have not yet reached a terminal state.""" +def _default_k8s_api() -> k8s_client.CustomObjectsApi: + """Create a Kubernetes CustomObjects API client with auto-detected config. + + Two config loaders are tried in order: + - load_incluster_config(): for code running inside a Kubernetes pod. + - load_kube_config(): fallback for local development, testing and CI. + """ + try: + k8s_config.load_incluster_config() + except k8s_config.ConfigException: + k8s_config.load_kube_config() + return k8s_client.CustomObjectsApi() + + +def _get_namespace() -> str: + """Return the Kubernetes namespace for InternalRequest operations.""" + try: + return _NAMESPACE_FILE.read_text().strip() + except FileNotFoundError: + _, context = k8s_config.list_kube_config_contexts() + return context.get("context", {}).get("namespace", "default") + + def duration_to_seconds(duration: str) -> int: """Convert an `XhYmZs` duration string to seconds.""" match = _DURATION_RE.fullmatch(duration) @@ -231,20 +262,6 @@ def build_payload( return payload -def _log_subprocess_output( - result: subprocess.CompletedProcess[str], - *, - label: str, - stdout_level: int = logging.DEBUG, - stderr_level: int = logging.INFO, -) -> None: - """Log captured stdout/stderr from a subprocess result.""" - if result.stdout and result.stdout.strip(): - logger.log(stdout_level, "%s stdout: %s", label, result.stdout.strip()) - if result.stderr and result.stderr.strip(): - logger.log(stderr_level, "%s stderr: %s", label, result.stderr.strip()) - - def _pipelinerun_uid_from_labels(labels: Mapping[str, str]) -> str: """Return the pipelinerun-uid label value when present.""" return labels.get(PIPELINERUN_UID_LABEL, "") @@ -254,30 +271,29 @@ def _fetch_internal_requests( *, name: str | None, label_selector: str | None, + k8s_api: k8s_client.CustomObjectsApi, ) -> list[dict[str, Any]]: """Return InternalRequest objects as a list of parsed JSON dicts.""" + namespace = _get_namespace() + if name is not None: - result = run_cmd( - ["kubectl", "get", "internalrequest", name, "-o", "json"], - check=True, + item = k8s_api.get_namespaced_custom_object( + group=_IR_GROUP, + version=_IR_VERSION, + namespace=namespace, + plural=_IR_PLURAL, + name=name, ) - item = json.loads(result.stdout) return [item] - result = run_cmd( - [ - "kubectl", - "get", - "internalrequest", - "-l", - label_selector or "", - "-o", - "json", - ], - check=True, + result = k8s_api.list_namespaced_custom_object( + group=_IR_GROUP, + version=_IR_VERSION, + namespace=namespace, + plural=_IR_PLURAL, + label_selector=label_selector or "", ) - data = json.loads(result.stdout) - items = data.get("items") + items = result.get("items") return items if isinstance(items, list) else [] @@ -309,6 +325,7 @@ def wait_for_completion( name: str | None = None, label_selector: str | None = None, timeout: int = 600, + k8s_api: k8s_client.CustomObjectsApi | None = None, ) -> None: """Block until InternalRequests complete or *timeout* seconds elapse. @@ -319,6 +336,9 @@ def wait_for_completion( InternalRequestWaitError: When an IR fails or the wait times out. """ + if k8s_api is None: + k8s_api = _default_k8s_api() + has_name = bool(name) has_labels = bool(label_selector) if has_name == has_labels: @@ -333,6 +353,7 @@ def _poll_once() -> None: internal_requests = _fetch_internal_requests( name=name, label_selector=label_selector, + k8s_api=k8s_api, ) logger.info( "Found %d InternalRequests matching the name or label", @@ -401,8 +422,12 @@ def cleanup_existing_requests( *, pipeline: str, labels: Mapping[str, str], + k8s_api: k8s_client.CustomObjectsApi | None = None, ) -> None: """Delete prior InternalRequests for the same pipeline run and pipeline name.""" + if k8s_api is None: + k8s_api = _default_k8s_api() + pipelinerun_uid = _pipelinerun_uid_from_labels(labels) if not pipelinerun_uid: return @@ -410,10 +435,12 @@ def cleanup_existing_requests( label_selector = ( f"{PIPELINERUN_UID_LABEL}={pipelinerun_uid}," f"{PIPELINE_NAME_LABEL}={pipeline}" ) - items = _fetch_internal_requests(name=None, label_selector=label_selector) + items = _fetch_internal_requests(name=None, label_selector=label_selector, k8s_api=k8s_api) if not items: return + namespace = _get_namespace() + logger.info("Found existing InternalRequests from prior attempts. Cleaning up...") for item in items: if not isinstance(item, dict): @@ -422,18 +449,14 @@ def cleanup_existing_requests( if not isinstance(ir_name, str) or not ir_name: continue logger.info("Deleting InternalRequest %s...", ir_name) - result = run_cmd( - [ - "kubectl", - "delete", - "internalrequest", - ir_name, - "--wait=true", - "--timeout=60s", - ], - check=True, + k8s_api.delete_namespaced_custom_object( + group=_IR_GROUP, + version=_IR_VERSION, + namespace=namespace, + plural=_IR_PLURAL, + name=ir_name, ) - _log_subprocess_output(result, label=f"kubectl delete {ir_name}") + logger.info("Deleted InternalRequest %s", ir_name) logger.info( "Cleanup complete. Waiting %ds for PipelineRun cancellation to propagate...", @@ -442,39 +465,50 @@ def cleanup_existing_requests( time.sleep(CLEANUP_PROPAGATION_SLEEP_SECONDS) -def create_internal_request(payload: dict[str, Any]) -> str: +def create_internal_request( + payload: dict[str, Any], + *, + k8s_api: k8s_client.CustomObjectsApi | None = None, +) -> str: """Create an InternalRequest from *payload* and return its name.""" - result = run_cmd( - ["kubectl", "create", "-f", "-", "-o", "json"], - stdin=json.dumps(payload), - check=True, + if k8s_api is None: + k8s_api = _default_k8s_api() + + namespace = _get_namespace() + resource = k8s_api.create_namespaced_custom_object( + group=_IR_GROUP, + version=_IR_VERSION, + namespace=namespace, + plural=_IR_PLURAL, + body=payload, ) - _log_subprocess_output(result, label="kubectl create") - resource = json.loads(result.stdout) name = resource.get("metadata", {}).get("name") if not isinstance(name, str) or not name: - msg = "kubectl create did not return an InternalRequest name" + msg = "API did not return an InternalRequest name" raise RuntimeError(msg) + logger.info("Created InternalRequest: %s", name) return name -def fetch_results(internal_request_name: str) -> dict[str, Any]: - """Read InternalRequest ``status.results`` via kubectl.""" - result = run_cmd( - [ - "kubectl", - "get", - "internalrequest", - internal_request_name, - "-o=jsonpath={.status.results}", - ], - check=True, +def fetch_results( + internal_request_name: str, + *, + k8s_api: k8s_client.CustomObjectsApi | None = None, +) -> dict[str, Any]: + """Read InternalRequest ``status.results`` from the API.""" + if k8s_api is None: + k8s_api = _default_k8s_api() + + namespace = _get_namespace() + resource = k8s_api.get_namespaced_custom_object( + group=_IR_GROUP, + version=_IR_VERSION, + namespace=namespace, + plural=_IR_PLURAL, + name=internal_request_name, ) - raw = (result.stdout or "").strip() - if not raw: - return {} - parsed = json.loads(raw) - return parsed if isinstance(parsed, dict) else {} + results = resource.get("status", {}).get("results") + return results if isinstance(results, dict) else {} def create( @@ -489,6 +523,7 @@ def create( task_timeout: str = "0h55m0s", finally_timeout: str = "0h5m0s", cleanup: bool = True, + k8s_api: k8s_client.CustomObjectsApi | None = None, ) -> str: """Create an InternalRequest and optionally wait for it to complete. @@ -499,6 +534,9 @@ def create( pipeline run. Set this when multiple InternalRequests are created concurrently with the same labels. """ + if k8s_api is None: + k8s_api = _default_k8s_api() + if not pipeline: msg = "pipeline is required" raise ValueError(msg) @@ -521,7 +559,7 @@ def create( finally_timeout=finally_timeout, ) if cleanup: - cleanup_existing_requests(pipeline=pipeline, labels=merged_labels) + cleanup_existing_requests(pipeline=pipeline, labels=merged_labels, k8s_api=k8s_api) payload = build_payload( pipeline=pipeline, @@ -534,11 +572,11 @@ def create( finally_timeout=finally_timeout, service_account=service_account, ) - internal_request_name = create_internal_request(payload) + internal_request_name = create_internal_request(payload, k8s_api=k8s_api) logger.info("InternalRequest '%s' created.", internal_request_name) if sync: logger.info("Sync flag set to true. Waiting for the InternalRequest to complete.") - wait_for_completion(name=internal_request_name, timeout=timeout) + wait_for_completion(name=internal_request_name, timeout=timeout, k8s_api=k8s_api) return internal_request_name diff --git a/src/helpers/internal_request/tests/test_internal_request.py b/src/helpers/internal_request/tests/test_internal_request.py index d078a88b..67e83c6b 100644 --- a/src/helpers/internal_request/tests/test_internal_request.py +++ b/src/helpers/internal_request/tests/test_internal_request.py @@ -1,10 +1,8 @@ -"""Test internal_request_results helpers.""" +"""Test internal_request helpers.""" from __future__ import annotations -import json from pathlib import Path -from typing import Any from unittest import mock import pytest @@ -19,6 +17,14 @@ ) +@pytest.fixture() +def k8s_api(): + """Provide a mock Kubernetes CustomObjects API client.""" + api = mock.MagicMock() + with mock.patch.object(ir_module, "_get_namespace", return_value="test-ns"): + yield api + + def test_duration_to_seconds_parses_hms() -> None: """Convert XhYmZs durations to seconds.""" assert ir_module.duration_to_seconds("1h0m0s") == 3600 @@ -96,163 +102,128 @@ def test_build_payload_includes_required_fields() -> None: ) -def _completed_process(stdout: str, returncode: int = 0, stderr: str = "") -> mock.MagicMock: - """Build a fake subprocess.CompletedProcess with the given outputs.""" - result = mock.MagicMock() - result.stdout = stdout - result.stderr = stderr - result.returncode = returncode - return result - - -def test_cleanup_existing_requests_deletes_matching_irs() -> None: +def test_cleanup_existing_requests_deletes_matching_irs(k8s_api: mock.MagicMock) -> None: """Delete existing InternalRequests before creating a new one.""" - calls: list[list[str]] = [] - - def fake_run_cmd(cmd: list[str], **kwargs: Any) -> mock.MagicMock: - calls.append(cmd) - if cmd[0:3] == ["kubectl", "get", "internalrequest"]: - body = {"items": [{"metadata": {"name": "old-ir-1"}}]} - return _completed_process(json.dumps(body)) - return _completed_process("") + k8s_api.list_namespaced_custom_object.return_value = { + "items": [{"metadata": {"name": "old-ir-1"}}], + } - with ( - mock.patch.object(ir_module, "run_cmd", side_effect=fake_run_cmd), - mock.patch.object(ir_module.time, "sleep"), - ): + with mock.patch.object(ir_module.time, "sleep"): ir_module.cleanup_existing_requests( pipeline="create-advisory", labels={ir_module.PIPELINERUN_UID_LABEL: "uid-123"}, + k8s_api=k8s_api, ) - assert calls[0] == [ - "kubectl", - "get", - "internalrequest", - "-l", - ( - f"{ir_module.PIPELINERUN_UID_LABEL}=uid-123," - f"{ir_module.PIPELINE_NAME_LABEL}=create-advisory" - ), - "-o", - "json", - ] - assert calls[1][:4] == ["kubectl", "delete", "internalrequest", "old-ir-1"] + list_call = k8s_api.list_namespaced_custom_object.call_args + assert list_call.kwargs["label_selector"] == ( + f"{ir_module.PIPELINERUN_UID_LABEL}=uid-123," + f"{ir_module.PIPELINE_NAME_LABEL}=create-advisory" + ) + + del_call = k8s_api.delete_namespaced_custom_object.call_args + assert del_call.kwargs["name"] == "old-ir-1" + assert del_call.kwargs["namespace"] == "test-ns" -def test_cleanup_existing_requests_skips_without_pipelinerun_uid() -> None: +def test_cleanup_existing_requests_skips_without_pipelinerun_uid( + k8s_api: mock.MagicMock, +) -> None: """Skip cleanup when the pipelinerun-uid label is absent.""" - with mock.patch.object(ir_module, "run_cmd") as fake_run_cmd: - ir_module.cleanup_existing_requests( - pipeline="create-advisory", - labels={"other": "value"}, - ) - fake_run_cmd.assert_not_called() + ir_module.cleanup_existing_requests( + pipeline="create-advisory", + labels={"other": "value"}, + k8s_api=k8s_api, + ) + k8s_api.list_namespaced_custom_object.assert_not_called() -def test_cleanup_existing_requests_skips_when_no_matching_items() -> None: +def test_cleanup_existing_requests_skips_when_no_matching_items( + k8s_api: mock.MagicMock, +) -> None: """Skip deletion when no existing InternalRequests are found.""" - with mock.patch.object(ir_module, "run_cmd") as fake_run_cmd: - fake_run_cmd.return_value = _completed_process(json.dumps({"items": []})) - ir_module.cleanup_existing_requests( - pipeline="create-advisory", - labels={ir_module.PIPELINERUN_UID_LABEL: "uid-123"}, - ) + k8s_api.list_namespaced_custom_object.return_value = {"items": []} + + ir_module.cleanup_existing_requests( + pipeline="create-advisory", + labels={ir_module.PIPELINERUN_UID_LABEL: "uid-123"}, + k8s_api=k8s_api, + ) - fake_run_cmd.assert_called_once() + k8s_api.list_namespaced_custom_object.assert_called_once() + k8s_api.delete_namespaced_custom_object.assert_not_called() -def test_cleanup_existing_requests_skips_non_dict_items() -> None: +def test_cleanup_existing_requests_skips_non_dict_items(k8s_api: mock.MagicMock) -> None: """Ignore list entries that are not InternalRequest objects.""" - calls: list[list[str]] = [] - - def fake_run_cmd(cmd: list[str], **kwargs: Any) -> mock.MagicMock: - calls.append(cmd) - if cmd[0:3] == ["kubectl", "get", "internalrequest"]: - body = {"items": ["not-a-dict", {"metadata": {"name": "old-ir-1"}}]} - return _completed_process(json.dumps(body)) - return _completed_process("") + k8s_api.list_namespaced_custom_object.return_value = { + "items": ["not-a-dict", {"metadata": {"name": "old-ir-1"}}], + } - with ( - mock.patch.object(ir_module, "run_cmd", side_effect=fake_run_cmd), - mock.patch.object(ir_module.time, "sleep"), - ): + with mock.patch.object(ir_module.time, "sleep"): ir_module.cleanup_existing_requests( pipeline="create-advisory", labels={ir_module.PIPELINERUN_UID_LABEL: "uid-123"}, + k8s_api=k8s_api, ) - assert calls[1][:4] == ["kubectl", "delete", "internalrequest", "old-ir-1"] + del_call = k8s_api.delete_namespaced_custom_object.call_args + assert del_call.kwargs["name"] == "old-ir-1" -def test_cleanup_existing_requests_skips_invalid_ir_name() -> None: +def test_cleanup_existing_requests_skips_invalid_ir_name(k8s_api: mock.MagicMock) -> None: """Ignore InternalRequests whose metadata name is missing or not a string.""" - calls: list[list[str]] = [] - - def fake_run_cmd(cmd: list[str], **kwargs: Any) -> mock.MagicMock: - calls.append(cmd) - if cmd[0:3] == ["kubectl", "get", "internalrequest"]: - body = { - "items": [ - {"metadata": {}}, - {"metadata": {"name": ""}}, - {"metadata": {"name": 123}}, - ], - } - return _completed_process(json.dumps(body)) - return _completed_process("") + k8s_api.list_namespaced_custom_object.return_value = { + "items": [ + {"metadata": {}}, + {"metadata": {"name": ""}}, + {"metadata": {"name": 123}}, + ], + } - with ( - mock.patch.object(ir_module, "run_cmd", side_effect=fake_run_cmd), - mock.patch.object(ir_module.time, "sleep"), - ): + with mock.patch.object(ir_module.time, "sleep"): ir_module.cleanup_existing_requests( pipeline="create-advisory", labels={ir_module.PIPELINERUN_UID_LABEL: "uid-123"}, + k8s_api=k8s_api, ) - assert len(calls) == 1 + k8s_api.delete_namespaced_custom_object.assert_not_called() -def test_create_creates_internal_request_without_waiting() -> None: +def test_create_creates_internal_request_without_waiting(k8s_api: mock.MagicMock) -> None: """Create an InternalRequest and return its name when sync is false.""" - calls: list[list[str]] = [] - - def fake_run_cmd(cmd: list[str], **kwargs: Any) -> mock.MagicMock: - calls.append(cmd) - if cmd[0:2] == ["kubectl", "create"]: - body = {"metadata": {"name": "create-advisory-abc"}} - return _completed_process(json.dumps(body)) - return _completed_process(json.dumps({"items": []})) + k8s_api.list_namespaced_custom_object.return_value = {"items": []} + k8s_api.create_namespaced_custom_object.return_value = { + "metadata": {"name": "create-advisory-abc"}, + } - with ( - mock.patch.object(ir_module, "run_cmd", side_effect=fake_run_cmd), - mock.patch.object(ir_module.time, "sleep"), - ): - name = ir_module.create( - "create-advisory", - params={ - "taskGitUrl": "https://example.test/catalog", - "taskGitRevision": "main", - }, - sync=False, - ) + name = ir_module.create( + "create-advisory", + params={ + "taskGitUrl": "https://example.test/catalog", + "taskGitRevision": "main", + }, + sync=False, + k8s_api=k8s_api, + ) assert name == "create-advisory-abc" - assert any(cmd[0:2] == ["kubectl", "create"] for cmd in calls) + k8s_api.create_namespaced_custom_object.assert_called_once() -def test_create_requires_task_git_params() -> None: +def test_create_requires_task_git_params(k8s_api: mock.MagicMock) -> None: """Reject creation when git resolver params are missing.""" with pytest.raises(ValueError, match="taskGitUrl and taskGitRevision"): ir_module.create( "create-advisory", params={"componentGroup": "myapp"}, sync=False, + k8s_api=k8s_api, ) -def test_create_requires_pipeline() -> None: +def test_create_requires_pipeline(k8s_api: mock.MagicMock) -> None: """Reject creation when the pipeline name is empty.""" with pytest.raises(ValueError, match="pipeline is required"): ir_module.create( @@ -262,23 +233,19 @@ def test_create_requires_pipeline() -> None: "taskGitRevision": "main", }, sync=False, + k8s_api=k8s_api, ) -def test_create_internal_request_raises_when_name_missing() -> None: - """Raise when kubectl create does not return an InternalRequest name.""" - with ( - mock.patch.object( - ir_module, - "run_cmd", - return_value=_completed_process(json.dumps({"metadata": {}})), - ), - pytest.raises(RuntimeError, match="did not return an InternalRequest name"), - ): - ir_module.create_internal_request({"kind": "InternalRequest"}) +def test_create_internal_request_raises_when_name_missing(k8s_api: mock.MagicMock) -> None: + """Raise when the API does not return an InternalRequest name.""" + k8s_api.create_namespaced_custom_object.return_value = {"metadata": {}} + + with pytest.raises(RuntimeError, match="did not return an InternalRequest name"): + ir_module.create_internal_request({"kind": "InternalRequest"}, k8s_api=k8s_api) -def test_create_waits_when_sync_is_true() -> None: +def test_create_waits_when_sync_is_true(k8s_api: mock.MagicMock) -> None: """Wait for completion after creating the InternalRequest.""" with ( mock.patch.object(ir_module, "cleanup_existing_requests"), @@ -296,13 +263,14 @@ def test_create_waits_when_sync_is_true() -> None: "taskGitRevision": "main", }, sync=True, + k8s_api=k8s_api, ) assert name == "create-advisory-abc" - wait.assert_called_once_with(name="create-advisory-abc", timeout=3600) + wait.assert_called_once_with(name="create-advisory-abc", timeout=3600, k8s_api=k8s_api) -def test_create_skips_cleanup_when_cleanup_is_false() -> None: +def test_create_skips_cleanup_when_cleanup_is_false(k8s_api: mock.MagicMock) -> None: """Do not delete prior InternalRequests when cleanup is False.""" with ( mock.patch.object(ir_module, "cleanup_existing_requests") as cleanup, @@ -321,18 +289,21 @@ def test_create_skips_cleanup_when_cleanup_is_false() -> None: }, sync=True, cleanup=False, + k8s_api=k8s_api, ) cleanup.assert_not_called() -def test_wait_for_completion_requires_exactly_one_selector() -> None: +def test_wait_for_completion_requires_exactly_one_selector( + k8s_api: mock.MagicMock, +) -> None: """Reject calls that provide both or neither selector.""" with pytest.raises(ValueError, match="exactly one"): - wait_for_completion() + wait_for_completion(k8s_api=k8s_api) with pytest.raises(ValueError, match="exactly one"): - wait_for_completion(name="ir-1", label_selector="foo=bar") + wait_for_completion(name="ir-1", label_selector="foo=bar", k8s_api=k8s_api) def _patch_ir_output_path(tmp_path: Path, ir_name: str = "ir-1") -> tuple[mock._patch, Path]: @@ -342,7 +313,9 @@ def _patch_ir_output_path(tmp_path: Path, ir_name: str = "ir-1") -> tuple[mock._ return patch, output_path -def test_wait_for_completion_handles_running_before_success(tmp_path: Path) -> None: +def test_wait_for_completion_handles_running_before_success( + tmp_path: Path, k8s_api: mock.MagicMock +) -> None: """Poll again when an InternalRequest is still running.""" running_body = { "metadata": {"name": "ir-1"}, @@ -358,19 +331,15 @@ def test_wait_for_completion_handles_running_before_success(tmp_path: Path) -> N "pipelineRun": "pr-1", }, } - responses = [running_body, succeeded_body] + k8s_api.get_namespaced_custom_object.side_effect = [running_body, succeeded_body] output_patch, output_path = _patch_ir_output_path(tmp_path) - def fake_run_cmd(cmd: list[str], **kwargs: Any) -> mock.MagicMock: - return _completed_process(json.dumps(responses.pop(0))) - with ( output_patch, - mock.patch.object(ir_module, "run_cmd", side_effect=fake_run_cmd), mock.patch.object(retry.retry.time, "sleep") as sleep, mock.patch.object(ir_module.time, "time", side_effect=[0, 1]), ): - wait_for_completion(name="ir-1", timeout=600) + wait_for_completion(name="ir-1", timeout=600, k8s_api=k8s_api) sleep.assert_called_once_with(5) assert output_path.read_text(encoding="utf-8") == ( @@ -378,7 +347,9 @@ def fake_run_cmd(cmd: list[str], **kwargs: Any) -> mock.MagicMock: ) -def test_wait_for_completion_writes_output_json_on_success(tmp_path: Path) -> None: +def test_wait_for_completion_writes_output_json_on_success( + tmp_path: Path, k8s_api: mock.MagicMock +) -> None: """Write name and pipelineRun to the IR output file on success.""" ir_body = { "metadata": {"name": "ir-1"}, @@ -387,24 +358,23 @@ def test_wait_for_completion_writes_output_json_on_success(tmp_path: Path) -> No "pipelineRun": "pr-1", }, } + k8s_api.get_namespaced_custom_object.return_value = ir_body output_patch, output_path = _patch_ir_output_path(tmp_path) - def fake_run_cmd(cmd: list[str], **kwargs: Any) -> mock.MagicMock: - return _completed_process(json.dumps(ir_body)) - with ( output_patch, - mock.patch.object(ir_module, "run_cmd", side_effect=fake_run_cmd), mock.patch.object(retry.retry.time, "sleep"), ): - wait_for_completion(name="ir-1", timeout=600) + wait_for_completion(name="ir-1", timeout=600, k8s_api=k8s_api) assert output_path.read_text(encoding="utf-8") == ( '{"name": "ir-1", "pipelineRun": "pr-1"}\n' ) -def test_wait_for_completion_raises_on_failure(tmp_path: Path) -> None: +def test_wait_for_completion_raises_on_failure( + tmp_path: Path, k8s_api: mock.MagicMock +) -> None: """Raise InternalRequestWaitError when an IR completes unsuccessfully.""" ir_body = { "metadata": {"name": "ir-1"}, @@ -413,18 +383,15 @@ def test_wait_for_completion_raises_on_failure(tmp_path: Path) -> None: "pipelineRun": "pr-1", }, } + k8s_api.get_namespaced_custom_object.return_value = ir_body output_patch, output_path = _patch_ir_output_path(tmp_path) - def fake_run_cmd(cmd: list[str], **kwargs: Any) -> mock.MagicMock: - return _completed_process(json.dumps(ir_body)) - with ( output_patch, - mock.patch.object(ir_module, "run_cmd", side_effect=fake_run_cmd), mock.patch.object(retry.retry.time, "sleep"), pytest.raises(InternalRequestWaitError) as exc_info, ): - wait_for_completion(name="ir-1", timeout=600) + wait_for_completion(name="ir-1", timeout=600, k8s_api=k8s_api) assert exc_info.value.exit_code == EXIT_FAILED assert output_path.read_text(encoding="utf-8") == ( @@ -432,109 +399,134 @@ def fake_run_cmd(cmd: list[str], **kwargs: Any) -> mock.MagicMock: ) -def test_wait_for_completion_raises_on_timeout() -> None: +def test_wait_for_completion_raises_on_timeout(k8s_api: mock.MagicMock) -> None: """Raise InternalRequestWaitError when the wait timeout elapses.""" ir_body = { "metadata": {"name": "ir-1"}, "status": {"conditions": []}, } - - def fake_run_cmd(cmd: list[str], **kwargs: Any) -> mock.MagicMock: - return _completed_process(json.dumps(ir_body)) + k8s_api.get_namespaced_custom_object.return_value = ir_body with ( - mock.patch.object(ir_module, "run_cmd", side_effect=fake_run_cmd), mock.patch.object(retry.retry.time, "sleep"), mock.patch.object(ir_module.time, "time", side_effect=[0, 601]), pytest.raises(InternalRequestWaitError) as exc_info, ): - wait_for_completion(name="ir-1", timeout=600) + wait_for_completion(name="ir-1", timeout=600, k8s_api=k8s_api) assert exc_info.value.exit_code == EXIT_TIMEOUT -def test_wait_for_completion_keeps_polling_when_label_selector_matches_nothing() -> None: +def test_wait_for_completion_keeps_polling_when_label_selector_matches_nothing( + k8s_api: mock.MagicMock, +) -> None: """Keep polling until timeout when a label selector matches no InternalRequests.""" - empty_list_body = {"items": []} - - def fake_run_cmd(cmd: list[str], **kwargs: Any) -> mock.MagicMock: - return _completed_process(json.dumps(empty_list_body)) + k8s_api.list_namespaced_custom_object.return_value = {"items": []} with ( - mock.patch.object(ir_module, "run_cmd", side_effect=fake_run_cmd), mock.patch.object(retry.retry.time, "sleep"), mock.patch.object(ir_module.time, "time", side_effect=[0, 601]), pytest.raises(InternalRequestWaitError) as exc_info, ): - wait_for_completion(label_selector="foo=bar", timeout=600) + wait_for_completion(label_selector="foo=bar", timeout=600, k8s_api=k8s_api) assert exc_info.value.exit_code == EXIT_TIMEOUT -def test_fetch_results_handles_empty_stdout() -> None: - """Return an empty dict when kubectl prints no results.""" - with mock.patch.object( - ir_module, - "run_cmd", - return_value=_completed_process(""), - ): - assert ir_module.fetch_results("ir-1") == {} +def test_fetch_results_returns_empty_when_no_results(k8s_api: mock.MagicMock) -> None: + """Return an empty dict when the InternalRequest has no results.""" + k8s_api.get_namespaced_custom_object.return_value = { + "metadata": {"name": "ir-1"}, + "status": {}, + } + assert ir_module.fetch_results("ir-1", k8s_api=k8s_api) == {} -def test_fetch_results_parses_json() -> None: - """Parse InternalRequest status.results JSON from kubectl.""" - payload = {"result": "Success", "advisory_url": "url"} - with mock.patch.object( - ir_module, - "run_cmd", - return_value=_completed_process(json.dumps(payload)), - ): - assert ir_module.fetch_results("ir-1") == payload +def test_fetch_results_parses_results(k8s_api: mock.MagicMock) -> None: + """Return the status.results dict from the InternalRequest.""" + results = {"result": "Success", "advisory_url": "url"} + k8s_api.get_namespaced_custom_object.return_value = { + "metadata": {"name": "ir-1"}, + "status": {"results": results}, + } + + assert ir_module.fetch_results("ir-1", k8s_api=k8s_api) == results -def test_fetch_results_ignores_non_dict_json() -> None: - """Return an empty dict when kubectl output is not a JSON object.""" - with mock.patch.object( - ir_module, - "run_cmd", - return_value=_completed_process('["not","dict"]'), - ): - assert ir_module.fetch_results("ir-1") == {} +def test_fetch_results_ignores_non_dict_results(k8s_api: mock.MagicMock) -> None: + """Return an empty dict when status.results is not a dict.""" + k8s_api.get_namespaced_custom_object.return_value = { + "metadata": {"name": "ir-1"}, + "status": {"results": ["not", "dict"]}, + } + + assert ir_module.fetch_results("ir-1", k8s_api=k8s_api) == {} -def test_log_subprocess_output_logs_stderr_and_stdout() -> None: - """Log both stdout and stderr when present.""" - result = _completed_process(stdout="created ok", stderr="some warning") - with mock.patch.object(ir_module.logger, "log") as log: - ir_module._log_subprocess_output(result, label="kubectl create") - assert log.call_count == 2 - stdout_call, stderr_call = log.call_args_list - assert "stdout" in stdout_call[0][1] - assert "stderr" in stderr_call[0][1] +def test_create_internal_request_logs_created_name(k8s_api: mock.MagicMock) -> None: + """Log the name of the created InternalRequest.""" + k8s_api.create_namespaced_custom_object.return_value = { + "metadata": {"name": "ir-abc"}, + } + with mock.patch.object(ir_module.logger, "info") as log_info: + name = ir_module.create_internal_request({"kind": "InternalRequest"}, k8s_api=k8s_api) -def test_log_subprocess_output_skips_empty() -> None: - """Do not log when stdout and stderr are empty.""" - result = _completed_process(stdout="", stderr="") - with mock.patch.object(ir_module.logger, "log") as log: - ir_module._log_subprocess_output(result, label="kubectl create") + assert name == "ir-abc" + logged = any("ir-abc" in str(call) for call in log_info.call_args_list) + assert logged - log.assert_not_called() +def test_get_namespace_reads_from_service_account_file(tmp_path: Path) -> None: + """Read namespace from the in-cluster service account file.""" + ns_file = tmp_path / "namespace" + ns_file.write_text("my-ns\n") + with mock.patch.object(ir_module, "_NAMESPACE_FILE", ns_file): + assert ir_module._get_namespace() == "my-ns" -def test_create_internal_request_logs_subprocess_output() -> None: - """Log kubectl stderr after creating an InternalRequest.""" - body = {"metadata": {"name": "ir-abc"}} - result = _completed_process( - stdout=json.dumps(body), stderr="Warning: resource version changed" - ) + +def test_get_namespace_falls_back_to_kubeconfig(tmp_path: Path) -> None: + """Fall back to kubeconfig context when the SA file is missing.""" + ns_file = tmp_path / "namespace" + context = {"context": {"namespace": "dev-ns"}} with ( - mock.patch.object(ir_module, "run_cmd", return_value=result), - mock.patch.object(ir_module.logger, "log") as log, + mock.patch.object(ir_module, "_NAMESPACE_FILE", ns_file), + mock.patch.object( + ir_module.k8s_config, "list_kube_config_contexts", return_value=([], context) + ), ): - name = ir_module.create_internal_request({"kind": "InternalRequest"}) + assert ir_module._get_namespace() == "dev-ns" - assert name == "ir-abc" - stderr_logged = any("stderr" in str(call) for call in log.call_args_list) - assert stderr_logged + +def test_default_k8s_api_loads_incluster_config() -> None: + """Load in-cluster config and return a CustomObjectsApi.""" + with ( + mock.patch.object(ir_module.k8s_config, "load_incluster_config") as incluster, + mock.patch.object(ir_module.k8s_client, "CustomObjectsApi") as api_cls, + ): + result = ir_module._default_k8s_api() + + incluster.assert_called_once() + api_cls.assert_called_once() + assert result is api_cls.return_value + + +def test_default_k8s_api_falls_back_to_kubeconfig() -> None: + """Fall back to kubeconfig when in-cluster config is unavailable.""" + from kubernetes.config import ConfigException + + with ( + mock.patch.object( + ir_module.k8s_config, + "load_incluster_config", + side_effect=ConfigException, + ), + mock.patch.object(ir_module.k8s_config, "load_kube_config") as kubeconfig, + mock.patch.object(ir_module.k8s_client, "CustomObjectsApi") as api_cls, + ): + result = ir_module._default_k8s_api() + + kubeconfig.assert_called_once() + api_cls.assert_called_once() + assert result is api_cls.return_value