Coverage for src/backend/InvenTree/users/tasks.py: 52%
94 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-07 17:47 +0000
1"""Background tasks for the users app."""
3from typing import Any
5from django.contrib.auth.models import Group, Permission
6from django.contrib.contenttypes.models import ContentType
8import structlog
10from InvenTree.ready import canAppAccessDatabase
11from users.models import RuleSet
12from users.permissions import get_model_permission_string, split_permission
13from users.ruleset import RULESET_CHANGE_INHERIT, RULESET_CHOICES, RULESET_NAMES
15logger = structlog.get_logger('inventree')
18def rebuild_all_permissions() -> None:
19 """Rebuild all user permissions.
21 This function is called when a user is created or when a group is modified.
22 It rebuilds the permissions for all users in the system.
23 """
24 logger.info('Rebuilding permissions')
26 # Rebuild permissions for each group
27 for group in Group.objects.all(): 27 ↛ 28line 27 didn't jump to line 28 because the loop on line 27 never started
28 update_group_roles(group)
31def update_group_roles(group: Group, debug: bool = False) -> None:
32 """Update the roles for a particular group.
34 Arguments:
35 group: The group object to update roles for.
36 debug: Whether to enable debug logging
38 This function performs the following tasks:
39 - Remove any RuleSet objects which have become outdated
40 - Ensure that the group has a mapped RuleSet for each role
41 - Rebuild the permissions for the group, based on the assigned RuleSet objects
42 """
43 if not canAppAccessDatabase(allow_test=True): 43 ↛ 44line 43 didn't jump to line 44 because the condition on line 43 was never true
44 return # pragma: no cover
46 logger.info('Updating group roles for %s', group)
48 # Remove any outdated RuleSet objects
49 outdated_rules = group.rule_sets.exclude(name__in=RULESET_NAMES)
51 if outdated_rules.exists(): 51 ↛ 52line 51 didn't jump to line 52 because the condition on line 51 was never true
52 logger.info(
53 'Deleting %s outdated rulesets from group %s', outdated_rules.count(), group
54 )
55 outdated_rules.delete()
57 # Add any missing RuleSet objects
58 for rule in RULESET_NAMES:
59 if not group.rule_sets.filter(name=rule).exists():
60 logger.info('Adding ruleset %s to group %s', rule, group)
61 RuleSet.objects.create(group=group, name=rule)
63 # Update the permissions for the group
64 # List of permissions already associated with this group
65 group_permissions = set()
67 # Iterate through each permission already assigned to this group,
68 # and create a simplified permission key string
69 for p in group.permissions.all().prefetch_related('content_type'): 69 ↛ 70line 69 didn't jump to line 70 because the loop on line 69 never started
70 (permission, app, model) = p.natural_key()
71 permission_string = f'{app}.{permission}'
72 group_permissions.add(permission_string)
74 # List of permissions which must be added to the group
75 permissions_to_add = set()
77 # List of permissions which must be removed from the group
78 permissions_to_delete = set()
80 def add_model(name, action, allowed):
81 """Add a new model to the pile.
83 Args:
84 name: The name of the model e.g. part_part
85 action: The permission action e.g. view
86 allowed: Whether or not the action is allowed
87 """
88 if action not in ['view', 'add', 'change', 'delete']: # pragma: no cover 88 ↛ 89line 88 didn't jump to line 89 because the condition on line 88 was never true
89 raise ValueError(f'Action {action} is invalid')
91 permission_string = get_model_permission_string(model, action)
93 if allowed: 93 ↛ 95line 93 didn't jump to line 95 because the condition on line 93 was never true
94 # An 'allowed' action is always preferenced over a 'forbidden' action
95 if permission_string in permissions_to_delete:
96 permissions_to_delete.remove(permission_string)
98 permissions_to_add.add(permission_string)
100 elif permission_string not in permissions_to_add: 100 ↛ exitline 100 didn't return from function 'add_model' because the condition on line 100 was always true
101 permissions_to_delete.add(permission_string)
103 # Pre-fetch all the RuleSet objects
104 rulesets: dict[Any, RuleSet] = {
105 r.name: r for r in RuleSet.objects.filter(group=group).prefetch_related('group')
106 }
108 # Get all the rulesets associated with this group
109 for rule_name, _rule_label in RULESET_CHOICES:
110 if rule_name in rulesets: 110 ↛ 113line 110 didn't jump to line 113 because the condition on line 110 was always true
111 ruleset = rulesets[rule_name]
112 else:
113 try:
114 ruleset = RuleSet.objects.get(group=group, name=rule_name)
115 except RuleSet.DoesNotExist:
116 ruleset = RuleSet.objects.create(group=group, name=rule_name)
118 # Which database tables does this RuleSet touch?
119 models = ruleset.get_models()
121 for model in models:
122 # Keep track of the available permissions for each model
123 add_model(model, 'view', ruleset.can_view)
124 add_model(model, 'add', ruleset.can_add)
125 add_model(model, 'change', ruleset.can_change)
126 add_model(model, 'delete', ruleset.can_delete)
128 def get_permission_object(permission_string):
129 """Find the permission object in the database, from the simplified permission string.
131 Args:
132 permission_string: a simplified permission_string e.g. 'part.view_partcategory'
134 Returns the permission object in the database associated with the permission string
135 """
136 (app, perm) = permission_string.split('.')
138 perm, model = split_permission(app, perm)
139 permission = None
141 try:
142 content_type = ContentType.objects.get(app_label=app, model=model)
143 permission = Permission.objects.get(
144 content_type=content_type, codename=perm
145 )
146 except ContentType.DoesNotExist: # pragma: no cover
147 logger.warning("No ContentType found matching '%s' and '%s'", app, model)
148 except Permission.DoesNotExist:
149 logger.warning("No Permission found matching '%s' and '%s'", app, perm)
151 return permission
153 # Add any required permissions to the group
154 for perm in permissions_to_add: 154 ↛ 156line 154 didn't jump to line 156 because the loop on line 154 never started
155 # Ignore if permission is already in the group
156 if perm in group_permissions:
157 continue
159 if permission := get_permission_object(perm):
160 group.permissions.add(permission)
161 if debug: # pragma: no cover
162 logger.debug('Adding permission %s to group %s', perm, group.name)
164 # Remove any extra permissions from the group
165 for perm in permissions_to_delete:
166 # Ignore if the permission is not already assigned
167 if perm not in group_permissions: 167 ↛ 170line 167 didn't jump to line 170 because the condition on line 167 was always true
168 continue
170 if permission := get_permission_object(perm):
171 group.permissions.remove(permission)
172 if debug: # pragma: no cover
173 logger.debug('Removing permission %s from group %s', perm, group.name)
175 # Enable all action permissions for certain children models
176 # if parent model has 'change' permission
177 for parent, child in RULESET_CHANGE_INHERIT:
178 parent_child_string = f'{parent}_{child}'
180 # Check each type of permission
181 for action in ['view', 'change', 'add', 'delete']:
182 parent_perm = f'{parent}.{action}_{parent}'
184 if parent_perm in group_permissions: 184 ↛ 185line 184 didn't jump to line 185 because the condition on line 184 was never true
185 child_perm = f'{parent}.{action}_{child}'
187 # Check if child permission not already in group
188 if child_perm not in group_permissions:
189 # Create permission object
190 add_model(parent_child_string, action, ruleset.can_delete)
191 # Add to group
192 permission = get_permission_object(child_perm)
193 if permission:
194 group.permissions.add(permission)
195 logger.debug(
196 'Adding permission %s to group %s', child_perm, group.name
197 )