Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 7 additions & 8 deletions webservice/models/webservice_backend.py
Original file line number Diff line number Diff line change
Expand Up @@ -39,8 +39,6 @@ class WebserviceBackend(models.Model):
("web_application", "Web Application (Authorization Code Grant)"),
],
readonly=False,
store=True,
compute="_compute_oauth2_flow",
)
oauth2_clientid = fields.Char(string="Client ID", auth_type="oauth2")
oauth2_client_secret = fields.Char(string="Client Secret", auth_type="oauth2")
Expand Down Expand Up @@ -122,12 +120,6 @@ def _get_adapter_protocol(self):
protocol += f"+{self.auth_type}-{self.oauth2_flow}"
return protocol

@api.depends("auth_type")
def _compute_oauth2_flow(self):
for rec in self:
if rec.auth_type != "oauth2":
rec.oauth2_flow = False

@api.depends("auth_type", "oauth2_flow")
def _compute_redirect_url(self):
get_param = self.env["ir.config_parameter"].sudo().get_param
Expand Down Expand Up @@ -177,3 +169,10 @@ def _server_env_fields(self):
}
webservice_fields.update(base_fields)
return webservice_fields

def _compute_server_env(self):
# OVERRIDE: reset ``oauth2_flow`` when ``auth_type`` is not "oauth2", even if
# defined otherwise in server env vars
res = super()._compute_server_env()
self.filtered(lambda r: r.auth_type != "oauth2").oauth2_flow = None
return res
92 changes: 92 additions & 0 deletions webservice/tests/test_oauth2.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,11 +4,17 @@
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


Expand Down Expand Up @@ -213,3 +219,89 @@ def test_fetch_token_from_auth(self):
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_recordset() # 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
)