mirror of
https://github.com/krkn-chaos/krkn.git
synced 2026-08-25 09:27:36 +00:00
feat: add k8s trigger type for event-driven chaos (#1515)
* feat: add k8s trigger type for event-driven chaos Closes #1496 Signed-off-by: Saurabh Wattamwar <swattamw@redhat.com> * fix: address review comments on k8s trigger - Validate namespace against resource_api.namespaced to fail fast when a namespaced resource is missing the namespace config - Use numeric equality first in _compare for == and != operators so that 1.0 == 1 works consistently with >= and <= - Fix krknctl-input.json description for apiVersion field Signed-off-by: Saurabh Wattamwar <swattamw@redhat.com> * feat: add optional context field to k8s trigger Allows selecting a specific kubeconfig context for cross-cluster triggers. If omitted, uses the default context as before. Signed-off-by: Saurabh Wattamwar <swattamw@redhat.com> * fix: use kubecli.dyn_client instead of raw DynamicClient and catch ValueError - K8sTrigger now receives kubecli from TriggerManager and uses kubecli.dyn_client for centralized config and proxy support - TriggerManager passes kubecli through _build_trigger to K8sTrigger - run_kraken.py passes kubecli to TriggerManager - Added ValueError to the except clause in evaluate() to handle non-numeric comparisons and namespace validation errors gracefully Signed-off-by: Saurabh Wattamwar <swattamw@redhat.com> * docs: add k8s trigger example config Signed-off-by: Saurabh Wattamwar <swattamw@redhat.com> * fix: make kubecli required and remove unused context field from k8s trigger Address remaining review comments on PR #1515: - Make kubecli a required argument with early validation instead of silently failing when None - Remove _context field since _get_client() delegates to kubecli.dyn_client which is already initialized with the correct context - Remove TRIGGER_K8S_CONTEXT from krknctl-input.json - Update tests to pass mock kubecli and cover missing-kubecli validation Signed-off-by: Saurabh Wattamwar <swattamw@redhat.com> --------- Signed-off-by: Saurabh Wattamwar <swattamw@redhat.com>
This commit is contained in:
Saurabh Wattamwar
@@ -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"
|
||||
@@ -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"
|
||||
}
|
||||
]
|
||||
@@ -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"]
|
||||
|
||||
@@ -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})"
|
||||
)
|
||||
@@ -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:
|
||||
|
||||
+1
-1
@@ -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(),
|
||||
|
||||
@@ -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()
|
||||
Reference in New Issue
Block a user