[ISV-5275] Prepare the integration test framework scaffolding (2/2) (#754)

* [ISV-5275] Prepare the integration test framework scaffolding (2/2)

Signed-off-by: Maurizio Porrato <mporrato@redhat.com>

Author: mporrato

operator-pipeline-images/operatorcert/integration/testcase.py

@@ -0,0 +1,101 @@
+"""
+Framework for the definition of integration test cases as python classes.
+"""
+
+import logging
+from typing import TypeVar
+
+from operatorcert.integration.config import Config
+from colorama import Fore, Style, init as colorama_init
+
+
+LOGGER = logging.getLogger("operator-cert")
+
+
+class BaseTestCase:
+    """
+    The base class for user-defined integration test cases
+    """
+
+    def __init__(self, config: Config, logger: logging.Logger) -> None:
+        self.config = config
+        self.logger = logger
+
+    def setup(self) -> None:
+        """
+        This method is the first to be called in the test execution and should
+        create the resources required by the test case
+        """
+
+    def watch(self) -> None:
+        """
+        This method is called after `setup()` and should watch the execution
+        of the pipeline and either return when the pipeline has finished or
+        raise an exception if it reaches an unexpected state
+        """
+
+    def validate(self) -> None:
+        """
+        This method is called after `watch()` terminates and should check
+        the state of all the resources involved in the test case and raise
+        an exception if the actual state does not match the expected state
+        """
+
+    def cleanup(self) -> None:
+        """
+        This method is called at the very end of the test case, even if a
+        previous step raised an exception; it should be used to free up
+        any resources created during the execution of the test
+        """
+
+    def run(self) -> None:
+        """
+        Execute the test case; `setup()`, `watch()`, `validate()` and
+        `cleanup()` are called in order
+        """
+        try:
+            self.setup()
+            self.watch()
+            self.validate()
+        except Exception as e:
+            raise e
+        finally:
+            self.cleanup()
+
+
+_test_cases = []
+
+
+_T = TypeVar("_T", bound=type)
+
+
+def integration_test_case(test_class: _T) -> _T:
+    """
+    Decorator used to register a class as a test case
+    """
+    _test_cases.append(test_class)
+    return test_class
+
+
+def run_tests(config: Config) -> int:
+    """
+    Executes all the test cases that have been registered using the
+    `integration_test_case` decorator
+
+    Return:
+        number of test cases that failed
+    """
+    colorama_init()
+    failed = 0
+    for test_class in _test_cases:
+        test_name = test_class.__name__
+        print(f"Running {test_name} ", end="")
+        try:
+            test_instance = test_class(config, LOGGER)
+            test_instance.run()
+            print(f"{Fore.GREEN}PASS{Style.RESET_ALL}")
+        except Exception as e:  # pylint: disable=broad-except
+            print(f"{Fore.RED}FAIL{Style.RESET_ALL}")
+            LOGGER.error("Test %s failed:", test_name, exc_info=e)
+            failed += 1
+    return failed

operator-pipeline-images/tests/integration/test_testcase.py

@@ -0,0 +1,64 @@
+from unittest.mock import patch, MagicMock
+
+import pytest
+
+from operatorcert.integration.config import Config
+from operatorcert.integration.testcase import (
+    BaseTestCase,
+    integration_test_case,
+    run_tests,
+)
+
+
+@patch("operatorcert.integration.testcase.BaseTestCase.setup")
+@patch("operatorcert.integration.testcase.BaseTestCase.watch")
+@patch("operatorcert.integration.testcase.BaseTestCase.validate")
+@patch("operatorcert.integration.testcase.BaseTestCase.cleanup")
+def test_basetestcase(
+    mock_cleanup: MagicMock,
+    mock_validate: MagicMock,
+    mock_watch: MagicMock,
+    mock_setup: MagicMock,
+) -> None:
+    t = BaseTestCase(MagicMock(), MagicMock())
+
+    # happy path
+    t.run()
+    mock_setup.assert_called_once()
+    mock_watch.assert_called_once()
+    mock_validate.assert_called_once()
+    mock_cleanup.assert_called_once()
+
+    # test raises exception
+    for m in (mock_setup, mock_watch, mock_validate, mock_cleanup):
+        m.reset_mock()
+
+    mock_watch.side_effect = Exception()
+    with pytest.raises(Exception):
+        t.run()
+    mock_setup.assert_called_once()
+    mock_watch.assert_called_once()
+    mock_validate.assert_not_called()
+    mock_cleanup.assert_called_once()
+
+
+@patch("operatorcert.integration.testcase._test_cases")
+def test_integration_test_case(mock_test_cases: MagicMock) -> None:
+    fake_class = MagicMock(spec=BaseTestCase)
+    integration_test_case(fake_class)
+    mock_test_cases.append.assert_called_once_with(fake_class)
+
+
+class GoodTest(BaseTestCase):
+    pass
+
+
+class BadTest(BaseTestCase):
+    def watch(self) -> None:
+        raise Exception()
+
+
+@patch("operatorcert.integration.testcase._test_cases", [GoodTest, BadTest, GoodTest])
+def test_run() -> None:
+    fake_config = MagicMock(spec=Config)
+    assert run_tests(fake_config) == 1

pdm.lock

@@ -101,6 +101,7 @@ operatorcert-dev = [
     "jmespath>=1.0.1",
     "kubernetes>=31.0.0",
     "ansible-lint>=24.10.0",
+    "colorama>=0.4.6",
 ]
 tox = ["tox>=4.16.0", "tox-pdm>=0.7.2"]