add junit into own util (#1529)

krkn/utils/__init__.py

@@ -22,3 +22,4 @@ from .functions import (
     ScenarioTelemetry,
     KrknTelemetryOpenshift
 )
+from .junit import validate_junit_options, write_junit_file

krkn/utils/junit.py

@@ -0,0 +1,83 @@
+# Copyright 2026 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 os
+import time
+
+from krkn_lib.utils.functions import get_junit_test_case
+
+
+def validate_junit_options(junit_testcase, junit_testcase_path):
+    """Validate junit CLI options. Returns (junit_error, junit_normalized_path)."""
+    junit_error = False
+    junit_normalized_path = None
+
+    if junit_testcase_path and not junit_testcase:
+        logging.error(
+            "please set junit test case description with --junit-testcase [description] option"
+        )
+        junit_error = True
+
+    if junit_testcase and not junit_testcase_path:
+        logging.error(
+            "please set junit test case path with --junit-testcase-path [path] option"
+        )
+        junit_error = True
+
+    if junit_testcase and junit_testcase_path:
+        junit_normalized_path = os.path.normpath(junit_testcase_path)
+
+        if not os.path.exists(junit_normalized_path):
+            logging.error(
+                f"{junit_normalized_path} do not exists, please select a valid path"
+            )
+            junit_error = True
+
+        if not os.path.isdir(junit_normalized_path):
+            logging.error(
+                f"{junit_normalized_path} is a file, please select a valid folder path"
+            )
+            junit_error = True
+
+        if not os.access(junit_normalized_path, os.W_OK):
+            logging.error(
+                f"{junit_normalized_path} is not writable, please select a valid path"
+            )
+            junit_error = True
+
+    return junit_error, junit_normalized_path
+
+
+def write_junit_file(
+    junit_normalized_path,
+    success,
+    elapsed_seconds,
+    test_case_description,
+    test_stdout,
+    test_version=None,
+):
+    """Write a junit XML testcase file to junit_normalized_path."""
+    junit_testcase_xml = get_junit_test_case(
+        success=success,
+        time=int(elapsed_seconds),
+        test_suite_name="chaos-krkn",
+        test_case_description=test_case_description,
+        test_stdout=test_stdout,
+        test_version=test_version,
+    )
+    junit_testcase_file_path = f"{junit_normalized_path}/junit_krkn_{int(time.time())}.xml"
+    logging.info(f"writing junit XML testcase in {junit_testcase_file_path}")
+    with open(junit_testcase_file_path, "w") as stream:
+        stream.write(junit_testcase_xml)

run_kraken.py

@@ -23,19 +23,20 @@ warnings.filterwarnings(
 import atexit
 import datetime
 import json
+import logging
+import optparse
 import os
+import queue
 import shutil
 import sys
 import tempfile
-import yaml
-import logging
-import optparse
-from colorlog import ColoredFormatter
-import pyfiglet
-import uuid
 import time
-import queue
-from typing import Optional, Dict
+import uuid
+from typing import Optional
+
+import pyfiglet
+import yaml
+from colorlog import ColoredFormatter
 
 from krkn import cerberus
 from krkn_lib.elastic.krkn_elastic import KrknElastic
@@ -58,9 +59,9 @@ from krkn_lib.telemetry.ocp import KrknTelemetryOpenshift
 from krkn_lib.models.telemetry import ChaosRunTelemetry
 from krkn_lib.models.k8s import ResiliencyReport
 from krkn_lib.utils import SafeLogger
-from krkn_lib.utils.functions import get_yaml_item_value, get_junit_test_case
+from krkn_lib.utils.functions import get_yaml_item_value
 
-from krkn.utils import TeeLogHandler, ErrorCollectionHandler
+from krkn.utils import TeeLogHandler, ErrorCollectionHandler, validate_junit_options, write_junit_file
 from krkn.health_checks import HealthCheckFactory
 from krkn.scenario_plugins.scenario_plugin_factory import (
     ScenarioPluginFactory,
@@ -944,51 +945,13 @@ if __name__ == "__main__":
     )
     option_error = False
 
-    # used to check if there is any missing or wrong parameter that prevents
-    # the creation of the junit file
-    junit_error = False
-    junit_normalized_path = None
-    retval = 0
     junit_start_time = time.time()
-    # checks if both mandatory options for junit are set
-    if options.junit_testcase_path and not options.junit_testcase:
-        logging.error(
-            "please set junit test case description with --junit-testcase [description] option"
-        )
+    retval = 0
+    junit_error, junit_normalized_path = validate_junit_options(
+        options.junit_testcase, options.junit_testcase_path
+    )
+    if junit_error:
         option_error = True
-        junit_error = True
-
-    if options.junit_testcase and not options.junit_testcase_path:
-        logging.error(
-            "please set junit test case path with --junit-testcase-path [path] option"
-        )
-        option_error = True
-        junit_error = True
-
-    # normalized path
-    if options.junit_testcase:
-        junit_normalized_path = os.path.normpath(options.junit_testcase_path)
-
-        if not os.path.exists(junit_normalized_path):
-            logging.error(
-                f"{junit_normalized_path} do not exists, please select a valid path"
-            )
-            option_error = True
-            junit_error = True
-
-        if not os.path.isdir(junit_normalized_path):
-            logging.error(
-                f"{junit_normalized_path} is a file, please select a valid folder path"
-            )
-            option_error = True
-            junit_error = True
-
-        if not os.access(junit_normalized_path, os.W_OK):
-            logging.error(
-                f"{junit_normalized_path} is not writable, please select a valid path"
-            )
-            option_error = True
-            junit_error = True
 
     if options.cfg is None:
         logging.error("Please check if you have passed the config")
@@ -1003,21 +966,14 @@ if __name__ == "__main__":
 
     junit_endtime = time.time()
 
-    # checks the minimum required parameters to write the junit file
     if junit_normalized_path and not junit_error:
-        junit_testcase_xml = get_junit_test_case(
-            success=True if retval == 0 else False,
-            time=int(junit_endtime - junit_start_time),
-            test_suite_name="chaos-krkn",
+        write_junit_file(
+            junit_normalized_path=junit_normalized_path,
+            success=retval == 0,
+            elapsed_seconds=junit_endtime - junit_start_time,
             test_case_description=options.junit_testcase,
             test_stdout=tee_handler.get_output(),
             test_version=options.junit_testcase_version,
         )
-        junit_testcase_file_path = (
-            f"{junit_normalized_path}/junit_krkn_{int(time.time())}.xml"
-        )
-        logging.info(f"writing junit XML testcase in {junit_testcase_file_path}")
-        with open(junit_testcase_file_path, "w") as stream:
-            stream.write(junit_testcase_xml)
 
     sys.exit(retval)

tests/test_junit_utils.py

@@ -0,0 +1,200 @@
+#!/usr/bin/env python3
+"""
+Tests for krkn/utils/junit.py — validate_junit_options and write_junit_file.
+
+Usage:
+    python -m coverage run -a -m unittest tests/test_junit_utils.py -v
+"""
+
+import os
+import shutil
+import stat
+import sys
+import tempfile
+import types
+import unittest
+from unittest.mock import MagicMock, patch
+
+
+# ---------------------------------------------------------------------------
+# Stub external dependencies so no krkn_lib install is needed
+# ---------------------------------------------------------------------------
+
+def _inject(name, **attrs):
+    mod = types.ModuleType(name)
+    for k, v in attrs.items():
+        setattr(mod, k, v)
+    sys.modules.setdefault(name, mod)
+    return sys.modules[name]
+
+
+_inject("krkn_lib")
+_inject("krkn_lib.utils")
+_inject("krkn_lib.utils.functions",
+        get_junit_test_case=MagicMock(return_value="<xml/>"),
+        get_yaml_item_value=MagicMock())
+_inject("krkn_lib.k8s", KrknKubernetes=MagicMock())
+_inject("krkn_lib.ocp", KrknOpenshift=MagicMock())
+_inject("krkn_lib.models.telemetry", ScenarioTelemetry=MagicMock(), ChaosRunTelemetry=MagicMock())
+_inject("krkn_lib.telemetry.ocp", KrknTelemetryOpenshift=MagicMock())
+_inject("krkn_lib.telemetry.k8s", KrknTelemetryKubernetes=MagicMock())
+_inject("tzlocal")
+_inject("tzlocal.unix", get_localzone=MagicMock(return_value="UTC"))
+
+from krkn.utils.junit import validate_junit_options, write_junit_file  # noqa: E402
+
+
+# ===========================================================================
+# validate_junit_options
+# ===========================================================================
+
+class TestValidateJunitOptions(unittest.TestCase):
+
+    def test_neither_set_returns_no_error_and_no_path(self):
+        junit_error, path = validate_junit_options(None, None)
+        self.assertFalse(junit_error)
+        self.assertIsNone(path)
+
+    def test_path_only_returns_error(self):
+        junit_error, path = validate_junit_options(None, "/some/path")
+        self.assertTrue(junit_error)
+        self.assertIsNone(path)
+
+    def test_testcase_only_no_path_returns_error_without_crash(self):
+        # If junit_testcase_path is None, normpath(None) must not be called.
+        junit_error, path = validate_junit_options("my test", None)
+        self.assertTrue(junit_error)
+        self.assertIsNone(path)
+
+    def test_valid_dir_returns_no_error(self):
+        with tempfile.TemporaryDirectory() as tmpdir:
+            junit_error, path = validate_junit_options("my test", tmpdir)
+        self.assertFalse(junit_error)
+        self.assertIsNotNone(path)
+
+    def test_path_is_normalized(self):
+        with tempfile.TemporaryDirectory() as tmpdir:
+            unnormalized = tmpdir + "/."
+            _, path = validate_junit_options("my test", unnormalized)
+        self.assertEqual(path, os.path.normpath(unnormalized))
+
+    def test_nonexistent_path_returns_error(self):
+        junit_error, _ = validate_junit_options("my test", "/nonexistent/path/xyz")
+        self.assertTrue(junit_error)
+
+    def test_file_instead_of_dir_returns_error(self):
+        with tempfile.NamedTemporaryFile() as tmp:
+            junit_error, _ = validate_junit_options("my test", tmp.name)
+        self.assertTrue(junit_error)
+
+    def test_nonwritable_dir_returns_error(self):
+        with tempfile.TemporaryDirectory() as tmpdir:
+            os.chmod(tmpdir, stat.S_IRUSR | stat.S_IXUSR)
+            try:
+                junit_error, _ = validate_junit_options("my test", tmpdir)
+                self.assertTrue(junit_error)
+            finally:
+                os.chmod(tmpdir, stat.S_IRWXU)
+
+
+# ===========================================================================
+# write_junit_file
+# ===========================================================================
+
+class TestWriteJunitFile(unittest.TestCase):
+
+    def setUp(self):
+        self.tmpdir = tempfile.mkdtemp()
+
+    def tearDown(self):
+        shutil.rmtree(self.tmpdir, ignore_errors=True)
+
+    def _get_written_file(self):
+        files = [
+            f for f in os.listdir(self.tmpdir)
+            if f.startswith("junit_krkn_") and f.endswith(".xml")
+        ]
+        self.assertEqual(len(files), 1, f"Expected exactly one junit XML file, got: {files}")
+        return os.path.join(self.tmpdir, files[0])
+
+    @patch("krkn.utils.junit.get_junit_test_case", return_value="<xml>success</xml>")
+    def test_writes_file_on_success(self, _mock):
+        write_junit_file(
+            junit_normalized_path=self.tmpdir,
+            success=True,
+            elapsed_seconds=10.5,
+            test_case_description="chaos run",
+            test_stdout="some output",
+        )
+        with open(self._get_written_file()) as f:
+            self.assertEqual(f.read(), "<xml>success</xml>")
+
+    @patch("krkn.utils.junit.get_junit_test_case", return_value="<xml>failure</xml>")
+    def test_writes_file_on_failure(self, _mock):
+        write_junit_file(
+            junit_normalized_path=self.tmpdir,
+            success=False,
+            elapsed_seconds=5.0,
+            test_case_description="chaos run",
+            test_stdout="error output",
+        )
+        with open(self._get_written_file()) as f:
+            self.assertEqual(f.read(), "<xml>failure</xml>")
+
+    @patch("krkn.utils.junit.get_junit_test_case", return_value="<xml/>")
+    def test_passes_correct_args_to_get_junit_test_case(self, mock_get):
+        write_junit_file(
+            junit_normalized_path=self.tmpdir,
+            success=True,
+            elapsed_seconds=42.9,
+            test_case_description="my scenario",
+            test_stdout="stdout here",
+            test_version="v1.2.3",
+        )
+        mock_get.assert_called_once_with(
+            success=True,
+            time=42,
+            test_suite_name="chaos-krkn",
+            test_case_description="my scenario",
+            test_stdout="stdout here",
+            test_version="v1.2.3",
+        )
+
+    @patch("krkn.utils.junit.get_junit_test_case", return_value="<xml/>")
+    def test_elapsed_seconds_truncated_to_int(self, mock_get):
+        write_junit_file(
+            junit_normalized_path=self.tmpdir,
+            success=True,
+            elapsed_seconds=99.99,
+            test_case_description="t",
+            test_stdout="",
+        )
+        self.assertEqual(mock_get.call_args[1]["time"], 99)
+
+    @patch("krkn.utils.junit.get_junit_test_case", return_value="<xml/>")
+    def test_file_name_matches_expected_pattern(self, _mock):
+        write_junit_file(
+            junit_normalized_path=self.tmpdir,
+            success=True,
+            elapsed_seconds=1.0,
+            test_case_description="t",
+            test_stdout="",
+        )
+        files = os.listdir(self.tmpdir)
+        self.assertEqual(len(files), 1)
+        self.assertRegex(files[0], r"^junit_krkn_\d+\.xml$")
+
+    @patch("krkn.utils.junit.get_junit_test_case", return_value="<xml/>")
+    def test_version_defaults_to_none(self, mock_get):
+        write_junit_file(
+            junit_normalized_path=self.tmpdir,
+            success=True,
+            elapsed_seconds=1.0,
+            test_case_description="t",
+            test_stdout="",
+        )
+        self.assertIsNone(mock_get.call_args[1]["test_version"])
+
+
+if __name__ == "__main__":
+    unittest.main()