from django.test import TestCase, tag from django.contrib.auth.models import User from django.contrib.gis.geos import MultiPolygon, Polygon from django.urls import reverse import json import os from unittest import skipUnless RUN_DXF_TESTS = os.getenv('RUN_DXF_TESTS', '0') == '1' from common.models import Thematic, UserConfig, UserThematics from assets.models import NatureLocation, TrafficLightIntersection, TrafficLightRadarModel, TrafficLightRadar, AssetCategory from assets.models.buildings import PublicBuilding, BuildingAsset, BuildingEquipment class NatureLocationsGeoJSONTest(TestCase): def setUp(self): # Create a thematic self.thematic = Thematic.objects.create( code="nature", name_fr="Nature", name_nl="Natuur" ) # Create a user self.user = User.objects.create_user(username="testuser", password="password123") self.config = UserConfig.objects.create(user=self.user) # Give thematic access to user UserThematics.objects.create( user_config=self.config, thematic=self.thematic, can_view_assets=True ) # Create a NatureLocation in Brussels (around lat/lon: 50.85, 4.35) # In SRID 3812, Brussels coordinates are around x=648000, y=670000. # Let's create a small multipolygon around it. poly = Polygon(((648000, 670000), (648100, 670000), (648100, 670100), (648000, 670100), (648000, 670000)), srid=3812) multipoly = MultiPolygon(poly, srid=3812) # Build the WGS84 geojson geom_4326 = multipoly.clone() geom_4326.transform(4326) self.location = NatureLocation.objects.create( code="LOC-001", name_fr="Test Location", geom=multipoly, geojson=geom_4326.json, is_highway=False ) def test_nature_locations_geojson_with_bbox_intersection(self): self.client.force_login(self.user) # Coordinates in degrees (WGS84, SRID 4326): # x=148000, y=170000 in Lambert 2008 (SRID 3812) is roughly lon=4.3508, lat=50.8465 # Let's verify that a bounding box around it intersects and returns the location # A polygon in SRID 4326: bbox = { "type": "Feature", "geometry": { "type": "Polygon", "coordinates": [[ [4.34, 50.84], [4.36, 50.84], [4.36, 50.86], [4.34, 50.86], [4.34, 50.84] ]] } } response = self.client.post( reverse("assets:nature_locations_geojson"), data=json.dumps({"filter_geom": bbox}), content_type="application/json" ) self.assertEqual(response.status_code, 200) data = response.json() self.assertEqual(data["type"], "FeatureCollection") # Ensure our location is in the results features = data["features"] self.assertTrue(any(f["properties"]["code"] == "LOC-001" for f in features)) from django.test import override_settings @override_settings(STORAGES={ "default": { "BACKEND": "django.core.files.storage.FileSystemStorage", }, "staticfiles": { "BACKEND": "django.contrib.staticfiles.storage.StaticFilesStorage", }, }) class TrafficLightRadarTest(TestCase): def setUp(self): # Create a thematic for traffic lights self.thematic, _ = Thematic.objects.get_or_create( code="trafficlights", defaults={ "name_fr": "Feux de signalisation", "name_nl": "Verkeerslichten" } ) self.user = User.objects.create_user(username="testuser", password="password123") self.config = UserConfig.objects.create(user=self.user, limit_assets_to_contracts=False) # Give thematic access to user UserThematics.objects.create( user_config=self.config, thematic=self.thematic, can_view_assets=True ) # Create category for radar self.category, _ = AssetCategory.objects.get_or_create( code="TL_RADAR_PREVENTIF", defaults={ "name_fr": "Radar Préventif", "name_nl": "Preventieve Radar", "thematic": self.thematic } ) # Create a spatial point for the intersection from django.contrib.gis.geos import Point geom_intersection = Point(648000, 670000, srid=3812) # Create intersection with is_radar=True self.intersection = TrafficLightIntersection.objects.create( code="ROM001", name_fr="Test Radar Intersection", geom=geom_intersection, lon=4.3508, lat=50.8465, is_radar=True ) # Create a model self.radar_model = TrafficLightRadarModel.objects.create( code="RAD-MODEL-01", name_fr="Model 1", category=self.category ) # Create a radar pole from assets.models.trafficlights import TrafficLightRadarPole self.radar_pole = TrafficLightRadarPole.objects.create( code="RP_ROM001", intersection=self.intersection, geom=geom_intersection, status="active" ) # Create the radar asset self.radar = TrafficLightRadar.objects.create( code="RAD_PREV_ROM001", category=self.category, model=self.radar_model, intersection=self.intersection, pole=self.radar_pole, geom=geom_intersection, status="active" ) def test_intersection_is_radar(self): # Verify fields self.assertTrue(self.intersection.is_radar) self.assertEqual(self.radar.intersection, self.intersection) self.assertEqual(self.intersection.radars.count(), 1) def test_geojson_contains_is_radar(self): self.client.force_login(self.user) response = self.client.get(reverse("assets:trafficlights_intersections_geojson")) self.assertEqual(response.status_code, 200) data = response.json() # Verify intersection features = data["features"] radar_features = [f for f in features if f["properties"]["code"] == "ROM001"] self.assertEqual(len(radar_features), 1) self.assertTrue(radar_features[0]["properties"]["is_radar"]) def test_detail_view_includes_radars(self): self.client.force_login(self.user) response = self.client.get(reverse("assets:trafficlights_intersections_detail", args=[self.intersection.id])) self.assertEqual(response.status_code, 200) # Check context self.assertIn("assets_geojson", response.context) assets_geojson = response.context["assets_geojson"] self.assertEqual(assets_geojson["type"], "FeatureCollection") radar_assets = [f for f in assets_geojson["features"] if f["properties"]["asset_type"] == "trafficlightradar"] self.assertEqual(len(radar_assets), 1) self.assertEqual(radar_assets[0]["properties"]["code"], "RAD_PREV_ROM001") def test_detail_view_includes_radar_poles(self): self.client.force_login(self.user) response = self.client.get(reverse("assets:trafficlights_intersections_detail", args=[self.intersection.id])) self.assertEqual(response.status_code, 200) # Check assets list in context self.assertIn("assets", response.context) assets = response.context["assets"] self.assertIn("poles", assets) # Verify the radar pole is in the poles list self.assertTrue(any(p.code == "RP_ROM001" for p in assets["poles"])) # Check GeoJSON has the correct asset_type and details self.assertIn("assets_geojson", response.context) assets_geojson = response.context["assets_geojson"] radar_pole_assets = [f for f in assets_geojson["features"] if f["properties"]["asset_type"] == "trafficlightradarpole"] self.assertEqual(len(radar_pole_assets), 1) self.assertEqual(radar_pole_assets[0]["properties"]["code"], "RP_ROM001") def test_detail_view_includes_georeferenced_documents(self): from django.contrib.contenttypes.models import ContentType from documents.models import ManagedDocument, DocumentAttachment, DocumentFolder, DocumentFolderAttachment # Ensure we have a default folder for the intersection to enable managed documents folder = DocumentFolder.objects.create(name="Intersection Folder", slug="intersection-folder", created_by=self.user) ct_tli = ContentType.objects.get_for_model(self.intersection.__class__) DocumentFolderAttachment.objects.create( folder=folder, content_type=ct_tli, object_id=self.intersection.id, is_default=True ) # Create a georeferenced document and attach it doc = ManagedDocument.objects.create( title="Plan Georef", georeference_coords=[[4.1, 50.1], [4.2, 50.1], [4.2, 50.2], [4.1, 50.2]], georeference_page=1, created_by=self.user ) doc.folders.add(folder) DocumentAttachment.objects.create( document=doc, content_type=ct_tli, object_id=self.intersection.id ) self.client.force_login(self.user) response = self.client.get(reverse("assets:trafficlights_intersections_detail", args=[self.intersection.id])) self.assertEqual(response.status_code, 200) self.assertIn("georeferenced_documents_json", response.context) import json georef_docs = json.loads(response.context["georeferenced_documents_json"]) self.assertEqual(len(georef_docs), 1) self.assertEqual(georef_docs[0]["title"], "Plan Georef") self.assertEqual(georef_docs[0]["coords"], [[4.1, 50.1], [4.2, 50.1], [4.2, 50.2], [4.1, 50.2]]) def test_unlink_parent_from_child(self): # 1. Create a parent PublicLightingPole from assets.models.publiclighting import PublicLightingStreet, PublicLightingPoleModel, PublicLightingPole pl_street = PublicLightingStreet.objects.create(code="PL-ST-01") pl_model = PublicLightingPoleModel.objects.create(code="PL-MOD-01") pl_pole = PublicLightingPole.objects.create(code="PL-POLE-01", model=pl_model, street=pl_street) # 2. Link parent to our radar child self.radar.publiclighting_pole = pl_pole self.radar.save() # Check it is linked self.radar.refresh_from_db() self.assertEqual(self.radar.publiclighting_pole, pl_pole) # 3. Give edit permission to user for trafficlights thematic UserThematics.objects.filter(user_config=self.config, thematic=self.thematic).update(can_edit_assets=True) # 4. Log in the user self.client.force_login(self.user) # 5. Call set-parent endpoint to unlink (parent_id=None/null) response = self.client.post( reverse("assets:set_parent_asset", kwargs={"asset_model": "trafficlightradar", "asset_id": self.radar.id}), data=json.dumps({ "field_name": "publiclighting_pole", "parent_model": "publiclightingpole", "parent_id": None }), content_type="application/json" ) self.assertEqual(response.status_code, 200) data = response.json() self.assertTrue(data["success"]) self.assertEqual(data["field_name"], "publiclighting_pole") # 6. Verify database state is updated (unlinked) self.radar.refresh_from_db() self.assertIsNone(self.radar.publiclighting_pole) def test_radar_position_sync_when_support_pole_moves(self): # Move the radar pole from django.contrib.gis.geos import Point new_geom = Point(648100, 670100, srid=3812) new_geom_4326 = new_geom.transform(4326, clone=True) self.radar_pole.geom = new_geom self.radar_pole.save() # Check if the child radar position updated automatically self.radar.refresh_from_db() self.assertEqual(self.radar.geom, new_geom) self.assertAlmostEqual(self.radar.lon, new_geom_4326.x, places=5) self.assertAlmostEqual(self.radar.lat, new_geom_4326.y, places=5) self.assertEqual(self.radar.geojson, new_geom_4326.geojson) class CategoryAllowedModelTest(TestCase): def setUp(self): # Create user & config self.user = User.objects.create_user(username="testuser_allowed", password="password123") self.config = UserConfig.objects.create(user=self.user, limit_assets_to_contracts=False) # Create thematic self.thematic, _ = Thematic.objects.get_or_create( code="trafficlights", defaults={"name_fr": "Feux", "name_nl": "Verkeerslichten"} ) # Give access UserThematics.objects.create( user_config=self.config, thematic=self.thematic, can_view_assets=True, can_edit_assets=True ) # Get/create category self.category, _ = AssetCategory.objects.get_or_create( code="TL_RADAR_PREVENTIF", defaults={ "name_fr": "Radar Préventif", "name_nl": "Preventieve Radar", "thematic": self.thematic } ) # Associate content types to category from django.contrib.contenttypes.models import ContentType from assets.models.trafficlights import TrafficLightRadar, TrafficLightAccessory self.ct_radar = ContentType.objects.get_for_model(TrafficLightRadar) self.ct_accessory = ContentType.objects.get_for_model(TrafficLightAccessory) self.category.allowed_models.add(self.ct_radar) self.category.allowed_models.add(self.ct_accessory) def test_get_categories_allowed_models(self): self.client.force_login(self.user) response = self.client.get(reverse("assets:get_asset_categories", args=[self.thematic.code])) self.assertEqual(response.status_code, 200) data = response.json() # Verify the radar category returns allowed models categories = data.get("categories", []) radar_cat = next((c for c in categories if c["id"] == self.category.id), None) self.assertIsNotNone(radar_cat) allowed_models = radar_cat.get("allowed_models", []) self.assertEqual(len(allowed_models), 2) radar_model = next(m for m in allowed_models if m["model_name"] == "trafficlightradar") accessory_model = next(m for m in allowed_models if m["model_name"] == "trafficlightaccessory") # TrafficLightRadar requires a model template self.assertTrue(radar_model["is_model_required"]) # TrafficLightAccessory does not require a model template self.assertFalse(accessory_model["is_model_required"]) from assets.views.trafficlights import parse_dxf_poles @tag('slow', 'dxf') @skipUnless(RUN_DXF_TESTS, "Skip slow DXF tests by default") class TrafficLightDXFImportTest(TestCase): def setUp(self): # Create a thematic for traffic lights self.thematic, _ = Thematic.objects.get_or_create( code="trafficlights", defaults={ "name_fr": "Feux de signalisation", "name_nl": "Verkeerslichten" } ) self.user = User.objects.create_user(username="testuser", password="password123") self.config = UserConfig.objects.create(user=self.user, limit_assets_to_contracts=False) # Give thematic access to user UserThematics.objects.create( user_config=self.config, thematic=self.thematic, can_view_assets=True, can_edit_assets=True ) # Create a spatial point for the intersection from django.contrib.gis.geos import Point geom_intersection = Point(648000, 670000, srid=3812) # Create intersection with code 'SB0239' matching the DXF file name self.intersection = TrafficLightIntersection.objects.create( code="SB0239", name_fr="Carrefour SB0239", geom=geom_intersection, lon=4.3508, lat=50.8465, status="active" ) # Create category for TrafficLightPole self.category, _ = AssetCategory.objects.get_or_create( code="TL_POLE", defaults={ "name_fr": "Poteau de signalisation", "name_nl": "Paal", "thematic": self.thematic } ) # Create a pole model from assets.models import TrafficLightPoleModel self.pole_model = TrafficLightPoleModel.objects.create( code="POM-01", name_fr="Poteau Standard", category=self.category ) # Create the existing active pole asset that Choice A / Choice B should target from assets.models import TrafficLightPole self.existing_pole = TrafficLightPole.objects.create( code="SB0239_T3", # matches label T3 in DXF category=self.category, model=self.pole_model, intersection=self.intersection, geom=geom_intersection, lon=4.3508, lat=50.8465, status="active" ) def test_parse_dxf_poles_real_file(self): # Verify parse_dxf_poles works on the actual DXF import os from django.conf import settings external_files_dir = os.path.join(os.path.dirname(settings.BASE_DIR), 'external_files') file_path = os.path.join(external_files_dir, 'SB0239_2026-02-01.dxf') if os.path.exists(file_path): poles = parse_dxf_poles(file_path) self.assertTrue(len(poles) > 0) # Ensure coordinates and labels are populated for pole in poles: self.assertIn('x_31370', pole) self.assertIn('y_31370', pole) self.assertIn('lon', pole) self.assertIn('lat', pole) self.assertIn('label', pole) def test_import_dxf_views_flow(self): import os from django.conf import settings external_files_dir = os.path.join(os.path.dirname(settings.BASE_DIR), 'external_files') file_path = os.path.join(external_files_dir, 'SB0239_2026-02-01.dxf') if not os.path.exists(file_path): self.skipTest("Real DXF file for test not found") self.client.force_login(self.user) # 1. GET: Select page url_import = reverse("assets:import_dxf", args=[self.intersection.id]) response = self.client.get(url_import) self.assertEqual(response.status_code, 200) self.assertContains(response, "SB0239_2026-02-01.dxf") # 2. POST: Select file redirects to GET page with selected dxf_file response = self.client.post(url_import, {"dxf_file": "SB0239_2026-02-01.dxf"}) self.assertEqual(response.status_code, 302) self.assertRedirects(response, f"{url_import}?dxf_file=SB0239_2026-02-01.dxf") # 3. POST: Add staged poles via AJAX url_add = reverse("assets:add_staged_pole", args=[self.intersection.id]) # Add pole 1 (replaces existing by code) response = self.client.post(url_add, { "code": "SB0239_T3", "model_id": str(self.pole_model.id), "lon": "4.35081", "lat": "50.84651", "dxf_file": "SB0239_2026-02-01.dxf" }) self.assertEqual(response.status_code, 200) data = response.json() self.assertTrue(data["success"]) pole_1_id = data["pole"]["id"] # Add pole 2 (new pole) response = self.client.post(url_add, { "code": "SB0239_P02", "model_id": str(self.pole_model.id), "lon": "4.35085", "lat": "50.84655", "dxf_file": "SB0239_2026-02-01.dxf" }) self.assertEqual(response.status_code, 200) data = response.json() self.assertTrue(data["success"]) pole_2_id = data["pole"]["id"] # Verify staged assets are created in 'to_be_approved' and 'to_be_validated' from assets.models import TrafficLightPole staged_poles = TrafficLightPole.objects.filter(intersection=self.intersection, status="to_be_approved") self.assertEqual(staged_poles.count(), 2) self.assertTrue(staged_poles.filter(code="SB0239_T3").exists()) self.assertTrue(staged_poles.filter(code="SB0239_P02").exists()) for sp in staged_poles: self.assertEqual(sp.validation_status, "to_be_validated") self.assertEqual(sp.note, "Créé manuellement sur plan DXF : SB0239_2026-02-01.dxf") # Test deleting pole 2 via AJAX (verifying delete view) url_delete = reverse("assets:delete_staged_pole", args=[self.intersection.id, pole_2_id]) response = self.client.post(url_delete) self.assertEqual(response.status_code, 200) self.assertTrue(response.json()["success"]) self.assertEqual(TrafficLightPole.objects.filter(intersection=self.intersection, status="to_be_approved").count(), 1) # Re-add pole 2 response = self.client.post(url_add, { "code": "SB0239_P02", "model_id": str(self.pole_model.id), "lon": "4.35085", "lat": "50.84655", "dxf_file": "SB0239_2026-02-01.dxf" }) self.assertEqual(response.status_code, 200) # 4. POST: Integrate Choice A (replace by code) url_integrate = reverse("assets:integrate_dxf", args=[self.intersection.id]) response = self.client.post(url_integrate, {"choice": "choice_a"}) self.assertEqual(response.status_code, 302) # Staged assets should now be active/validated self.assertEqual(TrafficLightPole.objects.filter(intersection=self.intersection, status="to_be_approved").count(), 0) active_poles = TrafficLightPole.objects.filter(intersection=self.intersection, status="active") self.assertEqual(active_poles.count(), 2) self.assertTrue(active_poles.filter(code="SB0239_T3").exists()) self.assertTrue(active_poles.filter(code="SB0239_P02").exists()) # The old active pole should have been archived/removed and its code changed/uninstallation date set removed_pole = TrafficLightPole.objects.filter(intersection=self.intersection, status="removed").first() self.assertIsNotNone(removed_pole) self.assertIn("SB0239_T3", removed_pole.code) self.assertIsNotNone(removed_pole.uninstallation_date) def test_integrate_choice_b_full_replacement(self): self.client.force_login(self.user) # Stage two poles via AJAX url_add = reverse("assets:add_staged_pole", args=[self.intersection.id]) self.client.post(url_add, { "code": "SB0239_T3", "model_id": str(self.pole_model.id), "lon": "4.35081", "lat": "50.84651", "dxf_file": "SB0239_2026-02-01.dxf" }) self.client.post(url_add, { "code": "SB0239_P02", "model_id": str(self.pole_model.id), "lon": "4.35085", "lat": "50.84655", "dxf_file": "SB0239_2026-02-01.dxf" }) # POST: Integrate Choice B (replace all) url_integrate = reverse("assets:integrate_dxf", args=[self.intersection.id]) response = self.client.post(url_integrate, {"choice": "choice_b"}) self.assertEqual(response.status_code, 302) # Staged assets should now be active from assets.models import TrafficLightPole self.assertEqual(TrafficLightPole.objects.filter(intersection=self.intersection, status="to_be_approved").count(), 0) # All old active poles should have been removed removed_poles = TrafficLightPole.objects.filter(intersection=self.intersection, status="removed") self.assertEqual(removed_poles.count(), 1) self.assertEqual(removed_poles.first().id, self.existing_pole.id) # Staged poles should be active active_poles = TrafficLightPole.objects.filter(intersection=self.intersection, status="active") self.assertEqual(active_poles.count(), 2) def test_save_dxf_layers(self): import os from django.conf import settings external_files_dir = os.path.join(os.path.dirname(settings.BASE_DIR), 'external_files') file_path = os.path.join(external_files_dir, 'SB0239_2026-02-01.dxf') if not os.path.exists(file_path): self.skipTest("Real DXF file for test not found") self.client.force_login(self.user) # Trigger layout GeoJSON generation and caching url_geojson = reverse("assets:dxf_layout_geojson", args=[self.intersection.id]) self.client.get(f"{url_geojson}?file=SB0239_2026-02-01.dxf") # Save a subset of layers (e.g. only 'IGN_Signalisation_Feu_Poteau') url_save = reverse("assets:save_dxf_layers", args=[self.intersection.id]) response = self.client.post(url_save, { "dxf_file": "SB0239_2026-02-01.dxf", "layers": "IGN_Signalisation_Feu_Poteau" }) self.assertEqual(response.status_code, 200) self.assertTrue(response.json()["success"]) # Querying the dxf-layout-geojson endpoint should now return ONLY features from the saved layer url_geojson = reverse("assets:dxf_layout_geojson", args=[self.intersection.id]) response_geojson = self.client.get(f"{url_geojson}?file=SB0239_2026-02-01.dxf") self.assertEqual(response_geojson.status_code, 200) data = response_geojson.json() self.assertEqual(data["selected_layers"], ["IGN_Signalisation_Feu_Poteau"]) for feature in data["features"]: self.assertEqual(feature["properties"]["Layer"], "IGN_Signalisation_Feu_Poteau") # Verify that the overlay ManagedDocument was created and georeferenced from documents.models import ManagedDocument from django.contrib.contenttypes.models import ContentType from assets.models import TrafficLightIntersectionDocument if getattr(TrafficLightIntersectionDocument, "DOCUMENT_MANAGEMENT_SYNC", False): from documents.models import DocumentAttachment ct_tli = ContentType.objects.get_for_model(self.intersection) dxf_attachment = DocumentAttachment.objects.filter( content_type=ct_tli, object_id=self.intersection.id, document__versions__file__contains="SB0239_2026-02-01" ).first() self.assertIsNotNone(dxf_attachment) dxf_source_id = dxf_attachment.document.id else: from assets.models import TrafficLightIntersectionDocument doc_obj = TrafficLightIntersectionDocument.objects.filter( intersection=self.intersection, file__contains="SB0239_2026-02-01" ).first() self.assertIsNotNone(doc_obj) self.assertIsNotNone(doc_obj.managed_document) dxf_source_id = doc_obj.managed_document.id overlay_doc = ManagedDocument.objects.filter( unique_params__dxf_source_id=dxf_source_id, is_archived=False ).first() self.assertIsNotNone(overlay_doc) self.assertIsNotNone(overlay_doc.georeference_coords) # Verify that the generated overlay file has the correct unsuffixed name and exists on disk import os from django.conf import settings overlay_filename = "sb0239_2026-02-01.overlay.png" overlay_path = os.path.join(settings.MEDIA_ROOT, 'dxf_layouts', overlay_filename) self.assertTrue(os.path.exists(overlay_path)) def test_save_dxf_layers_managed_document(self): import os from django.conf import settings from django.core.files.base import ContentFile from documents.models import ManagedDocument, DocumentVersion, DocumentAttachment from django.contrib.contenttypes.models import ContentType external_files_dir = os.path.join(os.path.dirname(settings.BASE_DIR), 'external_files') dxf_path = os.path.join(external_files_dir, 'SB0239_2026-02-01.dxf') if not os.path.exists(dxf_path): self.skipTest("Real DXF file for test not found") self.client.force_login(self.user) # Create a ManagedDocument and version mimicking direct upload via Document Management managed_doc = ManagedDocument.objects.create( title="Direct Upload", created_by=self.user ) with open(dxf_path, 'rb') as f: dxf_content = ContentFile(f.read(), name="SB0239_2026-02-01_direct.dxf") version = DocumentVersion.objects.create( document=managed_doc, file=dxf_content, uploaded_by=self.user, status=DocumentVersion.STATUS_APPROVED ) # Attach it to the intersection ct = ContentType.objects.get_for_model(self.intersection) DocumentAttachment.objects.create( document=managed_doc, content_type=ct, object_id=self.intersection.id, attached_by=self.user ) # Trigger saving layers on this managed document using the filename of the version dxf_filename = os.path.basename(version.file.name) url_save = reverse("assets:save_dxf_layers", args=[self.intersection.id]) response = self.client.post(url_save, { "dxf_file": dxf_filename, "layers": "IGN_Signalisation_Feu_Poteau" }) self.assertEqual(response.status_code, 200) self.assertTrue(response.json()["success"]) def test_dxf_document_deletion(self): import os from django.conf import settings from documents.models import ManagedDocument, DocumentVersion from assets.models import TrafficLightIntersectionDocument, TrafficLightPole self.user.is_superuser = True self.user.save() self.client.force_login(self.user) # Create a mock DXF document doc = TrafficLightIntersectionDocument.objects.create( intersection=self.intersection, file="uploaded_test.dxf", document_type='other', uploaded_by=self.user ) # Create the ManagedDocument and current version associated m_doc = ManagedDocument.objects.create( title="uploaded_test.dxf", created_by=self.user ) DocumentVersion.objects.create( document=m_doc, file="uploaded_test.dxf", version_number=1, uploaded_by=self.user ) # Link them doc.managed_document = m_doc doc.save() # Create a mock overlay document overlay_doc = ManagedDocument.objects.create( title="uploaded_test.overlay.png", created_by=self.user, unique_params={"format": "overlay_png", "dxf_source_id": m_doc.id} ) # Create a mock imported pole with the dxf filename in note pole = TrafficLightPole.objects.create( intersection=self.intersection, model=self.pole_model, code="TEST_POLE_DEL", geom="POINT(0 0)", status="active", note="Importé depuis le plan DXF : uploaded_test.dxf", updated_by=self.user ) # Delete document via client DocumentDeleteView url_delete = reverse("documents:delete", args=[m_doc.pk]) # Confirm delete request with checkboxes checked response = self.client.post(url_delete, { "delete_overlay": "on", "delete_vectors": "on" }) self.assertEqual(response.status_code, 302) # The document, overlay document, legacy intersection doc, and imported pole should be deleted self.assertFalse(ManagedDocument.objects.filter(pk=m_doc.pk).exists()) self.assertFalse(ManagedDocument.objects.filter(pk=overlay_doc.pk).exists()) self.assertFalse(TrafficLightIntersectionDocument.objects.filter(pk=doc.pk).exists()) self.assertFalse(TrafficLightPole.objects.filter(pk=pole.pk).exists()) def test_import_dxf_via_upload(self): self.client.force_login(self.user) import os from django.conf import settings from django.core.files.uploadedfile import SimpleUploadedFile # Read the real DXF file read_only_dxf_dir = os.path.join(os.path.dirname(settings.BASE_DIR), 'external_files') real_dxf_path = os.path.join(read_only_dxf_dir, 'SB0239_2026-02-01.dxf') if not os.path.exists(real_dxf_path): self.skipTest("Real DXF file for test not found") with open(real_dxf_path, 'rb') as f: dxf_content = f.read() # Create SimpleUploadedFile uploaded_file = SimpleUploadedFile("uploaded_test.dxf", dxf_content, content_type="application/dxf") url_import = reverse("assets:import_dxf", args=[self.intersection.id]) response = self.client.post(url_import, {"dxf_file_upload": uploaded_file}) self.assertEqual(response.status_code, 302) # Extract temp filename from redirect URL redirect_url = response.url self.assertIn("?dxf_file=temp_", redirect_url) temp_filename = redirect_url.split("?dxf_file=")[1] # The file should exist in the dxf_temp directory temp_file_path = os.path.join(settings.PRIVATE_MEDIA_ROOT, 'dxf_temp', temp_filename) self.assertTrue(os.path.exists(temp_file_path)) # Now simulate Step 2: POST to save_dxf_layers url_save = reverse("assets:save_dxf_layers", args=[self.intersection.id]) response_save = self.client.post(url_save, { "dxf_file": temp_filename, "layers": "IGN_Signalisation_Feu_Poteau" }) self.assertEqual(response_save.status_code, 200) self.assertTrue(response_save.json()["success"]) # Verify the document was saved to the database and file exists from assets.models import TrafficLightIntersectionDocument if getattr(TrafficLightIntersectionDocument, "DOCUMENT_MANAGEMENT_SYNC", False): from django.contrib.contenttypes.models import ContentType from documents.models import ManagedDocument ct_tli = ContentType.objects.get_for_model(self.intersection) m_doc = ManagedDocument.objects.filter( attachments__content_type=ct_tli, attachments__object_id=self.intersection.id ).first() self.assertIsNotNone(m_doc) latest_v = m_doc.latest_version self.assertIsNotNone(latest_v) self.assertTrue(os.path.exists(latest_v.file.path)) dxf_filename = os.path.basename(latest_v.file.name) dxf_file_path = latest_v.file.path else: from assets.models import TrafficLightIntersectionDocument doc = TrafficLightIntersectionDocument.objects.filter(intersection=self.intersection).first() self.assertIsNotNone(doc) self.assertTrue(os.path.exists(doc.file.path)) dxf_filename = os.path.basename(doc.file.name) dxf_file_path = doc.file.path # Check overlay image and layout geojson existence overlay_name = dxf_filename.lower().replace('.dxf', '.overlay.png') overlay_path = os.path.join(settings.MEDIA_ROOT, 'dxf_layouts', overlay_name) self.assertTrue(os.path.exists(overlay_path)) geojson_name = dxf_filename.lower().replace('.dxf', '.layout.geojson') geojson_path = os.path.join(settings.PRIVATE_MEDIA_ROOT, 'dxf_layouts', geojson_name) self.assertTrue(os.path.exists(geojson_path)) # Cleanup saved file after test try: os.remove(dxf_file_path) except Exception: pass try: os.remove(temp_file_path) except Exception: pass @tag('slow', 'dxf') @skipUnless(RUN_DXF_TESTS, "Skip slow DXF tests by default") class TrafficLightCrossPlanImportTest(TestCase): def setUp(self): from assets.models import ( TrafficLightPoleModel, TrafficLightLanternModel, TrafficLightDetectorModel, TrafficLightCableModel ) # Create a thematic for traffic lights self.thematic, _ = Thematic.objects.get_or_create( code="trafficlights", defaults={ "name_fr": "Feux de signalisation", "name_nl": "Verkeerslichten" } ) # Create a user self.user = User.objects.create_user(username="testuser", password="password123") self.config = UserConfig.objects.create(user=self.user) # Give view/edit permissions to user UserThematics.objects.create( user_config=self.config, thematic=self.thematic, can_view_assets=True, can_edit_assets=True ) # Create intersection SWB01 in DB self.intersection = TrafficLightIntersection.objects.create( code="SWB01", status="active" ) # Setup basic models so fallback works TrafficLightPoleModel.objects.get_or_create( code="poteau_droit", defaults={"name_fr": "Poteau droit, gris 3.1m", "name_nl": "Poteau droit, gris 3.1m"} ) TrafficLightLanternModel.objects.get_or_create( code="3v200", defaults={"name_fr": "3V200", "name_nl": "3V200", "voltage": 230} ) TrafficLightDetectorModel.objects.get_or_create( code="det_model", defaults={"name_fr": "Drukknop", "name_nl": "Drukknop"} ) TrafficLightCableModel.objects.get_or_create( code="cab_model", defaults={"name_fr": "SVAVB 61x1,5mm²", "name_nl": "SVAVB 61x1,5mm²", "number_of_strands": 61} ) def test_import_cross_plan_get(self): self.client.force_login(self.user) url = reverse("assets:import_cross_plan", args=[self.intersection.id]) response = self.client.get(url) self.assertEqual(response.status_code, 200) self.assertTemplateUsed(response, "assets/trafficlights_import_cross_plan.html") self.assertEqual(response.context["step"], "select") def test_import_cross_plan_post_and_integrate(self): self.client.force_login(self.user) import os from django.conf import settings from django.core.files.uploadedfile import SimpleUploadedFile from assets.models.trafficlights import ( TrafficLightPole, TrafficLightCable, TrafficLightLantern, TrafficLightDetector ) # Path to real Excel file real_excel_path = os.path.join(os.path.dirname(settings.BASE_DIR), 'external_files', '18ac47df841-2808f-DOC-23270.xlsx') if not os.path.exists(real_excel_path): self.skipTest("Real Excel file for test not found") with open(real_excel_path, 'rb') as f: excel_content = f.read() uploaded_file = SimpleUploadedFile("test_cross_plan.xlsx", excel_content, content_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet") # 1. Test POST import view (Analysis & Comparison Preview) url_import = reverse("assets:import_cross_plan", args=[self.intersection.id]) response = self.client.post(url_import, {"cross_plan_file_upload": uploaded_file}) self.assertEqual(response.status_code, 200) self.assertEqual(response.context["step"], "preview") self.assertTrue(len(response.context["poles"]) > 0) self.assertTrue(len(response.context["lanterns"]) > 0) # 2. Test POST integrate view (Global replace all action) url_integrate = reverse("assets:integrate_cross_plan", args=[self.intersection.id]) response_integrate = self.client.post(url_integrate, { "cross_plan_filename": response.context["cross_plan_filename"], "global_action": "replace_all" }) self.assertEqual(response_integrate.status_code, 302) # Redirects back to detail page # Verify created assets in DB self.assertTrue(TrafficLightPole.objects.filter(intersection=self.intersection, status="active").count() > 0) self.assertTrue(TrafficLightCable.objects.filter(intersection=self.intersection, status="active").count() > 0) self.assertTrue(TrafficLightLantern.objects.filter(pole__intersection=self.intersection, status="active").count() > 0) self.assertTrue(TrafficLightDetector.objects.filter(intersection=self.intersection, status="active").count() > 0) def test_prefix_match_and_rename_lantern_unchecked(self): from assets.models import TrafficLightPole, TrafficLightLantern, TrafficLightPoleModel, TrafficLightLanternModel, AssetCategory pole_model = TrafficLightPoleModel.objects.first() lantern_model = TrafficLightLanternModel.objects.first() category_pole = AssetCategory.objects.filter(code="TL_POLE").first() category_lantern = AssetCategory.objects.filter(code="TL_LANTERN").first() or AssetCategory.objects.filter(code="TL_LANTERNE").first() # Pre-create pole in DB pole = TrafficLightPole.objects.create( intersection=self.intersection, code="SWB01_A01", model=pole_model, category=category_pole, status="active" ) # Pre-create lantern with old nomenclature (no phase suffix) old_lantern = TrafficLightLantern.objects.create( pole=pole, code="SWB01_A01_LAN01", model=lantern_model, category=category_lantern, status="active" ) self.client.force_login(self.user) import os from django.conf import settings from django.core.files.uploadedfile import SimpleUploadedFile real_excel_path = os.path.join(os.path.dirname(settings.BASE_DIR), 'external_files', '18ac47df841-2808f-DOC-23270.xlsx') if not os.path.exists(real_excel_path): self.skipTest("Real Excel file for test not found") with open(real_excel_path, 'rb') as f: excel_content = f.read() uploaded_file = SimpleUploadedFile("test_cross_plan.xlsx", excel_content, content_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet") # POST to import_cross_plan to analyze url_import = reverse("assets:import_cross_plan", args=[self.intersection.id]) response = self.client.post(url_import, {"cross_plan_file_upload": uploaded_file}) self.assertEqual(response.status_code, 200) # 1. Post integrate WITHOUT checking the selection checkbox url_integrate = reverse("assets:integrate_cross_plan", args=[self.intersection.id]) response_integrate = self.client.post(url_integrate, { "cross_plan_filename": response.context["cross_plan_filename"], "action_lantern_SWB01_A01_LAN01_T1": "replace", "db_id_lantern_SWB01_A01_LAN01_T1": str(old_lantern.id) # Not passing selected_lantern_SWB01_A01_LAN01_T1 }) self.assertEqual(response_integrate.status_code, 302) # Old lantern must still be active (skipped) old_lantern.refresh_from_db() self.assertEqual(old_lantern.status, "active") def test_prefix_match_and_rename_lantern_checked(self): from assets.models import TrafficLightPole, TrafficLightLantern, TrafficLightPoleModel, TrafficLightLanternModel, AssetCategory pole_model = TrafficLightPoleModel.objects.first() lantern_model = TrafficLightLanternModel.objects.first() category_pole = AssetCategory.objects.filter(code="TL_POLE").first() category_lantern = AssetCategory.objects.filter(code="TL_LANTERN").first() or AssetCategory.objects.filter(code="TL_LANTERNE").first() # Pre-create pole in DB pole = TrafficLightPole.objects.create( intersection=self.intersection, code="SWB01_A01", model=pole_model, category=category_pole, status="active" ) # Pre-create lantern with old nomenclature (no phase suffix) old_lantern = TrafficLightLantern.objects.create( pole=pole, code="SWB01_A01_LAN01", model=lantern_model, category=category_lantern, status="active" ) self.client.force_login(self.user) import os from django.conf import settings from django.core.files.uploadedfile import SimpleUploadedFile real_excel_path = os.path.join(os.path.dirname(settings.BASE_DIR), 'external_files', '18ac47df841-2808f-DOC-23270.xlsx') if not os.path.exists(real_excel_path): self.skipTest("Real Excel file for test not found") with open(real_excel_path, 'rb') as f: excel_content = f.read() uploaded_file = SimpleUploadedFile("test_cross_plan.xlsx", excel_content, content_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet") # POST to import_cross_plan to analyze url_import = reverse("assets:import_cross_plan", args=[self.intersection.id]) response = self.client.post(url_import, {"cross_plan_file_upload": uploaded_file}) self.assertEqual(response.status_code, 200) # 2. Post integrate WITH checking the selection checkbox url_integrate = reverse("assets:integrate_cross_plan", args=[self.intersection.id]) response_integrate = self.client.post(url_integrate, { "cross_plan_filename": response.context["cross_plan_filename"], "selected_lantern_SWB01_A01_LAN01_T1": "on", "action_lantern_SWB01_A01_LAN01_T1": "replace", "db_id_lantern_SWB01_A01_LAN01_T1": str(old_lantern.id) }) self.assertEqual(response_integrate.status_code, 302) # Old lantern must now be archived old_lantern.refresh_from_db() self.assertEqual(old_lantern.status, "archived") # A new replacement lantern must be active with the updated code new_active = TrafficLightLantern.objects.filter(pole=pole, code="SWB01_A01_LAN01_T1", status="active").first() self.assertIsNotNone(new_active) self.assertEqual(old_lantern.replaced_by, new_active) def test_parse_telecom_detectors(self): import openpyxl import tempfile from assets.utils.cross_plan_parser import parse_cross_plan # Create a mock cross plan workbook wb = openpyxl.Workbook() ws = wb.active ws.title = "kruisjesplan" # Row 1: Sleutel ws.cell(row=1, column=1, value="Sleutel:") ws.cell(row=1, column=2, value="SWB01") # Cable 2 section start ws.cell(row=5, column=1, value="Câble 02 TWAVB 10x2x0,8 mm²") # Row 6: Richting and Kringen headers ws.cell(row=6, column=3, value="richting") ws.cell(row=6, column=4, value="kringen") # Row 7: Pole headers ws.cell(row=7, column=5, value="B01") ws.cell(row=7, column=6, value="A01") # Rows for telecom detectors # Row 8: RC ws.cell(row=8, column=1, value="com") ws.cell(row=8, column=2, value="1") ws.cell(row=8, column=3, value="RC") ws.cell(row=8, column=4, value="") ws.cell(row=8, column=6, value="X") # Row 9: RD ws.cell(row=9, column=1, value="com") ws.cell(row=9, column=2, value="2") ws.cell(row=9, column=3, value="RD") ws.cell(row=9, column=4, value="") ws.cell(row=9, column=6, value="X") # Row 10: Rm T2,B1,B2 ws.cell(row=10, column=1, value="com") ws.cell(row=10, column=2, value="3") ws.cell(row=10, column=3, value="Rm T2,B1,B2") ws.cell(row=10, column=4, value="") ws.cell(row=10, column=5, value="X") # Save workbook to temp file with tempfile.NamedTemporaryFile(suffix=".xlsx", delete=False) as tmp: tmp_path = tmp.name wb.save(tmp_path) try: deduced = parse_cross_plan(tmp_path) # Check parsed detectors detectors = deduced["detectors"] self.assertTrue(len(detectors) >= 3) # Verify RC rc_det = next((d for d in detectors if d["code"] == "SWB01_A01_RC"), None) self.assertIsNotNone(rc_det) self.assertEqual(rc_det["model_name"], "TM60") self.assertEqual(rc_det["pole_code"], "SWB01_A01") # Verify RD rd_det = next((d for d in detectors if d["code"] == "SWB01_A01_RD"), None) self.assertIsNotNone(rd_det) self.assertEqual(rd_det["model_name"], "TM60") self.assertEqual(rd_det["pole_code"], "SWB01_A01") # Verify Rm T2 rm_det = next((d for d in detectors if d["code"] == "SWB01_B01_RMT2"), None) self.assertIsNotNone(rm_det) self.assertEqual(rm_det["model_name"], "TM60") self.assertEqual(rm_det["pole_code"], "SWB01_B01") finally: import os if os.path.exists(tmp_path): os.remove(tmp_path) def test_detector_prefix_match_and_rename(self): from assets.models import TrafficLightPole, TrafficLightDetector, TrafficLightPoleModel, TrafficLightDetectorModel, AssetCategory pole_model = TrafficLightPoleModel.objects.first() detector_model = TrafficLightDetectorModel.objects.first() category_pole = AssetCategory.objects.filter(code="TL_POLE").first() category_detector = AssetCategory.objects.filter(code="TL_DETECTOR").first() or AssetCategory.objects.filter(code="TL_DETECTEUR").first() # Pre-create pole in DB pole = TrafficLightPole.objects.create( intersection=self.intersection, code="SWB01_B01", model=pole_model, category=category_pole, status="active" ) # Pre-create detector with old nomenclature (no pole suffix) old_detector = TrafficLightDetector.objects.create( intersection=self.intersection, pole=pole, code="SWB01-RA", model=detector_model, category=category_detector, status="active" ) self.client.force_login(self.user) import os from django.conf import settings from django.core.files.uploadedfile import SimpleUploadedFile real_excel_path = os.path.join(os.path.dirname(settings.BASE_DIR), 'external_files', '18ac47df841-2808f-DOC-23270.xlsx') if not os.path.exists(real_excel_path): self.skipTest("Real Excel file for test not found") with open(real_excel_path, 'rb') as f: excel_content = f.read() uploaded_file = SimpleUploadedFile("test_cross_plan.xlsx", excel_content, content_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet") # POST to import_cross_plan to analyze url_import = reverse("assets:import_cross_plan", args=[self.intersection.id]) response = self.client.post(url_import, {"cross_plan_file_upload": uploaded_file}) self.assertEqual(response.status_code, 200) # Check detectors in preview context preview_detectors = response.context["detectors"] target_item = next((item for item in preview_detectors if item["code"] == "SWB01_B01_RA"), None) self.assertIsNotNone(target_item) self.assertEqual(target_item["status"], "modified") self.assertEqual(target_item["db_id"], old_detector.id) self.assertIn("Renommer le code", target_item["details_diff"]) # Integrate with checked checkbox url_integrate = reverse("assets:integrate_cross_plan", args=[self.intersection.id]) response_integrate = self.client.post(url_integrate, { "cross_plan_filename": response.context["cross_plan_filename"], "selected_detector_SWB01_B01_RA": "on", "action_detector_SWB01_B01_RA": "replace", "db_id_detector_SWB01_B01_RA": str(old_detector.id) }) self.assertEqual(response_integrate.status_code, 302) # Old detector must now be archived old_detector.refresh_from_db() self.assertEqual(old_detector.status, "archived") # A new replacement detector must be active with the updated code new_active = TrafficLightDetector.objects.filter(intersection=self.intersection, pole=pole, code="SWB01_B01_RA", status="active").first() self.assertIsNotNone(new_active) self.assertEqual(old_detector.replaced_by, new_active) def test_multiple_cross_plan_uploads(self): import openpyxl import tempfile from assets.models import TrafficLightPole, TrafficLightLantern, TrafficLightDetector # Workbook 1: Lantern A on pole B01 wb1 = openpyxl.Workbook() ws1 = wb1.active ws1.title = "kruisjesplan" ws1.cell(row=1, column=1, value="Sleutel:") ws1.cell(row=1, column=2, value="SWB01") ws1.cell(row=5, column=1, value="Câble 01 SVAVB") ws1.cell(row=6, column=3, value="richting") ws1.cell(row=6, column=4, value="kringen") ws1.cell(row=7, column=5, value="B01") ws1.cell(row=8, column=1, value="com") ws1.cell(row=8, column=2, value="1") ws1.cell(row=8, column=3, value="A") ws1.cell(row=8, column=4, value="") ws1.cell(row=8, column=5, value="X") # Workbook 2: Detector RC on pole B01 wb2 = openpyxl.Workbook() ws2 = wb2.active ws2.title = "kruisjesplan" ws2.cell(row=1, column=1, value="Sleutel:") ws2.cell(row=1, column=2, value="SWB01") ws2.cell(row=5, column=1, value="Câble 02 TWAVB") ws2.cell(row=6, column=3, value="richting") ws2.cell(row=6, column=4, value="kringen") ws2.cell(row=7, column=5, value="B01") ws2.cell(row=8, column=1, value="com") ws2.cell(row=8, column=2, value="1") ws2.cell(row=8, column=3, value="RC") ws2.cell(row=8, column=4, value="") ws2.cell(row=8, column=5, value="X") with tempfile.NamedTemporaryFile(suffix=".xlsx", delete=False) as tmp1: tmp1_path = tmp1.name wb1.save(tmp1_path) with tempfile.NamedTemporaryFile(suffix=".xlsx", delete=False) as tmp2: tmp2_path = tmp2.name wb2.save(tmp2_path) try: self.client.force_login(self.user) from django.core.files.uploadedfile import SimpleUploadedFile with open(tmp1_path, 'rb') as f1, open(tmp2_path, 'rb') as f2: file1 = SimpleUploadedFile("plan_lanterns.xlsx", f1.read(), content_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet") file2 = SimpleUploadedFile("plan_detectors.xlsx", f2.read(), content_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet") # Upload both files url_import = reverse("assets:import_cross_plan", args=[self.intersection.id]) response = self.client.post(url_import, { "cross_plan_file_upload": [file1, file2] }) self.assertEqual(response.status_code, 200) # Check that BOTH lanterns and detectors are in the preview context preview_lanterns = response.context["lanterns"] preview_detectors = response.context["detectors"] # Lantern A should be present self.assertTrue(any(l["code"] == "SWB01_B01_LAN01_A" for l in preview_lanterns)) # Detector RC should be present self.assertTrue(any(d["code"] == "SWB01_B01_RC" for d in preview_detectors)) # Now integrate url_integrate = reverse("assets:integrate_cross_plan", args=[self.intersection.id]) response_integrate = self.client.post(url_integrate, { "cross_plan_filename": response.context["cross_plan_filename"], "selected_pole_SWB01_B01": "on", "action_pole_SWB01_B01": "add", "selected_lantern_SWB01_B01_LAN01_A": "on", "action_lantern_SWB01_B01_LAN01_A": "add", "selected_detector_SWB01_B01_RC": "on", "action_detector_SWB01_B01_RC": "add" }) self.assertEqual(response_integrate.status_code, 302) # Verify that both are created in the database pole = TrafficLightPole.objects.filter(intersection=self.intersection, code="SWB01_B01", status="active").first() self.assertIsNotNone(pole) lantern = TrafficLightLantern.objects.filter(pole=pole, code="SWB01_B01_LAN01_A", status="active").first() self.assertIsNotNone(lantern) detector = TrafficLightDetector.objects.filter(intersection=self.intersection, pole=pole, code="SWB01_B01_RC", status="active").first() self.assertIsNotNone(detector) finally: import os for path in (tmp1_path, tmp2_path): if os.path.exists(path): os.remove(path) def test_parse_cross_plan_french_and_new_codes(self): import openpyxl import tempfile from assets.utils.cross_plan_parser import parse_cross_plan # Workbook 1: Using Clé: and code SB403 wb = openpyxl.Workbook() ws = wb.active ws.title = "Plan de croix" ws.cell(row=1, column=3, value="Clé:") ws.cell(row=1, column=4, value="SB403") ws.cell(row=2, column=1, value="Câble 01 SVAVB") ws.cell(row=3, column=3, value="richting") ws.cell(row=3, column=4, value="kringen") ws.cell(row=4, column=5, value="B01") ws.cell(row=5, column=1, value="com") ws.cell(row=5, column=2, value="1") ws.cell(row=5, column=3, value="A") ws.cell(row=5, column=4, value="") ws.cell(row=5, column=5, value="X") with tempfile.NamedTemporaryFile(suffix=".xlsx", delete=False) as tmp: tmp_path = tmp.name wb.save(tmp_path) try: deduced = parse_cross_plan(tmp_path) self.assertEqual(deduced["intersection_code"], "SB403") self.assertEqual(len(deduced["poles"]), 1) self.assertEqual(deduced["poles"][0]["code"], "SB403_B01") finally: import os if os.path.exists(tmp_path): os.remove(tmp_path) def test_parse_cross_plan_fallback_regex(self): import openpyxl import tempfile from assets.utils.cross_plan_parser import parse_cross_plan # Workbook 2: No label, but code SB403 in the first cells wb = openpyxl.Workbook() ws = wb.active ws.title = "kruisjesplan" ws.cell(row=1, column=4, value="SB403") ws.cell(row=2, column=1, value="Câble 01 SVAVB") ws.cell(row=3, column=3, value="richting") ws.cell(row=3, column=4, value="kringen") ws.cell(row=4, column=5, value="B01") ws.cell(row=5, column=1, value="com") ws.cell(row=5, column=2, value="1") ws.cell(row=5, column=3, value="A") ws.cell(row=5, column=4, value="") ws.cell(row=5, column=5, value="X") with tempfile.NamedTemporaryFile(suffix=".xlsx", delete=False) as tmp: tmp_path = tmp.name wb.save(tmp_path) try: deduced = parse_cross_plan(tmp_path) self.assertEqual(deduced["intersection_code"], "SB403") finally: import os if os.path.exists(tmp_path): os.remove(tmp_path) def test_parse_cross_plan_pushbuttons(self): import openpyxl import tempfile from assets.utils.cross_plan_parser import parse_cross_plan # Workbook 3: Testing various pushbutton nomenclatures (PBa1, BPf4, Dka) wb = openpyxl.Workbook() ws = wb.active ws.title = "kruisjesplan" ws.cell(row=1, column=1, value="Sleutel:") ws.cell(row=1, column=2, value="SWB01") ws.cell(row=5, column=1, value="Câble 01 SVAVB") ws.cell(row=6, column=3, value="richting") ws.cell(row=6, column=4, value="kringen") ws.cell(row=7, column=5, value="B01") # Row 8: PBa1 ws.cell(row=8, column=1, value="com") ws.cell(row=8, column=2, value="1") ws.cell(row=8, column=3, value="PBa1") ws.cell(row=8, column=5, value="X") # Row 9: BPf4 ws.cell(row=9, column=1, value="com") ws.cell(row=9, column=2, value="2") ws.cell(row=9, column=3, value="BPf4") ws.cell(row=9, column=5, value="X") # Row 10: Dka ws.cell(row=10, column=1, value="com") ws.cell(row=10, column=2, value="3") ws.cell(row=10, column=3, value="Dka") ws.cell(row=10, column=5, value="X") with tempfile.NamedTemporaryFile(suffix=".xlsx", delete=False) as tmp: tmp_path = tmp.name wb.save(tmp_path) try: deduced = parse_cross_plan(tmp_path) detectors = deduced["detectors"] self.assertEqual(len(detectors), 3) # Check PBa1 pb_det = next((d for d in detectors if d["code"] == "SWB01_B01_PBa1"), None) self.assertIsNotNone(pb_det) self.assertEqual(pb_det["name"], "bouton poussoir NF (jaune) sur B01 - Phase a") self.assertEqual(pb_det["model_name"], "Drukknop") self.assertEqual(pb_det["phases"], ["a"]) # Check BPf4 bp_det = next((d for d in detectors if d["code"] == "SWB01_B01_BPf4"), None) self.assertIsNotNone(bp_det) self.assertEqual(bp_det["name"], "bouton poussoir NO (bleu) sur B01 - Phase f") self.assertEqual(bp_det["model_name"], "Drukknop") self.assertEqual(bp_det["phases"], ["f"]) # Check Dka dk_det = next((d for d in detectors if d["code"] == "SWB01_B01_DKa"), None) self.assertIsNotNone(dk_det) self.assertEqual(dk_det["name"], "Drukknop sur B01 - Phase a") self.assertEqual(dk_det["model_name"], "Drukknop") self.assertEqual(dk_det["phases"], ["a"]) finally: import os if os.path.exists(tmp_path): os.remove(tmp_path) def test_cross_plan_custom_names_and_pole_short_code(self): import openpyxl import tempfile from assets.models import TrafficLightPole, TrafficLightDetector # Create mock workbook wb = openpyxl.Workbook() ws = wb.active ws.title = "kruisjesplan" ws.cell(row=1, column=1, value="Sleutel:") ws.cell(row=1, column=2, value="SWB01") ws.cell(row=5, column=1, value="Câble 01 SVAVB") ws.cell(row=6, column=3, value="richting") ws.cell(row=6, column=4, value="kringen") ws.cell(row=7, column=5, value="B01") ws.cell(row=7, column=6, value="B02") ws.cell(row=8, column=1, value="com") ws.cell(row=8, column=2, value="1") ws.cell(row=8, column=3, value="PBa1") ws.cell(row=8, column=5, value="X") with tempfile.NamedTemporaryFile(suffix=".xlsx", delete=False) as tmp: tmp_path = tmp.name wb.save(tmp_path) try: self.client.force_login(self.user) from django.core.files.uploadedfile import SimpleUploadedFile with open(tmp_path, 'rb') as f: uploaded_file = SimpleUploadedFile("plan_custom_names.xlsx", f.read(), content_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet") # 1. Preview url_import = reverse("assets:import_cross_plan", args=[self.intersection.id]) response = self.client.post(url_import, { "cross_plan_file_upload": uploaded_file }) self.assertEqual(response.status_code, 200) # Check that pole defaults to short code name in preview preview_poles = response.context["poles"] b01_prev = next((p for p in preview_poles if p["code"] == "SWB01_B01"), None) self.assertIsNotNone(b01_prev) self.assertEqual(b01_prev["name"], "B01") b02_prev = next((p for p in preview_poles if p["code"] == "SWB01_B02"), None) self.assertIsNotNone(b02_prev) self.assertEqual(b02_prev["name"], "B02") # 2. Integrate with customized names url_integrate = reverse("assets:integrate_cross_plan", args=[self.intersection.id]) response_integrate = self.client.post(url_integrate, { "cross_plan_filename": response.context["cross_plan_filename"], "selected_pole_SWB01_B01": "on", "action_pole_SWB01_B01": "add", "name_pole_SWB01_B01": "Mon Poteau B01", "selected_pole_SWB01_B02": "on", "action_pole_SWB01_B02": "add", # no name override for B02, should default to "B02" "selected_detector_SWB01_B01_PBa1": "on", "action_detector_SWB01_B01_PBa1": "add", "name_detector_SWB01_B01_PBa1": "Mon BP Jaune" }) self.assertEqual(response_integrate.status_code, 302) # Verify saved assets and names p1 = TrafficLightPole.objects.filter(intersection=self.intersection, code="SWB01_B01", status="active").first() self.assertIsNotNone(p1) self.assertEqual(p1.name_fr, "Mon Poteau B01") p2 = TrafficLightPole.objects.filter(intersection=self.intersection, code="SWB01_B02", status="active").first() self.assertIsNotNone(p2) self.assertEqual(p2.name_fr, "B02") det = TrafficLightDetector.objects.filter(intersection=self.intersection, pole=p1, code="SWB01_B01_PBa1", status="active").first() self.assertIsNotNone(det) self.assertEqual(det.name, "Mon BP Jaune") finally: import os if os.path.exists(tmp_path): os.remove(tmp_path) def test_import_cross_plan_from_existing_document(self): from documents.models import DocumentTag, ManagedDocument, DocumentVersion, DocumentAttachment from django.contrib.contenttypes.models import ContentType from django.core.files.uploadedfile import SimpleUploadedFile import os from django.conf import settings # 1. Create the DocumentTag 'cross_plan' tag, _ = DocumentTag.objects.get_or_create( slug='cross_plan', defaults={ 'name_fr': 'Plan de croix', 'name_nl': 'Kruisjesplan', 'show_in_asset_detail': True } ) # 2. Create the ManagedDocument and version with real Excel real_excel_path = os.path.join(os.path.dirname(settings.BASE_DIR), 'external_files', '18ac47df841-2808f-DOC-23270.xlsx') if not os.path.exists(real_excel_path): self.skipTest("Real Excel file for test not found") with open(real_excel_path, 'rb') as f: excel_content = f.read() uploaded_file = SimpleUploadedFile("18ac47df841-2808f-DOC-23270.xlsx", excel_content, content_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet") doc = ManagedDocument.objects.create( title="Plan de croix existant", created_by=self.user ) doc.tags.add(tag) version = DocumentVersion.objects.create( document=doc, file=uploaded_file, uploaded_by=self.user, version_number=1 ) # 3. Attach to intersection ct_tli = ContentType.objects.get_for_model(self.intersection.__class__) DocumentAttachment.objects.create( document=doc, content_type=ct_tli, object_id=self.intersection.id ) self.client.force_login(self.user) # 4. Check GET request contains the plan choice url_import = reverse("assets:import_cross_plan", args=[self.intersection.id]) response = self.client.get(url_import) self.assertEqual(response.status_code, 200) self.assertIn("cross_plan_choices", response.context) choices = response.context["cross_plan_choices"] self.assertTrue(any(c["id"] == f"managed-{doc.id}" for c in choices)) # 5. POST to analyze using existing plan response_post = self.client.post(url_import, { "existing_cross_plans": [f"managed-{doc.id}"] }) self.assertEqual(response_post.status_code, 200) self.assertEqual(response_post.context["step"], "preview") self.assertTrue(len(response_post.context["poles"]) > 0) def test_parse_cross_plan_pedestrian_hp(self): import openpyxl import tempfile from assets.utils.cross_plan_parser import parse_cross_plan # Workbook: Pedestrian phase aF with HP line wb = openpyxl.Workbook() ws = wb.active ws.title = "kruisjesplan" ws.cell(row=1, column=1, value="Sleutel:") ws.cell(row=1, column=2, value="SWB01") ws.cell(row=5, column=1, value="Câble 01 SVAVB") ws.cell(row=6, column=3, value="richting") ws.cell(row=6, column=4, value="kringen") ws.cell(row=7, column=5, value="B01") # Row 8: Pedestrian phase aF ws.cell(row=8, column=1, value="com") ws.cell(row=8, column=2, value="1") ws.cell(row=8, column=3, value="aF") ws.cell(row=8, column=5, value="X") # Row 9: HP ws.cell(row=9, column=1, value="com") ws.cell(row=9, column=2, value="2") ws.cell(row=9, column=3, value="HP") ws.cell(row=9, column=5, value="X") with tempfile.NamedTemporaryFile(suffix=".xlsx", delete=False) as tmp: tmp_path = tmp.name wb.save(tmp_path) try: deduced = parse_cross_plan(tmp_path) lanterns = deduced["lanterns"] self.assertEqual(len(lanterns), 1) self.assertEqual(lanterns[0]["model_name"], "2V200 piéton+HP") self.assertEqual(lanterns[0]["code"], "SWB01_B01_LAN01_aF") finally: import os if os.path.exists(tmp_path): os.remove(tmp_path) def test_parse_cross_plan_double_cross_lanterns(self): import openpyxl import tempfile from assets.utils.cross_plan_parser import parse_cross_plan # Workbook with pole B01 (with arm) and double crosses (XX) for vehicle phase A, pedestrian a, and tram T1 wb = openpyxl.Workbook() ws = wb.active ws.title = "kruisjesplan" ws.cell(row=1, column=1, value="Sleutel:") ws.cell(row=1, column=2, value="SWB01") ws.cell(row=5, column=1, value="Câble 01 SVAVB") ws.cell(row=6, column=3, value="richting") ws.cell(row=6, column=4, value="kringen") ws.cell(row=7, column=5, value="B01") # Double crosses for vehicle phase A ws.cell(row=8, column=3, value="A") ws.cell(row=8, column=4, value="Y") ws.cell(row=8, column=5, value="XX") # Double crosses for pedestrian phase a ws.cell(row=9, column=3, value="a") ws.cell(row=9, column=4, value="g") ws.cell(row=9, column=5, value="XX") # Double crosses for tram phase T1 ws.cell(row=10, column=3, value="T1") ws.cell(row=10, column=4, value="Y") ws.cell(row=10, column=5, value="XX") with tempfile.NamedTemporaryFile(suffix=".xlsx", delete=False) as tmp: tmp_path = tmp.name wb.save(tmp_path) try: deduced = parse_cross_plan(tmp_path) lanterns = deduced["lanterns"] # We expect two lanterns for A: SWB01_B01_LAN01_A (3V200) and SWB01_B01_LAN02_A (3V300) lan01 = next((l for l in lanterns if l["code"] == "SWB01_B01_LAN01_A"), None) self.assertIsNotNone(lan01) self.assertEqual(lan01["model_name"], "3V200") lan02 = next((l for l in lanterns if l["code"] == "SWB01_B01_LAN02_A"), None) self.assertIsNotNone(lan02) self.assertEqual(lan02["model_name"], "3V300") # We expect two standard pedestrian lanterns for a: LAN05_a and LAN06_a (2V200) lan_a1 = next((l for l in lanterns if l["code"] == "SWB01_B01_LAN05_a"), None) self.assertIsNotNone(lan_a1) self.assertEqual(lan_a1["model_name"], "2V200") lan_a2 = next((l for l in lanterns if l["code"] == "SWB01_B01_LAN06_a"), None) self.assertIsNotNone(lan_a2) self.assertEqual(lan_a2["model_name"], "2V200") # We expect two tram lanterns for T1: LAN03_T1 (3V200 tram) and LAN04_T1 (3V300 tram) lan_t1 = next((l for l in lanterns if l["code"] == "SWB01_B01_LAN03_T1"), None) self.assertIsNotNone(lan_t1) self.assertEqual(lan_t1["model_name"], "3V200 tram tout droite") lan_t2 = next((l for l in lanterns if l["code"] == "SWB01_B01_LAN04_T1"), None) self.assertIsNotNone(lan_t2) self.assertEqual(lan_t2["model_name"], "3V300 tram tout droit") finally: import os if os.path.exists(tmp_path): os.remove(tmp_path) def test_parse_cross_plan_detector_model_mapping(self): import openpyxl import tempfile from assets.utils.cross_plan_parser import parse_cross_plan wb = openpyxl.Workbook() ws = wb.active ws.title = "kruisjesplan" ws.cell(row=1, column=1, value="Sleutel:") ws.cell(row=1, column=2, value="SWB01") ws.cell(row=5, column=1, value="Câble 02 TWAVB") ws.cell(row=6, column=3, value="richting") ws.cell(row=6, column=4, value="kringen") ws.cell(row=7, column=5, value="B01") # Row 8: RC (starts with R, but not Rp) -> should map to TM60 ws.cell(row=8, column=3, value="RC") ws.cell(row=8, column=5, value="X") # Row 9: RpA (starts with Rp) -> should map to TMA122 ws.cell(row=9, column=3, value="RpA") ws.cell(row=9, column=5, value="X") with tempfile.NamedTemporaryFile(suffix=".xlsx", delete=False) as tmp: tmp_path = tmp.name wb.save(tmp_path) try: deduced = parse_cross_plan(tmp_path) detectors = deduced["detectors"] self.assertEqual(len(detectors), 2) det_rc = next((d for d in detectors if d["code"] == "SWB01_B01_RC"), None) self.assertIsNotNone(det_rc) self.assertEqual(det_rc["model_name"], "TM60") det_rpa = next((d for d in detectors if d["code"] == "SWB01_B01_RPA"), None) self.assertIsNotNone(det_rpa) self.assertEqual(det_rpa["model_name"], "TMA-122 M") finally: import os if os.path.exists(tmp_path): os.remove(tmp_path) def test_parse_cross_plan_video_detectors(self): import openpyxl import tempfile from assets.utils.cross_plan_parser import parse_cross_plan wb = openpyxl.Workbook() ws = wb.active ws.title = "kruisjesplan" ws.cell(row=1, column=1, value="Sleutel:") ws.cell(row=1, column=2, value="SWB01") ws.cell(row=5, column=1, value="Câble 02 TWAVB") # Header indicating a Detector section ws.cell(row=6, column=3, value="Detecteur") ws.cell(row=6, column=4, value="I/O") ws.cell(row=7, column=5, value="B01") # Row 8: VAF VT VA (starts with V) -> should map to Traficam wide angle ws.cell(row=8, column=3, value="VAF VT VA") ws.cell(row=8, column=5, value="X") with tempfile.NamedTemporaryFile(suffix=".xlsx", delete=False) as tmp: tmp_path = tmp.name wb.save(tmp_path) try: deduced = parse_cross_plan(tmp_path) detectors = deduced["detectors"] self.assertEqual(len(detectors), 1) det_video = next((d for d in detectors if d["code"] == "SWB01_B01_VAFVTVA"), None) self.assertIsNotNone(det_video) self.assertEqual(det_video["model_name"], "Traficam wide angle") # Verify no lanterns were created since it was in the Detecteur section self.assertEqual(len(deduced["lanterns"]), 0) finally: import os if os.path.exists(tmp_path): os.remove(tmp_path) class AssetDefaultPositionTest(TestCase): def test_assign_default_position_hierarchy(self): from django.contrib.gis.geos import Polygon, MultiPolygon from assets.models.structures import StructureLocation, Structure, StructureGeoAsset, StructureAssetModel from assets.views.asset_editing import _assign_default_position # Create a parent location with geometry parent_geom = MultiPolygon( Polygon(((0, 0), (0, 10), (10, 10), (10, 0), (0, 0))), srid=3812 ) parent_loc = StructureLocation.objects.create( name_fr="Parent Location", geom=parent_geom, lon=5.0, lat=50.0 ) # Create a child location without geometry child_loc = StructureLocation.objects.create( name_fr="Child Location", parent_location=parent_loc ) # Create structure structure = Structure.objects.create( name_fr="Structure Test", main_location=parent_loc ) child_loc.structure = structure child_loc.save() # Create model model = StructureAssetModel.objects.create( name_fr="Model Test" ) # Create asset without geometry initially asset = StructureGeoAsset( code="TEST-001", model=model, location=child_loc, structure=structure ) # Call _assign_default_position _assign_default_position(asset, child_loc) # Assert coordinates/geometry are set from parent location self.assertIsNotNone(asset.geom) self.assertEqual(asset.geom.srid, 3812) self.assertEqual(asset.geom.x, parent_geom.centroid.x) self.assertEqual(asset.geom.y, parent_geom.centroid.y) self.assertIsNotNone(asset.lon) self.assertIsNotNone(asset.lat) self.assertIsNotNone(asset.geojson) class AssetValidationPermissionTest(TestCase): def setUp(self): from django.contrib.auth.models import User from common.models import UserConfig, Thematic, UserThematics from assets.models import AssetCategory from assets.models.structures import StructureGeoAsset, StructureAssetModel self.user = User.objects.create_user(username="validator_user", password="password") self.user_config = UserConfig.objects.create(user=self.user, is_intern=True) self.thematic = Thematic.objects.create(code="structures", name_fr="Ouvrages d'art") self.user_thematic = UserThematics.objects.create( user_config=self.user_config, thematic=self.thematic, can_view_assets=True, can_edit_assets=True, can_validate_assets=False # Start with validation disabled ) self.category = AssetCategory.objects.create( thematic=self.thematic, code="CAT", name_fr="Category" ) self.model = StructureAssetModel.objects.create(name_fr="Model") self.asset = StructureGeoAsset.objects.create( code="TEST-VAL-01", model=self.model, category=self.category, status="active" ) def test_validation_permission(self): from assets.permissions import can_validate_asset, get_allowed_update_fields_for_asset # Without permission self.assertFalse(can_validate_asset(self.user, self.asset)) fields = get_allowed_update_fields_for_asset(self.user, self.asset) self.assertNotIn("validation_status", fields) # With permission self.user_thematic.can_validate_assets = True self.user_thematic.save() self.assertTrue(can_validate_asset(self.user, self.asset)) fields = get_allowed_update_fields_for_asset(self.user, self.asset) self.assertIn("validation_status", fields) def test_bulk_edit_assets(self): import json from django.urls import reverse # Login/authenticate the user self.client.force_login(self.user) url = reverse('assets:bulk_edit_assets') payload = { "assets": [{"model": "structuregeoasset", "object_id": self.asset.pk}], "field_name": "validation_status", "new_value": "validated" } # 1. Without permission response = self.client.post( url, data=json.dumps(payload), content_type="application/json" ) self.assertEqual(response.status_code, 200) data = response.json() self.assertTrue(data['success']) self.assertEqual(data['updated_count'], 0) self.assertEqual(data['skipped_count'], 1) # 2. Enable permission self.user_thematic.can_validate_assets = True self.user_thematic.save() response = self.client.post( url, data=json.dumps(payload), content_type="application/json" ) self.assertEqual(response.status_code, 200) data = response.json() self.assertTrue(data['success']) self.assertEqual(data['updated_count'], 1) self.assertEqual(data['skipped_count'], 0) # Verify database update self.asset.refresh_from_db() self.assertEqual(self.asset.validation_status, 'validated') @tag('slow', 'dxf') @skipUnless(RUN_DXF_TESTS, "Skip slow DXF tests by default") class TrafficLightDXFLayerSelectionTest(TestCase): def setUp(self): from common.models import Thematic, UserConfig, UserThematics from assets.models import TrafficLightIntersection self.thematic, _ = Thematic.objects.get_or_create( code="trafficlights", defaults={ "name_fr": "Feux de signalisation", "name_nl": "Verkeerslichten" } ) self.user = User.objects.create_user(username="testuser", password="password123") self.config = UserConfig.objects.create(user=self.user, limit_assets_to_contracts=False) UserThematics.objects.create( user_config=self.config, thematic=self.thematic, can_view_assets=True, can_edit_assets=True ) self.intersection = TrafficLightIntersection.objects.create( code="SB0239", name_fr="Carrefour SB0239", status="active" ) # Create a real temporary DXF file in settings.PRIVATE_MEDIA_ROOT / 'dxf_files' import os import ezdxf from django.conf import settings self.dxf_dir = os.path.join(settings.PRIVATE_MEDIA_ROOT, 'dxf_files') os.makedirs(self.dxf_dir, exist_ok=True) self.dxf_filename = "test_layers.dxf" self.dxf_path = os.path.join(self.dxf_dir, self.dxf_filename) # Create DXF with ezdxf and add some geometry in different layers doc = ezdxf.new() doc.layers.new('LAYER_A', dxfattribs={'color': 2}) doc.layers.new('LAYER_B', dxfattribs={'color': 5}) msp = doc.modelspace() # Add entity to LAYER_A msp.add_line((0, 0), (10, 10), dxfattribs={'layer': 'LAYER_A'}) # Add entity to LAYER_B msp.add_line((10, 10), (20, 20), dxfattribs={'layer': 'LAYER_B'}) doc.saveas(self.dxf_path) def tearDown(self): import os if os.path.exists(self.dxf_path): try: os.remove(self.dxf_path) except Exception: pass # Also clean up the cached layouts created by tests from django.conf import settings cache_dir = os.path.join(settings.PRIVATE_MEDIA_ROOT, 'dxf_layouts') if os.path.exists(cache_dir): import glob for f in glob.glob(os.path.join(cache_dir, "test_layers.*")): try: os.remove(f) except Exception: pass def test_dxf_layers_endpoint(self): self.client.force_login(self.user) url = reverse("assets:dxf_layers", args=[self.intersection.id]) response = self.client.get(url, {"file": self.dxf_filename}) self.assertEqual(response.status_code, 200) data = response.json() self.assertIn("layers", data) layers = data["layers"] # Should have LAYER_A and LAYER_B layer_names = [l["name"] for l in layers] self.assertIn("LAYER_A", layer_names) self.assertIn("LAYER_B", layer_names) # Colors should match (LAYER_A color ACI 2 is yellow #ffff00, LAYER_B ACI 5 is blue #0000ff) layer_a = next(l for l in layers if l["name"] == "LAYER_A") layer_b = next(l for l in layers if l["name"] == "LAYER_B") self.assertEqual(layer_a["color"], "#ffff00") self.assertEqual(layer_b["color"], "#0000ff") def test_dxf_layout_geojson_with_all_layers(self): self.client.force_login(self.user) url = reverse("assets:dxf_layout_geojson", args=[self.intersection.id]) # All layers (default) response = self.client.get(url, {"file": self.dxf_filename, "layers": "all"}) self.assertEqual(response.status_code, 200) data = response.json() self.assertIn("features", data) features = data["features"] # Should have features for both LAYER_A and LAYER_B feature_layers = [f["properties"]["Layer"] for f in features] self.assertIn("LAYER_A", feature_layers) self.assertIn("LAYER_B", feature_layers) def test_dxf_layout_geojson_with_subset_layers(self): self.client.force_login(self.user) url = reverse("assets:dxf_layout_geojson", args=[self.intersection.id]) # Subset: only LAYER_A response = self.client.get(url, {"file": self.dxf_filename, "layers": "LAYER_A"}) self.assertEqual(response.status_code, 200) data = response.json() features = data["features"] feature_layers = [f["properties"]["Layer"] for f in features] self.assertIn("LAYER_A", feature_layers) self.assertNotIn("LAYER_B", feature_layers) def test_dxf_layout_geojson_with_no_layers(self): self.client.force_login(self.user) url = reverse("assets:dxf_layout_geojson", args=[self.intersection.id]) # None layers response = self.client.get(url, {"file": self.dxf_filename, "layers": "none"}) self.assertEqual(response.status_code, 200) data = response.json() features = data["features"] self.assertEqual(len(features), 0) def test_import_dxf_existing_plan_selection_flow(self): self.client.force_login(self.user) # 1. First trigger layout caching for a subset of layers url_layout = reverse("assets:dxf_layout_geojson", args=[self.intersection.id]) response_layout = self.client.get(url_layout, {"file": self.dxf_filename, "layers": "LAYER_A"}) self.assertEqual(response_layout.status_code, 200) # 2. POST to import_dxf indicating we selected this existing file url_import = reverse("assets:import_dxf", args=[self.intersection.id]) response_import = self.client.post(url_import, {"dxf_file": self.dxf_filename}) self.assertEqual(response_import.status_code, 200) # 3. Verify context variables passed to template self.assertTrue(response_import.context["is_existing_plan"]) self.assertTrue(response_import.context["has_cached_layout"]) self.assertIsNotNone(response_import.context["previous_selection"]) # previous_selection should decode back to ['LAYER_A'] import json prev_selection = json.loads(response_import.context["previous_selection"]) self.assertEqual(prev_selection, ["LAYER_A"]) def test_layout_cache_overwritten_on_selection_change(self): self.client.force_login(self.user) import os import json from django.conf import settings # 1. First request for LAYER_A only url_layout = reverse("assets:dxf_layout_geojson", args=[self.intersection.id]) response_a = self.client.get(url_layout, {"file": self.dxf_filename, "layers": "LAYER_A"}) self.assertEqual(response_a.status_code, 200) data_a = response_a.json() self.assertEqual(data_a["selected_layers"], ["LAYER_A"]) # Ensure single cache file exists geojson_dir = os.path.join(settings.PRIVATE_MEDIA_ROOT, 'dxf_layouts') cache_path = os.path.join(geojson_dir, f"{self.dxf_filename.lower().replace('.dxf', '.layout.geojson')}") self.assertTrue(os.path.exists(cache_path)) # 2. Verify cache contents with open(cache_path, "r", encoding="utf-8") as f: content_a = json.load(f) self.assertEqual(content_a["selected_layers"], ["LAYER_A"]) # 3. Second request for LAYER_B only (different selection) response_b = self.client.get(url_layout, {"file": self.dxf_filename, "layers": "LAYER_B"}) self.assertEqual(response_b.status_code, 200) data_b = response_b.json() self.assertEqual(data_b["selected_layers"], ["LAYER_B"]) # 4. Verify cache file was overwritten with new selection with open(cache_path, "r", encoding="utf-8") as f: content_b = json.load(f) self.assertEqual(content_b["selected_layers"], ["LAYER_B"]) def test_import_dxf_document_association_and_filter(self): import os from assets.models import TrafficLightIntersectionDocument from django.core.files.uploadedfile import SimpleUploadedFile self.client.force_login(self.user) # 1. Create a dummy DXF file upload associated with the intersection dummy_dxf = SimpleUploadedFile("carrefour_plan.dxf", b"Dummy DXF content", content_type="application/dxf") url_import = reverse("assets:import_dxf", args=[self.intersection.id]) response_post = self.client.post(url_import, {"dxf_file_upload": dummy_dxf}) # Verify document was created dxf_docs = TrafficLightIntersectionDocument.objects.filter(intersection=self.intersection) self.assertEqual(dxf_docs.count(), 1) doc = dxf_docs.first() self.assertTrue(doc.file.name.endswith(".dxf")) # 2. Get request to import_dxf selection step response_get = self.client.get(url_import) self.assertEqual(response_get.status_code, 200) # Verify the file is listed in the dxf_files list context self.assertIn(os.path.basename(doc.file.name), response_get.context["dxf_files"]) # Clean up database document file try: doc.file.delete() doc.delete() except Exception: pass class TrafficLightRadarRelationTests(TestCase): def setUp(self): from django.contrib.gis.geos import Point from assets.models import TrafficLightIntersection, TrafficLightRadarModel, TrafficLightRadarPoleModel, TrafficLightRadarPole, TrafficLightPole, TrafficLightPoleModel self.intersection = TrafficLightIntersection.objects.create( code="INT-TEST", name_fr="Carrefour Test", geom=Point(4.35, 50.85, srid=4326) ) # Transform intersection geom to 3812 as done in DB self.intersection.geom.transform(3812) self.intersection.save() self.radar_model = TrafficLightRadarModel.objects.create( code="RM-001", name_fr="Modèle Radar Test" ) self.pole_model = TrafficLightRadarPoleModel.objects.create( code="PM-001", name_fr="Modèle Poteau Radar Test" ) self.tl_pole_model = TrafficLightPoleModel.objects.create( code="TPM-001", name_fr="Modèle Poteau Tricolore Test" ) def test_radar_geom_alignment_with_support(self): from django.contrib.gis.geos import Point from assets.models import TrafficLightRadar, TrafficLightRadarPole # 1. Create a support pole with specific geometry support_geom = Point(4.351, 50.851, srid=4326) support_geom.transform(3812) pole = TrafficLightRadarPole.objects.create( code="POLE-001", intersection=self.intersection, model=self.pole_model, geom=support_geom ) # 2. Create a radar initially at a different position initial_geom = Point(4.35, 50.85, srid=4326) initial_geom.transform(3812) radar = TrafficLightRadar.objects.create( code="RADAR-001", intersection=self.intersection, model=self.radar_model, geom=initial_geom ) self.assertNotEqual(radar.geom, support_geom) # 3. Link the radar to the support pole and save radar.pole = pole radar.save() # 4. Verify position was aligned to support pole self.assertEqual(radar.geom, support_geom) def test_radar_connection_lines_geojson_helper(self): from django.contrib.gis.geos import Point from assets.models import TrafficLightRadar, TrafficLightRadarPole, TrafficLightPole from assets.views.core import get_connection_lines_geojson # 1. Create support pole pole_geom = Point(4.351, 50.851, srid=4326) pole_geom.transform(3812) pole = TrafficLightRadarPole.objects.create( code="POLE-001", intersection=self.intersection, model=self.pole_model, geom=pole_geom ) # 2. Create power connection pole (trafficlights_pole) at a different location power_geom = Point(4.352, 50.852, srid=4326) power_geom.transform(3812) tl_pole = TrafficLightPole.objects.create( code="TLPOLE-001", intersection=self.intersection, model=self.tl_pole_model, geom=power_geom ) # 3. Create radar linked to both radar = TrafficLightRadar.objects.create( code="RADAR-001", intersection=self.intersection, model=self.radar_model, pole=pole, power_trafficlights_pole=tl_pole, geom=pole_geom ) # 4. Call helper geojson = get_connection_lines_geojson(radar) self.assertIsNotNone(geojson) self.assertEqual(geojson["type"], "FeatureCollection") self.assertEqual(len(geojson["features"]), 1) feat = geojson["features"][0] self.assertEqual(feat["properties"]["relation_type"], "electrical_connection") self.assertEqual(feat["properties"]["parent_code"], "TLPOLE-001") # Coordinates should be in EPSG:4326 (WGS84) expected_p1 = pole_geom.clone() expected_p1.transform(4326) expected_p2 = power_geom.clone() expected_p2.transform(4326) coords = feat["geometry"]["coordinates"] self.assertAlmostEqual(coords[0][0], expected_p1.x, places=5) self.assertAlmostEqual(coords[0][1], expected_p1.y, places=5) self.assertAlmostEqual(coords[1][0], expected_p2.x, places=5) self.assertAlmostEqual(coords[1][1], expected_p2.y, places=5) class TrafficLightPoleSignPanelsTest(TestCase): def setUp(self): from assets.models import TrafficLightIntersection, TrafficLightPoleModel, TrafficLightPole from assets.models.sign import SignPanelModel from common.models import Thematic self.thematic = Thematic.objects.get_or_create(code="trafficlights", defaults={"name_fr": "Trafic"})[0] self.sign_thematic = Thematic.objects.get_or_create(code="sign", defaults={"name_fr": "Signalisation"})[0] self.intersection = TrafficLightIntersection.objects.create( code="INT-001", name_fr="Test Intersection" ) self.pole_model = TrafficLightPoleModel.objects.create( code="TPM-001", name_fr="Modèle Poteau Tricolore Test" ) self.pole = TrafficLightPole.objects.create( code="POLE-001", intersection=self.intersection, model=self.pole_model ) self.panel_model = SignPanelModel.objects.create( code="SPM-001", name_fr="Modèle Panneau Test" ) def test_sorted_signpanels(self): from assets.models.sign import SignPanel # Create sign panels attached to the traffic light pole panel2 = SignPanel.objects.create( code="PANEL-002", model=self.panel_model, trafficlight_pole=self.pole ) panel1 = SignPanel.objects.create( code="PANEL-001", model=self.panel_model, trafficlight_pole=self.pole ) # Verify that sorted_signpanels returns them sorted by code panels = list(self.pole.sorted_signpanels) self.assertEqual(len(panels), 2) self.assertEqual(panels[0], panel1) self.assertEqual(panels[1], panel2) def test_radar_pole_sorted_signpanels(self): from assets.models import TrafficLightRadarPole, TrafficLightRadarPoleModel radar_pole_model = TrafficLightRadarPoleModel.objects.create( code="RPM-001", name_fr="Modèle Radar Pole Test" ) radar_pole = TrafficLightRadarPole.objects.create( code="RPOLE-001", intersection=self.intersection, model=radar_pole_model ) from assets.models.buildings import BuildingAsset class PublicBuildingTestCase(TestCase): def setUp(self): # Complexe sportif (Localisation Racine) self.complex = PublicBuilding.objects.create( code="CMP-001", name_fr="Complexe Sportif Municipal", name_nl="Gemeentelijk Sportcomplex", address="Rue du Sport 1, 1000 Bruxelles", owner="Ville de Bruxelles", usage_type="sports", location_type="site", building_status="in_service", is_erp=True, erp_category="1", erp_type="X", geom=MultiPolygon(Polygon(((0, 0), (0, 10), (10, 10), (10, 0), (0, 0)))), ) # Bloc A (Localisation Enfant) self.bloc_a = PublicBuilding.objects.create( code="BLOC-A", name_fr="Bloc A", parent_location=self.complex, location_type="building", ) # Étage 0 (Localisation Enfant du Bloc A) self.etage_0 = PublicBuilding.objects.create( code="ETG-0", name_fr="Étage 0", parent_location=self.bloc_a, location_type="level", level_number=0, ) def test_public_building_location_hierarchy(self): self.assertEqual(self.complex.code, "CMP-001") self.assertEqual(self.complex.location_type, "site") self.assertEqual(self.bloc_a.parent_location, self.complex) self.assertEqual(self.etage_0.parent_location, self.bloc_a) self.assertIn(self.bloc_a, self.complex.get_child_locations()) def test_building_asset_tree_with_civil_engineering(self): # Génie civil / Bâti A (Asset Racine) civil_asset = BuildingAsset.objects.create( building=self.complex, location=self.bloc_a, code="BATI-A", name_fr="Bâti Bloc A", asset_type="civil_engineering" ) # Asset Enfant: Façade Nord facade_asset = BuildingAsset.objects.create( building=self.complex, location=self.bloc_a, parent_asset=civil_asset, code="FAC-N", name_fr="Façade Nord", asset_type="facade" ) # Asset Enfant: Chaudière hvac_asset = BuildingAsset.objects.create( building=self.complex, location=self.etage_0, parent_asset=civil_asset, code="CVC-01", name_fr="Chaudière", asset_type="hvac" ) self.assertEqual(civil_asset.asset_type, "civil_engineering") self.assertEqual(facade_asset.parent_asset, civil_asset) self.assertEqual(hvac_asset.parent_asset, civil_asset) self.assertIn(facade_asset, civil_asset.get_child_assets()) self.assertIn(hvac_asset, civil_asset.get_child_assets()) class PublicBuildingEditViewTest(TestCase): def setUp(self): self.thematic = Thematic.objects.create( code="publicbuildings", name_fr="Bâtiments Publics", name_nl="Openbare Gebouwen" ) self.building = PublicBuilding.objects.create( code="BAT-999", name_fr="Bâtiment de Test", owner="Commune Test" ) # User without edit permissions self.viewer_user = User.objects.create_user(username="viewer", password="password123") self.viewer_config = UserConfig.objects.create(user=self.viewer_user) UserThematics.objects.create( user_config=self.viewer_config, thematic=self.thematic, can_view_assets=True, can_edit_assets=False ) # User with edit permissions self.editor_user = User.objects.create_user(username="editor", password="password123") self.editor_config = UserConfig.objects.create(user=self.editor_user) UserThematics.objects.create( user_config=self.editor_config, thematic=self.thematic, can_view_assets=True, can_edit_assets=True ) def test_publicbuilding_edit_permission_denied_for_viewer(self): self.client.login(username="viewer", password="password123") url = reverse('assets:publicbuilding_edit', kwargs={'pk': self.building.pk}) response = self.client.get(url) self.assertEqual(response.status_code, 403) def test_publicbuilding_edit_allowed_for_editor(self): self.client.login(username="editor", password="password123") url = reverse('assets:publicbuilding_edit', kwargs={'pk': self.building.pk}) response = self.client.get(url) self.assertEqual(response.status_code, 200) self.assertContains(response, "BAT-999") def test_publicbuilding_edit_post_updates_building(self): self.client.login(username="editor", password="password123") url = reverse('assets:publicbuilding_edit', kwargs={'pk': self.building.pk}) post_data = { 'code': 'BAT-999', 'location_type': 'building', 'name_fr': 'Bâtiment Modifié', 'name_nl': 'Gewijzigd Gebouw', 'address': 'Rue du Test 123', 'owner': 'Nouveau Propriétaire', 'usage_type': 'administrative', 'building_status': 'in_service', 'is_erp': True, } response = self.client.post(url, post_data) self.assertRedirects(response, reverse('assets:publicbuilding_detail', kwargs={'pk': self.building.pk})) self.building.refresh_from_db() self.assertEqual(self.building.name_fr, 'Bâtiment Modifié') self.assertEqual(self.building.address, 'Rue du Test 123') self.assertEqual(self.building.owner, 'Nouveau Propriétaire') class GeoSyncMixinTest(TestCase): def setUp(self): from common.models import Municipality from assets.models import PublicLightingStreet, PublicLightingPoleModel, PublicLightingPole, ITSLocation self.municipality = Municipality.objects.create(name_fr="Bruxelles", name_nl="Brussel", fid="1000") self.street = PublicLightingStreet.objects.create( code="STREET-001", municipality=self.municipality, ) self.pole_model = PublicLightingPoleModel.objects.create( code="POLEMODEL-001", name_fr="Modèle Poteau", ) self.pole_class = PublicLightingPole self.its_location_class = ITSLocation def test_update_lon_lat_updates_geojson_and_geom(self): pole = self.pole_class.objects.create( code="PLPOLE-SYNC-1", street=self.street, model=self.pole_model, lon=4.3520, lat=50.8475, ) self.assertIsNotNone(pole.geojson) self.assertIn('"type": "Point"', pole.geojson) self.assertIsNotNone(pole.geom) self.assertEqual(pole.geom.srid, 3812) def test_update_geom_updates_geojson_and_coords(self): from django.contrib.gis.geos import Point pt_3812 = Point(648819, 670788, srid=3812) pole = self.pole_class.objects.create( code="PLPOLE-SYNC-2", street=self.street, model=self.pole_model, geom=pt_3812, ) self.assertIsNotNone(pole.geojson) self.assertIsNotNone(pole.lon) self.assertIsNotNone(pole.lat) self.assertAlmostEqual(pole.lon, 4.35, places=1) self.assertAlmostEqual(pole.lat, 50.85, places=1) def test_polygon_concave_uses_point_on_surface(self): from django.contrib.gis.geos import GEOSGeometry, Point, MultiPolygon # U-shaped polygon in WGS84 transformed to 3812 u_poly_wgs84 = GEOSGeometry('POLYGON ((4.3500 50.8500, 4.3500 50.8510, 4.3510 50.8510, 4.3510 50.8508, 4.3502 50.8508, 4.3502 50.8500, 4.3500 50.8500))', srid=4326) u_poly_3812 = MultiPolygon(u_poly_wgs84.transform(3812, clone=True)) location = self.its_location_class.objects.create( code="ITSLOC-SYNC-1", geom=u_poly_3812, ) self.assertIsNotNone(location.geojson) self.assertIsNotNone(location.lon) self.assertIsNotNone(location.lat) point_result = Point(location.lon, location.lat, srid=4326) self.assertTrue(u_poly_wgs84.intersects(point_result) or u_poly_wgs84.contains(point_result)) def test_update_fields_includes_spatial_changes(self): pole = self.pole_class.objects.create( code="PLPOLE-SYNC-3", street=self.street, model=self.pole_model, lon=4.3520, lat=50.8475, ) pole.lon = 4.3600 pole.save(update_fields=['lon']) pole.refresh_from_db() self.assertAlmostEqual(pole.lon, 4.3600, places=4) self.assertIn('4.36', pole.geojson)