PySqlit/tests/test_transaction.py at master · Python51888/PySqlit · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
"""Unit tests for pysqlit/transaction.py module."""

import pytest
import tempfile
import os

from pysqlit.transaction import TransactionManager, IsolationLevel


class TestIsolationLevel:
    """Test cases for IsolationLevel enum."""

    def test_isolation_level_values(self):
        """Test isolation level enum values."""
        assert IsolationLevel.READ_UNCOMMITTED.value == "READ_UNCOMMITTED"
        assert IsolationLevel.READ_COMMITTED.value == "READ_COMMITTED"
        assert IsolationLevel.REPEATABLE_READ.value == "REPEATABLE_READ"
        assert IsolationLevel.SERIALIZABLE.value == "SERIALIZABLE"


class TestTransactionManager:
    """Test cases for TransactionManager class."""

    def test_transaction_manager_creation(self, temp_db_path):
        """Test transaction manager creation."""
        from pysqlit.concurrent_storage import ConcurrentPager

        pager = ConcurrentPager(temp_db_path)
        manager = TransactionManager(pager)
        assert manager.pager == pager
        pager.close()

    def test_begin_transaction(self, temp_db_path):
        """Test beginning transaction."""
        from pysqlit.concurrent_storage import ConcurrentPager

        pager = ConcurrentPager(temp_db_path)
        manager = TransactionManager(pager)

        tx_id = manager.begin_transaction()
        assert isinstance(tx_id, int)
        assert tx_id > 0

        pager.close()

    def test_begin_transaction_with_isolation_level(self, temp_db_path):
        """Test beginning transaction with specific isolation level."""
        from pysqlit.concurrent_storage import ConcurrentPager

        pager = ConcurrentPager(temp_db_path)
        manager = TransactionManager(pager)

        tx_id = manager.begin_transaction(IsolationLevel.READ_COMMITTED)
        assert isinstance(tx_id, int)

        pager.close()

    def test_commit_transaction(self, temp_db_path):
        """Test committing transaction."""
        from pysqlit.concurrent_storage import ConcurrentPager

        pager = ConcurrentPager(temp_db_path)
        manager = TransactionManager(pager)

        tx_id = manager.begin_transaction()
        manager.commit_transaction(tx_id)

        pager.close()

    def test_rollback_transaction(self, temp_db_path):
        """Test rolling back transaction."""
        from pysqlit.concurrent_storage import ConcurrentPager

        pager = ConcurrentPager(temp_db_path)
        manager = TransactionManager(pager)

        tx_id = manager.begin_transaction()
        manager.rollback_transaction(tx_id)

        pager.close()

    def test_is_in_transaction(self, temp_db_path):
        """Test checking if in transaction."""
        from pysqlit.concurrent_storage import ConcurrentPager

        pager = ConcurrentPager(temp_db_path)
        manager = TransactionManager(pager)

        tx_id = manager.begin_transaction()
        assert manager.is_in_transaction(tx_id) is True

        manager.commit_transaction(tx_id)
        assert manager.is_in_transaction(tx_id) is False

        pager.close()

    def test_get_active_transaction_count(self, temp_db_path):
        """Test getting active transaction count."""
        from pysqlit.concurrent_storage import ConcurrentPager

        pager = ConcurrentPager(temp_db_path)
        manager = TransactionManager(pager)

        assert manager.get_active_transaction_count() == 0

        tx_id1 = manager.begin_transaction()
        assert manager.get_active_transaction_count() == 1

        tx_id2 = manager.begin_transaction()
        assert manager.get_active_transaction_count() == 2

        manager.commit_transaction(tx_id1)
        assert manager.get_active_transaction_count() == 1

        manager.rollback_transaction(tx_id2)
        assert manager.get_active_transaction_count() == 0

        pager.close()

    def test_multiple_transactions(self, temp_db_path):
        """Test multiple concurrent transactions."""
        from pysqlit.concurrent_storage import ConcurrentPager

        pager = ConcurrentPager(temp_db_path)
        manager = TransactionManager(pager)

        tx_ids = []
        for i in range(5):
            tx_id = manager.begin_transaction()
            tx_ids.append(tx_id)

        assert len(tx_ids) == 5
        assert len(set(tx_ids)) == 5  # All IDs should be unique

        for tx_id in tx_ids:
            manager.commit_transaction(tx_id)

        pager.close()

    def test_transaction_isolation_levels(self, temp_db_path):
        """Test different isolation levels."""
        from pysqlit.concurrent_storage import ConcurrentPager

        pager = ConcurrentPager(temp_db_path)
        manager = TransactionManager(pager)

        levels = [
            IsolationLevel.READ_UNCOMMITTED,
            IsolationLevel.READ_COMMITTED,
            IsolationLevel.REPEATABLE_READ,
            IsolationLevel.SERIALIZABLE
        ]

        for level in levels:
            tx_id = manager.begin_transaction(level)
            assert tx_id > 0
            manager.commit_transaction(tx_id)

        pager.close()