From ce12c52a5d4fbe7a4f3d6f002d36e285de5c255e Mon Sep 17 00:00:00 2001 From: Julien Castiaux Date: Tue, 8 Feb 2022 07:33:24 +0000 Subject: [PATCH] [ADD] test_http: Core HTTP test suite Backport from the master-httpocalypse-juc test suite in order to observe the perf of the master branch and to compare them with the httpocalypse. closes odoo/odoo#84156 Task: 2571224 Signed-off-by: Raphael Collet --- odoo/addons/test_http/__init__.py | 2 + odoo/addons/test_http/__manifest__.py | 16 + odoo/addons/test_http/controllers.py | 111 +++++ odoo/addons/test_http/data.xml | 43 ++ odoo/addons/test_http/ir.model.access.csv | 6 + odoo/addons/test_http/models.py | 57 +++ .../addons/test_http/static/src/img/gizeh.png | Bin 0 -> 814 bytes .../addons/test_http/static/src/img/gizeh.svg | 22 + odoo/addons/test_http/tests/__init__.py | 1 + odoo/addons/test_http/tests/test_http.py | 466 ++++++++++++++++++ odoo/addons/test_http/views.xml | 25 + odoo/tests/common.py | 9 +- 12 files changed, 757 insertions(+), 1 deletion(-) create mode 100644 odoo/addons/test_http/__init__.py create mode 100644 odoo/addons/test_http/__manifest__.py create mode 100644 odoo/addons/test_http/controllers.py create mode 100644 odoo/addons/test_http/data.xml create mode 100644 odoo/addons/test_http/ir.model.access.csv create mode 100644 odoo/addons/test_http/models.py create mode 100644 odoo/addons/test_http/static/src/img/gizeh.png create mode 100644 odoo/addons/test_http/static/src/img/gizeh.svg create mode 100644 odoo/addons/test_http/tests/__init__.py create mode 100644 odoo/addons/test_http/tests/test_http.py create mode 100644 odoo/addons/test_http/views.xml diff --git a/odoo/addons/test_http/__init__.py b/odoo/addons/test_http/__init__.py new file mode 100644 index 00000000000..91c5580fed3 --- /dev/null +++ b/odoo/addons/test_http/__init__.py @@ -0,0 +1,2 @@ +from . import controllers +from . import models diff --git a/odoo/addons/test_http/__manifest__.py b/odoo/addons/test_http/__manifest__.py new file mode 100644 index 00000000000..ae26765291f --- /dev/null +++ b/odoo/addons/test_http/__manifest__.py @@ -0,0 +1,16 @@ +# Part of Odoo. See LICENSE file for full copyright and licensing details. +{ + 'name': 'Test HTTP', + 'version': '1.0', + 'category': 'Hidden/Tests', + 'description': """A module to test HTTP""", + 'depends': ['base', 'web', 'web_tour'], + 'installable': True, + 'auto_install': False, + 'data': [ + 'data.xml', + 'ir.model.access.csv', + 'views.xml' + ], + 'license': 'LGPL-3', +} diff --git a/odoo/addons/test_http/controllers.py b/odoo/addons/test_http/controllers.py new file mode 100644 index 00000000000..9c64517f92d --- /dev/null +++ b/odoo/addons/test_http/controllers.py @@ -0,0 +1,111 @@ +# Part of Odoo. See LICENSE file for full copyright and licensing details. +import json +import logging +from odoo import http +from odoo.exceptions import UserError +from odoo.http import request + +_logger = logging.getLogger(__name__) + + +CT_JSON = {'Content-Type': 'application/json'} +WSGI_SAFE_KEYS = {'PATH_INFO', 'QUERY_STRING', 'RAW_URI', 'SCRIPT_NAME', 'wsgi.url_scheme'} + + +class TestHttp(http.Controller): + + # ===================================================== + # Greeting + # ===================================================== + @http.route(['/test_http/greeting', '/test_http/greeting-none'], type='http', auth='none') + def greeting_none(self): + return "Tek'ma'te" + + @http.route('/test_http/greeting-public', type='http', auth='public') + def greeting_public(self): + assert request.env.user, "ORM should be initialized" + return "Tek'ma'te" + + @http.route('/test_http/greeting-user', type='http', auth='user') + def greeting_user(self): + assert request.env.user, "ORM should be initialized" + return "Tek'ma'te" + + @http.route('/test_http/wsgi_environ', type='http', auth='none') + def wsgi_environ(self): + environ = { + key: val for key, val in request.httprequest.environ.items() + if (key.startswith('HTTP_') # headers + or key.startswith('REMOTE_') + or key.startswith('REQUEST_') + or key.startswith('SERVER_') + or key.startswith('werkzeug.proxy_fix.') + or key in WSGI_SAFE_KEYS) + } + + return request.make_response( + json.dumps(environ, indent=4), + headers=list(CT_JSON.items()) + ) + + # ===================================================== + # Echo-Reply + # ===================================================== + @http.route('/test_http/echo-http-get', type='http', auth='none', methods=['GET']) + def echo_http_get(self, **kwargs): + return str(kwargs) + + @http.route('/test_http/echo-http-post', type='http', auth='none', methods=['POST'], csrf=False) + def echo_http_post(self, **kwargs): + return str(kwargs) + + @http.route('/test_http/echo-http-csrf', type='http', auth='none', methods=['POST'], csrf=True) + def echo_http_csrf(self, **kwargs): + return str(kwargs) + + @http.route('/test_http/echo-json', type='json', auth='none', methods=['POST'], csrf=False) + def echo_json(self, **kwargs): + return kwargs + + @http.route('/test_http/echo-json-context', type='json', auth='user', methods=['POST'], csrf=False) + def echo_json_context(self, **kwargs): + return request.env.context + + # ===================================================== + # Models + # ===================================================== + @http.route('/test_http/', auth='public') + def galaxy(self, galaxy): + if not galaxy.exists(): + raise UserError('The Ancients did not settle there.') + + return http.request.render('test_http.tmpl_galaxy', { + 'galaxy': galaxy, + 'stargates': http.request.env['test_http.stargate'].search([ + ('galaxy_id', '=', galaxy.id) + ]), + }) + + @http.route('/test_http//', auth='user') + def stargate(self, galaxy, gate): + if not gate.exists(): + raise UserError("The goa'uld destroyed the gate") + + return http.request.render('test_http.tmpl_stargate', { + 'gate': gate + }) + + # ===================================================== + # Cors + # ===================================================== + @http.route('/test_http/cors_http_default', type='http', auth='none', cors='*') + def cors_http(self): + return "Hello" + + @http.route('/test_http/cors_http_methods', type='http', auth='none', methods=['GET', 'PUT'], cors='*') + def cors_http_verbs(self, **kwargs): + return "Hello" + + @http.route('/test_http/cors_json', type='json', auth='none', cors='*') + def cors_json(self, **kwargs): + return {} diff --git a/odoo/addons/test_http/data.xml b/odoo/addons/test_http/data.xml new file mode 100644 index 00000000000..0fe47642439 --- /dev/null +++ b/odoo/addons/test_http/data.xml @@ -0,0 +1,43 @@ + + + + Milky Way + + + + Pegasus + + + + + Earth + sq5Abt + + + + + Abydos + r7fwcu + + + + + Dakara + gs38x4 + + + + + Lantea + u6iz7d + + + + + + Athos + mfp2rt + + + + diff --git a/odoo/addons/test_http/ir.model.access.csv b/odoo/addons/test_http/ir.model.access.csv new file mode 100644 index 00000000000..3a8de2beb06 --- /dev/null +++ b/odoo/addons/test_http/ir.model.access.csv @@ -0,0 +1,6 @@ +id,name,model_id:id,group_id:id,perm_read,perm_write,perm_create,perm_unlink +access_test_http_stargate_portal,test_http.stargate,model_test_http_stargate,base.group_public,1,0,0,0 +access_test_http_stargate_user,test_http.stargate,model_test_http_stargate,base.group_user,1,1,1,1 +access_test_http_galaxy_portal,test_http.galaxy,model_test_http_galaxy,base.group_public,1,0,0,0 +access_test_http_galaxy_user,test_http.galaxy,model_test_http_galaxy,base.group_user,1,1,0,0 +access_test_http_galaxy_system,test_http.galaxy,model_test_http_galaxy,base.group_system,1,1,1,1 diff --git a/odoo/addons/test_http/models.py b/odoo/addons/test_http/models.py new file mode 100644 index 00000000000..4419b25924e --- /dev/null +++ b/odoo/addons/test_http/models.py @@ -0,0 +1,57 @@ +# Part of Odoo. See LICENSE file for full copyright and licensing details. + +from odoo import models, fields, api + + +MILKY_WAY_REGIONS = ['P3X', 'P4X', 'P2X', 'P5C'] +PEGASUS_REGIONS = ['M4R', 'P3Y', 'M6R'] + + +class Stargate(models.Model): + _name = 'test_http.stargate' + _description = 'Stargate' + + name = fields.Char(required=True, store=True, compute='_compute_name', readonly=False, help="The stargate/planet common name.") + address = fields.Char(required=True, help="The stargate's 6-glyphs address used by other stargates to dial this one.") + sgc_designation = fields.Char(store=True, compute='_compute_sgc_designation', help="The SGC designation name of this stargate.") + galaxy_id = fields.Many2one('test_http.galaxy', required=True, help="The galaxy where this stargate is.") + has_galaxy_crystal = fields.Boolean(store=True, compute='_compute_has_galaxy_crystal', readonly=False, help="Whether this stargate can dial other galaxies.") + + _sql_constraints = [ + ('address_length', 'CHECK(LENGTH(address) = 6)', "Local addresses have 6 glyphs"), + ] + + @api.depends('galaxy_id') + def _compute_has_galaxy_crystal(self): + milky_way = self.env.ref('test_http.milky_way') + for gate in self: + gate.has_galaxy_crystal = gate.galaxy_id == milky_way + + @api.depends('sgc_designation') + def _compute_name(self): + for gate in self: + if not gate.name: + gate.name = gate.sgc_designation + + @api.depends('address') + def _compute_sgc_designation(self): + """ Forge a sgc designation that look like a real one. """ + for gate in self: + if gate.galaxy_id.name not in ('Milky Way', 'Pegasus'): + gate.sgc_designation = False + continue + + region_part = ( + PEGASUS_REGIONS[gate.id % len(PEGASUS_REGIONS)] + if gate.galaxy_id.name == 'Pegasus' + else MILKY_WAY_REGIONS[gate.id % len(MILKY_WAY_REGIONS)] + ) + local_part = str(int.from_bytes(gate.address.encode(), 'big'))[:3] + gate.sgc_designation = f'{region_part}-{local_part}' + + +class Galaxy(models.Model): + _name = 'test_http.galaxy' + _description = 'Galaxy' + + name = fields.Char(required=True, help='The galaxy common name.') diff --git a/odoo/addons/test_http/static/src/img/gizeh.png b/odoo/addons/test_http/static/src/img/gizeh.png new file mode 100644 index 0000000000000000000000000000000000000000..e990d7b0f9c13ed65959b374be73c9743ca2faaa GIT binary patch literal 814 zcmV+}1JV46P)+;3owPy15fv(x9Ysh}ors_oA62jII; zdl`7)ye^muYVra~&hyHUKB}!0(#IMHap(9xpY|QVl+%8?fxcqXj627F0EduyzXaJA zzBp|^fwh)AMzSB6rh4t4@w`7~&0S2okek)JiM>a5W259u3u?NZz;0v@7y%w4zeYwQ z*@{f51xX9=6v;?8TJ};GdSdqoSxy%WN0_?`oww?!1`0?*=e>Yg;DAM6v1kR};+nny zT-4rA0(o2JqR1h~NM-gOk}Ng>_qETTAZf?E;}ovR*T}D-Xa(MA?Sl<>7eOcR4>9)# z_iUh3aN5*G7*8weWgTu78kMXa*ZP%!_SD@)(1e_Sm8qLC#@y2SKSMS=YxHYo-cdd8 zgJ?x67HZs;)Z9hTjhpM2fsPu+b~t0yMI1KnE`m+KH_gC_Iy6pd<9`9$jJb=T8OiTe z@H^CZuE;rKmCh@|3j)T_ch(7#l?vM&4W)M@vAYO*wP3SAj{(gdM=SL+b30OPP=N*v zYG2ZH{{lsW+JLOpqgs$rV3|Sv<-ki#e~frba%`??;!{YG$}~l)OEN + + + + + + + + + + + + + + + diff --git a/odoo/addons/test_http/tests/__init__.py b/odoo/addons/test_http/tests/__init__.py new file mode 100644 index 00000000000..451f37f515b --- /dev/null +++ b/odoo/addons/test_http/tests/__init__.py @@ -0,0 +1 @@ +from . import test_http diff --git a/odoo/addons/test_http/tests/test_http.py b/odoo/addons/test_http/tests/test_http.py new file mode 100644 index 00000000000..8968f61457b --- /dev/null +++ b/odoo/addons/test_http/tests/test_http.py @@ -0,0 +1,466 @@ +# -*- coding: utf-8 -*- +# Part of Odoo. See LICENSE file for full copyright and licensing details. + +import json +import unittest +from html.parser import HTMLParser +from unittest.mock import patch +from urllib.parse import urlparse +from socket import gethostbyname + +import odoo +from odoo.http import WebRequest +from odoo.tests import tagged +from odoo.tests.common import HOST, HttpCase, new_test_user +from odoo.tools import config, file_open, mute_logger +from odoo.tools.func import lazy_property +from odoo.addons.test_http.controllers import CT_JSON + + +# pylint: disable=W0223(abstract-method) +class HtmlTokenizer(HTMLParser): + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self.tokens = [] + + @classmethod + def _attrs_to_str(cls, attrs): + out = [] + for key, value in attrs: + out.append(f"{key}={value!r}" if value else key) + return " ".join(out) + + def handle_starttag(self, tag, attrs): + self.tokens.append(f"<{tag} {self._attrs_to_str(attrs)}>") + + def handle_endtag(self, tag): + self.tokens.append(f"") + + def handle_startendtag(self, tag, attrs): + # HTML5 instead of XHTML + self.handle_starttag(tag, attrs) + + def handle_data(self, data): + data = data.strip() + if data: + self.tokens.append(data) + + @classmethod + def tokenize(cls, source_str): + """ + Parse the source html into a list of tokens. Only tags and + tags data are conserved, other elements such as comments are + discarded. + """ + tokenizer = cls() + tokenizer.feed(source_str) + return tokenizer.tokens + + +class TestHttpBase(HttpCase): + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.addClassCleanup(lazy_property.reset_all, odoo.http.root) + cls.classPatch(odoo.conf, 'server_wide_modules', ['base', 'web', 'test_http']) + lazy_property.reset_all(odoo.http.root) + + def db_url_open(self, url, *args, allow_redirects=False, **kwargs): + return self.url_open(url, *args, allow_redirects=allow_redirects, **kwargs) + + def nodb_url_open(self, url, *args, allow_redirects=False, **kwargs): + with patch('odoo.http.db_list') as db_list, \ + patch('odoo.http.db_filter') as db_filter, \ + patch('odoo.http.db_monodb') as db_monodb: + db_list.return_value = [] + db_filter.return_value = [] + db_monodb.return_value = None + return self.url_open(url, *args, allow_redirects=allow_redirects, **kwargs) + + +@tagged('post_install', '-at_install') +class TestHttpGreeting(TestHttpBase): + def test_greeting0_matrix(self): + new_test_user(self.env, 'jackoneill', context={'lang': 'en_US'}) + test_matrix = [ + # path, database, login, expected_code, expected_re_pattern + ('/test_http/greeting', False, None, 200, r"Tek'ma'te"), + ('/test_http/greeting', True, None, 200, r"Tek'ma'te"), + ('/test_http/greeting', True, 'public', 200, r"Tek'ma'te"), + ('/test_http/greeting', True, 'jackoneill', 200, r"Tek'ma'te"), + ('/test_http/greeting-none', False, None, 200, r"Tek'ma'te"), + ('/test_http/greeting-none', True, None, 200, r"Tek'ma'te"), + ('/test_http/greeting-none', True, 'public', 200, r"Tek'ma'te"), + ('/test_http/greeting-none', True, 'jackoneill', 200, r"Tek'ma'te"), + ('/test_http/greeting-public', False, None, 404, r"Not Found"), + ('/test_http/greeting-public', True, None, 200, r"Tek'ma'te"), + ('/test_http/greeting-public', True, 'public', 200, r"Tek'ma'te"), + ('/test_http/greeting-public', True, 'jackoneill', 200, r"Tek'ma'te"), + ('/test_http/greeting-user', False, None, 404, r"Not Found"), + ('/test_http/greeting-user', True, None, 303, r".*/web/login.*"), + ('/test_http/greeting-user', True, 'public', 303, r".*/web/login.*"), + ('/test_http/greeting-user', True, 'jackoneill', 200, r"Tek'ma'te"), + ] + + for path, withdb, login, expected_code, expected_pattern in test_matrix: + with self.subTest(path=path, withdb=withdb, login=login): + if withdb: + if login == 'public': + self.authenticate(None, None) + elif login: + self.authenticate(login, login) + res = self.db_url_open(path, allow_redirects=False) + else: + res = self.nodb_url_open(path, allow_redirects=False) + + self.assertEqual(res.status_code, expected_code) + self.assertRegex(res.text, expected_pattern) + + if withdb and login: + self.logout(keep_db=False) + + def test_greeting1_headers_nodb(self): + res = self.nodb_url_open('/test_http/greeting') + self.assertEqual(res.status_code, 200) + self.assertEqual(res.headers.get('Content-Type'), 'text/html; charset=utf-8') + self.assertEqual(res.text, "Tek'ma'te") + + def test_greeting2_headers_db(self): + new_test_user(self.env, 'jackoneill', context={'lang': 'en_US'}) + self.authenticate('jackoneill', 'jackoneill') + res = self.db_url_open('/test_http/greeting') + self.assertEqual(res.status_code, 200) + self.assertEqual(res.headers.get('Content-Type'), 'text/html; charset=utf-8') + self.assertEqual(res.text, "Tek'ma'te") + + +@tagged('post_install', '-at_install') +class TestHttpStatic(TestHttpBase): + def test_static0_png_image(self): + res = self.nodb_url_open("/test_http/static/src/img/gizeh.png") + self.assertEqual(res.status_code, 200) + self.assertEqual(res.headers.get('Content-Length'), '814') + self.assertEqual(res.headers.get('Content-Type'), 'image/png') + cache_control = set(res.headers.get('Cache-Control', '').split(', ')) + self.assertEqual(cache_control, {'public', 'max-age=604800'}) # one week + with file_open('test_http/static/src/img/gizeh.png', 'rb') as file: + self.assertEqual(res.content, file.read()) + + def test_static1_svg_image(self): + res = self.nodb_url_open("/test_http/static/src/img/gizeh.svg") + self.assertEqual(res.status_code, 200) + self.assertEqual(res.headers.get('Content-Length'), '1529') + self.assertEqual(res.headers.get('Content-Type'), 'image/svg+xml') + cache_control = set(res.headers.get('Cache-Control', '').split(', ')) + self.assertEqual(cache_control, {'public', 'max-age=604800'}) # one week + with file_open('test_http/static/src/img/gizeh.svg', 'rb') as file: + self.assertEqual(res.content, file.read()) + + def test_static2_not_found(self): + res = self.nodb_url_open("/test_http/static/i-dont-exist") + self.assertEqual(res.status_code, 404) + + def test_static3_attachment(self): + with file_open('test_http/static/src/img/gizeh.svg', 'rb') as file: + content = file.read() + + attachment = self.env['ir.attachment'].create({ + 'name': 'point_of_origin.svg', + 'type': 'binary', + 'raw': content, + 'res_model': 'test_http.stargate', + 'res_id': self.ref('test_http.earth'), + }) + attachment['url'] = f'/test_http/{attachment["checksum"]}' + + res = self.db_url_open(attachment['url']) + self.assertEqual(res.headers.get('Content-Length'), '1529') + self.assertEqual(res.headers.get('Content-Type'), 'image/svg+xml; charset=utf-8') + self.assertEqual(res.headers.get('Content-Security-Policy'), "default-src 'none'") + self.assertEqual(res.content, content) + + +@tagged('post_install', '-at_install') +class TestHttpEchoReplyHttpNoDB(TestHttpBase): + def test_echohttp0_get_qs_nodb(self): + res = self.nodb_url_open('/test_http/echo-http-get?race=Asgard') + self.assertEqual(res.status_code, 200) + self.assertEqual(res.text, "{'race': 'Asgard'}") + + def test_echohttp1_get_form_nodb(self): + res = self.nodb_url_open('/test_http/echo-http-get', data={'commander': 'Thor'}) + self.assertEqual(res.status_code, 405) + + def test_echohttp2_post_qs_nodb(self): + res = self.nodb_url_open('/test_http/echo-http-post?race=Asgard') + self.assertEqual(res.status_code, 405) + + def test_echohttp3_post_qs_form_nodb(self): + res = self.nodb_url_open('/test_http/echo-http-post?race=Asgard', data={'commander': 'Thor'}) + self.assertEqual(res.status_code, 200) + self.assertEqual(res.text, "{'race': 'Asgard', 'commander': 'Thor'}") + + @mute_logger('odoo.http') + def test_echohttp4_post_json_nodb(self): + res = self.nodb_url_open('/test_http/echo-http-post', data='{}', headers=CT_JSON) + self.assertIn("Bad Request", res.text) + + @unittest.skip("Broken in master, fixed in httpocalypse.") + def test_echohttp5_post_csrf(self): + res = self.nodb_url_open('/test_http/echo-http-csrf?race=Asgard', data={'commander': 'Thor'}) + self.assertEqual(res.status_code, 303) + self.assertEqual(urlparse(res.headers.get('Location', '')).path, '/web/database/selector') + + +@tagged('post_install', '-at_install') +class TestHttpEchoReplyJsonNoDB(TestHttpBase): + def test_echojson0_qs_json_nodb(self): + payload = json.dumps({ + 'jsonrpc': '2.0', + 'id': 1234, + 'params': { + 'commander': 'Thor', + }, + }) + res = self.nodb_url_open("/test_http/echo-json?race=Asgard", data=payload, headers=CT_JSON) + self.assertEqual(res.status_code, 200) + self.assertEqual(res.text, '{"jsonrpc": "2.0", "id": 1234, "result": {"commander": "Thor"}}') + + def test_echojson1_http_get_nodb(self): + res = self.nodb_url_open('/test_http/echo-json') # GET + self.assertEqual(res.status_code, 405) + + @mute_logger('odoo.http') + def test_echojson2_http_post_nodb(self): + res = self.nodb_url_open('/test_http/echo-json', data={'race': 'Asgard'}) # POST + self.assertIn("Bad Request", res.text) + + +@tagged('post_install', '-at_install') +class TestHttpEchoReplyHttpWithDB(TestHttpBase): + def setUp(self): + super().setUp() + self.jackoneill = new_test_user(self.env, 'jackoneill', context={'lang': 'en_US'}) + self.authenticate('jackoneill', 'jackoneill') + + def test_echohttp0_get_qs_db(self): + res = self.db_url_open('/test_http/echo-http-get?race=Asgard') + self.assertEqual(res.status_code, 200) + self.assertEqual(res.text, "{'race': 'Asgard'}") + + def test_echohttp1_get_form_db(self): + res = self.db_url_open('/test_http/echo-http-get', data={'commander': 'Thor'}) + self.assertEqual(res.status_code, 405) + + def test_echohttp2_post_qs_db(self): + res = self.db_url_open('/test_http/echo-http-post?race=Asgard') + self.assertEqual(res.status_code, 405) + + def test_echohttp3_post_qs_form_db(self): + res = self.db_url_open('/test_http/echo-http-post?race=Asgard', data={'commander': 'Thor'}) + self.assertEqual(res.status_code, 200) + self.assertEqual(res.text, "{'race': 'Asgard', 'commander': 'Thor'}") + + @mute_logger('odoo.http') + def test_echohttp4_post_json_db(self): + res = self.db_url_open('/test_http/echo-http-post', data='{}', headers=CT_JSON) + self.assertIn("Bad Request", res.text) + + @mute_logger('odoo.http') + def test_echohttp5_post_no_csrf(self): + res = self.db_url_open('/test_http/echo-http-csrf?race=Asgard', data={'commander': 'Thor'}) + self.assertEqual(res.status_code, 400) + self.assertIn("Session expired (invalid CSRF token)", res.text) + + @mute_logger('odoo.http') + def test_echohttp6_post_bad_csrf(self): + res = self.db_url_open('/test_http/echo-http-csrf?race=Asgard', data={'commander': 'Thor', 'csrf_token': 'bad token'}) + self.assertEqual(res.status_code, 400) + self.assertIn("Session expired (invalid CSRF token)", res.text) + + @mute_logger('odoo.http') + def test_echohttp7_post_good_csrf(self): + res = self.db_url_open('/test_http/echo-http-csrf?race=Asgard', data={'commander': 'Thor', 'csrf_token': WebRequest.csrf_token(self)}) + self.assertEqual(res.status_code, 200) + self.assertEqual(res.text, "{'race': 'Asgard', 'commander': 'Thor'}") + + + +@tagged('post_install', '-at_install') +class TestHttpEchoReplyJsonWithDB(TestHttpBase): + def setUp(self): + super().setUp() + self.jackoneill = new_test_user(self.env, 'jackoneill', context={'lang': 'en_US'}) + self.authenticate('jackoneill', 'jackoneill') + + def test_echojson0_qs_json_db(self): + payload = json.dumps({ + 'jsonrpc': '2.0', + 'id': 1234, + 'params': { + 'commander': 'Thor', + }, + }) + res = self.db_url_open('/test_http/echo-json?race=Asgard', data=payload, headers=CT_JSON) + self.assertEqual(res.status_code, 200) + self.assertEqual(res.text, '{"jsonrpc": "2.0", "id": 1234, "result": {"commander": "Thor"}}') + + def test_echojson1_http_get_db(self): + res = self.db_url_open('/test_http/echo-json') # GET + self.assertEqual(res.status_code, 405) + + @mute_logger('odoo.http') + def test_echojson2_http_post_db(self): + res = self.db_url_open('/test_http/echo-json', data={'race': 'Asgard'}) # POST + self.assertIn("Bad Request", res.text) + + def test_echojson3_context_db(self): + payload = json.dumps({ + "jsonrpc": "2.0", + "id": 0, + "params": { + "context": { + "name": "Thor" + }, + "race": "Asgard", + }, + }) + res = self.db_url_open("/test_http/echo-json-context", data=payload, headers=CT_JSON) + self.assertEqual(res.status_code, 200) + self.assertEqual(res.text, '{"jsonrpc": "2.0", "id": 0, "result": {"name": "Thor"}}') + + +@tagged('post_install', '-at_install') +class TestHttpModels(TestHttpBase): + def setUp(self): + super().setUp() + self.jackoneill = new_test_user(self.env, 'jackoneill', context={'lang': 'en_US'}) + self.authenticate('jackoneill', 'jackoneill') + + def test_models0_galaxy_ok(self): + milky_way = self.env.ref('test_http.milky_way') + + res = self.url_open(f"/test_http/{milky_way.id}") + + self.assertEqual(res.status_code, 200) + self.assertEqual( + HtmlTokenizer.tokenize(res.text), + HtmlTokenizer.tokenize('''\ +

