diff --git a/test-requirements.txt b/test-requirements.txt
index 7d60827d033..8e4144e2102 100644
--- a/test-requirements.txt
+++ b/test-requirements.txt
@@ -2,3 +2,4 @@ invoice2data==0.3.5
freezegun
odoo_test_helper
pypdf>=3.15.0,<5.0
+httpretty
diff --git a/webservice/__manifest__.py b/webservice/__manifest__.py
index 4cfef1d84fa..f9d2b0bdb7c 100644
--- a/webservice/__manifest__.py
+++ b/webservice/__manifest__.py
@@ -8,7 +8,7 @@
"name": "WebService",
"summary": """
Defines webservice abstract definition to be used generally""",
- "version": "13.0.1.0.0",
+ "version": "12.0.1.0.0",
"license": "AGPL-3",
"development_status": "Production/Stable",
"maintainers": ["etobella"],
diff --git a/webservice/security/ir_rule.xml b/webservice/security/ir_rule.xml
index b59e4e5f94f..9942a3c843f 100644
--- a/webservice/security/ir_rule.xml
+++ b/webservice/security/ir_rule.xml
@@ -5,6 +5,6 @@
['|',('company_id','=',False),('company_id', 'in', company_ids)]
+ >['|',('company_id','=',False), ('company_id', 'child_of', [user.company_id.id])]
diff --git a/webservice/tests/common.py b/webservice/tests/common.py
index 220ce3131d6..5d14ed8b3bd 100644
--- a/webservice/tests/common.py
+++ b/webservice/tests/common.py
@@ -1,6 +1,7 @@
# Copyright 2020 Creu Blanca
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl).
-
+import mock
+from contextlib import contextmanager
from odoo.tests.common import tagged
from odoo.addons.component.tests.common import TransactionComponentCase
diff --git a/webservice/tests/test_oauth2.py b/webservice/tests/test_oauth2.py
deleted file mode 100644
index 3d7f05b984c..00000000000
--- a/webservice/tests/test_oauth2.py
+++ /dev/null
@@ -1,302 +0,0 @@
-# Copyright 2023 Camptocamp SA
-# @author Alexandre Fayolle
-# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl).
-import json
-import os
-import time
-from unittest import mock
-from urllib.parse import quote
-
-import responses
-from oauthlib.oauth2.rfc6749.errors import InvalidGrantError
-
-from odoo.tests.common import Form
-
-from odoo.addons.server_environment import server_env
-from odoo.addons.server_environment.models import server_env_mixin
-
-from .common import CommonWebService, mock_cursor
-
-
-class TestWebServiceOauth2BackendApplication(CommonWebService):
- def _setup_records(self):
-
- res = super(TestWebServiceOauth2BackendApplication, self)._setup_records()
- self.url = "https://localhost.demo.odoo/"
- os.environ["SERVER_ENV_CONFIG"] = "\n".join(
- [
- "[webservice_backend.test_oauth2_back]",
- "auth_type = oauth2",
- "oauth2_flow = backend_application",
- "oauth2_clientid = some_client_id",
- "oauth2_client_secret = shh_secret",
- f"oauth2_token_url = {self.url}oauth2/token",
- f"oauth2_audience = {self.url}",
- ]
- )
- self.webservice = self.env["webservice.backend"].create(
- {
- "name": "WebService OAuth2",
- "tech_name": "test_oauth2_back",
- "auth_type": "oauth2",
- "protocol": "http",
- "url": self.url,
- "oauth2_flow": "backend_application",
- "content_type": "application/xml",
- "oauth2_clientid": "some_client_id",
- "oauth2_client_secret": "shh_secret",
- "oauth2_token_url": f"{self.url}oauth2/token",
- "oauth2_audience": self.url,
- }
- )
- return res
-
- def test_get_adapter_protocol(self):
- protocol = self.webservice._get_adapter_protocol()
- self.assertEqual(protocol, "http+oauth2-backend_application")
-
- @responses.activate
- def test_fetch_token(self):
- duration = 3600
- expires_timestamp = time.time() + duration
- responses.add(
- responses.POST,
- f"{self.url}oauth2/token",
- json={
- "access_token": "cool_token",
- "expires_at": expires_timestamp,
- "expires_in": duration,
- "token_type": "Bearer",
- },
- )
- responses.add(responses.GET, f"{self.url}endpoint", body="OK")
-
- with mock_cursor(self.env.cr):
- result = self.webservice.call("get", url=f"{self.url}endpoint")
- self.webservice.invalidate_cache()
- self.assertTrue("cool_token" in self.webservice.oauth2_token)
- self.assertEqual(result, b"OK")
-
- @responses.activate
- def test_update_token(self):
- duration = 3600
- self.webservice.oauth2_token = json.dumps(
- {
- "access_token": "old_token",
- "expires_at": time.time() + 10, # in the near future
- "expires_in": duration,
- "token_type": "Bearer",
- }
- )
- self.webservice.flush()
-
- expires_timestamp = time.time() + duration
- responses.add(
- responses.POST,
- f"{self.url}oauth2/token",
- json={
- "access_token": "cool_token",
- "expires_at": expires_timestamp,
- "expires_in": duration,
- "token_type": "Bearer",
- },
- )
- responses.add(responses.GET, f"{self.url}endpoint", body="OK")
-
- with mock_cursor(self.env.cr):
- result = self.webservice.call("get", url=f"{self.url}endpoint")
- self.env.cr.commit.assert_called_once_with() # one call with no args
-
- self.webservice.invalidate_cache()
- self.assertTrue("cool_token" in self.webservice.oauth2_token)
- self.assertEqual(result, b"OK")
-
- @responses.activate
- def test_update_token_with_error(self):
- duration = 3600
- self.webservice.oauth2_token = json.dumps(
- {
- "access_token": "old_token",
- "expires_at": time.time() + 10, # in the near future
- "expires_in": duration,
- "token_type": "Bearer",
- }
- )
- self.webservice.flush()
-
- responses.add(
- responses.POST,
- f"{self.url}oauth2/token",
- json={"error": "invalid_grant", "error_description": "invalid grant"},
- status=404,
- )
- responses.add(responses.GET, f"{self.url}endpoint", body="NOK", status=403)
-
- with mock_cursor(self.env.cr):
- with self.assertRaises(InvalidGrantError):
- self.webservice.call("get", url=f"{self.url}endpoint")
- self.env.cr.commit.assert_not_called()
- self.env.cr.close.assert_called_once_with() # one call with no args
-
- self.webservice.invalidate_cache()
- self.assertTrue("old_token" in self.webservice.oauth2_token)
-
-
-class TestWebServiceOauth2WebApplication(CommonWebService):
- def _setup_records(self):
- res = super(TestWebServiceOauth2WebApplication, self)._setup_records()
- self.url = "https://localhost.demo.odoo/"
- os.environ["SERVER_ENV_CONFIG"] = "\n".join(
- [
- "[webservice_backend.test_oauth2_web]",
- "auth_type = oauth2",
- "oauth2_flow = web_application",
- "oauth2_clientid = some_client_id",
- "oauth2_client_secret = shh_secret",
- f"oauth2_token_url = {self.url}oauth2/token",
- f"oauth2_audience = {self.url}",
- f"oauth2_authorization_url = {self.url}/authorize",
- ]
- )
- self.webservice = self.env["webservice.backend"].create(
- {
- "name": "WebService OAuth2",
- "tech_name": "test_oauth2_web",
- "auth_type": "oauth2",
- "protocol": "http",
- "url": self.url,
- "oauth2_flow": "web_application",
- "content_type": "application/xml",
- "oauth2_clientid": "some_client_id",
- "oauth2_client_secret": "shh_secret",
- "oauth2_token_url": f"{self.url}oauth2/token",
- "oauth2_audience": self.url,
- "oauth2_authorization_url": f"{self.url}/authorize",
- }
- )
- return res
-
- def test_get_adapter_protocol(self):
- protocol = self.webservice._get_adapter_protocol()
- self.assertEqual(protocol, "http+oauth2-web_application")
-
- def test_authorization_code(self):
- action = self.webservice.button_authorize()
- expected_action = {
- "type": "ir.actions.act_url",
- "target": "self",
- "url": "https://localhost.demo.odoo//authorize?response_type=code&"
- "client_id=some_client_id&"
- f"redirect_uri={quote(self.webservice.redirect_url, safe='')}&state=",
- }
- self.assertEqual(action["type"], expected_action["type"])
- self.assertEqual(action["target"], expected_action["target"])
- self.assertTrue(
- action["url"].startswith(expected_action["url"]),
- f"Got url:\n{action['url']}\nexpected:\n{expected_action['url']}",
- )
-
- @responses.activate
- def test_fetch_token_from_auth(self):
- duration = 3600
- expires_timestamp = time.time() + duration
- responses.add(
- responses.POST,
- self.webservice.oauth2_token_url,
- json={
- "access_token": "cool_token",
- "expires_at": expires_timestamp,
- "expires_in": duration,
- "token_type": "Bearer",
- },
- )
- code = "some code"
- adapter = self.webservice._get_adapter()
- token = adapter._fetch_token_from_authorization(code)
- self.assertEqual("cool_token", token["access_token"])
-
- def test_oauth2_flow_compute_with_server_env(self):
- """Check the ``compute`` method when updating server envs"""
- ws = self.webservice
- url = self.url
- for auth_type, oauth2_flow in [
- (tp, fl)
- for tp in ws._fields["auth_type"].get_values(ws.env)
- for fl in ws._fields["oauth2_flow"].get_values(ws.env)
- ]:
- # Update env with current ``auth_type`` and ``oauth2_flow``
- with mock.patch.dict(
- os.environ,
- {
- "SERVER_ENV_CONFIG": f"""
-[webservice_backend.test_oauth2_web]
-auth_type = {auth_type}
-oauth2_flow = {oauth2_flow}
-oauth2_clientid = some_client_id
-oauth2_client_secret = shh_secret
-oauth2_token_url = {url}oauth2/token
-oauth2_audience = {url}
-oauth2_authorization_url = {url}/authorize
-""",
- },
- ):
- server_env_mixin.serv_config = server_env._load_config() # Reload vars
- ws.invalidate_cache() # Avoid reading from cache
- if auth_type == "oauth2":
- self.assertEqual(ws.oauth2_flow, oauth2_flow)
- else:
- self.assertFalse(ws.oauth2_flow)
-
- def test_oauth2_flow_compute_with_ui(self):
- """Check the ``compute`` method when updating WS from UI"""
- ws = self.webservice
- url = self.url
- form_xmlid = "webservice.webservice_backend_form_view"
- for auth_type, oauth2_flow in [
- (tp, fl)
- for tp in ws._fields["auth_type"].get_values(ws.env)
- for fl in ws._fields["oauth2_flow"].get_values(ws.env)
- ]:
- next_ws_id = ws.sudo().search([], order="id desc", limit=1).id + 1
- # Create a new WS with each ``auth_type/oauth2_flow`` couple through UI
- with Form(ws.browse(), form_xmlid) as ws_form:
- # Common fields
- ws_form.name = "WebService Test UI"
- ws_form.tech_name = f"webservice_test_ui_{next_ws_id}"
- ws_form.protocol = "http"
- ws_form.url = url
- ws_form.content_type = "application/xml"
- ws_form.auth_type = auth_type
- # Auth type specific fields
- if auth_type == "api_key":
- ws_form.api_key = "Test Api Key"
- ws_form.api_key_header = "Test Api Key Header"
- if auth_type == "oauth2":
- ws_form.oauth2_flow = oauth2_flow
- ws_form.oauth2_clientid = "Test Client ID"
- ws_form.oauth2_client_secret = "Test Client Secret"
- ws_form.oauth2_token_url = f"{url}oauth2/token"
- if auth_type == "user_pwd":
- ws_form.username = "Test Username"
- ws_form.password = "Test Password"
- ws = ws_form.save()
- # Check that ``oauth2_flow`` is the expected one after creation only if the
- # ``auth_type`` is "oauth2", else it should be False
- self.assertEqual(
- ws.oauth2_flow, oauth2_flow if ws.auth_type == "oauth2" else False
- )
- # Change WS's ``auth_type`` through UI
- with Form(ws, form_xmlid) as ws_form:
- new_auth_type = "none" if ws.auth_type == "oauth2" else "oauth2"
- ws_form.auth_type = new_auth_type
- if new_auth_type == "oauth2":
- ws_form.oauth2_flow = oauth2_flow
- ws_form.oauth2_clientid = "Test Client ID"
- ws_form.oauth2_client_secret = "Test Client Secret"
- ws_form.oauth2_token_url = f"{url}oauth2/token"
- ws = ws_form.save()
- # Check that ``oauth2_flow`` is the expected one after update only if the
- # ``auth_type`` is "oauth2", else it should be False
- self.assertEqual(
- ws.oauth2_flow, oauth2_flow if ws.auth_type == "oauth2" else False
- )
diff --git a/webservice/tests/test_webservice.py b/webservice/tests/test_webservice.py
index 2ac5699d293..83df4ea6497 100644
--- a/webservice/tests/test_webservice.py
+++ b/webservice/tests/test_webservice.py
@@ -3,7 +3,7 @@
# @author Simone Orsi
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl).
-import responses
+import httpretty
from requests import auth, exceptions as http_exceptions
from odoo import exceptions
@@ -37,15 +37,19 @@ def test_auth_validation(self):
r"requires 'Username & password' authentication. "
r"However, the following field\(s\) are not valued: Username, Password"
)
- with self.assertRaisesRegex(exceptions.UserError, msg):
- self.webservice.write({"auth_type": "user_pwd"})
+ with self.assertRaisesRegex(exceptions.ValidationError, msg):
+ self.webservice.write(
+ {
+ "auth_type": "user_pwd",
+ }
+ )
msg = (
r"Webservice 'WebService' "
r"requires 'Username & password' authentication. "
r"However, the following field\(s\) are not valued: Password"
)
- with self.assertRaisesRegex(exceptions.UserError, msg):
+ with self.assertRaisesRegex(exceptions.ValidationError, msg):
self.webservice.write({"auth_type": "user_pwd", "username": "user"})
msg = (
@@ -53,149 +57,136 @@ def test_auth_validation(self):
r"requires 'API Key' authentication. "
r"However, the following field\(s\) are not valued: API Key, API Key header"
)
- with self.assertRaisesRegex(exceptions.UserError, msg):
- self.webservice.write({"auth_type": "api_key"})
+ with self.assertRaisesRegex(exceptions.ValidationError, msg):
+ self.webservice.write(
+ {
+ "auth_type": "api_key",
+ }
+ )
msg = (
r"Webservice 'WebService' "
r"requires 'API Key' authentication. "
r"However, the following field\(s\) are not valued: API Key header"
)
- with self.assertRaisesRegex(exceptions.UserError, msg):
- self.webservice.write({"auth_type": "api_key", "api_key": "foo"})
+ with self.assertRaisesRegex(exceptions.ValidationError, msg):
+ self.webservice.write(
+ {
+ "auth_type": "api_key",
+ "api_key": "foo",
+ }
+ )
- @responses.activate
+ @httpretty.activate
def test_web_service_get(self):
- responses.add(responses.GET, self.url, body="{}")
+ httpretty.register_uri(httpretty.GET, self.url, body="{}")
result = self.webservice.call("get")
self.assertEqual(result, b"{}")
- self.assertEqual(len(responses.calls), 1)
+ self.assertEqual(len(httpretty.latest_requests()), 1)
self.assertEqual(
- responses.calls[0].request.headers["Content-Type"], "application/xml"
+ httpretty.latest_requests()[0].headers["Content-Type"], "application/xml"
)
- @responses.activate
- def test_web_service_get_url_combine(self):
- endpoint = "api/test"
- responses.add(responses.GET, self.url + endpoint, body="{}")
- result = self.webservice.call("get", url="api/test")
- self.assertEqual(result, b"{}")
- self.assertEqual(len(responses.calls), 1)
- self.assertEqual(
- responses.calls[0].request.headers["Content-Type"], "application/xml"
- )
-
- @responses.activate
- def test_web_service_get_url_combine_full_url(self):
- endpoint = "api/test"
- responses.add(responses.GET, self.url + endpoint, body="{}")
- result = self.webservice.call("get", url="https://localhost.demo.odoo/api/test")
- self.assertEqual(result, b"{}")
- self.assertEqual(len(responses.calls), 1)
- self.assertEqual(
- responses.calls[0].request.headers["Content-Type"], "application/xml"
- )
-
- @responses.activate
+ @httpretty.activate
def test_web_service_post(self):
- responses.add(responses.POST, self.url, body="{}")
+ httpretty.register_uri(httpretty.POST, self.url, body="{}")
result = self.webservice.call("post", data="demo_response")
self.assertEqual(result, b"{}")
self.assertEqual(
- responses.calls[0].request.headers["Content-Type"], "application/xml"
+ httpretty.latest_requests()[0].headers["Content-Type"], "application/xml"
)
- self.assertEqual(responses.calls[0].request.body, "demo_response")
+ self.assertEqual(httpretty.latest_requests()[0].body, b"demo_response")
- @responses.activate
+ @httpretty.activate
def test_web_service_put(self):
- responses.add(responses.PUT, self.url, body="{}")
+ httpretty.register_uri(httpretty.PUT, self.url, body="{}")
result = self.webservice.call("put", data="demo_response")
self.assertEqual(result, b"{}")
self.assertEqual(
- responses.calls[0].request.headers["Content-Type"], "application/xml"
+ httpretty.latest_requests()[0].headers["Content-Type"], "application/xml"
)
- self.assertEqual(responses.calls[0].request.body, "demo_response")
+ self.assertEqual(httpretty.latest_requests()[0].body, b"demo_response")
- @responses.activate
+ @httpretty.activate
def test_web_service_backend_username(self):
self.webservice.write(
{"auth_type": "user_pwd", "username": "user", "password": "pass"}
)
- responses.add(responses.GET, self.url, body="{}")
+ httpretty.register_uri(httpretty.GET, self.url, body="{}")
result = self.webservice.call("get")
self.assertEqual(result, b"{}")
- self.assertEqual(len(responses.calls), 1)
+ self.assertEqual(len(httpretty.latest_requests()), 1)
self.assertEqual(
- responses.calls[0].request.headers["Content-Type"], "application/xml"
+ httpretty.latest_requests()[0].headers["Content-Type"], "application/xml"
)
data = auth._basic_auth_str("user", "pass")
- self.assertEqual(responses.calls[0].request.headers["Authorization"], data)
+ self.assertEqual(httpretty.latest_requests()[0].headers["Authorization"], data)
- @responses.activate
+ @httpretty.activate
def test_web_service_username(self):
self.webservice.write(
{"auth_type": "user_pwd", "username": "user", "password": "pass"}
)
- responses.add(responses.GET, self.url, body="{}")
+ httpretty.register_uri(httpretty.GET, self.url, body="{}")
result = self.webservice.call("get", auth=("user2", "pass2"))
self.assertEqual(result, b"{}")
- self.assertEqual(len(responses.calls), 1)
+ self.assertEqual(len(httpretty.latest_requests()), 1)
self.assertEqual(
- responses.calls[0].request.headers["Content-Type"], "application/xml"
+ httpretty.latest_requests()[0].headers["Content-Type"], "application/xml"
)
data = auth._basic_auth_str("user2", "pass2")
- self.assertEqual(responses.calls[0].request.headers["Authorization"], data)
+ self.assertEqual(httpretty.latest_requests()[0].headers["Authorization"], data)
- @responses.activate
+ @httpretty.activate
def test_web_service_backend_api_key(self):
self.webservice.write(
{"auth_type": "api_key", "api_key": "123xyz", "api_key_header": "Api-Key"}
)
- responses.add(responses.POST, self.url, body="{}")
+ httpretty.register_uri(httpretty.POST, self.url, body="{}")
result = self.webservice.call("post")
self.assertEqual(result, b"{}")
- self.assertEqual(len(responses.calls), 1)
+ self.assertEqual(len(httpretty.latest_requests()), 1)
self.assertEqual(
- responses.calls[0].request.headers["Content-Type"], "application/xml"
+ httpretty.latest_requests()[0].headers["Content-Type"], "application/xml"
)
- self.assertEqual(responses.calls[0].request.headers["Api-Key"], "123xyz")
+ self.assertEqual(httpretty.latest_requests()[0].headers["Api-Key"], "123xyz")
- @responses.activate
+ @httpretty.activate
def test_web_service_headers(self):
- responses.add(responses.GET, self.url, body="{}")
+ httpretty.register_uri(httpretty.GET, self.url, body="{}")
result = self.webservice.call("get", headers={"demo_header": "HEADER"})
self.assertEqual(result, b"{}")
- self.assertEqual(len(responses.calls), 1)
+ self.assertEqual(len(httpretty.latest_requests()), 1)
self.assertEqual(
- responses.calls[0].request.headers["Content-Type"], "application/xml"
+ httpretty.latest_requests()[0].headers["Content-Type"], "application/xml"
)
- self.assertEqual(responses.calls[0].request.headers["demo_header"], "HEADER")
+ self.assertEqual(httpretty.latest_requests()[0].headers["demo_header"], "HEADER")
- @responses.activate
+ @httpretty.activate
def test_web_service_call_args(self):
url = "https://custom.url"
- responses.add(responses.POST, url, body="{}")
+ httpretty.register_uri(httpretty.POST, url, body="{}")
result = self.webservice.call(
"post", url=url, headers={"demo_header": "HEADER"}
)
self.assertEqual(result, b"{}")
- self.assertEqual(len(responses.calls), 1)
+ self.assertEqual(len(httpretty.latest_requests()), 1)
self.assertEqual(
- responses.calls[0].request.headers["Content-Type"], "application/xml"
+ httpretty.latest_requests()[0].headers["Content-Type"], "application/xml"
)
- self.assertEqual(responses.calls[0].request.headers["demo_header"], "HEADER")
+ self.assertEqual(httpretty.latest_requests()[0].headers["demo_header"], "HEADER")
url = self.url + "custom/path"
self.webservice.url += "{endpoint}"
- responses.add(responses.POST, url, body="{}")
+ httpretty.register_uri(httpretty.POST, url, body="{}")
result = self.webservice.call(
"post",
url_params={"endpoint": "custom/path"},
headers={"demo_header": "HEADER"},
)
self.assertEqual(result, b"{}")
- self.assertEqual(len(responses.calls), 2)
+ self.assertEqual(len(httpretty.latest_requests()), 2)
self.assertEqual(
- responses.calls[0].request.headers["Content-Type"], "application/xml"
+ httpretty.latest_requests()[0].headers["Content-Type"], "application/xml"
)
- self.assertEqual(responses.calls[0].request.headers["demo_header"], "HEADER")
+ self.assertEqual(httpretty.latest_requests()[0].headers["demo_header"], "HEADER")
diff --git a/webservice/views/webservice_backend.xml b/webservice/views/webservice_backend.xml
index 9fcf304e609..19af3319434 100644
--- a/webservice/views/webservice_backend.xml
+++ b/webservice/views/webservice_backend.xml
@@ -89,7 +89,6 @@