[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 <rco@odoo.com>
This commit is contained in:
Julien Castiaux
@@ -0,0 +1,2 @@
|
||||
from . import controllers
|
||||
from . import models
|
||||
@@ -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',
|
||||
}
|
||||
@@ -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/<model("test_http.galaxy"):galaxy>', 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/<model("test_http.galaxy"):galaxy>/<model("test_http.stargate"):gate>', 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 {}
|
||||
@@ -0,0 +1,43 @@
|
||||
<odoo>
|
||||
<data>
|
||||
<record id="milky_way" model="test_http.galaxy">
|
||||
<field name="name">Milky Way</field>
|
||||
</record>
|
||||
|
||||
<record id="pegasus" model="test_http.galaxy">
|
||||
<field name="name">Pegasus</field>
|
||||
</record>
|
||||
|
||||
|
||||
<record id="earth" model="test_http.stargate">
|
||||
<field name="name">Earth</field>
|
||||
<field name="address">sq5Abt</field>
|
||||
<field name="galaxy_id" ref="test_http.milky_way"/>
|
||||
</record>
|
||||
|
||||
<record id="abydos" model="test_http.stargate">
|
||||
<field name="name">Abydos</field>
|
||||
<field name="address">r7fwcu</field>
|
||||
<field name="galaxy_id" ref="test_http.milky_way"/>
|
||||
</record>
|
||||
|
||||
<record id="dakara" model="test_http.stargate">
|
||||
<field name="name">Dakara</field>
|
||||
<field name="address">gs38x4</field>
|
||||
<field name="galaxy_id" ref="test_http.milky_way"/>
|
||||
</record>
|
||||
|
||||
<record id="lantea" model="test_http.stargate">
|
||||
<field name="name">Lantea</field>
|
||||
<field name="address">u6iz7d</field>
|
||||
<field name="galaxy_id" ref="test_http.pegasus"/>
|
||||
<field name="has_galaxy_crystal" eval="True"/>
|
||||
</record>
|
||||
|
||||
<record id="athos" model="test_http.stargate">
|
||||
<field name="name">Athos</field>
|
||||
<field name="address">mfp2rt</field>
|
||||
<field name="galaxy_id" ref="test_http.pegasus"/>
|
||||
</record>
|
||||
</data>
|
||||
</odoo>
|
||||
@@ -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
|
||||
|
@@ -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.')
|
||||
Binary file not shown.
|
After Width: | Height: | Size: 814 B |
@@ -0,0 +1,22 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<!DOCTYPE svg PUBLIC "-//W3C//DTD SVG 1.1//EN" "http://www.w3.org/Graphics/SVG/1.1/DTD/svg11.dtd">
|
||||
<!-- Creator: CorelDRAW -->
|
||||
<svg xmlns="http://www.w3.org/2000/svg" xml:space="preserve" width="63.509mm" height="63.509mm" style="shape-rendering:geometricPrecision; text-rendering:geometricPrecision; image-rendering:optimizeQuality; fill-rule:evenodd; clip-rule:evenodd"
|
||||
viewBox="0 0 63.509 63.509">
|
||||
<defs>
|
||||
<style type="text/css">
|
||||
<![CDATA[
|
||||
.fil1 {fill:none}
|
||||
.fil0 {fill:#1F1A17}
|
||||
]]>
|
||||
</style>
|
||||
</defs>
|
||||
<g id="Layer_x0020_1">
|
||||
<metadata id="CorelCorpID_0Corel-Layer"/>
|
||||
<g id="_150956456">
|
||||
<path id="_151090152" class="fil0" d="M31.6671 7.4461c4.8933,0 8.861,3.9677 8.861,8.8598 0,4.8932 -3.9677,8.8609 -8.861,8.8609 -4.892,0 -8.8597,-3.9677 -8.8597,-8.8609 0,-4.8921 3.9677,-8.8598 8.8597,-8.8598zm0.0006 4.3597c2.4855,0 4.5007,2.0151 4.5007,4.5007 0,2.4855 -2.0152,4.5006 -4.5007,4.5006 -2.4855,0 -4.5006,-2.0151 -4.5006,-4.5006 0,-2.4856 2.0151,-4.5007 4.5006,-4.5007z"/>
|
||||
<path id="_150956480" class="fil0" d="M44.2053 53.9278c-2.4381,4.0631 -2.4381,4.0631 -2.4381,4.0631l0 -0.0656 12.9247 0 -23.0242 -31.2605c-23.0047,31.2605 -23.0047,31.2605 -23.0047,31.2605l12.9071 0 0 0.0656c0,0 0,0 -2.4381,-4.0631l-0.0049 0.0067c3.1437,-4.2692 7.2299,-9.8181 12.5406,-17.0292l0 0c5.3108,7.2111 9.3982,12.7594 12.5425,17.0286l-0.0049 -0.0061z"/>
|
||||
</g>
|
||||
<rect class="fil1" x="-0.0864984" y="0.170202" width="63.509" height="63.509"/>
|
||||
</g>
|
||||
</svg>
|
||||
|
After Width: | Height: | Size: 1.5 KiB |
@@ -0,0 +1 @@
|
||||
from . import test_http
|
||||
@@ -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"</{tag}>")
|
||||
|
||||
def handle_startendtag(self, tag, attrs):
|
||||
# HTML5 <img> instead of XHTML <img/>
|
||||
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('''\
|
||||
<p>Milky Way</p>
|
||||
<ul>
|
||||
<li><a href="/test_http/1/1">Earth (P4X-126)</a></li>
|
||||
<li><a href="/test_http/1/2">Abydos (P2X-125)</a></li>
|
||||
<li><a href="/test_http/1/3">Dakara (P5C-113)</a></li>
|
||||
</ul>
|
||||
''')
|
||||
)
|
||||
|
||||
@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('''\
|
||||
<dl>
|
||||
<dt>name</dt><dd>Earth</dd>
|
||||
<dt>address</dt><dd>sq5Abt</dd>
|
||||
<dt>sgc_designation</dt><dd>P4X-126</dd>
|
||||
</dl>
|
||||
''')
|
||||
)
|
||||
|
||||
@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')
|
||||
@@ -0,0 +1,25 @@
|
||||
<odoo>
|
||||
<data>
|
||||
<template id="tmpl_galaxy">
|
||||
<p><t t-out="galaxy.name"/></p>
|
||||
<ul>
|
||||
<li t-foreach="stargates" t-as="gate">
|
||||
<a t-attf-href="/test_http/#{ galaxy.id }/#{ gate.id }">
|
||||
<t t-out="gate.name"/> (<t t-out="gate.sgc_designation"/>)
|
||||
</a>
|
||||
</li>
|
||||
</ul>
|
||||
</template>
|
||||
|
||||
<template id="tmpl_stargate">
|
||||
<dl>
|
||||
<t t-if="gate.galaxy_id.id != 1">
|
||||
<dt>galaxy</dt><dd><t t-out="gate.galaxy_id.name"/></dd>
|
||||
</t>
|
||||
<t t-foreach="('name', 'address', 'sgc_designation')" t-as="field">
|
||||
<dt><t t-out="field"/></dt><dd><t t-out="gate[field]"/></dd>
|
||||
</t>
|
||||
</dl>
|
||||
</template>
|
||||
</data>
|
||||
</odoo>
|
||||
@@ -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):
|
||||
|
||||
Reference in New Issue
Block a user