krkn/krkn/tests/test_classes/example_scenario_plugin.py

from krkn_lib.models.telemetry import ScenarioTelemetry
from krkn_lib.telemetry.ocp import KrknTelemetryOpenshift
from krkn.scenario_plugins.abstract_scenario_plugin import AbstractScenarioPlugin


# Each plugin must extend the AbstractScenarioPlugin abstract class
# and implement its methods. Also the naming conventions must be respected
# you can refer to the documentation for the details:
# https://github.com/krkn-chaos/krkn/blob/main/docs/scenario_plugin_api.md
class ExampleScenarioPlugin(AbstractScenarioPlugin):

    def run(
        self,
        run_uuid: str,
        scenario: str,
        krkn_config: dict[str, any],
        lib_telemetry: KrknTelemetryOpenshift,
        scenario_telemetry: ScenarioTelemetry,
    ) -> int:
        """
        :param run_uuid: the uuid of the chaos run generated by krkn for every single run
        :param scenario: the config file of the scenario that is currently executed
        :param krkn_config: the full dictionary representation of the `config.yaml`
        :param lib_telemetry: it is a composite object of all the
        [krkn-lib](https://krkn-chaos.github.io/krkn-lib-docs/modules.html)
        objects and methods needed by a krkn plugin to run.
        :param scenario_telemetry: the `ScenarioTelemetry` object of the scenario that is currently executed
        """

        pass

        try:
            # The scenario logic for each scenario must be placed
            # here. A try-except it is needed to catch exceptions
            # that may occur in this section and they shouldn't
            # be propagated outside (only int return value is admitted).

            # krkn-lib KrknKubernetes object containing all the kubernetes primitives
            # can be retrieved by the KrknTelemetryOpenshift object
            krkn_kubernetes = lib_telemetry.get_lib_kubernetes()

            # krkn-lib KrknOpenshift object containing all the OCP primitives
            # can be retrieved by the KrknTelemetryOpenshift object
            krkn_openshift = lib_telemetry.get_lib_ocp()

            # if the scenario succeeds the telemetry exit status is 0
            return 0
        except Exception as e:
            # if the scenario fails the telemetry exit status is 1
            return 1

    # Reflects the scenario type defined in the config.yaml
    # in the chaos_scenarios section and to which each class
    # responds.
    def get_scenario_types(self) -> list[str]:
        return ["example_scenarios"]