Coverage for pyTooling/Warning/__init_

1# ==================================================================================================================== #

2# _____ _ _ __ __ _ #

3# _ __ _ |_ _|__ ___ | (_)_ __ __ \ \ / /_ _ _ __ _ __ (_)_ __ __ _ #

4# | '_ \| | | || |/ _ \ / _ \| | | '_ \ / _` \ \ /\ / / _` | '__| '_ \| | '_ \ / _` | #

5# | |_) | |_| || | (_) | (_) | | | | | | (_| |\ V V / (_| | | | | | | | | | | (_| | #

6# | .__/ \__, ||_|\___/ \___/|_|_|_| |_|\__, (_)_/\_/ \__,_|_| |_| |_|_|_| |_|\__, | #

7# |_| |___/ |___/ |___/ #

8# ==================================================================================================================== #

9# Authors: #

10# Patrick Lehmann #

11# #

12# License: #

13# ==================================================================================================================== #

15# #

16# Licensed under the Apache License, Version 2.0 (the "License"); #

17# you may not use this file except in compliance with the License. #

18# You may obtain a copy of the License at #

19# #

20# http://www.apache.org/licenses/LICENSE-2.0 #

21# #

22# Unless required by applicable law or agreed to in writing, software #

23# distributed under the License is distributed on an "AS IS" BASIS, #

24# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. #

25# See the License for the specific language governing permissions and #

26# limitations under the License. #

27# #

28# SPDX-License-Identifier: Apache-2.0 #

29# ==================================================================================================================== #

30#

31"""

32A solution to send warnings like exceptions to a handler in the upper part of the call-stack.

34.. hint::

36 See :ref:`high-level help <WARNING>` for explanations and usage examples.

37"""

38from threading import local, Lock

39from types import TracebackType

40from typing import List, Callable, Optional as Nullable, Type, Iterator, Self, Iterable, Tuple, Union

42from pyTooling.Decorators import export, readonly

43from pyTooling.Common import getFullyQualifiedName

44from pyTooling.Exceptions import ExceptionBase

47__all__ = ["_threadLocalData", "AnyWarning"]

50_threadLocalData = local()

51"""A reference to the thread local data needed by the pyTooling.Warning classes."""

54@export

55class CriticalWarning(BaseException):

56 """

57 Base-exception of all critical warnings handled by :class:`WarningCollector`.

59 .. tip::

61 Critical warnings must be unhandled within a call hierarchy, otherwise a :exc:`UnhandledCriticalWarningException`

62 will be raised.

63 """

65 @readonly

66 def HasNotes(self) -> bool:

67 """

68 Read-only property to return if the warning has attached notes.

70 :returns: True, if the warning has attached notes.

71 """

72 return hasattr(self, "__notes__") and self.__notes__ is not None and len(self.__notes__) > 0

74 @readonly

75 def Notes(self) -> Tuple[str, ...]:

76 """

77 Read-only property to return warning's attached notes.

79 :returns: Attached notes.

80 """

81 return tuple(self.__notes__) if hasattr(self, "__notes__") else tuple()

84@export

85class Warning(BaseException):

86 """

87 Base-exception of all warnings handled by :class:`WarningCollector`.

89 .. tip::

91 Warnings can be unhandled within a call hierarchy.

92 """

94 @readonly

95 def HasNotes(self) -> bool:

96 """

97 Read-only property to return if the warning has attached notes.

99 :returns: True, if the warning has attached notes.

100 """

101 return hasattr(self, "__notes__") and self.__notes__ is not None and len(self.__notes__) > 0

102

103 @readonly

104 def Notes(self) -> Tuple[str, ...]:

105 """

106 Read-only property to return warning's attached notes.

107

108 :returns: Attached notes.

109 """

110 return tuple(self.__notes__) if hasattr(self, "__notes__") else tuple()

111

112

113AnyWarning = Union[CriticalWarning, Warning]

114

115

116@export

