[IMP] base: add _is_internal method to res.users

No method was readily available to know if a user is `internal` (has
group `base.group_user`), which was inconsistent with other base groups.

_is_internal is now used in the codebase where it is clear that
`.has_group('base.group_user')` is called on a single record.

Part-of: odoo/odoo#85703
This commit is contained in:
Florian Charlier
2022-06-14 09:35:57 +02:00
parent b5f3433adc
commit dae28c4b46
27 changed files with 36 additions and 32 deletions
+1 -1
View File
@@ -3307,7 +3307,7 @@ class AccountMove(models.Model):
def is_internal_partner(partner):
# Helper to know if the partner is an internal one.
return partner.user_ids and all(user.has_group('base.group_user') for user in partner.user_ids)
return partner.user_ids and all(user._is_internal() for user in partner.user_ids)
extra_domain = False
if custom_values.get('company_id'):
+1 -1
View File
@@ -149,7 +149,7 @@ class OAuthController(http.Controller):
resp.autocorrect_location_header = False
# Since /web is hardcoded, verify user has right to land on it
if werkzeug.urls.url_parse(resp.location).path == '/web' and not request.env.user.has_group('base.group_user'):
if werkzeug.urls.url_parse(resp.location).path == '/web' and not request.env.user._is_internal():
resp.location = '/'
return resp
except AttributeError:
+2 -2
View File
@@ -41,7 +41,7 @@ class ResPartner(models.Model):
""" proxy for function field towards actual implementation """
result = self.sudo()._get_signup_url_for_action()
for partner in self:
if any(u.has_group('base.group_user') for u in partner.user_ids if u != self.env.user):
if any(u._is_internal() for u in partner.user_ids if u != self.env.user):
self.env['res.users'].check_access_rights('write')
partner.signup_url = result.get(partner.id, False)
@@ -105,7 +105,7 @@ class ResPartner(models.Model):
""" Get a signup token related to the partner if signup is enabled.
If the partner already has a user, get the login parameter.
"""
if not self.env.user.has_group('base.group_user') and not self.env.is_admin():
if not self.env.user._is_internal() and not self.env.is_admin():
raise exceptions.AccessDenied()
res = defaultdict(dict)
@@ -31,7 +31,7 @@ class Users(models.Model):
otp_required = False
if ICP.get_param('auth_totp.policy') == 'all_required':
otp_required = True
elif ICP.get_param('auth_totp.policy') == 'employee_required' and self.has_group('base.group_user'):
elif ICP.get_param('auth_totp.policy') == 'employee_required' and self._is_internal():
otp_required = True
if otp_required:
return 'totp_mail'
+1 -1
View File
@@ -8,7 +8,7 @@ class Users(models.Model):
_inherit = 'res.users'
def get_totp_invite_url(self):
if not self.has_group('base.group_user'):
if not self._is_internal():
return '/my/security'
else:
return super(Users, self).get_totp_invite_url()
+1 -1
View File
@@ -9,7 +9,7 @@ class IrHttp(models.AbstractModel):
def session_info(self):
res = super(IrHttp, self).session_info()
if self.env.user.has_group('base.group_user'):
if self.env.user._is_internal():
res['max_time_between_keys_in_ms'] = int(
self.env['ir.config_parameter'].sudo().get_param('barcode.max_time_between_keys_in_ms', default='55'))
return res
+1 -1
View File
@@ -9,6 +9,6 @@ class IrHttp(models.AbstractModel):
def session_info(self):
result = super(IrHttp, self).session_info()
if request.env.user.has_group('base.group_user'):
if request.env.user._is_internal():
result['show_effect'] = bool(request.env['ir.config_parameter'].sudo().get_param('base_setup.show_effect'))
return result
+2 -2
View File
@@ -86,7 +86,7 @@ class Digest(models.Model):
# ------------------------------------------------------------
def action_subscribe(self):
if self.env.user.has_group('base.group_user') and self.env.user not in self.user_ids:
if self.env.user._is_internal() and self.env.user not in self.user_ids:
self._action_subscribe_users(self.env.user)
def _action_subscribe_users(self, users):
@@ -95,7 +95,7 @@ class Digest(models.Model):
self.sudo().user_ids |= users
def action_unsubscribe(self):
if self.env.user.has_group('base.group_user') and self.env.user in self.user_ids:
if self.env.user._is_internal() and self.env.user in self.user_ids:
self._action_unsubscribe_users(self.env.user)
def _action_unsubscribe_users(self, users):
@@ -11,7 +11,7 @@ class BusController(main.BusController):
@route('/longpolling/poll', type="json", auth="public")
def poll(self, channels, last, options=None):
if request.env.user.has_group('base.group_user'):
if request.env.user._is_internal():
ip_address = request.httprequest.remote_addr
users_log = request.env['res.users.log'].search_count([
('create_uid', '=', request.env.user.id),
+1 -1
View File
@@ -13,7 +13,7 @@ class Http(models.AbstractModel):
widget to apply, depending on the current company.
"""
result = super(Http, self).session_info()
if request.env.user.has_group('base.group_user'):
if request.env.user._is_internal():
company_ids = request.env.user.company_ids
for company in company_ids:
+1 -1
View File
@@ -12,7 +12,7 @@ class IrHttp(models.AbstractModel):
def session_info(self):
user = request.env.user
result = super(IrHttp, self).session_info()
if self.env.user.has_group('base.group_user'):
if self.env.user._is_internal():
result['notification_type'] = user.notification_type
assets_discuss_public_hash = HomeStaticTemplateHelpers.get_qweb_templates_checksum(debug=request.session.debug, bundle='mail.assets_discuss_public')
result['cache_hashes']['assets_discuss_public'] = assets_discuss_public_hash
+2 -2
View File
@@ -1773,7 +1773,7 @@ class MailThread(models.AbstractModel):
if filtered_attachment_ids:
filtered_attachment_ids.write({'res_model': model, 'res_id': res_id})
# prevent public and portal users from using attachments that are not theirs
if not self.env.user.has_group('base.group_user'):
if not self.env.user._is_internal():
attachment_ids = filtered_attachment_ids.ids
m2m_attachment_ids += [Command.link(id) for id in attachment_ids]
@@ -2264,7 +2264,7 @@ class MailThread(models.AbstractModel):
* either call the standard implementation _notify_cancel_by_type_generic
* or implements their own logic
"""
if not self.env.user.has_group('base.group_user'):
if not self.env.user._is_internal():
raise exceptions.AccessError(_("Access Denied"))
self.check_access_rights('read')
+1 -1
View File
@@ -9,6 +9,6 @@ class Http(models.AbstractModel):
def session_info(self):
res = super(Http, self).session_info()
if self.env.user.has_group('base.group_user'):
if self.env.user._is_internal():
res['odoobot_initialized'] = self.env.user.odoobot_state not in [False, 'not_initialized']
return res
+1 -1
View File
@@ -23,7 +23,7 @@ class Users(models.Model):
return super().SELF_READABLE_FIELDS + ['odoobot_state']
def _init_messaging(self):
if self.odoobot_state in [False, 'not_initialized'] and self.has_group('base.group_user'):
if self.odoobot_state in [False, 'not_initialized'] and self._is_internal():
self._init_odoobot()
return super()._init_messaging()
+1 -1
View File
@@ -173,7 +173,7 @@ class PortalChatter(http.Controller):
'options': {
'message_count': message_data['message_count'],
'is_user_public': is_user_public,
'is_user_employee': request.env.user.has_group('base.group_user'),
'is_user_employee': request.env.user._is_internal(),
'is_user_publisher': request.env.user.has_group('website.group_website_publisher'),
'display_composer': display_composer,
'partner_id': request.env.user.partner_id.id
+1 -1
View File
@@ -299,7 +299,7 @@ class CustomerPortal(Controller):
# Avoid using sudo or creating access_token when not necessary: internal
# users can create attachments, as opposed to public and portal users.
if not request.env.user.has_group('base.group_user'):
if not request.env.user._is_internal():
IrAttachment = IrAttachment.sudo().with_context(binary_field_real_user=IrAttachment.env.user)
access_token = IrAttachment._generate_access_token()
+1 -1
View File
@@ -100,7 +100,7 @@ class PortalWizardUser(models.TransientModel):
for portal_wizard_user in self:
user = portal_wizard_user.user_id
if user and user.has_group('base.group_user'):
if user and user._is_internal():
portal_wizard_user.is_internal = True
portal_wizard_user.is_portal = False
elif user and user.has_group('base.group_portal'):
+1 -1
View File
@@ -951,7 +951,7 @@ class Project(models.Model):
return False
if self.env.user.has_group('base.group_portal'):
return self.env.user.partner_id in self.collaborator_ids.partner_id
return self.env.user.has_group('base.group_user')
return self.env.user._is_internal()
def _add_collaborators(self, partners):
self.ensure_one()
+1 -1
View File
@@ -481,7 +481,7 @@ class Survey(models.Model):
# no signup possible -> should be a not public user (employee or portal users)
if not self.users_can_signup and (not user or user._is_public()):
raise exceptions.UserError(_('Creating token for external people is not allowed for surveys requesting authentication.'))
if self.access_mode == 'internal' and (not user or not user.has_group('base.group_user')):
if self.access_mode == 'internal' and (not user or not user._is_internal()):
raise exceptions.UserError(_('Creating token for anybody else than employees is not allowed for internal surveys.'))
if check_attempts and not self._has_attempts_left(partner or (user and user.partner_id), email, invite_token):
raise exceptions.UserError(_('No attempts left.'))
+1 -1
View File
@@ -201,7 +201,7 @@ def _get_login_redirect_url(uid, redirect=None):
def is_user_internal(uid):
return request.env['res.users'].sudo().browse(uid).has_group('base.group_user')
return request.env['res.users'].browse(uid)._is_internal()
def _local_web_translations(trans_file):
+1 -1
View File
@@ -25,7 +25,7 @@ class EditorCollaborationController(BusController):
res_id = int(match[3])
# Verify access to the edition channel.
if not request.env.user.has_group('base.group_user'):
if not request.env.user._is_internal():
raise AccessDenied()
document = request.env[model_name].browse([res_id])
+1 -1
View File
@@ -189,7 +189,7 @@ class Web_Editor(http.Controller):
def video_url_data(self, video_url, autoplay=False, loop=False,
hide_controls=False, hide_fullscreen=False, hide_yt_logo=False,
hide_dm_logo=False, hide_dm_share=False):
if not request.env.user.has_group('base.group_user'):
if not request.env.user._is_internal():
raise werkzeug.exceptions.Forbidden()
return get_video_url_data(
video_url, autoplay=autoplay, loop=loop,
+2 -2
View File
@@ -75,7 +75,7 @@ class Website(Home):
homepage_id = request.website._get_cached('homepage_id')
homepage = homepage_id and request.env['website.page'].browse(homepage_id)
if homepage and (homepage.sudo().is_visible or request.env.user.has_group('base.group_user')) and homepage.url != '/':
if homepage and (homepage.sudo().is_visible or request.env.user._is_internal()) and homepage.url != '/':
request.env['ir.http'].reroute(homepage.url)
website_page = request.env['ir.http']._serve_page()
@@ -127,7 +127,7 @@ class Website(Home):
the frontend
"""
if not redirect and request.params.get('login_success'):
if request.env['res.users'].browse(uid).has_group('base.group_user'):
if request.env['res.users'].browse(uid)._is_internal():
redirect = '/web?' + request.httprequest.query_string.decode()
else:
redirect = '/my'
+1 -1
View File
@@ -430,7 +430,7 @@ class IrAttachment(models.Model):
if self.env.is_superuser():
return True
# Always require an internal user (aka, employee) to access to a attachment
if not (self.env.is_admin() or self.env.user.has_group('base.group_user')):
if not (self.env.is_admin() or self.env.user._is_internal()):
raise AccessError(_("Sorry, you are not allowed to access this document."))
# collect the records to check (by model)
model_ids = defaultdict(set) # {model_name: set(ids)}
+1 -1
View File
@@ -608,7 +608,7 @@ class Partner(models.Model):
del vals['is_company']
result = result and super(Partner, self).write(vals)
for partner in self:
if any(u.has_group('base.group_user') for u in partner.user_ids if u != self.env.user):
if any(u._is_internal() for u in partner.user_ids if u != self.env.user):
self.env['res.users'].check_access_rights('write')
partner._fields_sync(vals)
return result
+6 -2
View File
@@ -916,6 +916,10 @@ class Users(models.Model):
'target': 'current',
}
def _is_internal(self):
self.ensure_one()
return not self.sudo().share
def _is_public(self):
self.ensure_one()
return self.has_group('base.group_public')
@@ -1157,9 +1161,9 @@ class UsersImplied(models.Model):
def write(self, values):
if not values.get('groups_id'):
return super(UsersImplied, self).write(values)
users_before = self.filtered(lambda u: u.has_group('base.group_user'))
users_before = self.filtered(lambda u: u._is_internal())
res = super(UsersImplied, self).write(values)
demoted_users = users_before.filtered(lambda u: not u.has_group('base.group_user'))
demoted_users = users_before.filtered(lambda u: not u._is_internal())
if demoted_users:
# demoted users are restricted to the assigned groups only
vals = {'groups_id': [Command.clear()] + values['groups_id']}
@@ -1419,7 +1419,7 @@ class TestFields(TransactionCaseWithUserDemo):
record.with_user(user0).foo = 'yes we can'
# add ir.rule to prevent access on record
self.assertTrue(user0.has_group('base.group_user'))
self.assertTrue(user0._is_internal())
rule = self.env['ir.rule'].create({
'model_id': self.env['ir.model']._get_id(record._name),
'groups': [self.env.ref('base.group_user').id],