Forward-port of 8f52bdf9f4
closes odoo/odoo#93335
X-original-commit: e766798cb11e4637d9a303c525c78cf8e60f20dd
Signed-off-by: Christophe Monniez (moc) <moc@odoo.com>
Signed-off-by: Julien Castiaux <juc@odoo.com>
166 lines
7.8 KiB
Python
166 lines
7.8 KiB
Python
# Part of Odoo. See LICENSE file for full copyright and licensing details.
|
|
|
|
import json
|
|
from socket import gethostbyname
|
|
from unittest.mock import patch
|
|
from urllib.parse import urlparse
|
|
|
|
import odoo
|
|
from odoo.http import root
|
|
from odoo.tests import tagged
|
|
from odoo.tests.common import HOST
|
|
from odoo.tools import config, file_path
|
|
from odoo.addons.test_http.controllers import CT_JSON
|
|
|
|
from .test_common import TestHttpBase
|
|
|
|
|
|
@tagged('post_install', '-at_install')
|
|
class TestHttpMisc(TestHttpBase):
|
|
def test_misc0_redirect(self):
|
|
res = self.nodb_url_open('/test_http//greeting')
|
|
self.assertEqual(res.status_code, 301)
|
|
self.assertEqual(urlparse(res.headers.get('Location', '')).path, '/test_http/greeting')
|
|
|
|
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_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")
|
|
|
|
|
|
@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.assertEqual(urlparse(res.headers.get('Location', '')).path, '/web/database/selector')
|
|
|
|
def test_ensure_db1_grant_db(self):
|
|
res = self.multidb_url_open('/test_http/ensure_db?db=db0', timeout=10000)
|
|
res.raise_for_status()
|
|
self.assertEqual(res.status_code, 302)
|
|
self.assertEqual(urlparse(res.headers.get('Location', '')).path, '/test_http/ensure_db')
|
|
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.assertEqual(urlparse(res.headers.get('Location', '')).path, '/test_http/ensure_db')
|
|
|
|
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')
|