Milky Way

+ + ''') + ) + + @unittest.skip('500 without website, 404 with website, 400 in httpocalypse. Cmon master...') + def test_models1_galaxy_ko(self): + res = self.url_open("/test_http/404") # unknown galaxy + self.assertEqual(res.status_code, 400) + self.assertIn('The Ancients did not settle there.', res.text) + + def test_models2_stargate_ok(self): + milky_way = self.env.ref('test_http.milky_way') + earth = self.env.ref('test_http.earth') + + res = self.url_open(f'/test_http/{milky_way.id}/{earth.id}') + + self.assertEqual(res.status_code, 200) + self.assertEqual( + HtmlTokenizer.tokenize(res.text), + HtmlTokenizer.tokenize('''\ +
+
name
Earth
+
address
sq5Abt
+
sgc_designation
P4X-126
+
+ ''') + ) + + @unittest.skip('500 without website, 404 with website, 400 in httpocalypse. Cmon master...') + def test_models3_stargate_ko(self): + milky_way = self.env.ref('test_http.milky_way') + res = self.url_open(f'/test_http/{milky_way.id}/9999') # unknown gate + self.assertEqual(res.status_code, 400) + self.assertIn("The goa'uld destroyed the gate", res.text) + + +@tagged('post_install', '-at_install') +class TestHttpMisc(TestHttpBase): + @unittest.skip('In master it is in the http_routing override') + 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) + + +@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') + + @unittest.skip("Broken in master, fixed in httpocalypse.") + 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') diff --git a/odoo/addons/test_http/views.xml b/odoo/addons/test_http/views.xml new file mode 100644 index 00000000000..fdd71f0768d --- /dev/null +++ b/odoo/addons/test_http/views.xml @@ -0,0 +1,25 @@ + + + + + + + diff --git a/odoo/tests/common.py b/odoo/tests/common.py index 07f089906c5..5ae56dc0db9 100644 --- a/odoo/tests/common.py +++ b/odoo/tests/common.py @@ -417,6 +417,13 @@ class BaseCase(unittest.TestCase, metaclass=MetaCase): patcher.start() self.addCleanup(patcher.stop) + @classmethod + def classPatch(cls, obj, key, val): + """ Do the patch ``setattr(obj, key, val)``, and prepare cleanup. """ + patcher = patch.object(obj, key, val) # this is unittest.mock.patch + patcher.start() + cls.addClassCleanup(patcher.stop) + @contextmanager def with_user(self, login): """ Change user for a given test, like with self.with_user() ... """ @@ -1556,7 +1563,7 @@ class HttpCase(TransactionCase): odoo.tools.misc.dumpstacks() def logout(self, keep_db=True): - self.session.logout(keep_db=True) + self.session.logout(keep_db=keep_db) odoo.http.root.session_store.save(self.session) def authenticate(self, user, password):