(self, level, tag, expected, *, skip=False)
| 62 | self.assert_log(level, tag, line, **kwargs) |
| 63 | |
| 64 | def assert_log(self, level, tag, expected, *, skip=False): |
| 65 | deadline = time() + LOOPBACK_TIMEOUT |
| 66 | while True: |
| 67 | try: |
| 68 | line = self.logcat_queue.get(timeout=(deadline - time())) |
| 69 | except queue.Empty: |
| 70 | raise self.failureException( |
| 71 | f"line not found: {expected!r}" |
| 72 | ) from None |
| 73 | if match := re.fullmatch(fr"(.)/{tag}: (.*)", line): |
| 74 | try: |
| 75 | self.assertEqual(level, match[1]) |
| 76 | self.assertEqual(expected, match[2]) |
| 77 | break |
| 78 | except AssertionError: |
| 79 | if not skip: |
| 80 | raise |
| 81 | |
| 82 | def tearDown(self): |
| 83 | self.logcat_process.terminate() |
no test coverage detected