843 lines
35 KiB
Python
843 lines
35 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, UserContractAccess, UserContractStatusPermission
|
|
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)
|
|
|
|
def test_unassign_intervention_moves_it_to_backlog(self):
|
|
from interventions.models import InterventionAssignment
|
|
now = timezone.now()
|
|
itv = Intervention.objects.create(
|
|
title='Intervention Today Team A',
|
|
status='to_be_processed',
|
|
maintain_type='corrective',
|
|
assigned_team=self.team_a,
|
|
planned_begin_time=now,
|
|
planned_end_time=now + timedelta(hours=2),
|
|
)
|
|
self.assertEqual(InterventionAssignment.objects.filter(intervention=itv, date=now.date()).count(), 1)
|
|
|
|
update_url = reverse('interventions:interventions_schedule_update')
|
|
payload = {
|
|
'intervention_id': itv.id,
|
|
'planned_begin_time': now.isoformat(),
|
|
'planned_end_time': (now + timedelta(hours=2)).isoformat(),
|
|
'assigned_member_id': None,
|
|
'assigned_team_id': None,
|
|
}
|
|
res = self.client.post(update_url, data=json.dumps(payload), content_type='application/json')
|
|
self.assertEqual(res.status_code, 200)
|
|
|
|
itv.refresh_from_db()
|
|
self.assertIsNone(itv.assigned_team)
|
|
self.assertIsNone(itv.assigned_member)
|
|
self.assertEqual(itv.status, 'to_be_planned')
|
|
self.assertEqual(InterventionAssignment.objects.filter(intervention=itv, date=now.date()).count(), 0)
|
|
|
|
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', [])
|
|
self.assertIn(itv.id, [item['id'] for item in backlog_corrective])
|
|
|
|
for team in data.get('teams', []):
|
|
for day_key, day_data in team.get('days', {}).items():
|
|
itv_ids_in_day = [item['id'] for item in day_data.get('interventions', [])]
|
|
self.assertNotIn(itv.id, itv_ids_in_day)
|
|
|
|
def test_in_preparation_interventions_only_in_backlog(self):
|
|
now = timezone.now()
|
|
itv = Intervention.objects.create(
|
|
title='In Preparation With Schedule',
|
|
status='in_preparation',
|
|
maintain_type='corrective',
|
|
planned_begin_time=now,
|
|
planned_end_time=now + timedelta(hours=2),
|
|
)
|
|
|
|
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', [])
|
|
self.assertIn(itv.id, [item['id'] for item in backlog_corrective])
|
|
|
|
for team in data.get('teams', []):
|
|
for day_key, day_data in team.get('days', {}).items():
|
|
itv_ids_in_day = [item['id'] for item in day_data.get('interventions', [])]
|
|
self.assertNotIn(itv.id, itv_ids_in_day)
|
|
|
|
def test_backlog_sorting_order(self):
|
|
itv_low_old = Intervention.objects.create(
|
|
title='Low Priority Old',
|
|
status='to_be_planned',
|
|
maintain_type='corrective',
|
|
priority='4',
|
|
)
|
|
itv_urgent_old = Intervention.objects.create(
|
|
title='Urgent Priority Old',
|
|
status='to_be_planned',
|
|
maintain_type='corrective',
|
|
priority='1',
|
|
)
|
|
itv_urgent_new = Intervention.objects.create(
|
|
title='Urgent Priority New',
|
|
status='to_be_planned',
|
|
maintain_type='corrective',
|
|
priority='1',
|
|
)
|
|
|
|
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', [])
|
|
ids = [item['id'] for item in backlog_corrective if item['id'] in {itv_low_old.id, itv_urgent_old.id, itv_urgent_new.id}]
|
|
|
|
self.assertEqual(ids, [itv_urgent_new.id, itv_urgent_old.id, itv_low_old.id])
|
|
|
|
def test_member_history_ajax_permissions_and_data(self):
|
|
from contracts.models import CompanyMember, CompanyMemberHistory
|
|
member = CompanyMember.objects.create(name="Jean Dupont", user=self.user)
|
|
member.teams.add(self.team_a)
|
|
CompanyMemberHistory.objects.create(
|
|
member=member,
|
|
team=self.team_a,
|
|
join_date=timezone.now().date() - timedelta(days=10)
|
|
)
|
|
url = reverse('interventions:member_history_ajax', args=[member.id])
|
|
response = self.client.get(url)
|
|
self.assertEqual(response.status_code, 200)
|
|
data = response.json()
|
|
|
|
self.assertIn('member', data)
|
|
self.assertEqual(data['member']['name'], "Jean Dupont")
|
|
self.assertIn('team_history', data)
|
|
self.assertTrue(len(data['team_history']) >= 1)
|
|
self.assertEqual(data['team_history'][0]['team_name'], self.team_a.name)
|
|
|
|
def test_unassign_sans_equipe_reverts_status_to_backlog(self):
|
|
itv = Intervention.objects.create(
|
|
title='Unassigned Test',
|
|
status='to_be_processed',
|
|
maintain_type='corrective',
|
|
priority='2',
|
|
planned_begin_time=timezone.now(),
|
|
planned_end_time=timezone.now() + timedelta(hours=2),
|
|
)
|
|
url = reverse('interventions:interventions_schedule_update')
|
|
payload = {
|
|
'intervention_id': itv.id,
|
|
'member_id': None,
|
|
'team_id': None,
|
|
'unassign': True,
|
|
}
|
|
response = self.client.post(url, json.dumps(payload), content_type='application/json')
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
itv.refresh_from_db()
|
|
self.assertEqual(itv.status, 'to_be_planned')
|
|
self.assertIsNone(itv.planned_begin_time)
|
|
self.assertIsNone(itv.planned_end_time)
|
|
self.assertIsNone(itv.assigned_team)
|
|
self.assertIsNone(itv.assigned_member)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
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)
|
|
|
|
def test_calendar_events_external_manager_backlog_visibility(self):
|
|
User = get_user_model()
|
|
ext_user = User.objects.create_user(username='ext-mgr', password='pwd')
|
|
role_ext, _ = Role.objects.get_or_create(name='external_manager')
|
|
ext_config = UserConfig.objects.create(user=ext_user, is_intern=False)
|
|
ext_config.roles.add(role_ext)
|
|
|
|
from common.models import UserContractAccess, UserContractStatusPermission
|
|
from contracts.models import UserCompanyTeamAccess
|
|
company = Company.objects.create(name='Company Ext')
|
|
contract = Contract.objects.create(
|
|
company=company,
|
|
contract_number="C-TEST-01",
|
|
start_date=timezone.now().date() - timedelta(days=30),
|
|
end_date=timezone.now().date() + timedelta(days=365)
|
|
)
|
|
contract_access = UserContractAccess.objects.create(
|
|
user_config=ext_config,
|
|
contract=contract,
|
|
can_view_interventions=True
|
|
)
|
|
UserContractStatusPermission.objects.create(
|
|
user_contract=contract_access,
|
|
status='to_be_processed',
|
|
can_view=True,
|
|
can_change_status_to=False,
|
|
)
|
|
UserThematics.objects.create(
|
|
user_config=ext_config,
|
|
thematic=self.thematic,
|
|
can_view_interventions=True
|
|
)
|
|
team = CompanyTeam.objects.create(name='Team Ext', company=company)
|
|
team.thematics.add(self.thematic)
|
|
UserCompanyTeamAccess.objects.create(
|
|
user_config=ext_config,
|
|
team=team,
|
|
can_view=True
|
|
)
|
|
|
|
itv_backlog = Intervention.objects.create(
|
|
title='Intervention for External Manager',
|
|
status='to_be_processed',
|
|
maintain_type='corrective',
|
|
contract=contract,
|
|
thematic=self.thematic,
|
|
)
|
|
|
|
self.client.login(username='ext-mgr', password='pwd')
|
|
url = reverse('interventions:interventions_events') + '?view_mode=teams_daily'
|
|
response = self.client.get(url)
|
|
self.assertEqual(response.status_code, 200)
|
|
data = response.json()
|
|
self.assertIn('backlog_groups', data)
|
|
corrective_backlog = data['backlog_groups'].get('corrective', [])
|
|
corrective_ids = [i['id'] for i in corrective_backlog]
|
|
self.assertIn(itv_backlog.id, corrective_ids)
|
|
|
|
|
|
class CalendarContractFilterTests(TestCase):
|
|
def setUp(self):
|
|
User = get_user_model()
|
|
self.thematic = Thematic.objects.create(code='lighting', name_fr='Éclairage public', name_nl='Openbare verlichting')
|
|
self.company = Company.objects.create(name='Provider Corp')
|
|
|
|
# Deux contrats sur la même thématique
|
|
today = timezone.now().date()
|
|
self.contract_allowed = Contract.objects.create(
|
|
contract_number='CTR-ALLOWED',
|
|
company=self.company,
|
|
start_date=today - timedelta(days=10),
|
|
end_date=today + timedelta(days=365),
|
|
is_active=True,
|
|
)
|
|
self.contract_allowed.thematics.add(self.thematic)
|
|
|
|
self.contract_forbidden = Contract.objects.create(
|
|
contract_number='CTR-FORBIDDEN',
|
|
company=self.company,
|
|
start_date=today - timedelta(days=10),
|
|
end_date=today + timedelta(days=365),
|
|
is_active=True,
|
|
)
|
|
self.contract_forbidden.thematics.add(self.thematic)
|
|
|
|
# Utilisateur externe
|
|
self.external_user = User.objects.create_user(username='external-user', password='pwd')
|
|
self.role_external_manager = Role.objects.create(name='external_manager')
|
|
self.external_config = UserConfig.objects.create(
|
|
user=self.external_user,
|
|
is_intern=False,
|
|
default_thematic=self.thematic,
|
|
)
|
|
self.external_config.roles.add(self.role_external_manager)
|
|
UserThematics.objects.create(
|
|
user_config=self.external_config,
|
|
thematic=self.thematic,
|
|
can_view_interventions=True,
|
|
can_edit_interventions=True,
|
|
)
|
|
# Accès uniquement à contract_allowed
|
|
UserContractAccess.objects.create(
|
|
user_config=self.external_config,
|
|
contract=self.contract_allowed,
|
|
can_view_interventions=True,
|
|
)
|
|
|
|
# Utilisateur admin
|
|
self.admin_user = User.objects.create_user(username='admin-user', password='pwd')
|
|
self.role_admin = Role.objects.get_or_create(name='admin')[0]
|
|
self.admin_config = UserConfig.objects.create(
|
|
user=self.admin_user,
|
|
is_intern=True,
|
|
default_thematic=self.thematic,
|
|
)
|
|
self.admin_config.roles.add(self.role_admin)
|
|
UserThematics.objects.create(
|
|
user_config=self.admin_config,
|
|
thematic=self.thematic,
|
|
can_view_interventions=True,
|
|
can_edit_interventions=True,
|
|
)
|
|
|
|
def test_external_user_only_sees_accessible_contracts_in_calendar(self):
|
|
self.client.login(username='external-user', password='pwd')
|
|
url = reverse('interventions:interventions_calendar')
|
|
response = self.client.get(url, follow=True)
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
contract_choices = response.context['contract_choices']
|
|
contract_ids = [c[0] for c in contract_choices]
|
|
|
|
# L'externe ne doit voir que CTR-ALLOWED, pas CTR-FORBIDDEN
|
|
self.assertIn(self.contract_allowed.id, contract_ids)
|
|
self.assertNotIn(self.contract_forbidden.id, contract_ids)
|
|
|
|
def test_admin_user_sees_all_contracts_for_thematic_in_calendar(self):
|
|
self.client.login(username='admin-user', password='pwd')
|
|
url = reverse('interventions:interventions_calendar')
|
|
response = self.client.get(url, follow=True)
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
contract_choices = response.context['contract_choices']
|
|
contract_ids = [c[0] for c in contract_choices]
|
|
|
|
# L'admin voit tous les contrats de la thématique
|
|
self.assertIn(self.contract_allowed.id, contract_ids)
|
|
self.assertIn(self.contract_forbidden.id, contract_ids)
|
|
|
|
def test_internal_user_with_limit_interventions_to_contracts_only_sees_accessible(self):
|
|
User = get_user_model()
|
|
intern_user = User.objects.create_user(username='intern-limited', password='pwd')
|
|
intern_config = UserConfig.objects.create(
|
|
user=intern_user,
|
|
is_intern=True,
|
|
limit_interventions_to_contracts=True,
|
|
default_thematic=self.thematic,
|
|
)
|
|
UserThematics.objects.create(
|
|
user_config=intern_config,
|
|
thematic=self.thematic,
|
|
can_view_interventions=True,
|
|
can_edit_interventions=True,
|
|
)
|
|
UserContractAccess.objects.create(
|
|
user_config=intern_config,
|
|
contract=self.contract_allowed,
|
|
can_view_interventions=True,
|
|
)
|
|
|
|
self.client.login(username='intern-limited', password='pwd')
|
|
url = reverse('interventions:interventions_calendar')
|
|
response = self.client.get(url, follow=True)
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
contract_choices = response.context['contract_choices']
|
|
contract_ids = [c[0] for c in contract_choices]
|
|
|
|
self.assertIn(self.contract_allowed.id, contract_ids)
|
|
self.assertNotIn(self.contract_forbidden.id, contract_ids)
|
|
|
|
def test_internal_user_without_limit_interventions_to_contracts_sees_all(self):
|
|
User = get_user_model()
|
|
intern_user = User.objects.create_user(username='intern-unlimited', password='pwd')
|
|
intern_config = UserConfig.objects.create(
|
|
user=intern_user,
|
|
is_intern=True,
|
|
limit_interventions_to_contracts=False,
|
|
default_thematic=self.thematic,
|
|
)
|
|
UserThematics.objects.create(
|
|
user_config=intern_config,
|
|
thematic=self.thematic,
|
|
can_view_interventions=True,
|
|
can_edit_interventions=True,
|
|
)
|
|
|
|
self.client.login(username='intern-unlimited', password='pwd')
|
|
url = reverse('interventions:interventions_calendar')
|
|
response = self.client.get(url, follow=True)
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
contract_choices = response.context['contract_choices']
|
|
contract_ids = [c[0] for c in contract_choices]
|
|
|
|
self.assertIn(self.contract_allowed.id, contract_ids)
|
|
self.assertIn(self.contract_forbidden.id, contract_ids)
|
|
|
|
|
|
class ExternalManagerScheduleUpdateTests(TestCase):
|
|
def setUp(self):
|
|
User = get_user_model()
|
|
self.company_ext = Company.objects.create(name='Prestataire Ext A')
|
|
self.company_other = Company.objects.create(name='Prestataire Ext B')
|
|
|
|
self.ext_user = User.objects.create_user(username='a2manager', password='pwd')
|
|
self.role_ext = Role.objects.create(name='external_manager')
|
|
self.ext_config = UserConfig.objects.create(
|
|
user=self.ext_user,
|
|
is_intern=False,
|
|
company=self.company_ext,
|
|
)
|
|
self.ext_config.roles.add(self.role_ext)
|
|
|
|
today = timezone.now().date()
|
|
self.thematic = Thematic.objects.create(code='roads_ext_mgr', name_fr='Routes Ext', name_nl='Wegen Ext')
|
|
self.contract = Contract.objects.create(
|
|
contract_number='CTR-EXT-A',
|
|
company=self.company_ext,
|
|
start_date=today - timedelta(days=10),
|
|
end_date=today + timedelta(days=365),
|
|
is_active=True,
|
|
)
|
|
self.contract.thematics.add(self.thematic)
|
|
|
|
UserThematics.objects.create(
|
|
user_config=self.ext_config,
|
|
thematic=self.thematic,
|
|
can_view_interventions=True,
|
|
can_edit_interventions=False,
|
|
)
|
|
|
|
self.uca = UserContractAccess.objects.create(
|
|
user_config=self.ext_config,
|
|
contract=self.contract,
|
|
can_view_interventions=True,
|
|
)
|
|
|
|
# Status permissions: can_view=True, can_edit=False, can_change_status_to=False
|
|
UserContractStatusPermission.objects.create(
|
|
user_contract=self.uca,
|
|
status='to_be_planned',
|
|
can_view=True,
|
|
can_edit=False,
|
|
can_change_status_to=True,
|
|
)
|
|
UserContractStatusPermission.objects.create(
|
|
user_contract=self.uca,
|
|
status='to_be_processed',
|
|
can_view=True,
|
|
can_edit=False,
|
|
can_change_status_to=False,
|
|
)
|
|
UserContractStatusPermission.objects.create(
|
|
user_contract=self.uca,
|
|
status='assigned',
|
|
can_view=True,
|
|
can_edit=False,
|
|
can_change_status_to=False,
|
|
)
|
|
|
|
self.team_ext_1 = CompanyTeam.objects.create(name='Equipe 1', company=self.company_ext)
|
|
self.team_ext_2 = CompanyTeam.objects.create(name='Equipe 2', company=self.company_ext)
|
|
self.member_ext_1 = CompanyMember.objects.create(name='Membre 1', user=self.ext_user)
|
|
self.member_ext_1.teams.add(self.team_ext_1)
|
|
self.member_ext_2 = CompanyMember.objects.create(name='Membre 2', user=self.ext_user)
|
|
self.member_ext_2.teams.add(self.team_ext_2)
|
|
|
|
self.team_other = CompanyTeam.objects.create(name='Equipe B', company=self.company_other)
|
|
self.member_other = CompanyMember.objects.create(name='Membre B', user=self.ext_user)
|
|
self.member_other.teams.add(self.team_other)
|
|
|
|
now = timezone.now()
|
|
self.itv = Intervention.objects.create(
|
|
title='Intervention Prestataire A',
|
|
thematic=self.thematic,
|
|
contract=self.contract,
|
|
status='to_be_processed',
|
|
maintain_type='corrective',
|
|
assigned_team=self.team_ext_1,
|
|
assigned_member=self.member_ext_1,
|
|
planned_begin_time=now + timedelta(hours=1),
|
|
planned_end_time=now + timedelta(hours=3),
|
|
)
|
|
|
|
self.update_url = reverse('interventions:interventions_schedule_update')
|
|
self.client.login(username='a2manager', password='pwd')
|
|
|
|
def test_external_manager_can_reschedule_to_be_processed_intervention(self):
|
|
new_begin = timezone.now() + timedelta(days=2)
|
|
new_end = new_begin + timedelta(hours=2)
|
|
|
|
payload = {
|
|
'intervention_id': self.itv.id,
|
|
'planned_begin_time': new_begin.isoformat(),
|
|
'planned_end_time': new_end.isoformat(),
|
|
}
|
|
res = self.client.post(self.update_url, data=json.dumps(payload), content_type='application/json')
|
|
self.assertEqual(res.status_code, 200)
|
|
self.assertTrue(res.json().get('success'))
|
|
|
|
self.itv.refresh_from_db()
|
|
self.assertEqual(self.itv.planned_begin_time.date(), new_begin.date())
|
|
self.assertEqual(self.itv.status, 'to_be_processed')
|
|
|
|
def test_external_manager_can_reassign_member_and_team_on_to_be_processed_intervention(self):
|
|
new_begin = timezone.now() + timedelta(days=1)
|
|
new_end = new_begin + timedelta(hours=2)
|
|
|
|
payload = {
|
|
'intervention_id': self.itv.id,
|
|
'planned_begin_time': new_begin.isoformat(),
|
|
'planned_end_time': new_end.isoformat(),
|
|
'assigned_member_id': self.member_ext_2.id,
|
|
'assigned_team_id': self.team_ext_2.id,
|
|
}
|
|
res = self.client.post(self.update_url, data=json.dumps(payload), content_type='application/json')
|
|
self.assertEqual(res.status_code, 200)
|
|
self.assertTrue(res.json().get('success'))
|
|
|
|
self.itv.refresh_from_db()
|
|
self.assertEqual(self.itv.assigned_member_id, self.member_ext_2.id)
|
|
self.assertEqual(self.itv.assigned_team_id, self.team_ext_2.id)
|
|
self.assertEqual(self.itv.status, 'to_be_processed')
|
|
|
|
def test_external_manager_can_unassign_to_be_processed_intervention_to_backlog(self):
|
|
payload = {
|
|
'intervention_id': self.itv.id,
|
|
'assigned_member_id': None,
|
|
'assigned_team_id': None,
|
|
'planned_begin_time': None,
|
|
'planned_end_time': None,
|
|
'unassign': True,
|
|
}
|
|
res = self.client.post(self.update_url, data=json.dumps(payload), content_type='application/json')
|
|
self.assertEqual(res.status_code, 200)
|
|
self.assertTrue(res.json().get('success'))
|
|
|
|
self.itv.refresh_from_db()
|
|
self.assertEqual(self.itv.status, 'to_be_planned')
|
|
self.assertIsNone(self.itv.assigned_member)
|
|
self.assertIsNone(self.itv.assigned_team)
|
|
self.assertIsNone(self.itv.planned_begin_time)
|
|
self.assertIsNone(self.itv.planned_end_time)
|
|
|
|
def test_external_manager_unassign_without_to_be_planned_permission_keeps_current_status(self):
|
|
# Disable can_change_status_to on to_be_planned
|
|
UserContractStatusPermission.objects.filter(
|
|
user_contract=self.uca,
|
|
status='to_be_planned',
|
|
).update(can_change_status_to=False)
|
|
|
|
payload = {
|
|
'intervention_id': self.itv.id,
|
|
'assigned_member_id': None,
|
|
'assigned_team_id': None,
|
|
'unassign': True,
|
|
}
|
|
res = self.client.post(self.update_url, data=json.dumps(payload), content_type='application/json')
|
|
self.assertEqual(res.status_code, 200)
|
|
self.assertTrue(res.json().get('success'))
|
|
|
|
self.itv.refresh_from_db()
|
|
# Status remains to_be_processed since user cannot transition to to_be_planned
|
|
self.assertEqual(self.itv.status, 'to_be_processed')
|
|
self.assertIsNone(self.itv.assigned_member)
|
|
self.assertIsNone(self.itv.assigned_team)
|
|
self.assertIsNone(self.itv.planned_begin_time)
|
|
self.assertIsNone(self.itv.planned_end_time)
|
|
|
|
def test_external_manager_cannot_assign_member_of_different_company(self):
|
|
new_begin = timezone.now() + timedelta(days=1)
|
|
new_end = new_begin + timedelta(hours=2)
|
|
|
|
payload = {
|
|
'intervention_id': self.itv.id,
|
|
'planned_begin_time': new_begin.isoformat(),
|
|
'planned_end_time': new_end.isoformat(),
|
|
'assigned_member_id': self.member_other.id,
|
|
'assigned_team_id': self.team_other.id,
|
|
}
|
|
res = self.client.post(self.update_url, data=json.dumps(payload), content_type='application/json')
|
|
self.assertEqual(res.status_code, 403)
|
|
self.assertIn("entreprise", res.json().get('error', '').lower())
|
|
|
|
def test_external_manager_can_unassign_assigned_status_intervention_to_backlog(self):
|
|
itv_assigned = Intervention.objects.create(
|
|
title='Intervention Status Assigned',
|
|
thematic=self.thematic,
|
|
contract=self.contract,
|
|
status='assigned',
|
|
maintain_type='corrective',
|
|
assigned_team=self.team_ext_1,
|
|
assigned_member=self.member_ext_1,
|
|
planned_begin_time=timezone.now() + timedelta(hours=1),
|
|
planned_end_time=timezone.now() + timedelta(hours=3),
|
|
)
|
|
|
|
payload = {
|
|
'intervention_id': itv_assigned.id,
|
|
'assigned_member_id': None,
|
|
'assigned_team_id': None,
|
|
'unassign': True,
|
|
}
|
|
res = self.client.post(self.update_url, data=json.dumps(payload), content_type='application/json')
|
|
self.assertEqual(res.status_code, 200)
|
|
|
|
itv_assigned.refresh_from_db()
|
|
# Since 'assigned' -> 'to_be_planned' is not an allowed transition in ALLOWED_TRANSITIONS,
|
|
# the status remains 'assigned' while unassigning member/team/dates
|
|
self.assertEqual(itv_assigned.status, 'assigned')
|
|
self.assertIsNone(itv_assigned.assigned_member)
|
|
self.assertIsNone(itv_assigned.assigned_team)
|
|
|
|
def test_internal_manager_cannot_assign_members_in_calendar(self):
|
|
User = get_user_model()
|
|
mgr_user = User.objects.create_user(username='internal-mgr', password='pwd')
|
|
role_mgr = Role.objects.create(name='manager')
|
|
mgr_config = UserConfig.objects.create(user=mgr_user, is_intern=True)
|
|
mgr_config.roles.add(role_mgr)
|
|
|
|
uca_mgr = UserContractAccess.objects.create(
|
|
user_config=mgr_config,
|
|
contract=self.contract,
|
|
can_view_interventions=True,
|
|
)
|
|
UserContractStatusPermission.objects.create(
|
|
user_contract=uca_mgr,
|
|
status='to_be_processed',
|
|
can_view=True,
|
|
can_edit=True,
|
|
)
|
|
|
|
self.client.login(username='internal-mgr', password='pwd')
|
|
payload = {
|
|
'intervention_id': self.itv.id,
|
|
'planned_begin_time': (timezone.now() + timedelta(days=1)).isoformat(),
|
|
'planned_end_time': (timezone.now() + timedelta(days=1, hours=2)).isoformat(),
|
|
'assigned_member_id': self.member_ext_2.id,
|
|
}
|
|
res = self.client.post(self.update_url, data=json.dumps(payload), content_type='application/json', HTTP_ACCEPT_LANGUAGE='fr')
|
|
self.assertEqual(res.status_code, 403)
|
|
self.assertIn("assigner", res.json().get('error', '').lower())
|
|
|
|
|
|
|
|
|
|
|