pengyuthon/odoo_source
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
80098df3dddd65295988221477aea8a5dba7bc5c
odoo_source/odoo/addons/test_http/tests/test_misc.py
T
Mathieu Walravens 1d87710fdc [FIX] http: rewind file upload on serialization failure 
Before this commit:
When uploading a file, if the transaction fails due to a serialization
failure, Odoo will retry the request. However, if a file upload is read
during the transaction, the file pointer will be at the end of the file,
and calling `.read()` again returns an empty bytes object.

After this commit:
Upon retrying the request, rewind uploads to the beginning of the file,
if the file supports it.

opw-3228200

closes odoo/odoo#120180

X-original-commit: ac59ef0668122ad71dffbb5575250c767a0a56ec
Signed-off-by: Julien Castiaux (juc) <juc@odoo.com>
2023-04-28 20:29:55 +02:00

237 lines
11 KiB
Python
Raw Blame History

# 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
from urllib.parse import urlparse
import odoo
from odoo.http import root
from odoo.tests import tagged
from odoo.tests.common import HOST, new_test_user, get_db_name
from odoo.tools import config, file_path
from odoo.addons.test_http.controllers import CT_JSON
from odoo.addons.test_http.utils import TEST_IP
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, 404)
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_geoip(self):
res = self.nodb_url_open('/test_http/geoip')
res.raise_for_status()
self.assertEqual(res.json(), {
'city': None,
'country_code': None,
'country_name': None,
'latitude': None,
'longitude': None,
'region': None,
'time_zone': None,
})
# Fake client IP using proxy_mode and a forged X-Forwarded-For http header
headers = {
'Host': '',
'X-Forwarded-For': TEST_IP,
'X-Forwarded-Host': 'odoo.com',
'X-Forwarded-Proto': 'https'
}
with patch.dict('odoo.tools.config.options', {'proxy_mode': True}):
res = self.nodb_url_open('/test_http/geoip', headers=headers)
res.raise_for_status()
self.assertEqual(res.json(), {
'city': None,
'country_code': 'FR',
'country_name': 'France',
'latitude': 48.8582,
'longitude': 2.3387,
'region': None,
'time_zone': 'Europe/Paris',
})
def test_misc6_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.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')
Reference in New Issue View Git Blame Copy Permalink