tests/utils/lttngtest/tap_generator.py

   1 #!/usr/bin/env python3
   2 #
   3 # Copyright (C) 2022 Jérémie Galarneau <jeremie.galarneau@efficios.com>
   4 #
   5 # SPDX-License-Identifier: GPL-2.0-only
   6 #
   7
   8 import contextlib
   9 import sys
  10 from typing import Iterator, Optional
  11
  12
  13 class InvalidTestPlan(RuntimeError):
  14     def __init__(self, msg):
  15         # type: (str) -> None
  16         super().__init__(msg)
  17
  18
  19 class BailOut(RuntimeError):
  20     def __init__(self, msg):
  21         # type: (str) -> None
  22         super().__init__(msg)
  23
  24
  25 class TestCase:
  26     def __init__(
  27         self,
  28         tap_generator,  # type: "TapGenerator"
  29         description,  # type: str
  30     ):
  31         self._tap_generator = tap_generator  # type: "TapGenerator"
  32         self._result = None  # type: Optional[bool]
  33         self._description = description  # type: str
  34
  35     @property
  36     def result(self):
  37         # type: () -> Optional[bool]
  38         return self._result
  39
  40     @property
  41     def description(self):
  42         # type: () -> str
  43         return self._description
  44
  45     def _set_result(self, result):
  46         # type: (bool) -> None
  47         if self._result is not None:
  48             raise RuntimeError("Can't set test case result twice")
  49
  50         self._result = result
  51         self._tap_generator.test(result, self._description)
  52
  53     def success(self):
  54         # type: () -> None
  55         self._set_result(True)
  56
  57     def fail(self):
  58         # type: () -> None
  59         self._set_result(False)
  60
  61
  62 # Produces a test execution report in the TAP format.
  63 class TapGenerator:
  64     def __init__(self, total_test_count):
  65         # type: (int) -> None
  66         if total_test_count <= 0:
  67             raise ValueError("Test count must be greater than zero")
  68
  69         self._total_test_count = total_test_count  # type: int
  70         self._last_test_case_id = 0  # type: int
  71         self._printed_plan = False  # type: bool
  72         self._has_failure = False  # type: bool
  73
  74     def __del__(self):
  75         if self.remaining_test_cases > 0:
  76             self.bail_out(
  77                 "Missing {remaining_test_cases} test cases".format(
  78                     remaining_test_cases=self.remaining_test_cases
  79                 )
  80             )
  81
  82     @property
  83     def remaining_test_cases(self):
  84         # type: () -> int
  85         return self._total_test_count - self._last_test_case_id
  86
  87     def _print(self, msg):
  88         # type: (str) -> None
  89         if not self._printed_plan:
  90             print(
  91                 "1..{total_test_count}".format(total_test_count=self._total_test_count),
  92                 flush=True,
  93             )
  94             self._printed_plan = True
  95
  96         print(msg, flush=True)
  97
  98     def skip_all(self, reason):
  99         # type: (str) -> None
 100         if self._last_test_case_id != 0:
 101             raise RuntimeError("Can't skip all tests after running test cases")
 102
 103         if reason:
 104             self._print("1..0 # Skip all: {reason}".format(reason=reason))
 105
 106         self._last_test_case_id = self._total_test_count
 107
 108     def skip(self, reason, skip_count=1):
 109         # type: (str, int) -> None
 110         for i in range(skip_count):
 111             self._last_test_case_id = self._last_test_case_id + 1
 112             self._print(
 113                 "ok {test_number} # Skip: {reason}".format(
 114                     reason=reason, test_number=(i + self._last_test_case_id)
 115                 )
 116             )
 117
 118     def bail_out(self, reason):
 119         # type: (str) -> None
 120         self._print("Bail out! {reason}".format(reason=reason))
 121         self._last_test_case_id = self._total_test_count
 122         raise BailOut(reason)
 123
 124     def test(self, result, description):
 125         # type: (bool, str) -> None
 126         if self._last_test_case_id == self._total_test_count:
 127             raise InvalidTestPlan("Executing too many tests")
 128
 129         if result is False:
 130             self._has_failure = True
 131
 132         result_string = "ok" if result else "not ok"
 133         self._last_test_case_id = self._last_test_case_id + 1
 134         self._print(
 135             "{result_string} {case_id} - {description}".format(
 136                 result_string=result_string,
 137                 case_id=self._last_test_case_id,
 138                 description=description,
 139             )
 140         )
 141
 142     def ok(self, description):
 143         # type: (str) -> None
 144         self.test(True, description)
 145
 146     def fail(self, description):
 147         # type: (str) -> None
 148         self.test(False, description)
 149
 150     @property
 151     def is_successful(self):
 152         # type: () -> bool
 153         return (
 154             self._last_test_case_id == self._total_test_count and not self._has_failure
 155         )
 156
 157     @contextlib.contextmanager
 158     def case(self, description):
 159         # type: (str) -> Iterator[TestCase]
 160         test_case = TestCase(self, description)
 161         try:
 162             yield test_case
 163         except Exception as e:
 164             self.diagnostic(
 165                 "Exception `{exception_type}` thrown during test case `{description}`, marking as failure.".format(
 166                     description=test_case.description, exception_type=type(e).__name__
 167                 )
 168             )
 169
 170             if str(e) != "":
 171                 self.diagnostic(str(e))
 172
 173             test_case.fail()
 174         finally:
 175             if test_case.result is None:
 176                 test_case.success()
 177
 178     def diagnostic(self, msg):
 179         # type: (str) -> None
 180         print("# {msg}".format(msg=msg), file=sys.stderr, flush=True)