693 lines
26 KiB
Python
693 lines
26 KiB
Python
import json
|
||
from datetime import timedelta
|
||
from unittest import mock
|
||
|
||
from django.test import TestCase
|
||
from django.contrib.auth import get_user_model
|
||
from django.contrib import messages
|
||
from django.contrib.messages import get_messages
|
||
from django.core.files.uploadedfile import SimpleUploadedFile
|
||
|
||
from common.models import UserConfig, Role, Thematic, UserThematics
|
||
from contracts.models import Company, CompanyTeam, CompanyMember, Contract, ContractPost, ContractOrder, ContractOrderQuote, ContractOrderQuoteItem
|
||
from assets.models import AssetCategory
|
||
from interventions.models import (
|
||
Intervention,
|
||
InterventionContractPost,
|
||
InterventionPlanificationTimeLine,
|
||
InterventionRealizationTimeLine,
|
||
Operation,
|
||
InterventionAsset,
|
||
OperationTemplate,
|
||
OperationTemplateItem,
|
||
OperationStatusSet,
|
||
OperationStatus,
|
||
STATUS_ORDERS,
|
||
InterventionDocument,
|
||
)
|
||
from django.contrib.contenttypes.models import ContentType
|
||
from interventions.views import _group_operations_by_asset_and_template, _create_equipment_operations_for_assets
|
||
from interventions.permissions import can_manage_occupations, can_delete_document
|
||
from django.urls import reverse
|
||
from django.utils import timezone
|
||
from decimal import Decimal
|
||
|
||
|
||
|
||
class ConfirmCallTests(TestCase):
|
||
def setUp(self):
|
||
User = get_user_model()
|
||
self.user = User.objects.create_user(username='caller', 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.intervention = Intervention.objects.create(title='Call intervention', status='in_preparation')
|
||
|
||
def test_confirm_call_persists_payload(self):
|
||
self.client.login(username='caller', password='pwd')
|
||
url = reverse('interventions:confirm_call', args=[self.intervention.id])
|
||
tz = timezone.get_current_timezone()
|
||
call_dt = timezone.now().astimezone(tz).replace(second=0, microsecond=0)
|
||
payload = {
|
||
'call_received': False,
|
||
'call_datetime': call_dt.strftime('%Y-%m-%dT%H:%M'),
|
||
}
|
||
|
||
response = self.client.post(url, data=json.dumps(payload), content_type='application/json')
|
||
|
||
self.assertEqual(response.status_code, 200)
|
||
self.intervention.refresh_from_db()
|
||
self.assertFalse(self.intervention.init_call_received)
|
||
self.assertEqual(self.intervention.init_call_confirmed_time, call_dt)
|
||
|
||
def test_confirm_call_defaults_to_now_and_received(self):
|
||
self.client.login(username='caller', password='pwd')
|
||
url = reverse('interventions:confirm_call', args=[self.intervention.id])
|
||
|
||
response = self.client.post(url, data=json.dumps({}), content_type='application/json')
|
||
|
||
self.assertEqual(response.status_code, 200)
|
||
self.intervention.refresh_from_db()
|
||
self.assertTrue(self.intervention.init_call_received)
|
||
self.assertIsNotNone(self.intervention.init_call_confirmed_time)
|
||
|
||
|
||
|
||
|
||
|
||
class ApplyQuoteTests(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)
|
||
|
||
# Minimal contract/order/posts/quote setup
|
||
self.company = Company.objects.create(name='Comp')
|
||
from datetime import date, timedelta
|
||
today = date.today()
|
||
self.contract = Contract.objects.create(
|
||
company=self.company,
|
||
contract_number='C-001',
|
||
start_date=today - timedelta(days=30),
|
||
end_date=today + timedelta(days=365)
|
||
)
|
||
self.post1 = ContractPost.objects.create(
|
||
contract=self.contract,
|
||
order_number=1,
|
||
code='P1',
|
||
description_fr='Poste 1',
|
||
description_nl='Post 1',
|
||
unit='u',
|
||
unit_price=Decimal('10.00')
|
||
)
|
||
self.post2 = ContractPost.objects.create(
|
||
contract=self.contract,
|
||
order_number=2,
|
||
code='P2',
|
||
description_fr='Poste 2',
|
||
description_nl='Post 2',
|
||
unit='u',
|
||
unit_price=Decimal('20.00')
|
||
)
|
||
|
||
self.order = ContractOrder.objects.create(
|
||
contract=self.contract,
|
||
order_code='ORD-1',
|
||
order_status='pending',
|
||
order_date=today,
|
||
delivery_date=today + timedelta(days=7)
|
||
)
|
||
self.quote = ContractOrderQuote.objects.create(order=self.order, is_finalized=True)
|
||
ContractOrderQuoteItem.objects.create(quote=self.quote, contract_post=self.post1, quantity=Decimal('5.00'))
|
||
ContractOrderQuoteItem.objects.create(quote=self.quote, contract_post=self.post2, quantity=Decimal('3.00'))
|
||
|
||
self.intervention = Intervention.objects.create(title='With order', contract=self.contract, order=self.order, status='to_be_processed')
|
||
|
||
def test_apply_quote_adds_missing_posts_without_duplicates(self):
|
||
self.client.login(username='admin', password='pwd')
|
||
|
||
# Pre-create one link to ensure no duplicate is created
|
||
InterventionContractPost.objects.create(intervention=self.intervention, contract_post=self.post1, quantity=Decimal('1.00'))
|
||
|
||
url = reverse('interventions:intervention_apply_quote', args=[self.intervention.id])
|
||
resp = self.client.post(url)
|
||
self.assertEqual(resp.status_code, 200)
|
||
data = resp.json()
|
||
self.assertTrue(data.get('success'))
|
||
|
||
# We already had post1; post2 should be created, not duplicated
|
||
posts = InterventionContractPost.objects.filter(intervention=self.intervention).values_list('contract_post_id', flat=True)
|
||
self.assertSetEqual(set(posts), {self.post1.id, self.post2.id})
|
||
# created count should be 1 (only post2)
|
||
self.assertEqual(data.get('created'), 1)
|
||
|
||
|
||
|
||
|
||
class TimelineEventManagementTests(TestCase):
|
||
def setUp(self):
|
||
User = get_user_model()
|
||
self.user = User.objects.create_user(username='timeline-admin', 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.intervention = Intervention.objects.create(
|
||
title='Timeline control',
|
||
status='to_be_processed',
|
||
)
|
||
|
||
self.url = reverse('interventions:update_intervention', args=[self.intervention.id])
|
||
self.client.login(username='timeline-admin', password='pwd')
|
||
|
||
def test_planification_future_event_can_be_updated(self):
|
||
future_time = timezone.now() + timedelta(days=2)
|
||
event = InterventionPlanificationTimeLine.objects.create(
|
||
intervention=self.intervention,
|
||
event_time=future_time,
|
||
event_type='planned_begin',
|
||
event_description='Initial planning',
|
||
)
|
||
|
||
new_time = future_time + timedelta(days=1)
|
||
response = self.client.post(self.url, {
|
||
'field': 'planification',
|
||
'action': 'update',
|
||
'event_id': str(event.id),
|
||
'event_time': new_time.isoformat(),
|
||
'event_type': 'planned_end',
|
||
'event_description': 'Updated planning',
|
||
})
|
||
|
||
self.assertEqual(response.status_code, 200)
|
||
payload = response.json()
|
||
self.assertTrue(payload.get('success'))
|
||
|
||
event.refresh_from_db()
|
||
self.assertEqual(event.event_type, 'planned_end')
|
||
self.assertEqual(event.event_description, 'Updated planning')
|
||
self.assertEqual(event.event_time, new_time)
|
||
|
||
self.intervention.refresh_from_db()
|
||
self.assertEqual(self.intervention.planned_end_time, event.event_time)
|
||
|
||
def test_planification_past_event_cannot_be_deleted(self):
|
||
past_event = InterventionPlanificationTimeLine.objects.create(
|
||
intervention=self.intervention,
|
||
event_time=timezone.now() - timedelta(days=1),
|
||
event_type='planned_begin',
|
||
event_description='Past event',
|
||
)
|
||
|
||
response = self.client.post(self.url, {
|
||
'field': 'planification',
|
||
'action': 'delete',
|
||
'event_id': str(past_event.id),
|
||
})
|
||
|
||
self.assertEqual(response.status_code, 400)
|
||
payload = response.json()
|
||
self.assertFalse(payload.get('success'))
|
||
self.assertTrue(
|
||
InterventionPlanificationTimeLine.objects.filter(id=past_event.id).exists()
|
||
)
|
||
|
||
def test_realization_future_event_can_be_deleted(self):
|
||
event = InterventionRealizationTimeLine.objects.create(
|
||
intervention=self.intervention,
|
||
event_time=timezone.now() + timedelta(hours=6),
|
||
event_type='real_begin',
|
||
event_description='Future real event',
|
||
)
|
||
|
||
response = self.client.post(self.url, {
|
||
'field': 'realization',
|
||
'action': 'delete',
|
||
'event_id': str(event.id),
|
||
})
|
||
|
||
self.assertEqual(response.status_code, 200)
|
||
self.assertTrue(response.json().get('success'))
|
||
self.assertFalse(
|
||
InterventionRealizationTimeLine.objects.filter(id=event.id).exists()
|
||
)
|
||
|
||
|
||
|
||
|
||
class InterventionStatusTransitionTests(TestCase):
|
||
def setUp(self):
|
||
User = get_user_model()
|
||
self.admin = User.objects.create_user(username='admin_user', password='pwd')
|
||
self.role_admin = Role.objects.create(name='admin')
|
||
self.admin_config = UserConfig.objects.create(user=self.admin, is_intern=True)
|
||
self.admin_config.roles.add(self.role_admin)
|
||
|
||
self.manager = User.objects.create_user(username='manager_user', password='pwd')
|
||
self.role_manager = Role.objects.create(name='manager')
|
||
self.manager_config = UserConfig.objects.create(user=self.manager, is_intern=True)
|
||
self.manager_config.roles.add(self.role_manager)
|
||
|
||
self.thematic = Thematic.objects.create(code='roads', name_fr='Routes', name_nl='Wegen')
|
||
UserThematics.objects.create(
|
||
user_config=self.manager_config,
|
||
thematic=self.thematic,
|
||
can_view_interventions=True,
|
||
can_edit_interventions=True,
|
||
)
|
||
|
||
self.company = Company.objects.create(name='Provider Co')
|
||
from datetime import date
|
||
today = date.today()
|
||
self.contract = Contract.objects.create(
|
||
company=self.company,
|
||
contract_number='C-TEST',
|
||
start_date=today - timedelta(days=30),
|
||
end_date=today + timedelta(days=365)
|
||
)
|
||
|
||
from common.models import UserContractAccess, UserContractStatusPermission
|
||
self.uca = UserContractAccess.objects.create(
|
||
user_config=self.manager_config,
|
||
contract=self.contract,
|
||
can_view_interventions=True,
|
||
)
|
||
|
||
UserContractStatusPermission.objects.create(
|
||
user_contract=self.uca,
|
||
status='canceled',
|
||
can_change_status_to=True,
|
||
)
|
||
|
||
UserContractStatusPermission.objects.create(
|
||
user_contract=self.uca,
|
||
status='to_be_processed',
|
||
can_change_status_to=True,
|
||
)
|
||
|
||
def test_manager_cannot_cancel_intervention_assigned_or_beyond(self):
|
||
itv_to_be_processed = Intervention.objects.create(
|
||
title='Test active',
|
||
status='to_be_processed',
|
||
contract=self.contract,
|
||
thematic=self.thematic,
|
||
)
|
||
from interventions.permissions import user_allowed_transitions
|
||
allowed_for_to_be_processed = user_allowed_transitions(self.manager, 'to_be_processed', intervention=itv_to_be_processed)
|
||
self.assertIn('canceled', allowed_for_to_be_processed)
|
||
|
||
itv_assigned = Intervention.objects.create(
|
||
title='Test assigned',
|
||
status='assigned',
|
||
contract=self.contract,
|
||
thematic=self.thematic,
|
||
)
|
||
allowed_for_assigned = user_allowed_transitions(self.manager, 'assigned', intervention=itv_assigned)
|
||
self.assertNotIn('canceled', allowed_for_assigned)
|
||
|
||
itv_in_progress = Intervention.objects.create(
|
||
title='Test in progress',
|
||
status='in_progress',
|
||
contract=self.contract,
|
||
thematic=self.thematic,
|
||
)
|
||
allowed_for_in_progress = user_allowed_transitions(self.manager, 'in_progress', intervention=itv_in_progress)
|
||
self.assertNotIn('canceled', allowed_for_in_progress)
|
||
|
||
def test_admin_can_cancel_intervention_regardless_of_status(self):
|
||
itv_assigned = Intervention.objects.create(
|
||
title='Test assigned admin',
|
||
status='assigned',
|
||
contract=self.contract,
|
||
thematic=self.thematic,
|
||
)
|
||
from interventions.permissions import user_allowed_transitions
|
||
allowed_for_assigned = user_allowed_transitions(self.admin, 'assigned', intervention=itv_assigned)
|
||
self.assertIn('canceled', allowed_for_assigned)
|
||
|
||
def test_cancellation_resets_linked_in_progress_observations(self):
|
||
from observations.models import Observation
|
||
from interventions.views import attempt_status_update
|
||
|
||
# Create intervention
|
||
itv = Intervention.objects.create(
|
||
title='Test cancel intervention',
|
||
status='to_be_processed',
|
||
contract=self.contract,
|
||
thematic=self.thematic,
|
||
)
|
||
|
||
# Create linked observation in_progress
|
||
obs_in_progress = Observation.objects.create(
|
||
description='Test obs en cours',
|
||
latitude=50.85,
|
||
longitude=4.35,
|
||
status='in_progress',
|
||
intervention=itv,
|
||
thematic=self.thematic,
|
||
)
|
||
|
||
# Create linked observation not in_progress (e.g. to_process)
|
||
obs_to_process = Observation.objects.create(
|
||
description='Test obs a traiter',
|
||
latitude=50.85,
|
||
longitude=4.35,
|
||
status='to_process',
|
||
intervention=itv,
|
||
thematic=self.thematic,
|
||
)
|
||
|
||
# Cancel the intervention
|
||
success, message, email_failed = attempt_status_update(
|
||
request=None,
|
||
intervention=itv,
|
||
user=self.admin,
|
||
new_status='canceled',
|
||
cancellation_type='to_redirect',
|
||
)
|
||
self.assertTrue(success)
|
||
|
||
# Refresh from db
|
||
obs_in_progress.refresh_from_db()
|
||
obs_to_process.refresh_from_db()
|
||
|
||
# The in_progress observation must have transitioned to to_process
|
||
self.assertEqual(obs_in_progress.status, 'to_process')
|
||
# The to_process observation must remain unchanged
|
||
self.assertEqual(obs_to_process.status, 'to_process')
|
||
|
||
# Check that a comment was added to the transitioned observation
|
||
self.assertTrue(obs_in_progress.comments.filter(comment__contains=itv.code).exists())
|
||
self.assertFalse(obs_to_process.comments.filter(comment__contains=itv.code).exists())
|
||
|
||
def test_create_intervention_from_already_linked_observation(self):
|
||
from observations.models import Observation
|
||
from interventions.models import Symptom, AssetCategory
|
||
from common.models import UserContractStatusPermission
|
||
from django.urls import reverse
|
||
|
||
# Create first intervention
|
||
itv1 = Intervention.objects.create(
|
||
title='First intervention',
|
||
status='canceled',
|
||
contract=self.contract,
|
||
thematic=self.thematic,
|
||
)
|
||
|
||
# Create observation linked to itv1
|
||
obs = Observation.objects.create(
|
||
description='Test obs duplicate creation',
|
||
latitude=50.85,
|
||
longitude=4.35,
|
||
status='to_process',
|
||
intervention=itv1,
|
||
thematic=self.thematic,
|
||
)
|
||
|
||
# 1. Test redirect from create_intervention_from_observation view
|
||
self.client.login(username='manager_user', password='pwd')
|
||
url = reverse('observations:create_intervention_from_observation', args=[obs.id])
|
||
response = self.client.post(url)
|
||
self.assertEqual(response.status_code, 302)
|
||
# It should redirect to the add screen, not the detail screen of itv1
|
||
self.assertIn(reverse('interventions:intervention_add', kwargs={'thematic_code': self.thematic.code}), response.url)
|
||
self.assertIn(f"observation_id={obs.id}", response.url)
|
||
|
||
# 2. Test saving the new intervention updates the observation's intervention link
|
||
add_url = reverse('interventions:intervention_add', kwargs={'thematic_code': self.thematic.code})
|
||
category = AssetCategory.objects.create(
|
||
code='CAT-ADD',
|
||
name_fr='Cat-Add',
|
||
name_nl='Cat-Add',
|
||
thematic=self.thematic,
|
||
)
|
||
symptom = Symptom.objects.create(
|
||
code='SYMP-ADD',
|
||
name_fr='Symp-Add',
|
||
name_nl='Symp-Add',
|
||
thematic=self.thematic,
|
||
)
|
||
UserContractStatusPermission.objects.create(
|
||
user_contract=self.uca,
|
||
status='in_preparation',
|
||
can_change_status_to=True,
|
||
)
|
||
|
||
post_data = {
|
||
'title': 'New intervention for linked obs',
|
||
'type': 'fixing',
|
||
'status': 'in_preparation',
|
||
'symptom': symptom.id,
|
||
'asset_category': category.id,
|
||
'maintain_type': 'corrective',
|
||
'lon': '4.35',
|
||
'lat': '50.85',
|
||
'priority': '3',
|
||
'origin_type': 'demand',
|
||
'contract': self.contract.id,
|
||
'observation_id': obs.id,
|
||
}
|
||
|
||
response = self.client.post(add_url, data=post_data)
|
||
self.assertEqual(response.status_code, 302)
|
||
|
||
# Get the new intervention
|
||
itv2 = Intervention.objects.get(title='New intervention for linked obs')
|
||
|
||
# Verify the observation is now linked to the new intervention itv2
|
||
obs.refresh_from_db()
|
||
self.assertEqual(obs.intervention, itv2)
|
||
|
||
|
||
|
||
|
||
class AutoCloseInterventionTests(TestCase):
|
||
def setUp(self):
|
||
from common.models import UserConfig, Role, Thematic, UserThematics
|
||
from contracts.models import Company, Contract
|
||
from interventions.models import Intervention
|
||
from observations.models import Observation
|
||
|
||
User = get_user_model()
|
||
self.user = User.objects.create_user(username='autoclose_tester', 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')
|
||
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='Test Company')
|
||
|
||
from datetime import date
|
||
self.contract_auto = Contract.objects.create(
|
||
company=self.company,
|
||
contract_number='C-AUTO-CLOSE',
|
||
start_date=date.today() - timedelta(days=100),
|
||
end_date=date.today() + timedelta(days=100),
|
||
is_active=True,
|
||
)
|
||
|
||
self.contract_no_auto = Contract.objects.create(
|
||
company=self.company,
|
||
contract_number='C-NO-AUTO-CLOSE',
|
||
start_date=date.today() - timedelta(days=100),
|
||
end_date=date.today() + timedelta(days=100),
|
||
is_active=True,
|
||
)
|
||
|
||
def test_auto_close_expired_interventions(self):
|
||
from django.core.management import call_command
|
||
from interventions.models import Intervention, InterventionTimeLine, Symptom, SymptomContractAutoClose
|
||
from observations.models import Observation, ObservationComment
|
||
|
||
# Create a symptom and a rule (symptom × contract_auto, delay 5 days)
|
||
symptom = Symptom.objects.create(
|
||
code='SYMP-TEST',
|
||
name_fr='Symptôme test',
|
||
name_nl='Symptoom test',
|
||
thematic=self.thematic,
|
||
)
|
||
SymptomContractAutoClose.objects.create(
|
||
symptom=symptom,
|
||
contract=self.contract_auto,
|
||
delay_days=5,
|
||
)
|
||
|
||
# 1. Create eligible intervention that is expired
|
||
past_time = timezone.now() - timedelta(days=10)
|
||
itv_expired = Intervention.objects.create(
|
||
title='Expired intervention',
|
||
status='finished',
|
||
contract=self.contract_auto,
|
||
thematic=self.thematic,
|
||
symptom=symptom,
|
||
)
|
||
# Manually create TimelineEvent to simulate completion time in the past
|
||
InterventionTimeLine.objects.create(
|
||
intervention=itv_expired,
|
||
event_time=past_time,
|
||
event_type='status_change',
|
||
to_status='finished',
|
||
)
|
||
|
||
# Link an observation
|
||
obs = Observation.objects.create(
|
||
description='Obs linked to expired intervention',
|
||
status='to_process',
|
||
intervention=itv_expired,
|
||
thematic=self.thematic,
|
||
latitude=50.85,
|
||
longitude=4.35,
|
||
)
|
||
|
||
# 2. Create eligible intervention that is NOT expired
|
||
recent_time = timezone.now() - timedelta(days=2)
|
||
itv_recent = Intervention.objects.create(
|
||
title='Recent intervention',
|
||
status='finished',
|
||
contract=self.contract_auto,
|
||
thematic=self.thematic,
|
||
symptom=symptom,
|
||
)
|
||
InterventionTimeLine.objects.create(
|
||
intervention=itv_recent,
|
||
event_time=recent_time,
|
||
event_type='status_change',
|
||
to_status='finished',
|
||
)
|
||
|
||
# 3. Create eligible intervention on contract_no_auto (no rule → should NOT be closed)
|
||
itv_no_auto = Intervention.objects.create(
|
||
title='No auto close intervention',
|
||
status='finished',
|
||
contract=self.contract_no_auto,
|
||
thematic=self.thematic,
|
||
symptom=symptom,
|
||
)
|
||
InterventionTimeLine.objects.create(
|
||
intervention=itv_no_auto,
|
||
event_time=past_time,
|
||
event_type='status_change',
|
||
to_status='finished',
|
||
)
|
||
|
||
# 4. Create intervention in a non-eligible status (e.g. in_progress)
|
||
itv_non_eligible = Intervention.objects.create(
|
||
title='In progress intervention',
|
||
status='in_progress',
|
||
contract=self.contract_auto,
|
||
thematic=self.thematic,
|
||
symptom=symptom,
|
||
)
|
||
InterventionTimeLine.objects.create(
|
||
intervention=itv_non_eligible,
|
||
event_time=past_time,
|
||
event_type='status_change',
|
||
to_status='in_progress',
|
||
)
|
||
|
||
# Run the management command
|
||
call_command('close_expired_interventions')
|
||
|
||
# Refresh from database
|
||
itv_expired.refresh_from_db()
|
||
itv_recent.refresh_from_db()
|
||
itv_no_auto.refresh_from_db()
|
||
itv_non_eligible.refresh_from_db()
|
||
obs.refresh_from_db()
|
||
|
||
# Check results
|
||
self.assertEqual(itv_expired.status, 'closed')
|
||
self.assertEqual(obs.status, 'closed')
|
||
# Check that comment was created
|
||
self.assertTrue(ObservationComment.objects.filter(observation=obs, comment__contains=itv_expired.code).exists())
|
||
|
||
self.assertEqual(itv_recent.status, 'finished')
|
||
self.assertEqual(itv_no_auto.status, 'finished')
|
||
self.assertEqual(itv_non_eligible.status, 'in_progress')
|
||
|
||
def test_auto_close_expired_interventions_by_symptom(self):
|
||
from django.core.management import call_command
|
||
from interventions.models import Intervention, InterventionTimeLine, Symptom, SymptomContractAutoClose
|
||
from observations.models import Observation, ObservationComment
|
||
|
||
# Create symptom and rule: symptom_auto × contract_no_auto (delay 3 days)
|
||
symptom_auto = Symptom.objects.create(
|
||
code='SYMP-AUTO-CLOSE',
|
||
name_fr='Symptôme auto-clôture',
|
||
name_nl='Symptoom auto-clôture',
|
||
thematic=self.thematic,
|
||
)
|
||
SymptomContractAutoClose.objects.create(
|
||
symptom=symptom_auto,
|
||
contract=self.contract_no_auto,
|
||
delay_days=3,
|
||
)
|
||
|
||
# 1. Create eligible intervention that is expired under the rule (past 4 days)
|
||
past_time = timezone.now() - timedelta(days=4)
|
||
itv_expired = Intervention.objects.create(
|
||
title='Expired intervention by symptom',
|
||
status='finished',
|
||
contract=self.contract_no_auto,
|
||
thematic=self.thematic,
|
||
symptom=symptom_auto,
|
||
)
|
||
InterventionTimeLine.objects.create(
|
||
intervention=itv_expired,
|
||
event_time=past_time,
|
||
event_type='status_change',
|
||
to_status='finished',
|
||
)
|
||
|
||
# Link an observation
|
||
obs = Observation.objects.create(
|
||
description='Obs linked to symptom expired intervention',
|
||
status='to_process',
|
||
intervention=itv_expired,
|
||
thematic=self.thematic,
|
||
latitude=50.85,
|
||
longitude=4.35,
|
||
)
|
||
|
||
# 2. Create eligible intervention that is NOT expired (2 days past completion)
|
||
recent_time = timezone.now() - timedelta(days=2)
|
||
itv_recent = Intervention.objects.create(
|
||
title='Recent intervention by symptom',
|
||
status='finished',
|
||
contract=self.contract_no_auto,
|
||
thematic=self.thematic,
|
||
symptom=symptom_auto,
|
||
)
|
||
InterventionTimeLine.objects.create(
|
||
intervention=itv_recent,
|
||
event_time=recent_time,
|
||
event_type='status_change',
|
||
to_status='finished',
|
||
)
|
||
|
||
# Run the management command
|
||
call_command('close_expired_interventions')
|
||
|
||
# Refresh from database
|
||
itv_expired.refresh_from_db()
|
||
itv_recent.refresh_from_db()
|
||
obs.refresh_from_db()
|
||
|
||
# Check results
|
||
self.assertEqual(itv_expired.status, 'closed')
|
||
self.assertEqual(obs.status, 'closed')
|
||
self.assertTrue(ObservationComment.objects.filter(observation=obs, comment__contains=itv_expired.code).exists())
|
||
self.assertEqual(itv_recent.status, 'finished')
|
||
|
||
|
||
|
||
|