diff --git a/config/trigger_k8s_example.yaml b/config/trigger_k8s_example.yaml new file mode 100644 index 00000000..81d6a883 --- /dev/null +++ b/config/trigger_k8s_example.yaml @@ -0,0 +1,17 @@ +kraken: + chaos_scenarios: + - pod_disruption_scenarios: + - scenarios/kube/pod.yml + +triggers: + mode: all_of + timeout: 120 + interval: 5 + on_timeout: fail + conditions: + - type: k8s + apiVersion: apps/v1 + kind: Deployment + name: nginx + namespace: default + condition: "status.readyReplicas >= 1" diff --git a/containers/krknctl-input.json b/containers/krknctl-input.json index d54b2a51..5ec02647 100644 --- a/containers/krknctl-input.json +++ b/containers/krknctl-input.json @@ -823,5 +823,55 @@ "default": "", "required": "false", "group": "triggers" + }, + { + "name": "trigger-k8s-api-version", + "short_description": "K8s trigger API version", + "description": "Kubernetes API version for the resource to watch (e.g. apps/v1, kubevirt.io/v1). Required when using a k8s trigger condition.", + "variable": "TRIGGER_K8S_API_VERSION", + "type": "string", + "default": "", + "required": "false", + "group": "triggers" + }, + { + "name": "trigger-k8s-kind", + "short_description": "K8s trigger resource kind", + "description": "Kubernetes resource kind to watch (e.g. Deployment, Pod, VirtualMachineInstanceMigration)", + "variable": "TRIGGER_K8S_KIND", + "type": "string", + "default": "", + "required": "false", + "group": "triggers" + }, + { + "name": "trigger-k8s-namespace", + "short_description": "K8s trigger namespace", + "description": "Namespace of the resource to watch. Leave empty for cluster-scoped resources like Nodes.", + "variable": "TRIGGER_K8S_NAMESPACE", + "type": "string", + "default": "", + "required": "false", + "group": "triggers" + }, + { + "name": "trigger-k8s-name", + "short_description": "K8s trigger resource name", + "description": "Name of the specific Kubernetes resource to watch", + "variable": "TRIGGER_K8S_NAME", + "type": "string", + "default": "", + "required": "false", + "group": "triggers" + }, + { + "name": "trigger-k8s-condition", + "short_description": "K8s trigger condition", + "description": "Condition expression to evaluate against the resource (e.g. status.phase == Running, status.readyReplicas >= 1)", + "variable": "TRIGGER_K8S_CONDITION", + "type": "string", + "default": "", + "required": "false", + "group": "triggers" } ] \ No newline at end of file diff --git a/krkn/scenario_plugins/triggers/__init__.py b/krkn/scenario_plugins/triggers/__init__.py index 913c3f69..af3d5d02 100644 --- a/krkn/scenario_plugins/triggers/__init__.py +++ b/krkn/scenario_plugins/triggers/__init__.py @@ -14,6 +14,7 @@ from krkn.scenario_plugins.triggers.abstract_trigger import AbstractTrigger from krkn.scenario_plugins.triggers.command_trigger import CommandTrigger from krkn.scenario_plugins.triggers.http_trigger import HttpTrigger +from krkn.scenario_plugins.triggers.k8s_trigger import K8sTrigger from krkn.scenario_plugins.triggers.trigger_manager import TriggerManager -__all__ = ["AbstractTrigger", "CommandTrigger", "HttpTrigger", "TriggerManager"] +__all__ = ["AbstractTrigger", "CommandTrigger", "HttpTrigger", "K8sTrigger", "TriggerManager"] diff --git a/krkn/scenario_plugins/triggers/k8s_trigger.py b/krkn/scenario_plugins/triggers/k8s_trigger.py new file mode 100644 index 00000000..62ec879c --- /dev/null +++ b/krkn/scenario_plugins/triggers/k8s_trigger.py @@ -0,0 +1,240 @@ +# Copyright 2025 The Krkn Authors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +import logging +import re + +from kubernetes.dynamic import DynamicClient +from kubernetes.dynamic.exceptions import ( + NotFoundError, + ResourceNotFoundError, +) + +from krkn.scenario_plugins.triggers.abstract_trigger import AbstractTrigger + +OPERATORS = ("==", "!=", ">=", "<=", ">", "<") +OPERATOR_RE = re.compile(r"\s*(==|!=|>=|<=|>|<)\s*") + + +def _resolve_path(obj, path: str): + """Walk a dot-separated path on a dict/ResourceInstance. + + Returns the value at the path, or raises KeyError / IndexError + if any segment is missing. + """ + current = obj + for segment in path.split("."): + if isinstance(current, dict): + current = current[segment] + elif isinstance(current, (list, tuple)): + current = current[int(segment)] + else: + current = getattr(current, segment) + return current + + +def _coerce(value: str): + """Coerce a string value to int, float, bool, None, or leave as str.""" + if value.lower() == "true": + return True + if value.lower() == "false": + return False + if value.lower() == "none" or value.lower() == "null": + return None + try: + return int(value) + except ValueError: + pass + try: + return float(value) + except ValueError: + pass + return value + + +def _compare(actual, operator: str, expected): + """Compare actual value against expected using the given operator.""" + try: + actual_num = float(actual) + expected_num = float(expected) + is_numeric = True + except (TypeError, ValueError): + is_numeric = False + + if operator == "==": + if is_numeric: + return actual_num == expected_num + return str(actual) == str(expected) + if operator == "!=": + if is_numeric: + return actual_num != expected_num + return str(actual) != str(expected) + + if not is_numeric: + raise ValueError( + f"cannot compare non-numeric values with '{operator}': " + f"actual={actual!r}, expected={expected!r}" + ) + if operator == ">=": + return actual_num >= expected_num + if operator == "<=": + return actual_num <= expected_num + if operator == ">": + return actual_num > expected_num + if operator == "<": + return actual_num < expected_num + raise ValueError(f"unsupported operator: '{operator}'") + + +def _parse_condition(condition: str) -> tuple: + """Parse 'field.path == value' into (path, operator, expected_value).""" + match = OPERATOR_RE.search(condition) + if not match: + raise ValueError( + f"condition must contain an operator " + f"({', '.join(OPERATORS)}): got {condition!r}" + ) + operator = match.group(1) + path = condition[: match.start()].strip() + raw_value = condition[match.end() :].strip() + if not path: + raise ValueError(f"condition is missing a field path: {condition!r}") + if not raw_value: + raise ValueError(f"condition is missing an expected value: {condition!r}") + return path, operator, _coerce(raw_value) + + +class K8sTrigger(AbstractTrigger): + """Trigger that waits for a Kubernetes resource to match a condition. + + Kind-agnostic: works with built-in resources and CRDs through the + same code path using the Kubernetes dynamic client. + """ + + def __init__(self, trigger_config: dict, kubecli): + if kubecli is None: + raise ValueError("k8s trigger requires a kubecli instance") + + self._api_version = trigger_config.get("apiVersion") + if not self._api_version: + raise ValueError("k8s trigger requires 'apiVersion'") + + self._kind = trigger_config.get("kind") + if not self._kind: + raise ValueError("k8s trigger requires 'kind'") + + self._name = trigger_config.get("name") + if not self._name: + raise ValueError("k8s trigger requires 'name'") + + raw_condition = trigger_config.get("condition") + if not raw_condition: + raise ValueError("k8s trigger requires 'condition'") + + self._namespace = trigger_config.get("namespace") + self._path, self._operator, self._expected = _parse_condition( + raw_condition + ) + self._raw_condition = raw_condition + + self._kubecli = kubecli + self._last_result: bool | None = None + + def _get_client(self) -> DynamicClient: + return self._kubecli.dyn_client + + def evaluate(self) -> bool: + try: + dyn = self._get_client() + resource_api = dyn.resources.get( + api_version=self._api_version, kind=self._kind + ) + + if resource_api.namespaced and not self._namespace: + raise ValueError( + f"{self._api_version}/{self._kind} is namespaced " + f"but no namespace was specified" + ) + + if self._namespace: + resource = resource_api.get( + name=self._name, namespace=self._namespace + ) + else: + resource = resource_api.get(name=self._name) + + actual = _resolve_path(resource, self._path) + met = _compare(actual, self._operator, self._expected) + + logging.debug( + "k8s trigger: %s/%s %s.%s=%r (condition: %s) -> %s", + self._kind, + self._name, + self._path, + self._operator, + actual, + self._raw_condition, + met, + ) + except NotFoundError: + logging.debug( + "k8s trigger: resource %s/%s not found yet", + self._kind, + self._name, + ) + met = False + except ResourceNotFoundError: + logging.error( + "k8s trigger: API resource %s %s not registered on cluster", + self._api_version, + self._kind, + ) + met = False + except (KeyError, IndexError, AttributeError, ValueError) as e: + logging.debug( + "k8s trigger: field path '%s' not present on %s/%s: %s", + self._path, + self._kind, + self._name, + e, + ) + met = False + except Exception as e: + logging.error("k8s trigger unexpected error: %s", e) + met = False + + if met != self._last_result: + if met: + logging.info( + "trigger condition satisfied: %s/%s %s", + self._kind, + self._name, + self._raw_condition, + ) + else: + logging.info( + "trigger condition not satisfied: %s/%s %s", + self._kind, + self._name, + self._raw_condition, + ) + self._last_result = met + return met + + def describe(self) -> str: + ns_part = f" namespace={self._namespace}" if self._namespace else "" + return ( + f"k8s trigger: {self._api_version}/{self._kind} " + f"'{self._name}'{ns_part} " + f"(condition: {self._raw_condition})" + ) diff --git a/krkn/scenario_plugins/triggers/trigger_manager.py b/krkn/scenario_plugins/triggers/trigger_manager.py index 29052d55..adb09825 100644 --- a/krkn/scenario_plugins/triggers/trigger_manager.py +++ b/krkn/scenario_plugins/triggers/trigger_manager.py @@ -17,6 +17,7 @@ import time from krkn.scenario_plugins.triggers.abstract_trigger import AbstractTrigger from krkn.scenario_plugins.triggers.command_trigger import CommandTrigger from krkn.scenario_plugins.triggers.http_trigger import HttpTrigger +from krkn.scenario_plugins.triggers.k8s_trigger import K8sTrigger VALID_MODES = {"all_of", "any_of"} VALID_ON_TIMEOUT = {"skip", "fail", "run_anyway"} @@ -30,7 +31,7 @@ DEFAULT_ON_TIMEOUT = "skip" class TriggerManager: """Orchestrates polling across multiple triggers.""" - def __init__(self, trigger_config: dict): + def __init__(self, trigger_config: dict, kubecli=None): conditions = trigger_config.get("conditions") if not conditions: raise ValueError( @@ -74,9 +75,13 @@ class TriggerManager: if self._interval <= 0: raise ValueError(f"interval must be positive, got {self._interval}") + self._kubecli = kubecli + self._triggers: list[AbstractTrigger] = [] for condition in trigger_config["conditions"]: - self._triggers.append(self._build_trigger(condition)) + self._triggers.append( + self._build_trigger(condition, kubecli=kubecli) + ) # Track per-trigger satisfaction state for get_status self._trigger_states: list[bool | None] = [None] * len(self._triggers) @@ -86,7 +91,9 @@ class TriggerManager: return self._on_timeout @staticmethod - def _build_trigger(condition_config: dict) -> AbstractTrigger: + def _build_trigger( + condition_config: dict, kubecli=None + ) -> AbstractTrigger: """Factory method that creates a trigger from a condition config.""" trigger_type = condition_config.get("type") if not trigger_type: @@ -98,6 +105,9 @@ class TriggerManager: if trigger_type == "http": return HttpTrigger(condition_config) + if trigger_type == "k8s": + return K8sTrigger(condition_config, kubecli=kubecli) + raise ValueError(f"unknown trigger type: '{trigger_type}'") def wait_for_triggers(self) -> bool: diff --git a/run_kraken.py b/run_kraken.py index b78491f9..901fec6c 100644 --- a/run_kraken.py +++ b/run_kraken.py @@ -455,7 +455,7 @@ def main(options, command: Optional[str], out: Optional[dict] = None) -> int: trigger_config = config.get("triggers") if trigger_config: try: - trigger_manager = TriggerManager(trigger_config) + trigger_manager = TriggerManager(trigger_config, kubecli=kubecli) logging.info( "waiting for triggers before starting chaos:\n%s", trigger_manager.describe(), diff --git a/tests/test_triggers/test_k8s_trigger.py b/tests/test_triggers/test_k8s_trigger.py new file mode 100644 index 00000000..eeb27533 --- /dev/null +++ b/tests/test_triggers/test_k8s_trigger.py @@ -0,0 +1,498 @@ +#!/usr/bin/env python3 + +""" +Test suite for K8sTrigger class + +Usage: + python -m coverage run -a -m unittest tests/test_triggers/test_k8s_trigger.py -v + +Assisted By: Claude Code +""" + +import unittest +from unittest.mock import MagicMock, patch + +from krkn.scenario_plugins.triggers.k8s_trigger import ( + K8sTrigger, + _coerce, + _compare, + _parse_condition, + _resolve_path, +) +from kubernetes.dynamic.exceptions import ( + NotFoundError, + ResourceNotFoundError, +) + + +class TestResolvePath(unittest.TestCase): + """Tests for the _resolve_path helper.""" + + def test_simple_dict_path(self): + obj = {"status": {"phase": "Running"}} + self.assertEqual(_resolve_path(obj, "status.phase"), "Running") + + def test_nested_dict(self): + obj = {"spec": {"replicas": 3}} + self.assertEqual(_resolve_path(obj, "spec.replicas"), 3) + + def test_list_index(self): + obj = {"items": ["a", "b", "c"]} + self.assertEqual(_resolve_path(obj, "items.1"), "b") + + def test_missing_key_raises(self): + obj = {"status": {}} + with self.assertRaises(KeyError): + _resolve_path(obj, "status.phase") + + def test_attribute_access(self): + inner = MagicMock() + inner.phase = "Succeeded" + obj = {"status": inner} + self.assertEqual(_resolve_path(obj, "status.phase"), "Succeeded") + + def test_single_segment(self): + obj = {"phase": "Running"} + self.assertEqual(_resolve_path(obj, "phase"), "Running") + + +class TestCoerce(unittest.TestCase): + """Tests for the _coerce helper.""" + + def test_integer(self): + self.assertEqual(_coerce("42"), 42) + self.assertIsInstance(_coerce("42"), int) + + def test_float(self): + self.assertEqual(_coerce("3.14"), 3.14) + self.assertIsInstance(_coerce("3.14"), float) + + def test_true(self): + self.assertTrue(_coerce("true")) + self.assertTrue(_coerce("True")) + self.assertTrue(_coerce("TRUE")) + + def test_false(self): + self.assertFalse(_coerce("false")) + self.assertFalse(_coerce("False")) + + def test_none(self): + self.assertIsNone(_coerce("none")) + self.assertIsNone(_coerce("null")) + self.assertIsNone(_coerce("None")) + + def test_string(self): + self.assertEqual(_coerce("Running"), "Running") + + +class TestCompare(unittest.TestCase): + """Tests for the _compare helper.""" + + def test_eq_string(self): + self.assertTrue(_compare("Running", "==", "Running")) + self.assertFalse(_compare("Pending", "==", "Running")) + + def test_ne_string(self): + self.assertTrue(_compare("Pending", "!=", "Running")) + self.assertFalse(_compare("Running", "!=", "Running")) + + def test_eq_numeric_as_string(self): + self.assertTrue(_compare(3, "==", 3)) + self.assertTrue(_compare("3", "==", "3")) + + def test_eq_numeric_float_int(self): + """1.0 == 1 should be True via numeric comparison.""" + self.assertTrue(_compare(1.0, "==", 1)) + self.assertTrue(_compare("1.0", "==", "1")) + + def test_ne_numeric_float_int(self): + """1.0 != 1 should be False via numeric comparison.""" + self.assertFalse(_compare(1.0, "!=", 1)) + + def test_gte(self): + self.assertTrue(_compare(3, ">=", 3)) + self.assertTrue(_compare(4, ">=", 3)) + self.assertFalse(_compare(2, ">=", 3)) + + def test_lte(self): + self.assertTrue(_compare(3, "<=", 3)) + self.assertTrue(_compare(2, "<=", 3)) + self.assertFalse(_compare(4, "<=", 3)) + + def test_gt(self): + self.assertTrue(_compare(4, ">", 3)) + self.assertFalse(_compare(3, ">", 3)) + + def test_lt(self): + self.assertTrue(_compare(2, "<", 3)) + self.assertFalse(_compare(3, "<", 3)) + + def test_numeric_comparison_non_numeric_raises(self): + with self.assertRaises(ValueError): + _compare("abc", ">", 3) + + +class TestParseCondition(unittest.TestCase): + """Tests for the _parse_condition helper.""" + + def test_eq(self): + path, op, val = _parse_condition("status.phase == Running") + self.assertEqual(path, "status.phase") + self.assertEqual(op, "==") + self.assertEqual(val, "Running") + + def test_ne(self): + path, op, val = _parse_condition("status.phase != Pending") + self.assertEqual(path, "status.phase") + self.assertEqual(op, "!=") + self.assertEqual(val, "Pending") + + def test_gte_with_int(self): + path, op, val = _parse_condition("status.readyReplicas >= 1") + self.assertEqual(path, "status.readyReplicas") + self.assertEqual(op, ">=") + self.assertEqual(val, 1) + + def test_no_spaces(self): + path, op, val = _parse_condition("status.phase==Running") + self.assertEqual(path, "status.phase") + self.assertEqual(op, "==") + self.assertEqual(val, "Running") + + def test_missing_operator_raises(self): + with self.assertRaises(ValueError) as ctx: + _parse_condition("status.phase Running") + self.assertIn("operator", str(ctx.exception)) + + def test_missing_path_raises(self): + with self.assertRaises(ValueError) as ctx: + _parse_condition("== Running") + self.assertIn("field path", str(ctx.exception)) + + def test_missing_value_raises(self): + with self.assertRaises(ValueError) as ctx: + _parse_condition("status.phase ==") + self.assertIn("expected value", str(ctx.exception)) + + def test_boolean_value(self): + _, _, val = _parse_condition("status.ready == true") + self.assertIs(val, True) + + +class TestK8sTriggerInit(unittest.TestCase): + """Tests for K8sTrigger constructor validation.""" + + def _mock_kubecli(self): + kubecli = MagicMock() + kubecli.dyn_client = MagicMock() + return kubecli + + def test_missing_kubecli_raises(self): + with self.assertRaises(ValueError) as ctx: + K8sTrigger({ + "type": "k8s", + "apiVersion": "apps/v1", + "kind": "Deployment", + "name": "nginx", + "condition": "status.phase == Running", + }, kubecli=None) + self.assertIn("kubecli", str(ctx.exception)) + + def test_missing_api_version_raises(self): + with self.assertRaises(ValueError) as ctx: + K8sTrigger({ + "type": "k8s", + "kind": "Deployment", + "name": "nginx", + "condition": "status.phase == Running", + }, kubecli=self._mock_kubecli()) + self.assertIn("apiVersion", str(ctx.exception)) + + def test_missing_kind_raises(self): + with self.assertRaises(ValueError) as ctx: + K8sTrigger({ + "type": "k8s", + "apiVersion": "apps/v1", + "name": "nginx", + "condition": "status.phase == Running", + }, kubecli=self._mock_kubecli()) + self.assertIn("kind", str(ctx.exception)) + + def test_missing_name_raises(self): + with self.assertRaises(ValueError) as ctx: + K8sTrigger({ + "type": "k8s", + "apiVersion": "apps/v1", + "kind": "Deployment", + "condition": "status.phase == Running", + }, kubecli=self._mock_kubecli()) + self.assertIn("name", str(ctx.exception)) + + def test_missing_condition_raises(self): + with self.assertRaises(ValueError) as ctx: + K8sTrigger({ + "type": "k8s", + "apiVersion": "apps/v1", + "kind": "Deployment", + "name": "nginx", + }, kubecli=self._mock_kubecli()) + self.assertIn("condition", str(ctx.exception)) + + def test_valid_config(self): + trigger = K8sTrigger({ + "type": "k8s", + "apiVersion": "apps/v1", + "kind": "Deployment", + "name": "nginx", + "namespace": "default", + "condition": "status.readyReplicas >= 1", + }, kubecli=self._mock_kubecli()) + self.assertEqual(trigger._api_version, "apps/v1") + self.assertEqual(trigger._kind, "Deployment") + self.assertEqual(trigger._name, "nginx") + self.assertEqual(trigger._namespace, "default") + self.assertEqual(trigger._path, "status.readyReplicas") + self.assertEqual(trigger._operator, ">=") + self.assertEqual(trigger._expected, 1) + + def test_namespace_optional(self): + trigger = K8sTrigger({ + "type": "k8s", + "apiVersion": "v1", + "kind": "Node", + "name": "worker-1", + "condition": "status.phase == Ready", + }, kubecli=self._mock_kubecli()) + self.assertIsNone(trigger._namespace) + + +class TestK8sTriggerEvaluate(unittest.TestCase): + """Tests for K8sTrigger.evaluate() with mocked kubecli.""" + + def _make_kubecli(self): + """Create a mock kubecli with a mock dyn_client.""" + kubecli = MagicMock() + kubecli.dyn_client = MagicMock() + return kubecli + + def _make_trigger(self, kubecli=None, **overrides): + if kubecli is None: + kubecli = self._make_kubecli() + config = { + "type": "k8s", + "apiVersion": "apps/v1", + "kind": "Deployment", + "name": "nginx", + "namespace": "default", + "condition": "status.readyReplicas >= 1", + } + config.update(overrides) + return K8sTrigger(config, kubecli=kubecli), kubecli + + def test_condition_met(self): + """Resource matches condition -> returns True.""" + trigger, kubecli = self._make_trigger() + mock_api = MagicMock() + kubecli.dyn_client.resources.get.return_value = mock_api + mock_api.get.return_value = {"status": {"readyReplicas": 3}} + + self.assertTrue(trigger.evaluate()) + + kubecli.dyn_client.resources.get.assert_called_once_with( + api_version="apps/v1", kind="Deployment" + ) + mock_api.get.assert_called_once_with( + name="nginx", namespace="default" + ) + + def test_condition_not_met(self): + """Resource does not match condition -> returns False.""" + trigger, kubecli = self._make_trigger() + mock_api = MagicMock() + kubecli.dyn_client.resources.get.return_value = mock_api + mock_api.get.return_value = {"status": {"readyReplicas": 0}} + + self.assertFalse(trigger.evaluate()) + + def test_string_equality(self): + """String equality condition works.""" + trigger, kubecli = self._make_trigger( + condition="status.phase == Running", + kind="VirtualMachineInstanceMigration", + apiVersion="kubevirt.io/v1", + ) + mock_api = MagicMock() + kubecli.dyn_client.resources.get.return_value = mock_api + mock_api.get.return_value = {"status": {"phase": "Running"}} + + self.assertTrue(trigger.evaluate()) + + def test_resource_not_found(self): + """Resource does not exist yet -> returns False.""" + trigger, kubecli = self._make_trigger() + mock_api = MagicMock() + kubecli.dyn_client.resources.get.return_value = mock_api + mock_api.get.side_effect = NotFoundError(MagicMock(status=404)) + + self.assertFalse(trigger.evaluate()) + + def test_api_resource_not_registered(self): + """CRD not installed on cluster -> returns False.""" + trigger, kubecli = self._make_trigger( + apiVersion="kubevirt.io/v1", + kind="VirtualMachineInstanceMigration", + ) + kubecli.dyn_client.resources.get.side_effect = ResourceNotFoundError( + "Resource not found" + ) + + self.assertFalse(trigger.evaluate()) + + def test_field_path_missing(self): + """Field path doesn't exist on resource -> returns False.""" + trigger, kubecli = self._make_trigger() + mock_api = MagicMock() + kubecli.dyn_client.resources.get.return_value = mock_api + mock_api.get.return_value = {"status": {}} + + self.assertFalse(trigger.evaluate()) + + def test_unexpected_error(self): + """Unexpected API error -> returns False, no crash.""" + trigger, kubecli = self._make_trigger() + kubecli.dyn_client.resources.get.side_effect = ConnectionError("refused") + + self.assertFalse(trigger.evaluate()) + + def test_namespaced_resource_without_namespace(self): + """Namespaced resource with no namespace configured -> returns False.""" + kubecli = self._make_kubecli() + mock_api = MagicMock() + mock_api.namespaced = True + kubecli.dyn_client.resources.get.return_value = mock_api + + trigger = K8sTrigger({ + "type": "k8s", + "apiVersion": "apps/v1", + "kind": "Deployment", + "name": "nginx", + "condition": "status.readyReplicas >= 1", + }, kubecli=kubecli) + self.assertFalse(trigger.evaluate()) + mock_api.get.assert_not_called() + + def test_cluster_scoped_resource(self): + """No namespace -> calls get() without namespace kwarg.""" + kubecli = self._make_kubecli() + mock_api = MagicMock() + mock_api.namespaced = False + kubecli.dyn_client.resources.get.return_value = mock_api + mock_api.get.return_value = {"status": {"phase": "Ready"}} + + trigger = K8sTrigger({ + "type": "k8s", + "apiVersion": "v1", + "kind": "Node", + "name": "worker-1", + "condition": "status.phase == Ready", + }, kubecli=kubecli) + trigger.evaluate() + + mock_api.get.assert_called_once_with(name="worker-1") + + def test_crd_same_code_path(self): + """CRD uses the exact same code path as built-in resources.""" + kubecli = self._make_kubecli() + mock_api = MagicMock() + kubecli.dyn_client.resources.get.return_value = mock_api + mock_api.get.return_value = {"status": {"phase": "Running"}} + + trigger = K8sTrigger({ + "type": "k8s", + "apiVersion": "kubevirt.io/v1", + "kind": "VirtualMachineInstanceMigration", + "name": "test-migration", + "namespace": "vm-ns", + "condition": "status.phase == Running", + }, kubecli=kubecli) + self.assertTrue(trigger.evaluate()) + + kubecli.dyn_client.resources.get.assert_called_once_with( + api_version="kubevirt.io/v1", + kind="VirtualMachineInstanceMigration", + ) + + def test_value_error_from_compare(self): + """ValueError from _compare (non-numeric > operator) -> returns False.""" + trigger, kubecli = self._make_trigger( + condition="status.phase > 1", + ) + mock_api = MagicMock() + kubecli.dyn_client.resources.get.return_value = mock_api + mock_api.get.return_value = {"status": {"phase": "Running"}} + + self.assertFalse(trigger.evaluate()) + + +class TestK8sTriggerDescribe(unittest.TestCase): + """Tests for K8sTrigger.describe().""" + + def _mock_kubecli(self): + kubecli = MagicMock() + kubecli.dyn_client = MagicMock() + return kubecli + + def test_describe_with_namespace(self): + trigger = K8sTrigger({ + "type": "k8s", + "apiVersion": "apps/v1", + "kind": "Deployment", + "name": "nginx", + "namespace": "default", + "condition": "status.readyReplicas >= 1", + }, kubecli=self._mock_kubecli()) + desc = trigger.describe() + self.assertIn("apps/v1", desc) + self.assertIn("Deployment", desc) + self.assertIn("nginx", desc) + self.assertIn("default", desc) + self.assertIn("readyReplicas >= 1", desc) + + def test_describe_without_namespace(self): + trigger = K8sTrigger({ + "type": "k8s", + "apiVersion": "v1", + "kind": "Node", + "name": "worker-1", + "condition": "status.phase == Ready", + }, kubecli=self._mock_kubecli()) + desc = trigger.describe() + self.assertIn("Node", desc) + self.assertIn("worker-1", desc) + self.assertNotIn("namespace", desc) + + +class TestK8sTriggerClientInit(unittest.TestCase): + """Tests for K8sTrigger._get_client() delegation to kubecli.""" + + def test_get_client_returns_kubecli_dyn_client(self): + """_get_client() returns kubecli.dyn_client.""" + kubecli = MagicMock() + mock_dyn = MagicMock() + kubecli.dyn_client = mock_dyn + + trigger = K8sTrigger({ + "type": "k8s", + "apiVersion": "v1", + "kind": "Pod", + "name": "test", + "namespace": "default", + "condition": "status.phase == Running", + }, kubecli=kubecli) + + result = trigger._get_client() + self.assertIs(result, mock_dyn) + + +if __name__ == "__main__": + unittest.main()