117class UnhandledWarningException(ExceptionBase): # FIXME: to be removed in v9.0.0

118 """

119 Deprecated.

120

121 .. deprecated:: v9.0.0

122

123 Please use :exc:`UnhandledCriticalWarningException`.

124 """

125

126

127@export

128class UnhandledCriticalWarningException(UnhandledWarningException):

129 """

130 This exception is raised when a critical warning isn't handled by a :class:`WarningCollector` within the

131 call-hierarchy.

132 """

133

134

135@export

136class UnhandledExceptionException(UnhandledWarningException):

137 """

138 This exception is raised when an exception isn't handled by a :class:`WarningCollector` within the call-hierarchy.

139 """

140

141

142@export

143class WarningCollector:

144 """

145 A context manager to collect warnings within the call hierarchy.

146 """

147 _parent: Nullable["WarningCollector"] #: Parent WarningCollector

148 _warnings: List[BaseException] #: List of collected warnings (and exceptions).

149 _handler: Nullable[Callable[[BaseException], bool]] #: Optional handler function, which is called per collected warning.

150

151 __slots__ = ("_parent", "_warnings", "_handler")

152

153 def __init__(

154 self,

155 warnings: Nullable[List[BaseException]] = None,

156 handler: Nullable[Callable[[BaseException], bool]] = None

157 ) -> None:

158 """

159 Initializes a warning collector.

160

161 :param warnings: An optional reference to a list of warnings, which can be modified (appended) by this warning

162 collector. If ``None``, an internal list is created and can be referenced by the collector's

163 instance.

164 :param handler: An optional handler function, which processes the current warning and decides if a warning should

165 be reraised as an exception.

166 :raises TypeError: If optional parameter 'warnings' is not of type list.

167 :raises TypeError: If optional parameter 'handler' is not a callable.

168 """

169 if warnings is None:

170 warnings = []

171 elif not isinstance(warnings, list): 171 ↛ 172line 171 didn't jump to line 172 because the condition on line 171 was never true

172 ex = TypeError(f"Parameter 'warnings' is not a list.")

173 ex.add_note(f"Got type '{getFullyQualifiedName(warnings)}'.")

174 raise ex

175

176 if handler is not None and not isinstance(handler, Callable): 176 ↛ 177line 176 didn't jump to line 177 because the condition on line 176 was never true

177 ex = TypeError(f"Parameter 'handler' is not callable.")

178 ex.add_note(f"Got type '{getFullyQualifiedName(handler)}'.")

179 raise ex

180

181 self._parent = None

182 self._warnings = warnings

183 self._handler = handler

184

185 def __len__(self) -> int:

186 """

187 Returns the number of collected warnings.

188

189 :returns: Number of collected warnings.

190 """

191 return len(self._warnings)

192

193 def __iter__(self) -> Iterator[Warning | CriticalWarning | Exception]:

194 """

195 Return an iterator over all collected warnings.

196

197 :returns: Iterator over the collected warnings.

198 """

199 return iter(self._warnings)

200

201 def __getitem__(self, index: int) -> Warning | CriticalWarning | Exception:

202 """

203 Access a collected warning by index.

204

205 :param index: Index of the warning.

206 :returns: Collected warning.

207 """

208 return self._warnings[index]

209

210 def __enter__(self) -> Self:

211 """

212 Enter the warning collector context.

213

214 :returns: The warning collector instance.

215 """

216 global _threadLocalData

217

218 try:

219 self._parent = _threadLocalData.warningCollector

220 except AttributeError:

221 pass

222

223 _threadLocalData.warningCollector = self

224

225 return self

226

227 def __exit__(

228 self,

229 exc_type: Nullable[Type[BaseException]] = None,

230 exc_val: Nullable[BaseException] = None,

231 exc_tb: Nullable[TracebackType] = None

232 ) -> Nullable[bool]:

233 """

234 Exit the warning collector context.

235

236 :param exc_type: Exception type

237 :param exc_val: Exception instance

238 :param exc_tb: Exception's traceback.

239 :returns: ``None``

240 """

