1781 lines
73 KiB
Python
1781 lines
73 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,
|
|
InterventionTimeLine,
|
|
InterventionLink,
|
|
InterventionNote,
|
|
Symptom,
|
|
Operation,
|
|
InterventionAsset,
|
|
OperationTemplate,
|
|
OperationTemplateItem,
|
|
OperationStatusSet,
|
|
OperationStatus,
|
|
STATUS_ORDERS,
|
|
InterventionDocument,
|
|
get_type_choices,
|
|
)
|
|
from django.db.models import Q
|
|
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_homepage_on_pause_interventions_filtering(self):
|
|
# Create an intervention on_pause, with contract using manager, but manager is None (not this user)
|
|
itv_other_mgr = Intervention.objects.create(
|
|
title='Other Manager on pause',
|
|
status='on_pause',
|
|
contract=self.contract,
|
|
thematic=self.thematic,
|
|
intervention_manager=None,
|
|
symptom=self.symptom,
|
|
)
|
|
# Create an intervention on_pause, with contract using manager, managed by self.user
|
|
itv_self_mgr = Intervention.objects.create(
|
|
title='Self Manager on pause',
|
|
status='on_pause',
|
|
contract=self.contract,
|
|
thematic=self.thematic,
|
|
intervention_manager=self.user,
|
|
symptom=self.symptom,
|
|
)
|
|
# Create an intervention on_pause, but contract does not use manager
|
|
contract_no_mgr = Contract.objects.create(
|
|
company=self.company,
|
|
contract_number='C-NOMGR-PAUSE',
|
|
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 on pause',
|
|
status='on_pause',
|
|
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)
|
|
|
|
# On pause list/counter should only contain self-managed and no-manager interventions
|
|
on_pause_list = response.context['intervention_on_pause_list']
|
|
self.assertIn(itv_self_mgr, on_pause_list)
|
|
self.assertIn(itv_no_mgr, on_pause_list)
|
|
self.assertNotIn(itv_other_mgr, on_pause_list)
|
|
|
|
self.assertEqual(response.context['intervention_on_pause_count'], 2)
|
|
self.assertEqual(response.context['total_on_pause_count'], 3)
|
|
self.assertTrue(response.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_pause_notification_sent_to_manager_when_paused_by_other(self):
|
|
from notifications.models import NotificationRecipient
|
|
User = get_user_model()
|
|
other_user = User.objects.create_user(username='technician_user', password='pwd')
|
|
other_config = UserConfig.objects.create(user=other_user, is_intern=True)
|
|
other_config.roles.add(self.role_admin)
|
|
UserThematics.objects.create(
|
|
user_config=other_config,
|
|
thematic=self.thematic,
|
|
can_view_interventions=True,
|
|
can_edit_interventions=True,
|
|
)
|
|
intervention = Intervention.objects.create(
|
|
title='Pause Notification Test',
|
|
status='in_progress',
|
|
contract=self.contract,
|
|
thematic=self.thematic,
|
|
symptom=self.symptom,
|
|
intervention_manager=self.user,
|
|
)
|
|
|
|
from interventions.views import attempt_status_update
|
|
request = mock.Mock()
|
|
success, msg, reload = attempt_status_update(
|
|
request,
|
|
intervention,
|
|
other_user,
|
|
'on_pause',
|
|
pause_reason='order_material',
|
|
)
|
|
self.assertTrue(success, msg)
|
|
intervention.refresh_from_db()
|
|
self.assertEqual(intervention.status, 'on_pause')
|
|
|
|
recipients = NotificationRecipient.objects.filter(user=self.user)
|
|
self.assertTrue(recipients.exists())
|
|
notif = recipients.last().notification
|
|
self.assertIn(intervention.code, notif.title)
|
|
self.assertEqual(notif.notification_type, 'alert')
|
|
from interventions.models import PAUSE_REASON_CHOICES
|
|
expected_reason = str(dict(PAUSE_REASON_CHOICES).get('order_material'))
|
|
self.assertIn(expected_reason, notif.content)
|
|
self.assertEqual(notif.sender, other_user)
|
|
|
|
def test_pause_notification_not_sent_when_paused_by_manager_themself(self):
|
|
from notifications.models import NotificationRecipient
|
|
intervention = Intervention.objects.create(
|
|
title='Pause By Manager Test',
|
|
status='in_progress',
|
|
contract=self.contract,
|
|
thematic=self.thematic,
|
|
symptom=self.symptom,
|
|
intervention_manager=self.user,
|
|
)
|
|
|
|
from interventions.views import attempt_status_update
|
|
request = mock.Mock()
|
|
success, msg, reload = attempt_status_update(
|
|
request,
|
|
intervention,
|
|
self.user,
|
|
'on_pause',
|
|
pause_reason='waiting_for_supplier',
|
|
)
|
|
self.assertTrue(success, msg)
|
|
intervention.refresh_from_db()
|
|
self.assertEqual(intervention.status, 'on_pause')
|
|
|
|
# No notification should be sent to the manager themselves
|
|
recipients = NotificationRecipient.objects.filter(user=self.user)
|
|
self.assertFalse(recipients.exists())
|
|
|
|
def test_pause_notification_not_sent_when_no_manager(self):
|
|
from notifications.models import NotificationRecipient
|
|
User = get_user_model()
|
|
other_user = User.objects.create_user(username='tech2', password='pwd')
|
|
other_config = UserConfig.objects.create(user=other_user, is_intern=True)
|
|
other_config.roles.add(self.role_admin)
|
|
UserThematics.objects.create(
|
|
user_config=other_config,
|
|
thematic=self.thematic,
|
|
can_view_interventions=True,
|
|
can_edit_interventions=True,
|
|
)
|
|
intervention = Intervention.objects.create(
|
|
title='Pause No Manager Test',
|
|
status='in_progress',
|
|
contract=self.contract,
|
|
thematic=self.thematic,
|
|
symptom=self.symptom,
|
|
intervention_manager=None,
|
|
)
|
|
|
|
from interventions.views import attempt_status_update
|
|
request = mock.Mock()
|
|
success, msg, reload = attempt_status_update(
|
|
request,
|
|
intervention,
|
|
other_user,
|
|
'on_pause',
|
|
pause_reason='reschedule',
|
|
)
|
|
self.assertTrue(success, msg)
|
|
intervention.refresh_from_db()
|
|
self.assertEqual(intervention.status, 'on_pause')
|
|
|
|
self.assertEqual(NotificationRecipient.objects.count(), 0)
|
|
|
|
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)
|
|
|
|
|
|
class InterventionDuplicationTests(TestCase):
|
|
def setUp(self):
|
|
User = get_user_model()
|
|
self.admin_user = User.objects.create_superuser(username='dup-admin', password='pwd', email='admin@example.com')
|
|
self.admin_role = Role.objects.create(name='admin')
|
|
self.admin_config = UserConfig.objects.create(user=self.admin_user, is_intern=True)
|
|
self.admin_config.roles.add(self.admin_role)
|
|
|
|
self.thematic = Thematic.objects.create(code='roads_dup', name_fr='Routes Dup', name_nl='Wegen Dup')
|
|
self.symptom = Symptom.objects.create(thematic=self.thematic, name_fr='Nid de poule', name_nl='Kuil', is_active=True)
|
|
|
|
self.company = Company.objects.create(name='Provider Co')
|
|
self.team = CompanyTeam.objects.create(name='Team A', company=self.company)
|
|
self.member = CompanyMember.objects.create(user=self.admin_user, name='Member A')
|
|
self.member.teams.add(self.team)
|
|
|
|
from datetime import date, timedelta
|
|
self.contract = Contract.objects.create(
|
|
company=self.company,
|
|
contract_number='C-DUP-01',
|
|
start_date=date.today() - timedelta(days=10),
|
|
end_date=date.today() + timedelta(days=100),
|
|
)
|
|
|
|
from django.utils import timezone
|
|
now = timezone.now()
|
|
|
|
self.itv = Intervention.objects.create(
|
|
title='Intervention originale',
|
|
description='Description détaillée',
|
|
init_description='Origine signalement',
|
|
maintain_type='corrective',
|
|
type='repair',
|
|
origin_type='damage',
|
|
priority='1',
|
|
thematic=self.thematic,
|
|
symptom=self.symptom,
|
|
contract=self.contract,
|
|
assigned_provider=self.company,
|
|
assigned_team=self.team,
|
|
assigned_member=self.member,
|
|
status='in_progress',
|
|
status_order=21,
|
|
begin_time=now - timedelta(hours=2),
|
|
end_time=now + timedelta(hours=1),
|
|
planned_begin_time=now - timedelta(hours=3),
|
|
planned_end_time=now + timedelta(hours=2),
|
|
expected_begin_time=now - timedelta(hours=4),
|
|
expected_end_time=now + timedelta(hours=3),
|
|
transmission_time=now - timedelta(days=1),
|
|
address='Rue Royale 10, 1000 Bruxelles',
|
|
location_code='LOC-001',
|
|
lon=4.35,
|
|
lat=50.85,
|
|
created_by=self.admin_user,
|
|
)
|
|
|
|
# Notes
|
|
self.origin_note = InterventionNote.objects.create(
|
|
intervention=self.itv,
|
|
content='Note origine importante',
|
|
note_time=now,
|
|
note_type=InterventionNote.NoteType.ORIGIN,
|
|
note_author=self.admin_user,
|
|
)
|
|
self.work_note = InterventionNote.objects.create(
|
|
intervention=self.itv,
|
|
content='Note travaux réalisés à ne pas copier',
|
|
note_time=now,
|
|
note_type=InterventionNote.NoteType.WORK_DONE,
|
|
note_author=self.admin_user,
|
|
)
|
|
|
|
# Document
|
|
self.doc = InterventionDocument.objects.create(
|
|
intervention=self.itv,
|
|
file=SimpleUploadedFile('test_dup_doc.txt', b'hello world duplicated file'),
|
|
uploaded_by=self.admin_user,
|
|
description='Doc description',
|
|
)
|
|
|
|
# Operation template & Operation
|
|
self.template = OperationTemplate.objects.create(name='Template Dup', thematic=self.thematic)
|
|
self.itv.equipment_operation_templates.add(self.template)
|
|
self.op = Operation.objects.create(
|
|
intervention=self.itv,
|
|
name_fr='Vérifier enrobé',
|
|
name_nl='Asfalt controleren',
|
|
template_group='Template Dup',
|
|
order=1,
|
|
is_custom=False,
|
|
)
|
|
|
|
def test_duplication_success(self):
|
|
self.client.force_login(self.admin_user)
|
|
url = reverse('interventions:duplicate_intervention', kwargs={'intervention_id': self.itv.id})
|
|
response = self.client.post(url)
|
|
|
|
# Should redirect to the new intervention detail
|
|
new_itv = Intervention.objects.exclude(id=self.itv.id).first()
|
|
self.assertIsNotNone(new_itv)
|
|
self.assertRedirects(response, reverse('interventions:intervention_detail', kwargs={'intervention_id': new_itv.id}))
|
|
|
|
# General properties copied
|
|
self.assertEqual(new_itv.title, self.itv.title)
|
|
self.assertEqual(new_itv.description, self.itv.description)
|
|
self.assertEqual(new_itv.init_description, self.itv.init_description)
|
|
self.assertEqual(new_itv.thematic, self.itv.thematic)
|
|
self.assertEqual(new_itv.symptom, self.itv.symptom)
|
|
self.assertEqual(new_itv.priority, self.itv.priority)
|
|
self.assertEqual(new_itv.maintain_type, self.itv.maintain_type)
|
|
self.assertEqual(new_itv.type, self.itv.type)
|
|
self.assertEqual(new_itv.origin_type, self.itv.origin_type)
|
|
self.assertEqual(new_itv.address, self.itv.address)
|
|
self.assertEqual(new_itv.location_code, self.itv.location_code)
|
|
self.assertEqual(new_itv.lon, self.itv.lon)
|
|
self.assertEqual(new_itv.lat, self.itv.lat)
|
|
|
|
# Status reset to in_preparation
|
|
self.assertEqual(new_itv.status, 'in_preparation')
|
|
self.assertEqual(new_itv.status_order, 1)
|
|
|
|
# Contract, provider, team, members and dates NOT copied
|
|
self.assertIsNone(new_itv.contract)
|
|
self.assertIsNone(new_itv.order)
|
|
self.assertIsNone(new_itv.assigned_provider)
|
|
self.assertIsNone(new_itv.assigned_team)
|
|
self.assertIsNone(new_itv.assigned_member)
|
|
self.assertIsNone(new_itv.intervention_manager)
|
|
self.assertIsNone(new_itv.begin_time)
|
|
self.assertIsNone(new_itv.end_time)
|
|
self.assertIsNone(new_itv.planned_begin_time)
|
|
self.assertIsNone(new_itv.planned_end_time)
|
|
self.assertIsNone(new_itv.expected_begin_time)
|
|
self.assertIsNone(new_itv.expected_end_time)
|
|
self.assertIsNone(new_itv.transmission_time)
|
|
|
|
# Template & operations copied
|
|
self.assertEqual(list(new_itv.equipment_operation_templates.all()), [self.template])
|
|
self.assertEqual(new_itv.operations.count(), 1)
|
|
new_op = new_itv.operations.first()
|
|
self.assertEqual(new_op.name_fr, self.op.name_fr)
|
|
self.assertIsNone(new_op.status)
|
|
|
|
# Notes: only origin note copied
|
|
self.assertEqual(new_itv.notes.count(), 1)
|
|
copied_note = new_itv.notes.first()
|
|
self.assertEqual(copied_note.note_type, InterventionNote.NoteType.ORIGIN)
|
|
self.assertEqual(copied_note.content, self.origin_note.content)
|
|
|
|
# Documents copied
|
|
self.assertEqual(new_itv.documents.count(), 1)
|
|
copied_doc = new_itv.documents.first()
|
|
self.assertIn('test_dup_doc', copied_doc.file.name)
|
|
self.assertEqual(copied_doc.description, self.doc.description)
|
|
|
|
# Link exists between original and new
|
|
link_exists = InterventionLink.objects.filter(
|
|
Q(first=self.itv, second=new_itv) | Q(first=new_itv, second=self.itv)
|
|
).exists()
|
|
self.assertTrue(link_exists)
|
|
|
|
# Timeline creation event exists
|
|
timeline_exists = InterventionTimeLine.objects.filter(
|
|
intervention=new_itv,
|
|
event_type='creation',
|
|
).exists()
|
|
self.assertTrue(timeline_exists)
|
|
|
|
def test_duplication_mobile_redirect(self):
|
|
self.client.force_login(self.admin_user)
|
|
url = reverse('interventions:duplicate_intervention', kwargs={'intervention_id': self.itv.id})
|
|
response = self.client.post(url, {'variant': 'mobile'})
|
|
|
|
new_itv = Intervention.objects.exclude(id=self.itv.id).first()
|
|
self.assertIsNotNone(new_itv)
|
|
self.assertRedirects(response, reverse('mobile:intervention_detail_mobile', kwargs={'intervention_id': new_itv.id}))
|
|
|
|
def test_duplication_with_associated_observation(self):
|
|
from observations.models import Observation
|
|
from interventions.models import SourceCategory
|
|
source_cat, _ = SourceCategory.objects.get_or_create(name_fr='Observation', defaults={'name_nl': 'Observatie'})
|
|
|
|
# Create an observation linked to original intervention
|
|
obs = Observation.objects.create(
|
|
thematic=self.thematic,
|
|
status='in_progress',
|
|
created_by=self.admin_user,
|
|
description="Test Observation with Intervention",
|
|
intervention=self.itv,
|
|
latitude=50.85,
|
|
longitude=4.35,
|
|
)
|
|
self.itv.source_category = source_cat
|
|
self.itv.source_ref = str(obs.id)
|
|
self.itv.save()
|
|
|
|
# Check initial associations
|
|
self.assertIn(self.itv, obs.get_associated_interventions())
|
|
self.assertIn(obs, self.itv.get_associated_observations())
|
|
|
|
# Duplicate intervention
|
|
self.client.force_login(self.admin_user)
|
|
url = reverse('interventions:duplicate_intervention', kwargs={'intervention_id': self.itv.id})
|
|
response = self.client.post(url)
|
|
self.assertEqual(response.status_code, 302)
|
|
|
|
new_itv = Intervention.objects.exclude(id=self.itv.id).first()
|
|
self.assertIsNotNone(new_itv)
|
|
|
|
# Both interventions should be associated with the observation
|
|
obs_interventions = list(obs.get_associated_interventions())
|
|
self.assertIn(self.itv, obs_interventions)
|
|
self.assertIn(new_itv, obs_interventions)
|
|
|
|
# Both interventions should retrieve the observation
|
|
self.assertIn(obs, self.itv.get_associated_observations())
|
|
self.assertIn(obs, new_itv.get_associated_observations())
|
|
|
|
# Check observation detail view displays visible_interventions
|
|
obs_url = reverse('observations:observation_detail', kwargs={'observation_id': obs.id})
|
|
obs_response = self.client.get(obs_url)
|
|
self.assertEqual(obs_response.status_code, 200)
|
|
self.assertIn('visible_interventions', obs_response.context)
|
|
visible_itv_ids = [item['intervention'].id for item in obs_response.context['visible_interventions']]
|
|
self.assertIn(self.itv.id, visible_itv_ids)
|
|
self.assertIn(new_itv.id, visible_itv_ids)
|
|
|
|
|
|
class InterventionGeojsonExternalFilterTests(TestCase):
|
|
def setUp(self):
|
|
User = get_user_model()
|
|
self.user = User.objects.create_user(username='ext_filter_user', password='pwd')
|
|
role_admin = Role.objects.create(name='admin')
|
|
self.company_user = Company.objects.create(name='Ma Société')
|
|
self.company_other = Company.objects.create(name='Autre Société')
|
|
self.thematic = Thematic.objects.create(code='THEM_GEO_EXT', name_fr='Geo Ext', name_nl='Geo Ext')
|
|
|
|
self.user_config = UserConfig.objects.create(
|
|
user=self.user,
|
|
is_intern=True,
|
|
company=self.company_user,
|
|
default_thematic=self.thematic,
|
|
limit_interventions_to_contracts=False
|
|
)
|
|
self.user_config.roles.add(role_admin)
|
|
UserThematics.objects.create(
|
|
user_config=self.user_config,
|
|
thematic=self.thematic,
|
|
can_edit_interventions=True,
|
|
can_view_interventions=True
|
|
)
|
|
|
|
# Internal intervention
|
|
self.itv_internal = Intervention.objects.create(
|
|
code='ITV-INT-001',
|
|
title='Interne',
|
|
status='to_be_processed',
|
|
thematic=self.thematic,
|
|
assigned_provider=self.company_user
|
|
)
|
|
|
|
# External intervention
|
|
self.itv_external = Intervention.objects.create(
|
|
code='ITV-EXT-001',
|
|
title='Externe',
|
|
status='to_be_processed',
|
|
thematic=self.thematic,
|
|
assigned_provider=self.company_other
|
|
)
|
|
|
|
self.url = reverse('interventions:interventions_geojson')
|
|
|
|
def test_filter_external_interventions_geojson(self):
|
|
self.client.force_login(self.user)
|
|
|
|
# 1. Without filter: both returned
|
|
resp_all = self.client.get(self.url)
|
|
self.assertEqual(resp_all.status_code, 200)
|
|
data_all = json.loads(resp_all.content)
|
|
ids_all = [f['properties']['id'] for f in data_all['features']]
|
|
self.assertIn(self.itv_internal.id, ids_all)
|
|
self.assertIn(self.itv_external.id, ids_all)
|
|
|
|
# 2. With external=1 filter: only external returned
|
|
resp_ext = self.client.get(self.url + '?external=1')
|
|
self.assertEqual(resp_ext.status_code, 200)
|
|
data_ext = json.loads(resp_ext.content)
|
|
ids_ext = [f['properties']['id'] for f in data_ext['features']]
|
|
self.assertNotIn(self.itv_internal.id, ids_ext)
|
|
self.assertIn(self.itv_external.id, ids_ext)
|
|
|
|
|
|
class TrafficlightsReprogrammingTypeTests(TestCase):
|
|
def setUp(self):
|
|
User = get_user_model()
|
|
self.user = User.objects.create_user(username='admin_tl', 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.tl_thematic, _ = Thematic.objects.get_or_create(code='trafficlights', defaults={'name_fr': 'Feux tricolores', 'name_nl': 'Verkeerslichten'})
|
|
self.struct_thematic, _ = Thematic.objects.get_or_create(code='structures', defaults={'name_fr': "Ouvrages d'art", 'name_nl': 'Kunstwerken'})
|
|
|
|
UserThematics.objects.create(user_config=self.user_config, thematic=self.tl_thematic, can_edit_interventions=True)
|
|
UserThematics.objects.create(user_config=self.user_config, thematic=self.struct_thematic, can_edit_interventions=True)
|
|
|
|
self.itv_tl = Intervention.objects.create(
|
|
code='ITV-TL-001',
|
|
title='Intervention Feux',
|
|
status='in_preparation',
|
|
thematic=self.tl_thematic,
|
|
type='fixing',
|
|
created_by=self.user
|
|
)
|
|
self.itv_struct = Intervention.objects.create(
|
|
code='ITV-ST-001',
|
|
title='Intervention Ouvrage',
|
|
status='in_preparation',
|
|
thematic=self.struct_thematic,
|
|
type='fixing',
|
|
created_by=self.user
|
|
)
|
|
|
|
def test_get_type_choices(self):
|
|
tl_choices = [c[0] for c in get_type_choices(self.tl_thematic)]
|
|
self.assertIn('reprogramming', tl_choices)
|
|
self.assertIn('fixing', tl_choices)
|
|
|
|
tl_str_choices = [c[0] for c in get_type_choices('trafficlights')]
|
|
self.assertIn('reprogramming', tl_str_choices)
|
|
|
|
struct_choices = [c[0] for c in get_type_choices(self.struct_thematic)]
|
|
self.assertNotIn('reprogramming', struct_choices)
|
|
self.assertIn('fixing', struct_choices)
|
|
|
|
struct_str_choices = [c[0] for c in get_type_choices('structures')]
|
|
self.assertNotIn('reprogramming', struct_str_choices)
|
|
|
|
none_choices = [c[0] for c in get_type_choices(None)]
|
|
self.assertNotIn('reprogramming', none_choices)
|
|
|
|
def test_intervention_form_choices(self):
|
|
from interventions.views.forms import InterventionForm
|
|
form_tl = InterventionForm(thematic=self.tl_thematic, user=self.user)
|
|
tl_choices = [c[0] for c in form_tl.fields['type'].choices]
|
|
self.assertIn('reprogramming', tl_choices)
|
|
|
|
form_struct = InterventionForm(thematic=self.struct_thematic, user=self.user)
|
|
struct_choices = [c[0] for c in form_struct.fields['type'].choices]
|
|
self.assertNotIn('reprogramming', struct_choices)
|
|
|
|
def test_preventive_intervention_form_choices(self):
|
|
from interventions.views.forms import PreventiveInterventionForm
|
|
form_tl = PreventiveInterventionForm(thematic=self.tl_thematic, user=self.user)
|
|
tl_choices = [c[0] for c in form_tl.fields['type'].choices]
|
|
self.assertIn('reprogramming', tl_choices)
|
|
|
|
form_struct = PreventiveInterventionForm(thematic=self.struct_thematic, user=self.user)
|
|
struct_choices = [c[0] for c in form_struct.fields['type'].choices]
|
|
self.assertNotIn('reprogramming', struct_choices)
|
|
|
|
def test_intervention_detail_context(self):
|
|
self.client.force_login(self.user)
|
|
|
|
resp_tl = self.client.get(reverse('interventions:intervention_detail', kwargs={'intervention_id': self.itv_tl.id}))
|
|
self.assertEqual(resp_tl.status_code, 200)
|
|
tl_type_choices = [c[0] for c in resp_tl.context['type_choices']]
|
|
self.assertIn('reprogramming', tl_type_choices)
|
|
|
|
resp_struct = self.client.get(reverse('interventions:intervention_detail', kwargs={'intervention_id': self.itv_struct.id}))
|
|
self.assertEqual(resp_struct.status_code, 200)
|
|
struct_type_choices = [c[0] for c in resp_struct.context['type_choices']]
|
|
self.assertNotIn('reprogramming', struct_type_choices)
|
|
|
|
def test_update_intervention_type_reprogramming(self):
|
|
self.client.force_login(self.user)
|
|
|
|
# Updating trafficlights intervention to reprogramming -> SUCCESS
|
|
update_url_tl = reverse('interventions:update_intervention', kwargs={'intervention_id': self.itv_tl.id})
|
|
resp = self.client.post(
|
|
update_url_tl,
|
|
data=json.dumps({'type': 'reprogramming'}),
|
|
content_type='application/json'
|
|
)
|
|
self.assertEqual(resp.status_code, 200)
|
|
self.itv_tl.refresh_from_db()
|
|
self.assertEqual(self.itv_tl.type, 'reprogramming')
|
|
|
|
# Updating structures intervention to reprogramming -> REJECTED
|
|
update_url_struct = reverse('interventions:update_intervention', kwargs={'intervention_id': self.itv_struct.id})
|
|
resp = self.client.post(
|
|
update_url_struct,
|
|
data=json.dumps({'type': 'reprogramming'}),
|
|
content_type='application/json'
|
|
)
|
|
self.itv_struct.refresh_from_db()
|
|
self.assertEqual(self.itv_struct.type, 'fixing')
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|