Reduce the number of Makefiles in 'doc/'

[babeltrace.git] / tests / data / cli / exit_status / bt_plugin_test_cli_exit_status.py

# SPDX-License-Identifier: GPL-2.0-only
#
# Copyright (C) 2019 EfficiOS Inc.
#

import os
import time
import signal

import bt2

bt2.register_plugin(__name__, "test_exit_status")


class StatusIter(bt2._UserMessageIterator):
    def __init__(self, config, output_port):
        self.case = output_port.user_data["case"]

    def __next__(self):
        if self.case == "STOP":
            raise bt2.Stop()
        if self.case == "INTERRUPTED":
            os.kill(os.getpid(), signal.SIGINT)

            # Wait until the graph is in the interrupted state.
            timeout_s = 10
            for _ in range(timeout_s * 10):
                if self._is_interrupted:
                    raise bt2.TryAgain()

                time.sleep(0.1)

            raise Exception(
                "{} was not interrupted after {} seconds".format(
                    self.__class__.__name__, timeout_s
                )
            )

        elif self.case == "ERROR":
            raise TypeError("Raising type error")
        else:
            raise ValueError("Invalid parameter")


@bt2.plugin_component_class
class StatusSrc(bt2._UserSourceComponent, message_iterator_class=StatusIter):
    def __init__(self, config, params, obj):
        self._add_output_port("out", {"case": params["case"]})