diff --git a/loko/interventions/tests/test_timeline_workflow.py b/loko/interventions/tests/test_timeline_workflow.py index 6639455..3768f36 100644 --- a/loko/interventions/tests/test_timeline_workflow.py +++ b/loko/interventions/tests/test_timeline_workflow.py @@ -689,5 +689,190 @@ class AutoCloseInterventionTests(TestCase): self.assertEqual(itv_recent.status, 'finished') +class ProviderEmailNotificationTests(TestCase): + def setUp(self): + User = get_user_model() + self.role_admin, _ = Role.objects.get_or_create(name='admin') + self.role_manager, _ = Role.objects.get_or_create(name='manager') + self.role_ext_manager, _ = Role.objects.get_or_create(name='external_manager') + + # Company provider A & B + self.company_provider = Company.objects.create(name="Prestataire Alpha") + self.company_other = Company.objects.create(name="Prestataire Beta") + + # Teams + self.team_provider = CompanyTeam.objects.create(name="Equipe Alpha", company=self.company_provider) + self.team_other = CompanyTeam.objects.create(name="Equipe Beta", company=self.company_other) + + # Internal user (SPRB / Admin) + self.internal_user = User.objects.create_user(username='internal_user', password='pwd') + self.internal_config = UserConfig.objects.create(user=self.internal_user, is_intern=True) + self.internal_config.roles.add(self.role_admin) + + # External user from provider company + self.provider_user = User.objects.create_user(username='provider_user', password='pwd') + self.provider_config = UserConfig.objects.create( + user=self.provider_user, + is_intern=False, + company=self.company_provider, + ) + self.provider_config.roles.add(self.role_ext_manager) + self.provider_member = CompanyMember.objects.create(name="Membre Alpha", user=self.provider_user) + self.provider_member.teams.add(self.team_provider) + + # External user from other company + self.other_user = User.objects.create_user(username='other_user', password='pwd') + self.other_config = UserConfig.objects.create( + user=self.other_user, + is_intern=False, + company=self.company_other, + ) + self.other_config.roles.add(self.role_ext_manager) + self.other_member = CompanyMember.objects.create(name="Membre Beta", user=self.other_user) + self.other_member.teams.add(self.team_other) + + from datetime import date + from common.models import UserContractAccess, UserContractStatusPermission + from interventions.models import Symptom + + self.thematic = Thematic.objects.create(code='VOI', name_fr='Voirie', name_nl='Wegen') + self.symptom = Symptom.objects.create(code='SYM-01', name_fr='Nid de poule', name_nl='Kuil', thematic=self.thematic) + + self.contract = Contract.objects.create( + contract_number='CTR-2026-ALPHA', + company=self.company_provider, + notification_emails='prestataire@alpha.be', + send_email_notifications=True, + start_date=date(2025, 1, 1), + end_date=date(2030, 1, 1), + ) + self.contract.thematics.add(self.thematic) + self.internal_config.thematics.add(self.thematic) + self.internal_config.contracts.add(self.contract) + self.provider_config.thematics.add(self.thematic) + self.provider_config.contracts.add(self.contract) + + # Grant status permissions for provider on contract + uca_provider = UserContractAccess.objects.get(user_config=self.provider_config, contract=self.contract) + for st in ['in_preparation', 'to_be_processed', 'assigned', 'in_progress', 'finished']: + UserContractStatusPermission.objects.create( + user_contract=uca_provider, + status=st, + can_view=True, + can_edit=True, + can_change_status_to=True, + ) + + def test_is_user_same_as_provider_helper(self): + from interventions.views.helpers import is_user_same_as_provider + + intervention = Intervention.objects.create( + title='Intervention Test Provider', + maintain_type='corrective', + contract=self.contract, + assigned_provider=self.company_provider, + thematic=self.thematic, + symptom=self.symptom, + created_by=self.internal_user, + ) + + # Internal user is never considered the external provider + self.assertFalse(is_user_same_as_provider(self.internal_user, intervention)) + + # Provider user belonging to company_provider is recognized + self.assertTrue(is_user_same_as_provider(self.provider_user, intervention)) + + # External user from other company is not recognized + self.assertFalse(is_user_same_as_provider(self.other_user, intervention)) + + # Unauthenticated / None user + self.assertFalse(is_user_same_as_provider(None, intervention)) + + @mock.patch('interventions.views.workflow.is_smtp_available', return_value=True) + @mock.patch('django.core.mail.EmailMultiAlternatives.send', return_value=1) + def test_provider_repassing_finished_to_to_be_processed_does_not_send_email(self, mock_email_send, mock_smtp): + from interventions.views.workflow import attempt_status_update + from interventions.models import InterventionEmailLog + + # Intervention created by internal user, finished by contractor + intervention = Intervention.objects.create( + title='Interv Finished Repassed to To Be Processed', + maintain_type='corrective', + status='finished', + status_order=STATUS_ORDERS['finished'], + contract=self.contract, + assigned_provider=self.company_provider, + thematic=self.thematic, + symptom=self.symptom, + created_by=self.internal_user, + ) + + request = mock.Mock() + request.user = self.provider_user + request.POST = {} + request.content_type = 'application/x-www-form-urlencoded' + request.build_absolute_uri = lambda uri: f"http://localhost{uri}" + + # Provider user repasses intervention from 'finished' to 'to_be_processed' + success, message, email_failed = attempt_status_update( + request, + intervention, + self.provider_user, + 'to_be_processed', + ) + + self.assertTrue(success) + intervention.refresh_from_db() + self.assertEqual(intervention.status, 'to_be_processed') + + # No email should be sent and no transmission log created + mock_email_send.assert_not_called() + self.assertFalse(InterventionEmailLog.objects.filter(intervention=intervention).exists()) + + @mock.patch('interventions.views.workflow.is_smtp_available', return_value=True) + @mock.patch('django.core.mail.EmailMultiAlternatives.send', return_value=1) + def test_internal_user_passing_to_to_be_processed_sends_email(self, mock_email_send, mock_smtp): + from interventions.views.workflow import attempt_status_update + from interventions.models import InterventionEmailLog + + intervention = Intervention.objects.create( + title='Interv by internal user', + maintain_type='corrective', + status='in_preparation', + status_order=STATUS_ORDERS['in_preparation'], + contract=self.contract, + assigned_provider=self.company_provider, + thematic=self.thematic, + symptom=self.symptom, + created_by=self.internal_user, + ) + + request = mock.Mock() + request.user = self.internal_user + request.POST = {} + request.content_type = 'application/x-www-form-urlencoded' + request.build_absolute_uri = lambda uri: f"http://localhost{uri}" + + success, message, email_failed = attempt_status_update( + request, + intervention, + self.internal_user, + 'to_be_processed', + ) + + self.assertTrue(success) + intervention.refresh_from_db() + self.assertEqual(intervention.status, 'to_be_processed') + + # Email should be sent to the contract recipient + mock_email_send.assert_called_once() + self.assertTrue(InterventionEmailLog.objects.filter( + intervention=intervention, + email_type='transmission', + success=True, + ).exists()) + + + diff --git a/loko/interventions/views/helpers.py b/loko/interventions/views/helpers.py index cb538eb..15e4c21 100644 --- a/loko/interventions/views/helpers.py +++ b/loko/interventions/views/helpers.py @@ -216,6 +216,7 @@ __all__ = [ 'get_intervention_attachments', 'inspection_configuration_geojson', 'is_smtp_available', + 'is_user_same_as_provider', 'resolve_asset_category', 'resolve_asset_spatial_data', 'resolve_coords_from_hierarchy', @@ -3573,6 +3574,51 @@ def resolve_asset_category(asset, symptom=None): +def is_user_same_as_provider(user, intervention) -> bool: + """ + Vérifie si l'utilisateur appartient à la même organisation que le prestataire assigné à l'intervention. + """ + if not user or not getattr(user, 'is_authenticated', False): + return False + + if getattr(user, 'is_superuser', False): + return False + + user_config = getattr(user, 'config', None) + # Les utilisateurs internes (ex: SPRB / régie) ne sont pas considérés comme le prestataire externe + if user_config and getattr(user_config, 'is_intern', False): + return False + + # Récupérer l'ID des sociétés prestataires liées à l'intervention + provider_ids = set() + if getattr(intervention, 'assigned_provider_id', None): + provider_ids.add(intervention.assigned_provider_id) + if getattr(intervention, 'contract_id', None) and getattr(intervention.contract, 'company_id', None): + provider_ids.add(intervention.contract.company_id) + if getattr(intervention, 'assigned_team_id', None) and getattr(intervention.assigned_team, 'company_id', None): + provider_ids.add(intervention.assigned_team.company_id) + + if not provider_ids: + return False + + # 1. Vérifier via user_config.company + if user_config and user_config.company_id in provider_ids: + return True + + # 2. Vérifier via CompanyMember (équipes de la société) + from contracts.models import CompanyMember + if CompanyMember.objects.filter(user=user, teams__company_id__in=provider_ids).exists(): + return True + + # 3. Vérifier via UserContractAccess + if user_config and user_config.contracts.filter(company_id__in=provider_ids).exists(): + return True + + return False + + + + def get_contract_recipients(contract, intervention=None) -> List[str]: """Retourne une liste unique d'adresses email associées au contrat.""" diff --git a/loko/interventions/views/workflow.py b/loko/interventions/views/workflow.py index cba19a6..960f6b4 100644 --- a/loko/interventions/views/workflow.py +++ b/loko/interventions/views/workflow.py @@ -777,9 +777,10 @@ def attempt_status_update(request, intervention, user, new_status, pause_reason= if intervention.intervention_sheet: pass # déjà générée, ne rien faire - elif intervention.created_by and intervention.created_by.config and (not intervention.created_by.config.is_intern) \ - and intervention.assigned_provider and intervention.created_by.company == intervention.assigned_provider: - pass # ne pas générer de fiche ni envoyer d'email pour les interventions créées par un utilisateur externe pour sa propre compagnie (évite les boucles d'envoi) + elif is_user_same_as_provider(user, intervention) or ( + intervention.created_by and is_user_same_as_provider(intervention.created_by, intervention) + ): + pass # ne pas générer de fiche ni envoyer d'email si l'intervention est passée à ce statut ou créée par un utilisateur appartenant à la société prestataire (évite les boucles d'envoi) else: buffer = io.BytesIO()