(
self,
emitter: EventEmitter,
event: str,
error: Union[Error, Callable[..., Error]],
predicate: Callable = None,
)
| 63 | ) |
| 64 | |
| 65 | def reject_on_event( |
| 66 | self, |
| 67 | emitter: EventEmitter, |
| 68 | event: str, |
| 69 | error: Union[Error, Callable[..., Error]], |
| 70 | predicate: Callable = None, |
| 71 | ) -> None: |
| 72 | def on_match() -> None: |
| 73 | self._reject(error() if callable(error) else error) |
| 74 | |
| 75 | def listener(event_data: Any = None) -> None: |
| 76 | self._evaluate_predicate(predicate, event_data, on_match) |
| 77 | |
| 78 | emitter.on(event, listener) |
| 79 | self._registered_listeners.append((emitter, event, listener)) |
| 80 | |
| 81 | def reject_on_timeout(self, timeout: float, message: str) -> None: |
| 82 | if timeout == 0: |
no test coverage detected