0001-tests-prevent-Database-is-locked-error-during-concur.patch
tests/test_default_adapter.py | ||
---|---|---|
1 |
import threading |
|
2 | 1 |
import pytest |
3 | 2 |
import re |
4 | 3 |
import lasso |
4 |
from multiprocessing.pool import ThreadPool |
|
5 | 5 | |
6 | 6 |
from django.contrib import auth |
7 | 7 |
from django.db import connection |
... | ... | |
57 | 57 | |
58 | 58 |
def test_lookup_user_transaction(transactional_db, concurrency): |
59 | 59 |
adapter = DefaultAdapter() |
60 | ||
61 |
def map_threads(f, l): |
|
62 |
threads = [] |
|
63 |
for i in l: |
|
64 |
threads.append(threading.Thread(target=f, args=(i,))) |
|
65 |
threads[-1].start() |
|
66 |
for thread in threads: |
|
67 |
thread.join() |
|
68 |
users = [] |
|
60 |
p = ThreadPool(concurrency) |
|
69 | 61 | |
70 | 62 |
def f(i): |
71 |
users.append(adapter.lookup_user(idp, saml_attributes)) |
|
72 |
connection.close() |
|
73 |
map_threads(f, range(concurrency)) |
|
63 |
# sqlite has a default lock timeout of 5s seconds between different access to the same in |
|
64 |
# memory DB |
|
65 |
if connection.vendor == 'sqlite': |
|
66 |
connection.cursor().execute('PRAGMA busy_timeout = 400000') |
|
67 |
try: |
|
68 |
return adapter.lookup_user(idp, saml_attributes) |
|
69 |
finally: |
|
70 |
connection.close() |
|
71 |
users = p.map(f, range(concurrency)) |
|
72 | ||
74 | 73 |
assert len(users) == concurrency |
75 | 74 |
assert len(set(user.pk for user in users)) == 1 |
76 | 75 | |
77 |
- |