From 33ef94902b36521c48851cc7549126292bccb6b3 Mon Sep 17 00:00:00 2001 From: kdeterme Date: Fri, 28 Aug 2026 10:57:41 +0200 Subject: [PATCH] refactor: replace ContentType lookup with robust filtering and validation across asset and intervention views --- loko/assets/permissions.py | 16 ++++-- loko/assets/tests.py | 71 ++++++++++++++++++++++++++ loko/assets/views/asset_editing.py | 8 ++- loko/assets/views/clean.py | 9 ++-- loko/assets/views/controlcenters.py | 10 ++-- loko/assets/views/export.py | 14 +++-- loko/assets/views/its.py | 10 ++-- loko/assets/views/nature.py | 13 +++-- loko/assets/views/publiclighting.py | 12 ++--- loko/assets/views/roads.py | 12 ++--- loko/assets/views/trafficlights.py | 12 ++--- loko/assets/views/water.py | 9 ++-- loko/documents/permissions.py | 7 +-- loko/interventions/views/detail.py | 15 +++--- loko/interventions/views/general.py | 6 ++- loko/interventions/views/helpers.py | 6 ++- loko/interventions/views/preventive.py | 12 +++-- 17 files changed, 175 insertions(+), 67 deletions(-) diff --git a/loko/assets/permissions.py b/loko/assets/permissions.py index 1decb21..942adea 100644 --- a/loko/assets/permissions.py +++ b/loko/assets/permissions.py @@ -688,10 +688,20 @@ def asset_edit_permission_required(view_func): return HttpResponseForbidden(_("Missing asset information")) try: - content_type = ContentType.objects.get(model=asset_model.lower()) - model_class = content_type.model_class() + cts = ContentType.objects.filter(model=asset_model.lower()) + if not cts.exists(): + return HttpResponseForbidden(_("Invalid asset type")) + cts_sorted = sorted(cts, key=lambda ct: (0 if ct.app_label == 'assets' else 1)) + model_class = None + for ct in cts_sorted: + mc = ct.model_class() + if mc is not None: + model_class = mc + break + if model_class is None: + return HttpResponseForbidden(_("Invalid asset type")) asset = get_object_or_404(model_class, pk=asset_id) - except ContentType.DoesNotExist: + except Exception: return HttpResponseForbidden(_("Invalid asset type")) if not can_edit_asset(request.user, asset): diff --git a/loko/assets/tests.py b/loko/assets/tests.py index da1c3d8..c144ae9 100644 --- a/loko/assets/tests.py +++ b/loko/assets/tests.py @@ -4029,6 +4029,77 @@ class ReplaceAssetStructureGeoAssetTest(TestCase): self.assertEqual(new_asset.name_fr, "Asset Original") +class ExportAssetsTest(TestCase): + def setUp(self): + from django.contrib.auth import get_user_model + from common.models import Thematic, UserConfig, UserThematics + from assets.models import AssetCategory, Structure, StructureLocation, StructureAssetModel, StructureGeoAsset + + self.user = get_user_model().objects.create_user( + username="test_export_user", + email="export_test@example.com", + password="testpassword", + is_staff=True, + ) + self.user_config = UserConfig.objects.create(user=self.user) + self.thematic = Thematic.objects.get_or_create(code="structures", defaults={"name_fr": "Structures"})[0] + UserThematics.objects.create( + user_config=self.user_config, + thematic=self.thematic, + can_view_assets=True, + can_edit_assets=True, + ) + + self.struct = Structure.objects.create(code="EXP01", name_fr="Pont Export") + self.loc = StructureLocation.objects.create(code="EXP01A", name_fr="Loc Export", structure=self.struct) + self.category = AssetCategory.objects.create(thematic=self.thematic, code="CAT_EXP", name_fr="Cat Export") + self.model = StructureAssetModel.objects.create(code="MOD_EXP", name_fr="Mod Export", category=self.category) + self.asset = StructureGeoAsset.objects.create( + code="EXP01A-000001", + name_fr="Asset Pour Export", + structure=self.struct, + location=self.loc, + model=self.model, + category=self.category, + status="active" + ) + + def test_export_assets_csv_success(self): + from django.urls import reverse + + self.client.force_login(self.user) + url = reverse('assets:export_assets') + ids_json = json.dumps([f"structuregeoasset:{self.asset.id}"]) + + response = self.client.post(url, {'ids': ids_json, 'format': 'csv'}) + self.assertEqual(response.status_code, 200) + self.assertEqual(response['Content-Type'], 'text/csv; charset=utf-8') + content = response.content.decode('utf-8') + self.assertIn("EXP01A-000001", content) + self.assertIn("Asset Pour Export", content) + + def test_export_assets_with_duplicate_content_types(self): + from django.urls import reverse + from django.contrib.contenttypes.models import ContentType + + # Create a second duplicate ContentType with another app_label to simulate MultipleObjectsReturned + ContentType.objects.create( + app_label='other_app', + model='structuregeoasset' + ) + + self.client.force_login(self.user) + url = reverse('assets:export_assets') + ids_json = json.dumps([f"structuregeoasset:{self.asset.id}"]) + + # This should not raise MultipleObjectsReturned and succeed + response = self.client.post(url, {'ids': ids_json, 'format': 'csv'}) + self.assertEqual(response.status_code, 200) + content = response.content.decode('utf-8') + self.assertIn("EXP01A-000001", content) + + + diff --git a/loko/assets/views/asset_editing.py b/loko/assets/views/asset_editing.py index 45d0061..e0c280e 100644 --- a/loko/assets/views/asset_editing.py +++ b/loko/assets/views/asset_editing.py @@ -725,8 +725,12 @@ def _generate_archive_asset_code(old_asset): if location and hasattr(location, 'get_direct_child_assets'): existing_codes = set() assets_dict = location.get_direct_child_assets() - for key, queryset in assets_dict.items(): - for asset in queryset: + if isinstance(assets_dict, dict): + for key, queryset in assets_dict.items(): + for asset in queryset: + existing_codes.add(asset.code) + elif hasattr(assets_dict, '__iter__'): + for asset in assets_dict: existing_codes.add(asset.code) if new_code not in existing_codes: diff --git a/loko/assets/views/clean.py b/loko/assets/views/clean.py index 796cd27..027b7f4 100644 --- a/loko/assets/views/clean.py +++ b/loko/assets/views/clean.py @@ -468,10 +468,11 @@ def clean_assets_geojson(request): @check_thematic_access('clean') def clean_assets_detail(request, asset_model, asset_id): - try: - content_type = ContentType.objects.get(model=asset_model) - model_class = content_type.model_class() - except ContentType.DoesNotExist: + content_type = ContentType.objects.filter(app_label='assets', model=asset_model.lower()).first() or ContentType.objects.filter(model=asset_model.lower()).first() + if not content_type: + raise Http404(_("Unknown model: %(asset_model)s") % {'asset_model': asset_model}) + model_class = content_type.model_class() + if not model_class: raise Http404(_("Unknown model: %(asset_model)s") % {'asset_model': asset_model}) asset = get_object_or_404(model_class, id=asset_id) diff --git a/loko/assets/views/controlcenters.py b/loko/assets/views/controlcenters.py index a422564..5117de1 100644 --- a/loko/assets/views/controlcenters.py +++ b/loko/assets/views/controlcenters.py @@ -520,11 +520,11 @@ def controlcenters_locations_detail(request, location_id): @check_thematic_access('controlcenters') def controlcenters_assets_detail(request, asset_model, asset_id): - try: - # Récupérer le ContentType pour le modèle fourni - content_type = ContentType.objects.get(model=asset_model) - model_class = content_type.model_class() - except ContentType.DoesNotExist: + content_type = ContentType.objects.filter(app_label='assets', model=asset_model.lower()).first() or ContentType.objects.filter(model=asset_model.lower()).first() + if not content_type: + raise Http404(_("Modèle '%(model)s' inconnu.") % {"model": asset_model}) + model_class = content_type.model_class() + if not model_class: raise Http404(_("Modèle '%(model)s' inconnu.") % {"model": asset_model}) # Récupérer l’objet asset diff --git a/loko/assets/views/export.py b/loko/assets/views/export.py index 2523415..6a04a41 100644 --- a/loko/assets/views/export.py +++ b/loko/assets/views/export.py @@ -152,14 +152,22 @@ def export_assets(request): items: list[tuple[str, object]] = [] for model_name, asset_ids in groups.items(): try: - ct = ContentType.objects.get(model=model_name) - model_class = ct.model_class() + cts = ContentType.objects.filter(model=model_name.lower()) + if not cts.exists(): + continue + cts_sorted = sorted(cts, key=lambda ct: (0 if ct.app_label == 'assets' else 1)) + model_class = None + for ct in cts_sorted: + mc = ct.model_class() + if mc is not None: + model_class = mc + break if model_class is None: continue for asset in model_class.objects.filter(pk__in=asset_ids): if can_view_asset(request.user, asset): items.append((model_name, asset)) - except ContentType.DoesNotExist: + except Exception: continue timestamp = localtime(now()).strftime('%Y%m%d_%H%M%S') diff --git a/loko/assets/views/its.py b/loko/assets/views/its.py index 94cd5a3..c0385bf 100644 --- a/loko/assets/views/its.py +++ b/loko/assets/views/its.py @@ -523,11 +523,11 @@ def its_locations_detail(request, location_id): @check_thematic_access('its') def its_assets_detail(request, asset_model, asset_id): - try: - # Récupérer le ContentType pour le modèle fourni - content_type = ContentType.objects.get(model=asset_model) - model_class = content_type.model_class() - except ContentType.DoesNotExist: + content_type = ContentType.objects.filter(app_label='assets', model=asset_model.lower()).first() or ContentType.objects.filter(model=asset_model.lower()).first() + if not content_type: + raise Http404(f"Modèle '{asset_model}' inconnu.") + model_class = content_type.model_class() + if not model_class: raise Http404(f"Modèle '{asset_model}' inconnu.") # Récupérer l’objet asset diff --git a/loko/assets/views/nature.py b/loko/assets/views/nature.py index f13a73b..7ab5425 100644 --- a/loko/assets/views/nature.py +++ b/loko/assets/views/nature.py @@ -215,13 +215,12 @@ def nature_assets(request): @check_thematic_access('nature') def nature_assets_detail(request, asset_model, asset_id): - - try: - # Récupérer le ContentType pour le modèle fourni - content_type = ContentType.objects.get(model=asset_model) - model_class = content_type.model_class() - except ContentType.DoesNotExist: - raise Http404(_("Modèle %(asset_model)s inconnu.") % (asset_model)) + content_type = ContentType.objects.filter(app_label='assets', model=asset_model.lower()).first() or ContentType.objects.filter(model=asset_model.lower()).first() + if not content_type: + raise Http404(_("Modèle %(asset_model)s inconnu.") % {'asset_model': asset_model}) + model_class = content_type.model_class() + if not model_class: + raise Http404(_("Modèle %(asset_model)s inconnu.") % {'asset_model': asset_model}) # Récupérer l’objet asset asset = get_object_or_404(model_class, id=asset_id) diff --git a/loko/assets/views/publiclighting.py b/loko/assets/views/publiclighting.py index 2e7ba9b..66dcbf0 100644 --- a/loko/assets/views/publiclighting.py +++ b/loko/assets/views/publiclighting.py @@ -257,12 +257,12 @@ def publiclighting_assets(request): @check_thematic_access('publiclighting') def publiclighting_assets_detail(request, asset_model, asset_id): - try: - # Récupérer le ContentType pour le modèle fourni (ex: 'trafficlightpole') - content_type = ContentType.objects.get(model=asset_model) - model_class = content_type.model_class() - except ContentType.DoesNotExist: - raise Http404(_("Modèle %(asset_model)s inconnu.") % (asset_model)) + content_type = ContentType.objects.filter(app_label='assets', model=asset_model.lower()).first() or ContentType.objects.filter(model=asset_model.lower()).first() + if not content_type: + raise Http404(_("Modèle %(asset_model)s inconnu.") % {'asset_model': asset_model}) + model_class = content_type.model_class() + if not model_class: + raise Http404(_("Modèle %(asset_model)s inconnu.") % {'asset_model': asset_model}) # Récupérer l’objet asset asset = get_object_or_404(model_class, id=asset_id) diff --git a/loko/assets/views/roads.py b/loko/assets/views/roads.py index 0f440a0..0a9aed1 100644 --- a/loko/assets/views/roads.py +++ b/loko/assets/views/roads.py @@ -520,12 +520,12 @@ def roads_streets_detail(request, street_id): @check_thematic_access('roads') def roads_assets_detail(request, asset_model, asset_id): - try: - # Récupérer le ContentType pour le modèle fourni - content_type = ContentType.objects.get(model=asset_model) - model_class = content_type.model_class() - except ContentType.DoesNotExist: - raise Http404(_("Modèle %(asset_model)s inconnu.") % (asset_model)) + content_type = ContentType.objects.filter(app_label='assets', model=asset_model.lower()).first() or ContentType.objects.filter(model=asset_model.lower()).first() + if not content_type: + raise Http404(_("Modèle %(asset_model)s inconnu.") % {'asset_model': asset_model}) + model_class = content_type.model_class() + if not model_class: + raise Http404(_("Modèle %(asset_model)s inconnu.") % {'asset_model': asset_model}) # Récupérer l’objet asset asset = get_object_or_404(model_class, id=asset_id) diff --git a/loko/assets/views/trafficlights.py b/loko/assets/views/trafficlights.py index ba8f386..b270d07 100644 --- a/loko/assets/views/trafficlights.py +++ b/loko/assets/views/trafficlights.py @@ -1354,12 +1354,12 @@ def trafficlights_assets_geojson(request): @check_thematic_access('trafficlights') def trafficlights_assets_detail(request, asset_model, asset_id): - try: - # Récupérer le ContentType pour le modèle fourni (ex: 'trafficlightpole') - content_type = ContentType.objects.get(model=asset_model) - model_class = content_type.model_class() - except ContentType.DoesNotExist: - raise Http404(_("Modèle %(asset_model)s inconnu.") % (asset_model)) + content_type = ContentType.objects.filter(app_label='assets', model=asset_model.lower()).first() or ContentType.objects.filter(model=asset_model.lower()).first() + if not content_type: + raise Http404(_("Modèle %(asset_model)s inconnu.") % {'asset_model': asset_model}) + model_class = content_type.model_class() + if not model_class: + raise Http404(_("Modèle %(asset_model)s inconnu.") % {'asset_model': asset_model}) # Récupérer l’objet asset asset = get_object_or_404(model_class, id=asset_id) diff --git a/loko/assets/views/water.py b/loko/assets/views/water.py index 60fb797..0d5c7e9 100644 --- a/loko/assets/views/water.py +++ b/loko/assets/views/water.py @@ -527,10 +527,11 @@ def zirw_detail(request, location_id): @check_thematic_access('water') def water_assets_detail(request, asset_model, asset_id): - try: - content_type = ContentType.objects.get(model=asset_model) - model_class = content_type.model_class() - except ContentType.DoesNotExist: + content_type = ContentType.objects.filter(app_label='assets', model=asset_model.lower()).first() or ContentType.objects.filter(model=asset_model.lower()).first() + if not content_type: + raise Http404(_("Modèle %(asset_model)s inconnu.") % {'asset_model': asset_model}) + model_class = content_type.model_class() + if not model_class: raise Http404(_("Modèle %(asset_model)s inconnu.") % {'asset_model': asset_model}) asset = get_object_or_404(model_class, id=asset_id) diff --git a/loko/documents/permissions.py b/loko/documents/permissions.py index 1664ba0..1f6357e 100644 --- a/loko/documents/permissions.py +++ b/loko/documents/permissions.py @@ -518,9 +518,10 @@ def _get_content_type_ids_for_user(user, *, edit_only: bool) -> list[int]: model_names = _THEMATIC_MODEL_MAPPING.get(thematic_code, []) for model_name in model_names: try: - ct = ContentType.objects.get(model=model_name) - result.append(ct.id) - except ContentType.DoesNotExist: + ct = ContentType.objects.filter(app_label='assets', model=model_name.lower()).first() or ContentType.objects.filter(model=model_name.lower()).first() + if ct: + result.append(ct.id) + except Exception: continue if not hasattr(user, '_folder_cache'): diff --git a/loko/interventions/views/detail.py b/loko/interventions/views/detail.py index a4d0e66..7e262a7 100644 --- a/loko/interventions/views/detail.py +++ b/loko/interventions/views/detail.py @@ -1849,10 +1849,9 @@ def interventions_add(request, thematic_code): if equip_ct and equip_id: asset_pairs = [(int(equip_ct), int(equip_id))] elif equip_type and equip_id: - # ATTENTION: .get(model=...) suppose unicité du model name à travers les apps - # Préfère envoyer equip_ct depuis le front si possible. - ct = ContentType.objects.get(model=equip_type.lower()) - asset_pairs = [(ct.id, int(equip_id))] + ct = ContentType.objects.filter(app_label='assets', model=equip_type.lower()).first() or ContentType.objects.filter(model=equip_type.lower()).first() + if ct: + asset_pairs = [(ct.id, int(equip_id))] first_asset_obj = None for ct_id, obj_id in asset_pairs: @@ -3446,8 +3445,8 @@ def check_asset_templates_for_intervention(request, intervention_id): }, status=400) # Récupérer l'asset - ct = ContentType.objects.get(model=asset_type) - model_class = ct.model_class() + ct = ContentType.objects.filter(app_label='assets', model=asset_type.lower()).first() or ContentType.objects.filter(model=asset_type.lower()).first() + model_class = ct.model_class() if ct else None if not model_class: return JsonResponse({ 'success': False, @@ -3700,8 +3699,8 @@ def create_operations_from_selected_template(request, intervention_id): }, status=400) # Récupérer l'asset - ct = ContentType.objects.get(model=asset_type) - model_class = ct.model_class() + ct = ContentType.objects.filter(app_label='assets', model=asset_type.lower()).first() or ContentType.objects.filter(model=asset_type.lower()).first() + model_class = ct.model_class() if ct else None if not model_class: return JsonResponse({ 'success': False, diff --git a/loko/interventions/views/general.py b/loko/interventions/views/general.py index bf2b8a1..d95be85 100644 --- a/loko/interventions/views/general.py +++ b/loko/interventions/views/general.py @@ -1282,8 +1282,12 @@ def ajax_get_replacement_models_by_category(request): try: # Get the asset ContentType - ct = ContentType.objects.get(model=asset_model) + ct = ContentType.objects.filter(app_label='assets', model=asset_model.lower()).first() or ContentType.objects.filter(model=asset_model.lower()).first() + if not ct: + return JsonResponse({'error': f"Unknown model: {asset_model}"}, status=400) model_class = ct.model_class() + if not model_class: + return JsonResponse({'error': f"Unknown model: {asset_model}"}, status=400) # Get the asset instance asset_obj = model_class.objects.get(pk=asset_id) diff --git a/loko/interventions/views/helpers.py b/loko/interventions/views/helpers.py index 991ac6b..3e4253a 100644 --- a/loko/interventions/views/helpers.py +++ b/loko/interventions/views/helpers.py @@ -3163,8 +3163,12 @@ def batch_fetch_assets_with_locations(selected_assets_array): models_map = {} for model_name, ids in by_model.items(): - ct = ContentType.objects.get(model=model_name) + ct = ContentType.objects.filter(app_label='assets', model=model_name.lower()).first() or ContentType.objects.filter(model=model_name.lower()).first() + if not ct: + continue Model = ct.model_class() + if not Model: + continue # détecter quelles FKs de localisation existent réellement sur le modèle related_fields = [] diff --git a/loko/interventions/views/preventive.py b/loko/interventions/views/preventive.py index 4825a22..28ed66a 100644 --- a/loko/interventions/views/preventive.py +++ b/loko/interventions/views/preventive.py @@ -555,7 +555,9 @@ def intervention_add_preventive(request, thematic_code): asset_id = a['id'] # ContentType de l'asset - ct_asset = ContentType.objects.get(model=model_name) + ct_asset = ContentType.objects.filter(app_label='assets', model=model_name.lower()).first() or ContentType.objects.filter(model=model_name.lower()).first() + if not ct_asset: + continue # Objet asset préchargé (avec ses FKs) asset_obj = assets_map.get(model_name, {}).get(asset_id) @@ -671,7 +673,9 @@ def intervention_add_preventive(request, thematic_code): for a in selected_assets_array: model_name = a['asset_type'] asset_id = a['id'] - ct_asset = ContentType.objects.get(model=model_name) + ct_asset = ContentType.objects.filter(app_label='assets', model=model_name.lower()).first() or ContentType.objects.filter(model=model_name.lower()).first() + if not ct_asset: + continue InterventionAsset.objects.get_or_create( intervention=intervention, @@ -783,7 +787,9 @@ def intervention_add_preventive(request, thematic_code): for a in selected_assets_array: model_name = a['asset_type'] asset_id = a['id'] - ct_asset = ContentType.objects.get(model=model_name) + ct_asset = ContentType.objects.filter(app_label='assets', model=model_name.lower()).first() or ContentType.objects.filter(model=model_name.lower()).first() + if not ct_asset: + continue asset_obj = assets_map.get(model_name, {}).get(asset_id) # Filtrer les templates applicables à cet asset