diff --git a/loko/interventions/tests/test_control_assignment.py b/loko/interventions/tests/test_control_assignment.py
new file mode 100644
index 0000000..6ccab79
--- /dev/null
+++ b/loko/interventions/tests/test_control_assignment.py
@@ -0,0 +1,208 @@
+from datetime import date, timedelta
+from django.test import TestCase, RequestFactory
+from django.contrib.auth import get_user_model
+from django.utils import timezone
+from django.contrib.gis.geos import MultiPolygon, Polygon, Point
+
+from common.models import Role, UserConfig, Thematic, UserContractAccess
+from contracts.models import (
+ Company, CompanyTeam, CompanyMember, CompanyMemberHistory, CompanyMemberAbsence, Contract
+)
+from interventions.models import Intervention, InterventionContractPost, ContractPost
+from interventions.permissions import can_view_intervention, can_control_intervention
+from interventions.services_control_assignment import (
+ resolve_control_team_and_member,
+ resolve_control_team_and_member_for_intervention,
+)
+
+
+class ControlAssignmentTests(TestCase):
+ def setUp(self):
+ self.User = get_user_model()
+ self.role_controller, _ = Role.objects.get_or_create(name='controller')
+ self.role_tech, _ = Role.objects.get_or_create(name='technician')
+
+ # Thématiques
+ self.thematic_structures, _ = Thematic.objects.get_or_create(
+ code='structures', defaults={'name_fr': 'Ouvrages d\'art', 'name_nl': 'Kunstwerken'}
+ )
+ self.thematic_roads, _ = Thematic.objects.get_or_create(
+ code='roads', defaults={'name_fr': 'Voirie', 'name_nl': 'Wegen'}
+ )
+
+ # Société & Prestataire
+ self.company = Company.objects.create(name='Prestataire Structures')
+ self.org_intern = Company.objects.create(name='Organisme Interne Contrôle')
+
+ # Équipes opérationnelles terrain
+ self.team_ops = CompanyTeam.objects.create(
+ name='Équipe Opérationnelle',
+ company=self.company,
+ operation_types=[CompanyTeam.OPERATION_INTERVENTIONS]
+ )
+
+ # Équipe de contrôle avec géométrie (SRID 3812)
+ # Bbox approximative autour de Bruxelles en Lambert 72
+ poly = Polygon(((140000, 160000), (160000, 160000), (160000, 180000), (140000, 180000), (140000, 160000)))
+ self.ctrl_geom = MultiPolygon([poly], srid=3812)
+ self.team_ctrl = CompanyTeam.objects.create(
+ name='Équipe Contrôle Structures',
+ company=self.org_intern,
+ operation_types=[CompanyTeam.OPERATION_CONTROLS],
+ geom=self.ctrl_geom
+ )
+
+ # Membre contrôleur
+ self.user_ctrl = self.User.objects.create_user(username='ctrl_agent', password='pwd')
+ self.config_ctrl = UserConfig.objects.create(user=self.user_ctrl, is_intern=True)
+ self.config_ctrl.roles.add(self.role_controller)
+ self.member_ctrl = CompanyMember.objects.create(user=self.user_ctrl, name='Agent Contrôle 1')
+ self.member_ctrl.teams.add(self.team_ctrl)
+
+ # Planification du contrôleur dans l'historique
+ self.today = timezone.now().date()
+ CompanyMemberHistory.objects.create(
+ member=self.member_ctrl,
+ team=self.team_ctrl,
+ join_date=self.today - timedelta(days=30),
+ leave_date=None
+ )
+
+ # Contrat avec attribution automatique du contrôle
+ self.contract_auto = Contract.objects.create(
+ company=self.company,
+ contract_number='C-STRUCT-AUTO',
+ start_date=self.today - timedelta(days=100),
+ end_date=self.today + timedelta(days=100),
+ is_active=True,
+ auto_assign_control_team=True,
+ )
+ self.contract_auto.thematics.add(self.thematic_structures)
+
+ # Contrat SANS attribution automatique
+ self.contract_no_auto = Contract.objects.create(
+ company=self.company,
+ contract_number='C-STRUCT-NOAUTO',
+ start_date=self.today - timedelta(days=100),
+ end_date=self.today + timedelta(days=100),
+ is_active=True,
+ auto_assign_control_team=False,
+ )
+ self.contract_no_auto.thematics.add(self.thematic_structures)
+
+ def test_auto_assignment_structures_corrective(self):
+ """Vérifie l'attribution automatique pour une intervention corrective Structures avec contrat activé."""
+ # Coordonnées au centre de la zone (Bruxelles ~ 4.35, 50.85 -> en Lambert 72 ~ 150000, 170000)
+ p_poly = Polygon(((149000, 169000), (151000, 169000), (151000, 171000), (149000, 171000), (149000, 169000)))
+ geom = MultiPolygon([p_poly], srid=3812)
+
+ itv = Intervention.objects.create(
+ maintain_type='corrective',
+ thematic=self.thematic_structures,
+ contract=self.contract_auto,
+ geom=geom,
+ planned_begin_time=timezone.now(),
+ )
+
+ self.assertEqual(itv.assigned_control_team, self.team_ctrl)
+ self.assertEqual(itv.assigned_controller, self.member_ctrl)
+
+ def test_no_auto_assignment_if_not_corrective(self):
+ """Une intervention préventive ne doit pas recevoir d'attribution automatique de contrôle."""
+ itv = Intervention.objects.create(
+ maintain_type='preventive',
+ thematic=self.thematic_structures,
+ contract=self.contract_auto,
+ planned_begin_time=timezone.now(),
+ )
+ self.assertIsNone(itv.assigned_control_team)
+ self.assertIsNone(itv.assigned_controller)
+
+ def test_no_auto_assignment_if_not_structures(self):
+ """Une intervention d'une autre thématique ne doit pas recevoir d'attribution automatique de contrôle."""
+ itv = Intervention.objects.create(
+ maintain_type='corrective',
+ thematic=self.thematic_roads,
+ contract=self.contract_auto,
+ planned_begin_time=timezone.now(),
+ )
+ self.assertIsNone(itv.assigned_control_team)
+ self.assertIsNone(itv.assigned_controller)
+
+ def test_no_auto_assignment_if_contract_not_enabled(self):
+ """Un contrat sans auto_assign_control_team ne doit pas déclencher l'attribution."""
+ itv = Intervention.objects.create(
+ maintain_type='corrective',
+ thematic=self.thematic_structures,
+ contract=self.contract_no_auto,
+ planned_begin_time=timezone.now(),
+ )
+ self.assertIsNone(itv.assigned_control_team)
+ self.assertIsNone(itv.assigned_controller)
+
+ def test_fallback_calendar_without_geometry(self):
+ """Si aucune géométrie n'est définie, le fallback calendrier doit sélectionner l'équipe candidate."""
+ itv = Intervention.objects.create(
+ maintain_type='corrective',
+ thematic=self.thematic_structures,
+ contract=self.contract_auto,
+ geom=None,
+ planned_begin_time=timezone.now(),
+ )
+ self.assertEqual(itv.assigned_control_team, self.team_ctrl)
+ self.assertEqual(itv.assigned_controller, self.member_ctrl)
+
+ def test_strict_fallback_when_no_member_available(self):
+ """Si le contrôleur est absent à la date cible et aucun autre agent disponible, attribue (None, None)."""
+ target_date = self.today + timedelta(days=5)
+ # Enregistrer une absence
+ CompanyMemberAbsence.objects.create(
+ member=self.member_ctrl,
+ date=target_date,
+ reason='conge'
+ )
+
+ team, member = resolve_control_team_and_member(
+ target_geom=self.ctrl_geom,
+ scheduled_date=target_date,
+ thematic=self.thematic_structures
+ )
+ self.assertIsNone(team)
+ self.assertIsNone(member)
+
+ def test_does_not_overwrite_manual_control_assignment(self):
+ """Ne doit pas écraser une équipe de contrôle déjà assignée manuellement."""
+ team_other = CompanyTeam.objects.create(
+ name='Autre Contrôle',
+ company=self.org_intern,
+ operation_types=[CompanyTeam.OPERATION_CONTROLS]
+ )
+ itv = Intervention.objects.create(
+ maintain_type='corrective',
+ thematic=self.thematic_structures,
+ contract=self.contract_auto,
+ assigned_control_team=team_other,
+ planned_begin_time=timezone.now(),
+ )
+ self.assertEqual(itv.assigned_control_team, team_other)
+
+ def test_controller_permissions_and_control(self):
+ """Le contrôleur assigné doit pouvoir visualiser et contrôler l'intervention au statut processed."""
+ itv = Intervention.objects.create(
+ maintain_type='corrective',
+ thematic=self.thematic_structures,
+ contract=self.contract_auto,
+ status='processed',
+ assigned_control_team=self.team_ctrl,
+ assigned_controller=self.member_ctrl,
+ )
+
+ # Permissions de vue
+ self.assertTrue(can_view_intervention(self.user_ctrl, itv))
+ # Permissions de contrôle (statut processed)
+ self.assertTrue(can_control_intervention(self.user_ctrl, itv))
+
+ # Au statut in_progress, le contrôle n'est pas encore ouvert
+ itv.status = 'in_progress'
+ itv.save(update_fields=['status'])
+ self.assertFalse(can_control_intervention(self.user_ctrl, itv))
diff --git a/loko/interventions/views/detail.py b/loko/interventions/views/detail.py
index 6af23cb..214dc35 100644
--- a/loko/interventions/views/detail.py
+++ b/loko/interventions/views/detail.py
@@ -2480,7 +2480,14 @@ def intervention_posts_bulk_update(request, intervention_id):
# Vérifier les permissions selon le type de modification
can_update_posts = user_config.roles.filter(name__in=['admin', 'external_manager']).exists()
- can_control_posts = user_config.roles.filter(name__in=['admin', 'manager', 'controller']).exists()
+ is_in_control_team = False
+ company_member = CompanyMember.objects.filter(user=request.user).first()
+ if company_member:
+ is_in_control_team = (
+ intervention.assigned_controller == company_member
+ or bool(intervention.assigned_control_team and intervention.assigned_control_team in company_member.teams.all())
+ )
+ can_control_posts = user_config.roles.filter(name__in=['admin', 'manager', 'controller']).exists() or is_in_control_team
updated_count = 0
errors = []
diff --git a/loko/interventions/views/general.py b/loko/interventions/views/general.py
index c214d3d..b10f5f8 100644
--- a/loko/interventions/views/general.py
+++ b/loko/interventions/views/general.py
@@ -414,6 +414,8 @@ def index(request):
'selected_contracts': request.GET.getlist('contract'),
'selected_teams': request.GET.getlist('assigned_team'),
'selected_members': request.GET.getlist('assigned_member'),
+ 'selected_control_teams': request.GET.getlist('assigned_control_team'),
+ 'selected_controllers': request.GET.getlist('assigned_controller'),
'selected_pause_reasons': pause_reasons,
'selected_preparation_states': preparation_states,
'begin_date_min': begin_date_min or '',
@@ -465,6 +467,8 @@ def interventions_geojson(request):
codes = _expand_csv_params(request.GET.getlist('code') + request.GET.getlist('codes'))
assigned_teams = _expand_csv_params(request.GET.getlist('assigned_team'))
assigned_members = _expand_csv_params(request.GET.getlist('assigned_member'))
+ assigned_control_teams = _expand_csv_params(request.GET.getlist('assigned_control_team'))
+ assigned_controllers = _expand_csv_params(request.GET.getlist('assigned_controller'))
pause_reasons = _expand_csv_params(request.GET.getlist('pause_reason'))
intervention_managers = _expand_csv_params(request.GET.getlist('intervention_manager'))
preparation_states = _expand_csv_params(request.GET.getlist('preparation_state'))
@@ -545,6 +549,10 @@ def interventions_geojson(request):
qs = qs.filter(assigned_member_id__in=assigned_members)
if assigned_teams:
qs = qs.filter(Q(assigned_team_id__in=assigned_teams) | Q(assigned_member__teams__in=assigned_teams)).distinct()
+ if assigned_controllers:
+ qs = qs.filter(assigned_controller_id__in=assigned_controllers)
+ if assigned_control_teams:
+ qs = qs.filter(Q(assigned_control_team_id__in=assigned_control_teams) | Q(assigned_controller__teams__in=assigned_control_teams)).distinct()
if intervention_managers:
qs = qs.filter(intervention_manager_id__in=intervention_managers)
if followed_only == '1' and request.user.is_authenticated:
@@ -1007,6 +1015,8 @@ def export_interventions(request):
(_("Fournisseur assigné"), lambda i: (i.assigned_provider.name if i.assigned_provider else '')),
(_("Equipe assignée"), lambda i: (i.assigned_team.name if i.assigned_team else '')),
(_("Membre assigné"), lambda i: (i.assigned_member.name if i.assigned_member else '')),
+ (_("Équipe de contrôle"), lambda i: (i.assigned_control_team.name if i.assigned_control_team else '')),
+ (_("Contrôleur assigné"), lambda i: (i.assigned_controller.name if i.assigned_controller else '')),
(_("Contrat"), lambda i: (i.contract.contract_number if i.contract else '')),
(_("Commande"), lambda i: (i.order.order_code if i.order else '')),
(_("Début prévu"), lambda i: dj_format_date(i.expected_begin_time, 'Y-m-d H:i') if i.expected_begin_time else ''),