241 global _threadLocalData

242

243 _threadLocalData.warningCollector = self._parent

244

245 return False

246

247 @property

248 def Parent(self) -> Nullable[Self]:

249 """

250 Property to access the parent warning collector.

251

252 :returns: The parent warning collector or ``None``.

253 """

254 return self._parent

255

256 @Parent.setter

257 def Parent(self, value: Self) -> None:

258 self._parent = value

259

260 @readonly

261 def Warnings(self) -> List[Warning | CriticalWarning | Exception]:

262 """

263 Read-only property to access the list of collected warnings.

264

265 :returns: A list of collected warnings.

266 """

267 return self._warnings

268

269 def AddWarning(self, warning: Warning | CriticalWarning | Exception) -> bool:

270 """

271 Add a warning to the list of warnings managed by this warning collector.

272

273 :param warning: The warning to add to the collectors internal warning list.

274 :returns: Return ``True`` if the warning collector has a local handler callback and this handler returned

275 ``True``; otherwise ``False``.

276 :raises ValueError: If parameter ``warning`` is None.

277 :raises TypeError: If parameter ``warning`` is not of type :class:`Warning`.

278 """

279 if warning is None: 279 ↛ 280line 279 didn't jump to line 280 because the condition on line 279 was never true

280 raise ValueError("Parameter 'warning' is None.")

281 elif not isinstance(warning, (Warning, CriticalWarning, Exception)): 281 ↛ 282line 281 didn't jump to line 282 because the condition on line 281 was never true

282 ex = TypeError(f"Parameter 'warning' is not of type 'Warning', 'CriticalWarning' or 'Exception'.")

283 ex.add_note(f"Got type '{getFullyQualifiedName(warning)}'.")

284 raise ex

285

286 self._warnings.append(warning)

287

288 return False if self._handler is None else self._handler(warning)

289

290 @classmethod

291 def Raise(

292 cls,

293 warning: Warning | CriticalWarning | Exception,

294 cause: Nullable[Exception] = None,

295 *,

296 notes: Nullable[str | Iterable[str]] = None

297 ) -> None:

298 """

299 Walk the callstack frame by frame upwards and search for the first warning collector.

300

301 :param warning: Warning to send upwards in the call stack.

302 :param cause: Optional, root cause to be added to the warning.

303 :param notes: optional, a single note or a list of notes to be added to the warning.

304 :raises Exception: If warning should be converted to an exception.

305 :raises UnhandledExceptionException: If no warning collector was found along the call-hierarchy to collect and

306 handle an exception.

307 :raises UnhandledCriticalWarningException: If no warning collector was found along the call-hierarchy to collect and

308 handle a critical warning.

309 """

310 global _threadLocalData

311

312 if cause is not None: 312 ↛ 313line 312 didn't jump to line 313 because the condition on line 312 was never true

313 warning.__cause__ = cause

314

315 if notes is not None:

316 if isinstance(notes, str): 316 ↛ 317line 316 didn't jump to line 317 because the condition on line 316 was never true

317 warning.add_note(notes)

318 else:

319 for note in notes:

320 warning.add_note(note)

321

322 try:

323 warningCollector = _threadLocalData.warningCollector

324 if warningCollector.AddWarning(warning):

325 raise Exception(f"Warning: {warning}") from warning

326 except AttributeError:

327 ex = None

328 if isinstance(warning, Exception):

329 ex = UnhandledExceptionException(f"Unhandled Exception: {warning}")

330 elif isinstance(warning, CriticalWarning):

331 ex = UnhandledCriticalWarningException(f"Unhandled Critical Warning: {warning}")

332

333 if ex is not None:

334 ex.add_note(f"Add a 'with'-statement using '{cls.__name__}' somewhere up the call-hierarchy to receive and collect warnings.")

335 raise ex from warning

336

337

338@export

339class SupervisedWarningCollectorException(ExceptionBase):

340 pass

341

342

