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, UserContractAccess 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 OperationTemplateTests(TestCase): """Test the operation template functionality including template types.""" def setUp(self): from interventions.models import OperationTemplate, OperationTemplateItem, OperationStatusSet from common.models import Thematic # Create a test thematic self.thematic = Thematic.objects.create( code='test', name_fr='Test', name_nl='Test' ) # Create generic template self.generic_template = OperationTemplate.objects.create( name='Generic Inspection', description='Generic inspection template', thematic=self.thematic ) # Create equipment template self.equipment_template = OperationTemplate.objects.create( name='Equipment Check', description='Equipment check template', thematic=self.thematic ) # Create status set self.status_set = OperationStatusSet.objects.create(name='Standard') # Add items to templates OperationTemplateItem.objects.create( template=self.generic_template, name_fr='Vérification générale', name_nl='Algemene controle', status_set=self.status_set, order=1 ) OperationTemplateItem.objects.create( template=self.equipment_template, name_fr='Vérification équipement', name_nl='Apparatuurcontrole', status_set=self.status_set, order=1 ) def test_template_items_creation(self): """Test that template items are created correctly.""" generic_items = self.generic_template.items.all() equipment_items = self.equipment_template.items.all() self.assertEqual(generic_items.count(), 1) self.assertEqual(equipment_items.count(), 1) self.assertEqual(generic_items.first().name_fr, 'Vérification générale') self.assertEqual(equipment_items.first().name_fr, 'Vérification équipement') def test_operation_with_template_group(self): """Test that operations can be created with template group.""" from interventions.models import Intervention, Operation intervention = Intervention.objects.create( title='Test Intervention', status='in_preparation', maintain_type='preventive' ) # Create operation with template group operation = Operation.objects.create( intervention=intervention, name_fr='Test operation', name_nl='Test operatie', template_group=self.generic_template.name, order=1 ) self.assertEqual(operation.template_group, 'Generic Inspection') self.assertEqual(intervention.operations.count(), 1) # Test ordering by template_group operation2 = Operation.objects.create( intervention=intervention, name_fr='Test operation 2', name_nl='Test operatie 2', template_group=self.equipment_template.name, order=1 ) operations = intervention.operations.all() self.assertEqual(operations.count(), 2) # Operations should be ordered by template_group first self.assertEqual(operations[0].template_group, 'Equipment Check') self.assertEqual(operations[1].template_group, 'Generic Inspection') def test_get_asset_model_filters(self): """Test that get_asset_model_filters returns the correct dictionary with asset model IDs.""" from interventions.models import OperationTemplateAssetModelFilter from django.contrib.contenttypes.models import ContentType from common.models import Thematic ct = ContentType.objects.get_for_model(Thematic) # Create a filter entry filter_entry = OperationTemplateAssetModelFilter.objects.create( template=self.equipment_template, asset_model_content_type=ct, asset_model_object_id=999 ) filters = self.equipment_template.get_asset_model_filters() self.assertIsNotNone(filters) self.assertIn('thematic', filters) self.assertIn(999, filters['thematic']) class OperationGroupingTests(TestCase): def setUp(self): self.intervention = Intervention.objects.create( title='Grouped intervention', status='in_preparation', maintain_type='preventive' ) self.company = Company.objects.create(name='Asset Company') self.asset_ct = ContentType.objects.get_for_model(Company) self.intervention_asset = InterventionAsset.objects.create( intervention=self.intervention, content_type=self.asset_ct, object_id=self.company.id, ) def test_operations_grouped_by_asset_and_template(self): op_asset_a = Operation.objects.create( intervention=self.intervention, name_fr='Contrôle câble', name_nl='Kabelcontrole', template_group='Template A', order=1, asset_content_type=self.asset_ct, asset_object_id=self.company.id, ) op_asset_b = Operation.objects.create( intervention=self.intervention, name_fr='Contrôle boîtier', name_nl='Kastcontrole', template_group='Template B', order=2, asset_content_type=self.asset_ct, asset_object_id=self.company.id, ) op_generic = Operation.objects.create( intervention=self.intervention, name_fr='Inspection générale', name_nl='Algemene inspectie', template_group=None, order=3, ) linked_assets = [self.intervention_asset] operations = [op_asset_a, op_asset_b, op_generic] grouped = _group_operations_by_asset_and_template(operations, linked_assets) self.assertEqual(len(grouped), 2) self.assertFalse(grouped[0]['has_asset']) asset_group = next(group for group in grouped if group['has_asset']) self.assertEqual(asset_group['asset'], self.company) self.assertEqual(len(asset_group['template_groups']), 2) self.assertListEqual( [tg['name'] for tg in asset_group['template_groups']], ['Template A', 'Template B'] ) self.assertEqual(asset_group['template_groups'][0]['operations'][0], op_asset_a) generic_group = next(group for group in grouped if not group['has_asset']) self.assertEqual(len(generic_group['template_groups']), 1) self.assertIsNone(generic_group['template_groups'][0]['name']) self.assertEqual(generic_group['template_groups'][0]['operations'][0], op_generic) class PreventiveInterventionFormTests(TestCase): def test_preventive_intervention_form_status_choices(self): from interventions.views.forms import PreventiveInterventionForm # Test default choices (type = 'care' or default) form_default = PreventiveInterventionForm() status_field_default = form_default.fields['status'] choices_default = [val for val, label in status_field_default.choices] self.assertIn('in_preparation', choices_default) self.assertIn('to_be_approved', choices_default) self.assertIn('to_be_planned', choices_default) self.assertNotIn('to_be_processed', choices_default) self.assertNotIn('assigned', choices_default) # Test choices when type = 'inspection' in initial data form_inspection = PreventiveInterventionForm(initial={'type': 'inspection'}) status_field_inspection = form_inspection.fields['status'] choices_inspection = [val for val, label in status_field_inspection.choices] self.assertIn('in_preparation', choices_inspection) self.assertIn('to_be_approved', choices_inspection) self.assertIn('to_be_planned', choices_inspection) self.assertIn('to_be_processed', choices_inspection) self.assertIn('assigned', choices_inspection) # Test choices when type = 'inspection' in submitted data (POST) form_post = PreventiveInterventionForm(data={'type': 'inspection'}) status_field_post = form_post.fields['status'] choices_post = [val for val, label in status_field_post.choices] self.assertIn('in_preparation', choices_post) self.assertIn('to_be_approved', choices_post) self.assertIn('to_be_planned', choices_post) self.assertIn('to_be_processed', choices_post) self.assertIn('assigned', choices_post) class PreventiveDraftFormDataTests(TestCase): def test_multiple_operation_templates_are_preserved(self): User = get_user_model() user = User.objects.create_user(username='preventive_creator', password='pwd') role_admin, _ = Role.objects.get_or_create(name='admin') user_config = UserConfig.objects.create(user=user, is_intern=True) user_config.roles.add(role_admin) thematic = Thematic.objects.create(code='test', name_fr='Test', name_nl='Test') UserThematics.objects.create( user_config=user_config, thematic=thematic, can_view_assets=True, can_edit_interventions=True, ) self.client.login(username='preventive_creator', password='pwd') url = reverse('interventions:intervention_preventive_draft', kwargs={'thematic_code': 'test'}) response = self.client.post(url, { 'title': 'Préventive', 'operation_templates': ['1', '2'], }) self.assertEqual(response.status_code, 302) session = self.client.session drafts = session.get('intervention_drafts', {}) self.assertEqual(len(drafts), 1) draft_data = next(iter(drafts.values())) self.assertEqual(draft_data['form_data'].get('operation_templates'), ['1', '2']) class EquipmentTemplateOperationsTests(TestCase): def setUp(self): self.thematic = Thematic.objects.create(code='thema', name_fr='Thema', name_nl='Thema') self.status_set = OperationStatusSet.objects.create(name='Default') self.template = OperationTemplate.objects.create( name='Equipment Template', thematic=self.thematic, ) OperationTemplateItem.objects.create( template=self.template, name_fr='Inspection A', name_nl='Inspectie A', status_set=self.status_set, order=1, ) OperationTemplateItem.objects.create( template=self.template, name_fr='Inspection B', name_nl='Inspectie B', status_set=self.status_set, order=2, ) self.intervention = Intervention.objects.create( title='Preventive', status='in_preparation', maintain_type='preventive', ) self.company = Company.objects.create(name='Asset Company') self.asset_ct = ContentType.objects.get_for_model(Company) self.intervention.equipment_operation_templates.add(self.template) def _call_helper(self): return _create_equipment_operations_for_assets( self.intervention, [(self.asset_ct, self.company)], ) def test_operations_created_for_new_asset(self): created = self._call_helper() self.assertEqual(created, 2) operations = Operation.objects.filter( intervention=self.intervention, asset_content_type=self.asset_ct, asset_object_id=self.company.id, ).order_by('order') self.assertEqual(operations.count(), 2) self.assertListEqual( list(operations.values_list('name_fr', flat=True)), ['Inspection A', 'Inspection B'], ) self.assertTrue(all(op.template_group == self.template.name for op in operations)) def test_operations_not_duplicated_for_same_asset(self): first_run = self._call_helper() self.assertEqual(first_run, 2) second_run = self._call_helper() self.assertEqual(second_run, 0) operations = Operation.objects.filter( intervention=self.intervention, asset_content_type=self.asset_ct, asset_object_id=self.company.id, ) self.assertEqual(operations.count(), 2) class AssetOperationCleanupTests(TestCase): def setUp(self): self.User = get_user_model() self.user = self.User.objects.create_user(username='asset-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) self.thematic = Thematic.objects.create(code='road', name_fr='Route', name_nl='Weg') self.intervention = Intervention.objects.create( title='Asset cleanup', status='in_preparation', maintain_type='preventive', thematic=self.thematic, ) self.company = Company.objects.create(name='Linked asset company') self.asset_ct = ContentType.objects.get_for_model(Company) InterventionAsset.objects.create( intervention=self.intervention, content_type=self.asset_ct, object_id=self.company.id, ) self.status_set = OperationStatusSet.objects.create(name='Ops') self.status_ok = OperationStatus.objects.create( status_set=self.status_set, label_fr='OK', label_nl='OK', value='ok', order=1, ) self.operation_one = Operation.objects.create( intervention=self.intervention, name_fr='Op FR 1', name_nl='Op NL 1', order=1, template_group='Template A', asset_content_type=self.asset_ct, asset_object_id=self.company.id, ) self.operation_two = Operation.objects.create( intervention=self.intervention, name_fr='Op FR 2', name_nl='Op NL 2', order=2, template_group='Template A', asset_content_type=self.asset_ct, asset_object_id=self.company.id, ) self.url = reverse('interventions:intervention_link_assets', args=[self.intervention.id]) def _post_assets(self, assets): payload = json.dumps(assets) return self.client.post(self.url, {'assets': payload}) def test_operations_removed_when_asset_detached(self): self.client.login(username='asset-admin', password='pwd') response = self._post_assets([]) self.assertEqual(response.status_code, 302) remaining_assets = InterventionAsset.objects.filter(intervention=self.intervention) self.assertFalse(remaining_assets.exists()) self.assertFalse(Operation.objects.filter(intervention=self.intervention).exists()) success_messages = [m for m in get_messages(response.wsgi_request) if m.level == messages.SUCCESS] self.assertTrue(success_messages) def test_warning_emitted_when_operations_had_values(self): self.client.login(username='asset-admin', password='pwd') self.operation_one.status = self.status_ok self.operation_one.save(update_fields=['status']) response = self._post_assets([]) warning_messages = [m for m in get_messages(response.wsgi_request) if m.level == messages.WARNING] self.assertTrue(warning_messages) # Aucun nettoyage car au moins une opération contenait des valeurs remaining_ops = Operation.objects.filter(intervention=self.intervention) self.assertEqual(remaining_ops.count(), 2) class ExternalManagerPreventivePermissionsTests(TestCase): def setUp(self): User = get_user_model() self.role_ext_manager, _ = Role.objects.get_or_create(name='external_manager') self.role_manager, _ = Role.objects.get_or_create(name='manager') self.thematic = Thematic.objects.create(code='test_thema', name_fr='Test', name_nl='Test') self.provider = Company.objects.create(name="Test Provider") from datetime import date, timedelta today = date.today() self.contract_with_rights = Contract.objects.create( contract_number="C-001", company=self.provider, is_active=True, start_date=today - timedelta(days=30), end_date=today + timedelta(days=365) ) self.contract_with_rights.thematics.add(self.thematic) self.contract_without_rights = Contract.objects.create( contract_number="C-002", company=self.provider, is_active=True, start_date=today - timedelta(days=30), end_date=today + timedelta(days=365) ) self.contract_without_rights.thematics.add(self.thematic) # 1. User with creation rights self.user_with_rights = User.objects.create_user(username='ext_with_rights', password='pwd') self.uc_with_rights = UserConfig.objects.create(user=self.user_with_rights, is_intern=False) self.uc_with_rights.roles.add(self.role_ext_manager) UserThematics.objects.create( user_config=self.uc_with_rights, thematic=self.thematic, can_view_assets=True, can_edit_interventions=True, ) UserContractAccess.objects.create( user_config=self.uc_with_rights, contract=self.contract_with_rights, can_create_interventions=True ) # 2. User without creation rights self.user_without_rights = User.objects.create_user(username='ext_without_rights', password='pwd') self.uc_without_rights = UserConfig.objects.create(user=self.user_without_rights, is_intern=False) self.uc_without_rights.roles.add(self.role_ext_manager) UserThematics.objects.create( user_config=self.uc_without_rights, thematic=self.thematic, can_view_assets=True, can_edit_interventions=True, ) UserContractAccess.objects.create( user_config=self.uc_without_rights, contract=self.contract_without_rights, can_create_interventions=False ) # 3. An internal manager to view the detail page and verify the badge self.internal_manager = User.objects.create_user(username='int_manager', password='pwd') self.uc_int = UserConfig.objects.create(user=self.internal_manager, is_intern=True) self.uc_int.roles.add(self.role_manager) def test_access_to_add_preventive_view(self): # User with rights should access self.client.login(username='ext_with_rights', password='pwd') url = reverse('interventions:intervention_add_preventive', kwargs={'thematic_code': 'test_thema'}) # We need to set up a mock draft in the session session = self.client.session session['intervention_drafts'] = { 'draft-1': { 'locations': [], 'assets': [], 'form_data': {} } } session.save() response = self.client.get(f"{url}?draft_id=draft-1") self.assertEqual(response.status_code, 200) # User without rights should be Forbidden (403) self.client.login(username='ext_without_rights', password='pwd') response = self.client.get(f"{url}?draft_id=draft-1") self.assertEqual(response.status_code, 403) def test_contracts_filtered_in_form(self): self.client.login(username='ext_with_rights', password='pwd') url = reverse('interventions:intervention_add_preventive', kwargs={'thematic_code': 'test_thema'}) session = self.client.session session['intervention_drafts'] = { 'draft-1': { 'locations': [], 'assets': [], 'form_data': {} } } session.save() response = self.client.get(f"{url}?draft_id=draft-1") self.assertEqual(response.status_code, 200) # Verify that contract_with_rights is in form's queryset, but contract_without_rights is not contracts_queryset = response.context['form'].fields['contract'].queryset self.assertIn(self.contract_with_rights, contracts_queryset) self.assertNotIn(self.contract_without_rights, contracts_queryset) def test_preventive_intervention_created_by_provider(self): self.client.login(username='ext_with_rights', password='pwd') url = reverse('interventions:intervention_add_preventive', kwargs={'thematic_code': 'test_thema'}) session = self.client.session session['intervention_drafts'] = { 'draft-1': { 'locations': [], 'assets': [], 'form_data': {} } } session.save() # Submit creation form post_data = { 'title': 'Test Preventive Creation', 'status': 'in_preparation', 'type': 'care', 'contract': self.contract_with_rights.id, 'expected_begin_time': '2026-08-13T12:00', 'expected_end_time': '2026-08-13T13:00', 'recurrence_count': 0, 'type_of_selected_elements': 'locations', 'intervention_creation_mode': 'single_for_all', 'draft_id': 'draft-1', } response = self.client.post(url, post_data) self.assertEqual(response.status_code, 302) # Check created intervention intervention = Intervention.objects.filter(title='Test Preventive Creation').first() self.assertIsNotNone(intervention) self.assertEqual(intervention.created_by, self.user_with_rights) self.assertTrue(intervention.created_by_provider) self.assertEqual(intervention.maintain_type, 'preventive') # Check that the badge is shown for internal manager from interventions.permissions import should_show_created_by_provider_badge self.assertTrue(should_show_created_by_provider_badge(self.internal_manager, intervention))