tests/data/cli/exit-status/bt_plugin_test_cli_exit_status.py

   1 # SPDX-License-Identifier: GPL-2.0-only
   2 #
   3 # Copyright (C) 2019 EfficiOS Inc.
   4 #
   5
   6 import os
   7 import time
   8 import signal
   9
  10 import bt2
  11
  12 bt2.register_plugin(__name__, "test-exit-status")
  13
  14
  15 class StatusIter(bt2._UserMessageIterator):
  16     def __init__(self, config, output_port):
  17         self.case = output_port.user_data["case"]
  18
  19     def __next__(self):
  20         if self.case == "STOP":
  21             raise bt2.Stop()
  22         if self.case == "INTERRUPTED":
  23             os.kill(os.getpid(), signal.SIGINT)
  24
  25             # Wait until the graph is in the interrupted state.
  26             timeout_s = 10
  27             for _ in range(timeout_s * 10):
  28                 if self._is_interrupted:
  29                     raise bt2.TryAgain()
  30
  31                 time.sleep(0.1)
  32
  33             raise Exception(
  34                 "{} was not interrupted after {} seconds".format(
  35                     self.__class__.__name__, timeout_s
  36                 )
  37             )
  38
  39         elif self.case == "ERROR":
  40             raise TypeError("Raising type error")
  41         else:
  42             raise ValueError("Invalid parameter")
  43
  44
  45 @bt2.plugin_component_class
  46 class StatusSrc(bt2._UserSourceComponent, message_iterator_class=StatusIter):
  47     def __init__(self, config, params, obj):
  48         self._add_output_port("out", {"case": params["case"]})