343@export

344class SupervisedWarningCollector(WarningCollector):

345 """

346 A context manager to collect warnings within the call hierarchy.

347 """

348 _supervisor: Nullable["ThreadSupervisor"]

349 _exceptionHandler: Nullable[Callable[[BaseException], bool]]

350 _finallyHandler: Nullable[Callable[[], None]]

351

352 __slots__ = ("_supervisor", "_exceptionHandler", "_finallyHandler")

353

354 def __init__(

355 self,

356 warnings: Nullable[List[BaseException]] = None,

357 handler: Nullable[Callable[[BaseException], bool]] = None,

358 /,

359 supervisor: Nullable["ThreadSupervisor"] = None,

360 exceptionHandler: Nullable[Callable[[BaseException], bool]] = None,

361 finallyHandler: Nullable[Callable[[], None]] = None

362 ) -> None:

363 """

364 Initializes a warning collector.

365

366 :param warnings: An optional reference to a list of warnings, which can be modified (appended) by this warning

367 collector. If ``None``, an internal list is created and can be referenced by the collector's

368 instance.

369 :param handler: An optional handler function, which processes the current warning and decides if a warning should

370 be reraised as an exception.

371 :raises TypeError: If optional parameter 'warnings' is not of type list.

372 :raises TypeError: If optional parameter 'handler' is not a callable.

373 """

374 super().__init__(warnings, handler)

375

376 self._supervisor = supervisor

377 self._exceptionHandler = exceptionHandler

378 self._finallyHandler = finallyHandler

379

380 def __enter__(self) -> Self:

381 """

382 Enter the warning collector context.

383

384 :returns: The warning collector instance.

385 """

386 global _threadLocalData

387

388 if hasattr(_threadLocalData, "warningCollector") and _threadLocalData.warningCollector is not None:

389 raise SupervisedWarningCollectorException("This warning collector is not the top-most warning collector within the current thread.")

390

391 _threadLocalData.warningCollector = self

392

393 return self

394

395 def __exit__(

396 self,

397 exc_type: Nullable[Type[BaseException]] = None,

398 exc_val: Nullable[BaseException] = None,

399 exc_tb: Nullable[TracebackType] = None

400 ) -> Nullable[bool]:

401 """

402 Exit the warning collector context.

403

404 :param exc_type: Exception type

405 :param exc_val: Exception instance

406 :param exc_tb: Exception's traceback.

407 :returns: ``None``

408 """

409 global _threadLocalData

410

411 _threadLocalData.warningCollector = None

412

413 if self._supervisor is not None:

414 result = True

415 if len(self._warnings) > 0:

416 self._supervisor.AddWarnings(self._warnings)

417

418 if exc_val is not None:

419 self._supervisor.AddException("", exc_val)

420

421 if self._exceptionHandler is not None:

422 result = self._exceptionHandler(exc_val)

423 else:

424 result = None

425

426 if self._finallyHandler is not None:

427 self._finallyHandler()

428

429 return result

430

431

432@export

433class SupervisedThreadException(ExceptionBase):

434 """

435 The exception is raise if a supervised thread received an unhandled exception which got collected by

436 :class:`ExceptionCollector`.

437 """

438 _threadName: str

439

440 def __init__(self, threadName: str, message: str, /, cause: Nullable[BaseException] = None) -> None:

441 super().__init__(message)

442 self._threadName = threadName

443 self.__cause__ = cause

444

445 @readonly

446 def ThreadName(self) -> str:

447 """

448 Read-only property to access the name of the thread that raised the exception (:attr:`_threadName`).

449

450 :returns: Name of the thread.

451 """

452 return self._threadName

453

454

455@export

456class ThreadSupervisor:

