(self)
| 472 | |
| 473 | @tb.with_connection_options(record_class=CustomRecord) |
| 474 | async def test_record_subclass_04(self): |
| 475 | r = await self.con.fetchrow( |
| 476 | "SELECT 1 as a, '2' as b", |
| 477 | record_class=asyncpg.Record, |
| 478 | ) |
| 479 | self.assertIs(type(r), asyncpg.Record) |
| 480 | |
| 481 | r = await self.con.fetch( |
| 482 | "SELECT 1 as a, '2' as b", |
| 483 | record_class=asyncpg.Record, |
| 484 | ) |
| 485 | self.assertIs(type(r[0]), asyncpg.Record) |
| 486 | |
| 487 | async with self.con.transaction(): |
| 488 | cur = await self.con.cursor( |
| 489 | "SELECT 1 as a, '2' as b", |
| 490 | record_class=asyncpg.Record, |
| 491 | ) |
| 492 | r = await cur.fetchrow() |
| 493 | self.assertIs(type(r), asyncpg.Record) |
| 494 | |
| 495 | cur = await self.con.cursor( |
| 496 | "SELECT 1 as a, '2' as b", |
| 497 | record_class=asyncpg.Record, |
| 498 | ) |
| 499 | r = await cur.fetch(1) |
| 500 | self.assertIs(type(r[0]), asyncpg.Record) |
| 501 | |
| 502 | async with self.con.transaction(): |
| 503 | cur = self.con.cursor( |
| 504 | "SELECT 1 as a, '2' as b", |
| 505 | record_class=asyncpg.Record, |
| 506 | ) |
| 507 | async for r in cur: |
| 508 | self.assertIs(type(r), asyncpg.Record) |
| 509 | |
| 510 | ps = await self.con.prepare( |
| 511 | "SELECT 1 as a, '2' as b", |
| 512 | record_class=asyncpg.Record, |
| 513 | ) |
| 514 | r = await ps.fetchrow() |
| 515 | self.assertIs(type(r), asyncpg.Record) |
| 516 | |
| 517 | r = await ps.fetch() |
| 518 | self.assertIs(type(r[0]), asyncpg.Record) |
| 519 | |
| 520 | async def test_record_subclass_05(self): |
| 521 | class MyRecord(asyncpg.Record): |
nothing calls this directly
no test coverage detected