From 23936a5767653acf354902b8973701a97715f93e Mon Sep 17 00:00:00 2001 From: kdeterme Date: Fri, 4 Sep 2026 11:15:15 +0200 Subject: [PATCH] refactor: restrict is_blocked_due_to_contract flag to specific intervention statuses and sync via model save method --- loko/contracts/models.py | 28 ++++ ...07_cleanup_blocked_status_interventions.py | 27 ++++ loko/interventions/models.py | 5 +- .../tests/test_blocked_contracts.py | 130 +++++++++++++++++- loko/interventions/views/general.py | 2 +- 5 files changed, 189 insertions(+), 3 deletions(-) create mode 100644 loko/interventions/migrations/0007_cleanup_blocked_status_interventions.py diff --git a/loko/contracts/models.py b/loko/contracts/models.py index 6dbf168..687946b 100644 --- a/loko/contracts/models.py +++ b/loko/contracts/models.py @@ -408,6 +408,34 @@ class Contract(models.Model): help_text=_("Date et heure auxquelles le contrat a été marqué comme bloqué."), ) + def save(self, *args, **kwargs): + is_new = self._state.adding + old_is_blocked = False + if not is_new and self.pk: + old_is_blocked = Contract.objects.filter(pk=self.pk).values_list('is_blocked', flat=True).first() or False + + if self.is_blocked and not self.blocked_at: + self.blocked_at = timezone.now() + elif not self.is_blocked: + self.blocked_at = None + + super().save(*args, **kwargs) + + if not is_new and old_is_blocked != self.is_blocked: + from interventions.models import Intervention + if self.is_blocked: + Intervention.objects.filter( + contract=self, + status__in=Intervention.BLOCKED_CONTRACT_DISPLAY_STATUSES + ).update(is_blocked_due_to_contract=True) + Intervention.objects.filter( + contract=self + ).exclude( + status__in=Intervention.BLOCKED_CONTRACT_DISPLAY_STATUSES + ).update(is_blocked_due_to_contract=False) + else: + Intervention.objects.filter(contract=self).update(is_blocked_due_to_contract=False) + def __str__(self): return f"Contrat {self.contract_number} - {self.company.name}" diff --git a/loko/interventions/migrations/0007_cleanup_blocked_status_interventions.py b/loko/interventions/migrations/0007_cleanup_blocked_status_interventions.py new file mode 100644 index 0000000..b149452 --- /dev/null +++ b/loko/interventions/migrations/0007_cleanup_blocked_status_interventions.py @@ -0,0 +1,27 @@ +from django.db import migrations + + +def cleanup_blocked_flag(apps, schema_editor): + Intervention = apps.get_model('interventions', 'Intervention') + # Les interventions qui ne sont pas en préparation, à approuver ou à planifier + # ne doivent pas avoir is_blocked_due_to_contract=True + Intervention.objects.filter( + is_blocked_due_to_contract=True + ).exclude( + status__in=['in_preparation', 'to_be_approved', 'to_be_planned'] + ).update(is_blocked_due_to_contract=False) + + +def reverse_noop(apps, schema_editor): + pass + + +class Migration(migrations.Migration): + + dependencies = [ + ('interventions', '0006_intervention_is_blocked_due_to_contract'), + ] + + operations = [ + migrations.RunPython(cleanup_blocked_flag, reverse_noop), + ] diff --git a/loko/interventions/models.py b/loko/interventions/models.py index 2396f30..f3fdf00 100644 --- a/loko/interventions/models.py +++ b/loko/interventions/models.py @@ -496,9 +496,12 @@ class Intervention(models.Model): # Gestion de la sécurité sur contrat bloqué if self.contract and getattr(self.contract, 'is_blocked', False): - self.is_blocked_due_to_contract = True if is_new and self.status in ('to_be_processed', 'assigned', 'in_progress'): self.status = 'in_preparation' + if self.status in self.BLOCKED_CONTRACT_DISPLAY_STATUSES: + self.is_blocked_due_to_contract = True + else: + self.is_blocked_due_to_contract = False else: self.is_blocked_due_to_contract = False diff --git a/loko/interventions/tests/test_blocked_contracts.py b/loko/interventions/tests/test_blocked_contracts.py index 9424568..df4e520 100644 --- a/loko/interventions/tests/test_blocked_contracts.py +++ b/loko/interventions/tests/test_blocked_contracts.py @@ -62,7 +62,7 @@ class BlockedContractsTests(TestCase): uca_active = UserContractAccess.objects.create(user_config=self.internal_cfg, contract=self.contract_active, can_view_interventions=True) uca_blocked = UserContractAccess.objects.create(user_config=self.internal_cfg, contract=self.contract_blocked, can_view_interventions=True) from common.models import UserContractStatusPermission - for st in ['in_preparation', 'in_progress', 'to_be_processed']: + for st in ['in_preparation', 'in_progress', 'to_be_processed', 'processed', 'validated']: UserContractStatusPermission.objects.create(user_contract=uca_active, status=st, can_view=True, can_change_status_to=True) UserContractStatusPermission.objects.create(user_contract=uca_blocked, status=st, can_view=True, can_change_status_to=True) @@ -311,3 +311,131 @@ class BlockedContractsTests(TestCase): f"show_contract_blocked_warning devrait être False sur contrat actif pour le statut {status}" ) + def test_is_blocked_due_to_contract_not_set_for_processed_and_subsequent_statuses(self): + """ + Vérifie qu'une intervention dont le statut est traité ou ultérieur (ex: validé) + ne reçoit pas l'attribut is_blocked_due_to_contract=True lors de son enregistrement, + même si son contrat est bloqué. + """ + # 1. Création d'une intervention passée au statut processed + itv = Intervention.objects.create( + title='Intervention traitée', + contract=self.contract_blocked, + status='processed', + thematic=self.thematic, + asset_category=self.category, + ) + self.assertFalse(itv.is_blocked_due_to_contract) + + # 2. Transition vers validated + request = self.factory.post('/workflow/update-status/') + request.user = self.internal_user + success, msg, _ = attempt_status_update(request, itv, self.internal_user, 'validated') + self.assertTrue(success) + itv.refresh_from_db() + self.assertEqual(itv.status, 'validated') + self.assertFalse(itv.is_blocked_due_to_contract) + + def test_external_user_cannot_view_blocked_intervention_regardless_of_status(self): + """ + Vérifie qu'un externe ne peut jamais voir une intervention marquée is_blocked_due_to_contract=True, + quel que soit le statut. + """ + for status in ['in_preparation', 'to_be_approved', 'to_be_planned', 'validated']: + itv = Intervention.objects.create( + title=f'Intervention test {status}', + contract=self.contract_blocked, + assigned_provider=self.company, + status=status, + thematic=self.thematic, + asset_category=self.category, + ) + # Forcer is_blocked_due_to_contract=True pour tester la permission + Intervention.objects.filter(pk=itv.pk).update(is_blocked_due_to_contract=True) + itv.refresh_from_db() + + self.assertFalse( + can_view_intervention(self.external_user, itv), + f"L'externe ne devrait pas pouvoir voir une intervention avec is_blocked_due_to_contract=True au statut {status}" + ) + qs = filter_viewable_interventions_for_user(self.external_user) + self.assertNotIn( + itv, + qs, + f"L'intervention avec is_blocked_due_to_contract=True au statut {status} ne devrait pas apparaître dans le queryset de l'externe" + ) + + def test_contract_save_syncs_is_blocked_due_to_contract_only_on_target_statuses(self): + """ + Vérifie que le blocage d'un contrat via Contract.save() ne bloque que les interventions + en préparation, à approuver ou à planifier, et pas celles qui sont déjà traitées / validées. + """ + today = timezone.now().date() + contract = Contract.objects.create( + contract_number='CTR-SYNC-TEST', + company=self.company, + start_date=today, + end_date=today + timezone.timedelta(days=365), + is_active=True, + is_blocked=False, + ) + contract.thematics.add(self.thematic) + + itv_prep = Intervention.objects.create( + title='Itv en préparation', + contract=contract, + status='in_preparation', + thematic=self.thematic, + asset_category=self.category, + ) + itv_app = Intervention.objects.create( + title='Itv à approuver', + contract=contract, + status='to_be_approved', + thematic=self.thematic, + asset_category=self.category, + ) + itv_plan = Intervention.objects.create( + title='Itv à planifier', + contract=contract, + status='to_be_planned', + thematic=self.thematic, + asset_category=self.category, + ) + itv_val = Intervention.objects.create( + title='Itv validée', + contract=contract, + status='validated', + thematic=self.thematic, + asset_category=self.category, + ) + + # Bloquer le contrat + contract.is_blocked = True + contract.save() + + itv_prep.refresh_from_db() + itv_app.refresh_from_db() + itv_plan.refresh_from_db() + itv_val.refresh_from_db() + + self.assertTrue(itv_prep.is_blocked_due_to_contract) + self.assertTrue(itv_app.is_blocked_due_to_contract) + self.assertTrue(itv_plan.is_blocked_due_to_contract) + self.assertFalse(itv_val.is_blocked_due_to_contract) + + # Débloquer le contrat + contract.is_blocked = False + contract.save() + + itv_prep.refresh_from_db() + itv_app.refresh_from_db() + itv_plan.refresh_from_db() + itv_val.refresh_from_db() + + self.assertFalse(itv_prep.is_blocked_due_to_contract) + self.assertFalse(itv_app.is_blocked_due_to_contract) + self.assertFalse(itv_plan.is_blocked_due_to_contract) + self.assertFalse(itv_val.is_blocked_due_to_contract) + + diff --git a/loko/interventions/views/general.py b/loko/interventions/views/general.py index 0bc2da8..c214d3d 100644 --- a/loko/interventions/views/general.py +++ b/loko/interventions/views/general.py @@ -771,7 +771,7 @@ def interventions_geojson(request): "assigned_provider": r['assigned_provider__name'], "contract_number": r['contract__contract_number'], "contract_provider": r['contract__company__name'], - "is_blocked_due_to_contract": bool(r['is_blocked_due_to_contract'] or r['contract__is_blocked']), + "is_blocked_due_to_contract": bool(r['is_blocked_due_to_contract'] or (r['contract__is_blocked'] and r['status'] in ('in_preparation', 'to_be_approved', 'to_be_planned'))), "origin_type": r['origin_type_label'], "source_category": r[source_category_field] if r[source_category_field] else '', "team": r['assigned_team__name'] or '',