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.osv.orm')
  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.osv.orm')
  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     @mute_logger('openerp.osv.orm')
  81     def test_search_read(self):
  82         # simple search_read
  83         self.partner.create(self.cr, UID, {'name': 'MyPartner1'})
  84         found = self.partner.search_read(self.cr, UID, [['name', '=', 'MyPartner1']], ['name'])
  85         self.assertEqual(len(found), 1)
  86         self.assertEqual(found[0]['name'], 'MyPartner1')
  87         self.assertTrue('id' in found[0])
  88
  89         # search_read correct order
  90         self.partner.create(self.cr, UID, {'name': 'MyPartner2'})
  91         found = self.partner.search_read(self.cr, UID, [['name', 'like', 'MyPartner']], ['name'], order="name")
  92         self.assertEqual(len(found), 2)
  93         self.assertEqual(found[0]['name'], 'MyPartner1')
  94         self.assertEqual(found[1]['name'], 'MyPartner2')
  95         found = self.partner.search_read(self.cr, UID, [['name', 'like', 'MyPartner']], ['name'], order="name desc")
  96         self.assertEqual(len(found), 2)
  97         self.assertEqual(found[0]['name'], 'MyPartner2')
  98         self.assertEqual(found[1]['name'], 'MyPartner1')
  99
 100         # search_read that finds nothing
 101         found = self.partner.search_read(self.cr, UID, [['name', '=', 'Does not exists']], ['name'])
 102         self.assertEqual(len(found), 0)
 103
 104     def test_groupby_date(self):
 105         partners = dict(
 106             A='2012-11-19',
 107             B='2012-12-17',
 108             C='2012-12-31',
 109             D='2013-01-07',
 110             E='2013-01-14',
 111             F='2013-01-28',
 112             G='2013-02-11',
 113         )
 114
 115         all_partners = []
 116         partners_by_day = defaultdict(set)
 117         partners_by_month = defaultdict(set)
 118         partners_by_year = defaultdict(set)
 119
 120         for name, date in partners.items():
 121             p = self.partner.create(self.cr, UID, dict(name=name, date=date))
 122             all_partners.append(p)
 123             partners_by_day[date].add(p)
 124             partners_by_month[date.rsplit('-', 1)[0]].add(p)
 125             partners_by_year[date.split('-', 1)[0]].add(p)
 126
 127         def read_group(interval, domain=None):
 128             main_domain = [('id', 'in', all_partners)]
 129             if domain:
 130                 domain = ['&'] + main_domain + domain
 131             else:
 132                 domain = main_domain
 133
 134             rg = self.partner.read_group(self.cr, self.uid, domain, ['date'], 'date' + ':' + interval)
 135             result = {}
 136             for r in rg:
 137                 result[r['date']] = set(self.partner.search(self.cr, self.uid, r['__domain']))
 138             return result
 139
 140         self.assertEqual(len(read_group('day')), len(partners_by_day))
 141         self.assertEqual(len(read_group('month')), len(partners_by_month))
 142         self.assertEqual(len(read_group('year')), len(partners_by_year))
 143
 144
 145 class TestInherits(common.TransactionCase):
 146     """ test the behavior of the orm for models that use _inherits;
 147         specifically: res.users, that inherits from res.partner
 148     """
 149
 150     def setUp(self):
 151         super(TestInherits, self).setUp()
 152         self.partner = self.registry('res.partner')
 153         self.user = self.registry('res.users')
 154
 155     def test_create(self):
 156         """ creating a user should automatically create a new partner """
 157         partners_before = self.partner.search(self.cr, UID, [])
 158         foo_id = self.user.create(self.cr, UID, {'name': 'Foo', 'login': 'foo', 'password': 'foo'})
 159         foo = self.user.browse(self.cr, UID, foo_id)
 160
 161         self.assertNotIn(foo.partner_id.id, partners_before)
 162
 163     def test_create_with_ancestor(self):
 164         """ creating a user with a specific 'partner_id' should not create a new partner """
 165         par_id = self.partner.create(self.cr, UID, {'name': 'Foo'})
 166         partners_before = self.partner.search(self.cr, UID, [])
 167         foo_id = self.user.create(self.cr, UID, {'partner_id': par_id, 'login': 'foo', 'password': 'foo'})
 168         partners_after = self.partner.search(self.cr, UID, [])
 169
 170         self.assertEqual(set(partners_before), set(partners_after))
 171
 172         foo = self.user.browse(self.cr, UID, foo_id)
 173         self.assertEqual(foo.name, 'Foo')
 174         self.assertEqual(foo.partner_id.id, par_id)
 175
 176     @mute_logger('openerp.osv.orm')
 177     def test_read(self):
 178         """ inherited fields should be read without any indirection """
 179         foo_id = self.user.create(self.cr, UID, {'name': 'Foo', 'login': 'foo', 'password': 'foo'})
 180         foo_values, = self.user.read(self.cr, UID, [foo_id])
 181         partner_id = foo_values['partner_id'][0]
 182         partner_values, = self.partner.read(self.cr, UID, [partner_id])
 183         self.assertEqual(foo_values['name'], partner_values['name'])
 184
 185         foo = self.user.browse(self.cr, UID, foo_id)
 186         self.assertEqual(foo.name, foo.partner_id.name)
 187
 188     @mute_logger('openerp.osv.orm')
 189     def test_copy(self):
 190         """ copying a user should automatically copy its partner, too """
 191         foo_id = self.user.create(self.cr, UID, {'name': 'Foo', 'login': 'foo', 'password': 'foo'})
 192         foo_before, = self.user.read(self.cr, UID, [foo_id])
 193         bar_id = self.user.copy(self.cr, UID, foo_id, {'login': 'bar', 'password': 'bar'})
 194         foo_after, = self.user.read(self.cr, UID, [foo_id])
 195
 196         self.assertEqual(foo_before, foo_after)
 197
 198         foo, bar = self.user.browse(self.cr, UID, [foo_id, bar_id])
 199         self.assertEqual(bar.login, 'bar')
 200         self.assertNotEqual(foo.id, bar.id)
 201         self.assertNotEqual(foo.partner_id.id, bar.partner_id.id)
 202
 203     @mute_logger('openerp.osv.orm')
 204     def test_copy_with_ancestor(self):
 205         """ copying a user with 'parent_id' in defaults should not duplicate the partner """
 206         foo_id = self.user.create(self.cr, UID, {'name': 'Foo', 'login': 'foo', 'password': 'foo'})
 207         par_id = self.partner.create(self.cr, UID, {'name': 'Bar'})
 208
 209         foo_before, = self.user.read(self.cr, UID, [foo_id])
 210         partners_before = self.partner.search(self.cr, UID, [])
 211         bar_id = self.user.copy(self.cr, UID, foo_id, {'partner_id': par_id, 'login': 'bar'})
 212         foo_after, = self.user.read(self.cr, UID, [foo_id])
 213         partners_after = self.partner.search(self.cr, UID, [])
 214
 215         self.assertEqual(foo_before, foo_after)
 216         self.assertEqual(set(partners_before), set(partners_after))
 217
 218         foo, bar = self.user.browse(self.cr, UID, [foo_id, bar_id])
 219         self.assertNotEqual(foo.id, bar.id)
 220         self.assertEqual(bar.partner_id.id, par_id)
 221         self.assertEqual(bar.login, 'bar', "login is given from copy parameters")
 222         self.assertEqual(bar.password, foo.password, "password is given from original record")
 223         self.assertEqual(bar.name, 'Bar', "name is given from specific partner")
 224
 225
 226
 227 CREATE = lambda values: (0, False, values)
 228 UPDATE = lambda id, values: (1, id, values)
 229 DELETE = lambda id: (2, id, False)
 230 FORGET = lambda id: (3, id, False)
 231 LINK_TO = lambda id: (4, id, False)
 232 DELETE_ALL = lambda: (5, False, False)
 233 REPLACE_WITH = lambda ids: (6, False, ids)
 234
 235 def sorted_by_id(list_of_dicts):
 236     "sort dictionaries by their 'id' field; useful for comparisons"
 237     return sorted(list_of_dicts, key=lambda d: d.get('id'))
 238
 239 class TestO2MSerialization(common.TransactionCase):
 240     """ test the orm method 'write' on one2many fields """
 241
 242     def setUp(self):
 243         super(TestO2MSerialization, self).setUp()
 244         self.partner = self.registry('res.partner')
 245
 246     def test_no_command(self):
 247         " empty list of commands yields an empty list of records "
 248         results = self.partner.resolve_2many_commands(
 249             self.cr, UID, 'child_ids', [])
 250
 251         self.assertEqual(results, [])
 252
 253     def test_CREATE_commands(self):
 254         " returns the VALUES dict as-is "
 255         values = [{'foo': 'bar'}, {'foo': 'baz'}, {'foo': 'baq'}]
 256         results = self.partner.resolve_2many_commands(
 257             self.cr, UID, 'child_ids', map(CREATE, values))
 258
 259         self.assertEqual(results, values)
 260
 261     def test_LINK_TO_command(self):
 262         " reads the records from the database, records are returned with their ids. "
 263         ids = [
 264             self.partner.create(self.cr, UID, {'name': 'foo'}),
 265             self.partner.create(self.cr, UID, {'name': 'bar'}),
 266             self.partner.create(self.cr, UID, {'name': 'baz'})
 267         ]
 268         commands = map(LINK_TO, ids)
 269
 270         results = self.partner.resolve_2many_commands(
 271             self.cr, UID, 'child_ids', commands, ['name'])
 272
 273         self.assertEqual(sorted_by_id(results), sorted_by_id([
 274             {'id': ids[0], 'name': 'foo'},
 275             {'id': ids[1], 'name': 'bar'},
 276             {'id': ids[2], 'name': 'baz'}
 277         ]))
 278
 279     def test_bare_ids_command(self):
 280         " same as the equivalent LINK_TO commands "
 281         ids = [
 282             self.partner.create(self.cr, UID, {'name': 'foo'}),
 283             self.partner.create(self.cr, UID, {'name': 'bar'}),
 284             self.partner.create(self.cr, UID, {'name': 'baz'})
 285         ]
 286
 287         results = self.partner.resolve_2many_commands(
 288             self.cr, UID, 'child_ids', ids, ['name'])
 289
 290         self.assertEqual(sorted_by_id(results), sorted_by_id([
 291             {'id': ids[0], 'name': 'foo'},
 292             {'id': ids[1], 'name': 'bar'},
 293             {'id': ids[2], 'name': 'baz'}
 294         ]))
 295
 296     def test_UPDATE_command(self):
 297         " take the in-db records and merge the provided information in "
 298         id_foo = self.partner.create(self.cr, UID, {'name': 'foo'})
 299         id_bar = self.partner.create(self.cr, UID, {'name': 'bar'})
 300         id_baz = self.partner.create(self.cr, UID, {'name': 'baz', 'city': 'tag'})
 301
 302         results = self.partner.resolve_2many_commands(
 303             self.cr, UID, 'child_ids', [
 304                 LINK_TO(id_foo),
 305                 UPDATE(id_bar, {'name': 'qux', 'city': 'tagtag'}),
 306                 UPDATE(id_baz, {'name': 'quux'})
 307             ], ['name', 'city'])
 308
 309         self.assertEqual(sorted_by_id(results), sorted_by_id([
 310             {'id': id_foo, 'name': 'foo', 'city': False},
 311             {'id': id_bar, 'name': 'qux', 'city': 'tagtag'},
 312             {'id': id_baz, 'name': 'quux', 'city': 'tag'}
 313         ]))
 314
 315     def test_DELETE_command(self):
 316         " deleted records are not returned at all. "
 317         ids = [
 318             self.partner.create(self.cr, UID, {'name': 'foo'}),
 319             self.partner.create(self.cr, UID, {'name': 'bar'}),
 320             self.partner.create(self.cr, UID, {'name': 'baz'})
 321         ]
 322         commands = [DELETE(ids[0]), DELETE(ids[1]), DELETE(ids[2])]
 323
 324         results = self.partner.resolve_2many_commands(
 325             self.cr, UID, 'child_ids', commands, ['name'])
 326
 327         self.assertEqual(results, [])
 328
 329     def test_mixed_commands(self):
 330         ids = [
 331             self.partner.create(self.cr, UID, {'name': name})
 332             for name in ['NObar', 'baz', 'qux', 'NOquux', 'NOcorge', 'garply']
 333         ]
 334
 335         results = self.partner.resolve_2many_commands(
 336             self.cr, UID, 'child_ids', [
 337                 CREATE({'name': 'foo'}),
 338                 UPDATE(ids[0], {'name': 'bar'}),
 339                 LINK_TO(ids[1]),
 340                 DELETE(ids[2]),
 341                 UPDATE(ids[3], {'name': 'quux',}),
 342                 UPDATE(ids[4], {'name': 'corge'}),
 343                 CREATE({'name': 'grault'}),
 344                 LINK_TO(ids[5])
 345             ], ['name'])
 346
 347         self.assertEqual(sorted_by_id(results), sorted_by_id([
 348             {'name': 'foo'},
 349             {'id': ids[0], 'name': 'bar'},
 350             {'id': ids[1], 'name': 'baz'},
 351             {'id': ids[3], 'name': 'quux'},
 352             {'id': ids[4], 'name': 'corge'},
 353             {'name': 'grault'},
 354             {'id': ids[5], 'name': 'garply'}
 355         ]))
 356
 357     def test_LINK_TO_pairs(self):
 358         "LINK_TO commands can be written as pairs, instead of triplets"
 359         ids = [
 360             self.partner.create(self.cr, UID, {'name': 'foo'}),
 361             self.partner.create(self.cr, UID, {'name': 'bar'}),
 362             self.partner.create(self.cr, UID, {'name': 'baz'})
 363         ]
 364         commands = map(lambda id: (4, id), ids)
 365
 366         results = self.partner.resolve_2many_commands(
 367             self.cr, UID, 'child_ids', commands, ['name'])
 368
 369         self.assertEqual(sorted_by_id(results), sorted_by_id([
 370             {'id': ids[0], 'name': 'foo'},
 371             {'id': ids[1], 'name': 'bar'},
 372             {'id': ids[2], 'name': 'baz'}
 373         ]))
 374
 375     def test_singleton_commands(self):
 376         "DELETE_ALL can appear as a singleton"
 377         results = self.partner.resolve_2many_commands(
 378             self.cr, UID, 'child_ids', [DELETE_ALL()], ['name'])
 379
 380         self.assertEqual(results, [])
 381
 382 # vim:expandtab:smartindent:tabstop=4:softtabstop=4:shiftwidth=4: