239 lines
8.6 KiB
Python
239 lines
8.6 KiB
Python
|
# -*- coding: utf-8 -*-
|
||
|
|
# Part of Odoo. See LICENSE file for full copyright and licensing details.
|
||
|
|
import collections
|
||
|
|
import time
|
||
|
|
from xmlrpc.client import Binary
|
||
|
|
|
||
|
|
from odoo.exceptions import AccessDenied, AccessError
|
||
|
|
from odoo.http import _request_stack
|
||
|
|
|
||
|
|
import odoo
|
||
|
|
import odoo.tools
|
||
|
|
from odoo.tests import common
|
||
|
|
from odoo.service import common as auth, model
|
||
|
|
from odoo.tools import DotDict
|
||
|
|
|
||
|
|
|
||
|
|
@common.tagged('post_install', '-at_install')
|
||
|
|
class TestXMLRPC(common.HttpCase):
|
||
|
|
|
||
|
|
def setUp(self):
|
||
|
|
super(TestXMLRPC, self).setUp()
|
||
|
|
self.admin_uid = self.env.ref('base.user_admin').id
|
||
|
|
|
||
|
|
def xmlrpc(self, model, method, *args, **kwargs):
|
||
|
|
return self.xmlrpc_object.execute_kw(
|
||
|
|
common.get_db_name(), self.admin_uid, 'admin',
|
||
|
|
model, method, args, kwargs
|
||
|
|
)
|
||
|
|
|
||
|
|
def test_01_xmlrpc_login(self):
|
||
|
|
""" Try to login on the common service. """
|
||
|
|
db_name = common.get_db_name()
|
||
|
|
uid = self.xmlrpc_common.login(db_name, 'admin', 'admin')
|
||
|
|
self.assertEqual(uid, self.admin_uid)
|
||
|
|
|
||
|
|
def test_xmlrpc_ir_model_search(self):
|
||
|
|
""" Try a search on the object service. """
|
||
|
|
o = self.xmlrpc_object
|
||
|
|
db_name = common.get_db_name()
|
||
|
|
ids = o.execute(db_name, self.admin_uid, 'admin', 'ir.model', 'search', [])
|
||
|
|
self.assertIsInstance(ids, list)
|
||
|
|
ids = o.execute(db_name, self.admin_uid, 'admin', 'ir.model', 'search', [], {})
|
||
|
|
self.assertIsInstance(ids, list)
|
||
|
|
|
||
|
|
def test_xmlrpc_read_group(self):
|
||
|
|
groups = self.xmlrpc_object.execute(
|
||
|
|
common.get_db_name(), self.admin_uid, 'admin',
|
||
|
|
'res.partner', 'read_group', [], ['is_company', 'color'], ['parent_id']
|
||
|
|
)
|
||
|
|
|
||
|
|
def test_xmlrpc_name_search(self):
|
||
|
|
self.xmlrpc_object.execute(
|
||
|
|
common.get_db_name(), self.admin_uid, 'admin',
|
||
|
|
'res.partner', 'name_search', "admin"
|
||
|
|
)
|
||
|
|
|
||
|
|
def test_xmlrpc_html_field(self):
|
||
|
|
sig = '<p>bork bork bork <span style="font-weight: bork">bork</span><br></p>'
|
||
|
|
r = self.env['res.users'].create({
|
||
|
|
'name': 'bob',
|
||
|
|
'login': 'bob',
|
||
|
|
'signature': sig
|
||
|
|
})
|
||
|
|
self.assertEqual(str(r.signature), sig)
|
||
|
|
[x] = self.xmlrpc('res.users', 'read', r.id, ['signature'])
|
||
|
|
self.assertEqual(x['signature'], sig)
|
||
|
|
|
||
|
|
def test_xmlrpc_frozendict_marshalling(self):
|
||
|
|
""" Test that the marshalling of a frozendict object works properly over XMLRPC """
|
||
|
|
self.env.ref('base.user_admin').tz = "Europe/Brussels"
|
||
|
|
ctx = self.xmlrpc_object.execute(
|
||
|
|
common.get_db_name(), self.admin_uid, 'admin',
|
||
|
|
'res.users', 'context_get',
|
||
|
|
)
|
||
|
|
self.assertEqual(ctx['lang'], 'en_US')
|
||
|
|
self.assertEqual(ctx['tz'], 'Europe/Brussels')
|
||
|
|
|
||
|
|
def test_xmlrpc_defaultdict_marshalling(self):
|
||
|
|
"""
|
||
|
|
Test that the marshalling of a collections.defaultdict object
|
||
|
|
works properly over XMLRPC
|
||
|
|
"""
|
||
|
|
self.patch(self.registry['res.users'], 'context_get',
|
||
|
|
odoo.api.model(lambda *_: collections.defaultdict(int)))
|
||
|
|
self.assertEqual(self.xmlrpc('res.users', 'context_get'), {})
|
||
|
|
|
||
|
|
def test_xmlrpc_remove_control_characters(self):
|
||
|
|
record = self.env['res.users'].create({
|
||
|
|
'name': 'bob with a control character: \x03',
|
||
|
|
'login': 'bob',
|
||
|
|
})
|
||
|
|
self.assertEqual(record.name, 'bob with a control character: \x03')
|
||
|
|
[record_data] = self.xmlrpc('res.users', 'read', record.id, ['name'])
|
||
|
|
self.assertEqual(record_data['name'], 'bob with a control character: ')
|
||
|
|
|
||
|
|
def test_jsonrpc_read_group(self):
|
||
|
|
self._json_call(
|
||
|
|
common.get_db_name(), self.admin_uid, 'admin',
|
||
|
|
'res.partner', 'read_group', [], ['is_company', 'color'], ['parent_id']
|
||
|
|
)
|
||
|
|
|
||
|
|
def test_jsonrpc_name_search(self):
|
||
|
|
# well that's some sexy sexy call right there
|
||
|
|
self._json_call(
|
||
|
|
common.get_db_name(),
|
||
|
|
self.admin_uid, 'admin',
|
||
|
|
'res.partner', 'name_search', 'admin'
|
||
|
|
)
|
||
|
|
|
||
|
|
def _json_call(self, *args):
|
||
|
|
self.opener.post("http://%s:%s/jsonrpc" % (common.HOST, odoo.tools.config['http_port']), json={
|
||
|
|
'jsonrpc': '2.0',
|
||
|
|
'id': None,
|
||
|
|
'method': 'call',
|
||
|
|
'params': {
|
||
|
|
'service': 'object',
|
||
|
|
'method': 'execute',
|
||
|
|
'args': args
|
||
|
|
}
|
||
|
|
})
|
||
|
|
|
||
|
|
def test_xmlrpc_attachment_raw(self):
|
||
|
|
ids = self.env['ir.attachment'].create({'name': 'n', 'raw': b'\x01\x09'}).ids
|
||
|
|
[att] = self.xmlrpc_object.execute(
|
||
|
|
common.get_db_name(), self.admin_uid, 'admin',
|
||
|
|
'ir.attachment', 'read', ids, ['raw'])
|
||
|
|
self.assertEqual(att['raw'], '\t',
|
||
|
|
"on read, binary data should be decoded as a string and stripped from control character")
|
||
|
|
|
||
|
|
# really just for the test cursor
|
||
|
|
@common.tagged('post_install', '-at_install')
|
||
|
|
class TestAPIKeys(common.HttpCase):
|
||
|
|
@classmethod
|
||
|
|
def setUpClass(cls):
|
||
|
|
super().setUpClass()
|
||
|
|
cls._user = cls.env['res.users'].create({
|
||
|
|
'name': "Bylan",
|
||
|
|
'login': 'byl',
|
||
|
|
'password': 'ananananan',
|
||
|
|
'tz': 'Australia/Eucla',
|
||
|
|
})
|
||
|
|
|
||
|
|
def setUp(self):
|
||
|
|
super().setUp()
|
||
|
|
# needs a fake request in order to call methods protected with check_identity
|
||
|
|
fake_req = DotDict({
|
||
|
|
# various things go and access request items
|
||
|
|
'httprequest': DotDict({
|
||
|
|
'environ': {'REMOTE_ADDR': 'localhost'},
|
||
|
|
'cookies': {},
|
||
|
|
}),
|
||
|
|
# bypass check_identity flow
|
||
|
|
'session': {'identity-check-last': time.time()},
|
||
|
|
'geoip': {},
|
||
|
|
})
|
||
|
|
_request_stack.push(fake_req)
|
||
|
|
self.addCleanup(_request_stack.pop)
|
||
|
|
|
||
|
|
def test_trivial(self):
|
||
|
|
uid = auth.dispatch('authenticate', [self.env.cr.dbname, 'byl', 'ananananan', {}])
|
||
|
|
self.assertEqual(uid, self._user.id)
|
||
|
|
|
||
|
|
ctx = model.dispatch('execute_kw', [
|
||
|
|
self.env.cr.dbname, uid, 'ananananan',
|
||
|
|
'res.users', 'context_get', []
|
||
|
|
])
|
||
|
|
self.assertEqual(ctx['tz'], 'Australia/Eucla')
|
||
|
|
|
||
|
|
def test_wrongpw(self):
|
||
|
|
# User.authenticate raises but RPC.authenticate returns False
|
||
|
|
uid = auth.dispatch('authenticate', [self.env.cr.dbname, 'byl', 'aws', {}])
|
||
|
|
self.assertFalse(uid)
|
||
|
|
with self.assertRaises(AccessDenied):
|
||
|
|
model.dispatch('execute_kw', [
|
||
|
|
self.env.cr.dbname, self._user.id, 'aws',
|
||
|
|
'res.users', 'context_get', []
|
||
|
|
])
|
||
|
|
|
||
|
|
def test_key(self):
|
||
|
|
env = self.env(user=self._user)
|
||
|
|
r = env['res.users.apikeys.description'].create({
|
||
|
|
'name': 'a',
|
||
|
|
}).make_key()
|
||
|
|
k = r['context']['default_key']
|
||
|
|
|
||
|
|
uid = auth.dispatch('authenticate', [self.env.cr.dbname, 'byl', 'ananananan', {}])
|
||
|
|
self.assertEqual(uid, self._user.id)
|
||
|
|
|
||
|
|
uid = auth.dispatch('authenticate', [self.env.cr.dbname, 'byl', k, {}])
|
||
|
|
self.assertEqual(uid, self._user.id)
|
||
|
|
|
||
|
|
ctx = model.dispatch('execute_kw', [
|
||
|
|
self.env.cr.dbname, uid, k,
|
||
|
|
'res.users', 'context_get', []
|
||
|
|
])
|
||
|
|
self.assertEqual(ctx['tz'], 'Australia/Eucla')
|
||
|
|
|
||
|
|
def test_delete(self):
|
||
|
|
env = self.env(user=self._user)
|
||
|
|
env['res.users.apikeys.description'].create({'name': 'b',}).make_key()
|
||
|
|
env['res.users.apikeys.description'].create({'name': 'b',}).make_key()
|
||
|
|
env['res.users.apikeys.description'].create({'name': 'b',}).make_key()
|
||
|
|
k0, k1, k2 = env['res.users.apikeys'].search([])
|
||
|
|
|
||
|
|
# user can remove their own keys
|
||
|
|
k0.remove()
|
||
|
|
self.assertFalse(k0.exists())
|
||
|
|
|
||
|
|
# admin can remove user keys
|
||
|
|
k1.with_user(self.env.ref('base.user_admin')).remove ()
|
||
|
|
self.assertFalse(k1.exists())
|
||
|
|
|
||
|
|
# other user can't remove user keys
|
||
|
|
u = self.env['res.users'].create({
|
||
|
|
'name': 'a',
|
||
|
|
'login': 'a',
|
||
|
|
'groups_id': self.env.ref('base.group_user').ids,
|
||
|
|
})
|
||
|
|
with self.assertRaises(AccessError):
|
||
|
|
k2.with_user(u).remove()
|
||
|
|
|
||
|
|
def test_disabled(self):
|
||
|
|
env = self.env(user=self._user)
|
||
|
|
k = env['res.users.apikeys.description'].create({'name': 'b',}).make_key()['context']['default_key']
|
||
|
|
|
||
|
|
self._user.active = False
|
||
|
|
|
||
|
|
with self.assertRaises(AccessDenied):
|
||
|
|
model.dispatch('execute_kw', [
|
||
|
|
self.env.cr.dbname, self._user.id, 'ananananan',
|
||
|
|
'res.users', 'context_get', []
|
||
|
|
])
|
||
|
|
|
||
|
|
with self.assertRaises(AccessDenied):
|
||
|
|
model.dispatch('execute_kw', [
|
||
|
|
self.env.cr.dbname, self._user.id, k,
|
||
|
|
'res.users', 'context_get', []
|
||
|
|
])
|