244 lines
10 KiB
Python
244 lines
10 KiB
Python
import json
|
|
from datetime import timedelta
|
|
from unittest import mock
|
|
|
|
from django.test import TestCase
|
|
from django.contrib.auth import get_user_model
|
|
from django.contrib import messages
|
|
from django.contrib.messages import get_messages
|
|
from django.core.files.uploadedfile import SimpleUploadedFile
|
|
|
|
from common.models import UserConfig, Role, Thematic, UserThematics
|
|
from contracts.models import Company, CompanyTeam, CompanyMember, Contract, ContractPost, ContractOrder, ContractOrderQuote, ContractOrderQuoteItem
|
|
from assets.models import AssetCategory
|
|
from interventions.models import (
|
|
Intervention,
|
|
InterventionContractPost,
|
|
InterventionPlanificationTimeLine,
|
|
InterventionRealizationTimeLine,
|
|
Operation,
|
|
InterventionAsset,
|
|
OperationTemplate,
|
|
OperationTemplateItem,
|
|
OperationStatusSet,
|
|
OperationStatus,
|
|
STATUS_ORDERS,
|
|
InterventionDocument,
|
|
)
|
|
from django.contrib.contenttypes.models import ContentType
|
|
from interventions.views import _group_operations_by_asset_and_template, _create_equipment_operations_for_assets
|
|
from interventions.permissions import can_manage_occupations, can_delete_document
|
|
from django.urls import reverse
|
|
from django.utils import timezone
|
|
from decimal import Decimal
|
|
|
|
|
|
|
|
class CalendarEventsTests(TestCase):
|
|
def setUp(self):
|
|
User = get_user_model()
|
|
self.user = User.objects.create_user(username='calendar-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)
|
|
|
|
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)
|
|
|
|
# Member belonging to both Team A and Team B
|
|
self.member = CompanyMember.objects.create(user=self.user, name='Multi Team Member')
|
|
self.member.teams.add(self.team_a, self.team_b)
|
|
|
|
self.client.login(username='calendar-user', password='pwd')
|
|
|
|
def test_calendar_events_view_mode_members_no_duplicates(self):
|
|
# Request calendar events in member view mode
|
|
url = reverse('interventions:interventions_events') + '?view_mode=members'
|
|
response = self.client.get(url)
|
|
self.assertEqual(response.status_code, 200)
|
|
data = response.json()
|
|
|
|
# data has 'teams' key
|
|
teams = data.get('teams', [])
|
|
|
|
# Verify that for each team, the member is present only once
|
|
for team in teams:
|
|
member_ids = [m['id'] for m in team.get('members', [])]
|
|
if team['id'] in [self.team_a.id, self.team_b.id]:
|
|
# Ensure the member is not duplicated in the same team
|
|
self.assertEqual(member_ids.count(self.member.id), 1)
|
|
|
|
def test_calendar_events_task_replication_fix(self):
|
|
# Create an intervention assigned to the member and specifically to Team A
|
|
itv = Intervention.objects.create(
|
|
title='Team A Task',
|
|
status='to_be_processed',
|
|
assigned_member=self.member,
|
|
assigned_team=self.team_a,
|
|
planned_begin_time=timezone.now() + timedelta(hours=1),
|
|
planned_end_time=timezone.now() + timedelta(hours=3),
|
|
)
|
|
|
|
url = reverse('interventions:interventions_events') + '?view_mode=members'
|
|
response = self.client.get(url)
|
|
self.assertEqual(response.status_code, 200)
|
|
data = response.json()
|
|
|
|
teams = data.get('teams', [])
|
|
|
|
# Find Team A and Team B in the response
|
|
team_a_data = next((t for t in teams if t['id'] == self.team_a.id), None)
|
|
team_b_data = next((t for t in teams if t['id'] == self.team_b.id), None)
|
|
|
|
self.assertIsNotNone(team_a_data)
|
|
self.assertIsNotNone(team_b_data)
|
|
|
|
# Member should have the intervention under Team A
|
|
member_in_team_a = next((m for m in team_a_data['members'] if m['id'] == self.member.id), None)
|
|
self.assertIsNotNone(member_in_team_a)
|
|
itv_ids_in_team_a = [i['id'] for i in member_in_team_a['interventions']]
|
|
self.assertIn(itv.id, itv_ids_in_team_a)
|
|
|
|
# Member should NOT have the intervention under Team B
|
|
member_in_team_b = next((m for m in team_b_data['members'] if m['id'] == self.member.id), None)
|
|
self.assertIsNotNone(member_in_team_b)
|
|
itv_ids_in_team_b = [i['id'] for i in member_in_team_b['interventions']]
|
|
self.assertNotIn(itv.id, itv_ids_in_team_b)
|
|
|
|
def test_schedule_update_coerces_ids(self):
|
|
itv = Intervention.objects.create(
|
|
title='Test Coercion',
|
|
status='to_be_planned',
|
|
)
|
|
url = reverse('interventions:interventions_schedule_update')
|
|
# Send payload with string IDs
|
|
payload = {
|
|
'intervention_id': itv.id,
|
|
'planned_begin_time': (timezone.now() + timedelta(hours=1)).isoformat(),
|
|
'planned_end_time': (timezone.now() + timedelta(hours=3)).isoformat(),
|
|
'assigned_member_id': str(self.member.id),
|
|
'assigned_team_id': str(self.team_b.id),
|
|
}
|
|
response = self.client.post(
|
|
url,
|
|
data=json.dumps(payload),
|
|
content_type='application/json'
|
|
)
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertTrue(response.json().get('success'))
|
|
|
|
itv.refresh_from_db()
|
|
self.assertEqual(itv.assigned_member_id, self.member.id)
|
|
# It should be Team B, not Team A (which would be the fallback if the string ID was not matched)
|
|
self.assertEqual(itv.assigned_team_id, self.team_b.id)
|
|
|
|
def test_schedule_update_rejects_past_dates(self):
|
|
itv = Intervention.objects.create(
|
|
title='Test Past Date update',
|
|
status='to_be_planned',
|
|
)
|
|
url = reverse('interventions:interventions_schedule_update')
|
|
payload = {
|
|
'intervention_id': itv.id,
|
|
'planned_begin_time': (timezone.now() - timedelta(days=1)).isoformat(),
|
|
'planned_end_time': (timezone.now() + timedelta(hours=1)).isoformat(),
|
|
'assigned_member_id': str(self.member.id),
|
|
'assigned_team_id': str(self.team_a.id),
|
|
}
|
|
response = self.client.post(
|
|
url,
|
|
data=json.dumps(payload),
|
|
content_type='application/json',
|
|
HTTP_ACCEPT_LANGUAGE='fr'
|
|
)
|
|
self.assertEqual(response.status_code, 400)
|
|
self.assertIn("passé", response.json().get('error', '').lower())
|
|
|
|
def test_backlog_includes_past_unfinished_interventions(self):
|
|
past_unfinished_itv = Intervention.objects.create(
|
|
title='Past Unfinished corrective',
|
|
status='to_be_processed',
|
|
maintain_type='corrective',
|
|
planned_begin_time=timezone.now() - timedelta(days=2),
|
|
planned_end_time=timezone.now() - timedelta(days=2) + timedelta(hours=2),
|
|
assigned_team=self.team_a
|
|
)
|
|
|
|
url = reverse('interventions:interventions_events') + '?view_mode=teams_daily'
|
|
response = self.client.get(url)
|
|
self.assertEqual(response.status_code, 200)
|
|
data = response.json()
|
|
|
|
backlog_corrective = data.get('backlog_groups', {}).get('corrective', [])
|
|
backlog_ids = [item['id'] for item in backlog_corrective]
|
|
self.assertIn(past_unfinished_itv.id, backlog_ids)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
class CalendarSearchTests(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)
|
|
|
|
self.thematic = Thematic.objects.create(code='roads', name_fr='Routes', name_nl='Wegen')
|
|
UserThematics.objects.create(
|
|
user_config=self.user_config,
|
|
thematic=self.thematic,
|
|
can_view_interventions=True,
|
|
can_edit_interventions=True,
|
|
)
|
|
|
|
self.intv_matching = Intervention.objects.create(
|
|
title='Matching Search Keyword',
|
|
status='to_be_processed',
|
|
thematic=self.thematic,
|
|
expected_begin_time=timezone.now(),
|
|
expected_end_time=timezone.now() + timedelta(hours=2)
|
|
)
|
|
self.intv_other = Intervention.objects.create(
|
|
title='Completely Unrelated Title',
|
|
status='to_be_processed',
|
|
thematic=self.thematic,
|
|
expected_begin_time=timezone.now(),
|
|
expected_end_time=timezone.now() + timedelta(hours=2)
|
|
)
|
|
|
|
self.client.login(username='admin', password='pwd')
|
|
|
|
def test_interventions_calendar_selected_search(self):
|
|
url = reverse('interventions:interventions_calendar') + '?q=Keyword'
|
|
response = self.client.get(url)
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response.context['selected_search'], 'Keyword')
|
|
|
|
def test_interventions_events_filtering(self):
|
|
url = reverse('interventions:interventions_events') + '?q=Keyword'
|
|
response = self.client.get(url)
|
|
self.assertEqual(response.status_code, 200)
|
|
events = response.json()
|
|
|
|
# Should only contain the matching intervention
|
|
event_ids = [e['id'] for e in events]
|
|
self.assertIn(self.intv_matching.id, event_ids)
|
|
self.assertNotIn(self.intv_other.id, event_ids)
|
|
|
|
def test_interventions_geojson_filtering(self):
|
|
url = reverse('interventions:interventions_geojson') + '?q=Keyword'
|
|
response = self.client.get(url)
|
|
self.assertEqual(response.status_code, 200)
|
|
geojson = response.json()
|
|
|
|
# GeoJSON is a FeatureCollection, check properties inside features
|
|
features = geojson.get('features', [])
|
|
feature_ids = [f['properties']['id'] for f in features]
|
|
self.assertIn(self.intv_matching.id, feature_ids)
|
|
self.assertNotIn(self.intv_other.id, feature_ids)
|
|
|
|
|