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

Revision 525, 14.2 kB (checked in by ged, 3 years 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.