-
Notifications
You must be signed in to change notification settings - Fork 28
Expand file tree
/
Copy pathtest_SQLRelatedJoin.py
More file actions
114 lines (91 loc) · 3.2 KB
/
test_SQLRelatedJoin.py
File metadata and controls
114 lines (91 loc) · 3.2 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
import pytest
from sqlobject import RelatedJoin, SQLObject, SQLRelatedJoin, StringCol, \
ForeignKey
from sqlobject.sqlbuilder import Alias
from sqlobject.tests.dbtest import setupClass, supports
class Fighter(SQLObject):
class sqlmeta:
idName = 'fighter_id' # test on a non-standard way
name = StringCol()
tourtments = RelatedJoin('Tourtment')
class Tourtment(SQLObject):
class sqlmeta:
table = 'competition' # test on a non-standard way
name = StringCol()
fightersAsList = RelatedJoin('Fighter')
fightersAsSResult = SQLRelatedJoin('Fighter')
def createAllTables():
setupClass(Fighter)
setupClass(Tourtment)
def createData():
createAllTables()
# create some tourtments
t1 = Tourtment(name='Tourtment #1')
t2 = Tourtment(name='Tourtment #2')
t3 = Tourtment(name='Tourtment #3')
# create some fighters
gokou = Fighter(name='gokou')
vegeta = Fighter(name='vegeta')
gohan = Fighter(name='gohan')
trunks = Fighter(name='trunks')
# relating them
t1.addFighter(gokou)
t1.addFighter(vegeta)
t1.addFighter(gohan)
t2.addFighter(gokou)
t2.addFighter(vegeta)
t2.addFighter(trunks)
t3.addFighter(gohan)
t3.addFighter(trunks)
return t1, t2, t3, gokou, vegeta, gohan, trunks
def test_1():
t1, t2, t3, gokou, vegeta, gohan, trunks = createData()
for i, j in zip(t1.fightersAsList, t1.fightersAsSResult):
assert i is j
assert len(t2.fightersAsList) == t2.fightersAsSResult.count()
def test_related_join_transaction():
if not supports('transactions'):
pytest.skip("Transactions aren't supported")
createAllTables()
trans = Tourtment._connection.transaction()
try:
t1 = Tourtment(name='Tourtment #1', connection=trans)
t1.addFighter(Fighter(name='Jim', connection=trans))
assert t1.fightersAsSResult.count() == 1
assert t1.fightersAsSResult[0]._connection == trans
finally:
trans.commit(True)
Tourtment._connection.autoCommit = True
def test_related_join_filter():
t1, t2, t3, gokou, vegeta, gohan, trunks = createData()
filteredFighters = t1.fightersAsSResult.filter(
Fighter.q.name.startswith('go')
).orderBy('id')
for i, j in zip(filteredFighters, [gokou, gohan]):
assert i is j
class RecursiveGroup(SQLObject):
name = StringCol(length=255, unique=True)
subgroups = SQLRelatedJoin(
'RecursiveGroup',
otherColumn='group_id',
intermediateTable='rec_group_map',
createRelatedTable=False,
)
class RecGroupMap(SQLObject):
recursive_group = ForeignKey('RecursiveGroup')
group = ForeignKey('RecursiveGroup')
def test_rec_group():
setupClass([RecursiveGroup, RecGroupMap])
a = RecursiveGroup(name='a')
a1 = RecursiveGroup(name='a1')
a.addRecursiveGroup(a1)
a2 = RecursiveGroup(name='a2')
a.addRecursiveGroup(a2)
assert sorted(a.subgroups, key=lambda x: x.name) == [a1, a2]
pytest.raises(
ValueError,
a.subgroups.filter,
RecursiveGroup.q.name == 'a1',
)
rgroupAlias = Alias(RecursiveGroup, '_SO_SQLRelatedJoin_OtherTable')
assert list(a.subgroups.filter(rgroupAlias.q.name == 'a1')) == [a1]