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
|
from django.test import TestCase
from cacheops import invalidate_all
from cacheops.transaction import transaction_states
class BaseTestCase(TestCase):
def setUp(self):
# Emulate not being in transaction by tricking system to ignore its pretest level.
# TestCase wraps each test into 1 or 2 transaction(s) altering cacheops behavior.
# The alternative is using TransactionTestCase, which is 10x slow.
from funcy import empty
transaction_states._states, self._states \
= empty(transaction_states._states), transaction_states._states
invalidate_all()
def tearDown(self):
transaction_states._states = self._states
def make_inc(deco=lambda x: x):
calls = [0]
@deco
def inc(_=None, **kw):
calls[0] += 1
return calls[0]
inc.get = lambda: calls[0]
return inc
# Thread utilities
from threading import Thread
class ThreadWithReturnValue(Thread):
def __init__(self, *args, **kwargs):
super(ThreadWithReturnValue, self).__init__(*args, **kwargs)
self._return = None
self._exc = None
def run(self):
try:
self._return = self._target(*self._args, **self._kwargs)
except Exception as e:
self._exc = e
finally:
# Django does not drop postgres connections opened in new threads.
# This leads to postgres complaining about db accessed when we try to destroy it.
# See https://code.djangoproject.com/ticket/22420#comment:18
from django.db import connection
connection.close()
def join(self, *args, **kwargs):
super(ThreadWithReturnValue, self).join(*args, **kwargs)
if self._exc:
raise self._exc
return self._return
def run_in_thread(target):
t = ThreadWithReturnValue(target=target)
t.start()
return t.join()
|
