mirror of
https://github.com/krkn-chaos/krkn.git
synced 2026-08-25 09:27:36 +00:00
feat: add event-driven chaos triggers with command trigger (Phase 1) (#1484)
* feat: add event-driven chaos triggers with command trigger (Phase 1) Adds a pluggable trigger system that gates chaos injection on user-defined preconditions. A top-level `triggers` block in config.yaml is evaluated before the chaos loop, with configurable polling mode (all_of/any_of), timeout, interval, and on_timeout behavior (skip/fail/run_anyway). Phase 1 implements the command trigger type which runs a shell command and checks its exit code. - AbstractTrigger base class for pluggable trigger types - CommandTrigger with input validation (expected_rc int coercion) - TriggerManager with positive-number validation for timeout/interval - Trigger evaluation runs before health check plugins start - Broad exception handling in evaluate() prevents UnboundLocalError - Debug logging of command returncode/stderr for observability - 33 unit tests covering all functionality and edge cases Closes #1483 Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> Signed-off-by: ddjain <darjain@redhat.com> * fix: address review feedback on trigger validation - Validate conditions is a list, not a string - Expose on_timeout as a property, use it in run_kraken.py instead of re-reading raw config - Validate expected_rc is in 0-255 Unix range - Add 4 new tests (37 trigger tests total) Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> Signed-off-by: ddjain <darjain@redhat.com> --------- Signed-off-by: ddjain <darjain@redhat.com> Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Darshan Jain
co-authored by
Claude Opus 4.6
Claude Opus 4.6
parent
99870a2fc7
commit
f62d2e1c36
@@ -0,0 +1,18 @@
|
||||
# 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.
|
||||
from krkn.scenario_plugins.triggers.abstract_trigger import AbstractTrigger
|
||||
from krkn.scenario_plugins.triggers.command_trigger import CommandTrigger
|
||||
from krkn.scenario_plugins.triggers.trigger_manager import TriggerManager
|
||||
|
||||
__all__ = ["AbstractTrigger", "CommandTrigger", "TriggerManager"]
|
||||
@@ -0,0 +1,28 @@
|
||||
# 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.
|
||||
from abc import ABC, abstractmethod
|
||||
|
||||
|
||||
class AbstractTrigger(ABC):
|
||||
"""Base class for all trigger types."""
|
||||
|
||||
@abstractmethod
|
||||
def evaluate(self) -> bool:
|
||||
"""Returns True if the trigger condition is met."""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def describe(self) -> str:
|
||||
"""Human-readable description for logging."""
|
||||
pass
|
||||
@@ -0,0 +1,95 @@
|
||||
# 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 subprocess
|
||||
|
||||
from krkn.scenario_plugins.triggers.abstract_trigger import AbstractTrigger
|
||||
|
||||
COMMAND_TIMEOUT_SECONDS = 30
|
||||
|
||||
|
||||
class CommandTrigger(AbstractTrigger):
|
||||
"""Trigger that runs a shell command and checks its exit code."""
|
||||
|
||||
def __init__(self, config: dict):
|
||||
self._cmd = config.get("cmd") or config.get("inline")
|
||||
if not self._cmd:
|
||||
raise ValueError(
|
||||
"command trigger requires either 'cmd' or 'inline' field"
|
||||
)
|
||||
try:
|
||||
self._expected_rc = int(config.get("expected_rc", 0))
|
||||
except (TypeError, ValueError):
|
||||
raise ValueError(
|
||||
f"expected_rc must be an integer, got {config.get('expected_rc')!r}"
|
||||
)
|
||||
if not 0 <= self._expected_rc <= 255:
|
||||
raise ValueError(
|
||||
f"expected_rc must be between 0 and 255, got {self._expected_rc}"
|
||||
)
|
||||
self._last_result: bool | None = None
|
||||
|
||||
def evaluate(self) -> bool:
|
||||
try:
|
||||
result = subprocess.run(
|
||||
self._cmd,
|
||||
shell=True,
|
||||
timeout=COMMAND_TIMEOUT_SECONDS,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
)
|
||||
met = result.returncode == self._expected_rc
|
||||
logging.debug(
|
||||
f"command trigger: rc={result.returncode} "
|
||||
f"expected={self._expected_rc} cmd='{self._cmd}'"
|
||||
)
|
||||
if result.stderr:
|
||||
logging.debug(
|
||||
f"command trigger stderr: {result.stderr.strip()}"
|
||||
)
|
||||
except subprocess.TimeoutExpired:
|
||||
logging.warning(
|
||||
f"command trigger timed out after {COMMAND_TIMEOUT_SECONDS}s: "
|
||||
f"{self._cmd}"
|
||||
)
|
||||
met = False
|
||||
except FileNotFoundError:
|
||||
logging.error(
|
||||
f"command trigger binary not found: {self._cmd}"
|
||||
)
|
||||
met = False
|
||||
except Exception as e:
|
||||
logging.error(
|
||||
f"command trigger unexpected error: {e}: {self._cmd}"
|
||||
)
|
||||
met = False
|
||||
|
||||
# Log only on state change
|
||||
if met != self._last_result:
|
||||
if met:
|
||||
logging.info(
|
||||
f"trigger condition satisfied: {self._cmd}"
|
||||
)
|
||||
else:
|
||||
logging.info(
|
||||
f"trigger condition not satisfied: {self._cmd}"
|
||||
)
|
||||
self._last_result = met
|
||||
return met
|
||||
|
||||
def describe(self) -> str:
|
||||
return (
|
||||
f"command trigger: '{self._cmd}' "
|
||||
f"(expected rc={self._expected_rc})"
|
||||
)
|
||||
@@ -0,0 +1,169 @@
|
||||
# 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 time
|
||||
|
||||
from krkn.scenario_plugins.triggers.abstract_trigger import AbstractTrigger
|
||||
from krkn.scenario_plugins.triggers.command_trigger import CommandTrigger
|
||||
|
||||
VALID_MODES = {"all_of", "any_of"}
|
||||
VALID_ON_TIMEOUT = {"skip", "fail", "run_anyway"}
|
||||
|
||||
DEFAULT_TIMEOUT = 300
|
||||
DEFAULT_INTERVAL = 5
|
||||
DEFAULT_MODE = "all_of"
|
||||
DEFAULT_ON_TIMEOUT = "skip"
|
||||
|
||||
|
||||
class TriggerManager:
|
||||
"""Orchestrates polling across multiple triggers."""
|
||||
|
||||
def __init__(self, trigger_config: dict):
|
||||
conditions = trigger_config.get("conditions")
|
||||
if not conditions:
|
||||
raise ValueError(
|
||||
"trigger config must include a non-empty 'conditions' list"
|
||||
)
|
||||
if not isinstance(conditions, list):
|
||||
raise ValueError(
|
||||
"trigger 'conditions' must be a list, "
|
||||
f"got {type(conditions).__name__}"
|
||||
)
|
||||
|
||||
self._mode = trigger_config.get("mode", DEFAULT_MODE)
|
||||
if self._mode not in VALID_MODES:
|
||||
raise ValueError(
|
||||
f"invalid trigger mode '{self._mode}', "
|
||||
f"must be one of: {', '.join(sorted(VALID_MODES))}"
|
||||
)
|
||||
|
||||
self._on_timeout = trigger_config.get("on_timeout", DEFAULT_ON_TIMEOUT)
|
||||
if self._on_timeout not in VALID_ON_TIMEOUT:
|
||||
raise ValueError(
|
||||
f"invalid on_timeout '{self._on_timeout}', "
|
||||
f"must be one of: {', '.join(sorted(VALID_ON_TIMEOUT))}"
|
||||
)
|
||||
|
||||
self._timeout = trigger_config.get("timeout", DEFAULT_TIMEOUT)
|
||||
self._interval = trigger_config.get("interval", DEFAULT_INTERVAL)
|
||||
|
||||
try:
|
||||
self._timeout = float(self._timeout)
|
||||
self._interval = float(self._interval)
|
||||
except (TypeError, ValueError):
|
||||
raise ValueError(
|
||||
f"timeout and interval must be numeric, "
|
||||
f"got timeout={trigger_config.get('timeout')!r}, "
|
||||
f"interval={trigger_config.get('interval')!r}"
|
||||
)
|
||||
|
||||
if self._timeout <= 0:
|
||||
raise ValueError(
|
||||
f"timeout must be positive, got {self._timeout}"
|
||||
)
|
||||
if self._interval <= 0:
|
||||
raise ValueError(
|
||||
f"interval must be positive, got {self._interval}"
|
||||
)
|
||||
|
||||
self._triggers: list[AbstractTrigger] = []
|
||||
for condition in trigger_config["conditions"]:
|
||||
self._triggers.append(self._build_trigger(condition))
|
||||
|
||||
# Track per-trigger satisfaction state for get_status
|
||||
self._trigger_states: list[bool | None] = [None] * len(self._triggers)
|
||||
|
||||
@property
|
||||
def on_timeout(self) -> str:
|
||||
return self._on_timeout
|
||||
|
||||
@staticmethod
|
||||
def _build_trigger(condition_config: dict) -> AbstractTrigger:
|
||||
"""Factory method that creates a trigger from a condition config."""
|
||||
trigger_type = condition_config.get("type")
|
||||
if not trigger_type:
|
||||
raise ValueError("each condition must have a 'type' field")
|
||||
|
||||
if trigger_type == "command":
|
||||
return CommandTrigger(condition_config)
|
||||
|
||||
raise ValueError(f"unknown trigger type: '{trigger_type}'")
|
||||
|
||||
def wait_for_triggers(self) -> bool:
|
||||
"""Polls triggers until conditions are met or timeout expires.
|
||||
|
||||
Returns True if conditions were met, False if timed out.
|
||||
"""
|
||||
logging.info(
|
||||
f"waiting for triggers: mode={self._mode}, "
|
||||
f"timeout={self._timeout}s, interval={self._interval}s, "
|
||||
f"on_timeout={self._on_timeout}"
|
||||
)
|
||||
deadline = time.monotonic() + self._timeout
|
||||
|
||||
while time.monotonic() < deadline:
|
||||
results = []
|
||||
for i, trigger in enumerate(self._triggers):
|
||||
result = trigger.evaluate()
|
||||
self._trigger_states[i] = result
|
||||
results.append(result)
|
||||
|
||||
logging.debug(
|
||||
f"trigger poll: {[r for r in results]}"
|
||||
)
|
||||
|
||||
if self._mode == "all_of" and all(results):
|
||||
logging.info("all trigger conditions satisfied")
|
||||
return True
|
||||
elif self._mode == "any_of" and any(results):
|
||||
logging.info("at least one trigger condition satisfied")
|
||||
return True
|
||||
|
||||
remaining = deadline - time.monotonic()
|
||||
if remaining <= 0:
|
||||
break
|
||||
time.sleep(min(self._interval, remaining))
|
||||
|
||||
logging.warning(
|
||||
f"triggers timed out after {self._timeout}s "
|
||||
f"(on_timeout={self._on_timeout})"
|
||||
)
|
||||
return False
|
||||
|
||||
def describe(self) -> str:
|
||||
"""Human-readable summary of all triggers."""
|
||||
parts = [
|
||||
f"TriggerManager(mode={self._mode}, "
|
||||
f"timeout={self._timeout}s, interval={self._interval}s, "
|
||||
f"on_timeout={self._on_timeout})",
|
||||
]
|
||||
for i, trigger in enumerate(self._triggers):
|
||||
parts.append(f" [{i}] {trigger.describe()}")
|
||||
return "\n".join(parts)
|
||||
|
||||
def get_status(self) -> dict:
|
||||
"""Returns current state of each trigger for the signal server."""
|
||||
return {
|
||||
"mode": self._mode,
|
||||
"timeout": self._timeout,
|
||||
"interval": self._interval,
|
||||
"on_timeout": self._on_timeout,
|
||||
"triggers": [
|
||||
{
|
||||
"description": trigger.describe(),
|
||||
"satisfied": self._trigger_states[i],
|
||||
}
|
||||
for i, trigger in enumerate(self._triggers)
|
||||
],
|
||||
}
|
||||
@@ -66,6 +66,7 @@ from krkn.rollback.command import (
|
||||
list_rollback as list_rollback_command,
|
||||
execute_rollback as execute_rollback_command,
|
||||
)
|
||||
from krkn.scenario_plugins.triggers.trigger_manager import TriggerManager
|
||||
|
||||
# removes TripleDES warning
|
||||
import warnings
|
||||
@@ -412,6 +413,36 @@ def main(options, command: Optional[str]) -> int:
|
||||
logging.error(f"⛔ Class: {class_name} Module: {module_name}")
|
||||
logging.error(f"⚠️ {error}\n")
|
||||
|
||||
# Evaluate top-level triggers before starting health checks or chaos
|
||||
trigger_config = config.get("triggers")
|
||||
if trigger_config:
|
||||
try:
|
||||
trigger_manager = TriggerManager(trigger_config)
|
||||
logging.info(
|
||||
"waiting for triggers before starting chaos:\n%s",
|
||||
trigger_manager.describe(),
|
||||
)
|
||||
triggered = trigger_manager.wait_for_triggers()
|
||||
if not triggered:
|
||||
on_timeout = trigger_manager.on_timeout
|
||||
if on_timeout == "skip":
|
||||
logging.warning(
|
||||
"trigger timed out, skipping all scenarios"
|
||||
)
|
||||
chaos_scenarios = []
|
||||
elif on_timeout == "fail":
|
||||
logging.error(
|
||||
"trigger timed out, exiting with failure"
|
||||
)
|
||||
return 1
|
||||
else:
|
||||
logging.warning(
|
||||
"trigger timed out, running scenarios anyway"
|
||||
)
|
||||
except ValueError as e:
|
||||
logging.error("invalid trigger configuration: %s", e)
|
||||
return 1
|
||||
|
||||
# Start all health check plugins discovered via config_key_map.
|
||||
# Returns list of (plugin, worker_thread, telemetry_queue);
|
||||
# worker_thread is None for self-threading plugins (e.g. virt).
|
||||
|
||||
@@ -0,0 +1,177 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
"""
|
||||
Test suite for CommandTrigger class
|
||||
|
||||
Usage:
|
||||
python -m coverage run -a -m unittest tests/test_triggers/test_command_trigger.py -v
|
||||
|
||||
Assisted By: Claude Code
|
||||
"""
|
||||
|
||||
import subprocess
|
||||
import unittest
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from krkn.scenario_plugins.triggers.command_trigger import (
|
||||
COMMAND_TIMEOUT_SECONDS,
|
||||
CommandTrigger,
|
||||
)
|
||||
|
||||
|
||||
class TestCommandTrigger(unittest.TestCase):
|
||||
|
||||
def _make_trigger(self, **overrides):
|
||||
"""Build a CommandTrigger with sensible defaults."""
|
||||
config = {"cmd": "echo hello", "expected_rc": 0}
|
||||
config.update(overrides)
|
||||
return CommandTrigger(config)
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# evaluate() tests
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
@patch("krkn.scenario_plugins.triggers.command_trigger.subprocess.run")
|
||||
def test_evaluate_success(self, mock_run):
|
||||
"""Command exits 0, expected_rc=0 -> returns True."""
|
||||
mock_run.return_value = MagicMock(returncode=0)
|
||||
trigger = self._make_trigger(cmd="echo ok", expected_rc=0)
|
||||
|
||||
self.assertTrue(trigger.evaluate())
|
||||
mock_run.assert_called_once_with(
|
||||
"echo ok",
|
||||
shell=True,
|
||||
timeout=COMMAND_TIMEOUT_SECONDS,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
)
|
||||
|
||||
@patch("krkn.scenario_plugins.triggers.command_trigger.subprocess.run")
|
||||
def test_evaluate_failure(self, mock_run):
|
||||
"""Command exits 1, expected_rc=0 -> returns False."""
|
||||
mock_run.return_value = MagicMock(returncode=1)
|
||||
trigger = self._make_trigger(cmd="false", expected_rc=0)
|
||||
|
||||
self.assertFalse(trigger.evaluate())
|
||||
|
||||
@patch("krkn.scenario_plugins.triggers.command_trigger.subprocess.run")
|
||||
def test_evaluate_custom_expected_rc(self, mock_run):
|
||||
"""Command exits 42, expected_rc=42 -> returns True."""
|
||||
mock_run.return_value = MagicMock(returncode=42)
|
||||
trigger = self._make_trigger(cmd="exit 42", expected_rc=42)
|
||||
|
||||
self.assertTrue(trigger.evaluate())
|
||||
|
||||
@patch("krkn.scenario_plugins.triggers.command_trigger.subprocess.run")
|
||||
def test_evaluate_inline_command(self, mock_run):
|
||||
"""Uses 'inline' field instead of 'cmd'."""
|
||||
mock_run.return_value = MagicMock(returncode=0)
|
||||
trigger = CommandTrigger({"inline": "date"})
|
||||
|
||||
self.assertTrue(trigger.evaluate())
|
||||
mock_run.assert_called_once_with(
|
||||
"date",
|
||||
shell=True,
|
||||
timeout=COMMAND_TIMEOUT_SECONDS,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
)
|
||||
|
||||
@patch("krkn.scenario_plugins.triggers.command_trigger.subprocess.run")
|
||||
def test_cmd_takes_precedence_over_inline(self, mock_run):
|
||||
"""Both 'cmd' and 'inline' provided; 'cmd' is used."""
|
||||
mock_run.return_value = MagicMock(returncode=0)
|
||||
trigger = CommandTrigger({"cmd": "use-this", "inline": "not-this"})
|
||||
|
||||
trigger.evaluate()
|
||||
args, _ = mock_run.call_args
|
||||
self.assertEqual(args[0], "use-this")
|
||||
|
||||
@patch("krkn.scenario_plugins.triggers.command_trigger.subprocess.run")
|
||||
def test_evaluate_timeout(self, mock_run):
|
||||
"""Command hangs, subprocess times out -> returns False."""
|
||||
mock_run.side_effect = subprocess.TimeoutExpired(
|
||||
cmd="sleep 999", timeout=COMMAND_TIMEOUT_SECONDS
|
||||
)
|
||||
trigger = self._make_trigger(cmd="sleep 999")
|
||||
|
||||
self.assertFalse(trigger.evaluate())
|
||||
|
||||
@patch("krkn.scenario_plugins.triggers.command_trigger.subprocess.run")
|
||||
def test_evaluate_command_not_found(self, mock_run):
|
||||
"""Binary does not exist -> returns False."""
|
||||
mock_run.side_effect = FileNotFoundError("No such file")
|
||||
trigger = self._make_trigger(cmd="/nonexistent/binary")
|
||||
|
||||
self.assertFalse(trigger.evaluate())
|
||||
|
||||
@patch("krkn.scenario_plugins.triggers.command_trigger.subprocess.run")
|
||||
def test_evaluate_unexpected_exception(self, mock_run):
|
||||
"""Unexpected exception (e.g. PermissionError) -> returns False, no crash."""
|
||||
mock_run.side_effect = PermissionError("Permission denied")
|
||||
trigger = self._make_trigger(cmd="/restricted/script.sh")
|
||||
|
||||
self.assertFalse(trigger.evaluate())
|
||||
|
||||
@patch("krkn.scenario_plugins.triggers.command_trigger.subprocess.run")
|
||||
def test_evaluate_default_expected_rc(self, mock_run):
|
||||
"""No expected_rc in config -> defaults to 0."""
|
||||
mock_run.return_value = MagicMock(returncode=0)
|
||||
trigger = CommandTrigger({"cmd": "true"})
|
||||
|
||||
self.assertTrue(trigger.evaluate())
|
||||
self.assertEqual(trigger._expected_rc, 0)
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# describe() tests
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def test_describe(self):
|
||||
"""Returns a meaningful description string."""
|
||||
trigger = self._make_trigger(cmd="check_service.sh", expected_rc=0)
|
||||
description = trigger.describe()
|
||||
|
||||
self.assertIn("check_service.sh", description)
|
||||
self.assertIn("0", description)
|
||||
self.assertIsInstance(description, str)
|
||||
self.assertTrue(len(description) > 0)
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Validation tests
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def test_missing_cmd_and_inline(self):
|
||||
"""Neither 'cmd' nor 'inline' provided -> raises ValueError."""
|
||||
with self.assertRaises(ValueError) as ctx:
|
||||
CommandTrigger({})
|
||||
|
||||
self.assertIn("cmd", str(ctx.exception).lower())
|
||||
self.assertIn("inline", str(ctx.exception).lower())
|
||||
|
||||
def test_expected_rc_string_coerced_to_int(self):
|
||||
"""expected_rc='0' (string from envsubst) -> coerced to int 0."""
|
||||
trigger = CommandTrigger({"cmd": "true", "expected_rc": "0"})
|
||||
self.assertEqual(trigger._expected_rc, 0)
|
||||
self.assertIsInstance(trigger._expected_rc, int)
|
||||
|
||||
def test_expected_rc_invalid_raises(self):
|
||||
"""expected_rc='abc' -> raises ValueError."""
|
||||
with self.assertRaises(ValueError) as ctx:
|
||||
CommandTrigger({"cmd": "true", "expected_rc": "abc"})
|
||||
self.assertIn("expected_rc", str(ctx.exception))
|
||||
|
||||
def test_expected_rc_above_255_raises(self):
|
||||
"""expected_rc=999 -> raises ValueError (Unix rc range is 0-255)."""
|
||||
with self.assertRaises(ValueError) as ctx:
|
||||
CommandTrigger({"cmd": "true", "expected_rc": 999})
|
||||
self.assertIn("255", str(ctx.exception))
|
||||
|
||||
def test_expected_rc_negative_raises(self):
|
||||
"""expected_rc=-1 -> raises ValueError."""
|
||||
with self.assertRaises(ValueError) as ctx:
|
||||
CommandTrigger({"cmd": "true", "expected_rc": -1})
|
||||
self.assertIn("expected_rc", str(ctx.exception))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,370 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
"""
|
||||
Test suite for TriggerManager class
|
||||
|
||||
Usage:
|
||||
python -m coverage run -a -m unittest tests/test_triggers/test_trigger_manager.py -v
|
||||
|
||||
Assisted By: Claude Code
|
||||
"""
|
||||
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
from krkn.scenario_plugins.triggers.abstract_trigger import AbstractTrigger
|
||||
from krkn.scenario_plugins.triggers.trigger_manager import (
|
||||
DEFAULT_INTERVAL,
|
||||
DEFAULT_MODE,
|
||||
DEFAULT_ON_TIMEOUT,
|
||||
DEFAULT_TIMEOUT,
|
||||
TriggerManager,
|
||||
)
|
||||
|
||||
|
||||
class StubTrigger(AbstractTrigger):
|
||||
"""Concrete trigger for testing that returns a preconfigured value."""
|
||||
|
||||
def __init__(self, result: bool, name: str = "stub"):
|
||||
self._result = result
|
||||
self._name = name
|
||||
|
||||
def evaluate(self) -> bool:
|
||||
return self._result
|
||||
|
||||
def describe(self) -> str:
|
||||
return f"stub trigger '{self._name}' (result={self._result})"
|
||||
|
||||
def set_result(self, result: bool):
|
||||
self._result = result
|
||||
|
||||
|
||||
def _make_config(**overrides):
|
||||
"""Build a minimal valid trigger config dict with overrides."""
|
||||
config = {
|
||||
"mode": "all_of",
|
||||
"timeout": 10,
|
||||
"interval": 1,
|
||||
"on_timeout": "skip",
|
||||
"conditions": [
|
||||
{"type": "command", "cmd": "true"},
|
||||
],
|
||||
}
|
||||
config.update(overrides)
|
||||
return config
|
||||
|
||||
|
||||
class TestTriggerManager(unittest.TestCase):
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# wait_for_triggers() — mode tests
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
@patch.object(TriggerManager, "_build_trigger")
|
||||
def test_all_of_all_pass(self, mock_build):
|
||||
"""mode=all_of, all triggers pass -> returns True."""
|
||||
t1 = StubTrigger(True, "t1")
|
||||
t2 = StubTrigger(True, "t2")
|
||||
mock_build.side_effect = [t1, t2]
|
||||
|
||||
config = _make_config(
|
||||
mode="all_of",
|
||||
conditions=[
|
||||
{"type": "command", "cmd": "a"},
|
||||
{"type": "command", "cmd": "b"},
|
||||
],
|
||||
)
|
||||
manager = TriggerManager(config)
|
||||
self.assertTrue(manager.wait_for_triggers())
|
||||
|
||||
@patch("krkn.scenario_plugins.triggers.trigger_manager.time")
|
||||
@patch.object(TriggerManager, "_build_trigger")
|
||||
def test_all_of_one_fails(self, mock_build, mock_time):
|
||||
"""mode=all_of, one trigger stays False -> loops until timeout, returns False."""
|
||||
t1 = StubTrigger(True, "t1")
|
||||
t2 = StubTrigger(False, "t2")
|
||||
mock_build.side_effect = [t1, t2]
|
||||
|
||||
# Simulate: first call returns start time, subsequent calls advance past deadline
|
||||
call_count = 0
|
||||
|
||||
def advancing_monotonic():
|
||||
nonlocal call_count
|
||||
call_count += 1
|
||||
if call_count <= 2:
|
||||
return 0.0
|
||||
return 999.0 # past deadline
|
||||
|
||||
mock_time.monotonic.side_effect = advancing_monotonic
|
||||
mock_time.sleep = lambda x: None
|
||||
|
||||
config = _make_config(
|
||||
mode="all_of",
|
||||
timeout=10,
|
||||
conditions=[
|
||||
{"type": "command", "cmd": "a"},
|
||||
{"type": "command", "cmd": "b"},
|
||||
],
|
||||
)
|
||||
manager = TriggerManager(config)
|
||||
self.assertFalse(manager.wait_for_triggers())
|
||||
|
||||
@patch.object(TriggerManager, "_build_trigger")
|
||||
def test_any_of_one_passes(self, mock_build):
|
||||
"""mode=any_of, one trigger passes -> returns True immediately."""
|
||||
t1 = StubTrigger(False, "t1")
|
||||
t2 = StubTrigger(True, "t2")
|
||||
mock_build.side_effect = [t1, t2]
|
||||
|
||||
config = _make_config(
|
||||
mode="any_of",
|
||||
conditions=[
|
||||
{"type": "command", "cmd": "a"},
|
||||
{"type": "command", "cmd": "b"},
|
||||
],
|
||||
)
|
||||
manager = TriggerManager(config)
|
||||
self.assertTrue(manager.wait_for_triggers())
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# wait_for_triggers() — timeout tests
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
@patch("krkn.scenario_plugins.triggers.trigger_manager.time")
|
||||
@patch.object(TriggerManager, "_build_trigger")
|
||||
def test_timeout_returns_false(self, mock_build, mock_time):
|
||||
"""Triggers never pass, timeout expires -> returns False."""
|
||||
t1 = StubTrigger(False, "t1")
|
||||
mock_build.side_effect = [t1]
|
||||
|
||||
call_count = 0
|
||||
|
||||
def advancing_monotonic():
|
||||
nonlocal call_count
|
||||
call_count += 1
|
||||
if call_count <= 2:
|
||||
return 0.0
|
||||
return 999.0
|
||||
|
||||
mock_time.monotonic.side_effect = advancing_monotonic
|
||||
mock_time.sleep = lambda x: None
|
||||
|
||||
config = _make_config(
|
||||
timeout=10,
|
||||
conditions=[{"type": "command", "cmd": "a"}],
|
||||
)
|
||||
manager = TriggerManager(config)
|
||||
self.assertFalse(manager.wait_for_triggers())
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Default values
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
@patch.object(TriggerManager, "_build_trigger")
|
||||
def test_default_timeout(self, mock_build):
|
||||
"""No timeout in config -> defaults to 300."""
|
||||
mock_build.return_value = StubTrigger(True)
|
||||
config = {
|
||||
"conditions": [{"type": "command", "cmd": "true"}],
|
||||
}
|
||||
manager = TriggerManager(config)
|
||||
self.assertEqual(manager._timeout, DEFAULT_TIMEOUT)
|
||||
self.assertEqual(manager._timeout, 300)
|
||||
|
||||
@patch.object(TriggerManager, "_build_trigger")
|
||||
def test_default_interval(self, mock_build):
|
||||
"""No interval in config -> defaults to 5."""
|
||||
mock_build.return_value = StubTrigger(True)
|
||||
config = {
|
||||
"conditions": [{"type": "command", "cmd": "true"}],
|
||||
}
|
||||
manager = TriggerManager(config)
|
||||
self.assertEqual(manager._interval, DEFAULT_INTERVAL)
|
||||
self.assertEqual(manager._interval, 5)
|
||||
|
||||
@patch.object(TriggerManager, "_build_trigger")
|
||||
def test_default_mode(self, mock_build):
|
||||
"""No mode in config -> defaults to all_of."""
|
||||
mock_build.return_value = StubTrigger(True)
|
||||
config = {
|
||||
"conditions": [{"type": "command", "cmd": "true"}],
|
||||
}
|
||||
manager = TriggerManager(config)
|
||||
self.assertEqual(manager._mode, DEFAULT_MODE)
|
||||
self.assertEqual(manager._mode, "all_of")
|
||||
|
||||
@patch.object(TriggerManager, "_build_trigger")
|
||||
def test_default_on_timeout(self, mock_build):
|
||||
"""No on_timeout in config -> defaults to skip."""
|
||||
mock_build.return_value = StubTrigger(True)
|
||||
config = {
|
||||
"conditions": [{"type": "command", "cmd": "true"}],
|
||||
}
|
||||
manager = TriggerManager(config)
|
||||
self.assertEqual(manager._on_timeout, DEFAULT_ON_TIMEOUT)
|
||||
self.assertEqual(manager._on_timeout, "skip")
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Validation tests
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def test_invalid_mode_raises(self):
|
||||
"""mode='invalid' -> raises ValueError."""
|
||||
config = _make_config(mode="invalid")
|
||||
with self.assertRaises(ValueError) as ctx:
|
||||
TriggerManager(config)
|
||||
self.assertIn("invalid", str(ctx.exception))
|
||||
|
||||
def test_missing_conditions_raises(self):
|
||||
"""No 'conditions' key -> raises ValueError."""
|
||||
config = {"mode": "all_of", "timeout": 10}
|
||||
with self.assertRaises(ValueError) as ctx:
|
||||
TriggerManager(config)
|
||||
self.assertIn("conditions", str(ctx.exception))
|
||||
|
||||
def test_empty_conditions_raises(self):
|
||||
"""conditions=[] -> raises ValueError."""
|
||||
config = _make_config(conditions=[])
|
||||
with self.assertRaises(ValueError) as ctx:
|
||||
TriggerManager(config)
|
||||
self.assertIn("conditions", str(ctx.exception))
|
||||
|
||||
def test_conditions_not_a_list_raises(self):
|
||||
"""conditions='some string' -> raises ValueError."""
|
||||
config = _make_config(conditions="kubectl check something")
|
||||
with self.assertRaises(ValueError) as ctx:
|
||||
TriggerManager(config)
|
||||
self.assertIn("list", str(ctx.exception))
|
||||
|
||||
def test_on_timeout_property(self):
|
||||
"""on_timeout property returns validated value."""
|
||||
mock_build = patch.object(TriggerManager, "_build_trigger").start()
|
||||
mock_build.return_value = StubTrigger(True)
|
||||
config = _make_config(on_timeout="fail")
|
||||
manager = TriggerManager(config)
|
||||
self.assertEqual(manager.on_timeout, "fail")
|
||||
patch.stopall()
|
||||
|
||||
def test_unknown_trigger_type_raises(self):
|
||||
"""type='kafka' -> raises ValueError."""
|
||||
config = _make_config(
|
||||
conditions=[{"type": "kafka", "topic": "events"}],
|
||||
)
|
||||
with self.assertRaises(ValueError) as ctx:
|
||||
TriggerManager(config)
|
||||
self.assertIn("kafka", str(ctx.exception))
|
||||
|
||||
def test_negative_timeout_raises(self):
|
||||
"""timeout=-1 -> raises ValueError."""
|
||||
config = _make_config(timeout=-1)
|
||||
with self.assertRaises(ValueError) as ctx:
|
||||
TriggerManager(config)
|
||||
self.assertIn("timeout", str(ctx.exception))
|
||||
|
||||
def test_zero_timeout_raises(self):
|
||||
"""timeout=0 -> raises ValueError."""
|
||||
config = _make_config(timeout=0)
|
||||
with self.assertRaises(ValueError) as ctx:
|
||||
TriggerManager(config)
|
||||
self.assertIn("timeout", str(ctx.exception))
|
||||
|
||||
def test_negative_interval_raises(self):
|
||||
"""interval=-1 -> raises ValueError."""
|
||||
config = _make_config(interval=-1)
|
||||
with self.assertRaises(ValueError) as ctx:
|
||||
TriggerManager(config)
|
||||
self.assertIn("interval", str(ctx.exception))
|
||||
|
||||
def test_zero_interval_raises(self):
|
||||
"""interval=0 -> raises ValueError."""
|
||||
config = _make_config(interval=0)
|
||||
with self.assertRaises(ValueError) as ctx:
|
||||
TriggerManager(config)
|
||||
self.assertIn("interval", str(ctx.exception))
|
||||
|
||||
def test_string_timeout_raises(self):
|
||||
"""timeout='abc' -> raises ValueError."""
|
||||
config = _make_config(timeout="abc")
|
||||
with self.assertRaises(ValueError) as ctx:
|
||||
TriggerManager(config)
|
||||
self.assertIn("numeric", str(ctx.exception))
|
||||
|
||||
def test_string_interval_raises(self):
|
||||
"""interval='xyz' -> raises ValueError."""
|
||||
config = _make_config(interval="xyz")
|
||||
with self.assertRaises(ValueError) as ctx:
|
||||
TriggerManager(config)
|
||||
self.assertIn("numeric", str(ctx.exception))
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# get_status() tests
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
@patch.object(TriggerManager, "_build_trigger")
|
||||
def test_get_status(self, mock_build):
|
||||
"""get_status returns dict with trigger states."""
|
||||
t1 = StubTrigger(True, "t1")
|
||||
t2 = StubTrigger(False, "t2")
|
||||
mock_build.side_effect = [t1, t2]
|
||||
|
||||
config = _make_config(
|
||||
mode="any_of",
|
||||
timeout=60,
|
||||
interval=2,
|
||||
on_timeout="fail",
|
||||
conditions=[
|
||||
{"type": "command", "cmd": "a"},
|
||||
{"type": "command", "cmd": "b"},
|
||||
],
|
||||
)
|
||||
manager = TriggerManager(config)
|
||||
|
||||
status = manager.get_status()
|
||||
self.assertEqual(status["mode"], "any_of")
|
||||
self.assertEqual(status["timeout"], 60)
|
||||
self.assertEqual(status["interval"], 2)
|
||||
self.assertEqual(status["on_timeout"], "fail")
|
||||
self.assertEqual(len(status["triggers"]), 2)
|
||||
|
||||
# Before any evaluation, states should be None
|
||||
for trigger_info in status["triggers"]:
|
||||
self.assertIn("description", trigger_info)
|
||||
self.assertIn("satisfied", trigger_info)
|
||||
self.assertIsNone(trigger_info["satisfied"])
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# describe() tests
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
@patch.object(TriggerManager, "_build_trigger")
|
||||
def test_describe(self, mock_build):
|
||||
"""describe returns human-readable summary."""
|
||||
t1 = StubTrigger(True, "t1")
|
||||
t2 = StubTrigger(False, "t2")
|
||||
mock_build.side_effect = [t1, t2]
|
||||
|
||||
config = _make_config(
|
||||
mode="all_of",
|
||||
timeout=30,
|
||||
interval=3,
|
||||
on_timeout="skip",
|
||||
conditions=[
|
||||
{"type": "command", "cmd": "a"},
|
||||
{"type": "command", "cmd": "b"},
|
||||
],
|
||||
)
|
||||
manager = TriggerManager(config)
|
||||
desc = manager.describe()
|
||||
|
||||
self.assertIsInstance(desc, str)
|
||||
self.assertIn("all_of", desc)
|
||||
self.assertIn("30", desc)
|
||||
self.assertIn("3", desc)
|
||||
self.assertIn("skip", desc)
|
||||
# Should include each trigger's description
|
||||
self.assertIn("t1", desc)
|
||||
self.assertIn("t2", desc)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user