(
self, async_session: AsyncSession, metadata, use_scalar
)
| 102 | @testing.combinations((True,), (False,), argnames="use_scalar") |
| 103 | @testing.requires.sequences |
| 104 | async def test_sequence_execute( |
| 105 | self, async_session: AsyncSession, metadata, use_scalar |
| 106 | ): |
| 107 | seq = normalize_sequence( |
| 108 | config, Sequence("some_sequence", metadata=metadata) |
| 109 | ) |
| 110 | |
| 111 | sync_connection = (await async_session.connection()).sync_connection |
| 112 | |
| 113 | await (await async_session.connection()).run_sync(metadata.create_all) |
| 114 | |
| 115 | if use_scalar: |
| 116 | eq_( |
| 117 | await async_session.scalar(seq), |
| 118 | sync_connection.dialect.default_sequence_base, |
| 119 | ) |
| 120 | else: |
| 121 | with expect_deprecated( |
| 122 | r"Using the .execute\(\) method to invoke a " |
| 123 | r"DefaultGenerator object is deprecated; please use " |
| 124 | r"the .scalar\(\) method." |
| 125 | ): |
| 126 | eq_( |
| 127 | await async_session.execute(seq), |
| 128 | sync_connection.dialect.default_sequence_base, |
| 129 | ) |
| 130 | |
| 131 | @async_test |
| 132 | async def test_close_all(self, async_engine): |
nothing calls this directly
no test coverage detected