1135 lines
45 KiB
Python
1135 lines
45 KiB
Python
import json
|
||
from datetime import timedelta
|
||
from unittest import mock
|
||
|
||
from django.test import TestCase
|
||
from django.contrib.auth import get_user_model
|
||
from django.contrib import messages
|
||
from django.contrib.messages import get_messages
|
||
from django.core.files.uploadedfile import SimpleUploadedFile
|
||
|
||
from common.models import UserConfig, Role, Thematic, UserThematics
|
||
from contracts.models import Company, CompanyTeam, CompanyMember, Contract, ContractPost, ContractOrder, ContractOrderQuote, ContractOrderQuoteItem
|
||
from assets.models import AssetCategory
|
||
from interventions.models import (
|
||
Intervention,
|
||
InterventionContractPost,
|
||
InterventionPlanificationTimeLine,
|
||
InterventionRealizationTimeLine,
|
||
Operation,
|
||
InterventionAsset,
|
||
OperationTemplate,
|
||
OperationTemplateItem,
|
||
OperationStatusSet,
|
||
OperationStatus,
|
||
STATUS_ORDERS,
|
||
InterventionDocument,
|
||
)
|
||
from django.contrib.contenttypes.models import ContentType
|
||
from interventions.views import _group_operations_by_asset_and_template, _create_equipment_operations_for_assets
|
||
from interventions.permissions import can_manage_occupations, can_delete_document
|
||
from django.urls import reverse
|
||
from django.utils import timezone
|
||
from decimal import Decimal
|
||
|
||
|
||
|
||
class ConfirmCallTests(TestCase):
|
||
def setUp(self):
|
||
User = get_user_model()
|
||
self.user = User.objects.create_user(username='caller', password='pwd')
|
||
self.role_admin, _ = Role.objects.get_or_create(name='admin')
|
||
self.user_config = UserConfig.objects.create(user=self.user, is_intern=True)
|
||
self.user_config.roles.add(self.role_admin)
|
||
self.intervention = Intervention.objects.create(title='Call intervention', status='in_preparation')
|
||
|
||
def test_confirm_call_persists_payload(self):
|
||
self.client.login(username='caller', password='pwd')
|
||
url = reverse('interventions:confirm_call', args=[self.intervention.id])
|
||
tz = timezone.get_current_timezone()
|
||
call_dt = timezone.now().astimezone(tz).replace(second=0, microsecond=0)
|
||
payload = {
|
||
'call_received': False,
|
||
'call_datetime': call_dt.strftime('%Y-%m-%dT%H:%M'),
|
||
}
|
||
|
||
response = self.client.post(url, data=json.dumps(payload), content_type='application/json')
|
||
|
||
self.assertEqual(response.status_code, 200)
|
||
self.intervention.refresh_from_db()
|
||
self.assertFalse(self.intervention.init_call_received)
|
||
self.assertEqual(self.intervention.init_call_confirmed_time, call_dt)
|
||
|
||
def test_confirm_call_defaults_to_now_and_received(self):
|
||
self.client.login(username='caller', password='pwd')
|
||
url = reverse('interventions:confirm_call', args=[self.intervention.id])
|
||
|
||
response = self.client.post(url, data=json.dumps({}), content_type='application/json')
|
||
|
||
self.assertEqual(response.status_code, 200)
|
||
self.intervention.refresh_from_db()
|
||
self.assertTrue(self.intervention.init_call_received)
|
||
self.assertIsNotNone(self.intervention.init_call_confirmed_time)
|
||
|
||
|
||
|
||
|
||
|
||
class ApplyQuoteTests(TestCase):
|
||
def setUp(self):
|
||
User = get_user_model()
|
||
self.user = User.objects.create_user(username='admin', password='pwd')
|
||
self.role_admin = Role.objects.create(name='admin')
|
||
self.user_config = UserConfig.objects.create(user=self.user, is_intern=True)
|
||
self.user_config.roles.add(self.role_admin)
|
||
|
||
# Minimal contract/order/posts/quote setup
|
||
self.company = Company.objects.create(name='Comp')
|
||
from datetime import date, timedelta
|
||
today = date.today()
|
||
self.contract = Contract.objects.create(
|
||
company=self.company,
|
||
contract_number='C-001',
|
||
start_date=today - timedelta(days=30),
|
||
end_date=today + timedelta(days=365)
|
||
)
|
||
self.post1 = ContractPost.objects.create(
|
||
contract=self.contract,
|
||
order_number=1,
|
||
code='P1',
|
||
description_fr='Poste 1',
|
||
description_nl='Post 1',
|
||
unit='u',
|
||
unit_price=Decimal('10.00')
|
||
)
|
||
self.post2 = ContractPost.objects.create(
|
||
contract=self.contract,
|
||
order_number=2,
|
||
code='P2',
|
||
description_fr='Poste 2',
|
||
description_nl='Post 2',
|
||
unit='u',
|
||
unit_price=Decimal('20.00')
|
||
)
|
||
|
||
self.order = ContractOrder.objects.create(
|
||
contract=self.contract,
|
||
order_code='ORD-1',
|
||
order_status='pending',
|
||
order_date=today,
|
||
delivery_date=today + timedelta(days=7)
|
||
)
|
||
self.quote = ContractOrderQuote.objects.create(order=self.order, is_finalized=True)
|
||
ContractOrderQuoteItem.objects.create(quote=self.quote, contract_post=self.post1, quantity=Decimal('5.00'))
|
||
ContractOrderQuoteItem.objects.create(quote=self.quote, contract_post=self.post2, quantity=Decimal('3.00'))
|
||
|
||
self.intervention = Intervention.objects.create(title='With order', contract=self.contract, order=self.order, status='to_be_processed')
|
||
|
||
def test_apply_quote_adds_missing_posts_without_duplicates(self):
|
||
self.client.login(username='admin', password='pwd')
|
||
|
||
# Pre-create one link to ensure no duplicate is created
|
||
InterventionContractPost.objects.create(intervention=self.intervention, contract_post=self.post1, quantity=Decimal('1.00'))
|
||
|
||
url = reverse('interventions:intervention_apply_quote', args=[self.intervention.id])
|
||
resp = self.client.post(url)
|
||
self.assertEqual(resp.status_code, 200)
|
||
data = resp.json()
|
||
self.assertTrue(data.get('success'))
|
||
|
||
# We already had post1; post2 should be created, not duplicated
|
||
posts = InterventionContractPost.objects.filter(intervention=self.intervention).values_list('contract_post_id', flat=True)
|
||
self.assertSetEqual(set(posts), {self.post1.id, self.post2.id})
|
||
# created count should be 1 (only post2)
|
||
self.assertEqual(data.get('created'), 1)
|
||
|
||
|
||
|
||
|
||
class TimelineEventManagementTests(TestCase):
|
||
def setUp(self):
|
||
User = get_user_model()
|
||
self.user = User.objects.create_user(username='timeline-admin', password='pwd')
|
||
self.role_admin, _ = Role.objects.get_or_create(name='admin')
|
||
self.user_config = UserConfig.objects.create(user=self.user, is_intern=True)
|
||
self.user_config.roles.add(self.role_admin)
|
||
|
||
self.intervention = Intervention.objects.create(
|
||
title='Timeline control',
|
||
status='to_be_processed',
|
||
)
|
||
|
||
self.url = reverse('interventions:update_intervention', args=[self.intervention.id])
|
||
self.client.login(username='timeline-admin', password='pwd')
|
||
|
||
def test_planification_future_event_can_be_updated(self):
|
||
future_time = timezone.now() + timedelta(days=2)
|
||
event = InterventionPlanificationTimeLine.objects.create(
|
||
intervention=self.intervention,
|
||
event_time=future_time,
|
||
event_type='planned_begin',
|
||
event_description='Initial planning',
|
||
)
|
||
|
||
new_time = future_time + timedelta(days=1)
|
||
response = self.client.post(self.url, {
|
||
'field': 'planification',
|
||
'action': 'update',
|
||
'event_id': str(event.id),
|
||
'event_time': new_time.isoformat(),
|
||
'event_type': 'planned_end',
|
||
'event_description': 'Updated planning',
|
||
})
|
||
|
||
self.assertEqual(response.status_code, 200)
|
||
payload = response.json()
|
||
self.assertTrue(payload.get('success'))
|
||
|
||
event.refresh_from_db()
|
||
self.assertEqual(event.event_type, 'planned_end')
|
||
self.assertEqual(event.event_description, 'Updated planning')
|
||
self.assertEqual(event.event_time, new_time)
|
||
|
||
self.intervention.refresh_from_db()
|
||
self.assertEqual(self.intervention.planned_end_time, event.event_time)
|
||
|
||
def test_planification_past_event_cannot_be_deleted(self):
|
||
past_event = InterventionPlanificationTimeLine.objects.create(
|
||
intervention=self.intervention,
|
||
event_time=timezone.now() - timedelta(days=1),
|
||
event_type='planned_begin',
|
||
event_description='Past event',
|
||
)
|
||
|
||
response = self.client.post(self.url, {
|
||
'field': 'planification',
|
||
'action': 'delete',
|
||
'event_id': str(past_event.id),
|
||
})
|
||
|
||
self.assertEqual(response.status_code, 400)
|
||
payload = response.json()
|
||
self.assertFalse(payload.get('success'))
|
||
self.assertTrue(
|
||
InterventionPlanificationTimeLine.objects.filter(id=past_event.id).exists()
|
||
)
|
||
|
||
def test_realization_future_event_can_be_deleted(self):
|
||
event = InterventionRealizationTimeLine.objects.create(
|
||
intervention=self.intervention,
|
||
event_time=timezone.now() + timedelta(hours=6),
|
||
event_type='real_begin',
|
||
event_description='Future real event',
|
||
)
|
||
|
||
response = self.client.post(self.url, {
|
||
'field': 'realization',
|
||
'action': 'delete',
|
||
'event_id': str(event.id),
|
||
})
|
||
|
||
self.assertEqual(response.status_code, 200)
|
||
self.assertTrue(response.json().get('success'))
|
||
self.assertFalse(
|
||
InterventionRealizationTimeLine.objects.filter(id=event.id).exists()
|
||
)
|
||
|
||
|
||
|
||
|
||
class InterventionStatusTransitionTests(TestCase):
|
||
def setUp(self):
|
||
User = get_user_model()
|
||
self.admin = User.objects.create_user(username='admin_user', password='pwd')
|
||
self.role_admin = Role.objects.create(name='admin')
|
||
self.admin_config = UserConfig.objects.create(user=self.admin, is_intern=True)
|
||
self.admin_config.roles.add(self.role_admin)
|
||
|
||
self.manager = User.objects.create_user(username='manager_user', password='pwd')
|
||
self.role_manager = Role.objects.create(name='manager')
|
||
self.manager_config = UserConfig.objects.create(user=self.manager, is_intern=True)
|
||
self.manager_config.roles.add(self.role_manager)
|
||
|
||
self.thematic = Thematic.objects.create(code='roads', name_fr='Routes', name_nl='Wegen')
|
||
UserThematics.objects.create(
|
||
user_config=self.manager_config,
|
||
thematic=self.thematic,
|
||
can_view_interventions=True,
|
||
can_edit_interventions=True,
|
||
)
|
||
|
||
self.company = Company.objects.create(name='Provider Co')
|
||
from datetime import date
|
||
today = date.today()
|
||
self.contract = Contract.objects.create(
|
||
company=self.company,
|
||
contract_number='C-TEST',
|
||
start_date=today - timedelta(days=30),
|
||
end_date=today + timedelta(days=365)
|
||
)
|
||
|
||
from common.models import UserContractAccess, UserContractStatusPermission
|
||
self.uca = UserContractAccess.objects.create(
|
||
user_config=self.manager_config,
|
||
contract=self.contract,
|
||
can_view_interventions=True,
|
||
)
|
||
|
||
UserContractStatusPermission.objects.create(
|
||
user_contract=self.uca,
|
||
status='canceled',
|
||
can_change_status_to=True,
|
||
)
|
||
|
||
UserContractStatusPermission.objects.create(
|
||
user_contract=self.uca,
|
||
status='to_be_processed',
|
||
can_change_status_to=True,
|
||
)
|
||
|
||
def test_manager_cannot_cancel_intervention_assigned_or_beyond(self):
|
||
itv_to_be_processed = Intervention.objects.create(
|
||
title='Test active',
|
||
status='to_be_processed',
|
||
contract=self.contract,
|
||
thematic=self.thematic,
|
||
)
|
||
from interventions.permissions import user_allowed_transitions
|
||
allowed_for_to_be_processed = user_allowed_transitions(self.manager, 'to_be_processed', intervention=itv_to_be_processed)
|
||
self.assertIn('canceled', allowed_for_to_be_processed)
|
||
|
||
itv_assigned = Intervention.objects.create(
|
||
title='Test assigned',
|
||
status='assigned',
|
||
contract=self.contract,
|
||
thematic=self.thematic,
|
||
)
|
||
allowed_for_assigned = user_allowed_transitions(self.manager, 'assigned', intervention=itv_assigned)
|
||
self.assertNotIn('canceled', allowed_for_assigned)
|
||
|
||
itv_in_progress = Intervention.objects.create(
|
||
title='Test in progress',
|
||
status='in_progress',
|
||
contract=self.contract,
|
||
thematic=self.thematic,
|
||
)
|
||
allowed_for_in_progress = user_allowed_transitions(self.manager, 'in_progress', intervention=itv_in_progress)
|
||
self.assertNotIn('canceled', allowed_for_in_progress)
|
||
|
||
def test_admin_can_cancel_intervention_regardless_of_status(self):
|
||
itv_assigned = Intervention.objects.create(
|
||
title='Test assigned admin',
|
||
status='assigned',
|
||
contract=self.contract,
|
||
thematic=self.thematic,
|
||
)
|
||
from interventions.permissions import user_allowed_transitions
|
||
allowed_for_assigned = user_allowed_transitions(self.admin, 'assigned', intervention=itv_assigned)
|
||
self.assertIn('canceled', allowed_for_assigned)
|
||
|
||
def test_cancellation_resets_linked_in_progress_observations(self):
|
||
from observations.models import Observation
|
||
from interventions.views import attempt_status_update
|
||
|
||
# Create intervention
|
||
itv = Intervention.objects.create(
|
||
title='Test cancel intervention',
|
||
status='to_be_processed',
|
||
contract=self.contract,
|
||
thematic=self.thematic,
|
||
)
|
||
|
||
# Create linked observation in_progress
|
||
obs_in_progress = Observation.objects.create(
|
||
description='Test obs en cours',
|
||
latitude=50.85,
|
||
longitude=4.35,
|
||
status='in_progress',
|
||
intervention=itv,
|
||
thematic=self.thematic,
|
||
)
|
||
|
||
# Create linked observation not in_progress (e.g. to_process)
|
||
obs_to_process = Observation.objects.create(
|
||
description='Test obs a traiter',
|
||
latitude=50.85,
|
||
longitude=4.35,
|
||
status='to_process',
|
||
intervention=itv,
|
||
thematic=self.thematic,
|
||
)
|
||
|
||
# Cancel the intervention
|
||
success, message, email_failed = attempt_status_update(
|
||
request=None,
|
||
intervention=itv,
|
||
user=self.admin,
|
||
new_status='canceled',
|
||
cancellation_type='to_redirect',
|
||
)
|
||
self.assertTrue(success)
|
||
|
||
# Refresh from db
|
||
obs_in_progress.refresh_from_db()
|
||
obs_to_process.refresh_from_db()
|
||
|
||
# The in_progress observation must have transitioned to to_process
|
||
self.assertEqual(obs_in_progress.status, 'to_process')
|
||
# The to_process observation must remain unchanged
|
||
self.assertEqual(obs_to_process.status, 'to_process')
|
||
|
||
# Check that a comment was added to the transitioned observation
|
||
self.assertTrue(obs_in_progress.comments.filter(comment__contains=itv.code).exists())
|
||
self.assertFalse(obs_to_process.comments.filter(comment__contains=itv.code).exists())
|
||
|
||
def test_create_intervention_from_already_linked_observation(self):
|
||
from observations.models import Observation
|
||
from interventions.models import Symptom, AssetCategory
|
||
from common.models import UserContractStatusPermission
|
||
from django.urls import reverse
|
||
|
||
# Create first intervention
|
||
itv1 = Intervention.objects.create(
|
||
title='First intervention',
|
||
status='canceled',
|
||
contract=self.contract,
|
||
thematic=self.thematic,
|
||
)
|
||
|
||
# Create observation linked to itv1
|
||
obs = Observation.objects.create(
|
||
description='Test obs duplicate creation',
|
||
latitude=50.85,
|
||
longitude=4.35,
|
||
status='to_process',
|
||
intervention=itv1,
|
||
thematic=self.thematic,
|
||
)
|
||
|
||
# 1. Test redirect from create_intervention_from_observation view
|
||
self.client.login(username='manager_user', password='pwd')
|
||
url = reverse('observations:create_intervention_from_observation', args=[obs.id])
|
||
response = self.client.post(url)
|
||
self.assertEqual(response.status_code, 302)
|
||
# It should redirect to the add screen, not the detail screen of itv1
|
||
self.assertIn(reverse('interventions:intervention_add', kwargs={'thematic_code': self.thematic.code}), response.url)
|
||
self.assertIn(f"observation_id={obs.id}", response.url)
|
||
|
||
# 2. Test saving the new intervention updates the observation's intervention link
|
||
add_url = reverse('interventions:intervention_add', kwargs={'thematic_code': self.thematic.code})
|
||
category = AssetCategory.objects.create(
|
||
code='CAT-ADD',
|
||
name_fr='Cat-Add',
|
||
name_nl='Cat-Add',
|
||
thematic=self.thematic,
|
||
)
|
||
symptom = Symptom.objects.create(
|
||
code='SYMP-ADD',
|
||
name_fr='Symp-Add',
|
||
name_nl='Symp-Add',
|
||
thematic=self.thematic,
|
||
)
|
||
UserContractStatusPermission.objects.create(
|
||
user_contract=self.uca,
|
||
status='in_preparation',
|
||
can_change_status_to=True,
|
||
)
|
||
|
||
post_data = {
|
||
'title': 'New intervention for linked obs',
|
||
'type': 'fixing',
|
||
'status': 'in_preparation',
|
||
'symptom': symptom.id,
|
||
'asset_category': category.id,
|
||
'maintain_type': 'corrective',
|
||
'lon': '4.35',
|
||
'lat': '50.85',
|
||
'priority': '3',
|
||
'origin_type': 'demand',
|
||
'contract': self.contract.id,
|
||
'observation_id': obs.id,
|
||
}
|
||
|
||
response = self.client.post(add_url, data=post_data)
|
||
self.assertEqual(response.status_code, 302)
|
||
|
||
# Get the new intervention
|
||
itv2 = Intervention.objects.get(title='New intervention for linked obs')
|
||
|
||
# Verify the observation is now linked to the new intervention itv2
|
||
obs.refresh_from_db()
|
||
self.assertEqual(obs.intervention, itv2)
|
||
|
||
|
||
|
||
|
||
class AutoCloseInterventionTests(TestCase):
|
||
def setUp(self):
|
||
from common.models import UserConfig, Role, Thematic, UserThematics
|
||
from contracts.models import Company, Contract
|
||
from interventions.models import Intervention
|
||
from observations.models import Observation
|
||
|
||
User = get_user_model()
|
||
self.user = User.objects.create_user(username='autoclose_tester', password='pwd')
|
||
self.role_admin, _ = Role.objects.get_or_create(name='admin')
|
||
self.user_config = UserConfig.objects.create(user=self.user, is_intern=True)
|
||
self.user_config.roles.add(self.role_admin)
|
||
|
||
self.thematic = Thematic.objects.create(code='roads', name_fr='Voirie', name_nl='Wegen')
|
||
UserThematics.objects.create(
|
||
user_config=self.user_config,
|
||
thematic=self.thematic,
|
||
can_view_interventions=True,
|
||
can_edit_interventions=True,
|
||
)
|
||
|
||
self.company = Company.objects.create(name='Test Company')
|
||
|
||
from datetime import date
|
||
self.contract_auto = Contract.objects.create(
|
||
company=self.company,
|
||
contract_number='C-AUTO-CLOSE',
|
||
start_date=date.today() - timedelta(days=100),
|
||
end_date=date.today() + timedelta(days=100),
|
||
is_active=True,
|
||
)
|
||
|
||
self.contract_no_auto = Contract.objects.create(
|
||
company=self.company,
|
||
contract_number='C-NO-AUTO-CLOSE',
|
||
start_date=date.today() - timedelta(days=100),
|
||
end_date=date.today() + timedelta(days=100),
|
||
is_active=True,
|
||
)
|
||
|
||
def test_auto_close_expired_interventions(self):
|
||
from django.core.management import call_command
|
||
from interventions.models import Intervention, InterventionTimeLine, Symptom, SymptomContractAutoClose
|
||
from observations.models import Observation, ObservationComment
|
||
|
||
# Create a symptom and a rule (symptom × contract_auto, delay 5 days)
|
||
symptom = Symptom.objects.create(
|
||
code='SYMP-TEST',
|
||
name_fr='Symptôme test',
|
||
name_nl='Symptoom test',
|
||
thematic=self.thematic,
|
||
)
|
||
SymptomContractAutoClose.objects.create(
|
||
symptom=symptom,
|
||
contract=self.contract_auto,
|
||
delay_days=5,
|
||
)
|
||
|
||
# 1. Create eligible intervention that is expired
|
||
past_time = timezone.now() - timedelta(days=10)
|
||
itv_expired = Intervention.objects.create(
|
||
title='Expired intervention',
|
||
status='finished',
|
||
contract=self.contract_auto,
|
||
thematic=self.thematic,
|
||
symptom=symptom,
|
||
)
|
||
# Manually create TimelineEvent to simulate completion time in the past
|
||
InterventionTimeLine.objects.create(
|
||
intervention=itv_expired,
|
||
event_time=past_time,
|
||
event_type='status_change',
|
||
to_status='finished',
|
||
)
|
||
|
||
# Link an observation
|
||
obs = Observation.objects.create(
|
||
description='Obs linked to expired intervention',
|
||
status='to_process',
|
||
intervention=itv_expired,
|
||
thematic=self.thematic,
|
||
latitude=50.85,
|
||
longitude=4.35,
|
||
)
|
||
|
||
# 2. Create eligible intervention that is NOT expired
|
||
recent_time = timezone.now() - timedelta(days=2)
|
||
itv_recent = Intervention.objects.create(
|
||
title='Recent intervention',
|
||
status='finished',
|
||
contract=self.contract_auto,
|
||
thematic=self.thematic,
|
||
symptom=symptom,
|
||
)
|
||
InterventionTimeLine.objects.create(
|
||
intervention=itv_recent,
|
||
event_time=recent_time,
|
||
event_type='status_change',
|
||
to_status='finished',
|
||
)
|
||
|
||
# 3. Create eligible intervention on contract_no_auto (no rule → should NOT be closed)
|
||
itv_no_auto = Intervention.objects.create(
|
||
title='No auto close intervention',
|
||
status='finished',
|
||
contract=self.contract_no_auto,
|
||
thematic=self.thematic,
|
||
symptom=symptom,
|
||
)
|
||
InterventionTimeLine.objects.create(
|
||
intervention=itv_no_auto,
|
||
event_time=past_time,
|
||
event_type='status_change',
|
||
to_status='finished',
|
||
)
|
||
|
||
# 4. Create intervention in a non-eligible status (e.g. in_progress)
|
||
itv_non_eligible = Intervention.objects.create(
|
||
title='In progress intervention',
|
||
status='in_progress',
|
||
contract=self.contract_auto,
|
||
thematic=self.thematic,
|
||
symptom=symptom,
|
||
)
|
||
InterventionTimeLine.objects.create(
|
||
intervention=itv_non_eligible,
|
||
event_time=past_time,
|
||
event_type='status_change',
|
||
to_status='in_progress',
|
||
)
|
||
|
||
# Run the management command
|
||
call_command('close_expired_interventions')
|
||
|
||
# Refresh from database
|
||
itv_expired.refresh_from_db()
|
||
itv_recent.refresh_from_db()
|
||
itv_no_auto.refresh_from_db()
|
||
itv_non_eligible.refresh_from_db()
|
||
obs.refresh_from_db()
|
||
|
||
# Check results
|
||
self.assertEqual(itv_expired.status, 'closed')
|
||
self.assertEqual(obs.status, 'closed')
|
||
# Check that comment was created
|
||
self.assertTrue(ObservationComment.objects.filter(observation=obs, comment__contains=itv_expired.code).exists())
|
||
|
||
self.assertEqual(itv_recent.status, 'finished')
|
||
self.assertEqual(itv_no_auto.status, 'finished')
|
||
self.assertEqual(itv_non_eligible.status, 'in_progress')
|
||
|
||
def test_auto_close_expired_interventions_by_symptom(self):
|
||
from django.core.management import call_command
|
||
from interventions.models import Intervention, InterventionTimeLine, Symptom, SymptomContractAutoClose
|
||
from observations.models import Observation, ObservationComment
|
||
|
||
# Create symptom and rule: symptom_auto × contract_no_auto (delay 3 days)
|
||
symptom_auto = Symptom.objects.create(
|
||
code='SYMP-AUTO-CLOSE',
|
||
name_fr='Symptôme auto-clôture',
|
||
name_nl='Symptoom auto-clôture',
|
||
thematic=self.thematic,
|
||
)
|
||
SymptomContractAutoClose.objects.create(
|
||
symptom=symptom_auto,
|
||
contract=self.contract_no_auto,
|
||
delay_days=3,
|
||
)
|
||
|
||
# 1. Create eligible intervention that is expired under the rule (past 4 days)
|
||
past_time = timezone.now() - timedelta(days=4)
|
||
itv_expired = Intervention.objects.create(
|
||
title='Expired intervention by symptom',
|
||
status='finished',
|
||
contract=self.contract_no_auto,
|
||
thematic=self.thematic,
|
||
symptom=symptom_auto,
|
||
)
|
||
InterventionTimeLine.objects.create(
|
||
intervention=itv_expired,
|
||
event_time=past_time,
|
||
event_type='status_change',
|
||
to_status='finished',
|
||
)
|
||
|
||
# Link an observation
|
||
obs = Observation.objects.create(
|
||
description='Obs linked to symptom expired intervention',
|
||
status='to_process',
|
||
intervention=itv_expired,
|
||
thematic=self.thematic,
|
||
latitude=50.85,
|
||
longitude=4.35,
|
||
)
|
||
|
||
# 2. Create eligible intervention that is NOT expired (2 days past completion)
|
||
recent_time = timezone.now() - timedelta(days=2)
|
||
itv_recent = Intervention.objects.create(
|
||
title='Recent intervention by symptom',
|
||
status='finished',
|
||
contract=self.contract_no_auto,
|
||
thematic=self.thematic,
|
||
symptom=symptom_auto,
|
||
)
|
||
InterventionTimeLine.objects.create(
|
||
intervention=itv_recent,
|
||
event_time=recent_time,
|
||
event_type='status_change',
|
||
to_status='finished',
|
||
)
|
||
|
||
# Run the management command
|
||
call_command('close_expired_interventions')
|
||
|
||
# Refresh from database
|
||
itv_expired.refresh_from_db()
|
||
itv_recent.refresh_from_db()
|
||
obs.refresh_from_db()
|
||
|
||
# Check results
|
||
self.assertEqual(itv_expired.status, 'closed')
|
||
self.assertEqual(obs.status, 'closed')
|
||
self.assertTrue(ObservationComment.objects.filter(observation=obs, comment__contains=itv_expired.code).exists())
|
||
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())
|
||
|
||
|
||
class PhotoRequirementWorkflowTests(TestCase):
|
||
def setUp(self):
|
||
User = get_user_model()
|
||
from datetime import date
|
||
from common.models import UserContractAccess, UserContractStatusPermission
|
||
from interventions.models import Symptom
|
||
|
||
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')
|
||
self.role_technician, _ = Role.objects.get_or_create(name='technician')
|
||
|
||
self.company = Company.objects.create(name='Société Tech')
|
||
self.team_photos_required = CompanyTeam.objects.create(
|
||
name='Equipe Photos Requises',
|
||
company=self.company,
|
||
photos_required=True,
|
||
)
|
||
self.team_no_photos = CompanyTeam.objects.create(
|
||
name='Equipe Sans Photos',
|
||
company=self.company,
|
||
photos_required=False,
|
||
)
|
||
|
||
# Admin user
|
||
self.admin_user = User.objects.create_user(username='admin_ph_user', password='pwd')
|
||
self.admin_config = UserConfig.objects.create(user=self.admin_user, is_intern=True)
|
||
self.admin_config.roles.add(self.role_admin)
|
||
|
||
# Manager user
|
||
self.manager_user = User.objects.create_user(username='manager_ph_user', password='pwd')
|
||
self.manager_config = UserConfig.objects.create(user=self.manager_user, is_intern=True)
|
||
self.manager_config.roles.add(self.role_manager)
|
||
|
||
# External Manager user
|
||
self.ext_manager_user = User.objects.create_user(username='ext_manager_ph_user', password='pwd')
|
||
self.ext_manager_config = UserConfig.objects.create(user=self.ext_manager_user, is_intern=False, company=self.company)
|
||
self.ext_manager_config.roles.add(self.role_ext_manager)
|
||
self.ext_manager_member = CompanyMember.objects.create(name="Ext Mgr", user=self.ext_manager_user)
|
||
self.ext_manager_member.teams.add(self.team_photos_required)
|
||
|
||
# Technician user
|
||
self.tech_user = User.objects.create_user(username='tech_ph_user', password='pwd')
|
||
self.tech_config = UserConfig.objects.create(user=self.tech_user, is_intern=False, company=self.company)
|
||
self.tech_config.roles.add(self.role_technician)
|
||
self.tech_member = CompanyMember.objects.create(name="Technicien 1", user=self.tech_user)
|
||
self.tech_member.teams.add(self.team_photos_required)
|
||
|
||
# Thematic, Symptom, Contract
|
||
self.thematic = Thematic.objects.create(code='VOI_PH', name_fr='Voirie Photo', name_nl='Wegen Photo')
|
||
self.symptom = Symptom.objects.create(code='SYM-PH', name_fr='Symptome Photo', name_nl='Symptom Photo', thematic=self.thematic)
|
||
self.contract = Contract.objects.create(
|
||
contract_number='CTR-PHOTOS-TEST',
|
||
company=self.company,
|
||
start_date=date(2025, 1, 1),
|
||
end_date=date(2030, 1, 1),
|
||
)
|
||
self.contract.thematics.add(self.thematic)
|
||
|
||
for cfg in [self.admin_config, self.manager_config, self.ext_manager_config, self.tech_config]:
|
||
cfg.thematics.add(self.thematic)
|
||
cfg.contracts.add(self.contract)
|
||
uca = UserContractAccess.objects.get(user_config=cfg, contract=self.contract)
|
||
for st in ['in_preparation', 'to_be_processed', 'assigned', 'in_progress', 'on_pause', 'finished']:
|
||
UserContractStatusPermission.objects.create(
|
||
user_contract=uca,
|
||
status=st,
|
||
can_view=True,
|
||
can_edit=True,
|
||
can_change_status_to=True,
|
||
)
|
||
|
||
def test_is_photo_required_for_user_helper(self):
|
||
from interventions.views.helpers import _is_photo_required_for_user
|
||
|
||
intervention = Intervention.objects.create(
|
||
title='Intervention Photo Helper Test',
|
||
maintain_type='corrective',
|
||
status='in_progress',
|
||
status_order=STATUS_ORDERS['in_progress'],
|
||
contract=self.contract,
|
||
assigned_provider=self.company,
|
||
assigned_team=self.team_photos_required,
|
||
assigned_member=self.tech_member,
|
||
thematic=self.thematic,
|
||
symptom=self.symptom,
|
||
created_by=self.admin_user,
|
||
)
|
||
|
||
# Technician with team photos_required=True -> True
|
||
self.assertTrue(_is_photo_required_for_user(self.tech_user, intervention))
|
||
|
||
# Manager -> False
|
||
self.assertFalse(_is_photo_required_for_user(self.manager_user, intervention))
|
||
|
||
# External Manager -> False
|
||
self.assertFalse(_is_photo_required_for_user(self.ext_manager_user, intervention))
|
||
|
||
# Admin -> False
|
||
self.assertFalse(_is_photo_required_for_user(self.admin_user, intervention))
|
||
|
||
# If team photos_required=False -> False for technician as well
|
||
intervention.assigned_team = self.team_no_photos
|
||
intervention.save(update_fields=['assigned_team'])
|
||
self.assertFalse(_is_photo_required_for_user(self.tech_user, intervention))
|
||
|
||
def test_technician_blocked_from_finishing_without_photos(self):
|
||
from interventions.views.workflow import attempt_status_update
|
||
|
||
intervention = Intervention.objects.create(
|
||
title='Intervention Tech Blocked Test',
|
||
maintain_type='corrective',
|
||
status='in_progress',
|
||
status_order=STATUS_ORDERS['in_progress'],
|
||
contract=self.contract,
|
||
assigned_provider=self.company,
|
||
assigned_team=self.team_photos_required,
|
||
assigned_member=self.tech_member,
|
||
thematic=self.thematic,
|
||
symptom=self.symptom,
|
||
created_by=self.admin_user,
|
||
)
|
||
|
||
request = mock.Mock()
|
||
request.user = self.tech_user
|
||
request.POST = {}
|
||
request.content_type = 'application/x-www-form-urlencoded'
|
||
request.build_absolute_uri = lambda uri: f"http://localhost{uri}"
|
||
|
||
# Technician tries to finish without photos
|
||
success, message, _ = attempt_status_update(
|
||
request,
|
||
intervention,
|
||
self.tech_user,
|
||
'finished',
|
||
)
|
||
|
||
self.assertFalse(success)
|
||
self.assertIn("Au moins une photo", str(message))
|
||
intervention.refresh_from_db()
|
||
self.assertEqual(intervention.status, 'in_progress')
|
||
|
||
def test_technician_allowed_to_finish_with_photos(self):
|
||
from interventions.views.workflow import attempt_status_update
|
||
from common.models import Document
|
||
|
||
intervention = Intervention.objects.create(
|
||
title='Intervention Tech Photo Allowed Test',
|
||
maintain_type='corrective',
|
||
status='in_progress',
|
||
status_order=STATUS_ORDERS['in_progress'],
|
||
contract=self.contract,
|
||
assigned_provider=self.company,
|
||
assigned_team=self.team_photos_required,
|
||
assigned_member=self.tech_member,
|
||
thematic=self.thematic,
|
||
symptom=self.symptom,
|
||
created_by=self.admin_user,
|
||
)
|
||
|
||
# Attach document/photo
|
||
doc_file = SimpleUploadedFile("chantier.jpg", b"fake image content", content_type="image/jpeg")
|
||
InterventionDocument.objects.create(
|
||
intervention=intervention,
|
||
file=doc_file,
|
||
uploaded_by=self.tech_user,
|
||
)
|
||
|
||
request = mock.Mock()
|
||
request.user = self.tech_user
|
||
request.POST = {}
|
||
request.content_type = 'application/x-www-form-urlencoded'
|
||
request.build_absolute_uri = lambda uri: f"http://localhost{uri}"
|
||
|
||
success, message, _ = attempt_status_update(
|
||
request,
|
||
intervention,
|
||
self.tech_user,
|
||
'finished',
|
||
)
|
||
|
||
self.assertTrue(success)
|
||
intervention.refresh_from_db()
|
||
self.assertEqual(intervention.status, 'finished')
|
||
|
||
def test_manager_allowed_to_finish_without_photos(self):
|
||
from interventions.views.workflow import attempt_status_update
|
||
|
||
intervention = Intervention.objects.create(
|
||
title='Intervention Manager No Photo Test',
|
||
maintain_type='corrective',
|
||
status='in_progress',
|
||
status_order=STATUS_ORDERS['in_progress'],
|
||
contract=self.contract,
|
||
assigned_provider=self.company,
|
||
assigned_team=self.team_photos_required,
|
||
assigned_member=self.tech_member,
|
||
thematic=self.thematic,
|
||
symptom=self.symptom,
|
||
created_by=self.admin_user,
|
||
)
|
||
|
||
request = mock.Mock()
|
||
request.user = self.manager_user
|
||
request.POST = {}
|
||
request.content_type = 'application/x-www-form-urlencoded'
|
||
request.build_absolute_uri = lambda uri: f"http://localhost{uri}"
|
||
|
||
# Manager finishes intervention without any photo attached
|
||
success, message, _ = attempt_status_update(
|
||
request,
|
||
intervention,
|
||
self.manager_user,
|
||
'finished',
|
||
)
|
||
|
||
self.assertTrue(success)
|
||
intervention.refresh_from_db()
|
||
self.assertEqual(intervention.status, 'finished')
|
||
|
||
def test_external_manager_allowed_to_finish_without_photos(self):
|
||
from interventions.views.workflow import attempt_status_update
|
||
|
||
intervention = Intervention.objects.create(
|
||
title='Intervention Ext Manager No Photo Test',
|
||
maintain_type='corrective',
|
||
status='in_progress',
|
||
status_order=STATUS_ORDERS['in_progress'],
|
||
contract=self.contract,
|
||
assigned_provider=self.company,
|
||
assigned_team=self.team_photos_required,
|
||
assigned_member=self.tech_member,
|
||
thematic=self.thematic,
|
||
symptom=self.symptom,
|
||
created_by=self.admin_user,
|
||
)
|
||
|
||
request = mock.Mock()
|
||
request.user = self.ext_manager_user
|
||
request.POST = {}
|
||
request.content_type = 'application/x-www-form-urlencoded'
|
||
request.build_absolute_uri = lambda uri: f"http://localhost{uri}"
|
||
|
||
# External Manager finishes intervention without any photo attached
|
||
success, message, _ = attempt_status_update(
|
||
request,
|
||
intervention,
|
||
self.ext_manager_user,
|
||
'finished',
|
||
)
|
||
|
||
self.assertTrue(success)
|
||
intervention.refresh_from_db()
|
||
self.assertEqual(intervention.status, 'finished')
|
||
|
||
|
||
|
||
|
||
|
||
|