220 lines
10 KiB
Python
220 lines
10 KiB
Python
import json
|
|
from django.test import TestCase
|
|
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.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']:
|
|
UserContractStatusPermission.objects.create(user_contract=uca_active, status=st, can_view=True)
|
|
UserContractStatusPermission.objects.create(user_contract=uca_blocked, status=st, can_view=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)
|