(self)
| 15 | class TestQueryLogging(tb.ConnectedTestCase): |
| 16 | |
| 17 | async def test_logging_context(self): |
| 18 | queries = asyncio.Queue() |
| 19 | |
| 20 | def query_saver(record): |
| 21 | queries.put_nowait(record) |
| 22 | |
| 23 | log = LogCollector() |
| 24 | |
| 25 | with self.con.query_logger(query_saver): |
| 26 | self.assertEqual(len(self.con._query_loggers), 1) |
| 27 | await self.con.execute("SELECT 1") |
| 28 | with self.con.query_logger(log): |
| 29 | self.assertEqual(len(self.con._query_loggers), 2) |
| 30 | await self.con.execute("SELECT 2") |
| 31 | |
| 32 | r1 = await queries.get() |
| 33 | r2 = await queries.get() |
| 34 | self.assertEqual(r1.query, "SELECT 1") |
| 35 | self.assertEqual(r2.query, "SELECT 2") |
| 36 | self.assertEqual(len(log.records), 1) |
| 37 | self.assertEqual(log.records[0].query, "SELECT 2") |
| 38 | self.assertEqual(len(self.con._query_loggers), 0) |
| 39 | |
| 40 | async def test_error_logging(self): |
| 41 | log = LogCollector() |
nothing calls this directly
no test coverage detected