« back to all changes in this revision
Viewing changes to tests/test_autoref.py
- browse files at revision 1
- compare with another revision
- download diff
- download tarball
- view history from revision 1
- Committer: Clint Byrum
- Date: 2012-01-25 21:53:57 UTC
- Revision ID: clint@ubuntu.com-20120125215357-j603u9d5alv1bt9v
Tags: upstream-0.7.2
Import upstream version 0.7.2
- files added:
- mongokit
- mongokit.egg-info
- tests
added
removed
tests/test_autoref.py
1
#!/usr/bin/env python
2
# -*- coding: utf-8 -*-
3
# Copyright (c) 2009-2011, Nicolas Clairon
4
# All rights reserved.
5
# Redistribution and use in source and binary forms, with or without
6
# modification, are permitted provided that the following conditions are met:
7
#
8
# * Redistributions of source code must retain the above copyright
9
# notice, this list of conditions and the following disclaimer.
10
# * Redistributions in binary form must reproduce the above copyright
11
# notice, this list of conditions and the following disclaimer in the
12
# documentation and/or other materials provided with the distribution.
13
# * Neither the name of the University of California, Berkeley nor the
14
# names of its contributors may be used to endorse or promote products
15
# derived from this software without specific prior written permission.
16
#
17
# THIS SOFTWARE IS PROVIDED BY THE REGENTS AND CONTRIBUTORS ``AS IS'' AND ANY
18
# EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
19
# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
20
# DISCLAIMED. IN NO EVENT SHALL THE REGENTS AND CONTRIBUTORS BE LIABLE FOR ANY
21
# DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
22
# (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
23
# LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
24
# ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
25
# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
26
# SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
27
28
import unittest
29
import logging
30
31
logging.basicConfig(level=logging.DEBUG)
32
33
from mongokit import *
34
from pymongo.objectid import ObjectId
35
36
class AutoRefTestCase(unittest.TestCase):
37
"""Tests AutoRef case"""
38
def setUp(self):
39
self.connection = Connection()
40
self.col = self.connection['test']['mongokit']
41
42
def tearDown(self):
43
self.connection.drop_database('test')
44
self.connection.drop_database('test2')
45
46
def test_simple_autoref(self):
47
class DocA(Document):
48
structure = {
49
"a":{'foo':int},
50
}
51
self.connection.register([DocA])
52
53
doca = self.col.DocA()
54
doca['_id'] = 'doca'
55
doca['a']['foo'] = 3
56
doca.save()
57
58
class DocB(Document):
59
structure = {
60
"b":{"doc_a":DocA},
61
}
62
use_autorefs = True
63
self.connection.register([DocB])
64
65
docb = self.col.DocB()
66
# the structure is automaticly filled by the corresponding structure
67
assert docb == {'b': {'doc_a':None}}, docb
68
#docb.validate()
69
docb['_id'] = 'docb'
70
docb['b']['doc_a'] = 4
71
self.assertRaises(SchemaTypeError, docb.validate)
72
docb['b']['doc_a'] = doca
73
assert docb == {'b': {'doc_a': {'a': {'foo': 3}, '_id': 'doca'}}, '_id': 'docb'}
74
docb.save()
75
saved_docb = self.col.find_one({'_id':'docb'})
76
_docb = self.col.DocB.get_from_id('docb')
77
assert saved_docb['b']['doc_a'] == DBRef(database='test', collection='mongokit', id='doca'), saved_docb['b']['doc_a']
78
79
docb_list = list(self.col.DocB.fetch())
80
assert len(docb_list) == 1
81
new_docb = docb_list[0]
82
assert isinstance(new_docb['b']['doc_a'], DocA), new_docb['b']['doc_a'].__class__
83
assert docb == {'b': {'doc_a': {'a': {'foo': 3}, '_id': 'doca'}}, '_id': 'docb'}, docb
84
assert docb['b']['doc_a']['a']['foo'] == 3
85
docb['b']['doc_a']['a']['foo'] = 4
86
docb.save()
87
assert docb['b']['doc_a']['a']['foo'] == 4, docb
88
assert self.col.DocA.fetch().next()['a']['foo'] == 4
89
assert doca['a']['foo'] == 4, doca['a']['foo']
90
saved_docb = self.col.DocB.collection.find_one({'_id':'docb'})
91
assert saved_docb['b']['doc_a'] == DBRef(database='test', collection='mongokit', id='doca'), saved_docb['b']['doc_a']
92
assert self.col.DocB.fetch_one() == docb
93
assert self.col.DocB.find_one({'_id':'docb'}) == docb
94
95
def test_simple_autoref2(self):
96
class Embed(Document):
97
structure = {
98
'foo': dict,
99
'bar': int,
100
}
101
102
class Doc(Document):
103
structure = {
104
'embed':Embed,
105
'eggs': unicode,
106
}
107
use_autorefs = True
108
self.connection.register([Embed, Doc])
109
110
embed = self.col.Embed()
111
embed['foo'] = {'hello':u'monde'}
112
embed['bar'] = 3
113
embed.save()
114
115
doc = self.col.Doc()
116
doc['embed'] = embed
117
doc['eggs'] = u'arf'
118
doc.save()
119
120
assert doc == {'embed': {u'_id': embed['_id'], u'bar': 3, u'foo': {u'hello': u'monde'}}, '_id': doc['_id'], 'eggs': u'arf'}, doc
121
122
doc = self.col.Doc.fetch_one()
123
doc['embed']['foo']['hello'] = u'World'
124
doc.save()
125
126
assert doc == {'embed': {u'_id': embed['_id'], u'bar': 3, u'foo': {u'hello': u'World'}}, '_id': doc['_id'], 'eggs': u'arf'}, doc
127
assert self.col.Embed.fetch_one() == {u'_id': embed['_id'], u'bar': 3, u'foo': {u'hello': u'World'}}
128
129
def test_autoref_with_default_values(self):
130
class DocA(Document):
131
structure = {
132
"a":{'foo':int},
133
"abis":{'bar':int},
134
}
135
self.connection.register([DocA])
136
doca = self.col.DocA()
137
doca['_id'] = 'doca'
138
doca['a']['foo'] = 2
139
doca.save()
140
141
class DocB(Document):
142
structure = {
143
"b":{"doc_a":DocA},
144
}
145
use_autorefs = True
146
default_values = {'b.doc_a':doca}
147
self.connection.register([DocB])
148
149
docb = self.col.DocB()
150
assert docb == {'b': {'doc_a': {'a': {'foo': 2}, 'abis': {'bar': None}, '_id': 'doca'}}}, docb
151
docb.save()
152
153
def test_autoref_with_required_fields(self):
154
class DocA(Document):
155
structure = {
156
"a":{'foo':int},
157
"abis":{'bar':int},
158
}
159
required_fields = ['a.foo']
160
self.connection.register([DocA])
161
162
doca = self.col.DocA()
163
doca['_id'] = 'doca'
164
doca['a']['foo'] = 2
165
doca.save()
166
167
class DocB(Document):
168
db_name = "test"
169
collection_name = "mongokit"
170
structure = {
171
"b":{"doc_a":DocA},
172
}
173
use_autorefs = True
174
self.connection.register([DocB])
175
176
docb = self.col.DocB()
177
docb['b']['doc_a'] = doca
178
assert docb == {'b': {'doc_a': {'a': {'foo': 2}, 'abis': {'bar': None}, '_id': 'doca'}}}, docb
179
docb['_id'] = 'docb'
180
docb['b']['doc_a']['a']['foo'] = None
181
self.assertRaises(RequireFieldError, docb.validate)
182
docb['b']['doc_a']['a']['foo'] = 4
183
docb.save()
184
185
docb['b']['doc_a'] = None
186
docb.save()
187
188
def test_badautoref(self):
189
"""Test autoref enabled, but embed the wrong kind of document.
190
Assert that it tells us it's a bad embed.
191
"""
192
class EmbedDoc(Document):
193
structure = {
194
"spam": unicode
195
}
196
self.connection.register([EmbedDoc])
197
embed = self.col.EmbedDoc()
198
embed["spam"] = u"eggs"
199
embed.save()
200
assert embed
201
202
class EmbedOtherDoc(Document):
203
structure = {
204
"ham": unicode
205
}
206
self.connection.register([EmbedOtherDoc])
207
embedOther = self.connection.test.embed_other.EmbedOtherDoc()
208
embedOther["ham"] = u"eggs"
209
embedOther.save()
210
assert embedOther
211
212
class MyDoc(Document):
213
use_autorefs = True
214
structure = {
215
"bla":{
216
"foo":unicode,
217
"bar":int,
218
},
219
"spam": EmbedDoc,
220
}
221
use_autorefs = True
222
self.connection.register([MyDoc])
223
mydoc = self.connection.test.autoref.MyDoc()
224
mydoc["bla"]["foo"] = u"bar"
225
mydoc["bla"]["bar"] = 42
226
mydoc["spam"] = embedOther
227
228
self.assertRaises(SchemaTypeError, mydoc.save)
229
230
def test_badautoref_not_enabled(self):
231
# Test that, when autoref is disabled
232
# we refuse to allow a MongoDocument
233
# to be valid schema.
234
235
class EmbedDoc(Document):
236
structure = {
237
"spam": unicode
238
}
239
self.connection.register([EmbedDoc])
240
embed = self.connection.test['autoref.embed'].EmbedDoc()
241
embed["spam"] = u"eggs"
242
embed.save()
243
assert embed
244
245
class MyDoc(Document):
246
structure = {
247
"bla":{
248
"foo":unicode,
249
"bar":int,
250
},
251
"spam": EmbedDoc,
252
}
253
self.assertRaises(StructureError, self.connection.register, [MyDoc])
254
255
def test_subclass(self):
256
# Test autoref enabled, but embed a subclass.
257
# e.g. if we say EmbedDoc, a subclass of EmbedDoc
258
# is also valid.
259
260
class EmbedDoc(Document):
261
structure = {
262
"spam": unicode
263
}
264
self.connection.register([EmbedDoc])
265
embed = self.connection.test['autoref.embed'].EmbedDoc()
266
embed["spam"] = u"eggs"
267
embed.save()
268
269
class EmbedOtherDoc(EmbedDoc):
270
structure = {
271
"ham": unicode
272
}
273
self.connection.register([EmbedOtherDoc])
274
embedOther = self.connection.test['autoref.embed_other'].EmbedOtherDoc()
275
embedOther["ham"] = u"eggs"
276
embedOther.save()
277
assert embedOther
278
279
class MyDoc(Document):
280
use_autorefs = True
281
structure = {
282
"bla":{
283
"foo":unicode,
284
"bar":int,
285
},
286
"spam": EmbedDoc,
287
}
288
self.connection.register([MyDoc])
289
mydoc = self.connection.test.autoref.MyDoc()
290
mydoc["bla"]["foo"] = u"bar"
291
mydoc["bla"]["bar"] = 42
292
mydoc["spam"] = embedOther
293
294
mydoc.save()
295
assert mydoc['spam'].collection.name == "autoref.embed_other"
296
assert mydoc['spam'] == embedOther
297
298
def test_autoref_in_list(self):
299
class DocA(Document):
300
structure = {
301
"a":{'foo':int},
302
}
303
self.connection.register([DocA])
304
305
doca = self.col.DocA()
306
doca['_id'] = 'doca'
307
doca['a']['foo'] = 3
308
doca.save()
309
310
doca2 = self.col.DocA()
311
doca2['_id'] = 'doca2'
312
doca2['a']['foo'] = 5
313
doca2.save()
314
315
class DocB(Document):
316
structure = {
317
"b":{"doc_a":[DocA]},
318
}
319
use_autorefs = True
320
self.connection.register([DocB])
321
322
docb = self.col.DocB()
323
# the structure is automaticly filled by the corresponding structure
324
assert docb == {'b': {'doc_a':[]}}, docb
325
docb.validate()
326
docb['_id'] = 'docb'
327
docb['b']['doc_a'].append(u'bla')
328
self.assertRaises(SchemaTypeError, docb.validate)
329
docb['b']['doc_a'] = []
330
docb['b']['doc_a'].append(doca)
331
assert docb == {'b': {'doc_a': [{'a': {'foo': 3}, '_id': 'doca'}]}, '_id': 'docb'}
332
docb.save()
333
assert isinstance(docb.collection.find_one({'_id':'docb'})['b']['doc_a'][0], DBRef), type(docb.collection.find_one({'_id':'docb'})['b']['doc_a'][0])
334
335
assert docb == {'b': {'doc_a': [{'a': {'foo': 3}, '_id': 'doca'}]}, '_id': 'docb'}
336
assert docb['b']['doc_a'][0]['a']['foo'] == 3
337
docb['b']['doc_a'][0]['a']['foo'] = 4
338
docb.save()
339
assert docb['b']['doc_a'][0]['a']['foo'] == 4, docb['b']['doc_a'][0]['a']['foo']
340
assert doca['a']['foo'] == 4, doca['a']['foo']
341
342
docb['b']['doc_a'].append(doca2)
343
assert docb == {'b': {'doc_a': [{'a': {'foo': 4}, '_id': 'doca'}, {'a': {'foo': 5}, '_id': 'doca2'}]}, '_id': 'docb'}
344
docb.validate()
345
346
def test_autoref_retrieval(self):
347
class DocA(Document):
348
structure = {
349
"a":{'foo':int},
350
}
351
self.connection.register([DocA])
352
353
doca = self.col.DocA()
354
doca['_id'] = 'doca'
355
doca['a']['foo'] = 3
356
doca.save()
357
358
class DocB(Document):
359
structure = {
360
"b":{
361
"doc_a":DocA,
362
"deep": {"doc_a_deep":DocA},
363
"deeper": {"doc_a_deeper":DocA,
364
"inner":{"doc_a_deepest":DocA}}
365
},
366
367
}
368
use_autorefs = True
369
self.connection.register([DocB])
370
371
docb = self.col.DocB()
372
# the structure is automaticly filled by the corresponding structure
373
docb['_id'] = 'docb'
374
docb['b']['doc_a'] = doca
375
376
# create a few deeper docas
377
deep = self.col.DocA()
378
#deep['_id'] = 'deep'
379
deep['a']['foo'] = 5
380
deep.save()
381
docb['b']['deep']['doc_a_deep'] = deep
382
deeper = self.col.DocA()
383
deeper['_id'] = 'deeper'
384
deeper['a']['foo'] = 8
385
deeper.save()
386
docb['b']['deeper']['doc_a_deeper'] = deeper
387
deepest = self.col.DocA()
388
deepest['_id'] = 'deepest'
389
#deepest['_id'] = 'deeper'
390
deepest['a']['foo'] = 18
391
deepest.save()
392
docb['b']['deeper']['inner']['doc_a_deepest'] = deepest
393
394
docb.save()
395
396
# now, does retrieval function as expected?
397
test_doc = self.col.DocB.get_from_id(docb['_id'])
398
assert isinstance(test_doc['b']['doc_a'], DocA), type(test_doc['b']['doc_a'])
399
assert test_doc['b']['doc_a']['a']['foo'] == 3
400
assert isinstance(test_doc['b']['deep']['doc_a_deep'], DocA)
401
assert test_doc['b']['deep']['doc_a_deep']['a']['foo'] == 5
402
assert isinstance(test_doc['b']['deeper']['doc_a_deeper'], DocA)
403
assert test_doc['b']['deeper']['doc_a_deeper']['a']['foo'] == 8, test_doc
404
assert isinstance(test_doc['b']['deeper']['inner']['doc_a_deepest'], DocA)
405
assert test_doc['b']['deeper']['inner']['doc_a_deepest']['a']['foo'] == 18
406
407
def test_autoref_with_same_embed_id(self):
408
class DocA(Document):
409
structure = {
410
"a":{'foo':int},
411
}
412
self.connection.register([DocA])
413
414
doca = self.col.DocA()
415
doca['_id'] = 'doca'
416
doca['a']['foo'] = 3
417
doca.save()
418
419
class DocB(Document):
420
structure = {
421
"b":{
422
"doc_a":DocA,
423
"deep": {"doc_a_deep":DocA},
424
},
425
426
}
427
use_autorefs = True
428
self.connection.register([DocB])
429
430
docb = self.col.DocB()
431
docb['_id'] = 'docb'
432
docb['b']['doc_a'] = doca
433
# create a few deeper docas
434
deep = self.col.DocA()
435
deep['_id'] = 'doca' # XXX same id of doca, this will be erased by doca when saving docb
436
deep['a']['foo'] = 5
437
deep.save()
438
docb['b']['deep']['doc_a_deep'] = deep
439
440
docb.save()
441
442
test_doc = self.col.DocB.get_from_id(docb['_id'])
443
assert test_doc['b']['doc_a']['a']['foo'] == 3, test_doc['b']['doc_a']['a']
444
assert test_doc['b']['deep']['doc_a_deep']['a']['foo'] == 3, test_doc['b']['deep']['doc_a_deep']['a']['foo']
445
446
def test_autorefs_embed_in_list_with_bad_reference(self):
447
class User(Document):
448
structure = {'name':unicode}
449
self.connection.register([User])
450
451
class Group(Document):
452
use_autorefs = True
453
structure = {
454
'name':unicode,
455
'members':[User], #users
456
}
457
self.connection.register([User, Group])
458
459
user = self.col.User()
460
user['_id'] = u'fixe'
461
user['name'] = u'fixe'
462
user.save()
463
464
user2 = self.col.User()
465
user['_id'] = u'namlook'
466
user2['name'] = u'namlook'
467
user2.save()
468
469
group = self.col.Group()
470
group['members'].append(user)
471
self.assertRaises(AutoReferenceError, group.save)
472
473
def test_autorefs_with_dynamic_collection(self):
474
class DocA(Document):
475
structure = {'a':unicode}
476
477
class DocB(Document):
478
structure = {'b':DocA}
479
use_autorefs = True
480
self.connection.register([DocA, DocB])
481
482
doca = self.connection.test.doca.DocA()
483
doca['a'] = u'bla'
484
doca.save()
485
486
docb = self.connection.test.docb.DocB()
487
docb['b'] = doca
488
docb.save()
489
490
assert docb['b']['a'] == 'bla'
491
assert docb['b'].collection.name == "doca"
492
493
doca2 = self.connection.test.doca2.DocA()
494
doca2['a'] = u'foo'
495
doca2.save()
496
497
docb2 = self.connection.test.docb.DocB()
498
docb2['b'] = doca2
499
docb2.save()
500
501
assert docb2['b']['a'] == 'foo'
502
assert docb2['b'].collection.name == 'doca2'
503
assert docb2.collection.name == 'docb'
504
505
assert list(self.connection.test.docb.DocB.fetch()) == [docb, docb2]
506
507
def test_autorefs_with_dynamic_db(self):
508
class DocA(Document):
509
structure = {'a':unicode}
510
511
class DocB(Document):
512
structure = {'b':DocA}
513
use_autorefs = True
514
self.connection.register([DocA, DocB])
515
516
doca = self.connection.dba.mongokit.DocA()
517
doca['a'] = u'bla'
518
doca.save()
519
520
docb = self.connection.dbb.mongokit.DocB()
521
docb['b'] = doca
522
docb.save()
523
524
assert docb['b']['a'] == 'bla'
525
docb = self.connection.dbb.mongokit.DocB.get_from_id(docb['_id'])
526
assert isinstance(docb['b'], DocA)
527
528
def test_autoref_without_validation(self):
529
class DocA(Document):
530
structure = {
531
"a":{'foo':int},
532
}
533
self.connection.register([DocA])
534
535
doca = self.col.DocA()
536
doca['_id'] = 'doca'
537
doca['a']['foo'] = 3
538
doca.save()
539
540
class DocB(Document):
541
structure = {
542
"b":{"doc_a":DocA},
543
}
544
use_autorefs = True
545
skip_validation = True
546
self.connection.register([DocB])
547
548
docb = self.col.DocB()
549
docb['_id'] = 'docb'
550
docb['b']['doc_a'] = doca
551
docb.save()
552
553
def test_autoref_updated(self):
554
class DocA(Document):
555
structure = {
556
"a":{'foo':int},
557
}
558
self.connection.register([DocA])
559
560
doca = self.col.DocA()
561
doca['_id'] = 'doca'
562
doca['a']['foo'] = 3
563
doca.save()
564
565
doca2 = self.col.DocA()
566
doca2['_id'] = 'doca2'
567
doca2['a']['foo'] = 6
568
doca2.save()
569
570
class DocB(Document):
571
structure = {
572
"b":{"doc_a":[DocA]},
573
}
574
use_autorefs = True
575
self.connection.register([DocB])
576
577
docb = self.col.DocB()
578
docb['_id'] = 'docb'
579
docb.save()
580
assert docb == {'b': {'doc_a': []}, '_id': 'docb'}
581
docb['b']['doc_a'] = [doca, doca2]
582
docb.save()
583
assert docb == {'b': {'doc_a': [{u'a': {u'foo': 3}, u'_id': u'doca'}, {u'a': {u'foo': 6}, u'_id': u'doca2'}]}, '_id': 'docb'}
584
docb['b']['doc_a'].pop(0)
585
docb.save()
586
assert docb == {'b': {'doc_a': [{u'a': {u'foo': 6}, u'_id': u'doca2'}]}, '_id': 'docb'}
587
fetched_docb = self.col.DocB.get_from_id('docb')
588
assert fetched_docb == {u'_id': u'docb', u'b': {u'doc_a': [{u'a': {u'foo': 6}, u'_id': u'doca2'}]}}
589
590
docb = self.col.DocB()
591
docb['_id'] = 'docb'
592
docb.save()
593
assert docb == {'b': {'doc_a': []}, '_id': 'docb'}
594
docb['b']['doc_a'] = [doca, doca2]
595
docb.save()
596
assert docb == {'b': {'doc_a': [{u'a': {u'foo': 3}, u'_id': u'doca'}, {u'a': {u'foo': 6}, u'_id': u'doca2'}]}, '_id': 'docb'}, docb
597
docb['b']['doc_a'].pop(0)
598
docb['b']['doc_a'].append(doca)
599
docb.save()
600
assert docb == {'b': {'doc_a': [{u'a': {u'foo': 6}, u'_id': u'doca2'}, {u'a': {u'foo': 3}, u'_id': u'doca'}]}, '_id': 'docb'}, docb
601
fetched_docb = self.col.DocB.get_from_id('docb')
602
assert fetched_docb == {u'_id': u'docb', u'b': {u'doc_a': [{u'a': {u'foo': 6}, u'_id': u'doca2'}, {u'a': {u'foo': 3}, u'_id': u'doca'}]}}
603
604
def test_autoref_updated_with_default_values(self):
605
class DocA(Document):
606
structure = {
607
"a":{'foo':int},
608
"abis":{'bar':int},
609
}
610
default_values = {'a.foo':2}
611
required_fields = ['abis.bar']
612
613
self.connection.register([DocA])
614
doca = self.col.DocA()
615
doca['_id'] = 'doca'
616
doca['abis']['bar'] = 3
617
doca.save()
618
619
class DocB(Document):
620
structure = {
621
"b":{"doc_a":DocA},
622
}
623
use_autorefs = True
624
625
self.connection.register([DocB])
626
docb = self.col.DocB()
627
docb['_id'] = 'docb'
628
docb['b']['doc_a'] = doca
629
assert docb == {'b': {'doc_a': {'a': {'foo': 2}, 'abis': {'bar': 3}, '_id': 'doca'}}, '_id': 'docb'}, docb
630
docb['b']['doc_a']['a']['foo'] = 4
631
docb.save()
632
assert docb == {'b': {'doc_a': {'a': {'foo': 4}, 'abis': {'bar': 3}, '_id': 'doca'}}, '_id': 'docb'}, docb
633
assert doca['a']['foo'] == 4
634
635
def test_autoref_with_None(self):
636
class RootDocument(Document):
637
use_dot_notation=True
638
use_autorefs = True
639
structure = {}
640
641
class User(RootDocument):
642
collection_name = "users"
643
structure = {
644
"email": unicode,
645
"password": unicode,
646
}
647
required_fields = [ "email", "password" ]
648
indexes = [
649
{ "fields": "email",
650
"unique": True,
651
},
652
]
653
self.connection.register([User])
654
User = self.col.User
655
u = User()
656
u['email'] = u'....'
657
u['password'] = u'....'
658
u.save()
659
assert u['_id'] != None
660
661
class ExampleSession(RootDocument):
662
#collection_name = "sessions"
663
use_autorefs = True
664
structure = {
665
"user": User,
666
"token": unicode,
667
}
668
# raise an assertion because User is a CallableUser, not User
669
self.connection.register([ExampleSession])
670
ex = self.col.ExampleSession()
671
self.assertRaises(SchemaTypeError, ex.validate)
672
673
def test_autoref_without_database_specified(self):
674
class EmbedDoc(Document):
675
structure = {
676
"foo": unicode,
677
}
678
679
class Doc(Document):
680
use_dot_notation=True
681
use_autorefs = True
682
force_autorefs_current_db = True
683
structure = {
684
"embed": EmbedDoc,
685
}
686
self.connection.register([EmbedDoc, Doc])
687
688
embed = self.col.EmbedDoc()
689
embed['foo'] = u'bar'
690
embed.save()
691
692
raw_doc = {'embed':DBRef(collection=self.col.name, id=embed['_id'])}
693
self.col.insert(raw_doc)
694
695
doc = self.col.Doc.find_one({'_id':raw_doc['_id']})
696
697
def test_recreate_and_reregister_class_with_reference(self):
698
class CompanyDocument(Document):
699
collection_name = "test_companies"
700
use_autorefs = True
701
use_dot_notation = True
702
structure = {
703
"name": unicode,
704
}
705
706
class UserDocument(Document):
707
collection_name = "test_users"
708
use_autorefs = True
709
use_dot_notation = True
710
structure = {
711
"email": unicode,
712
"company": CompanyDocument,
713
}
714
715
class SessionDocument(Document):
716
collection_name = "test_sessions"
717
use_autorefs = True
718
use_dot_notation = True
719
structure = {
720
"token": unicode,
721
"owner": UserDocument,
722
}
723
self.connection.register([CompanyDocument, UserDocument, SessionDocument])
724
725
company = self.col.database[CompanyDocument.collection_name].CompanyDocument()
726
company.name = u"Company"
727
company.save()
728
729
company_owner = self.col.database[UserDocument.collection_name].UserDocument()
730
company_owner.email = u"manager@test.com"
731
company_owner.company = company
732
company_owner.save()
733
734
s = self.col.database[SessionDocument.collection_name].SessionDocument()
735
s.token = u'asddadsad'
736
s.owner = company_owner
737
s.save()
738
739
sbis= self.col.database[SessionDocument.collection_name].SessionDocument.find_one({"token": u"asddadsad" })
740
assert sbis == s, sbis
741
742
class CompanyDocument(Document):
743
collection_name = "test_companies"
744
use_autorefs = True
745
structure = {
746
"name": unicode,
747
}
748
749
class UserDocument(Document):
750
collection_name = "test_users"
751
use_autorefs = True
752
structure = {
753
"email": unicode,
754
"company": CompanyDocument,
755
}
756
757
class SessionDocument(Document):
758
collection_name = "test_sessions"
759
use_autorefs = True
760
structure = {
761
"token": unicode,
762
"owner": UserDocument,
763
}
764
self.connection.register([CompanyDocument, UserDocument, SessionDocument])
765
766
sbis= self.col.database[SessionDocument.collection_name].SessionDocument.find_one({"token": u"asddadsad" })
767
assert sbis == s, sbis
768
769
770
def test_nested_autorefs(self):
771
class DocA(Document):
772
structure = {
773
'name':unicode,
774
}
775
use_autorefs = True
776
777
class DocB(Document):
778
structure = {
779
'name': unicode,
780
'doca' : DocA,
781
}
782
use_autorefs = True
783
784
class DocC(Document):
785
structure = {
786
'name': unicode,
787
'docb': DocB,
788
'doca': DocA,
789
}
790
use_autorefs = True
791
792
class DocD(Document):
793
structure = {
794
'name': unicode,
795
'docc': DocC,
796
}
797
use_autorefs = True
798
self.connection.register([DocA, DocB, DocC, DocD])
799
800
doca = self.col.DocA()
801
doca['name'] = u'Test A'
802
doca.save()
803
804
docb = self.col.DocB()
805
docb['name'] = u'Test B'
806
docb['doca'] = doca
807
docb.save()
808
809
docc = self.col.DocC()
810
docc['name'] = u'Test C'
811
docc['docb'] = docb
812
docc['doca'] = doca
813
docc.save()
814
815
docd = self.col.DocD()
816
docd['name'] = u'Test D'
817
docd['docc'] = docc
818
docd.save()
819
820
doca = self.col.DocA.find_one({'name': 'Test A'})
821
docb = self.col.DocB.find_one({'name': 'Test B'})
822
docc = self.col.DocC.find_one({'name': 'Test C'})
823
docd = self.col.DocD.find_one({'name': 'Test D'})
824
825
826
def test_nested_autoref_in_list_and_dict(self):
827
class DocA(Document):
828
structure = {
829
'name':unicode,
830
}
831
use_autorefs = True
832
833
834
class DocB(Document):
835
structure = {
836
'name': unicode,
837
'test': [{
838
'something' : unicode,
839
'doca' : DocA,
840
}]
841
}
842
use_autorefs = True
843
844
self.connection.register([DocA, DocB])
845
846
doca = self.col.DocA()
847
doca['name'] = u'Test A'
848
doca.save()
849
850
docc = self.col.DocA()
851
docc['name'] = u'Test C'
852
docc.save()
853
854
docb = self.col.DocB()
855
docb['name'] = u'Test B'
856
docb['test'].append({u'something': u'foo', 'doca': doca})
857
docb['test'].append({u'something': u'foo', 'doca': docc})
858
docb.save()
859
860
raw_docb = self.col.find_one({'name':'Test B'})
861
assert isinstance(raw_docb['test'][0]['doca'], DBRef), raw_docb['test'][0]
862
863
def test_dereference(self):
864
865
class DocA(Document):
866
structure = {
867
'name':unicode,
868
}
869
use_autorefs = True
870
871
self.connection.register([DocA])
872
873
doca = self.col.DocA()
874
doca['name'] = u'Test A'
875
doca.save()
876
877
docb = self.connection.test2.mongokit.DocA()
878
docb['name'] = u'Test B'
879
docb.save()
880
881
dbref = doca.get_dbref()
882
883
self.assertRaises(TypeError, self.connection.test.dereference, 1)
884
self.assertRaises(ValueError, self.connection.test.dereference, docb.get_dbref(), DocA)
885
assert self.connection.test.dereference(dbref) == {'_id':doca['_id'], 'name': 'Test A'}
886
assert isinstance(self.connection.test.dereference(dbref), dict)
887
assert self.connection.test.dereference(dbref, DocA) == {'_id':doca['_id'], 'name': 'Test A'}
888
assert isinstance(self.connection.test.dereference(dbref, DocA), DocA)
889
890
def test_autorefs_with_list(self):
891
class VDocument(Document):
892
db_name = 'MyDB'
893
use_dot_notation = True
894
use_autorefs = True
895
skip_validation = True
896
897
def __init__(self, *args, **kwargs):
898
super(VDocument, self).__init__(*args, **kwargs)
899
900
def save(self, *args, **kwargs):
901
kwargs.update({'validate':True})
902
return super(VDocument, self).save(*args, **kwargs)
903
904
class H(VDocument):
905
structure = {'name':[ObjectId], 'blah':[unicode], 'foo': [{'x':unicode}]}
906
self.connection.register([H, VDocument])
907
908
h = self.col.H()
909
obj_id = ObjectId()
910
h.name.append(obj_id)
911
h.blah.append(u'some string')
912
h.foo.append({'x':u'hey'})
913
h.save()
914
assert h == {'blah': [u'some string'], 'foo': [{'x': u'hey'}], 'name': [obj_id], '_id': h['_id']}
915
916
def test_autorefs_with_list2(self):
917
class DocA(Document):
918
structure = {'name':unicode}
919
920
class DocB(Document):
921
structure = {
922
'docs':[{
923
'doca': [DocA],
924
'inc':int,
925
}],
926
}
927
use_autorefs = True
928
929
self.connection.register([DocA, DocB])
930
931
doca = self.col.DocA()
932
doca['_id'] = u'doca'
933
doca['name'] = u"foo"
934
doca.save()
935
936
self.col.insert(
937
{'_id': 'docb', 'docs':[
938
{
939
'doca':[DBRef(database='test', collection='mongokit', id='doca')],
940
'inc':2,
941
},
942
]
943
})
944
assert self.col.DocB.find_one({'_id':'docb'}) == {u'docs': [{u'doca': [{u'_id': u'doca', u'name': u'foo'}], u'inc': 2}], u'_id': u'docb'}
945
946
def test_autorefs_with_required(self):
947
import datetime
948
import uuid
949
950
@self.connection.register
951
class User(Document):
952
structure = {
953
'email': unicode,
954
}
955
956
@self.connection.register
957
class Event(Document):
958
structure = {
959
'user': User,
960
'title': unicode,
961
}
962
required_fields = ['user', 'title']
963
use_autorefs = True
964
965
user = self.connection.test.users.User()
966
user.save()
967
event = self.connection.test.events.Event()
968
event['user'] = user
969
event['title'] = u"Test"
970
event.validate()
971
event.save()
972
Loggerhead is a web-based interface for Breezy
Version: 2.0.1