441 lines
20 KiB
Python
441 lines
20 KiB
Python
import json
|
|
from django.test import TestCase, RequestFactory
|
|
from django.contrib.auth import get_user_model
|
|
from django.utils import timezone
|
|
|
|
from common.models import UserConfig, Role, Thematic, UserThematics, UserContractAccess
|
|
from contracts.models import Company, Contract, CompanyTeam, CompanyMember
|
|
from assets.models import AssetCategory
|
|
from observations.models import Observation
|
|
from interventions.models import Intervention, Symptom
|
|
from django.http import QueryDict
|
|
from interventions.permissions import can_view_intervention, filter_viewable_interventions_for_user, user_allowed_transitions
|
|
from interventions.views.workflow import attempt_status_update
|
|
from interventions.views.helpers import get_contract_recipients, get_contract_cc_recipients, apply_intervention_filters_from_querydict
|
|
|
|
|
|
class BlockedContractsTests(TestCase):
|
|
def setUp(self):
|
|
self.factory = RequestFactory()
|
|
self.User = get_user_model()
|
|
self.thematic, _ = Thematic.objects.get_or_create(code='trafficlights', defaults={'name': 'Feux de signalisation'})
|
|
self.category, _ = AssetCategory.objects.get_or_create(name_fr='Test Category', thematic=self.thematic)
|
|
|
|
# Roles
|
|
self.admin_role, _ = Role.objects.get_or_create(name='admin')
|
|
self.operator_role, _ = Role.objects.get_or_create(name='operator')
|
|
self.external_role, _ = Role.objects.get_or_create(name='external_manager')
|
|
|
|
# Company & Contract
|
|
self.company = Company.objects.create(name='Prestataire Alpha')
|
|
today = timezone.now().date()
|
|
self.contract_active = Contract.objects.create(
|
|
contract_number='CTR-2026-001',
|
|
company=self.company,
|
|
start_date=today,
|
|
end_date=today + timezone.timedelta(days=365),
|
|
is_active=True,
|
|
is_blocked=False,
|
|
notification_emails='prestataire@example.com',
|
|
cc_notification_emails='cc@example.com'
|
|
)
|
|
self.contract_active.thematics.add(self.thematic)
|
|
|
|
self.contract_blocked = Contract.objects.create(
|
|
contract_number='CTR-2026-BLOCKED',
|
|
company=self.company,
|
|
start_date=today,
|
|
end_date=today + timezone.timedelta(days=365),
|
|
is_active=True,
|
|
is_blocked=True,
|
|
blocked_reason='Gel budgétaire temporaire',
|
|
notification_emails='prestataire_blocked@example.com',
|
|
cc_notification_emails='cc_blocked@example.com'
|
|
)
|
|
self.contract_blocked.thematics.add(self.thematic)
|
|
|
|
# Internal user (operator)
|
|
self.internal_user = self.User.objects.create_user(username='internal_op', password='pwd')
|
|
self.internal_cfg = UserConfig.objects.create(user=self.internal_user, is_intern=True)
|
|
self.internal_cfg.roles.add(self.operator_role)
|
|
UserThematics.objects.create(user_config=self.internal_cfg, thematic=self.thematic, can_view_interventions=True, can_edit_interventions=True)
|
|
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', '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)
|
|
|
|
# External user (contractor)
|
|
self.external_user = self.User.objects.create_user(username='provider_user', password='pwd')
|
|
self.external_cfg = UserConfig.objects.create(user=self.external_user, is_intern=False, company=self.company)
|
|
self.external_cfg.roles.add(self.external_role)
|
|
self.team = CompanyTeam.objects.create(name='Equipe 1', company=self.company)
|
|
self.member = CompanyMember.objects.create(name='Jean Provider', user=self.external_user)
|
|
self.member.teams.add(self.team)
|
|
UserThematics.objects.create(user_config=self.external_cfg, thematic=self.thematic, can_view_interventions=True, can_edit_interventions=True)
|
|
UserContractAccess.objects.create(user_config=self.external_cfg, contract=self.contract_active, can_view_interventions=True)
|
|
UserContractAccess.objects.create(user_config=self.external_cfg, contract=self.contract_blocked, can_view_interventions=True)
|
|
|
|
def test_contract_display_name_and_attributes(self):
|
|
self.assertIn('[BLOQUÉ]', self.contract_blocked.display_name)
|
|
self.assertNotIn('[BLOQUÉ]', self.contract_active.display_name)
|
|
self.assertTrue(self.contract_blocked.is_blocked)
|
|
self.assertEqual(self.contract_blocked.blocked_reason, 'Gel budgétaire temporaire')
|
|
|
|
def test_intervention_creation_with_blocked_contract_forces_preparation(self):
|
|
itv = Intervention.objects.create(
|
|
title='Intervention Bloquée Test',
|
|
contract=self.contract_blocked,
|
|
status='to_be_processed',
|
|
thematic=self.thematic,
|
|
asset_category=self.category,
|
|
)
|
|
# Should be forced to in_preparation and is_blocked_due_to_contract=True
|
|
self.assertEqual(itv.status, 'in_preparation')
|
|
self.assertTrue(itv.is_blocked_due_to_contract)
|
|
self.assertTrue(itv.is_contract_blocked)
|
|
|
|
def test_intervention_permissions_visibility_for_external_provider(self):
|
|
# 1. Past/In-progress intervention on same contract (simulate existing intervention prior to block)
|
|
past_itv = Intervention.objects.create(
|
|
title='Intervention passée',
|
|
contract=self.contract_active,
|
|
assigned_provider=self.company,
|
|
status='in_progress',
|
|
thematic=self.thematic,
|
|
asset_category=self.category,
|
|
)
|
|
Intervention.objects.filter(pk=past_itv.pk).update(contract=self.contract_blocked, is_blocked_due_to_contract=False, status='in_progress')
|
|
past_itv.refresh_from_db()
|
|
# 2. Blocked intervention
|
|
blocked_itv = Intervention.objects.create(
|
|
title='Nouvelle intervention bloquée',
|
|
contract=self.contract_blocked,
|
|
assigned_provider=self.company,
|
|
status='in_preparation',
|
|
thematic=self.thematic,
|
|
asset_category=self.category,
|
|
)
|
|
|
|
# External user should see past_itv but NOT blocked_itv
|
|
self.assertTrue(can_view_intervention(self.external_user, past_itv))
|
|
self.assertFalse(can_view_intervention(self.external_user, blocked_itv))
|
|
|
|
# Queryset filter for external user
|
|
qs = filter_viewable_interventions_for_user(self.external_user)
|
|
self.assertIn(past_itv, qs)
|
|
self.assertNotIn(blocked_itv, qs)
|
|
|
|
# Internal user can view both
|
|
self.assertTrue(can_view_intervention(self.internal_user, past_itv))
|
|
self.assertTrue(can_view_intervention(self.internal_user, blocked_itv))
|
|
internal_qs = filter_viewable_interventions_for_user(self.internal_user)
|
|
self.assertIn(past_itv, internal_qs)
|
|
self.assertIn(blocked_itv, internal_qs)
|
|
|
|
def test_workflow_transitions_prevented_for_blocked_contract(self):
|
|
blocked_itv = Intervention.objects.create(
|
|
title='Intervention Bloquée',
|
|
contract=self.contract_blocked,
|
|
status='in_preparation',
|
|
thematic=self.thematic,
|
|
asset_category=self.category,
|
|
)
|
|
|
|
allowed_transitions = user_allowed_transitions(self.internal_user, 'in_preparation', intervention=blocked_itv)
|
|
self.assertNotIn('to_be_processed', allowed_transitions)
|
|
self.assertNotIn('assigned', allowed_transitions)
|
|
self.assertNotIn('in_progress', allowed_transitions)
|
|
|
|
# Direct call to attempt_status_update
|
|
success, msg, _ = attempt_status_update(None, blocked_itv, self.internal_user, 'to_be_processed')
|
|
self.assertFalse(success)
|
|
|
|
def test_email_recipients_disabled_for_blocked_contract(self):
|
|
# Active contract has recipients
|
|
recipients_active = get_contract_recipients(self.contract_active)
|
|
self.assertIn('prestataire@example.com', recipients_active)
|
|
|
|
# Blocked contract returns empty list
|
|
recipients_blocked = get_contract_recipients(self.contract_blocked)
|
|
self.assertEqual(recipients_blocked, [])
|
|
|
|
cc_blocked = get_contract_cc_recipients(self.contract_blocked)
|
|
self.assertEqual(cc_blocked, [])
|
|
|
|
def test_observation_auto_create_intervention_with_blocked_contract(self):
|
|
symptom = Symptom.objects.create(
|
|
name_fr='Panne de feu',
|
|
thematic=self.thematic,
|
|
contract=self.contract_blocked,
|
|
auto_create_intervention_status='to_be_processed',
|
|
)
|
|
|
|
obs = Observation.objects.create(
|
|
thematic=self.thematic,
|
|
symptom=symptom,
|
|
category=self.category,
|
|
description='Feu tricolore éteint',
|
|
longitude=4.35,
|
|
latitude=50.85,
|
|
created_by=self.internal_user,
|
|
)
|
|
|
|
obs.create_auto_intervention()
|
|
self.assertIsNotNone(obs.intervention)
|
|
itv = obs.intervention
|
|
self.assertEqual(itv.status, 'in_preparation')
|
|
self.assertTrue(itv.is_blocked_due_to_contract)
|
|
self.assertTrue(itv.is_contract_blocked)
|
|
|
|
def test_filter_by_is_blocked_due_to_contract(self):
|
|
itv_active = Intervention.objects.create(
|
|
title='Active',
|
|
contract=self.contract_active,
|
|
status='in_preparation',
|
|
thematic=self.thematic,
|
|
asset_category=self.category,
|
|
is_blocked_due_to_contract=False,
|
|
)
|
|
itv_blocked = Intervention.objects.create(
|
|
title='Bloquée',
|
|
contract=self.contract_blocked,
|
|
status='in_preparation',
|
|
thematic=self.thematic,
|
|
asset_category=self.category,
|
|
)
|
|
|
|
base_qs = Intervention.objects.all()
|
|
|
|
# Filter blocked = true
|
|
qdict_blocked = QueryDict('is_blocked_due_to_contract=1')
|
|
filtered_blocked = apply_intervention_filters_from_querydict(base_qs, qdict_blocked)
|
|
self.assertIn(itv_blocked, filtered_blocked)
|
|
self.assertNotIn(itv_active, filtered_blocked)
|
|
|
|
# Filter blocked = false
|
|
qdict_unblocked = QueryDict('is_blocked_due_to_contract=0')
|
|
filtered_unblocked = apply_intervention_filters_from_querydict(base_qs, qdict_unblocked)
|
|
self.assertIn(itv_active, filtered_unblocked)
|
|
self.assertNotIn(itv_blocked, filtered_unblocked)
|
|
|
|
def test_unblocking_when_changing_contract_to_active(self):
|
|
symptom = Symptom.objects.create(name_fr='Feu éteint', thematic=self.thematic)
|
|
# 1. Create intervention on blocked contract
|
|
itv = Intervention.objects.create(
|
|
title='Intervention initiale bloquée',
|
|
contract=self.contract_blocked,
|
|
assigned_provider=self.company,
|
|
status='in_preparation',
|
|
thematic=self.thematic,
|
|
asset_category=self.category,
|
|
symptom=symptom,
|
|
)
|
|
self.assertTrue(itv.is_blocked_due_to_contract)
|
|
self.assertTrue(itv.is_contract_blocked)
|
|
allowed_before = user_allowed_transitions(self.internal_user, 'in_preparation', intervention=itv)
|
|
self.assertNotIn('to_be_processed', allowed_before)
|
|
|
|
# 2. Reassign to active contract
|
|
itv.contract = self.contract_active
|
|
itv.save()
|
|
|
|
# 3. Verify flags are reset and workflow is unlocked
|
|
self.assertFalse(itv.is_blocked_due_to_contract)
|
|
self.assertFalse(itv.is_contract_blocked)
|
|
allowed_after = user_allowed_transitions(self.internal_user, 'in_preparation', intervention=itv)
|
|
self.assertIn('to_be_processed', allowed_after)
|
|
|
|
# 4. Verify transition to to_be_processed succeeds
|
|
request = self.factory.post('/workflow/update-status/')
|
|
request.user = self.internal_user
|
|
success, msg, _ = attempt_status_update(request, itv, self.internal_user, 'to_be_processed')
|
|
self.assertTrue(success)
|
|
itv.refresh_from_db()
|
|
self.assertEqual(itv.status, 'to_be_processed')
|
|
|
|
def test_show_contract_blocked_warning_only_for_prep_approval_planning(self):
|
|
"""
|
|
Vérifie que show_contract_blocked_warning est True uniquement pour :
|
|
- in_preparation
|
|
- to_be_approved
|
|
- to_be_planned
|
|
Et False pour tous les autres statuts (ex: validated, finished, in_progress, etc.).
|
|
"""
|
|
allowed_warning_statuses = {'in_preparation', 'to_be_approved', 'to_be_planned'}
|
|
other_statuses = {
|
|
'to_be_processed', 'assigned', 'in_progress', 'on_pause', 'finished',
|
|
'processed', 'to_be_corrected', 'corrected', 'validated', 'invoiced',
|
|
'closed', 'canceled'
|
|
}
|
|
|
|
# Pour une intervention sur contrat bloqué
|
|
for status in allowed_warning_statuses:
|
|
itv = Intervention(
|
|
title=f'Test {status}',
|
|
contract=self.contract_blocked,
|
|
status=status,
|
|
thematic=self.thematic,
|
|
asset_category=self.category,
|
|
)
|
|
self.assertTrue(
|
|
itv.show_contract_blocked_warning,
|
|
f"show_contract_blocked_warning devrait être True pour le statut {status}"
|
|
)
|
|
|
|
for status in other_statuses:
|
|
itv = Intervention(
|
|
title=f'Test {status}',
|
|
contract=self.contract_blocked,
|
|
status=status,
|
|
thematic=self.thematic,
|
|
asset_category=self.category,
|
|
)
|
|
self.assertFalse(
|
|
itv.show_contract_blocked_warning,
|
|
f"show_contract_blocked_warning devrait être False pour le statut {status}"
|
|
)
|
|
|
|
# Pour une intervention sur contrat actif (non bloqué)
|
|
for status in allowed_warning_statuses | other_statuses:
|
|
itv = Intervention(
|
|
title=f'Test {status} active',
|
|
contract=self.contract_active,
|
|
status=status,
|
|
thematic=self.thematic,
|
|
asset_category=self.category,
|
|
)
|
|
self.assertFalse(
|
|
itv.show_contract_blocked_warning,
|
|
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)
|
|
|
|
|