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