3 # Copyright (C) 2022 Jérémie Galarneau <jeremie.galarneau@efficios.com>
5 # SPDX-License-Identifier: GPL-2.0-only
12 from typing
import Iterator
, Optional
16 assert sys
.version_info
> (3, 3, 0)
17 # time.monotonic_ns is only available for python >= 3.8
18 return time
.monotonic() * 1000000000
21 class InvalidTestPlan(RuntimeError):
22 def __init__(self
, msg
):
27 class BailOut(RuntimeError):
28 def __init__(self
, msg
):
36 tap_generator
, # type: "TapGenerator"
37 description
, # type: str
39 self
._tap
_generator
= tap_generator
# type: "TapGenerator"
40 self
._result
= None # type: Optional[bool]
41 self
._description
= description
# type: str
45 # type: () -> Optional[bool]
49 def description(self
):
51 return self
._description
53 def _set_result(self
, result
):
54 # type: (bool) -> None
55 if self
._result
is not None:
56 raise RuntimeError("Can't set test case result twice")
59 self
._tap
_generator
.test(result
, self
._description
)
63 self
._set
_result
(True)
67 self
._set
_result
(False)
70 # Produces a test execution report in the TAP format.
72 def __init__(self
, total_test_count
):
74 if total_test_count
<= 0:
75 raise ValueError("Test count must be greater than zero")
77 self
._total
_test
_count
= total_test_count
# type: int
78 self
._last
_test
_case
_id
= 0 # type: int
79 self
._printed
_plan
= False # type: bool
80 self
._has
_failure
= False # type: bool
81 self
._time
_tests
= True # type: bool
82 if os
.getenv("TAP_AUTOTIME", "1") == "" or os
.getenv("TAP_AUTOTIME", "1") == "0":
83 self
._time
_tests
= False
84 self
._last
_time
= _get_time_ns()
87 if self
.remaining_test_cases
> 0:
89 "Missing {remaining_test_cases} test cases".format(
90 remaining_test_cases
=self
.remaining_test_cases
95 def remaining_test_cases(self
):
97 return self
._total
_test
_count
- self
._last
_test
_case
_id
99 def _print(self
, msg
):
100 # type: (str) -> None
101 if not self
._printed
_plan
:
103 "1..{total_test_count}".format(total_test_count
=self
._total
_test
_count
),
106 self
._printed
_plan
= True
108 print(msg
, flush
=True)
110 def skip_all(self
, reason
):
111 # type: (str) -> None
112 if self
._last
_test
_case
_id
!= 0:
113 raise RuntimeError("Can't skip all tests after running test cases")
116 self
._print
("1..0 # Skip all: {reason}".format(reason
=reason
))
118 self
._last
_test
_case
_id
= self
._total
_test
_count
120 def skip(self
, reason
, skip_count
=1):
121 # type: (str, int) -> None
122 for i
in range(skip_count
):
123 self
._last
_test
_case
_id
= self
._last
_test
_case
_id
+ 1
125 "ok {test_number} # Skip: {reason}".format(
126 reason
=reason
, test_number
=(i
+ self
._last
_test
_case
_id
)
130 def bail_out(self
, reason
):
131 # type: (str) -> None
132 self
._print
("Bail out! {reason}".format(reason
=reason
))
133 self
._last
_test
_case
_id
= self
._total
_test
_count
134 raise BailOut(reason
)
136 def test(self
, result
, description
):
137 # type: (bool, str) -> None
138 duration
= (_get_time_ns() - self
._last
_time
) / 1000000
139 if self
._last
_test
_case
_id
== self
._total
_test
_count
:
140 raise InvalidTestPlan("Executing too many tests")
143 self
._has
_failure
= True
145 result_string
= "ok" if result
else "not ok"
146 self
._last
_test
_case
_id
= self
._last
_test
_case
_id
+ 1
148 "{result_string} {case_id} - {description}".format(
149 result_string
=result_string
,
150 case_id
=self
._last
_test
_case
_id
,
151 description
=description
,
155 self
._print
("---\n duration_ms: {}\n...\n".format(duration
))
156 self
._last
_time
= _get_time_ns()
158 def ok(self
, description
):
159 # type: (str) -> None
160 self
.test(True, description
)
162 def fail(self
, description
):
163 # type: (str) -> None
164 self
.test(False, description
)
167 def is_successful(self
):
170 self
._last
_test
_case
_id
== self
._total
_test
_count
and not self
._has
_failure
173 @contextlib.contextmanager
174 def case(self
, description
):
175 # type: (str) -> Iterator[TestCase]
176 test_case
= TestCase(self
, description
)
179 except Exception as e
:
181 "Exception `{exception_type}` thrown during test case `{description}`, marking as failure.".format(
182 description
=test_case
.description
, exception_type
=type(e
).__name
__
187 self
.diagnostic(str(e
))
191 if test_case
.result
is None:
194 def diagnostic(self
, msg
):
195 # type: (str) -> None
196 print("# {msg}".format(msg
=msg
), file=sys
.stderr
, flush
=True)
This page took 0.036653 seconds and 4 git commands to generate.