457 """

458 Thread-safe collector of exceptions and warnings raised in worker threads for surfacing on another thread.

459

460 This thread supervisor should be used in combination with :class:`WarningCollector` to accumulate exceptions

461 (:class:`BaseException`) and warnings (:class:`CriticalWarning` or :class:`Warning`).

462

463 .. code-block:: python

464

465 @export

466 class MyThread(Thread):

467 def __init__(

468 self,

469 threadSupervisor: ThreadSupervisor,

470 stopEvent: Event

471 ) -> None:

472 super().__init__(name="MyThread", daemon=True)

473

474 self._threadSupervisor = threadSupervisor

475 self._stopEvent = stopEvent

476

477 def run(self) -> None:

478 def exceptionHandler(ex: BaseException) -> None:

479 self._stopEvent.set()

480

481 def finallyHandler() -> None:

482 # some finally code

483

484 with SupervisedWarningCollector(

485 supervisor=self._threadSupervisor,

486 exceptionHandler=exceptionHandler,

487 finallyHandler=finallyHandler

488 ) as warnings:

489 # Thread body

490

491 .. code-block:: python

492

493 def RunVivadoPipeline(

494 self,

495 ) -> List[AnyWarning]:

496 stopEvent = Event()

497 threadSupervisor = ThreadSupervisor()

498

499 myThread = MyThread(threadSupervisor, stopEvent)

500 myThread.start()

501

502 try:

503 myThread.join()

504 except KeyboardInterrupt:

505 stopEvent.set()

506 myThread.join(timeout=2.0)

507 raise

508

509 threadSupervisor.ReRaise()

510

511 return threadSupervisor.Warnings

512 """

513

514 _lock: Lock

515 _exceptions: List[Tuple[str, BaseException]]

516 _warnings: List[Tuple[str, AnyWarning]]

517

518 __slots__ = ("_lock", "_exceptions", "_warnings")

519

520 def __init__(self) -> None:

521 self._lock = Lock()

522 self._exceptions = []

523 self._warnings = []

524

525 @readonly

526 def HasWarning(self) -> bool:

527 """

528 Check if at least one warning was collected from a supervised thread.

529

530 :returns: ``True``, if at least one warning was collected.

531 """

532 with self._lock:

533 return len(self._warnings) > 0

534

535 @readonly

536 def HasExceptions(self) -> bool:

537 """

538 Check if at least one exception was collected from a supervised thread.

539

540 :returns: ``True``, if at least one exception was collected.

541 """

542 with self._lock:

543 return len(self._exceptions) > 0

544

545 @readonly

546 def Warnings(self) -> List[AnyWarning]:

547 """

548 Read-only property to return all warnings collected from supervised threads (:attr:`_warnings`).

549

550 :returns: List of collected warnings, without their thread names.

551 """

552 with self._lock:

553 return [warning for _, warning in self._warnings]

554

555 def AddWarning(self, threadName: str, warning: AnyWarning) -> None:

556 with self._lock:

557 self._warnings.append((threadName, warning))

558

559 def AddWarnings(self, threadName: str, warnings: List[AnyWarning]) -> None:

560 with self._lock:

561 self._warnings.extend((threadName, warning) for warning in warnings)

562

563 def AddException(self, threadName: str, ex: BaseException) -> None:

564 with self._lock:

565 self._exceptions.append((threadName, ex))

566

567 def ReRaise(self, unwrapped: bool = False) -> None:

568 with self._lock:

569 if len(self._exceptions) == 0:

570 return

571

572 exceptions = list(self._exceptions)

573

574 if len(exceptions) == 1:

575 threadName, ex = exceptions[0]

576 if unwrapped:

577 raise ex

578 else:

579 raise SupervisedThreadException(threadName, f"Thread '{threadName}' failed.") from ex

580

581 elif unwrapped:

582 raise ExceptionGroup(

583 "Multiple threads failed.",

584 [ex for _, ex in exceptions]

585 )

586 else:

587 raise ExceptionGroup(

588 "Multiple threads failed.",

589 [SupervisedThreadException(f"Thread '{threadName}' failed.", cause=ex) for threadName, ex in exceptions]

590 )

Coverage for pyTooling/Warning/init.py: 58%

185 statements