1# -*- coding: utf-8 -*- 2# 3# Copyright (C) 2007-2009 Christopher Lenz 4# All rights reserved. 5# 6# This software is licensed as described in the file COPYING, which 7# you should have received as part of this distribution. 8 9from decimal import Decimal 10import unittest 11 12from couchdb import design, mapping 13from couchdb.tests import testutil 14from datetime import datetime 15 16class DocumentTestCase(testutil.TempDatabaseMixin, unittest.TestCase): 17 18 def test_mutable_fields(self): 19 class Test(mapping.Document): 20 d = mapping.DictField() 21 a = Test() 22 b = Test() 23 a.d['x'] = True 24 self.assertTrue(a.d.get('x')) 25 self.assertFalse(b.d.get('x')) 26 27 def test_automatic_id(self): 28 class Post(mapping.Document): 29 title = mapping.TextField() 30 post = Post(title='Foo bar') 31 assert post.id is None 32 post.store(self.db) 33 assert post.id is not None 34 self.assertEqual('Foo bar', self.db[post.id]['title']) 35 36 def test_explicit_id_via_init(self): 37 class Post(mapping.Document): 38 title = mapping.TextField() 39 post = Post(id='foo_bar', title='Foo bar') 40 self.assertEqual('foo_bar', post.id) 41 post.store(self.db) 42 self.assertEqual('Foo bar', self.db['foo_bar']['title']) 43 44 def test_explicit_id_via_setter(self): 45 class Post(mapping.Document): 46 title = mapping.TextField() 47 post = Post(title='Foo bar') 48 post.id = 'foo_bar' 49 self.assertEqual('foo_bar', post.id) 50 post.store(self.db) 51 self.assertEqual('Foo bar', self.db['foo_bar']['title']) 52 53 def test_change_id_failure(self): 54 class Post(mapping.Document): 55 title = mapping.TextField() 56 post = Post(title='Foo bar') 57 post.store(self.db) 58 post = Post.load(self.db, post.id) 59 try: 60 post.id = 'foo_bar' 61 self.fail('Excepted AttributeError') 62 except AttributeError as e: 63 self.assertEqual('id can only be set on new documents', e.args[0]) 64 65 def test_batch_update(self): 66 class Post(mapping.Document): 67 title = mapping.TextField() 68 post1 = Post(title='Foo bar') 69 post2 = Post(title='Foo baz') 70 results = self.db.update([post1, post2]) 71 self.assertEqual(2, len(results)) 72 assert results[0][0] is True 73 assert results[1][0] is True 74 75 def test_store_existing(self): 76 class Post(mapping.Document): 77 title = mapping.TextField() 78 post = Post(title='Foo bar') 79 post.store(self.db) 80 post.store(self.db) 81 self.assertEqual(len(list(self.db.view('_all_docs'))), 1) 82 83 def test_old_datetime(self): 84 dt = mapping.DateTimeField() 85 assert dt._to_python('1880-01-01T00:00:00Z') 86 87 def test_datetime_with_microseconds(self): 88 dt = mapping.DateTimeField() 89 assert dt._to_python('2016-06-09T21:21:49.739248Z') 90 91 def test_datetime_to_json(self): 92 dt = mapping.DateTimeField() 93 d = datetime.now() 94 assert dt._to_json(d) 95 96 def test_get_has_default(self): 97 doc = mapping.Document() 98 doc.get('foo') 99 doc.get('foo', None) 100 101 102class ListFieldTestCase(testutil.TempDatabaseMixin, unittest.TestCase): 103 104 def test_to_json(self): 105 # See <http://code.google.com/p/couchdb-python/issues/detail?id=14> 106 class Post(mapping.Document): 107 title = mapping.TextField() 108 comments = mapping.ListField(mapping.DictField( 109 mapping.Mapping.build( 110 author = mapping.TextField(), 111 content = mapping.TextField(), 112 ) 113 )) 114 post = Post(title='Foo bar') 115 post.comments.append(author='myself', content='Bla bla') 116 post.comments = post.comments 117 self.assertEqual([{'content': 'Bla bla', 'author': 'myself'}], 118 post.comments) 119 120 def test_proxy_append(self): 121 class Thing(mapping.Document): 122 numbers = mapping.ListField(mapping.DecimalField) 123 thing = Thing(numbers=[Decimal('1.0'), Decimal('2.0')]) 124 thing.numbers.append(Decimal('3.0')) 125 self.assertEqual(3, len(thing.numbers)) 126 self.assertEqual(Decimal('3.0'), thing.numbers[2]) 127 128 def test_proxy_append_kwargs(self): 129 class Thing(mapping.Document): 130 numbers = mapping.ListField(mapping.DecimalField) 131 thing = Thing() 132 self.assertRaises(TypeError, thing.numbers.append, foo='bar') 133 134 def test_proxy_contains(self): 135 class Thing(mapping.Document): 136 numbers = mapping.ListField(mapping.DecimalField) 137 thing = Thing(numbers=[Decimal('1.0'), Decimal('2.0')]) 138 assert isinstance(thing.numbers, mapping.ListField.Proxy) 139 assert '1.0' not in thing.numbers 140 assert Decimal('1.0') in thing.numbers 141 142 def test_proxy_count(self): 143 class Thing(mapping.Document): 144 numbers = mapping.ListField(mapping.DecimalField) 145 thing = Thing(numbers=[Decimal('1.0'), Decimal('2.0')]) 146 self.assertEqual(1, thing.numbers.count(Decimal('1.0'))) 147 self.assertEqual(0, thing.numbers.count('1.0')) 148 149 def test_proxy_index(self): 150 class Thing(mapping.Document): 151 numbers = mapping.ListField(mapping.DecimalField) 152 thing = Thing(numbers=[Decimal('1.0'), Decimal('2.0')]) 153 self.assertEqual(0, thing.numbers.index(Decimal('1.0'))) 154 self.assertRaises(ValueError, thing.numbers.index, '3.0') 155 156 def test_proxy_insert(self): 157 class Thing(mapping.Document): 158 numbers = mapping.ListField(mapping.DecimalField) 159 thing = Thing(numbers=[Decimal('1.0'), Decimal('2.0')]) 160 thing.numbers.insert(0, Decimal('0.0')) 161 self.assertEqual(3, len(thing.numbers)) 162 self.assertEqual(Decimal('0.0'), thing.numbers[0]) 163 164 def test_proxy_insert_kwargs(self): 165 class Thing(mapping.Document): 166 numbers = mapping.ListField(mapping.DecimalField) 167 thing = Thing() 168 self.assertRaises(TypeError, thing.numbers.insert, 0, foo='bar') 169 170 def test_proxy_remove(self): 171 class Thing(mapping.Document): 172 numbers = mapping.ListField(mapping.DecimalField) 173 thing = Thing() 174 thing.numbers.append(Decimal('1.0')) 175 thing.numbers.remove(Decimal('1.0')) 176 177 def test_proxy_iter(self): 178 class Thing(mapping.Document): 179 numbers = mapping.ListField(mapping.DecimalField) 180 self.db['test'] = {'numbers': ['1.0', '2.0']} 181 thing = Thing.load(self.db, 'test') 182 assert isinstance(thing.numbers[0], Decimal) 183 184 def test_proxy_iter_dict(self): 185 class Post(mapping.Document): 186 comments = mapping.ListField(mapping.DictField) 187 self.db['test'] = {'comments': [{'author': 'Joe', 'content': 'Hey'}]} 188 post = Post.load(self.db, 'test') 189 assert isinstance(post.comments[0], dict) 190 191 def test_proxy_pop(self): 192 class Thing(mapping.Document): 193 numbers = mapping.ListField(mapping.DecimalField) 194 thing = Thing() 195 thing.numbers = [Decimal('%d' % i) for i in range(3)] 196 self.assertEqual(thing.numbers.pop(), Decimal('2.0')) 197 self.assertEqual(len(thing.numbers), 2) 198 self.assertEqual(thing.numbers.pop(0), Decimal('0.0')) 199 200 def test_proxy_slices(self): 201 class Thing(mapping.Document): 202 numbers = mapping.ListField(mapping.DecimalField) 203 thing = Thing() 204 thing.numbers = [Decimal('%d' % i) for i in range(5)] 205 ll = thing.numbers[1:3] 206 self.assertEqual(len(ll), 2) 207 self.assertEqual(ll[0], Decimal('1.0')) 208 thing.numbers[2:4] = [Decimal('%d' % i) for i in range(6, 8)] 209 self.assertEqual(thing.numbers[2], Decimal('6.0')) 210 self.assertEqual(thing.numbers[4], Decimal('4.0')) 211 self.assertEqual(len(thing.numbers), 5) 212 del thing.numbers[3:] 213 self.assertEqual(len(thing.numbers), 3) 214 215 def test_mutable_fields(self): 216 class Thing(mapping.Document): 217 numbers = mapping.ListField(mapping.DecimalField) 218 thing = Thing.wrap({'_id': 'foo', '_rev': 1}) # no numbers 219 thing.numbers.append('1.0') 220 thing2 = Thing(id='thing2') 221 self.assertEqual([i for i in thing2.numbers], []) 222 223 224all_map_func = 'function(doc) { emit(doc._id, doc); }' 225 226 227class WrappingTestCase(testutil.TempDatabaseMixin, unittest.TestCase): 228 229 class Item(mapping.Document): 230 with_include_docs = mapping.ViewField('test', all_map_func, 231 include_docs=True) 232 without_include_docs = mapping.ViewField('test', all_map_func) 233 234 def setUp(self): 235 super(WrappingTestCase, self).setUp() 236 design.ViewDefinition.sync_many( 237 self.db, [self.Item.with_include_docs, 238 self.Item.without_include_docs]) 239 240 def test_viewfield_property(self): 241 self.Item().store(self.db) 242 results = self.Item.with_include_docs(self.db) 243 self.assertEqual(type(results.rows[0]), self.Item) 244 results = self.Item.without_include_docs(self.db) 245 self.assertEqual(type(results.rows[0]), self.Item) 246 247 def test_view(self): 248 self.Item().store(self.db) 249 results = self.Item.view(self.db, 'test/without_include_docs') 250 self.assertEqual(type(results.rows[0]), self.Item) 251 results = self.Item.view(self.db, 'test/without_include_docs', 252 include_docs=True) 253 self.assertEqual(type(results.rows[0]), self.Item) 254 255 def test_wrapped_view(self): 256 self.Item().store(self.db) 257 results = self.db.view('_all_docs', wrapper=self.Item._wrap_row) 258 doc = results.rows[0] 259 self.db.delete(doc) 260 261 def test_query(self): 262 self.Item().store(self.db) 263 results = self.Item.query(self.db, all_map_func, None) 264 self.assertEqual(type(results.rows[0]), self.Item) 265 results = self.Item.query(self.db, all_map_func, None, include_docs=True) 266 self.assertEqual(type(results.rows[0]), self.Item) 267 268 269def suite(): 270 suite = unittest.TestSuite() 271 suite.addTest(testutil.doctest_suite(mapping)) 272 suite.addTest(unittest.makeSuite(DocumentTestCase, 'test')) 273 suite.addTest(unittest.makeSuite(ListFieldTestCase, 'test')) 274 suite.addTest(unittest.makeSuite(WrappingTestCase, 'test')) 275 return suite 276 277 278if __name__ == '__main__': 279 unittest.main(defaultTest='suite') 280