|
1 | 1 | from __future__ import annotations
|
2 | 2 |
|
3 |
| -import typing |
4 |
| - |
5 | 3 | import pytest
|
6 | 4 |
|
7 | 5 | from psqlpy import Cursor, IsolationLevel, PSQLPool, ReadVariant
|
8 | 6 | from psqlpy.exceptions import DBTransactionError, RustPSQLDriverPyBaseError
|
9 | 7 |
|
10 | 8 |
|
11 | 9 | @pytest.mark.anyio()
|
| 10 | +@pytest.mark.parametrize( |
| 11 | + ("isolation_level", "deferrable", "read_variant"), |
| 12 | + [ |
| 13 | + (None, None, None), |
| 14 | + (IsolationLevel.ReadCommitted, True, ReadVariant.ReadOnly), |
| 15 | + (IsolationLevel.ReadUncommitted, False, ReadVariant.ReadWrite), |
| 16 | + (IsolationLevel.RepeatableRead, True, ReadVariant.ReadOnly), |
| 17 | + (IsolationLevel.Serializable, False, ReadVariant.ReadWrite), |
| 18 | + ], |
| 19 | +) |
12 | 20 | async def test_transaction_init_parameters(
|
13 | 21 | psql_pool: PSQLPool,
|
14 | 22 | table_name: str,
|
| 23 | + isolation_level: IsolationLevel | None, |
| 24 | + deferrable: bool | None, |
| 25 | + read_variant: ReadVariant | None, |
15 | 26 | ) -> None:
|
16 | 27 | connection = await psql_pool.connection()
|
17 |
| - |
18 |
| - test_init_parameters: typing.Final[list[dict[str, typing.Any]]] = [ |
19 |
| - {"isolation_level": None, "deferrable": None, "read_variant": None}, |
20 |
| - { |
21 |
| - "isolation_level": IsolationLevel.ReadCommitted, |
22 |
| - "deferrable": True, |
23 |
| - "read_variant": ReadVariant.ReadOnly, |
24 |
| - }, |
25 |
| - { |
26 |
| - "isolation_level": IsolationLevel.ReadUncommitted, |
27 |
| - "deferrable": False, |
28 |
| - "read_variant": ReadVariant.ReadWrite, |
29 |
| - }, |
30 |
| - { |
31 |
| - "isolation_level": IsolationLevel.RepeatableRead, |
32 |
| - "deferrable": True, |
33 |
| - "read_variant": ReadVariant.ReadOnly, |
34 |
| - }, |
35 |
| - { |
36 |
| - "isolation_level": IsolationLevel.Serializable, |
37 |
| - "deferrable": False, |
38 |
| - "read_variant": ReadVariant.ReadWrite, |
39 |
| - }, |
40 |
| - ] |
41 |
| - for init_parameters in test_init_parameters: |
42 |
| - insert_exception = None |
43 |
| - async with connection.transaction( |
44 |
| - isolation_level=init_parameters.get("isolation_level"), |
45 |
| - deferrable=init_parameters.get("deferrable"), |
46 |
| - read_variant=init_parameters.get("read_variant"), |
47 |
| - ) as transaction: |
48 |
| - await transaction.execute("SELECT 1") |
49 |
| - try: |
50 |
| - await transaction.execute( |
51 |
| - f"INSERT INTO {table_name} VALUES ($1, $2)", |
52 |
| - parameters=[100, "test_name"], |
53 |
| - ) |
54 |
| - except RustPSQLDriverPyBaseError as exception: |
55 |
| - insert_exception = exception |
56 |
| - |
57 |
| - assert ( |
58 |
| - insert_exception is None |
59 |
| - or init_parameters.get("read_variant") is ReadVariant.ReadOnly |
| 28 | + async with connection.transaction( |
| 29 | + isolation_level=isolation_level, |
| 30 | + deferrable=deferrable, |
| 31 | + read_variant=read_variant, |
| 32 | + ) as transaction: |
| 33 | + await transaction.execute("SELECT 1") |
| 34 | + try: |
| 35 | + await transaction.execute( |
| 36 | + f"INSERT INTO {table_name} VALUES ($1, $2)", |
| 37 | + parameters=[100, "test_name"], |
60 | 38 | )
|
| 39 | + except RustPSQLDriverPyBaseError: |
| 40 | + assert read_variant is ReadVariant.ReadOnly |
61 | 41 |
|
62 | 42 |
|
63 | 43 | @pytest.mark.anyio()
|
|
0 commit comments