feat(assets): ignore auxiliary 'com' wiring and allow archive or delete in cross plan import
- Ignore communication and power supply rows ('com', 'comm', 'alim', etc.) in detector sections
- Add user-selectable 'archive' and 'delete' actions for all assets in cross plan preview
- Handle ProtectedError gracefully when deleting referenced assets
- Add unit tests for wire filtering and archive/delete integration actions
- Update and compile translations (fr, nl, en)
This commit is contained in:
parent
71154a0fb4
commit
1d99cdd09b
10 changed files with 543 additions and 35 deletions
Binary file not shown.
|
|
@ -8573,6 +8573,21 @@ msgstr "GIEP zones"
|
|||
msgid "Aucune localisation GIEP trouvée à proximité"
|
||||
msgstr "No location found"
|
||||
|
||||
msgid "Supprimer de la DB"
|
||||
msgstr "Delete from DB"
|
||||
|
||||
msgid "Supprimer l'existant de la DB"
|
||||
msgstr "Delete existing from DB"
|
||||
|
||||
msgid "Ne pas créer (Ignorer)"
|
||||
msgstr "Do not create (Ignore)"
|
||||
|
||||
msgid "Archiver l'existant (Ne pas recréer)"
|
||||
msgstr "Archive existing (do not recreate)"
|
||||
|
||||
msgid "Archiver l'existant"
|
||||
msgstr "Archive existing"
|
||||
|
||||
#~ msgid "Absent du fichier Excel (Sera conservé ou archivé)"
|
||||
#~ msgstr "Missing from Excel file (Will be kept or archived)"
|
||||
|
||||
|
|
|
|||
Binary file not shown.
|
|
@ -8624,6 +8624,21 @@ msgstr "Zones GIEP"
|
|||
msgid "Aucune localisation GIEP trouvée à proximité"
|
||||
msgstr "Aucune localisation GIEP trouvée à proximité"
|
||||
|
||||
msgid "Supprimer de la DB"
|
||||
msgstr "Supprimer de la DB"
|
||||
|
||||
msgid "Supprimer l'existant de la DB"
|
||||
msgstr "Supprimer l'existant de la DB"
|
||||
|
||||
msgid "Ne pas créer (Ignorer)"
|
||||
msgstr "Ne pas créer (Ignorer)"
|
||||
|
||||
msgid "Archiver l'existant (Ne pas recréer)"
|
||||
msgstr "Archiver l'existant (Ne pas recréer)"
|
||||
|
||||
msgid "Archiver l'existant"
|
||||
msgstr "Archiver l'existant"
|
||||
|
||||
#~ msgid "Absent du fichier Excel (Sera conservé ou archivé)"
|
||||
#~ msgstr "Absent du fichier Excel (Sera conservé ou archivé)"
|
||||
|
||||
|
|
|
|||
Binary file not shown.
|
|
@ -8572,6 +8572,21 @@ msgstr "GIEP-zones"
|
|||
msgid "Aucune localisation GIEP trouvée à proximité"
|
||||
msgstr "Geen locatie gevonden."
|
||||
|
||||
msgid "Supprimer de la DB"
|
||||
msgstr "Verwijderen uit de database"
|
||||
|
||||
msgid "Supprimer l'existant de la DB"
|
||||
msgstr "Bestaande verwijderen uit de database"
|
||||
|
||||
msgid "Ne pas créer (Ignorer)"
|
||||
msgstr "Niet aanmaken (Negeren)"
|
||||
|
||||
msgid "Archiver l'existant (Ne pas recréer)"
|
||||
msgstr "Bestaande archiveren (niet opnieuw aanmaken)"
|
||||
|
||||
msgid "Archiver l'existant"
|
||||
msgstr "Bestaande archiveren"
|
||||
|
||||
#~ msgid "Absent du fichier Excel (Sera conservé ou archivé)"
|
||||
#~ msgstr "Ontbreekt in het Excel-bestand (Wordt bewaard of gearchiveerd)"
|
||||
|
||||
|
|
|
|||
|
|
@ -342,20 +342,25 @@
|
|||
<select name="action_pole_db_{{ item.code }}" class="form-select form-select-sm rounded-2">
|
||||
<option value="keep">{% translate "Conserver en DB" %}</option>
|
||||
<option value="archive" selected>{% translate "Archiver" %}</option>
|
||||
<option value="delete">{% translate "Supprimer de la DB" %}</option>
|
||||
</select>
|
||||
{% else %}
|
||||
<select name="action_pole_{{ item.code }}" class="form-select form-select-sm rounded-2">
|
||||
{% if item.status == 'new' %}
|
||||
<option value="add" selected>{% translate "Créer l'asset" %}</option>
|
||||
<option value="keep">{% translate "Ignorer" %}</option>
|
||||
<option value="keep">{% translate "Ne pas créer (Ignorer)" %}</option>
|
||||
{% elif item.status == 'modified' %}
|
||||
<option value="modify" selected>{% translate "Mettre à jour l'asset" %}</option>
|
||||
<option value="replace">{% translate "Remplacer (Archiver + Créer)" %}</option>
|
||||
<option value="keep">{% translate "Conserver existant" %}</option>
|
||||
<option value="archive">{% translate "Archiver l'existant (Ne pas recréer)" %}</option>
|
||||
<option value="delete">{% translate "Supprimer l'existant de la DB" %}</option>
|
||||
{% elif item.status == 'identical' %}
|
||||
<option value="keep" selected>{% translate "Ne rien faire (Identique)" %}</option>
|
||||
<option value="modify">{% translate "Mettre à jour l'asset" %}</option>
|
||||
<option value="replace">{% translate "Remplacer (Archiver + Créer)" %}</option>
|
||||
<option value="archive">{% translate "Archiver l'existant" %}</option>
|
||||
<option value="delete">{% translate "Supprimer l'existant de la DB" %}</option>
|
||||
{% endif %}
|
||||
</select>
|
||||
{% endif %}
|
||||
|
|
@ -438,19 +443,25 @@
|
|||
<select name="action_lantern_db_{{ item.code }}" class="form-select form-select-sm rounded-2">
|
||||
<option value="keep">{% translate "Conserver en DB" %}</option>
|
||||
<option value="archive" selected>{% translate "Archiver" %}</option>
|
||||
<option value="delete">{% translate "Supprimer de la DB" %}</option>
|
||||
</select>
|
||||
{% else %}
|
||||
<select name="action_lantern_{{ item.code }}" class="form-select form-select-sm rounded-2">
|
||||
{% if item.status == 'new' %}
|
||||
<option value="add" selected>{% translate "Créer l'asset" %}</option>
|
||||
<option value="keep">{% translate "Ignorer" %}</option>
|
||||
<option value="keep">{% translate "Ne pas créer (Ignorer)" %}</option>
|
||||
{% elif item.status == 'modified' %}
|
||||
<option value="modify" selected>{% translate "Mettre à jour l'asset" %}</option>
|
||||
<option value="replace">{% translate "Remplacer (Archiver + Créer)" %}</option>
|
||||
<option value="keep">{% translate "Conserver existant" %}</option>
|
||||
<option value="archive">{% translate "Archiver l'existant (Ne pas recréer)" %}</option>
|
||||
<option value="delete">{% translate "Supprimer l'existant de la DB" %}</option>
|
||||
{% elif item.status == 'identical' %}
|
||||
<option value="keep" selected>{% translate "Ne rien faire (Identique)" %}</option>
|
||||
<option value="modify">{% translate "Mettre à jour l'asset" %}</option>
|
||||
<option value="replace">{% translate "Remplacer (Archiver + Créer)" %}</option>
|
||||
<option value="archive">{% translate "Archiver l'existant" %}</option>
|
||||
<option value="delete">{% translate "Supprimer l'existant de la DB" %}</option>
|
||||
{% endif %}
|
||||
</select>
|
||||
{% endif %}
|
||||
|
|
@ -537,19 +548,25 @@
|
|||
<select name="action_detector_db_{{ item.code }}" class="form-select form-select-sm rounded-2">
|
||||
<option value="keep" {% if item.action_default == 'keep' %}selected{% endif %}>{% translate "Conserver en DB" %}</option>
|
||||
<option value="archive" {% if item.action_default != 'keep' %}selected{% endif %}>{% translate "Archiver" %}</option>
|
||||
<option value="delete">{% translate "Supprimer de la DB" %}</option>
|
||||
</select>
|
||||
{% else %}
|
||||
<select name="action_detector_{{ item.code }}" class="form-select form-select-sm rounded-2">
|
||||
{% if item.status == 'new' %}
|
||||
<option value="add" selected>{% translate "Créer l'asset" %}</option>
|
||||
<option value="keep">{% translate "Ignorer" %}</option>
|
||||
<option value="keep">{% translate "Ne pas créer (Ignorer)" %}</option>
|
||||
{% elif item.status == 'modified' %}
|
||||
<option value="modify" selected>{% translate "Mettre à jour l'asset" %}</option>
|
||||
<option value="replace">{% translate "Remplacer (Archiver + Créer)" %}</option>
|
||||
<option value="keep">{% translate "Conserver existant" %}</option>
|
||||
<option value="archive">{% translate "Archiver l'existant (Ne pas recréer)" %}</option>
|
||||
<option value="delete">{% translate "Supprimer l'existant de la DB" %}</option>
|
||||
{% elif item.status == 'identical' %}
|
||||
<option value="keep" selected>{% translate "Ne rien faire (Identique)" %}</option>
|
||||
<option value="modify">{% translate "Mettre à jour l'asset" %}</option>
|
||||
<option value="replace">{% translate "Remplacer (Archiver + Créer)" %}</option>
|
||||
<option value="archive">{% translate "Archiver l'existant" %}</option>
|
||||
<option value="delete">{% translate "Supprimer l'existant de la DB" %}</option>
|
||||
{% endif %}
|
||||
</select>
|
||||
{% endif %}
|
||||
|
|
@ -652,20 +669,25 @@
|
|||
<select name="action_cable_db_{{ item.code }}" class="form-select form-select-sm rounded-2">
|
||||
<option value="keep">{% translate "Conserver en DB" %}</option>
|
||||
<option value="archive" selected>{% translate "Archiver" %}</option>
|
||||
<option value="delete">{% translate "Supprimer de la DB" %}</option>
|
||||
</select>
|
||||
{% else %}
|
||||
<select name="action_cable_{{ item.code }}" class="form-select form-select-sm rounded-2">
|
||||
{% if item.status == 'new' %}
|
||||
<option value="add" selected>{% translate "Créer l'asset" %}</option>
|
||||
<option value="keep">{% translate "Ignorer" %}</option>
|
||||
<option value="keep">{% translate "Ne pas créer (Ignorer)" %}</option>
|
||||
{% elif item.status == 'modified' %}
|
||||
<option value="modify" selected>{% translate "Mettre à jour l'asset" %}</option>
|
||||
<option value="replace">{% translate "Remplacer (Archiver + Créer)" %}</option>
|
||||
<option value="keep">{% translate "Conserver existant" %}</option>
|
||||
<option value="archive">{% translate "Archiver l'existant (Ne pas recréer)" %}</option>
|
||||
<option value="delete">{% translate "Supprimer l'existant de la DB" %}</option>
|
||||
{% elif item.status == 'identical' %}
|
||||
<option value="keep" selected>{% translate "Ne rien faire (Identique)" %}</option>
|
||||
<option value="modify">{% translate "Mettre à jour l'asset" %}</option>
|
||||
<option value="replace">{% translate "Remplacer (Archiver + Créer)" %}</option>
|
||||
<option value="archive">{% translate "Archiver l'existant" %}</option>
|
||||
<option value="delete">{% translate "Supprimer l'existant de la DB" %}</option>
|
||||
{% endif %}
|
||||
</select>
|
||||
{% endif %}
|
||||
|
|
@ -753,6 +775,7 @@
|
|||
<select name="action_other_{{ item.unique_key }}" class="form-select form-select-sm rounded-2">
|
||||
<option value="keep" selected>{% translate "Conserver (Par défaut)" %}</option>
|
||||
<option value="archive">{% translate "Archiver" %}</option>
|
||||
<option value="delete">{% translate "Supprimer de la DB" %}</option>
|
||||
</select>
|
||||
</td>
|
||||
</tr>
|
||||
|
|
|
|||
|
|
@ -3025,6 +3025,308 @@ class TrafficLightCrossPlanImportTest(TestCase):
|
|||
if os.path.exists(tmp_path):
|
||||
os.remove(tmp_path)
|
||||
|
||||
def test_parse_cross_plan_com_and_alim_wires_ignored_in_detection_sections(self):
|
||||
"""
|
||||
Verify that auxiliary wiring rows like 'Com', 'Comm', 'Alim', 'Contact', 'Radar'
|
||||
in detection cable / Detecteur sections are not parsed as detectors or lanterns.
|
||||
This fixes the bug where 'Com' lines were erroneously created as 'Traficam wide angle' detectors.
|
||||
"""
|
||||
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 48 TWAVB 10x2x0,8 mm²")
|
||||
ws.cell(row=6, column=3, value="Detecteur")
|
||||
ws.cell(row=6, column=4, value="kringen")
|
||||
ws.cell(row=7, column=5, value="A01")
|
||||
ws.cell(row=7, column=6, value="A05")
|
||||
|
||||
# Row 8: Com on A01
|
||||
ws.cell(row=8, column=2, value="1")
|
||||
ws.cell(row=8, column=3, value="Com")
|
||||
ws.cell(row=8, column=4, value="COM")
|
||||
ws.cell(row=8, column=5, value="X")
|
||||
|
||||
# Row 9: RpC on A01
|
||||
ws.cell(row=9, column=2, value="2")
|
||||
ws.cell(row=9, column=3, value="RpC")
|
||||
ws.cell(row=9, column=4, value="TMA-122 M")
|
||||
ws.cell(row=9, column=5, value="X")
|
||||
|
||||
# Row 10: Alim on A05
|
||||
ws.cell(row=10, column=2, value="3")
|
||||
ws.cell(row=10, column=3, value="Alim")
|
||||
ws.cell(row=10, column=4, value="24V")
|
||||
ws.cell(row=10, column=6, value="X")
|
||||
|
||||
# Row 11: RpB on A05
|
||||
ws.cell(row=11, column=2, value="4")
|
||||
ws.cell(row=11, column=3, value="RpB")
|
||||
ws.cell(row=11, column=4, value="TMA-122 M")
|
||||
ws.cell(row=11, column=6, 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"]
|
||||
detector_codes = [d["code"] for d in detectors]
|
||||
|
||||
# Only RpC on A01 and RpB on A05 should be recognized
|
||||
self.assertEqual(len(detectors), 2)
|
||||
self.assertIn("SWB01_A01_RPC", detector_codes)
|
||||
self.assertIn("SWB01_A05_RPB", detector_codes)
|
||||
|
||||
# Neither Com nor Alim should be in detectors
|
||||
self.assertNotIn("SWB01_A01_COM", detector_codes)
|
||||
self.assertNotIn("SWB01_A05_ALIM", detector_codes)
|
||||
self.assertFalse(any("COM" in code for code in detector_codes))
|
||||
self.assertFalse(any("ALIM" in code for code in detector_codes))
|
||||
self.assertEqual(len(deduced["lanterns"]), 0)
|
||||
finally:
|
||||
import os
|
||||
if os.path.exists(tmp_path):
|
||||
os.remove(tmp_path)
|
||||
|
||||
def test_integrate_cross_plan_archive_and_delete_actions(self):
|
||||
"""
|
||||
Verify that users can choose 'archive' or 'delete' for assets during cross plan integration:
|
||||
- When 'archive' is selected, the asset's status becomes 'archived'.
|
||||
- When 'delete' is selected, the asset is removed from the database.
|
||||
"""
|
||||
import openpyxl
|
||||
import tempfile
|
||||
from assets.models import (
|
||||
TrafficLightPole, TrafficLightLantern, TrafficLightDetector,
|
||||
TrafficLightCable, TrafficLightPoleModel, TrafficLightLanternModel,
|
||||
TrafficLightDetectorModel, TrafficLightCableModel
|
||||
)
|
||||
from assets.models.sign import SignPanel, SignPanelModel
|
||||
|
||||
pole_model = TrafficLightPoleModel.objects.first()
|
||||
lantern_model = TrafficLightLanternModel.objects.first()
|
||||
detector_model = TrafficLightDetectorModel.objects.first()
|
||||
cable_model = TrafficLightCableModel.objects.first()
|
||||
sign_model, _ = SignPanelModel.objects.get_or_create(code="sp_test_mod", defaults={"name_fr": "Model SP Test"})
|
||||
|
||||
# Assets to delete
|
||||
pole_del = TrafficLightPole.objects.create(intersection=self.intersection, code="SWB01_P_DEL", model=pole_model, status="active")
|
||||
lan_del = TrafficLightLantern.objects.create(pole=pole_del, code="SWB01_L_DEL", model=lantern_model, status="active")
|
||||
det_del = TrafficLightDetector.objects.create(intersection=self.intersection, code="SWB01_D_DEL", model=detector_model, status="active")
|
||||
cab_del = TrafficLightCable.objects.create(intersection=self.intersection, code="SWB01_C_DEL", model=cable_model, status="active")
|
||||
sp_del = SignPanel.objects.create(trafficlight_pole=pole_del, code="SWB01_SP_DEL", model=sign_model, status="active")
|
||||
pole_standalone_del = TrafficLightPole.objects.create(intersection=self.intersection, code="SWB01_P_STANDALONE_DEL", model=pole_model, status="active")
|
||||
|
||||
# Assets to archive
|
||||
pole_arc = TrafficLightPole.objects.create(intersection=self.intersection, code="SWB01_P_ARC", model=pole_model, status="active")
|
||||
lan_arc = TrafficLightLantern.objects.create(pole=pole_arc, code="SWB01_L_ARC", model=lantern_model, status="active")
|
||||
det_arc = TrafficLightDetector.objects.create(intersection=self.intersection, pole=pole_arc, code="SWB01_D_ARC", model=detector_model, status="active")
|
||||
cab_arc = TrafficLightCable.objects.create(intersection=self.intersection, code="SWB01_C_ARC", model=cable_model, status="active")
|
||||
sp_arc = SignPanel.objects.create(trafficlight_pole=pole_arc, code="SWB01_SP_ARC", model=sign_model, status="active")
|
||||
|
||||
# Minimal workbook so we can run import_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 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="P_ARC")
|
||||
ws.cell(row=8, column=3, value="A")
|
||||
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_test_archive_delete.xlsx", f.read(), content_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet")
|
||||
|
||||
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)
|
||||
|
||||
# Test POST integrate with delete on db-only assets and archive on others
|
||||
url_integrate = reverse("assets:integrate_cross_plan", args=[self.intersection.id])
|
||||
post_data = {
|
||||
"cross_plan_filename": response.context["cross_plan_filename"],
|
||||
# Delete DB-only standalone pole
|
||||
"selected_pole_db_SWB01_P_STANDALONE_DEL": "on",
|
||||
"action_pole_db_SWB01_P_STANDALONE_DEL": "delete",
|
||||
"db_id_pole_SWB01_P_STANDALONE_DEL": str(pole_standalone_del.id),
|
||||
# Delete DB-only lantern
|
||||
"selected_lantern_db_SWB01_L_DEL": "on",
|
||||
"action_lantern_db_SWB01_L_DEL": "delete",
|
||||
"db_id_lantern_SWB01_L_DEL": str(lan_del.id),
|
||||
# Delete DB-only detector
|
||||
"selected_detector_db_SWB01_D_DEL": "on",
|
||||
"action_detector_db_SWB01_D_DEL": "delete",
|
||||
"db_id_detector_SWB01_D_DEL": str(det_del.id),
|
||||
# Delete DB-only cable
|
||||
"selected_cable_db_SWB01_C_DEL": "on",
|
||||
"action_cable_db_SWB01_C_DEL": "delete",
|
||||
"db_id_cable_SWB01_C_DEL": str(cab_del.id),
|
||||
# Delete other asset (signpanel)
|
||||
f"selected_other_signpanel_{sp_del.id}": "on",
|
||||
f"action_other_signpanel_{sp_del.id}": "delete",
|
||||
|
||||
# Archive DB-only pole
|
||||
"selected_pole_db_SWB01_P_ARC": "on",
|
||||
"action_pole_db_SWB01_P_ARC": "archive",
|
||||
"db_id_pole_SWB01_P_ARC": str(pole_arc.id),
|
||||
# Archive DB-only lantern
|
||||
"selected_lantern_db_SWB01_L_ARC": "on",
|
||||
"action_lantern_db_SWB01_L_ARC": "archive",
|
||||
"db_id_lantern_SWB01_L_ARC": str(lan_arc.id),
|
||||
# Archive DB-only detector
|
||||
"selected_detector_db_SWB01_D_ARC": "on",
|
||||
"action_detector_db_SWB01_D_ARC": "archive",
|
||||
"db_id_detector_SWB01_D_ARC": str(det_arc.id),
|
||||
# Archive DB-only cable
|
||||
"selected_cable_db_SWB01_C_ARC": "on",
|
||||
"action_cable_db_SWB01_C_ARC": "archive",
|
||||
"db_id_cable_SWB01_C_ARC": str(cab_arc.id),
|
||||
# Archive other asset (signpanel)
|
||||
f"selected_other_signpanel_{sp_arc.id}": "on",
|
||||
f"action_other_signpanel_{sp_arc.id}": "archive",
|
||||
}
|
||||
response_integrate = self.client.post(url_integrate, post_data)
|
||||
self.assertEqual(response_integrate.status_code, 302)
|
||||
|
||||
# Verification of deleted items
|
||||
self.assertFalse(TrafficLightPole.objects.filter(pk=pole_standalone_del.id).exists())
|
||||
self.assertFalse(TrafficLightLantern.objects.filter(pk=lan_del.id).exists())
|
||||
self.assertFalse(TrafficLightDetector.objects.filter(pk=det_del.id).exists())
|
||||
self.assertFalse(TrafficLightCable.objects.filter(pk=cab_del.id).exists())
|
||||
self.assertFalse(SignPanel.objects.filter(pk=sp_del.id).exists())
|
||||
|
||||
# Verification of archived items
|
||||
pole_arc.refresh_from_db()
|
||||
self.assertEqual(pole_arc.status, "archived")
|
||||
lan_arc.refresh_from_db()
|
||||
self.assertEqual(lan_arc.status, "archived")
|
||||
det_arc.refresh_from_db()
|
||||
self.assertEqual(det_arc.status, "archived")
|
||||
cab_arc.refresh_from_db()
|
||||
self.assertEqual(cab_arc.status, "archived")
|
||||
sp_arc.refresh_from_db()
|
||||
self.assertEqual(sp_arc.status, "archived")
|
||||
|
||||
finally:
|
||||
import os
|
||||
if os.path.exists(tmp_path):
|
||||
os.remove(tmp_path)
|
||||
|
||||
def test_integrate_cross_plan_plan_asset_archive_and_delete_actions(self):
|
||||
"""
|
||||
Verify that for assets that exist both in DB and in the plan (status identical or modified),
|
||||
the user can select 'archive' (archiver l'existant sans recréer) or 'delete' (supprimer l'existant de la DB).
|
||||
"""
|
||||
import openpyxl
|
||||
import tempfile
|
||||
from assets.models import (
|
||||
TrafficLightPole, TrafficLightLantern, TrafficLightDetector,
|
||||
TrafficLightCable, TrafficLightPoleModel, TrafficLightLanternModel,
|
||||
TrafficLightDetectorModel, TrafficLightCableModel
|
||||
)
|
||||
from django.core.files.uploadedfile import SimpleUploadedFile
|
||||
|
||||
pole_model = TrafficLightPoleModel.objects.first()
|
||||
lantern_model = TrafficLightLanternModel.objects.first()
|
||||
detector_model = TrafficLightDetectorModel.objects.first()
|
||||
cable_model = TrafficLightCableModel.objects.first()
|
||||
|
||||
existing_pole = TrafficLightPole.objects.create(intersection=self.intersection, code="SWB01_A01", model=pole_model, status="active")
|
||||
existing_lantern = TrafficLightLantern.objects.create(pole=existing_pole, code="SWB01_A01_LAN01_A", model=lantern_model, status="active")
|
||||
existing_detector = TrafficLightDetector.objects.create(intersection=self.intersection, code="SWB01_A01_RPC", model=detector_model, status="active")
|
||||
existing_cable = TrafficLightCable.objects.create(intersection=self.intersection, code="SWB01_CAB01", model=cable_model, status="active")
|
||||
|
||||
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="A01")
|
||||
ws.cell(row=8, column=3, value="A")
|
||||
ws.cell(row=8, column=5, value="X")
|
||||
ws.cell(row=9, column=3, value="RpC")
|
||||
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:
|
||||
self.client.force_login(self.user)
|
||||
with open(tmp_path, "rb") as f:
|
||||
uploaded_file = SimpleUploadedFile("plan_test_plan_actions.xlsx", f.read(), content_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet")
|
||||
|
||||
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)
|
||||
|
||||
# Post integrate:
|
||||
# - detector: action="delete"
|
||||
# - lantern: action="archive"
|
||||
# - cable: action="delete"
|
||||
# - pole: action="archive"
|
||||
url_integrate = reverse("assets:integrate_cross_plan", args=[self.intersection.id])
|
||||
post_data = {
|
||||
"cross_plan_filename": response.context["cross_plan_filename"],
|
||||
"selected_detector_SWB01_A01_RPC": "on",
|
||||
"action_detector_SWB01_A01_RPC": "delete",
|
||||
"db_id_detector_SWB01_A01_RPC": str(existing_detector.id),
|
||||
|
||||
"selected_lantern_SWB01_A01_LAN01_A": "on",
|
||||
"action_lantern_SWB01_A01_LAN01_A": "archive",
|
||||
"db_id_lantern_SWB01_A01_LAN01_A": str(existing_lantern.id),
|
||||
|
||||
"selected_cable_SWB01_CAB01": "on",
|
||||
"action_cable_SWB01_CAB01": "delete",
|
||||
"db_id_cable_SWB01_CAB01": str(existing_cable.id),
|
||||
|
||||
"selected_pole_SWB01_A01": "on",
|
||||
"action_pole_SWB01_A01": "archive",
|
||||
"db_id_pole_SWB01_A01": str(existing_pole.id),
|
||||
}
|
||||
response_integrate = self.client.post(url_integrate, post_data)
|
||||
self.assertEqual(response_integrate.status_code, 302)
|
||||
|
||||
# Detector must be deleted
|
||||
self.assertFalse(TrafficLightDetector.objects.filter(pk=existing_detector.id).exists())
|
||||
# Cable must be deleted
|
||||
self.assertFalse(TrafficLightCable.objects.filter(pk=existing_cable.id).exists())
|
||||
|
||||
# Lantern must be archived
|
||||
existing_lantern.refresh_from_db()
|
||||
self.assertEqual(existing_lantern.status, "archived")
|
||||
|
||||
# Pole must be archived
|
||||
existing_pole.refresh_from_db()
|
||||
self.assertEqual(existing_pole.status, "archived")
|
||||
|
||||
finally:
|
||||
import os
|
||||
if os.path.exists(tmp_path):
|
||||
os.remove(tmp_path)
|
||||
|
||||
|
||||
|
||||
|
||||
class AssetDefaultPositionTest(TestCase):
|
||||
def test_assign_default_position_hierarchy(self):
|
||||
|
|
|
|||
|
|
@ -26,6 +26,38 @@ def is_hp_or_ls(text):
|
|||
return True
|
||||
return False
|
||||
|
||||
def is_wiring_or_non_equipment(text):
|
||||
"""
|
||||
Checks if the wire/direction/circuit text refers to auxiliary wiring, communication,
|
||||
power supply, contact lines, or reserve wires that must NOT be treated as equipment
|
||||
(neither lantern nor detector).
|
||||
In cross plans, 'Com' / 'Comm' / 'Communication' represents communication wiring/bus
|
||||
for detectors/cameras or radar power supplies, NOT an actual detector.
|
||||
"""
|
||||
if not text:
|
||||
return False
|
||||
t_str = str(text).strip()
|
||||
t_lower = t_str.lower()
|
||||
t_clean = re.sub(r'[\s/\\_\-\.:;]+', ' ', t_lower).strip()
|
||||
t_upper = re.sub(r'\s+', '', t_str.upper())
|
||||
|
||||
# 1. Direct keyword matches (word boundary)
|
||||
if re.search(r'\b(?:com|comm|communication|communicatie|alim|alimentation|voeding|tension|contact|reserve|reservee|réservé|réservée|réserve|libre|vrij|spaar|terre|masse|neutre)\b', t_clean, re.IGNORECASE):
|
||||
return True
|
||||
|
||||
# 2. String starts with or equals COM, COMM, ALIM, etc.
|
||||
if re.match(r'^(?:COM|COMM|ALIM|CONTACT|TENSION|RESERVE)\d*$', t_upper, re.IGNORECASE):
|
||||
return True
|
||||
|
||||
if t_upper.startswith(("COM_", "COMM_", "ALIM_", "COM-", "COMM-", "ALIM-")):
|
||||
return True
|
||||
|
||||
# 3. Radar power supply (not a detector code like RA, RB, RC, RD, RM, RP)
|
||||
if t_upper == "RADAR" or t_upper.startswith("RADAR") or (bool(re.search(r'\bradar\b', t_str, re.IGNORECASE)) and not bool(re.match(r'^(?:RA|RB|RC|RD|RM|RP)\d*$', t_upper, re.IGNORECASE))):
|
||||
return True
|
||||
|
||||
return False
|
||||
|
||||
def is_cyclopieton_phase(phase):
|
||||
"""
|
||||
Checks if a phase code corresponds to a combined pedestrian + cyclist signal (cyclo-piéton).
|
||||
|
|
@ -283,6 +315,11 @@ def parse_cross_plan(excel_path):
|
|||
pole_has_hp[p] = True
|
||||
continue
|
||||
|
||||
# Check if wire refers to communication, power supply, contact, reserve, etc.
|
||||
# It must NOT be treated as an equipment (neither lantern nor detector).
|
||||
if is_wiring_or_non_equipment(richting) or is_wiring_or_non_equipment(kringen):
|
||||
continue
|
||||
|
||||
# Detectors
|
||||
# Check if push button (e.g. Dka, Dkb, PBa1, BPf4...)
|
||||
if richting.lower().startswith(("dk", "pb", "bp")):
|
||||
|
|
@ -383,7 +420,7 @@ def parse_cross_plan(excel_path):
|
|||
# Richting codes are typically: A, B, C, T1, T2, F1, F2, a, b, c...
|
||||
# Ignore wiring/cabling categories like ALIM, Radar, Com, contact, HP/LS, etc.
|
||||
normalized = re.sub(r'\s+', ' ', richting).strip()
|
||||
if is_hp_or_ls(richting) or is_hp_or_ls(kringen) or re.search(r'(?:alim|com|contact|radar|tension|comm|\bhp\b|\bls\b|hp/ls|ls/hp)', normalized, re.IGNORECASE):
|
||||
if is_hp_or_ls(richting) or is_hp_or_ls(kringen) or is_wiring_or_non_equipment(richting) or is_wiring_or_non_equipment(kringen) or re.search(r'(?:alim|com|contact|radar|tension|comm|\bhp\b|\bls\b|hp/ls|ls/hp)', normalized, re.IGNORECASE):
|
||||
continue
|
||||
|
||||
if p not in pole_phases:
|
||||
|
|
|
|||
|
|
@ -4867,12 +4867,18 @@ def integrate_cross_plan(request, intersection_id):
|
|||
from django.shortcuts import get_object_or_404, redirect
|
||||
from django.contrib import messages
|
||||
from django.db import transaction
|
||||
import json
|
||||
from django.db.models import ProtectedError
|
||||
from assets.models.trafficlights import (
|
||||
TrafficLightIntersection, TrafficLightPole, TrafficLightLantern,
|
||||
TrafficLightDetector, TrafficLightCable, TrafficLightPoleModel,
|
||||
TrafficLightLanternModel, TrafficLightDetectorModel, TrafficLightCableModel
|
||||
TrafficLightLanternModel, TrafficLightDetectorModel, TrafficLightCableModel,
|
||||
TrafficLightController, TrafficLightControllerHardware, TrafficLightNetworkHardware,
|
||||
TrafficLightVoltageHardware, TrafficLightElectricalCabinet, TrafficLightAccessory,
|
||||
TrafficLightRadar, TrafficLightProgramming
|
||||
)
|
||||
from assets.models.sign import SignPanel
|
||||
from assets.models.its import ITSGeoAsset
|
||||
from assets.models.publiclighting import PublicLightingLightFixture
|
||||
|
||||
intersection = get_object_or_404(TrafficLightIntersection, pk=intersection_id)
|
||||
filename = request.POST.get('cross_plan_filename')
|
||||
|
|
@ -5406,6 +5412,18 @@ def integrate_cross_plan(request, intersection_id):
|
|||
PublicLightingLightFixture.objects.filter(traffic_light_pole=existing).update(traffic_light_pole=new_pole)
|
||||
except Exception:
|
||||
pass
|
||||
elif action == "archive" and existing:
|
||||
existing.status = 'archived'
|
||||
existing.updated_by = request.user
|
||||
existing.save()
|
||||
elif action == "delete" and existing:
|
||||
try:
|
||||
existing.delete()
|
||||
except ProtectedError:
|
||||
existing.status = 'archived'
|
||||
existing.updated_by = request.user
|
||||
existing.save()
|
||||
messages.warning(request, f"Le poteau {existing.code} n'a pas pu être supprimé car d'autres éléments y font référence ; il a été archivé à la place.")
|
||||
else:
|
||||
# 'keep' (Ne rien faire)
|
||||
if existing:
|
||||
|
|
@ -5424,6 +5442,17 @@ def integrate_cross_plan(request, intersection_id):
|
|||
TrafficLightPole.objects.filter(intersection=intersection, code=code, status='active').update(
|
||||
status='archived', updated_by=request.user
|
||||
)
|
||||
elif action == "delete":
|
||||
db_id = request.POST.get(f"db_id_pole_{code}")
|
||||
qs = TrafficLightPole.objects.filter(pk=db_id) if db_id else TrafficLightPole.objects.filter(intersection=intersection, code=code)
|
||||
for pole_to_del in qs:
|
||||
try:
|
||||
pole_to_del.delete()
|
||||
except ProtectedError:
|
||||
pole_to_del.status = 'archived'
|
||||
pole_to_del.updated_by = request.user
|
||||
pole_to_del.save()
|
||||
messages.warning(request, f"Le poteau {pole_to_del.code} n'a pas pu être supprimé car d'autres éléments y font référence ; il a été archivé à la place.")
|
||||
|
||||
# Fetch active poles to match parent relation for lanterns/detectors if not resolved above
|
||||
active_poles = {p.code: p for p in TrafficLightPole.objects.filter(intersection=intersection, status='active')}
|
||||
|
|
@ -5492,6 +5521,18 @@ def integrate_cross_plan(request, intersection_id):
|
|||
new_cable.save()
|
||||
existing.replaced_by = new_cable
|
||||
existing.save()
|
||||
elif action == "archive" and existing:
|
||||
existing.status = 'archived'
|
||||
existing.updated_by = request.user
|
||||
existing.save()
|
||||
elif action == "delete" and existing:
|
||||
try:
|
||||
existing.delete()
|
||||
except ProtectedError:
|
||||
existing.status = 'archived'
|
||||
existing.updated_by = request.user
|
||||
existing.save()
|
||||
messages.warning(request, f"Le câble {existing.code} n'a pas pu être supprimé car il est référencé ; il a été archivé à la place.")
|
||||
else:
|
||||
# 'keep' (Ne rien faire)
|
||||
if existing:
|
||||
|
|
@ -5509,6 +5550,17 @@ def integrate_cross_plan(request, intersection_id):
|
|||
TrafficLightCable.objects.filter(intersection=intersection, code=code, status='active').update(
|
||||
status='archived', updated_by=request.user
|
||||
)
|
||||
elif action == "delete":
|
||||
db_id = request.POST.get(f"db_id_cable_{code}")
|
||||
qs = TrafficLightCable.objects.filter(pk=db_id) if db_id else TrafficLightCable.objects.filter(intersection=intersection, code=code)
|
||||
for cable_to_del in qs:
|
||||
try:
|
||||
cable_to_del.delete()
|
||||
except ProtectedError:
|
||||
cable_to_del.status = 'archived'
|
||||
cable_to_del.updated_by = request.user
|
||||
cable_to_del.save()
|
||||
messages.warning(request, f"Le câble {cable_to_del.code} n'a pas pu être supprimé car il est référencé ; il a été archivé à la place.")
|
||||
|
||||
# --- 3. Lanterns integration ---
|
||||
for dl in deduced["lanterns"]:
|
||||
|
|
@ -5570,6 +5622,18 @@ def integrate_cross_plan(request, intersection_id):
|
|||
)
|
||||
existing.replaced_by = new_lantern
|
||||
existing.save()
|
||||
elif action == "archive" and existing:
|
||||
existing.status = 'archived'
|
||||
existing.updated_by = request.user
|
||||
existing.save()
|
||||
elif action == "delete" and existing:
|
||||
try:
|
||||
existing.delete()
|
||||
except ProtectedError:
|
||||
existing.status = 'archived'
|
||||
existing.updated_by = request.user
|
||||
existing.save()
|
||||
messages.warning(request, f"La lanterne {existing.code} n'a pas pu être supprimée car elle est référencée ; elle a été archivée à la place.")
|
||||
else:
|
||||
# 'keep'
|
||||
if existing:
|
||||
|
|
@ -5589,6 +5653,17 @@ def integrate_cross_plan(request, intersection_id):
|
|||
TrafficLightLantern.objects.filter(pole__intersection=intersection, code=code, status='active').update(
|
||||
status='archived', updated_by=request.user
|
||||
)
|
||||
elif action == "delete":
|
||||
db_id = request.POST.get(f"db_id_lantern_{code}")
|
||||
qs = TrafficLightLantern.objects.filter(pk=db_id) if db_id else TrafficLightLantern.objects.filter(pole__intersection=intersection, code=code)
|
||||
for lan_to_del in qs:
|
||||
try:
|
||||
lan_to_del.delete()
|
||||
except ProtectedError:
|
||||
lan_to_del.status = 'archived'
|
||||
lan_to_del.updated_by = request.user
|
||||
lan_to_del.save()
|
||||
messages.warning(request, f"La lanterne {lan_to_del.code} n'a pas pu être supprimée car elle est référencée ; elle a été archivée à la place.")
|
||||
|
||||
# --- 4. Detectors integration ---
|
||||
for dd in deduced["detectors"]:
|
||||
|
|
@ -5658,6 +5733,18 @@ def integrate_cross_plan(request, intersection_id):
|
|||
)
|
||||
existing.replaced_by = new_det
|
||||
existing.save()
|
||||
elif action == "archive" and existing:
|
||||
existing.status = 'archived'
|
||||
existing.updated_by = request.user
|
||||
existing.save()
|
||||
elif action == "delete" and existing:
|
||||
try:
|
||||
existing.delete()
|
||||
except ProtectedError:
|
||||
existing.status = 'archived'
|
||||
existing.updated_by = request.user
|
||||
existing.save()
|
||||
messages.warning(request, f"Le détecteur {existing.code} n'a pas pu être supprimé car il est référencé ; il a été archivé à la place.")
|
||||
else:
|
||||
# 'keep'
|
||||
if existing:
|
||||
|
|
@ -5678,6 +5765,17 @@ def integrate_cross_plan(request, intersection_id):
|
|||
TrafficLightDetector.objects.filter(intersection=intersection, code=code, status='active').update(
|
||||
status='archived', updated_by=request.user
|
||||
)
|
||||
elif action == "delete":
|
||||
db_id = request.POST.get(f"db_id_detector_{code}")
|
||||
qs = TrafficLightDetector.objects.filter(pk=db_id) if db_id else TrafficLightDetector.objects.filter(intersection=intersection, code=code)
|
||||
for det_to_del in qs:
|
||||
try:
|
||||
det_to_del.delete()
|
||||
except ProtectedError:
|
||||
det_to_del.status = 'archived'
|
||||
det_to_del.updated_by = request.user
|
||||
det_to_del.save()
|
||||
messages.warning(request, f"Le détecteur {det_to_del.code} n'a pas pu être supprimé car il est référencé ; il a été archivé à la place.")
|
||||
|
||||
# --- 5. Other associated assets integration ---
|
||||
for key in request.POST.keys():
|
||||
|
|
@ -5687,36 +5785,39 @@ def integrate_cross_plan(request, intersection_id):
|
|||
asset_type, asset_id_str = suffix.split("_", 1)
|
||||
is_selected = request.POST.get(f"selected_other_{suffix}") == "on"
|
||||
action = request.POST.get(key, "keep")
|
||||
if is_selected and action == "archive":
|
||||
if is_selected and action in ("archive", "delete"):
|
||||
try:
|
||||
asset_id = int(asset_id_str)
|
||||
if asset_type == "signpanel":
|
||||
from assets.models.sign import SignPanel
|
||||
SignPanel.objects.filter(pk=asset_id).update(status='archived', updated_by=request.user)
|
||||
elif asset_type == "itsasset":
|
||||
from assets.models.its import ITSGeoAsset
|
||||
ITSGeoAsset.objects.filter(pk=asset_id).update(status='archived', updated_by=request.user)
|
||||
elif asset_type == "controller":
|
||||
TrafficLightController.objects.filter(pk=asset_id).update(status='archived', updated_by=request.user)
|
||||
from assets.models.trafficlights import TrafficLightControllerHardware
|
||||
model_map = {
|
||||
"signpanel": SignPanel,
|
||||
"itsasset": ITSGeoAsset,
|
||||
"controller": TrafficLightController,
|
||||
"networkhardware": TrafficLightNetworkHardware,
|
||||
"voltagehardware": TrafficLightVoltageHardware,
|
||||
"electricalcabinet": TrafficLightElectricalCabinet,
|
||||
"accessory": TrafficLightAccessory,
|
||||
"radar": TrafficLightRadar,
|
||||
"programming": TrafficLightProgramming,
|
||||
"lightfixture": PublicLightingLightFixture,
|
||||
}
|
||||
model_cls = model_map.get(asset_type)
|
||||
if model_cls:
|
||||
if action == "archive":
|
||||
model_cls.objects.filter(pk=asset_id).update(status='archived', updated_by=request.user)
|
||||
if asset_type == "controller":
|
||||
TrafficLightControllerHardware.objects.filter(controller_id=asset_id).update(status='archived', updated_by=request.user)
|
||||
TrafficLightNetworkHardware.objects.filter(controller_id=asset_id).update(status='archived', updated_by=request.user)
|
||||
TrafficLightVoltageHardware.objects.filter(controller_id=asset_id).update(status='archived', updated_by=request.user)
|
||||
elif asset_type == "networkhardware":
|
||||
TrafficLightNetworkHardware.objects.filter(pk=asset_id).update(status='archived', updated_by=request.user)
|
||||
elif asset_type == "voltagehardware":
|
||||
TrafficLightVoltageHardware.objects.filter(pk=asset_id).update(status='archived', updated_by=request.user)
|
||||
elif asset_type == "electricalcabinet":
|
||||
TrafficLightElectricalCabinet.objects.filter(pk=asset_id).update(status='archived', updated_by=request.user)
|
||||
elif asset_type == "accessory":
|
||||
TrafficLightAccessory.objects.filter(pk=asset_id).update(status='archived', updated_by=request.user)
|
||||
elif asset_type == "radar":
|
||||
TrafficLightRadar.objects.filter(pk=asset_id).update(status='archived', updated_by=request.user)
|
||||
elif asset_type == "programming":
|
||||
TrafficLightProgramming.objects.filter(pk=asset_id).update(status='archived', updated_by=request.user)
|
||||
elif asset_type == "lightfixture":
|
||||
from assets.models.publiclighting import PublicLightingLightFixture
|
||||
PublicLightingLightFixture.objects.filter(pk=asset_id).update(status='archived', updated_by=request.user)
|
||||
elif action == "delete":
|
||||
obj = model_cls.objects.filter(pk=asset_id).first()
|
||||
if obj:
|
||||
try:
|
||||
obj.delete()
|
||||
except ProtectedError:
|
||||
obj.status = 'archived'
|
||||
obj.updated_by = request.user
|
||||
obj.save()
|
||||
messages.warning(request, f"L'équipement {obj} n'a pas pu être supprimé car il est référencé ; il a été archivé à la place.")
|
||||
except (ValueError, TypeError, Exception):
|
||||
pass
|
||||
|
||||
|
|
|
|||
Loading…
Reference in a new issue