-import unittest2
+from openerp.tools import mute_logger
+import common
+
+UID = common.ADMIN_USER_ID
+DB = common.DB
+
+
+class TestORM(common.TransactionCase):
+ """ test special behaviors of ORM CRUD functions
+
+ TODO: use real Exceptions types instead of Exception """
+
+ def setUp(self):
+ super(TestORM, self).setUp()
+ cr, uid = self.cr, self.uid
+ self.partner = self.registry('res.partner')
+ self.users = self.registry('res.users')
+ self.p1 = self.partner.name_create(cr, uid, 'W')[0]
+ self.p2 = self.partner.name_create(cr, uid, 'Y')[0]
+ self.ir_rule = self.registry('ir.rule')
+
+ # sample unprivileged user
+ employee_gid = self.ref('base.group_user')
+ self.uid2 = self.users.create(cr, uid, {'name': 'test user', 'login': 'test', 'groups_id': [4,employee_gid]})
+
+ @mute_logger('openerp.osv.orm')
+ def testAccessDeletedRecords(self):
+ """ Verify that accessing deleted records works as expected """
+ cr, uid, uid2, p1, p2 = self.cr, self.uid, self.uid2, self.p1, self.p2
+ self.partner.unlink(cr, uid, [p1])
+
+ # read() is expected to skip deleted records because our API is not
+ # transactional for a sequence of search()->read() performed from the
+ # client-side... a concurrent deletion could therefore cause spurious
+ # exceptions even when simply opening a list view!
+ # /!\ Using unprileged user to detect former side effects of ir.rules!
+ self.assertEqual([{'id': p2, 'name': 'Y'}], self.partner.read(cr, uid2, [p1,p2], ['name']), "read() should skip deleted records")
+ self.assertEqual([], self.partner.read(cr, uid2, [p1], ['name']), "read() should skip deleted records")
+
+ # Deleting an already deleted record should be simply ignored
+ self.assertTrue(self.partner.unlink(cr, uid, [p1]), "Re-deleting should be a no-op")
+
+ # Updating an already deleted record should raise, even as admin
+ with self.assertRaises(Exception):
+ self.partner.write(cr, uid, [p1], {'name': 'foo'})
+
+ @mute_logger('openerp.osv.orm')
+ def testAccessFilteredRecords(self):
+ """ Verify that accessing filtered records works as expected for non-admin user """
+ cr, uid, uid2, p1, p2 = self.cr, self.uid, self.uid2, self.p1, self.p2
+ partner_model = self.registry('ir.model').search(cr, uid, [('model','=','res.partner')])[0]
+ self.ir_rule.create(cr, uid, {'name': 'Y is invisible',
+ 'domain_force': [('id', '!=', p1)],
+ 'model_id': partner_model})
+ # search as unprivileged user
+ partners = self.partner.search(cr, uid2, [])
+ self.assertFalse(p1 in partners, "W should not be visible...")
+ self.assertTrue(p2 in partners, "... but Y should be visible")
+
+ # read as unprivileged user
+ with self.assertRaises(Exception):
+ self.partner.read(cr, uid2, [p1], ['name'])
+ # write as unprivileged user
+ with self.assertRaises(Exception):
+ self.partner.write(cr, uid2, [p1], {'name': 'foo'})
+ # unlink as unprivileged user
+ with self.assertRaises(Exception):
+ self.partner.unlink(cr, uid2, [p1])
+
+ # Prepare mixed case
+ self.partner.unlink(cr, uid, [p2])
+ # read mixed records: some deleted and some filtered
+ with self.assertRaises(Exception):
+ self.partner.read(cr, uid2, [p1,p2], ['name'])
+ # delete mixed records: some deleted and some filtered
+ with self.assertRaises(Exception):
+ self.partner.unlink(cr, uid2, [p1,p2])
+
+
+class TestInherits(common.TransactionCase):
+ """ test the behavior of the orm for models that use _inherits;
+ specifically: res.users, that inherits from res.partner
+ """
+
+ def setUp(self):
+ super(TestInherits, self).setUp()
+ self.partner = self.registry('res.partner')
+ self.user = self.registry('res.users')
+
+ def test_create(self):
+ """ creating a user should automatically create a new partner """
+ partners_before = self.partner.search(self.cr, UID, [])
+ foo_id = self.user.create(self.cr, UID, {'name': 'Foo', 'login': 'foo', 'password': 'foo'})
+ foo = self.user.browse(self.cr, UID, foo_id)
+
+ self.assertNotIn(foo.partner_id.id, partners_before)
+
+ def test_create_with_ancestor(self):
+ """ creating a user with a specific 'partner_id' should not create a new partner """
+ par_id = self.partner.create(self.cr, UID, {'name': 'Foo'})
+ partners_before = self.partner.search(self.cr, UID, [])
+ foo_id = self.user.create(self.cr, UID, {'partner_id': par_id, 'login': 'foo', 'password': 'foo'})
+ partners_after = self.partner.search(self.cr, UID, [])
+
+ self.assertEqual(set(partners_before), set(partners_after))
+
+ foo = self.user.browse(self.cr, UID, foo_id)
+ self.assertEqual(foo.name, 'Foo')
+ self.assertEqual(foo.partner_id.id, par_id)
+
+ @mute_logger('openerp.osv.orm')
+ def test_read(self):
+ """ inherited fields should be read without any indirection """
+ foo_id = self.user.create(self.cr, UID, {'name': 'Foo', 'login': 'foo', 'password': 'foo'})
+ foo_values, = self.user.read(self.cr, UID, [foo_id])
+ partner_id = foo_values['partner_id'][0]
+ partner_values, = self.partner.read(self.cr, UID, [partner_id])
+ self.assertEqual(foo_values['name'], partner_values['name'])
+
+ foo = self.user.browse(self.cr, UID, foo_id)
+ self.assertEqual(foo.name, foo.partner_id.name)
+
+ @mute_logger('openerp.osv.orm')
+ def test_copy(self):
+ """ copying a user should automatically copy its partner, too """
+ foo_id = self.user.create(self.cr, UID, {'name': 'Foo', 'login': 'foo', 'password': 'foo'})
+ foo_before, = self.user.read(self.cr, UID, [foo_id])
+ bar_id = self.user.copy(self.cr, UID, foo_id, {'login': 'bar', 'password': 'bar'})
+ foo_after, = self.user.read(self.cr, UID, [foo_id])
+
+ self.assertEqual(foo_before, foo_after)
+
+ foo, bar = self.user.browse(self.cr, UID, [foo_id, bar_id])
+ self.assertEqual(bar.login, 'bar')
+ self.assertNotEqual(foo.id, bar.id)
+ self.assertNotEqual(foo.partner_id.id, bar.partner_id.id)
+
+ @mute_logger('openerp.osv.orm')
+ def test_copy_with_ancestor(self):
+ """ copying a user with 'parent_id' in defaults should not duplicate the partner """
+ foo_id = self.user.create(self.cr, UID, {'name': 'Foo', 'login': 'foo', 'password': 'foo'})
+ par_id = self.partner.create(self.cr, UID, {'name': 'Bar'})
+
+ foo_before, = self.user.read(self.cr, UID, [foo_id])
+ partners_before = self.partner.search(self.cr, UID, [])
+ bar_id = self.user.copy(self.cr, UID, foo_id, {'partner_id': par_id, 'login': 'bar'})
+ foo_after, = self.user.read(self.cr, UID, [foo_id])
+ partners_after = self.partner.search(self.cr, UID, [])
+
+ self.assertEqual(foo_before, foo_after)
+ self.assertEqual(set(partners_before), set(partners_after))
+
+ foo, bar = self.user.browse(self.cr, UID, [foo_id, bar_id])
+ self.assertNotEqual(foo.id, bar.id)
+ self.assertEqual(bar.partner_id.id, par_id)
+ self.assertEqual(bar.login, 'bar', "login is given from copy parameters")
+ self.assertEqual(bar.password, foo.password, "password is given from original record")
+ self.assertEqual(bar.name, 'Bar', "name is given from specific partner")
-import openerp
-UID = 1
-DB = openerp.tools.config['db_name']
CREATE = lambda values: (0, False, values)
UPDATE = lambda id, values: (1, id, values)
DELETE_ALL = lambda: (5, False, False)
REPLACE_WITH = lambda ids: (6, False, ids)
-class TestO2MSerialization(unittest2.TestCase):
+def sorted_by_id(list_of_dicts):
+ "sort dictionaries by their 'id' field; useful for comparisons"
+ return sorted(list_of_dicts, key=lambda d: d.get('id'))
- def setUp(self):
- self.cr = openerp.modules.registry.RegistryManager.get(DB).db.cursor()
- self.partner = openerp.modules.registry.RegistryManager.get(DB)['res.partner']
- self.address = openerp.modules.registry.RegistryManager.get(DB)['res.partner.address']
+class TestO2MSerialization(common.TransactionCase):
+ """ test the orm method 'write' on one2many fields """
- def tearDown(self):
- self.cr.rollback()
- self.cr.close()
+ def setUp(self):
+ super(TestO2MSerialization, self).setUp()
+ self.partner = self.registry('res.partner')
def test_no_command(self):
" empty list of commands yields an empty list of records "
- results = self.partner.resolve_o2m_commands_to_record_dicts(
- self.cr, UID, 'address', [])
+ results = self.partner.resolve_2many_commands(
+ self.cr, UID, 'child_ids', [])
self.assertEqual(results, [])
def test_CREATE_commands(self):
" returns the VALUES dict as-is "
- results = self.partner.resolve_o2m_commands_to_record_dicts(
- self.cr, UID, 'address',
- map(CREATE, [{'foo': 'bar'}, {'foo': 'baz'}, {'foo': 'baq'}]))
- self.assertEqual(results, [
- {'foo': 'bar'},
- {'foo': 'baz'},
- {'foo': 'baq'}
- ])
+ values = [{'foo': 'bar'}, {'foo': 'baz'}, {'foo': 'baq'}]
+ results = self.partner.resolve_2many_commands(
+ self.cr, UID, 'child_ids', map(CREATE, values))
+
+ self.assertEqual(results, values)
def test_LINK_TO_command(self):
" reads the records from the database, records are returned with their ids. "
ids = [
- self.address.create(self.cr, UID, {'name': 'foo'}),
- self.address.create(self.cr, UID, {'name': 'bar'}),
- self.address.create(self.cr, UID, {'name': 'baz'})
+ self.partner.create(self.cr, UID, {'name': 'foo'}),
+ self.partner.create(self.cr, UID, {'name': 'bar'}),
+ self.partner.create(self.cr, UID, {'name': 'baz'})
]
commands = map(LINK_TO, ids)
- results = self.partner.resolve_o2m_commands_to_record_dicts(
- self.cr, UID, 'address', commands, ['name'])
+ results = self.partner.resolve_2many_commands(
+ self.cr, UID, 'child_ids', commands, ['name'])
- self.assertEqual(results, [
+ self.assertEqual(sorted_by_id(results), sorted_by_id([
{'id': ids[0], 'name': 'foo'},
{'id': ids[1], 'name': 'bar'},
{'id': ids[2], 'name': 'baz'}
- ])
+ ]))
def test_bare_ids_command(self):
" same as the equivalent LINK_TO commands "
ids = [
- self.address.create(self.cr, UID, {'name': 'foo'}),
- self.address.create(self.cr, UID, {'name': 'bar'}),
- self.address.create(self.cr, UID, {'name': 'baz'})
+ self.partner.create(self.cr, UID, {'name': 'foo'}),
+ self.partner.create(self.cr, UID, {'name': 'bar'}),
+ self.partner.create(self.cr, UID, {'name': 'baz'})
]
- results = self.partner.resolve_o2m_commands_to_record_dicts(
- self.cr, UID, 'address', ids, ['name'])
+ results = self.partner.resolve_2many_commands(
+ self.cr, UID, 'child_ids', ids, ['name'])
- self.assertEqual(results, [
+ self.assertEqual(sorted_by_id(results), sorted_by_id([
{'id': ids[0], 'name': 'foo'},
{'id': ids[1], 'name': 'bar'},
{'id': ids[2], 'name': 'baz'}
- ])
+ ]))
def test_UPDATE_command(self):
" take the in-db records and merge the provided information in "
- id_foo = self.address.create(self.cr, UID, {'name': 'foo'})
- id_bar = self.address.create(self.cr, UID, {'name': 'bar'})
- id_baz = self.address.create(self.cr, UID, {'name': 'baz', 'city': 'tag'})
+ id_foo = self.partner.create(self.cr, UID, {'name': 'foo'})
+ id_bar = self.partner.create(self.cr, UID, {'name': 'bar'})
+ id_baz = self.partner.create(self.cr, UID, {'name': 'baz', 'city': 'tag'})
- results = self.partner.resolve_o2m_commands_to_record_dicts(
- self.cr, UID, 'address', [
+ results = self.partner.resolve_2many_commands(
+ self.cr, UID, 'child_ids', [
LINK_TO(id_foo),
UPDATE(id_bar, {'name': 'qux', 'city': 'tagtag'}),
UPDATE(id_baz, {'name': 'quux'})
], ['name', 'city'])
- self.assertEqual(results, [
+ self.assertEqual(sorted_by_id(results), sorted_by_id([
{'id': id_foo, 'name': 'foo', 'city': False},
{'id': id_bar, 'name': 'qux', 'city': 'tagtag'},
{'id': id_baz, 'name': 'quux', 'city': 'tag'}
- ])
+ ]))
+
+ def test_DELETE_command(self):
+ " deleted records are not returned at all. "
+ ids = [
+ self.partner.create(self.cr, UID, {'name': 'foo'}),
+ self.partner.create(self.cr, UID, {'name': 'bar'}),
+ self.partner.create(self.cr, UID, {'name': 'baz'})
+ ]
+ commands = [DELETE(ids[0]), DELETE(ids[1]), DELETE(ids[2])]
+
+ results = self.partner.resolve_2many_commands(
+ self.cr, UID, 'child_ids', commands, ['name'])
+
+ self.assertEqual(results, [])
def test_mixed_commands(self):
ids = [
- self.address.create(self.cr, UID, {'name': name})
+ self.partner.create(self.cr, UID, {'name': name})
for name in ['NObar', 'baz', 'qux', 'NOquux', 'NOcorge', 'garply']
]
- results = self.partner.resolve_o2m_commands_to_record_dicts(
- self.cr, UID, 'address', [
+ results = self.partner.resolve_2many_commands(
+ self.cr, UID, 'child_ids', [
CREATE({'name': 'foo'}),
UPDATE(ids[0], {'name': 'bar'}),
LINK_TO(ids[1]),
- LINK_TO(ids[2]),
+ DELETE(ids[2]),
UPDATE(ids[3], {'name': 'quux',}),
UPDATE(ids[4], {'name': 'corge'}),
CREATE({'name': 'grault'}),
LINK_TO(ids[5])
], ['name'])
- self.assertEqual(results, [
+ self.assertEqual(sorted_by_id(results), sorted_by_id([
{'name': 'foo'},
{'id': ids[0], 'name': 'bar'},
{'id': ids[1], 'name': 'baz'},
- {'id': ids[2], 'name': 'qux'},
{'id': ids[3], 'name': 'quux'},
{'id': ids[4], 'name': 'corge'},
{'name': 'grault'},
{'id': ids[5], 'name': 'garply'}
- ])
+ ]))
def test_LINK_TO_pairs(self):
"LINK_TO commands can be written as pairs, instead of triplets"
ids = [
- self.address.create(self.cr, UID, {'name': 'foo'}),
- self.address.create(self.cr, UID, {'name': 'bar'}),
- self.address.create(self.cr, UID, {'name': 'baz'})
+ self.partner.create(self.cr, UID, {'name': 'foo'}),
+ self.partner.create(self.cr, UID, {'name': 'bar'}),
+ self.partner.create(self.cr, UID, {'name': 'baz'})
]
commands = map(lambda id: (4, id), ids)
- results = self.partner.resolve_o2m_commands_to_record_dicts(
- self.cr, UID, 'address', commands, ['name'])
+ results = self.partner.resolve_2many_commands(
+ self.cr, UID, 'child_ids', commands, ['name'])
- self.assertEqual(results, [
+ self.assertEqual(sorted_by_id(results), sorted_by_id([
{'id': ids[0], 'name': 'foo'},
{'id': ids[1], 'name': 'bar'},
{'id': ids[2], 'name': 'baz'}
- ])
+ ]))
def test_singleton_commands(self):
"DELETE_ALL can appear as a singleton"
+ results = self.partner.resolve_2many_commands(
+ self.cr, UID, 'child_ids', [DELETE_ALL()], ['name'])
- try:
- self.partner.resolve_o2m_commands_to_record_dicts(
- self.cr, UID, 'address', [(5,)], ['name'])
- except AssertionError:
- # 5 should fail with an assert error, but not e.g. a ValueError
- pass
-
- def test_invalid_commands(self):
- "Commands with uncertain semantics in this context should be forbidden"
-
- with self.assertRaises(AssertionError):
- self.partner.resolve_o2m_commands_to_record_dicts(
- self.cr, UID, 'address', [DELETE(42)], ['name'])
-
- with self.assertRaises(AssertionError):
- self.partner.resolve_o2m_commands_to_record_dicts(
- self.cr, UID, 'address', [FORGET(42)], ['name'])
-
- with self.assertRaises(AssertionError):
- self.partner.resolve_o2m_commands_to_record_dicts(
- self.cr, UID, 'address', [DELETE_ALL()], ['name'])
-
- with self.assertRaises(AssertionError):
- self.partner.resolve_o2m_commands_to_record_dicts(
- self.cr, UID, 'address', [REPLACE_WITH([42])], ['name'])
-
+ self.assertEqual(results, [])
# vim:expandtab:smartindent:tabstop=4:softtabstop=4:shiftwidth=4: