openerp/addons/base/tests/test_orm.py

   1 from collections import defaultdict
   2 from openerp.tools import mute_logger
   3 from openerp.tests import common
   4
   5 UID = common.ADMIN_USER_ID
   6 DB = common.DB
   7
   8
   9 class TestORM(common.TransactionCase):
  10     """ test special behaviors of ORM CRUD functions
  11
  12         TODO: use real Exceptions types instead of Exception """
  13
  14     def setUp(self):
  15         super(TestORM, self).setUp()
  16         cr, uid = self.cr, self.uid
  17         self.partner = self.registry('res.partner')
  18         self.users = self.registry('res.users')
  19         self.p1 = self.partner.name_create(cr, uid, 'W')[0]
  20         self.p2 = self.partner.name_create(cr, uid, 'Y')[0]
  21         self.ir_rule = self.registry('ir.rule')
  22
  23         # sample unprivileged user
  24         employee_gid = self.ref('base.group_user')
  25         self.uid2 = self.users.create(cr, uid, {'name': 'test user', 'login': 'test', 'groups_id': [4,employee_gid]})
  26
  27     @mute_logger('openerp.models')
  28     def testAccessDeletedRecords(self):
  29         """ Verify that accessing deleted records works as expected """
  30         cr, uid, uid2, p1, p2 = self.cr, self.uid, self.uid2, self.p1, self.p2
  31         self.partner.unlink(cr, uid, [p1])
  32
  33         # read() is expected to skip deleted records because our API is not
  34         # transactional for a sequence of search()->read() performed from the
  35         # client-side... a concurrent deletion could therefore cause spurious
  36         # exceptions even when simply opening a list view!
  37         # /!\ Using unprileged user to detect former side effects of ir.rules!
  38         self.assertEqual([{'id': p2, 'name': 'Y'}], self.partner.read(cr, uid2, [p1,p2], ['name']), "read() should skip deleted records")
  39         self.assertEqual([], self.partner.read(cr, uid2, [p1], ['name']), "read() should skip deleted records")
  40
  41         # Deleting an already deleted record should be simply ignored
  42         self.assertTrue(self.partner.unlink(cr, uid, [p1]), "Re-deleting should be a no-op")
  43
  44         # Updating an already deleted record should raise, even as admin
  45         with self.assertRaises(Exception):
  46             self.partner.write(cr, uid, [p1], {'name': 'foo'})
  47
  48     @mute_logger('openerp.models')
  49     def testAccessFilteredRecords(self):
  50         """ Verify that accessing filtered records works as expected for non-admin user """
  51         cr, uid, uid2, p1, p2 = self.cr, self.uid, self.uid2, self.p1, self.p2
  52         partner_model = self.registry('ir.model').search(cr, uid, [('model','=','res.partner')])[0]
  53         self.ir_rule.create(cr, uid, {'name': 'Y is invisible',
  54                                       'domain_force': [('id', '!=', p1)],
  55                                       'model_id': partner_model})
  56         # search as unprivileged user
  57         partners = self.partner.search(cr, uid2, [])
  58         self.assertFalse(p1 in partners, "W should not be visible...")
  59         self.assertTrue(p2 in partners, "... but Y should be visible")
  60
  61         # read as unprivileged user
  62         with self.assertRaises(Exception):
  63             self.partner.read(cr, uid2, [p1], ['name'])
  64         # write as unprivileged user
  65         with self.assertRaises(Exception):
  66             self.partner.write(cr, uid2, [p1], {'name': 'foo'})
  67         # unlink as unprivileged user
  68         with self.assertRaises(Exception):
  69             self.partner.unlink(cr, uid2, [p1])
  70
  71         # Prepare mixed case
  72         self.partner.unlink(cr, uid, [p2])
  73         # read mixed records: some deleted and some filtered
  74         with self.assertRaises(Exception):
  75             self.partner.read(cr, uid2, [p1,p2], ['name'])
  76         # delete mixed records: some deleted and some filtered
  77         with self.assertRaises(Exception):
  78             self.partner.unlink(cr, uid2, [p1,p2])
  79
  80     def test_multi_read(self):
  81         record_id = self.partner.create(self.cr, UID, {'name': 'MyPartner1'})
  82         records = self.partner.read(self.cr, UID, [record_id])
  83         self.assertIsInstance(records, list)
  84
  85     def test_one_read(self):
  86         record_id = self.partner.create(self.cr, UID, {'name': 'MyPartner1'})
  87         record = self.partner.read(self.cr, UID, record_id)
  88         self.assertIsInstance(record, dict)
  89
  90     @mute_logger('openerp.models')
  91     def test_search_read(self):
  92         # simple search_read
  93         self.partner.create(self.cr, UID, {'name': 'MyPartner1'})
  94         found = self.partner.search_read(self.cr, UID, [['name', '=', 'MyPartner1']], ['name'])
  95         self.assertEqual(len(found), 1)
  96         self.assertEqual(found[0]['name'], 'MyPartner1')
  97         self.assertTrue('id' in found[0])
  98
  99         # search_read correct order
 100         self.partner.create(self.cr, UID, {'name': 'MyPartner2'})
 101         found = self.partner.search_read(self.cr, UID, [['name', 'like', 'MyPartner']], ['name'], order="name")
 102         self.assertEqual(len(found), 2)
 103         self.assertEqual(found[0]['name'], 'MyPartner1')
 104         self.assertEqual(found[1]['name'], 'MyPartner2')
 105         found = self.partner.search_read(self.cr, UID, [['name', 'like', 'MyPartner']], ['name'], order="name desc")
 106         self.assertEqual(len(found), 2)
 107         self.assertEqual(found[0]['name'], 'MyPartner2')
 108         self.assertEqual(found[1]['name'], 'MyPartner1')
 109
 110         # search_read that finds nothing
 111         found = self.partner.search_read(self.cr, UID, [['name', '=', 'Does not exists']], ['name'])
 112         self.assertEqual(len(found), 0)
 113
 114     def test_exists(self):
 115         partner = self.partner.browse(self.cr, UID, [])
 116
 117         # check that records obtained from search exist
 118         recs = partner.search([])
 119         self.assertTrue(recs)
 120         self.assertEqual(recs.exists(), recs)
 121
 122         # check that there is no record with id 0
 123         recs = partner.browse([0])
 124         self.assertFalse(recs.exists())
 125
 126     def test_groupby_date(self):
 127         partners = dict(
 128             A='2012-11-19',
 129             B='2012-12-17',
 130             C='2012-12-31',
 131             D='2013-01-07',
 132             E='2013-01-14',
 133             F='2013-01-28',
 134             G='2013-02-11',
 135         )
 136
 137         all_partners = []
 138         partners_by_day = defaultdict(set)
 139         partners_by_month = defaultdict(set)
 140         partners_by_year = defaultdict(set)
 141
 142         for name, date in partners.items():
 143             p = self.partner.create(self.cr, UID, dict(name=name, date=date))
 144             all_partners.append(p)
 145             partners_by_day[date].add(p)
 146             partners_by_month[date.rsplit('-', 1)[0]].add(p)
 147             partners_by_year[date.split('-', 1)[0]].add(p)
 148
 149         def read_group(interval, domain=None):
 150             main_domain = [('id', 'in', all_partners)]
 151             if domain:
 152                 domain = ['&'] + main_domain + domain
 153             else:
 154                 domain = main_domain
 155
 156             rg = self.partner.read_group(self.cr, self.uid, domain, ['date'], 'date' + ':' + interval)
 157             result = {}
 158             for r in rg:
 159                 result[r['date:' + interval]] = set(self.partner.search(self.cr, self.uid, r['__domain']))
 160             return result
 161
 162         self.assertEqual(len(read_group('day')), len(partners_by_day))
 163         self.assertEqual(len(read_group('month')), len(partners_by_month))
 164         self.assertEqual(len(read_group('year')), len(partners_by_year))
 165
 166         rg = self.partner.read_group(self.cr, self.uid, [('id', 'in', all_partners)],
 167                         ['date'], ['date:month', 'date:day'], lazy=False)
 168         self.assertEqual(len(rg), len(all_partners))
 169
 170
 171 class TestInherits(common.TransactionCase):
 172     """ test the behavior of the orm for models that use _inherits;
 173         specifically: res.users, that inherits from res.partner
 174     """
 175
 176     def setUp(self):
 177         super(TestInherits, self).setUp()
 178         self.partner = self.registry('res.partner')
 179         self.user = self.registry('res.users')
 180
 181     def test_default(self):
 182         """ `default_get` cannot return a dictionary or a new id """
 183         defaults = self.user.default_get(self.cr, UID, ['partner_id'])
 184         if 'partner_id' in defaults:
 185             self.assertIsInstance(defaults['partner_id'], (bool, int, long))
 186
 187     def test_create(self):
 188         """ creating a user should automatically create a new partner """
 189         partners_before = self.partner.search(self.cr, UID, [])
 190         foo_id = self.user.create(self.cr, UID, {'name': 'Foo', 'login': 'foo', 'password': 'foo'})
 191         foo = self.user.browse(self.cr, UID, foo_id)
 192
 193         self.assertNotIn(foo.partner_id.id, partners_before)
 194
 195     def test_create_with_ancestor(self):
 196         """ creating a user with a specific 'partner_id' should not create a new partner """
 197         par_id = self.partner.create(self.cr, UID, {'name': 'Foo'})
 198         partners_before = self.partner.search(self.cr, UID, [])
 199         foo_id = self.user.create(self.cr, UID, {'partner_id': par_id, 'login': 'foo', 'password': 'foo'})
 200         partners_after = self.partner.search(self.cr, UID, [])
 201
 202         self.assertEqual(set(partners_before), set(partners_after))
 203
 204         foo = self.user.browse(self.cr, UID, foo_id)
 205         self.assertEqual(foo.name, 'Foo')
 206         self.assertEqual(foo.partner_id.id, par_id)
 207
 208     @mute_logger('openerp.models')
 209     def test_read(self):
 210         """ inherited fields should be read without any indirection """
 211         foo_id = self.user.create(self.cr, UID, {'name': 'Foo', 'login': 'foo', 'password': 'foo'})
 212         foo_values, = self.user.read(self.cr, UID, [foo_id])
 213         partner_id = foo_values['partner_id'][0]
 214         partner_values, = self.partner.read(self.cr, UID, [partner_id])
 215         self.assertEqual(foo_values['name'], partner_values['name'])
 216
 217         foo = self.user.browse(self.cr, UID, foo_id)
 218         self.assertEqual(foo.name, foo.partner_id.name)
 219
 220     @mute_logger('openerp.models')
 221     def test_copy(self):
 222         """ copying a user should automatically copy its partner, too """
 223         foo_id = self.user.create(self.cr, UID, {'name': 'Foo', 'login': 'foo', 'password': 'foo'})
 224         foo_before, = self.user.read(self.cr, UID, [foo_id])
 225         del foo_before['__last_update']
 226         bar_id = self.user.copy(self.cr, UID, foo_id, {'login': 'bar', 'password': 'bar'})
 227         foo_after, = self.user.read(self.cr, UID, [foo_id])
 228         del foo_after['__last_update']
 229
 230         self.assertEqual(foo_before, foo_after)
 231
 232         foo, bar = self.user.browse(self.cr, UID, [foo_id, bar_id])
 233         self.assertEqual(bar.login, 'bar')
 234         self.assertNotEqual(foo.id, bar.id)
 235         self.assertNotEqual(foo.partner_id.id, bar.partner_id.id)
 236
 237     @mute_logger('openerp.models')
 238     def test_copy_with_ancestor(self):
 239         """ copying a user with 'parent_id' in defaults should not duplicate the partner """
 240         foo_id = self.user.create(self.cr, UID, {'name': 'Foo', 'login': 'foo', 'password': 'foo',
 241                                                  'login_date': '2016-01-01', 'signature': 'XXX'})
 242         par_id = self.partner.create(self.cr, UID, {'name': 'Bar'})
 243
 244         foo_before, = self.user.read(self.cr, UID, [foo_id])
 245         del foo_before['__last_update']
 246         partners_before = self.partner.search(self.cr, UID, [])
 247         bar_id = self.user.copy(self.cr, UID, foo_id, {'partner_id': par_id, 'login': 'bar'})
 248         foo_after, = self.user.read(self.cr, UID, [foo_id])
 249         del foo_after['__last_update']
 250         partners_after = self.partner.search(self.cr, UID, [])
 251
 252         self.assertEqual(foo_before, foo_after)
 253         self.assertEqual(set(partners_before), set(partners_after))
 254
 255         foo, bar = self.user.browse(self.cr, UID, [foo_id, bar_id])
 256         self.assertNotEqual(foo.id, bar.id)
 257         self.assertEqual(bar.partner_id.id, par_id)
 258         self.assertEqual(bar.login, 'bar', "login is given from copy parameters")
 259         self.assertFalse(bar.login_date, "login_date should not be copied from original record")
 260         self.assertEqual(bar.name, 'Bar', "name is given from specific partner")
 261         self.assertEqual(bar.signature, foo.signature, "signature should be copied")
 262
 263
 264
 265 CREATE = lambda values: (0, False, values)
 266 UPDATE = lambda id, values: (1, id, values)
 267 DELETE = lambda id: (2, id, False)
 268 FORGET = lambda id: (3, id, False)
 269 LINK_TO = lambda id: (4, id, False)
 270 DELETE_ALL = lambda: (5, False, False)
 271 REPLACE_WITH = lambda ids: (6, False, ids)
 272
 273 def sorted_by_id(list_of_dicts):
 274     "sort dictionaries by their 'id' field; useful for comparisons"
 275     return sorted(list_of_dicts, key=lambda d: d.get('id'))
 276
 277 class TestO2MSerialization(common.TransactionCase):
 278     """ test the orm method 'write' on one2many fields """
 279
 280     def setUp(self):
 281         super(TestO2MSerialization, self).setUp()
 282         self.partner = self.registry('res.partner')
 283
 284     def test_no_command(self):
 285         " empty list of commands yields an empty list of records "
 286         results = self.partner.resolve_2many_commands(
 287             self.cr, UID, 'child_ids', [])
 288
 289         self.assertEqual(results, [])
 290
 291     def test_CREATE_commands(self):
 292         " returns the VALUES dict as-is "
 293         values = [{'foo': 'bar'}, {'foo': 'baz'}, {'foo': 'baq'}]
 294         results = self.partner.resolve_2many_commands(
 295             self.cr, UID, 'child_ids', map(CREATE, values))
 296
 297         self.assertEqual(results, values)
 298
 299     def test_LINK_TO_command(self):
 300         " reads the records from the database, records are returned with their ids. "
 301         ids = [
 302             self.partner.create(self.cr, UID, {'name': 'foo'}),
 303             self.partner.create(self.cr, UID, {'name': 'bar'}),
 304             self.partner.create(self.cr, UID, {'name': 'baz'})
 305         ]
 306         commands = map(LINK_TO, ids)
 307
 308         results = self.partner.resolve_2many_commands(
 309             self.cr, UID, 'child_ids', commands, ['name'])
 310
 311         self.assertEqual(sorted_by_id(results), sorted_by_id([
 312             {'id': ids[0], 'name': 'foo'},
 313             {'id': ids[1], 'name': 'bar'},
 314             {'id': ids[2], 'name': 'baz'}
 315         ]))
 316
 317     def test_bare_ids_command(self):
 318         " same as the equivalent LINK_TO commands "
 319         ids = [
 320             self.partner.create(self.cr, UID, {'name': 'foo'}),
 321             self.partner.create(self.cr, UID, {'name': 'bar'}),
 322             self.partner.create(self.cr, UID, {'name': 'baz'})
 323         ]
 324
 325         results = self.partner.resolve_2many_commands(
 326             self.cr, UID, 'child_ids', ids, ['name'])
 327
 328         self.assertEqual(sorted_by_id(results), sorted_by_id([
 329             {'id': ids[0], 'name': 'foo'},
 330             {'id': ids[1], 'name': 'bar'},
 331             {'id': ids[2], 'name': 'baz'}
 332         ]))
 333
 334     def test_UPDATE_command(self):
 335         " take the in-db records and merge the provided information in "
 336         id_foo = self.partner.create(self.cr, UID, {'name': 'foo'})
 337         id_bar = self.partner.create(self.cr, UID, {'name': 'bar'})
 338         id_baz = self.partner.create(self.cr, UID, {'name': 'baz', 'city': 'tag'})
 339
 340         results = self.partner.resolve_2many_commands(
 341             self.cr, UID, 'child_ids', [
 342                 LINK_TO(id_foo),
 343                 UPDATE(id_bar, {'name': 'qux', 'city': 'tagtag'}),
 344                 UPDATE(id_baz, {'name': 'quux'})
 345             ], ['name', 'city'])
 346
 347         self.assertEqual(sorted_by_id(results), sorted_by_id([
 348             {'id': id_foo, 'name': 'foo', 'city': False},
 349             {'id': id_bar, 'name': 'qux', 'city': 'tagtag'},
 350             {'id': id_baz, 'name': 'quux', 'city': 'tag'}
 351         ]))
 352
 353     def test_DELETE_command(self):
 354         " deleted records are not returned at all. "
 355         ids = [
 356             self.partner.create(self.cr, UID, {'name': 'foo'}),
 357             self.partner.create(self.cr, UID, {'name': 'bar'}),
 358             self.partner.create(self.cr, UID, {'name': 'baz'})
 359         ]
 360         commands = [DELETE(ids[0]), DELETE(ids[1]), DELETE(ids[2])]
 361
 362         results = self.partner.resolve_2many_commands(
 363             self.cr, UID, 'child_ids', commands, ['name'])
 364
 365         self.assertEqual(results, [])
 366
 367     def test_mixed_commands(self):
 368         ids = [
 369             self.partner.create(self.cr, UID, {'name': name})
 370             for name in ['NObar', 'baz', 'qux', 'NOquux', 'NOcorge', 'garply']
 371         ]
 372
 373         results = self.partner.resolve_2many_commands(
 374             self.cr, UID, 'child_ids', [
 375                 CREATE({'name': 'foo'}),
 376                 UPDATE(ids[0], {'name': 'bar'}),
 377                 LINK_TO(ids[1]),
 378                 DELETE(ids[2]),
 379                 UPDATE(ids[3], {'name': 'quux',}),
 380                 UPDATE(ids[4], {'name': 'corge'}),
 381                 CREATE({'name': 'grault'}),
 382                 LINK_TO(ids[5])
 383             ], ['name'])
 384
 385         self.assertEqual(sorted_by_id(results), sorted_by_id([
 386             {'name': 'foo'},
 387             {'id': ids[0], 'name': 'bar'},
 388             {'id': ids[1], 'name': 'baz'},
 389             {'id': ids[3], 'name': 'quux'},
 390             {'id': ids[4], 'name': 'corge'},
 391             {'name': 'grault'},
 392             {'id': ids[5], 'name': 'garply'}
 393         ]))
 394
 395     def test_LINK_TO_pairs(self):
 396         "LINK_TO commands can be written as pairs, instead of triplets"
 397         ids = [
 398             self.partner.create(self.cr, UID, {'name': 'foo'}),
 399             self.partner.create(self.cr, UID, {'name': 'bar'}),
 400             self.partner.create(self.cr, UID, {'name': 'baz'})
 401         ]
 402         commands = map(lambda id: (4, id), ids)
 403
 404         results = self.partner.resolve_2many_commands(
 405             self.cr, UID, 'child_ids', commands, ['name'])
 406
 407         self.assertEqual(sorted_by_id(results), sorted_by_id([
 408             {'id': ids[0], 'name': 'foo'},
 409             {'id': ids[1], 'name': 'bar'},
 410             {'id': ids[2], 'name': 'baz'}
 411         ]))
 412
 413     def test_singleton_commands(self):
 414         "DELETE_ALL can appear as a singleton"
 415         results = self.partner.resolve_2many_commands(
 416             self.cr, UID, 'child_ids', [DELETE_ALL()], ['name'])
 417
 418         self.assertEqual(results, [])
 419
 420 # vim:expandtab:smartindent:tabstop=4:softtabstop=4:shiftwidth=4: