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 os
  10 import sys
  11 import time
  12 from typing import Iterator, Optional
  13
  14
  15 def _get_time_ns():
  16     # type: () -> int
  17
  18     # time.monotonic is only available since Python 3.3. We don't support
  19     # those older versions so we can simply assert here.
  20     assert sys.version_info >= (3, 3, 0)
  21
  22     # time.monotonic_ns is only available for python >= 3.8,
  23     # so the value is multiplied by 10^9 to maintain compatibility with
  24     # older versions of the interpreter.
  25     return int(time.monotonic() * 1000000000)
  26
  27
  28 class InvalidTestPlan(RuntimeError):
  29     def __init__(self, msg):
  30         # type: (str) -> None
  31         super().__init__(msg)
  32
  33
  34 class BailOut(RuntimeError):
  35     def __init__(self, msg):
  36         # type: (str) -> None
  37         super().__init__(msg)
  38
  39
  40 class TestCase:
  41     def __init__(
  42         self,
  43         tap_generator,  # type: "TapGenerator"
  44         description,  # type: str
  45     ):
  46         self._tap_generator = tap_generator  # type: "TapGenerator"
  47         self._result = None  # type: Optional[bool]
  48         self._description = description  # type: str
  49
  50     @property
  51     def result(self):
  52         # type: () -> Optional[bool]
  53         return self._result
  54
  55     @property
  56     def description(self):
  57         # type: () -> str
  58         return self._description
  59
  60     def _set_result(self, result):
  61         # type: (bool) -> None
  62         if self._result is not None:
  63             raise RuntimeError("Can't set test case result twice")
  64
  65         self._result = result
  66         self._tap_generator.test(result, self._description)
  67
  68     def success(self):
  69         # type: () -> None
  70         self._set_result(True)
  71
  72     def fail(self):
  73         # type: () -> None
  74         self._set_result(False)
  75
  76
  77 # Produces a test execution report in the TAP format.
  78 class TapGenerator:
  79     def __init__(self, total_test_count):
  80         # type: (int) -> None
  81         if total_test_count <= 0:
  82             raise ValueError("Test count must be greater than zero")
  83
  84         self._total_test_count = total_test_count  # type: int
  85         self._last_test_case_id = 0  # type: int
  86         self._printed_plan = False  # type: bool
  87         self._has_failure = False  # type: bool
  88         self._time_tests = True  # type: bool
  89         if os.getenv("LTTNG_TESTS_TAP_AUTOTIME", "1") == "0":
  90             self._time_tests = False
  91         self._last_time = _get_time_ns()
  92
  93     def __del__(self):
  94         if self.remaining_test_cases > 0:
  95             self.bail_out(
  96                 "Missing {remaining_test_cases} test cases".format(
  97                     remaining_test_cases=self.remaining_test_cases
  98                 )
  99             )
 100
 101     @property
 102     def remaining_test_cases(self):
 103         # type: () -> int
 104         return self._total_test_count - self._last_test_case_id
 105
 106     def _print(self, msg):
 107         # type: (str) -> None
 108         if not self._printed_plan:
 109             print(
 110                 "1..{total_test_count}".format(total_test_count=self._total_test_count),
 111                 flush=True,
 112             )
 113             self._printed_plan = True
 114
 115         print(msg, flush=True)
 116
 117     def skip_all(self, reason):
 118         # type: (str) -> None
 119         if self._last_test_case_id != 0:
 120             raise RuntimeError("Can't skip all tests after running test cases")
 121
 122         if reason:
 123             self._print("1..0 # Skip all: {reason}".format(reason=reason))
 124
 125         self._last_test_case_id = self._total_test_count
 126
 127     def skip(self, reason, skip_count=1):
 128         # type: (str, int) -> None
 129         for i in range(skip_count):
 130             self._last_test_case_id = self._last_test_case_id + 1
 131             self._print(
 132                 "ok {test_number} # Skip: {reason}".format(
 133                     reason=reason, test_number=(self._last_test_case_id)
 134                 )
 135             )
 136
 137     def bail_out(self, reason):
 138         # type: (str) -> None
 139         self._print("Bail out! {reason}".format(reason=reason))
 140         self._last_test_case_id = self._total_test_count
 141         raise BailOut(reason)
 142
 143     def test(self, result, description):
 144         # type: (bool, str) -> None
 145         duration = (_get_time_ns() - self._last_time) / 1000000
 146         if self._last_test_case_id == self._total_test_count:
 147             raise InvalidTestPlan("Executing too many tests")
 148
 149         if result is False:
 150             self._has_failure = True
 151
 152         result_string = "ok" if result else "not ok"
 153         self._last_test_case_id = self._last_test_case_id + 1
 154         self._print(
 155             "{result_string} {case_id} - {description}".format(
 156                 result_string=result_string,
 157                 case_id=self._last_test_case_id,
 158                 description=description,
 159             )
 160         )
 161         if self._time_tests:
 162             self._print("---\n  duration_ms: {}\n...\n".format(duration))
 163         self._last_time = _get_time_ns()
 164
 165     def ok(self, description):
 166         # type: (str) -> None
 167         self.test(True, description)
 168
 169     def fail(self, description):
 170         # type: (str) -> None
 171         self.test(False, description)
 172
 173     @property
 174     def is_successful(self):
 175         # type: () -> bool
 176         return (
 177             self._last_test_case_id == self._total_test_count and not self._has_failure
 178         )
 179
 180     @contextlib.contextmanager
 181     def case(self, description):
 182         # type: (str) -> Iterator[TestCase]
 183         test_case = TestCase(self, description)
 184         try:
 185             yield test_case
 186         except Exception as e:
 187             self.diagnostic(
 188                 "Exception `{exception_type}` thrown during test case `{description}`, marking as failure.".format(
 189                     description=test_case.description, exception_type=type(e).__name__
 190                 )
 191             )
 192
 193             if str(e) != "":
 194                 self.diagnostic(str(e))
 195
 196             test_case.fail()
 197         finally:
 198             if test_case.result is None:
 199                 test_case.success()
 200
 201     def diagnostic(self, msg):
 202         # type: (str) -> None
 203         print("# {msg}".format(msg=msg), file=sys.stderr, flush=True)