diff --git a/loko/assets/permissions.py b/loko/assets/permissions.py index 382f27b..060967a 100644 --- a/loko/assets/permissions.py +++ b/loko/assets/permissions.py @@ -156,14 +156,124 @@ def _check_instance_permission(user, obj, permission_field, thematic_fallback_fn return False +def get_asset_contract_ids(asset): + """ + Retourne l'ensemble des IDs de contrats associés à une instance d'asset ou de localisation, + ou None si le modèle d'asset ne gère pas de contrats. + """ + if asset is None: + return None + + contract_ids = set() + has_contract_relation = False + + # 1. Clés étrangères directes + if hasattr(asset, 'maintenance_contract_id') and asset.maintenance_contract_id: + contract_ids.add(asset.maintenance_contract_id) + has_contract_relation = True + elif hasattr(asset, 'maintenance_contract'): + has_contract_relation = True + + if hasattr(asset, 'controller_maintenance_contract_id') and asset.controller_maintenance_contract_id: + contract_ids.add(asset.controller_maintenance_contract_id) + has_contract_relation = True + elif hasattr(asset, 'controller_maintenance_contract'): + has_contract_relation = True + + if hasattr(asset, 'contract_id') and asset.contract_id: + contract_ids.add(asset.contract_id) + has_contract_relation = True + elif hasattr(asset, 'contract'): + has_contract_relation = True + + # 2. Relations ManyToMany ou Reverse managers (contracts, etc.) + if hasattr(asset, 'contracts'): + has_contract_relation = True + try: + model = asset.contracts.model + if model.__name__ == 'Contract': + for cid in asset.contracts.values_list('pk', flat=True): + contract_ids.add(cid) + elif hasattr(model, 'contract_id'): + qs = asset.contracts.all() + if hasattr(model, 'status'): + qs = qs.filter(status='active') + for cid in qs.values_list('contract_id', flat=True): + contract_ids.add(cid) + except Exception: + pass + + for rel_name in [ + 'trafficlight_contracts', 'structure_contracts', 'its_locations_contracts', + 'its_assets_contracts', 'controlcenters_contracts', 'controlcenters_assets_contracts', + 'clean_location_contracts' + ]: + if hasattr(asset, rel_name): + has_contract_relation = True + try: + manager = getattr(asset, rel_name) + if hasattr(manager.model, 'status'): + cids = manager.filter(status='active').values_list('contract_id', flat=True) + else: + cids = manager.values_list('contract_id', flat=True) + contract_ids.update(cids) + except Exception: + pass + + # 3. Assets enfants liés à un parent (intersection, location, structure, control_center, pole, etc.) + for parent_attr in ['intersection', 'location', 'structure', 'control_center', 'building', 'pole']: + if hasattr(asset, parent_attr): + parent = getattr(asset, parent_attr) + if parent: + parent_cids = get_asset_contract_ids(parent) + if parent_cids is not None: + has_contract_relation = True + contract_ids.update(parent_cids) + + if not has_contract_relation: + return None + return contract_ids + + def can_view_asset(user, asset): """Check if user can view a specific asset.""" + if not user or not user.is_authenticated: + return False + + try: + user_config = UserConfig.objects.get(user=user) + except UserConfig.DoesNotExist: + user_config = None + + if getattr(user, 'is_superuser', False) or getattr(user, 'is_staff', False): + return True + if user_config and (user_config.has_role('admin') or user_config.has_role('top_manager')): + return True + def _fallback(): thematic = _get_asset_thematic(asset) if not thematic: return False can_view, _ = get_user_asset_thematic_access(user, thematic) - return can_view + if not can_view: + return False + + # Vérifier la restriction par contrat pour les utilisateurs externes + is_external = (user_config is None) or (not user_config.is_intern) + must_limit = is_external and (not user_config or user_config.limit_assets_to_contracts) + if must_limit and user_config: + asset_cids = get_asset_contract_ids(asset) + if asset_cids is not None: + from common.models import UserContractAccess + accessible_cids = set( + UserContractAccess.objects.filter( + user_config=user_config, + can_view_assets=True + ).values_list('contract_id', flat=True) + ) + if not (asset_cids & accessible_cids): + return False + return True return _check_instance_permission(user, asset, 'can_view', _fallback) @@ -276,27 +386,12 @@ def can_delete_location(user, location): def _get_asset_thematic(asset): - """Get the thematic for an asset.""" + """Get the thematic for an asset instance or model class.""" + if asset is None: + return None + from common.models import Thematic - - # Try from category - if hasattr(asset, 'category') and asset.category: - return asset.category.thematic - # Try direct thematic attribute - if hasattr(asset, 'thematic') and asset.thematic: - return asset.thematic - - # Try from parent building category - if hasattr(asset, 'building') and asset.building and hasattr(asset.building, 'category') and asset.building.category: - return asset.building.category.thematic - - # Try from model's category - if hasattr(asset, 'model') and asset.model and hasattr(asset.model, 'category') and asset.model.category: - return asset.model.category.thematic - - # Fallback: deduce thematic from asset class name - class_name = asset.__class__.__name__.lower() thematic_mappings = { 'naturerwiz': 'water', 'naturerwiasset': 'water', @@ -317,8 +412,60 @@ def _get_asset_thematic(asset): 'parking': 'parking', 'parkingspot': 'parking', 'parkinglocation': 'parking', + 'clean': 'clean', + 'cleanlocation': 'clean', + 'cleanlitterbin': 'clean', + 'cleanglasscontainer': 'clean', } + + if isinstance(asset, type): + class_name = asset.__name__.lower() + for prefix, thematic_code in thematic_mappings.items(): + if class_name.startswith(prefix): + try: + return Thematic.objects.get(code=thematic_code) + except Thematic.DoesNotExist: + pass + # Fallback: check AssetCategory via ContentType + from django.contrib.contenttypes.models import ContentType + from assets.models import AssetCategory + ct = ContentType.objects.filter(model=class_name).first() + if ct: + cat = AssetCategory.objects.filter(allowed_models=ct).first() + if cat and cat.thematic: + return cat.thematic + return None + + # Try from category + try: + if hasattr(asset, 'category') and asset.category and hasattr(asset.category, 'thematic'): + return asset.category.thematic + except Exception: + pass + + # Try direct thematic attribute + try: + if hasattr(asset, 'thematic') and asset.thematic and hasattr(asset.thematic, 'code'): + return asset.thematic + except Exception: + pass + + # Try from parent building category + try: + if hasattr(asset, 'building') and asset.building and hasattr(asset.building, 'category') and asset.building.category: + return asset.building.category.thematic + except Exception: + pass + # Try from model's category + try: + if hasattr(asset, 'model') and asset.model and hasattr(asset.model, 'category') and asset.model.category: + return asset.model.category.thematic + except Exception: + pass + + # Fallback: deduce thematic from asset class name + class_name = asset.__class__.__name__.lower() for prefix, thematic_code in thematic_mappings.items(): if class_name.startswith(prefix): try: diff --git a/loko/assets/templatetags/asset_tags.py b/loko/assets/templatetags/asset_tags.py index b1b065f..d597706 100644 --- a/loko/assets/templatetags/asset_tags.py +++ b/loko/assets/templatetags/asset_tags.py @@ -590,20 +590,26 @@ def asset_inspections(context, asset): from django.contrib.contenttypes.models import ContentType from inspections.models import Inspection + from inspections.permissions import filter_viewable_inspections_for_user from assets.permissions import can_inspect_asset from inspections.registry import get_inspection_partial_template - asset_ct = ContentType.objects.get_for_model(asset) - inspections = list(Inspection.objects.filter( - asset_content_type=asset_ct, - asset_object_id=asset.pk - ).select_related('inspector').prefetch_related('documents').order_by('-inspection_date')) - request = context.get('request') user = request.user if request else None - can_insp = False + asset_ct = ContentType.objects.get_for_model(asset) + base_qs = Inspection.objects.filter( + asset_content_type=asset_ct, + asset_object_id=asset.pk + ).select_related('inspector').prefetch_related('documents').order_by('-inspection_date') + + if user and user.is_authenticated: + inspections = list(filter_viewable_inspections_for_user(user, base_qs)) + else: + inspections = [] + + can_insp = False if user and user.is_authenticated: can_insp = can_inspect_asset(user, asset) diff --git a/loko/assets/views/parking.py b/loko/assets/views/parking.py index 784983b..4672184 100644 --- a/loko/assets/views/parking.py +++ b/loko/assets/views/parking.py @@ -344,9 +344,13 @@ def parking_assets_detail(request, asset_model, asset_id): content_type = ContentType.objects.get_for_model(ParkingSpot) from inspections.models import Inspection - inspections = Inspection.objects.filter( - asset_content_type=content_type, - asset_object_id=spot.id + from inspections.permissions import filter_viewable_inspections_for_user + inspections = filter_viewable_inspections_for_user( + request.user, + Inspection.objects.filter( + asset_content_type=content_type, + asset_object_id=spot.id + ) ).order_by('-inspection_date') from interventions.models import Intervention diff --git a/loko/inspections/apps.py b/loko/inspections/apps.py index a6c31ec..8b07c6f 100644 --- a/loko/inspections/apps.py +++ b/loko/inspections/apps.py @@ -12,9 +12,15 @@ class InspectionsConfig(AppConfig): from common.private_files.registry import register def guard_inspection_document(obj, user): - if not user.is_authenticated: + if not user or not user.is_authenticated: return False - return True + if getattr(user, 'is_superuser', False) or getattr(user, 'is_staff', False): + return True + user_config = getattr(user, 'config', None) + if user_config and (user_config.has_role('admin') or user_config.has_role('top_manager') or user_config.is_intern): + return True + from inspections.permissions import can_view_inspection + return can_view_inspection(user, getattr(obj, 'inspection', None)) register('inspections', 'InspectionDocument', 'file', guard_inspection_document) register('assets', 'InspectionDocument', 'file', guard_inspection_document) diff --git a/loko/inspections/permissions.py b/loko/inspections/permissions.py index 08bc773..d46ac6c 100644 --- a/loko/inspections/permissions.py +++ b/loko/inspections/permissions.py @@ -1,4 +1,23 @@ +from functools import reduce +from operator import or_ +from collections import defaultdict from django.conf import settings +from django.db.models import Q +from django.contrib.contenttypes.models import ContentType + + +def is_user_internal_or_admin(user): + """ + Détermine si un utilisateur est un utilisateur interne ou un administrateur ayant un accès global. + """ + if not user or not user.is_authenticated: + return False + if getattr(user, 'is_superuser', False) or getattr(user, 'is_staff', False): + return True + user_config = getattr(user, 'config', None) + if user_config and (user_config.has_role('admin') or user_config.has_role('top_manager') or user_config.is_intern): + return True + return False def can_edit_inspection(user, inspection): @@ -13,7 +32,7 @@ def can_edit_inspection(user, inspection): if getattr(user, 'is_superuser', False) or getattr(user, 'is_staff', False): return True - # Administrateur au sens configuration de rôle StreetUp + # Administrateur au sens configuration de rôle user_config = getattr(user, 'config', None) if user_config and hasattr(user_config, 'has_role'): if user_config.has_role('admin') or user_config.has_role('top_manager'): @@ -24,3 +43,262 @@ def can_edit_inspection(user, inspection): return True return False + + +def can_view_inspection(user, inspection): + """ + Vérifie si un utilisateur a le droit de visualiser une inspection. + - Administrateurs et utilisateurs internes (is_intern=True) : accès complet. + - Utilisateurs externes : seules les inspections qui les concernent sont visibles : + 1. L'utilisateur est l'auteur/inspecteur de l'inspection. + 2. L'inspection est liée à une mission d'intervention visible par l'utilisateur. + 3. L'inspection est liée à une intervention contrôlée visible par l'utilisateur. + 4. L'inspection est liée à un asset ou une intervention cible visible par l'utilisateur. + """ + if not user or not user.is_authenticated: + return False + + if is_user_internal_or_admin(user): + return True + + if not inspection: + return False + + # 1. Auteur / Inspecteur de l'inspection + if inspection.inspector_id == user.id: + return True + + # 2. Mission d'intervention + if inspection.mission_intervention_id: + from interventions.permissions import can_view_intervention + if can_view_intervention(user, inspection.mission_intervention): + return True + + # 3. Intervention de travaux contrôlée + if inspection.checked_intervention_id: + from interventions.permissions import can_view_intervention + if can_view_intervention(user, inspection.checked_intervention): + return True + + # 4. Cible (Asset ou Intervention) + if inspection.asset_content_type_id and inspection.asset_object_id: + if inspection.asset_content_type.app_label == 'interventions': + from interventions.permissions import can_view_intervention + target = inspection.asset + if target and can_view_intervention(user, target): + return True + else: + from assets.permissions import can_view_asset + target = inspection.asset + if target and can_view_asset(user, target): + return True + + return False + + +def get_contract_filtered_asset_qs(ModelClass, accessible_contract_ids): + """ + Retourne un QuerySet filtré des instances de ModelClass liées aux contrats autorisés, + ou None si ModelClass n'est pas un modèle gérant des contrats. + """ + class_name = ModelClass.__name__.lower() + + # Structures + if class_name == 'structure': + if not accessible_contract_ids: + return ModelClass.objects.none() + return ModelClass.objects.filter(contracts__status='active', contracts__contract_id__in=accessible_contract_ids) + elif class_name == 'structurelocation': + if not accessible_contract_ids: + return ModelClass.objects.none() + return ModelClass.objects.filter(structure__contracts__status='active', structure__contracts__contract_id__in=accessible_contract_ids) + elif class_name == 'structuregeoasset': + if not accessible_contract_ids: + return ModelClass.objects.none() + return ModelClass.objects.filter(location__structure__contracts__status='active', location__structure__contracts__contract_id__in=accessible_contract_ids) + + # ITS + elif class_name == 'itslocation': + if not accessible_contract_ids: + return ModelClass.objects.none() + return ModelClass.objects.filter(its_locations_contracts__status='active', its_locations_contracts__contract_id__in=accessible_contract_ids) + elif class_name == 'itsgeoasset': + if not accessible_contract_ids: + return ModelClass.objects.none() + return ModelClass.objects.filter(its_assets_contracts__status='active', its_assets_contracts__contract_id__in=accessible_contract_ids) + + # Control Centers + elif class_name == 'controlcenter': + if not accessible_contract_ids: + return ModelClass.objects.none() + return ModelClass.objects.filter(controlcenters_contracts__status='active', controlcenters_contracts__contract_id__in=accessible_contract_ids) + elif class_name == 'controlcenterlocation': + if not accessible_contract_ids: + return ModelClass.objects.none() + return ModelClass.objects.filter(control_center__controlcenters_contracts__status='active', control_center__controlcenters_contracts__contract_id__in=accessible_contract_ids) + elif class_name == 'controlcentergeoasset': + if not accessible_contract_ids: + return ModelClass.objects.none() + return ModelClass.objects.filter(controlcenters_assets_contracts__status='active', controlcenters_assets_contracts__contract_id__in=accessible_contract_ids) + + # Traffic Lights + elif class_name == 'trafficlightintersection': + if not accessible_contract_ids: + return ModelClass.objects.none() + return ModelClass.objects.filter(trafficlight_contracts__status='active', trafficlight_contracts__contract_id__in=accessible_contract_ids) + elif class_name in ('trafficlightpole', 'trafficlightlantern', 'trafficlightcable', 'trafficlightdetector', + 'trafficlightcontroller', 'trafficlightelectricalcabinet', 'trafficlightradar', + 'trafficlightaccessory'): + if not accessible_contract_ids: + return ModelClass.objects.none() + return ModelClass.objects.filter(intersection__trafficlight_contracts__status='active', intersection__trafficlight_contracts__contract_id__in=accessible_contract_ids) + elif class_name == 'trafficlightradarpole': + if not accessible_contract_ids: + return ModelClass.objects.none() + return ModelClass.objects.filter(pole__intersection__trafficlight_contracts__status='active', pole__intersection__trafficlight_contracts__contract_id__in=accessible_contract_ids) + + # Clean + elif class_name == 'cleanlocation': + if not accessible_contract_ids: + return ModelClass.objects.none() + return ModelClass.objects.filter(clean_location_contracts__contract_id__in=accessible_contract_ids) + elif class_name in ('cleanlitterbin', 'cleanglasscontainer'): + if not accessible_contract_ids: + return ModelClass.objects.none() + return ModelClass.objects.filter(location__clean_location_contracts__contract_id__in=accessible_contract_ids) + + # Artworks + elif class_name in ('artwork', 'monument', 'fountain'): + if not accessible_contract_ids: + return ModelClass.objects.none() + return ModelClass.objects.filter(maintenance_contract_id__in=accessible_contract_ids) + + # Modèles sans notion de contrat (ex: ParkingSpot, NatureRWIZ, Road, PublicLighting, Sign, etc.) + return None + + +def filter_viewable_inspections_for_user(user, qs=None): + """ + Filtre un QuerySet d'inspections pour ne renvoyer que celles visibles par l'utilisateur. + Pour les internes et administrateurs : toutes les inspections (qs complet). + Pour les externes : seules les inspections qui les concernent : + - Inspections créées par l'utilisateur (inspecteur) + - Inspections liées à une mission d'intervention ou intervention contrôlée visible + - Inspections dont l'asset ou intervention cible est visible + """ + from inspections.models import Inspection + + if qs is None: + qs = Inspection.objects.all() + + if not user or not user.is_authenticated: + return qs.none() + + if is_user_internal_or_admin(user): + return qs + + user_config = getattr(user, 'config', None) + conditions = [] + + # 1. Auteur / Inspecteur de l'inspection + conditions.append(Q(inspector=user)) + + # 2. Interventions liées visibles (missions d'intervention, contrôles post-travaux, ou cible intervention) + from interventions.permissions import filter_viewable_interventions_for_user + viewable_interventions = filter_viewable_interventions_for_user(user) + + conditions.append(Q(mission_intervention__in=viewable_interventions)) + conditions.append(Q(checked_intervention__in=viewable_interventions)) + + interv_ct = ContentType.objects.filter(app_label='interventions', model='intervention').first() + if interv_ct: + conditions.append(Q(asset_content_type=interv_ct, asset_object_id__in=viewable_interventions.values('pk'))) + + # 3. Assets cibles visibles + if user_config: + from common.models import UserThematics, UserContractAccess, UserAssetAccess, UserAssetTypeAccess, AssetTypePermissionConfig + from assets.permissions import _get_asset_thematic + + viewable_thematic_ids = set( + UserThematics.objects.filter(user_config=user_config, can_view_assets=True).values_list('thematic_id', flat=True) + ) + must_limit = not user_config.is_intern and user_config.limit_assets_to_contracts + accessible_contract_ids = list( + UserContractAccess.objects.filter(user_config=user_config, can_view_assets=True).values_list('contract_id', flat=True) + ) if must_limit else [] + + explicit_denies = defaultdict(set) + for row in UserAssetAccess.objects.filter(user_config=user_config, can_view=False).values_list('content_type_id', 'object_id'): + explicit_denies[row[0]].add(row[1]) + + explicit_allows = defaultdict(set) + for row in UserAssetAccess.objects.filter(user_config=user_config, can_view=True).values_list('content_type_id', 'object_id'): + explicit_allows[row[0]].add(row[1]) + + # ContentTypes présents dans les inspections ciblées + ct_ids = set(qs.values_list('asset_content_type_id', flat=True).distinct()) + + for ct_id in ct_ids: + if not ct_id: + continue + try: + ct = ContentType.objects.get(pk=ct_id) + except ContentType.DoesNotExist: + continue + + if ct.app_label == 'interventions': + continue + + ModelClass = ct.model_class() + if not ModelClass: + continue + + # Vérifier si AssetTypePermissionConfig exige des permissions explicites + requires_explicit = AssetTypePermissionConfig.objects.filter(content_type=ct, requires_explicit_permissions=True).exists() + type_access = UserAssetTypeAccess.objects.filter(user_config=user_config, content_type=ct).first() + + if type_access and not type_access.can_view: + allowed_ids = explicit_allows.get(ct.id, set()) + if allowed_ids: + conditions.append(Q(asset_content_type=ct, asset_object_id__in=allowed_ids)) + continue + + if requires_explicit: + allowed_ids = explicit_allows.get(ct.id, set()) + if allowed_ids: + conditions.append(Q(asset_content_type=ct, asset_object_id__in=allowed_ids)) + continue + + # Vérifier l'accès thématique + th = _get_asset_thematic(ModelClass) + if th and th.id not in viewable_thematic_ids: + allowed_ids = explicit_allows.get(ct.id, set()) + if allowed_ids: + conditions.append(Q(asset_content_type=ct, asset_object_id__in=allowed_ids)) + continue + + # Thématique autorisée (ou sans thématique assignée) + if must_limit: + filtered_asset_qs = get_contract_filtered_asset_qs(ModelClass, accessible_contract_ids) + if filtered_asset_qs is not None: + ct_q = Q(asset_content_type=ct, asset_object_id__in=filtered_asset_qs.values('pk')) + else: + ct_q = Q(asset_content_type=ct) + else: + ct_q = Q(asset_content_type=ct) + + denied_ids = explicit_denies.get(ct.id, set()) + if denied_ids: + ct_q &= ~Q(asset_object_id__in=denied_ids) + + allowed_ids = explicit_allows.get(ct.id, set()) + if allowed_ids: + ct_q |= Q(asset_content_type=ct, asset_object_id__in=allowed_ids) + + conditions.append(ct_q) + + if not conditions: + return qs.none() + + q = reduce(or_, conditions) + return qs.filter(q).distinct() diff --git a/loko/inspections/tests.py b/loko/inspections/tests.py index e8ab2a9..3fccdc2 100644 --- a/loko/inspections/tests.py +++ b/loko/inspections/tests.py @@ -496,3 +496,341 @@ class InspectionModelAndViewsTestCase(TestCase): self.assertIsNone(self.spot.last_inspection_date) +class ExternalUserInspectionVisibilityTestCase(TestCase): + def setUp(self): + self.client = Client() + + # Utilisateur interne + self.internal_user = User.objects.create_user( + username='intern_user', + password='password123', + first_name='Intern', + last_name='User' + ) + from common.models import UserConfig, Role, UserThematics, UserContractAccess + from contracts.models import Contract + from assets.models import Structure, StructureContract, AssetCategory + + config_intern = UserConfig.objects.create( + user=self.internal_user, + is_intern=True, + limit_assets_to_contracts=False + ) + + # Utilisateur externe + self.external_user = User.objects.create_user( + username='extern_user', + password='password123', + first_name='Extern', + last_name='User' + ) + self.external_config = UserConfig.objects.create( + user=self.external_user, + is_intern=False, + limit_assets_to_contracts=True + ) + role_ext, _ = Role.objects.get_or_create(name='external_manager') + self.external_config.roles.add(role_ext) + + # Thématiques + self.thematic_parking, _ = Thematic.objects.get_or_create( + code='parking', + defaults={'name_fr': 'Stationnement'} + ) + self.thematic_water, _ = Thematic.objects.get_or_create( + code='water', + defaults={'name_fr': 'Gestion de l\'Eau'} + ) + self.thematic_structures, _ = Thematic.objects.get_or_create( + code='structures', + defaults={'name_fr': 'Ouvrages d\'art'} + ) + + # L'externe a accès aux assets Parking et Structures, mais PAS à Water + UserThematics.objects.create( + user_config=self.external_config, + thematic=self.thematic_parking, + can_view_assets=True, + can_view_interventions=True + ) + UserThematics.objects.create( + user_config=self.external_config, + thematic=self.thematic_structures, + can_view_assets=True, + can_view_interventions=True + ) + UserThematics.objects.create( + user_config=self.external_config, + thematic=self.thematic_water, + can_view_assets=False, + can_view_interventions=False + ) + + # Contrats + from contracts.models import Company, Contract + self.company = Company.objects.create(name="Company Ext Test") + today = timezone.now().date() + self.contract_a = Contract.objects.create( + company=self.company, + contract_number="CTR-EXT-A", + start_date=today, + end_date=today + timezone.timedelta(days=365), + is_active=True + ) + self.contract_a.thematics.add(self.thematic_structures) + + self.contract_b = Contract.objects.create( + company=self.company, + contract_number="CTR-EXT-B", + start_date=today, + end_date=today + timezone.timedelta(days=365), + is_active=True + ) + self.contract_b.thematics.add(self.thematic_structures) + + # L'externe a accès au contrat A mais pas au B + uca_a = UserContractAccess.objects.create( + user_config=self.external_config, + contract=self.contract_a, + can_view_assets=True, + can_view_interventions=True + ) + from common.models import UserContractStatusPermission + UserContractStatusPermission.objects.create( + user_contract=uca_a, + status='in_progress', + can_view=True + ) + + # Assets + self.spot = ParkingSpot.objects.create(name_fr="Emplacement Ext") + self.spot_ct = ContentType.objects.get_for_model(ParkingSpot) + + self.rwiz = NatureRWIZ.objects.create(name_fr="Bassin Water") + self.rwiz_ct = ContentType.objects.get_for_model(NatureRWIZ) + + self.struct_a = Structure.objects.create(code="STR-A", name_fr="Pont A") + StructureContract.objects.create(structure=self.struct_a, contract=self.contract_a, status='active') + self.struct_ct = ContentType.objects.get_for_model(Structure) + + self.struct_b = Structure.objects.create(code="STR-B", name_fr="Pont B") + StructureContract.objects.create(structure=self.struct_b, contract=self.contract_b, status='active') + + # Interventions + self.interv_a = Intervention.objects.create( + title="Intervention Contrat A", + contract=self.contract_a, + thematic=self.thematic_structures, + status="in_progress" + ) + self.interv_b = Intervention.objects.create( + title="Intervention Contrat B", + contract=self.contract_b, + thematic=self.thematic_structures, + status="in_progress" + ) + + def test_external_user_thematic_filtering(self): + from inspections.permissions import can_view_inspection, filter_viewable_inspections_for_user + + insp_parking = Inspection.objects.create( + asset_content_type=self.spot_ct, + asset_object_id=self.spot.id, + inspector=self.internal_user, + result_status='compliant' + ) + insp_water = Inspection.objects.create( + asset_content_type=self.rwiz_ct, + asset_object_id=self.rwiz.id, + inspector=self.internal_user, + result_status='compliant' + ) + + # L'externe peut voir le parking (can_view_assets=True sur parking) + self.assertTrue(can_view_inspection(self.external_user, insp_parking)) + # L'externe ne peut pas voir water (can_view_assets=False sur water) + self.assertFalse(can_view_inspection(self.external_user, insp_water)) + + # QuerySet filtering + qs = filter_viewable_inspections_for_user(self.external_user) + self.assertIn(insp_parking, qs) + self.assertNotIn(insp_water, qs) + + def test_external_user_contract_filtering(self): + from inspections.permissions import can_view_inspection, filter_viewable_inspections_for_user + + insp_struct_a = Inspection.objects.create( + asset_content_type=self.struct_ct, + asset_object_id=self.struct_a.id, + inspector=self.internal_user, + result_status='compliant' + ) + insp_struct_b = Inspection.objects.create( + asset_content_type=self.struct_ct, + asset_object_id=self.struct_b.id, + inspector=self.internal_user, + result_status='compliant' + ) + + # L'externe a accès au contrat A -> voit insp_struct_a + self.assertTrue(can_view_inspection(self.external_user, insp_struct_a)) + # L'externe n'a pas accès au contrat B -> ne voit pas insp_struct_b + self.assertFalse(can_view_inspection(self.external_user, insp_struct_b)) + + # QuerySet filtering + qs = filter_viewable_inspections_for_user(self.external_user) + self.assertIn(insp_struct_a, qs) + self.assertNotIn(insp_struct_b, qs) + + def test_external_user_intervention_linked_inspection(self): + from inspections.permissions import can_view_inspection, filter_viewable_inspections_for_user + + # Inspection sur asset d'une thématique non autorisée, mais liée à une intervention autorisée (mission) + insp_mission = Inspection.objects.create( + asset_content_type=self.rwiz_ct, + asset_object_id=self.rwiz.id, + inspector=self.internal_user, + mission_intervention=self.interv_a, + result_status='compliant' + ) + # Inspection liée à une intervention non autorisée + insp_mission_unauth = Inspection.objects.create( + asset_content_type=self.rwiz_ct, + asset_object_id=self.rwiz.id, + inspector=self.internal_user, + mission_intervention=self.interv_b, + result_status='compliant' + ) + + self.assertTrue(can_view_inspection(self.external_user, insp_mission)) + self.assertFalse(can_view_inspection(self.external_user, insp_mission_unauth)) + + qs = filter_viewable_inspections_for_user(self.external_user) + self.assertIn(insp_mission, qs) + self.assertNotIn(insp_mission_unauth, qs) + + def test_external_user_is_inspector_always_visible(self): + from inspections.permissions import can_view_inspection, filter_viewable_inspections_for_user + + # Inspection réalisée par l'externe lui-même sur une thématique normalement restreinte + insp_by_extern = Inspection.objects.create( + asset_content_type=self.rwiz_ct, + asset_object_id=self.rwiz.id, + inspector=self.external_user, + result_status='compliant' + ) + + self.assertTrue(can_view_inspection(self.external_user, insp_by_extern)) + qs = filter_viewable_inspections_for_user(self.external_user) + self.assertIn(insp_by_extern, qs) + + def test_external_user_inspections_list_view(self): + insp_parking = Inspection.objects.create( + asset_content_type=self.spot_ct, + asset_object_id=self.spot.id, + inspector=self.internal_user, + result_status='compliant' + ) + insp_water = Inspection.objects.create( + asset_content_type=self.rwiz_ct, + asset_object_id=self.rwiz.id, + inspector=self.internal_user, + result_status='compliant' + ) + + self.client.login(username='extern_user', password='password123') + url = reverse('inspections:inspections_list') + response = self.client.get(url, {'tab': 'all'}) + self.assertEqual(response.status_code, 200) + + inspections = list(response.context['inspections']) + self.assertIn(insp_parking, inspections) + self.assertNotIn(insp_water, inspections) + + def test_external_user_inspection_detail_api(self): + insp_parking = Inspection.objects.create( + asset_content_type=self.spot_ct, + asset_object_id=self.spot.id, + inspector=self.internal_user, + result_status='compliant' + ) + insp_water = Inspection.objects.create( + asset_content_type=self.rwiz_ct, + asset_object_id=self.rwiz.id, + inspector=self.internal_user, + result_status='compliant' + ) + + self.client.login(username='extern_user', password='password123') + + # Inspection autorisée -> 200 + url_ok = reverse('inspections:get_inspection_detail_api', kwargs={'inspection_id': insp_parking.id}) + res_ok = self.client.get(url_ok) + self.assertEqual(res_ok.status_code, 200) + self.assertTrue(res_ok.json()['success']) + + # Inspection non autorisée -> 403 + url_forbidden = reverse('inspections:get_inspection_detail_api', kwargs={'inspection_id': insp_water.id}) + res_forbidden = self.client.get(url_forbidden) + self.assertEqual(res_forbidden.status_code, 403) + self.assertFalse(res_forbidden.json()['success']) + + def test_guard_inspection_document(self): + from inspections.models import InspectionDocument + from django.core.files.uploadedfile import SimpleUploadedFile + from inspections.apps import InspectionsConfig + + insp_parking = Inspection.objects.create( + asset_content_type=self.spot_ct, + asset_object_id=self.spot.id, + inspector=self.internal_user, + result_status='compliant' + ) + insp_water = Inspection.objects.create( + asset_content_type=self.rwiz_ct, + asset_object_id=self.rwiz.id, + inspector=self.internal_user, + result_status='compliant' + ) + + dummy_file = SimpleUploadedFile("doc.txt", b"content", content_type="text/plain") + doc_parking = InspectionDocument.objects.create(inspection=insp_parking, file=dummy_file) + doc_water = InspectionDocument.objects.create(inspection=insp_water, file=dummy_file) + + from inspections.permissions import can_view_inspection + self.assertTrue(can_view_inspection(self.external_user, doc_parking.inspection)) + self.assertFalse(can_view_inspection(self.external_user, doc_water.inspection)) + + def test_internal_user_sees_all_inspections(self): + from inspections.permissions import can_view_inspection, filter_viewable_inspections_for_user + + insp_parking = Inspection.objects.create( + asset_content_type=self.spot_ct, + asset_object_id=self.spot.id, + inspector=self.external_user, + result_status='compliant' + ) + insp_water = Inspection.objects.create( + asset_content_type=self.rwiz_ct, + asset_object_id=self.rwiz.id, + inspector=self.external_user, + result_status='compliant' + ) + insp_struct_b = Inspection.objects.create( + asset_content_type=self.struct_ct, + asset_object_id=self.struct_b.id, + inspector=self.external_user, + result_status='compliant' + ) + + self.assertTrue(can_view_inspection(self.internal_user, insp_parking)) + self.assertTrue(can_view_inspection(self.internal_user, insp_water)) + self.assertTrue(can_view_inspection(self.internal_user, insp_struct_b)) + + qs = filter_viewable_inspections_for_user(self.internal_user) + self.assertIn(insp_parking, qs) + self.assertIn(insp_water, qs) + self.assertIn(insp_struct_b, qs) + + + diff --git a/loko/inspections/views.py b/loko/inspections/views.py index cb21875..c31059d 100644 --- a/loko/inspections/views.py +++ b/loko/inspections/views.py @@ -49,7 +49,9 @@ def inspections_list(request): checked_interv_id = request.GET.get('checked_interv_id', '') search_q = request.GET.get('q', '').strip() - qs = Inspection.objects.all().select_related( + from .permissions import filter_viewable_inspections_for_user + qs = filter_viewable_inspections_for_user(request.user) + qs = qs.select_related( 'asset_content_type', 'inspector', 'mission_intervention', 'checked_intervention' ).prefetch_related('documents').order_by('-inspection_date') @@ -433,10 +435,11 @@ def get_inspection_detail_api(request, inspection_id): inspection_obj = get_object_or_404(Inspection, pk=inspection_id) leaf = inspection_obj.detailed_inspection - if not leaf.can_edit(request.user): + from .permissions import can_view_inspection + if not can_view_inspection(request.user, leaf): return JsonResponse({ 'success': False, - 'error': str(_("Permission refusée. Seuls l'auteur et les administrateurs peuvent modifier cette inspection.")) + 'error': str(_("Permission refusée. Vous n'avez pas accès à cette inspection.")) }, status=403) docs_data = [] @@ -665,7 +668,9 @@ def inspections_geojson(request): if period == 'disabled': return JsonResponse({'type': 'FeatureCollection', 'features': []}) - qs = Inspection.objects.all().select_related('asset_content_type', 'inspector', 'mission_intervention').prefetch_related('documents') + from .permissions import filter_viewable_inspections_for_user, get_contract_filtered_asset_qs + qs = filter_viewable_inspections_for_user(request.user) + qs = qs.select_related('asset_content_type', 'inspector', 'mission_intervention').prefetch_related('documents') if mission_id: qs = qs.filter(mission_intervention_id=mission_id) @@ -710,6 +715,19 @@ def inspections_geojson(request): features = [] + user_config = getattr(request.user, 'config', None) + is_external = (user_config is None) or (not user_config.is_intern) + must_limit = is_external and (not user_config or user_config.limit_assets_to_contracts) + accessible_contract_ids = [] + if must_limit and user_config: + from common.models import UserContractAccess + accessible_contract_ids = list( + UserContractAccess.objects.filter( + user_config=user_config, + can_view_assets=True + ).values_list('contract_id', flat=True) + ) + for ct_id in ct_ids: if not ct_id: continue @@ -720,6 +738,11 @@ def inspections_geojson(request): continue asset_qs = ModelClass.objects.filter(geom__isnull=False) + if must_limit: + contract_filtered_qs = get_contract_filtered_asset_qs(ModelClass, accessible_contract_ids) + if contract_filtered_qs is not None: + asset_qs = asset_qs.filter(pk__in=contract_filtered_qs.values('pk')) + if bbox_geom: bbox_geom_3812 = bbox_geom.transform(3812, clone=True) asset_qs = asset_qs.filter(geom__intersects=bbox_geom_3812)