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

Revision 477, 13.1 kB (checked in by ged, 4 years ago)

- Added a "table_kwargs" argument on ManyToMany relationships to pass any

extra keyword arguments to the underlying Table object (handles half of
ticket #94).

Line 
1"""
2test many to many relationships
3"""
4
5from elixir import *
6import elixir
7
8#-----------
9
10class TestManyToMany(object):
11    def setup(self):
12        metadata.bind = 'sqlite://'
13
14    def teardown(self):
15        cleanup_all(True)
16
17    def test_simple(self):
18        class A(Entity):
19            using_options(shortnames=True)
20            name = Field(String(60))
21            as_ = ManyToMany('A')
22            bs_ = ManyToMany('B')
23
24        class B(Entity):
25            using_options(shortnames=True)
26            name = Field(String(60))
27            as_ = ManyToMany('A')
28
29        setup_all(True)
30        A.mapper.compile()
31
32        # check m2m table was generated correctly
33        m2m_table = A.bs_.property.secondary
34        assert m2m_table.name in metadata.tables
35
36        # check column names
37        m2m_cols = m2m_table.columns
38        assert 'a_id' in m2m_cols
39        assert 'b_id' in m2m_cols
40
41        # check selfref m2m table column names were generated correctly
42        m2m_cols = A.as_.property.secondary.columns
43        assert 'as__id' in m2m_cols
44        assert 'inverse_id' in m2m_cols
45
46        # check the relationships work as expected
47        b1 = B(name='b1', as_=[A(name='a1')])
48
49        session.commit()
50        session.clear()
51
52        a = A.query.one()
53        b = B.query.one()
54
55        assert a in b.as_
56        assert b in a.bs_
57
58    def test_table_kwargs(self):
59        class A(Entity):
60            using_options(shortnames=True)
61            name = Field(String(60))
62            bs_ = ManyToMany('B', table_kwargs={'info': {'test': True}})
63
64        class B(Entity):
65            using_options(shortnames=True)
66            name = Field(String(60))
67            as_ = ManyToMany('A')
68
69        setup_all(True)
70        A.mapper.compile()
71
72        assert A.bs_.property.secondary.info['test'] is True
73
74    def test_custom_global_column_nameformat(self):
75        # this needs to be done before declaring the classes
76        elixir.options.M2MCOL_NAMEFORMAT = elixir.options.OLD_M2MCOL_NAMEFORMAT
77
78        class A(Entity):
79            bs_ = ManyToMany('B')
80
81        class B(Entity):
82            as_ = ManyToMany('A')
83
84        setup_all(True)
85
86        # revert to original format
87        elixir.options.M2MCOL_NAMEFORMAT = elixir.options.NEW_M2MCOL_NAMEFORMAT
88
89        # check m2m table was generated correctly
90        A.mapper.compile()
91        m2m_table = A.bs_.property.secondary
92        assert m2m_table.name in metadata.tables
93
94        # check column names
95        m2m_cols = m2m_table.columns
96        assert '%s_id' % A.table.name in m2m_cols
97        assert '%s_id' % B.table.name in m2m_cols
98
99    def test_alternate_column_formatter(self):
100        # this needs to be done before declaring the classes
101        elixir.options.M2MCOL_NAMEFORMAT = \
102            elixir.options.ALTERNATE_M2MCOL_NAMEFORMAT
103
104        class A(Entity):
105            as_ = ManyToMany('A')
106            bs_ = ManyToMany('B')
107
108        class B(Entity):
109            as_ = ManyToMany('A')
110
111        setup_all(True)
112        A.mapper.compile()
113
114        # revert to original format
115        elixir.options.M2MCOL_NAMEFORMAT = elixir.options.NEW_M2MCOL_NAMEFORMAT
116
117        # check m2m table column names were generated correctly
118        m2m_cols = A.bs_.property.secondary.columns
119        assert 'as__id' in m2m_cols
120        assert 'bs__id' in m2m_cols
121
122        # check selfref m2m table column names were generated correctly
123        m2m_cols = A.as_.property.secondary.columns
124        assert 'as__id' in m2m_cols
125        assert 'inverse_id' in m2m_cols
126
127    def test_upgrade(self):
128        elixir.options.M2MCOL_NAMEFORMAT = elixir.options.OLD_M2MCOL_NAMEFORMAT
129
130        class A(Entity):
131            using_options(shortnames=True)
132            name = Field(String(20))
133            links_to = ManyToMany('A')
134            is_linked_from = ManyToMany('A')
135            bs_ = ManyToMany('B')
136
137        class B(Entity):
138            using_options(shortnames=True)
139            name = Field(String(20))
140            as_ = ManyToMany('A')
141
142        setup_all(True)
143
144        a = A(name='a1', links_to=[A(name='a2')])
145
146        session.commit()
147        session.clear()
148
149        del A
150        del B
151
152        # do not drop the tables, that's the whole point!
153        cleanup_all()
154
155        # simulate a renaming of columns (as given by the migration aid)
156        # 'a_id1' to 'is_linked_from_id'.
157        # 'a_id2' to 'links_to_id'.
158        conn = metadata.bind.connect()
159        conn.execute("ALTER TABLE a_links_to__a_is_linked_from RENAME TO temp")
160        conn.execute("CREATE TABLE a_links_to__a_is_linked_from ("
161                        "is_linked_from_id INTEGER NOT NULL, "
162                        "links_to_id INTEGER NOT NULL, "
163                     "PRIMARY KEY (is_linked_from_id, links_to_id), "
164                     "CONSTRAINT a_fk1 FOREIGN KEY(is_linked_from_id) "
165                                      "REFERENCES a (id), "
166                     "CONSTRAINT a_fk2 FOREIGN KEY(links_to_id) "
167                                      "REFERENCES a (id))")
168        conn.execute("INSERT INTO a_links_to__a_is_linked_from "
169                     "(is_linked_from_id, links_to_id) "
170                     "SELECT a_id1, a_id2 FROM temp")
171        conn.close()
172
173        # ...
174        elixir.options.M2MCOL_NAMEFORMAT = elixir.options.NEW_M2MCOL_NAMEFORMAT
175#        elixir.options.MIGRATION_TO_07_AID = True
176
177        class A(Entity):
178            using_options(shortnames=True)
179            name = Field(String(20))
180            links_to = ManyToMany('A')
181            is_linked_from = ManyToMany('A')
182            bs_ = ManyToMany('B')
183
184        class B(Entity):
185            using_options(shortnames=True)
186            name = Field(String(20))
187            as_ = ManyToMany('A')
188
189        setup_all()
190
191        a1 = A.get_by(name='a1')
192        assert len(a1.links_to) == 1
193        assert not a1.is_linked_from
194
195        a2 = a1.links_to[0]
196        assert a2.name == 'a2'
197        assert not a2.links_to
198        assert a2.is_linked_from == [a1]
199
200    def test_manual_column_format(self):
201        class A(Entity):
202            using_options(tablename='aye')
203            name = Field(String(60))
204            bs_ = ManyToMany('B', column_format='%(entity)s_%(key)s')
205
206        class B(Entity):
207            using_options(tablename='bee')
208            name = Field(String(60))
209            as_ = ManyToMany('A', column_format='%(entity)s_%(key)s')
210
211        setup_all(True)
212
213        # check column names were generated correctly
214        A.mapper.compile()
215        m2m_cols = A.bs_.property.secondary.columns
216        assert 'a_id' in m2m_cols
217        assert 'b_id' in m2m_cols
218
219        # check the relationships work as expected
220        b1 = B(name='b1', as_=[A(name='a1')])
221
222        session.commit()
223        session.clear()
224
225        a = A.query.one()
226        b = B.query.one()
227
228        assert a in b.as_
229        assert b in a.bs_
230
231    def test_multi_pk_in_target(self):
232        class A(Entity):
233            key1 = Field(Integer, primary_key=True, autoincrement=False)
234            key2 = Field(String(40), primary_key=True)
235
236            bs_ = ManyToMany('B')
237
238        class B(Entity):
239            name = Field(String(60))
240            as_ = ManyToMany('A')
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_multi(self):
256        class A(Entity):
257            name = Field(String(100))
258
259            rel1 = ManyToMany('B')
260            rel2 = ManyToMany('B')
261
262        class B(Entity):
263            name = Field(String(20), primary_key=True)
264
265        setup_all(True)
266
267        b1 = B(name='b1')
268        a1 = A(name='a1', rel1=[B(name='b2'), b1],
269                          rel2=[B(name='b3'), B(name='b4'), b1])
270
271        session.commit()
272        session.clear()
273
274        a1 = A.query.one()
275        b1 = B.get_by(name='b1')
276        b2 = B.get_by(name='b2')
277
278        assert b1 in a1.rel1
279        assert b1 in a1.rel2
280        assert b2 in a1.rel1
281
282    def test_selfref(self):
283        class Person(Entity):
284            using_options(shortnames=True)
285            name = Field(String(30))
286
287            friends = ManyToMany('Person')
288
289        setup_all(True)
290
291        barney = Person(name="Barney")
292        homer = Person(name="Homer", friends=[barney])
293        barney.friends.append(homer)
294
295        session.commit()
296        session.clear()
297
298        homer = Person.get_by(name="Homer")
299        barney = Person.get_by(name="Barney")
300
301        assert homer in barney.friends
302        assert barney in homer.friends
303
304        m2m_cols = Person.friends.property.secondary.columns
305        assert 'friends_id' in m2m_cols
306        assert 'inverse_id' in m2m_cols
307
308    def test_bidirectional_selfref(self):
309        class Person(Entity):
310            using_options(shortnames=True)
311            name = Field(String(30))
312
313            friends = ManyToMany('Person')
314            is_friend_of = ManyToMany('Person')
315
316        setup_all(True)
317
318        barney = Person(name="Barney")
319        homer = Person(name="Homer", friends=[barney])
320        barney.friends.append(homer)
321
322        session.commit()
323        session.clear()
324
325        homer = Person.get_by(name="Homer")
326        barney = Person.get_by(name="Barney")
327
328        assert homer in barney.friends
329        assert barney in homer.friends
330
331        m2m_cols = Person.friends.property.secondary.columns
332        assert 'friends_id' in m2m_cols
333        assert 'is_friend_of_id' in m2m_cols
334
335    def test_has_and_belongs_to_many(self):
336        class A(Entity):
337            has_field('name', String(100))
338
339            has_and_belongs_to_many('bs', of_kind='B')
340
341        class B(Entity):
342            has_field('name', String(100), primary_key=True)
343
344        setup_all(True)
345
346        b1 = B(name='b1')
347        a1 = A(name='a1', bs=[B(name='b2'), b1])
348        a2 = A(name='a2', bs=[B(name='b3'), b1])
349        a3 = A(name='a3')
350
351        session.commit()
352        session.clear()
353
354        a1 = A.get_by(name='a1')
355        a2 = A.get_by(name='a2')
356        a3 = A.get_by(name='a3')
357        b1 = B.get_by(name='b1')
358        b2 = B.get_by(name='b2')
359
360        assert b1 in a1.bs
361        assert b2 in a1.bs
362        assert b1 in a2.bs
363        assert not a3.bs
364
365    def test_local_and_remote_colnames(self):
366        class A(Entity):
367            using_options(shortnames=True)
368            key1 = Field(Integer, primary_key=True, autoincrement=False)
369            key2 = Field(String(40), primary_key=True)
370
371            bs_ = ManyToMany('B', local_colname=['foo', 'bar'],
372                                  remote_colname="baz")
373
374        class B(Entity):
375            using_options(shortnames=True)
376            name = Field(String(60))
377            as_ = ManyToMany('A', remote_colname=['foo', 'bar'],
378                                  local_colname="baz")
379
380        setup_all(True)
381
382        b1 = B(name='b1', as_=[A(key1=10, key2='a1')])
383
384        session.commit()
385        session.clear()
386
387        a = A.query.one()
388        b = B.query.one()
389
390        assert a in b.as_
391        assert b in a.bs_
392
393    def test_manual_table_auto_joins(self):
394        from sqlalchemy import Table, Column, ForeignKey, ForeignKeyConstraint
395
396        a_b = Table('a_b', metadata,
397                    Column('a_key1', None),
398                    Column('a_key2', None),
399                    Column('b_id', None, ForeignKey('b.id')),
400                    ForeignKeyConstraint(['a_key1', 'a_key2'],
401                                         ['a.key1', 'a.key2']))
402
403        class A(Entity):
404            using_options(shortnames=True)
405            key1 = Field(Integer, primary_key=True, autoincrement=False)
406            key2 = Field(String(40), primary_key=True)
407
408            bs_ = ManyToMany('B', table=a_b)
409
410        class B(Entity):
411            using_options(shortnames=True)
412            name = Field(String(60))
413            as_ = ManyToMany('A', table=a_b)
414
415        setup_all(True)
416
417        b1 = B(name='b1', as_=[A(key1=10, key2='a1')])
418
419        session.commit()
420        session.clear()
421
422        a = A.query.one()
423        b = B.query.one()
424
425        assert a in b.as_
426        assert b in a.bs_
427
428    def test_manual_table_manual_joins(self):
429        from sqlalchemy import Table, Column, ForeignKey, \
430                               ForeignKeyConstraint, and_
431
432        a_b = Table('a_b', metadata,
433                    Column('a_key1', Integer),
434                    Column('a_key2', String(40)),
435                    Column('b_id', String(60)))
436
437        class A(Entity):
438            using_options(shortnames=True)
439            key1 = Field(Integer, primary_key=True, autoincrement=False)
440            key2 = Field(String(40), primary_key=True)
441
442            bs_ = ManyToMany('B', table=a_b,
443                             primaryjoin=lambda: and_(A.key1 == a_b.c.a_key1,
444                                                      A.key2 == a_b.c.a_key2),
445                             secondaryjoin=lambda: B.id == a_b.c.b_id,
446                             foreign_keys=[a_b.c.a_key1, a_b.c.a_key2,
447                                 a_b.c.b_id])
448
449        class B(Entity):
450            using_options(shortnames=True)
451            name = Field(String(60))
452
453        setup_all(True)
454
455        a1 = A(key1=10, key2='a1', bs_=[B(name='b1')])
456
457        session.commit()
458        session.clear()
459
460        a = A.query.one()
461        b = B.query.one()
462
463        assert b in a.bs_
Note: See TracBrowser for help on using the browser.