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 ExportInterventionsTests(TestCase): def setUp(self): User = get_user_model() self.user = User.objects.create_user(username='admin', password='pwd') # Give admin role to see everything 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.intv1 = Intervention.objects.create(title='Intervention A', status='to_be_processed') self.intv2 = Intervention.objects.create(title='Intervention B', status='in_progress') self.url = reverse('interventions:export_interventions') def test_export_csv(self): self.client.login(username='admin', password='pwd') import json resp = self.client.post(self.url, { 'ids': json.dumps([self.intv1.id, self.intv2.id]), 'format': 'csv' }) self.assertEqual(resp.status_code, 200) self.assertIn('text/csv', resp['Content-Type']) content = resp.content.decode('utf-8') self.assertIn('Intervention A', content) self.assertIn('Intervention B', content) def test_export_xlsx(self): self.client.login(username='admin', password='pwd') import json resp = self.client.post(self.url, { 'ids': json.dumps([self.intv1.id]), 'format': 'xlsx' }) self.assertEqual(resp.status_code, 200) self.assertIn('application/vnd.openxmlformats-officedocument.spreadsheetml.sheet', resp['Content-Type']) # XLSX starts with PK zip signature self.assertTrue(resp.content.startswith(b'PK')) class InterventionDocumentDeletionTests(TestCase): def setUp(self): User = get_user_model() self.owner = User.objects.create_user(username='doc-owner', password='pwd') self.manager = User.objects.create_user(username='doc-manager', password='pwd') manager_role = Role.objects.create(name='manager') self.manager_config = UserConfig.objects.create(user=self.manager, is_intern=True) self.manager_config.roles.add(manager_role) self.intervention = Intervention.objects.create(title='Doc intervention') self.document = InterventionDocument.objects.create( intervention=self.intervention, file=SimpleUploadedFile('document.txt', b'document content'), uploaded_by=self.owner, ) def test_internal_manager_can_delete_intervention_document(self): self.assertTrue(can_delete_document(self.manager, self.manager_config, self.document)) class PostsBulkUpdateTests(TestCase): def setUp(self): User = get_user_model() self.user = User.objects.create_user(username='provider', password='pwd') self.role_external = Role.objects.create(name='external_manager') self.thematic = Thematic.objects.create(code='roads', name_fr='Routes', name_nl='Wegen') self.user_config = UserConfig.objects.create(user=self.user, is_intern=True, default_thematic=self.thematic) self.user_config.roles.add(self.role_external) 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='Provider Co') self.team = CompanyTeam.objects.create(name='Provider Team', company=self.company) self.member = CompanyMember.objects.create(user=self.user, name='Provider Member') self.member.teams.add(self.team) from datetime import date, timedelta today = date.today() self.contract = Contract.objects.create( company=self.company, contract_number='C-POSTS', start_date=today - timedelta(days=30), end_date=today + timedelta(days=365) ) self.intervention = Intervention.objects.create( title='Bulk posts', status='to_be_processed', contract=self.contract, thematic=self.thematic, ) from common.models import UserContractAccess, UserContractStatusPermission self.uca = UserContractAccess.objects.create( user_config=self.user_config, contract=self.contract, ) UserContractStatusPermission.objects.create( user_contract=self.uca, status='to_be_processed', can_edit=True, ) self.post_existing = ContractPost.objects.create( contract=self.contract, order_number=1, code='EXIST', description_fr='Poste existant', description_nl='Bestaande post', unit='u', unit_price=Decimal('10.00') ) self.post_new = ContractPost.objects.create( contract=self.contract, order_number=2, code='NEW', description_fr='Nouveau poste', description_nl='Nieuwe post', unit='u', unit_price=Decimal('15.00') ) self.url = reverse('interventions:intervention_posts_bulk_update', args=[self.intervention.id]) def _post(self, payload): return self.client.post(self.url, data=json.dumps(payload), content_type='application/json') def test_provider_cannot_override_control_comment(self): InterventionContractPost.objects.create( intervention=self.intervention, contract_post=self.post_existing, quantity=Decimal('1.00'), control_comment='Controller only', ) self.client.login(username='provider', password='pwd') response = self._post({ 'posts': [{ 'contract_post_id': self.post_existing.id, 'quantity': '7.00', 'control_comment': 'Provider comment', }] }) self.assertEqual(response.status_code, 200) payload = response.json() self.assertTrue(payload.get('success')) post = InterventionContractPost.objects.get(intervention=self.intervention, contract_post=self.post_existing) self.assertEqual(post.quantity, Decimal('7.00')) self.assertEqual(post.control_comment, 'Controller only') def test_zero_quantity_new_post_is_ignored(self): self.client.login(username='provider', password='pwd') response = self._post({ 'posts': [{ 'contract_post_id': self.post_new.id, 'quantity': '0', 'comment': 'Should not be saved', }] }) self.assertEqual(response.status_code, 200) payload = response.json() self.assertTrue(payload.get('success')) self.assertEqual(payload.get('updated_count'), 0) self.assertFalse( InterventionContractPost.objects.filter(intervention=self.intervention, contract_post=self.post_new).exists() ) class InterventionCategoryUpdateTests(TestCase): def setUp(self): User = get_user_model() self.user = User.objects.create_user(username='category-editor', 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') self.category = AssetCategory.objects.create( code='RD-1', name_fr='Chaussée', name_nl='Wegdek', thematic=self.thematic, ) self.intervention = Intervention.objects.create( title='Editable category', thematic=self.thematic, status='to_be_processed', status_order=STATUS_ORDERS['to_be_processed'], ) self.url = reverse('interventions:update_intervention', args=[self.intervention.id]) self.client.login(username='category-editor', password='pwd') def test_category_can_be_updated_before_validation(self): response = self.client.post( self.url, data=json.dumps({'category': str(self.category.id)}), content_type='application/json', ) self.assertEqual(response.status_code, 200) payload = response.json() self.assertTrue(payload.get('success')) self.intervention.refresh_from_db() self.assertEqual(self.intervention.asset_category, self.category) def test_category_update_forbidden_after_validation(self): validated = Intervention.objects.create( title='Locked category', thematic=self.thematic, status='validated', status_order=STATUS_ORDERS['validated'], ) url = reverse('interventions:update_intervention', args=[validated.id]) response = self.client.post( url, data=json.dumps({'category': str(self.category.id)}), content_type='application/json', ) self.assertEqual(response.status_code, 403) validated.refresh_from_db() self.assertIsNone(validated.asset_category) class OperationCommentUpdateTests(TestCase): def setUp(self): self.User = get_user_model() self.admin_role = Role.objects.create(name='admin') self.admin_user = self.User.objects.create_user(username='op-admin', password='pwd') self.admin_config = UserConfig.objects.create(user=self.admin_user, is_intern=True) self.admin_config.roles.add(self.admin_role) self.readonly_user = self.User.objects.create_user(username='op-readonly', password='pwd') UserConfig.objects.create(user=self.readonly_user, is_intern=False) self.intervention = Intervention.objects.create(title='Operation comment', status='in_progress') self.operation = Operation.objects.create( intervention=self.intervention, name_fr='Contrôle visuel', name_nl='Visuele controle', ) self.url = reverse('interventions:operation_update_comment', args=[self.operation.id]) def test_admin_can_update_comment(self): self.client.login(username='op-admin', password='pwd') payload = {'comment': 'Nettoyé le 12/09'} response = self.client.post(self.url, data=json.dumps(payload), content_type='application/json') self.assertEqual(response.status_code, 200) body = response.json() self.assertTrue(body.get('success')) self.operation.refresh_from_db() self.assertEqual(self.operation.comment, payload['comment']) def test_update_comment_requires_permission(self): self.client.login(username='op-readonly', password='pwd') response = self.client.post(self.url, data=json.dumps({'comment': 'Test'}), content_type='application/json') self.assertEqual(response.status_code, 403) self.operation.refresh_from_db() self.assertEqual(self.operation.comment, '') class AdminInterventionFiltersTests(TestCase): def setUp(self): User = get_user_model() self.user = User.objects.create_user(username='admin_filter_user', 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) # Create some contracts and providers self.company1 = Company.objects.create(name='Company A') self.company2 = Company.objects.create(name='Company B') from datetime import date, timedelta today = date.today() self.contract1 = Contract.objects.create( company=self.company1, contract_number='C-FILTER-1', start_date=today - timedelta(days=30), end_date=today + timedelta(days=365) ) self.contract2 = Contract.objects.create( company=self.company2, contract_number='C-FILTER-2', start_date=today - timedelta(days=30), end_date=today + timedelta(days=365) ) self.url = reverse('interventions:interventions_index') def test_admin_sees_all_contracts_and_providers(self): self.client.login(username='admin_filter_user', password='pwd') response = self.client.get(self.url, follow=True) self.assertEqual(response.status_code, 200) # Check that provider choices and contract choices in the context contain all contracts/providers context = response.context provider_choices = context['provider_choices'] contract_choices = context['contract_choices'] provider_ids = [p[0] for p in provider_choices] contract_ids = [c[0] for c in contract_choices] self.assertIn(self.company1.id, provider_ids) self.assertIn(self.company2.id, provider_ids) self.assertIn(self.contract1.id, contract_ids) self.assertIn(self.contract2.id, contract_ids) def test_non_intern_user_can_access_index(self): non_intern_user = get_user_model().objects.create_user(username='external_filter_user', password='pwd') UserConfig.objects.create(user=non_intern_user, is_intern=False) self.client.login(username='external_filter_user', password='pwd') response = self.client.get(self.url, {'status': 'assigned'}) self.assertEqual(response.status_code, 200) class OverdueInterventionTests(TestCase): def setUp(self): User = get_user_model() self.user = User.objects.create_user(username='overdue_tester', 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='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.client.login(username='overdue_tester', password='pwd') def test_overdue_filters_exclude_on_pause(self): # Create an intervention that has been on_pause for more than 15 days from interventions.models import InterventionTimeLine past_time = timezone.now() - timedelta(days=20) itv = Intervention.objects.create( title='Paused intervention', status='on_pause', thematic=self.thematic, ) InterventionTimeLine.objects.create( intervention=itv, event_time=past_time, event_type='status_change', to_status='on_pause', ) # Query overdue from django.http import QueryDict from interventions.views import apply_intervention_filters_from_querydict qs = Intervention.objects.all() filtered_qs = apply_intervention_filters_from_querydict(qs, QueryDict('overdue=1')) self.assertNotIn(itv, filtered_qs) def test_overdue_filters_rescheduled_ignored(self): # Create an intervention that was started in the past (so begin_time is populated) # but was recently paused and then rescheduled to a future date. past_time = timezone.now() - timedelta(days=10) future_time = timezone.now() + timedelta(days=5) itv = Intervention.objects.create( title='Rescheduled intervention', status='to_be_processed', begin_time=past_time, planned_begin_time=future_time, thematic=self.thematic, ) # Query overdue from django.http import QueryDict from interventions.views import apply_intervention_filters_from_querydict qs = Intervention.objects.all() filtered_qs = apply_intervention_filters_from_querydict(qs, QueryDict('overdue=1')) self.assertNotIn(itv, filtered_qs) def test_overdue_filters_24h_buffer_planned(self): # Create an intervention planned 12 hours ago (less than 24h) recent_past = timezone.now() - timedelta(hours=12) itv_recent = Intervention.objects.create( title='Recently planned intervention', status='to_be_processed', planned_begin_time=recent_past, thematic=self.thematic, ) # Create an intervention planned 36 hours ago (more than 24h) older_past = timezone.now() - timedelta(hours=36) itv_older = Intervention.objects.create( title='Older planned intervention', status='to_be_processed', planned_begin_time=older_past, thematic=self.thematic, ) from django.http import QueryDict from interventions.views import apply_intervention_filters_from_querydict qs = Intervention.objects.all() filtered_qs = apply_intervention_filters_from_querydict(qs, QueryDict('overdue=1')) self.assertNotIn(itv_recent, filtered_qs) self.assertIn(itv_older, filtered_qs) class InterventionManagerTests(TestCase): def setUp(self): User = get_user_model() self.user = User.objects.create_user(username='manager_user', password='pwd') self.role_admin = Role.objects.get_or_create(name='admin')[0] 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, ) from interventions.models import Symptom self.symptom = Symptom.objects.create( code='roads_defect', name_fr='Défaut voirie', name_nl='Wegdefect', thematic=self.thematic, ) self.company = Company.objects.create(name='Test Company') from datetime import date self.contract = Contract.objects.create( company=self.company, contract_number='C-MGR', start_date=date.today(), end_date=date.today() + timedelta(days=365), uses_intervention_manager=True ) from common.models import UserContractAccess UserContractAccess.objects.create( user_config=self.user_config, contract=self.contract, can_approve=True, can_view_interventions=True, ) self.client.login(username='manager_user', password='pwd') def test_manager_assignment_on_approval(self): intervention = Intervention.objects.create( title='Approval Test', status='to_be_approved', contract=self.contract, thematic=self.thematic, symptom=self.symptom, ) from interventions.views import attempt_status_update # Mock request request = mock.Mock() success, msg, reload = attempt_status_update(request, intervention, self.user, 'to_be_processed') self.assertTrue(success) intervention.refresh_from_db() self.assertEqual(intervention.intervention_manager, self.user) def test_manager_not_assigned_if_uses_intervention_manager_false(self): self.contract.uses_intervention_manager = False self.contract.save() intervention = Intervention.objects.create( title='Approval Test False', status='to_be_approved', contract=self.contract, thematic=self.thematic, symptom=self.symptom, ) from interventions.views import attempt_status_update request = mock.Mock() success, msg, reload = attempt_status_update(request, intervention, self.user, 'to_be_processed') self.assertTrue(success) intervention.refresh_from_db() self.assertIsNone(intervention.intervention_manager) def test_manager_not_assigned_if_canceled(self): intervention = Intervention.objects.create( title='Cancel Test', status='to_be_approved', contract=self.contract, thematic=self.thematic, symptom=self.symptom, ) from interventions.views import attempt_status_update request = mock.Mock() success, msg, reload = attempt_status_update( request, intervention, self.user, 'canceled', cancellation_type='other', cancellation_reason='test cancel' ) self.assertTrue(success) intervention.refresh_from_db() self.assertIsNone(intervention.intervention_manager) def test_homepage_finished_interventions_filtering(self): # Create an intervention finished, with contract using manager, but manager is None (not this user) itv_other_mgr = Intervention.objects.create( title='Other Manager finished', status='finished', contract=self.contract, thematic=self.thematic, intervention_manager=None, symptom=self.symptom, ) # Create an intervention finished, with contract using manager, managed by self.user itv_self_mgr = Intervention.objects.create( title='Self Manager finished', status='finished', contract=self.contract, thematic=self.thematic, intervention_manager=self.user, symptom=self.symptom, ) # Create an intervention finished, but contract does not use manager contract_no_mgr = Contract.objects.create( company=self.company, contract_number='C-NOMGR', start_date=self.contract.start_date, end_date=self.contract.end_date, uses_intervention_manager=False ) # Grant view permission for the no-manager contract from common.models import UserContractAccess UserContractAccess.objects.create( user_config=self.user_config, contract=contract_no_mgr, can_view_interventions=True, ) itv_no_mgr = Intervention.objects.create( title='No Manager finished', status='finished', contract=contract_no_mgr, thematic=self.thematic, symptom=self.symptom, ) # Access the homepage response = self.client.get(reverse('home:index')) self.assertEqual(response.status_code, 200) # Finished list/counter should only contain self-managed and no-manager interventions finished_list = response.context['intervention_finished_list'] self.assertIn(itv_self_mgr, finished_list) self.assertIn(itv_no_mgr, finished_list) self.assertNotIn(itv_other_mgr, finished_list) self.assertEqual(response.context['intervention_finished_count'], 2) self.assertEqual(response.context['total_finished_count'], 3) self.assertTrue(response.context['is_manager_filter_applied']) # Test when no contracts use intervention manager self.contract.uses_intervention_manager = False self.contract.save() response2 = self.client.get(reverse('home:index')) self.assertFalse(response2.context['is_manager_filter_applied']) def test_geojson_filtering_by_manager(self): # Create an intervention finished managed by self.user itv_self_mgr = Intervention.objects.create( title='Self Manager', status='finished', contract=self.contract, thematic=self.thematic, intervention_manager=self.user, symptom=self.symptom, ) # Create an intervention finished with no manager itv_no_mgr = Intervention.objects.create( title='No Manager', status='finished', contract=self.contract, thematic=self.thematic, intervention_manager=None, symptom=self.symptom, ) # Access the geojson view passing our manager ID response = self.client.get( reverse('interventions:interventions_geojson') + f'?intervention_manager={self.user.id}' ) self.assertEqual(response.status_code, 200) data = response.json() # Only the intervention managed by self.user should be present in the features feature_codes = [f['properties']['code'] for f in data['features']] self.assertIn(itv_self_mgr.code, feature_codes) self.assertNotIn(itv_no_mgr.code, feature_codes) def test_pdf_report_contains_intervention_manager(self): itv = Intervention.objects.create( title='PDF Report Test', status='finished', contract=self.contract, thematic=self.thematic, intervention_manager=self.user, symptom=self.symptom, ) from interventions.views.views_pdf_reporting import InterventionReportPDF pdf = InterventionReportPDF() pdf.add_page() pdf.add_intervention_report_data(itv) output = pdf.output(dest='S') self.assertIsNotNone(output) def test_geojson_filtering_by_assigned_team(self): from contracts.models import CompanyTeam, CompanyMember from django.http import QueryDict from interventions.views import apply_intervention_filters_from_querydict # Create two teams team1 = CompanyTeam.objects.create(name='Team 1', company=self.company) team2 = CompanyTeam.objects.create(name='Team 2', company=self.company) # Create a member and assign them to team1 member1 = CompanyMember.objects.create(name='Member 1', user=self.user) member1.teams.add(team1) # Intervention 1: assigned to team2 directly itv_team2 = Intervention.objects.create( title='Itv Team 2', status='finished', contract=self.contract, thematic=self.thematic, assigned_team=team2, symptom=self.symptom, ) # Intervention 2: assigned to member1 (who is in team1) itv_member1 = Intervention.objects.create( title='Itv Member 1', status='finished', contract=self.contract, thematic=self.thematic, assigned_member=member1, symptom=self.symptom, ) # Intervention 3: no assignment itv_none = Intervention.objects.create( title='Itv None', status='finished', contract=self.contract, thematic=self.thematic, symptom=self.symptom, ) # 1. Access the geojson view passing team2 response = self.client.get( reverse('interventions:interventions_geojson') + f'?assigned_team={team2.id}' ) self.assertEqual(response.status_code, 200) data = response.json() feature_codes = [f['properties']['code'] for f in data['features']] self.assertIn(itv_team2.code, feature_codes) self.assertNotIn(itv_member1.code, feature_codes) self.assertNotIn(itv_none.code, feature_codes) # 2. Access the geojson view passing team1 (which the member is part of) response = self.client.get( reverse('interventions:interventions_geojson') + f'?assigned_team={team1.id}' ) self.assertEqual(response.status_code, 200) data = response.json() feature_codes = [f['properties']['code'] for f in data['features']] self.assertIn(itv_member1.code, feature_codes) self.assertNotIn(itv_team2.code, feature_codes) self.assertNotIn(itv_none.code, feature_codes) # 3. Access the geojson view passing both teams response = self.client.get( reverse('interventions:interventions_geojson') + f'?assigned_team={team1.id},{team2.id}' ) self.assertEqual(response.status_code, 200) data = response.json() feature_codes = [f['properties']['code'] for f in data['features']] self.assertIn(itv_team2.code, feature_codes) self.assertIn(itv_member1.code, feature_codes) self.assertNotIn(itv_none.code, feature_codes) # 4. Verify apply_intervention_filters_from_querydict qs = Intervention.objects.all() filtered_qs = apply_intervention_filters_from_querydict(qs, QueryDict(f'assigned_team={team2.id}')) self.assertIn(itv_team2, filtered_qs) self.assertNotIn(itv_member1, filtered_qs) filtered_qs = apply_intervention_filters_from_querydict(qs, QueryDict(f'assigned_team={team1.id}')) self.assertIn(itv_member1, filtered_qs) self.assertNotIn(itv_team2, filtered_qs) class InterventionAssignmentTests(TestCase): def setUp(self): from contracts.models import Company, CompanyTeam, CompanyMember from common.models import Role, UserConfig, Thematic, UserThematics self.User = get_user_model() self.user = self.User.objects.create_user(username='test-user', 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.company = Company.objects.create(name='Test Company') self.team_a = CompanyTeam.objects.create(name='Team A', company=self.company) self.team_b = CompanyTeam.objects.create(name='Team B', company=self.company) self.member = CompanyMember.objects.create(user=self.user, name='John Doe') self.member.teams.add(self.team_a) 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.client.login(username='test-user', password='pwd') @mock.patch('django.utils.timezone.now') def test_sync_daily_assignments_creation_and_update(self, mock_now): from datetime import datetime, date from interventions.models import Intervention, InterventionAssignment import pytz # Fix today to 2026-06-11 mock_now.return_value = datetime(2026, 6, 11, 12, 0, 0, tzinfo=pytz.UTC) # Create intervention planned from 2026-06-09 to 2026-06-12 (2 past days, today, 1 future day) itv = Intervention.objects.create( title='Test Assignment', status='to_be_processed', assigned_team=self.team_a, assigned_member=self.member, planned_begin_time=datetime(2026, 6, 9, 8, 0, 0, tzinfo=pytz.UTC), planned_end_time=datetime(2026, 6, 12, 17, 0, 0, tzinfo=pytz.UTC), thematic=self.thematic, ) # 4 assignments should be created (June 9, 10, 11, 12) assignments = InterventionAssignment.objects.filter(intervention=itv).order_by('date') self.assertEqual(assignments.count(), 4) dates = [a.date for a in assignments] self.assertEqual(dates, [ date(2026, 6, 9), date(2026, 6, 10), date(2026, 6, 11), date(2026, 6, 12) ]) for a in assignments: self.assertEqual(a.team, self.team_a) self.assertEqual(a.member, self.member) self.assertEqual(a.status, 'to_be_processed') # Now, modify the intervention: reassign to team_b and change status to 'finished' itv.assigned_team = self.team_b itv.assigned_member = None itv.status = 'finished' itv.save() # Check history preservation # For June 9 and 10 (past), team, member, status should remain Team A, John Doe, 'to_be_processed' for d in [date(2026, 6, 9), date(2026, 6, 10)]: a = InterventionAssignment.objects.get(intervention=itv, date=d) self.assertEqual(a.team, self.team_a) self.assertEqual(a.member, self.member) self.assertEqual(a.status, 'to_be_processed') # For June 11 (today) and June 12 (future), team, member, status should be updated to Team B, None, 'finished' for d in [date(2026, 6, 11), date(2026, 6, 12)]: a = InterventionAssignment.objects.get(intervention=itv, date=d) self.assertEqual(a.team, self.team_b) self.assertIsNone(a.member) self.assertEqual(a.status, 'finished') @mock.patch('django.utils.timezone.now') def test_sync_daily_assignments_cleanup(self, mock_now): from datetime import datetime, date from interventions.models import Intervention, InterventionAssignment import pytz # Fix today to 2026-06-11 mock_now.return_value = datetime(2026, 6, 11, 12, 0, 0, tzinfo=pytz.UTC) # Create intervention planned from 2026-06-11 to 2026-06-13 itv = Intervention.objects.create( title='Test Assignment Cleanup', status='to_be_processed', assigned_team=self.team_a, planned_begin_time=datetime(2026, 6, 11, 8, 0, 0, tzinfo=pytz.UTC), planned_end_time=datetime(2026, 6, 13, 17, 0, 0, tzinfo=pytz.UTC), thematic=self.thematic, ) self.assertEqual(InterventionAssignment.objects.filter(intervention=itv).count(), 3) # Now reschedule to only today (2026-06-11) itv.planned_end_time = datetime(2026, 6, 11, 17, 0, 0, tzinfo=pytz.UTC) itv.save() # June 12 and 13 (future dates) should be deleted, keeping only June 11 assignments = InterventionAssignment.objects.filter(intervention=itv) self.assertEqual(assignments.count(), 1) self.assertEqual(assignments.first().date, date(2026, 6, 11)) @mock.patch('django.utils.timezone.now') def test_teams_daily_ajax_view(self, mock_now): from datetime import datetime, date from interventions.models import Intervention, InterventionAssignment from contracts.models import CompanyMemberHistory import pytz # Fix today to 2026-06-11 (Thursday) # Monday is 2026-06-08, Sunday is 2026-06-14 mock_now.return_value = datetime(2026, 6, 11, 12, 0, 0, tzinfo=pytz.UTC) # Clear any auto-generated history CompanyMemberHistory.objects.all().delete() # Create team member history: member belongs to team_a from 2026-06-08 to 2026-06-10 CompanyMemberHistory.objects.create( member=self.member, team=self.team_a, join_date=date(2026, 6, 8), leave_date=date(2026, 6, 10), ) # Create intervention assigned to team_a on 2026-06-09 itv = Intervention.objects.create( title='Daily Task', status='to_be_processed', assigned_team=self.team_a, planned_begin_time=datetime(2026, 6, 9, 8, 0, 0, tzinfo=pytz.UTC), planned_end_time=datetime(2026, 6, 9, 17, 0, 0, tzinfo=pytz.UTC), thematic=self.thematic, ) # Create another intervention that is unassigned/scheduled itv_unassigned = Intervention.objects.create( title='Unassigned Task', status='to_be_processed', planned_begin_time=datetime(2026, 6, 10, 8, 0, 0, tzinfo=pytz.UTC), planned_end_time=datetime(2026, 6, 10, 17, 0, 0, tzinfo=pytz.UTC), thematic=self.thematic, ) url = reverse('interventions:interventions_events') + '?view_mode=teams_daily&date=2026-06-11' response = self.client.get(url) self.assertEqual(response.status_code, 200) data = response.json() self.assertEqual(data['view_mode'], 'teams_daily') self.assertEqual(data['start_date'], '2026-06-08') self.assertEqual(data['end_date'], '2026-06-14') # Check teams data teams = data['teams'] # We expect team_a and team_b, plus "Sans équipe" because we have an unassigned intervention team_a_entry = next((t for t in teams if t['id'] == self.team_a.id), None) self.assertIsNotNone(team_a_entry) # On 2026-06-09, team_a should have John Doe in members, and itv in interventions day_06_09 = team_a_entry['days']['2026-06-09'] day_06_09_names = [m['name'] if isinstance(m, dict) else m for m in day_06_09['members']] self.assertIn('John Doe', day_06_09_names) self.assertEqual(len(day_06_09['interventions']), 1) self.assertEqual(day_06_09['interventions'][0]['id'], itv.id) # On 2026-06-11, team_a should NOT have John Doe (he left on 2026-06-10) day_06_11 = team_a_entry['days']['2026-06-11'] day_06_11_names = [m['name'] if isinstance(m, dict) else m for m in day_06_11['members']] self.assertNotIn('John Doe', day_06_11_names) # Check "Sans équipe" row unassigned_entry = next((t for t in teams if t['id'] is None), None) self.assertIsNotNone(unassigned_entry) # On 2026-06-10, itv_unassigned should be in "Sans équipe" row day_06_10 = unassigned_entry['days']['2026-06-10'] self.assertEqual(len(day_06_10['interventions']), 1) self.assertEqual(day_06_10['interventions'][0]['id'], itv_unassigned.id) @mock.patch('django.utils.timezone.now') def test_teams_daily_ajax_view_member_deduplication(self, mock_now): from datetime import datetime, date from contracts.models import CompanyMemberHistory import pytz # Fix today to 2026-06-11 mock_now.return_value = datetime(2026, 6, 11, 12, 0, 0, tzinfo=pytz.UTC) CompanyMemberHistory.objects.all().delete() # Create duplicate overlapping history for John Doe in team_a on 2026-06-09 CompanyMemberHistory.objects.create( member=self.member, team=self.team_a, join_date=date(2026, 6, 8), leave_date=date(2026, 6, 10), ) CompanyMemberHistory.objects.create( member=self.member, team=self.team_a, join_date=date(2026, 6, 9), leave_date=date(2026, 6, 9), ) url = reverse('interventions:interventions_events') + '?view_mode=teams_daily&date=2026-06-11' response = self.client.get(url) self.assertEqual(response.status_code, 200) data = response.json() teams = data['teams'] team_a_entry = next((t for t in teams if t['id'] == self.team_a.id), None) self.assertIsNotNone(team_a_entry) # On 2026-06-09, members list should have ONLY ONE entry for 'John Doe' day_06_09 = team_a_entry['days']['2026-06-09'] day_06_09_names = [m['name'] if isinstance(m, dict) else m for m in day_06_09['members']] self.assertEqual(day_06_09_names, ['John Doe']) @mock.patch('django.utils.timezone.now') def test_teams_daily_ajax_view_member_transfer_resolution(self, mock_now): from datetime import datetime, date from contracts.models import CompanyMemberHistory import pytz # Fix today to 2026-06-11 mock_now.return_value = datetime(2026, 6, 11, 12, 0, 0, tzinfo=pytz.UTC) CompanyMemberHistory.objects.all().delete() # John Doe leaves team_a on 2026-06-12 CompanyMemberHistory.objects.create( member=self.member, team=self.team_a, join_date=date(2026, 6, 8), leave_date=date(2026, 6, 12), ) # John Doe joins team_b on 2026-06-12 CompanyMemberHistory.objects.create( member=self.member, team=self.team_b, join_date=date(2026, 6, 12), leave_date=None, ) url = reverse('interventions:interventions_events') + '?view_mode=teams_daily&date=2026-06-11' response = self.client.get(url) self.assertEqual(response.status_code, 200) data = response.json() teams = data['teams'] team_a_entry = next((t for t in teams if t['id'] == self.team_a.id), None) team_b_entry = next((t for t in teams if t['id'] == self.team_b.id), None) self.assertIsNotNone(team_a_entry) self.assertIsNotNone(team_b_entry) # On 2026-06-11, John Doe should still be in team_a and NOT in team_b day_06_11_a = team_a_entry['days']['2026-06-11'] day_06_11_b = team_b_entry['days']['2026-06-11'] self.assertIn('John Doe', [m['name'] if isinstance(m, dict) else m for m in day_06_11_a['members']]) self.assertNotIn('John Doe', [m['name'] if isinstance(m, dict) else m for m in day_06_11_b['members']]) # On 2026-06-12 (transfer day), John Doe should NOT be in team_a and should be in team_b day_06_12_a = team_a_entry['days']['2026-06-12'] day_06_12_b = team_b_entry['days']['2026-06-12'] self.assertNotIn('John Doe', [m['name'] if isinstance(m, dict) else m for m in day_06_12_a['members']]) self.assertIn('John Doe', [m['name'] if isinstance(m, dict) else m for m in day_06_12_b['members']]) @mock.patch('django.utils.timezone.now') def test_teams_daily_ajax_view_member_concurrent_resolution(self, mock_now): from datetime import datetime, date from contracts.models import CompanyMemberHistory import pytz # Fix today to 2026-06-11 mock_now.return_value = datetime(2026, 6, 11, 12, 0, 0, tzinfo=pytz.UTC) CompanyMemberHistory.objects.all().delete() # John Doe belongs to team_a CompanyMemberHistory.objects.create( member=self.member, team=self.team_a, join_date=date(2026, 6, 8), leave_date=None, ) # John Doe also belongs to team_b CompanyMemberHistory.objects.create( member=self.member, team=self.team_b, join_date=date(2026, 6, 10), leave_date=None, ) url = reverse('interventions:interventions_events') + '?view_mode=teams_daily&date=2026-06-11' response = self.client.get(url) self.assertEqual(response.status_code, 200) data = response.json() teams = data['teams'] team_a_entry = next((t for t in teams if t['id'] == self.team_a.id), None) team_b_entry = next((t for t in teams if t['id'] == self.team_b.id), None) self.assertIsNotNone(team_a_entry) self.assertIsNotNone(team_b_entry) # On 2026-06-11, John Doe should be in both team_a and team_b day_06_11_a = team_a_entry['days']['2026-06-11'] day_06_11_b = team_b_entry['days']['2026-06-11'] self.assertIn('John Doe', [m['name'] if isinstance(m, dict) else m for m in day_06_11_a['members']]) self.assertIn('John Doe', [m['name'] if isinstance(m, dict) else m for m in day_06_11_b['members']]) class InterventionSubscriptionTests(TestCase): def setUp(self): from django.contrib.auth import get_user_model from common.models import Role, UserConfig, Thematic, UserThematics from contracts.models import Contract self.User = get_user_model() self.user1 = self.User.objects.create_user(username='sub-user-1', password='pwd') self.user2 = self.User.objects.create_user(username='sub-user-2', password='pwd') self.role_admin, _ = Role.objects.get_or_create(name='admin') self.config1 = UserConfig.objects.create(user=self.user1, is_intern=True) self.config1.roles.add(self.role_admin) self.config2 = UserConfig.objects.create(user=self.user2, is_intern=True) self.config2.roles.add(self.role_admin) self.thematic = Thematic.objects.create(code='roads_sub', name_fr='Voirie Sub', name_nl='Wegen Sub', allow_planning_without_contract=True) # Grant view/edit permissions to the thematic for both users UserThematics.objects.create( user_config=self.config1, thematic=self.thematic, can_view_interventions=True, can_edit_interventions=True, ) UserThematics.objects.create( user_config=self.config2, thematic=self.thematic, can_view_interventions=True, can_edit_interventions=True, ) from interventions.models import Intervention self.itv = Intervention.objects.create( title='Test Sub Intervention', status='in_preparation', thematic=self.thematic, created_by=self.user1 ) def test_toggle_subscription_api(self): from interventions.models import InterventionSubscription self.client.login(username='sub-user-1', password='pwd') url = f'/interventions/{self.itv.id}/toggle-subscription/' # 1. First call should subscribe response = self.client.post(url) self.assertEqual(response.status_code, 200) self.assertTrue(response.json()['subscribed']) self.assertTrue(InterventionSubscription.objects.filter(user=self.user1, intervention=self.itv).exists()) # 2. Second call should unsubscribe response = self.client.post(url) self.assertEqual(response.status_code, 200) self.assertFalse(response.json()['subscribed']) self.assertFalse(InterventionSubscription.objects.filter(user=self.user1, intervention=self.itv).exists()) def test_state_change_notifications(self): from interventions.models import InterventionSubscription, Intervention from notifications.models import NotificationRecipient # User 1 subscribes to the intervention InterventionSubscription.objects.create(user=self.user1, intervention=self.itv) # User 2 makes a change to status self.client.login(username='sub-user-2', password='pwd') url = f'/interventions/{self.itv.id}/update-status/' # In preparation to to_be_planned is an allowed transition for admin response = self.client.post(url, {'new_status': 'to_be_planned'}) self.assertEqual(response.status_code, 200) self.itv.refresh_from_db() self.assertEqual(self.itv.status, 'to_be_planned') # User 1 (subscriber) should receive a notification self.assertTrue(NotificationRecipient.objects.filter(user=self.user1).exists()) notif = NotificationRecipient.objects.filter(user=self.user1).first().notification from interventions.models import STATUS_CHOICES target_label = dict(STATUS_CHOICES).get('to_be_planned') self.assertIn(str(target_label), notif.content) # User 2 (modifier) should NOT receive a notification self.assertFalse(NotificationRecipient.objects.filter(user=self.user2).exists()) def test_list_followed_only_filtering(self): from interventions.models import InterventionSubscription, Intervention from interventions.views import apply_intervention_filters_from_querydict from django.http import QueryDict # User 1 subscribes to self.itv InterventionSubscription.objects.create(user=self.user1, intervention=self.itv) # Create another intervention that User 1 is NOT subscribed to itv2 = Intervention.objects.create( title='Another Intervention', status='in_preparation', thematic=self.thematic, created_by=self.user1 ) qs = Intervention.objects.all() # Filter with followed_only=1 params = QueryDict('followed_only=1') filtered_qs = apply_intervention_filters_from_querydict(qs, params, user=self.user1) self.assertIn(self.itv, filtered_qs) self.assertNotIn(itv2, filtered_qs)