{{ message }}
Repository navigation
Expand file tree
/
Copy pathtest_transaction.py
More file actions
327 lines (262 loc) · 12.7 KB
/
Copy pathtest_transaction.py
File metadata and controls
327 lines (262 loc) · 12.7 KB
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
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
# Copyright 2010-2026 The pygit2 contributors
#
# This file is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License, version 2,
# as published by the Free Software Foundation.
#
# In addition to the permissions in the GNU General Public License,
# the authors give you unlimited permission to link the compiled
# version of this file into combinations with other programs,
# and to distribute those combinations without any restriction
# coming from the use of this file. (The General Public License
# restrictions do apply in other respects; for example, they cover
# modification of the file, and distribution when not linked into
# a combined executable.)
#
# This file is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
# General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; see the file COPYING. If not, write to
# the Free Software Foundation, 51 Franklin Street, Fifth Floor,
# Boston, MA 02110-1301, USA.
import threading
import pytest
from pygit2 import GitError, Oid, Repository
from pygit2.transaction import ReferenceTransaction
def test_transaction_context_manager(testrepo: Repository) -> None:
"""Test basic transaction with context manager."""
master_ref = testrepo.lookup_reference('refs/heads/master')
assert str(master_ref.target) == '2be5719152d4f82c7302b1c0932d8e5f0a4a0e98'
# Create a transaction and update a ref
new_target = Oid(hex='5ebeeebb320790caf276b9fc8b24546d63316533')
with testrepo.transaction() as txn:
txn.lock_ref('refs/heads/master')
txn.set_target('refs/heads/master', new_target, message='Test update')
# Verify the update was applied
master_ref = testrepo.lookup_reference('refs/heads/master')
assert master_ref.target == new_target
def test_transaction_rollback_on_exception(testrepo: Repository) -> None:
"""Test that transaction rolls back when exception is raised."""
master_ref = testrepo.lookup_reference('refs/heads/master')
original_target = master_ref.target
new_target = Oid(hex='5ebeeebb320790caf276b9fc8b24546d63316533')
# Transaction should not commit if exception is raised
with pytest.raises(RuntimeError):
with testrepo.transaction() as txn:
txn.lock_ref('refs/heads/master')
txn.set_target('refs/heads/master', new_target, message='Test update')
raise RuntimeError('Abort transaction')
# Verify the update was NOT applied
master_ref = testrepo.lookup_reference('refs/heads/master')
assert master_ref.target == original_target
def test_transaction_multiple_refs(testrepo: Repository) -> None:
"""Test updating multiple refs in a single transaction."""
master_ref = testrepo.lookup_reference('refs/heads/master')
i18n_ref = testrepo.lookup_reference('refs/heads/i18n')
new_master = Oid(hex='5ebeeebb320790caf276b9fc8b24546d63316533')
new_i18n = Oid(hex='2be5719152d4f82c7302b1c0932d8e5f0a4a0e98')
with testrepo.transaction() as txn:
txn.lock_ref('refs/heads/master')
txn.lock_ref('refs/heads/i18n')
txn.set_target('refs/heads/master', new_master, message='Update master')
txn.set_target('refs/heads/i18n', new_i18n, message='Update i18n')
# Verify both updates were applied
master_ref = testrepo.lookup_reference('refs/heads/master')
i18n_ref = testrepo.lookup_reference('refs/heads/i18n')
assert master_ref.target == new_master
assert i18n_ref.target == new_i18n
def test_transaction_symbolic_ref(testrepo: Repository) -> None:
"""Test updating symbolic reference in transaction."""
with testrepo.transaction() as txn:
txn.lock_ref('HEAD')
txn.set_symbolic_target('HEAD', 'refs/heads/i18n', message='Switch HEAD')
head = testrepo.lookup_reference('HEAD')
assert head.target == 'refs/heads/i18n'
# Restore HEAD to master
with testrepo.transaction() as txn:
txn.lock_ref('HEAD')
txn.set_symbolic_target('HEAD', 'refs/heads/master', message='Restore HEAD')
def test_transaction_remove_ref(testrepo: Repository) -> None:
"""Test removing a reference in a transaction."""
# Create a test ref
test_ref_name = 'refs/heads/test-transaction-delete'
target = Oid(hex='5ebeeebb320790caf276b9fc8b24546d63316533')
testrepo.create_reference(test_ref_name, target)
# Verify it exists
assert test_ref_name in testrepo.references
# Remove it in a transaction
with testrepo.transaction() as txn:
txn.lock_ref(test_ref_name)
txn.remove(test_ref_name)
# Verify it's gone
assert test_ref_name not in testrepo.references
def test_transaction_error_without_lock(testrepo: Repository) -> None:
"""Test that setting target without lock raises error."""
new_target = Oid(hex='5ebeeebb320790caf276b9fc8b24546d63316533')
with pytest.raises(KeyError, match='not locked'):
with testrepo.transaction() as txn:
# Try to set target without locking first
txn.set_target('refs/heads/master', new_target, message='Should fail')
def test_transaction_isolated_across_threads(testrepo: Repository) -> None:
"""Test that transactions from different threads are isolated."""
# Create two test refs
ref1_name = 'refs/heads/thread-test-1'
ref2_name = 'refs/heads/thread-test-2'
target1 = Oid(hex='5ebeeebb320790caf276b9fc8b24546d63316533')
target2 = Oid(hex='2be5719152d4f82c7302b1c0932d8e5f0a4a0e98')
testrepo.create_reference(ref1_name, target1)
testrepo.create_reference(ref2_name, target2)
results = []
errors = []
thread1_ref1_locked = threading.Event()
thread2_ref2_locked = threading.Event()
def update_ref1() -> None:
try:
with testrepo.transaction() as txn:
txn.lock_ref(ref1_name)
thread1_ref1_locked.set()
thread2_ref2_locked.wait(timeout=5)
txn.set_target(ref1_name, target2, message='Thread 1 update')
results.append('thread1_success')
except Exception as e:
errors.append(('thread1', str(e)))
def update_ref2() -> None:
try:
with testrepo.transaction() as txn:
txn.lock_ref(ref2_name)
thread2_ref2_locked.set()
thread1_ref1_locked.wait(timeout=5)
txn.set_target(ref2_name, target1, message='Thread 2 update')
results.append('thread2_success')
except Exception as e:
errors.append(('thread2', str(e)))
thread1 = threading.Thread(target=update_ref1)
thread2 = threading.Thread(target=update_ref2)
thread1.start()
thread2.start()
thread1.join()
thread2.join()
# Both threads should succeed - transactions are isolated
assert len(errors) == 0, f'Errors: {errors}'
assert 'thread1_success' in results
assert 'thread2_success' in results
# Verify both updates were applied
ref1 = testrepo.lookup_reference(ref1_name)
ref2 = testrepo.lookup_reference(ref2_name)
assert str(ref1.target) == '2be5719152d4f82c7302b1c0932d8e5f0a4a0e98'
assert str(ref2.target) == '5ebeeebb320790caf276b9fc8b24546d63316533'
def test_transaction_deadlock_prevention(testrepo: Repository) -> None:
"""Test that acquiring locks in different order raises error instead of deadlock."""
# Create two test refs
ref1_name = 'refs/heads/deadlock-test-1'
ref2_name = 'refs/heads/deadlock-test-2'
target = Oid(hex='5ebeeebb320790caf276b9fc8b24546d63316533')
testrepo.create_reference(ref1_name, target)
testrepo.create_reference(ref2_name, target)
thread1_ref1_locked = threading.Event()
thread2_ref2_locked = threading.Event()
errors = []
successes = []
def thread1_task() -> None:
try:
with testrepo.transaction() as txn:
txn.lock_ref(ref1_name)
thread1_ref1_locked.set()
thread2_ref2_locked.wait(timeout=5)
# this would cause a deadlock, so will throw (GitError)
txn.lock_ref(ref2_name)
# shouldn't get here
successes.append('thread1')
except Exception as e:
errors.append(('thread1', type(e).__name__, str(e)))
def thread2_task() -> None:
try:
with testrepo.transaction() as txn:
txn.lock_ref(ref2_name)
thread2_ref2_locked.set()
thread1_ref1_locked.wait(timeout=5)
# this would cause a deadlock, so will throw (GitError)
txn.lock_ref(ref2_name)
# shouldn't get here
successes.append('thread2')
except Exception as e:
errors.append(('thread2', type(e).__name__, str(e)))
thread1 = threading.Thread(target=thread1_task)
thread2 = threading.Thread(target=thread2_task)
thread1.start()
thread2.start()
thread1.join(timeout=5)
thread2.join(timeout=5)
# At least one thread should fail with an error (not deadlock)
# If both threads are still alive, we have a deadlock
assert not thread1.is_alive(), 'Thread 1 deadlocked'
assert not thread2.is_alive(), 'Thread 2 deadlocked'
# Both can't succeed.
# libgit2 doesn't *wait* for locks, so it's possible for neither to succeed
# if they both try to take the second lock at basically the same time.
# The other possibility is that one thread throws, exits its transaction,
# and the other thread is able to acquire the second lock.
assert len(successes) <= 1 and len(errors) >= 1, (
f'Successes: {successes}; errors: {errors}'
)
def test_transaction_commit_from_wrong_thread(testrepo: Repository) -> None:
"""Test that committing a transaction from wrong thread raises error."""
txn: ReferenceTransaction | None = None
def create_transaction() -> None:
nonlocal txn
txn = testrepo.transaction().__enter__()
ref_name = 'refs/heads/wrong-thread-test'
target = Oid(hex='5ebeeebb320790caf276b9fc8b24546d63316533')
testrepo.create_reference(ref_name, target)
txn.lock_ref(ref_name)
# Create transaction in thread 1
thread = threading.Thread(target=create_transaction)
thread.start()
thread.join()
assert txn is not None
with pytest.raises(RuntimeError):
# Try to commit from main thread (different from creator) doesn't cause libgit2 to crash,
# it raises an exception instead
txn.commit()
def test_transaction_nested_same_thread(testrepo: Repository) -> None:
"""Test that two concurrent transactions from same thread work with different refs."""
# Create test refs
ref1_name = 'refs/heads/nested-test-1'
ref2_name = 'refs/heads/nested-test-2'
target1 = Oid(hex='5ebeeebb320790caf276b9fc8b24546d63316533')
target2 = Oid(hex='2be5719152d4f82c7302b1c0932d8e5f0a4a0e98')
testrepo.create_reference(ref1_name, target1)
testrepo.create_reference(ref2_name, target2)
# Nested transactions should work as long as they don't conflict
with testrepo.transaction() as txn1:
txn1.lock_ref(ref1_name)
with testrepo.transaction() as txn2:
txn2.lock_ref(ref2_name)
txn2.set_target(ref2_name, target1, message='Inner transaction')
# Inner transaction committed, now update outer
txn1.set_target(ref1_name, target2, message='Outer transaction')
# Both updates should have been applied
ref1 = testrepo.lookup_reference(ref1_name)
ref2 = testrepo.lookup_reference(ref2_name)
assert str(ref1.target) == '2be5719152d4f82c7302b1c0932d8e5f0a4a0e98'
assert str(ref2.target) == '5ebeeebb320790caf276b9fc8b24546d63316533'
def test_transaction_nested_same_ref_conflict(testrepo: Repository) -> None:
"""Test that nested transactions fail when trying to lock the same ref."""
ref_name = 'refs/heads/nested-conflict-test'
target = Oid(hex='5ebeeebb320790caf276b9fc8b24546d63316533')
new_target = Oid(hex='2be5719152d4f82c7302b1c0932d8e5f0a4a0e98')
testrepo.create_reference(ref_name, target)
with testrepo.transaction() as txn1:
txn1.lock_ref(ref_name)
# Inner transaction should fail to lock the same ref
with pytest.raises(GitError):
with testrepo.transaction() as txn2:
txn2.lock_ref(ref_name)
# Outer transaction should still be able to complete
txn1.set_target(ref_name, new_target, message='Outer transaction')
# Outer transaction's update should have been applied
ref = testrepo.lookup_reference(ref_name)
assert ref.target == new_target
You can’t perform that action at this time.