(self, async_engine, case: testing.Variation)
| 1490 | @async_test |
| 1491 | @testing.variation("case", ["scalar_one", "scalar_one_or_none", "scalar"]) |
| 1492 | async def test_stream_scalar(self, async_engine, case: testing.Variation): |
| 1493 | users = self.tables.users |
| 1494 | async with async_engine.connect() as conn: |
| 1495 | result = await conn.stream( |
| 1496 | select(users).limit(1).order_by(users.c.user_name) |
| 1497 | ) |
| 1498 | |
| 1499 | if case.scalar_one: |
| 1500 | u1 = await result.scalar_one() |
| 1501 | elif case.scalar_one_or_none: |
| 1502 | u1 = await result.scalar_one_or_none() |
| 1503 | elif case.scalar: |
| 1504 | u1 = await result.scalar() |
| 1505 | else: |
| 1506 | case.fail() |
| 1507 | |
| 1508 | eq_(u1, 1) |
| 1509 | |
| 1510 | |
| 1511 | class TextSyncDBAPI(fixtures.TestBase): |
nothing calls this directly
no test coverage detected