Odoo18-Base/odoo/addons/test_http/tests/test_misc.py
2025-03-10 11:12:23 +07:00

261 lines
12 KiB
Python

# Part of Odoo. See LICENSE file for full copyright and licensing details.
import json
from io import StringIO
from socket import gethostbyname
from unittest.mock import patch
import odoo
from odoo.http import root, content_disposition
from odoo.tests import tagged
from odoo.tests.common import HOST, new_test_user, get_db_name, BaseCase
from odoo.tools import config, file_path, parse_version
from odoo.addons.test_http.controllers import CT_JSON
from .test_common import TestHttpBase
try:
from importlib import metadata
werkzeug_version = metadata.version('werkzeug')
except ImportError:
import werkzeug
werkzeug_version = werkzeug.__version__
@tagged('post_install', '-at_install')
class TestHttpMisc(TestHttpBase):
def test_misc0_redirect(self):
res = self.nodb_url_open('/test_http//greeting')
awaited_codes = [404]
if parse_version('2.2.0') <= parse_version(werkzeug_version) <= parse_version('3.0.1'):
# Bug in werkzeug from 2.2.0 up to 3.0.1 (shipped in Ubuntu Noble 24.04)
# not a big deal but should be removed once fixed upstream.
awaited_codes.append(308)
self.assertIn(res.status_code, awaited_codes)
def test_misc1_reverse_proxy(self):
# client <-> reverse-proxy <-> odoo
client_ip = '127.0.0.16'
reverseproxy_ip = gethostbyname(HOST)
host = 'mycompany.odoo.com'
headers = {
'Host': '',
'X-Forwarded-For': client_ip,
'X-Forwarded-Host': host,
'X-Forwarded-Proto': 'https'
}
# Don't trust client-sent forwarded headers
with patch.object(config, 'options', {**config.options, 'proxy_mode': False}):
res = self.nodb_url_open('/test_http/wsgi_environ', headers=headers)
self.assertEqual(res.status_code, 200)
self.assertEqual(res.json()['REMOTE_ADDR'], reverseproxy_ip)
self.assertEqual(res.json()['HTTP_HOST'], '')
# Trust proxy-sent forwarded headers
with patch.object(config, 'options', {**config.options, 'proxy_mode': True}):
res = self.nodb_url_open('/test_http/wsgi_environ', headers=headers)
self.assertEqual(res.status_code, 200)
self.assertEqual(res.json()['REMOTE_ADDR'], client_ip)
self.assertEqual(res.json()['HTTP_HOST'], host)
def test_misc2_local_redirect(self):
def local_redirect(path):
fake_req = odoo.tools.misc.DotDict(db=False)
return odoo.http.Request.redirect(fake_req, path, local=True).headers['Location']
self.assertEqual(local_redirect('https://www.example.com/hello?a=b'), '/hello?a=b')
self.assertEqual(local_redirect('/hello?a=b'), '/hello?a=b')
self.assertEqual(local_redirect('hello?a=b'), '/hello?a=b')
self.assertEqual(local_redirect('www.example.com/hello?a=b'), '/www.example.com/hello?a=b')
self.assertEqual(local_redirect('https://www.example.comhttps://www.example2.com/hello?a=b'), '/www.example2.com/hello?a=b')
self.assertEqual(local_redirect('https://https://www.example.com/hello?a=b'), '/www.example.com/hello?a=b')
def test_misc3_is_static_file(self):
uri = 'test_http/static/src/img/gizeh.png'
path = file_path(uri)
# Valid URLs
self.assertEqual(root.get_static_file(f'/{uri}'), path, "Valid file")
self.assertEqual(root.get_static_file(f'odoo.com/{uri}', host='odoo.com'), path, "Valid file with valid host")
self.assertEqual(root.get_static_file(f'http://odoo.com/{uri}', host='odoo.com'), path, "Valid file with valid host")
# Invalid URLs
self.assertIsNone(root.get_static_file('/test_http/i-dont-exist'), "File doesn't exist")
self.assertIsNone(root.get_static_file('/test_http/__manifest__.py'), "File is not static")
self.assertIsNone(root.get_static_file(f'odoo.com/{uri}'), "No host allowed")
self.assertIsNone(root.get_static_file(f'http://odoo.com/{uri}'), "No host allowed")
def test_misc4_rpc_qweb(self):
jack = new_test_user(self.env, 'jackoneill', context={'lang': 'en_US'})
milky_way = self.env.ref('test_http.milky_way')
payload = json.dumps({'jsonrpc': '2.0', 'method': 'call', 'id': None, 'params': {
'service': 'object', 'method': 'execute', 'args': [
get_db_name(), jack.id, 'jackoneill', 'test_http.galaxy', 'render', milky_way.id
]
}})
for method in (self.db_url_open, self.nodb_url_open):
with self.subTest(method=method.__name__):
res = method('/jsonrpc', data=payload, headers=CT_JSON)
res.raise_for_status()
res_rpc = res.json()
self.assertNotIn('error', res_rpc.keys(), res_rpc.get('error', {}).get('data', {}).get('message'))
self.assertIn(milky_way.name, res_rpc['result'], "QWeb template was correctly rendered")
def test_misc5_upload_file_retry(self):
from odoo.addons.test_http import controllers # pylint: disable=C0415
with patch.object(controllers, "should_fail", True), StringIO("Hello world!") as file:
res = self.url_open("/test_http/upload_file", files={"ufile": file}, timeout=None)
self.assertEqual(res.status_code, 200)
self.assertEqual(res.text, file.getvalue())
@tagged('post_install', '-at_install')
class TestHttpCors(TestHttpBase):
def test_cors0_http_default(self):
res_opt = self.opener.options(f'{self.base_url()}/test_http/cors_http_default', timeout=10, allow_redirects=False)
self.assertIn(res_opt.status_code, (200, 204))
self.assertEqual(res_opt.headers.get('Access-Control-Allow-Origin'), '*')
self.assertEqual(res_opt.headers.get('Access-Control-Allow-Methods'), 'GET, POST')
self.assertEqual(res_opt.headers.get('Access-Control-Max-Age'), '86400') # one day
self.assertEqual(res_opt.headers.get('Access-Control-Allow-Headers'), 'Origin, X-Requested-With, Content-Type, Accept, Authorization')
res_get = self.url_open('/test_http/cors_http_default')
self.assertEqual(res_get.status_code, 200)
self.assertEqual(res_get.headers.get('Access-Control-Allow-Origin'), '*')
self.assertEqual(res_get.headers.get('Access-Control-Allow-Methods'), 'GET, POST')
def test_cors1_http_methods(self):
res_opt = self.opener.options(f'{self.base_url()}/test_http/cors_http_methods', timeout=10, allow_redirects=False)
self.assertIn(res_opt.status_code, (200, 204))
self.assertEqual(res_opt.headers.get('Access-Control-Allow-Origin'), '*')
self.assertEqual(res_opt.headers.get('Access-Control-Allow-Methods'), 'GET, PUT')
self.assertEqual(res_opt.headers.get('Access-Control-Max-Age'), '86400') # one day
self.assertEqual(res_opt.headers.get('Access-Control-Allow-Headers'), 'Origin, X-Requested-With, Content-Type, Accept, Authorization')
res_post = self.url_open('/test_http/cors_http_methods')
self.assertEqual(res_post.status_code, 200)
self.assertEqual(res_post.headers.get('Access-Control-Allow-Origin'), '*')
self.assertEqual(res_post.headers.get('Access-Control-Allow-Methods'), 'GET, PUT')
def test_cors2_json(self):
res_opt = self.opener.options(f'{self.base_url()}/test_http/cors_json', timeout=10, allow_redirects=False)
self.assertIn(res_opt.status_code, (200, 204), res_opt.text)
self.assertEqual(res_opt.headers.get('Access-Control-Allow-Origin'), '*')
self.assertEqual(res_opt.headers.get('Access-Control-Allow-Methods'), 'POST')
self.assertEqual(res_opt.headers.get('Access-Control-Max-Age'), '86400') # one day
self.assertEqual(res_opt.headers.get('Access-Control-Allow-Headers'), 'Origin, X-Requested-With, Content-Type, Accept, Authorization')
res_post = self.url_open('/test_http/cors_json', data=json.dumps({'params': {}}), headers=CT_JSON)
self.assertEqual(res_post.status_code, 200)
self.assertEqual(res_post.headers.get('Access-Control-Allow-Origin'), '*')
self.assertEqual(res_post.headers.get('Access-Control-Allow-Methods'), 'POST')
@tagged('post_install', '-at_install')
class TestHttpEnsureDb(TestHttpBase):
def setUp(self):
super().setUp()
self.db_list = ['db0', 'db1']
def test_ensure_db0_db_selector(self):
res = self.multidb_url_open('/test_http/ensure_db')
res.raise_for_status()
self.assertEqual(res.status_code, 303)
self.assertURLEqual(res.headers.get('Location'), '/web/database/selector')
def test_ensure_db1_grant_db(self):
res = self.multidb_url_open('/test_http/ensure_db?db=db0')
res.raise_for_status()
self.assertEqual(res.status_code, 302)
self.assertURLEqual(res.headers.get('Location'), '/test_http/ensure_db?db=db0')
self.assertEqual(odoo.http.root.session_store.get(res.cookies['session_id']).db, 'db0')
# follow the redirection
res = self.multidb_url_open('/test_http/ensure_db')
res.raise_for_status()
self.assertEqual(res.status_code, 200)
self.assertEqual(res.text, 'db0')
def test_ensure_db2_use_session_db(self):
session = self.authenticate(None, None)
session.db = 'db0'
odoo.http.root.session_store.save(session)
res = self.multidb_url_open('/test_http/ensure_db')
res.raise_for_status()
self.assertEqual(res.status_code, 200)
self.assertEqual(res.text, 'db0')
def test_ensure_db3_change_db(self):
session = self.authenticate(None, None)
session.db = 'db0'
odoo.http.root.session_store.save(session)
res = self.multidb_url_open('/test_http/ensure_db?db=db1')
res.raise_for_status()
self.assertEqual(res.status_code, 302)
self.assertURLEqual(res.headers.get('Location'), '/test_http/ensure_db?db=db1')
new_session = odoo.http.root.session_store.get(res.cookies['session_id'])
self.assertNotEqual(session.sid, new_session.sid)
self.assertEqual(new_session.db, 'db1')
self.assertEqual(new_session.uid, None)
# follow redirection
self.opener.cookies['session_id'] = new_session.sid
res = self.multidb_url_open('/test_http/ensure_db')
res.raise_for_status()
self.assertEqual(res.status_code, 200)
self.assertEqual(res.text, 'db1')
def test_ensure_db4_unicode(self):
self.db_list = ["basededonnée1", "basededonnée2"] # é matters
res = self.multidb_url_open('/test_http/ensure_db?db=basededonnée1')
res.raise_for_status()
self.assertEqual(res.status_code, 302)
self.assertURLEqual(
res.headers.get('Location'),
'/test_http/ensure_db?db=basededonnée1')
self.assertEqual(
odoo.http.root.session_store.get(res.cookies['session_id']).db,
'basededonnée1')
# follow the redirection
res = self.multidb_url_open('/test_http/ensure_db')
res.raise_for_status()
self.assertEqual(res.status_code, 200)
self.assertEqual(res.text, 'basededonnée1')
class TestContentDisposition(BaseCase):
def test_content_disposition(self):
""" Test that content_disposition filename conforms to RFC 6266, RFC 5987 """
assertions = [
('foo bar.xls', 'foo%20bar.xls', 'Space character'),
('foo(bar).xls', 'foo%28bar%29.xls', 'Parenthesis'),
('foo<bar>.xls', 'foo%3Cbar%3E.xls', 'Angle brackets'),
('foo[bar].xls', 'foo%5Bbar%5D.xls', 'Brackets'),
('foo{bar}.xls', 'foo%7Bbar%7D.xls', 'Curly brackets'),
('foo@bar.xls', 'foo%40bar.xls', 'At sign'),
('foo,bar.xls', 'foo%2Cbar.xls', 'Comma sign'),
('foo;bar.xls', 'foo%3Bbar.xls', 'Semicolon sign'),
('foo:bar.xls', 'foo%3Abar.xls', 'Colon sign'),
('foo\\bar.xls', 'foo%5Cbar.xls', 'Backslash sign'),
('foo"bar.xls', 'foo%22bar.xls', 'Double quote sign'),
('foo/bar.xls', 'foo%2Fbar.xls', 'Slash sign'),
('foo?bar.xls', 'foo%3Fbar.xls', 'Question mark'),
('foo=bar.xls', 'foo%3Dbar.xls', 'Equal sign'),
('foo*bar.xls', 'foo%2Abar.xls', 'Star sign'),
("foo'bar.xls", 'foo%27bar.xls', 'Single-quote sign'),
('foo%bar.xls', 'foo%25bar.xls', 'Percent sign'),
]
for filename, pct_encoded, hint in assertions:
self.assertEqual(content_disposition(filename), f"attachment; filename*=UTF-8''{pct_encoded}", f'{hint} should be percent encoded')