root / elixir / trunk / tests / test_m2m.py @ 451

Revision 451, 7.5 kB (checked in by ged, 4 years ago)

simplified m2m column_format test

Line 
1"""
2test many to many relationships
3"""
4
5from elixir import *
6
7#-----------
8
9class TestManyToMany(object):
10    def setup(self):
11        metadata.bind = 'sqlite:///'
12
13    def teardown(self):
14        cleanup_all(True)
15
16    def test_simple(self):
17        class A(Entity):
18            name = Field(String(60))
19            bs_ = ManyToMany('B')
20
21        class B(Entity):
22            name = Field(String(60))
23            as_ = ManyToMany('A')
24
25        setup_all(True)
26
27        b1 = B(name='b1', as_=[A(name='a1')])
28
29        session.commit()
30        session.clear()
31
32        a = A.query.one()
33        b = B.query.one()
34
35        assert a in b.as_
36        assert b in a.bs_
37
38    def test_column_format(self):
39        class A(Entity):
40            using_options(tablename='aye')
41            name = Field(String(60))
42            bs_ = ManyToMany('B', column_format='%(entity)s_%(key)s')
43
44        class B(Entity):
45            using_options(tablename='bee')
46            name = Field(String(60))
47            as_ = ManyToMany('A', column_format='%(entity)s_%(key)s')
48
49        setup_all(True)
50
51        b1 = B(name='b1', as_=[A(name='a1')])
52
53        session.commit()
54        session.clear()
55
56        a = A.query.one()
57        b = B.query.one()
58
59        assert a in b.as_
60        assert b in a.bs_
61
62        m2m_cols = A.bs_.property.secondary.columns
63        assert 'a_id' in m2m_cols
64        assert 'b_id' in m2m_cols
65
66    def test_multi_pk_in_target(self):
67        class A(Entity):
68            key1 = Field(Integer, primary_key=True, autoincrement=False)
69            key2 = Field(String(40), primary_key=True)
70
71            bs_ = ManyToMany('B')
72
73        class B(Entity):
74            name = Field(String(60))
75            as_ = ManyToMany('A')
76
77        setup_all(True)
78
79        b1 = B(name='b1', as_=[A(key1=10, key2='a1')])
80
81        session.commit()
82        session.clear()
83
84        a = A.query.one()
85        b = B.query.one()
86
87        assert a in b.as_
88        assert b in a.bs_
89
90    def test_multi(self):
91        class A(Entity):
92            name = Field(String(100))
93
94            rel1 = ManyToMany('B')
95            rel2 = ManyToMany('B')
96
97        class B(Entity):
98            name = Field(String(20), primary_key=True)
99
100        setup_all(True)
101
102        b1 = B(name='b1')
103        a1 = A(name='a1', rel1=[B(name='b2'), b1],
104                          rel2=[B(name='b3'), B(name='b4'), b1])
105
106        session.commit()
107        session.clear()
108
109        a1 = A.query.one()
110        b1 = B.get_by(name='b1')
111        b2 = B.get_by(name='b2')
112
113        assert b1 in a1.rel1
114        assert b1 in a1.rel2
115        assert b2 in a1.rel1
116
117    def test_selfref(self):
118        class Person(Entity):
119            using_options(shortnames=True)
120            name = Field(String(30))
121
122            friends = ManyToMany('Person')
123
124        setup_all(True)
125
126        barney = Person(name="Barney")
127        homer = Person(name="Homer", friends=[barney])
128        barney.friends.append(homer)
129
130        session.commit()
131        session.clear()
132
133        homer = Person.get_by(name="Homer")
134        barney = Person.get_by(name="Barney")
135
136        assert homer in barney.friends
137        assert barney in homer.friends
138
139    def test_bidirectional_selfref(self):
140        class Person(Entity):
141            using_options(shortnames=True)
142            name = Field(String(30))
143
144            friends = ManyToMany('Person')
145            is_friend_of = ManyToMany('Person')
146
147        setup_all(True)
148
149        barney = Person(name="Barney")
150        homer = Person(name="Homer", friends=[barney])
151        barney.friends.append(homer)
152
153        session.commit()
154        session.clear()
155
156        homer = Person.get_by(name="Homer")
157        barney = Person.get_by(name="Barney")
158
159        assert homer in barney.friends
160        assert barney in homer.friends
161
162    def test_has_and_belongs_to_many(self):
163        class A(Entity):
164            has_field('name', String(100))
165
166            has_and_belongs_to_many('bs', of_kind='B')
167
168        class B(Entity):
169            has_field('name', String(100), primary_key=True)
170
171        setup_all(True)
172
173        b1 = B(name='b1')
174        a1 = A(name='a1', bs=[B(name='b2'), b1])
175        a2 = A(name='a2', bs=[B(name='b3'), b1])
176        a3 = A(name='a3')
177
178        session.commit()
179        session.clear()
180
181        a1 = A.get_by(name='a1')
182        a2 = A.get_by(name='a2')
183        a3 = A.get_by(name='a3')
184        b1 = B.get_by(name='b1')
185        b2 = B.get_by(name='b2')
186
187        assert b1 in a1.bs
188        assert b2 in a1.bs
189        assert b1 in a2.bs
190        assert not a3.bs
191
192    def test_local_and_remote_colnames(self):
193        class A(Entity):
194            using_options(shortnames=True)
195            key1 = Field(Integer, primary_key=True, autoincrement=False)
196            key2 = Field(String(40), primary_key=True)
197
198            bs_ = ManyToMany('B', local_colname=['foo', 'bar'],
199                                  remote_colname="baz")
200
201        class B(Entity):
202            using_options(shortnames=True)
203            name = Field(String(60))
204            as_ = ManyToMany('A', remote_colname=['foo', 'bar'],
205                                  local_colname="baz")
206
207        setup_all(True)
208
209        b1 = B(name='b1', as_=[A(key1=10, key2='a1')])
210
211        session.commit()
212        session.clear()
213
214        a = A.query.one()
215        b = B.query.one()
216
217        assert a in b.as_
218        assert b in a.bs_
219
220    def test_manual_table_auto_joins(self):
221        from sqlalchemy import Table, Column, ForeignKey, ForeignKeyConstraint
222
223        a_b = Table('a_b', metadata,
224                    Column('a_key1', None),
225                    Column('a_key2', None),
226                    Column('b_id', None, ForeignKey('b.id')),
227                    ForeignKeyConstraint(['a_key1', 'a_key2'],
228                                         ['a.key1', 'a.key2']))
229
230        class A(Entity):
231            using_options(shortnames=True)
232            key1 = Field(Integer, primary_key=True, autoincrement=False)
233            key2 = Field(String(40), primary_key=True)
234
235            bs_ = ManyToMany('B', table=a_b)
236
237        class B(Entity):
238            using_options(shortnames=True)
239            name = Field(String(60))
240            as_ = ManyToMany('A', table=a_b)
241
242        setup_all(True)
243
244        b1 = B(name='b1', as_=[A(key1=10, key2='a1')])
245
246        session.commit()
247        session.clear()
248
249        a = A.query.one()
250        b = B.query.one()
251
252        assert a in b.as_
253        assert b in a.bs_
254
255    def test_manual_table_manual_joins(self):
256        from sqlalchemy import Table, Column, ForeignKey, \
257                               ForeignKeyConstraint, and_
258
259        a_b = Table('a_b', metadata,
260                    Column('a_key1', Integer),
261                    Column('a_key2', String(40)),
262                    Column('b_id', String(60)))
263
264        class A(Entity):
265            using_options(shortnames=True)
266            key1 = Field(Integer, primary_key=True, autoincrement=False)
267            key2 = Field(String(40), primary_key=True)
268
269            bs_ = ManyToMany('B', table=a_b,
270                             primaryjoin=lambda: and_(A.key1 == a_b.c.a_key1,
271                                                      A.key2 == a_b.c.a_key2),
272                             secondaryjoin=lambda: B.id == a_b.c.b_id,
273                             foreign_keys=[a_b.c.a_key1, a_b.c.a_key2,
274                                 a_b.c.b_id])
275
276        class B(Entity):
277            using_options(shortnames=True)
278            name = Field(String(60))
279
280        setup_all(True)
281
282        a1 = A(key1=10, key2='a1', bs_=[B(name='b1')])
283
284        session.commit()
285        session.clear()
286
287        a = A.query.one()
288        b = B.query.one()
289
290        assert b in a.bs_
Note: See TracBrowser for help on using the browser.