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