root / elixir / trunk / tests / test_m2m.py

Revision 525, 14.2 kB (checked in by ged, 21 months ago)

Do not use None as column types in tests because this is unsupported in SA 0.6
for composite foreign keys

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