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')