fix: use latest finished status timeline date as effective end date for intervention filtering and summary views

This commit is contained in:
kdeterme 2026-08-31 11:17:42 +02:00
parent 32eec7b442
commit 5c042531e5
6 changed files with 222 additions and 8 deletions

View file

@ -1,5 +1,6 @@
import json import json
from datetime import timedelta from datetime import datetime, timedelta, date
from django.utils import timezone
from unittest import mock from unittest import mock
from django.test import TestCase from django.test import TestCase
@ -1773,6 +1774,182 @@ class TrafficlightsReprogrammingTypeTests(TestCase):
self.assertEqual(self.itv_struct.type, 'fixing') self.assertEqual(self.itv_struct.type, 'fixing')
class InterventionEndDateFilteringTests(TestCase):
def setUp(self):
self.User = get_user_model()
self.user = self.User.objects.create_user(username='date-filter-user', password='pwd')
self.role_admin, _ = Role.objects.get_or_create(name='admin')
self.config = UserConfig.objects.create(user=self.user, is_intern=True)
self.config.roles.add(self.role_admin)
self.thematic = Thematic.objects.create(
code='roads_date_test',
name_fr='Voirie Date Test',
name_nl='Wegen Date Test',
allow_planning_without_contract=True
)
UserThematics.objects.create(
user_config=self.config,
thematic=self.thematic,
can_view_interventions=True,
can_edit_interventions=True,
)
from interventions.models import Intervention, InterventionTimeLine
# itv1: passed to 'finished' on 2026-08-15 (end_time is None)
self.itv1 = Intervention.objects.create(
title='Finished on Aug 15',
status='finished',
thematic=self.thematic,
created_by=self.user,
)
InterventionTimeLine.objects.create(
intervention=self.itv1,
event_user=self.user,
event_time=timezone.make_aware(datetime(2026, 8, 15, 14, 30)),
event_type='status_change',
from_status='in_progress',
to_status='finished',
)
# itv2: passed to 'finished' on 2026-08-25 (end_time is None)
self.itv2 = Intervention.objects.create(
title='Finished on Aug 25',
status='finished',
thematic=self.thematic,
created_by=self.user,
)
InterventionTimeLine.objects.create(
intervention=self.itv2,
event_user=self.user,
event_time=timezone.make_aware(datetime(2026, 8, 25, 10, 0)),
event_type='status_change',
from_status='in_progress',
to_status='finished',
)
# itv3: not finished, has planned_end_time on 2026-08-15
self.itv3 = Intervention.objects.create(
title='Planned Aug 15',
status='in_progress',
thematic=self.thematic,
planned_end_time=timezone.make_aware(datetime(2026, 8, 15, 18, 0)),
created_by=self.user,
)
# itv4: passed to 'finished' on 2026-08-15, but has an old/different end_time on 2026-07-01
self.itv4 = Intervention.objects.create(
title='Finished Aug 15 with old end_time',
status='validated',
end_time=timezone.make_aware(datetime(2026, 7, 1, 9, 0)),
thematic=self.thematic,
created_by=self.user,
)
InterventionTimeLine.objects.create(
intervention=self.itv4,
event_user=self.user,
event_time=timezone.make_aware(datetime(2026, 8, 15, 16, 0)),
event_type='status_change',
from_status='in_progress',
to_status='finished',
)
def test_apply_intervention_filters_uses_finished_timeline_date(self):
from interventions.views import apply_intervention_filters_from_querydict
from django.http import QueryDict
qs = Intervention.objects.filter(thematic=self.thematic)
# Range 2026-08-10 to 2026-08-20
params = QueryDict('end_date_min=2026-08-10&end_date_max=2026-08-20')
filtered = apply_intervention_filters_from_querydict(qs, params, user=self.user)
# itv1 (finished Aug 15) and itv4 (finished Aug 15) and itv3 (planned Aug 15) should match
self.assertIn(self.itv1, filtered)
self.assertIn(self.itv3, filtered)
self.assertIn(self.itv4, filtered)
# itv2 (finished Aug 25) should NOT match
self.assertNotIn(self.itv2, filtered)
def test_apply_intervention_filters_prefers_finished_timeline_over_different_end_time(self):
from interventions.views import apply_intervention_filters_from_querydict
from django.http import QueryDict
qs = Intervention.objects.filter(thematic=self.thematic)
# Range in July: itv4 has end_time in July but its finished timeline date is in August
params_july = QueryDict('end_date_min=2026-07-01&end_date_max=2026-07-31')
filtered_july = apply_intervention_filters_from_querydict(qs, params_july, user=self.user)
self.assertNotIn(self.itv4, filtered_july)
# Range in August: itv4 must be included because it was finished on Aug 15
params_august = QueryDict('end_date_min=2026-08-01&end_date_max=2026-08-31')
filtered_august = apply_intervention_filters_from_querydict(qs, params_august, user=self.user)
self.assertIn(self.itv4, filtered_august)
def test_interventions_geojson_end_date_filter(self):
self.client.force_login(self.user)
url = reverse('interventions:interventions_geojson')
# Filter range: 2026-08-10 to 2026-08-20
response = self.client.get(url, {
'thematic': 'roads_date_test',
'end_date_min': '2026-08-10',
'end_date_max': '2026-08-20',
'status': 'in_progress,finished,validated',
})
self.assertEqual(response.status_code, 200)
data = response.json()
feature_ids = [f['id'] for f in data['features']]
self.assertIn(self.itv1.id, feature_ids)
self.assertIn(self.itv3.id, feature_ids)
self.assertIn(self.itv4.id, feature_ids)
self.assertNotIn(self.itv2.id, feature_ids)
def test_multiple_finished_transitions_uses_latest_date(self):
from interventions.models import Intervention, InterventionTimeLine
from interventions.views import apply_intervention_filters_from_querydict
from django.http import QueryDict
# Create an intervention finished in June, then reopened, then finished in August
itv_reopened = Intervention.objects.create(
title='Finished then Reopened then Finished',
status='finished',
thematic=self.thematic,
created_by=self.user,
)
InterventionTimeLine.objects.create(
intervention=itv_reopened,
event_user=self.user,
event_time=timezone.make_aware(datetime(2026, 6, 10, 10, 0)),
event_type='status_change',
from_status='in_progress',
to_status='finished',
)
InterventionTimeLine.objects.create(
intervention=itv_reopened,
event_user=self.user,
event_time=timezone.make_aware(datetime(2026, 8, 20, 11, 0)),
event_type='status_change',
from_status='to_be_corrected',
to_status='finished',
)
qs = Intervention.objects.filter(pk=itv_reopened.pk)
# In June: should NOT match because latest finished is in August
params_june = QueryDict('end_date_min=2026-06-01&end_date_max=2026-06-30')
self.assertNotIn(itv_reopened, apply_intervention_filters_from_querydict(qs, params_june, user=self.user))
# In August: should match because latest finished is Aug 20
params_august = QueryDict('end_date_min=2026-08-01&end_date_max=2026-08-31')
self.assertIn(itv_reopened, apply_intervention_filters_from_querydict(qs, params_august, user=self.user))

View file

@ -430,10 +430,15 @@ def intervention_detail(request, intervention_id):
.distinct() .distinct()
) )
# Récupérer les interventions des 3 derniers mois # Récupérer les interventions des 3 derniers mois
_latest_finished_sq = InterventionTimeLine.objects.filter(
intervention=OuterRef('pk'),
to_status='finished',
).order_by('-event_time').values('event_time')[:1]
related_interventions = ( related_interventions = (
Intervention.objects Intervention.objects
.annotate( .annotate(
effective_end=Coalesce( effective_end=Coalesce(
Subquery(_latest_finished_sq),
'end_time', 'end_time',
'planned_end_time', 'planned_end_time',
'expected_end_time', 'expected_end_time',

View file

@ -575,13 +575,18 @@ def interventions_geojson(request):
if preparation_states: if preparation_states:
qs = qs.filter(prep_state__in=preparation_states) qs = qs.filter(prep_state__in=preparation_states)
_latest_finished_sq = InterventionTimeLine.objects.filter(
intervention=OuterRef('pk'),
to_status='finished',
).order_by('-event_time').values('event_time')[:1]
# --- annotate once (DB does geometry + dates) # --- annotate once (DB does geometry + dates)
if date_type == 'planned': if date_type == 'planned':
effective_begin = Coalesce('begin_time', 'planned_begin_time', output_field=DateTimeField()) effective_begin = Coalesce('begin_time', 'planned_begin_time', output_field=DateTimeField())
effective_end = Coalesce('end_time', 'planned_end_time', output_field=DateTimeField()) effective_end = Coalesce(Subquery(_latest_finished_sq), 'end_time', 'planned_end_time', output_field=DateTimeField())
else: else:
effective_begin = Coalesce('begin_time', 'planned_begin_time', 'expected_begin_time', output_field=DateTimeField()) effective_begin = Coalesce('begin_time', 'planned_begin_time', 'expected_begin_time', output_field=DateTimeField())
effective_end = Coalesce('end_time', 'planned_end_time', 'expected_end_time', output_field=DateTimeField()) effective_end = Coalesce(Subquery(_latest_finished_sq), 'end_time', 'planned_end_time', 'expected_end_time', output_field=DateTimeField())
qs = qs.annotate( qs = qs.annotate(
geom_geojson=AsGeoJSON(Transform(Centroid('geom'), 4326)), geom_geojson=AsGeoJSON(Transform(Centroid('geom'), 4326)),

View file

@ -1789,13 +1789,18 @@ def apply_intervention_filters_from_querydict(qs, params, user=None):
if followed_only == '1' and user and user.is_authenticated: if followed_only == '1' and user and user.is_authenticated:
qs = qs.filter(subscriptions__user=user) qs = qs.filter(subscriptions__user=user)
_latest_finished_sq = InterventionTimeLine.objects.filter(
intervention=OuterRef('pk'),
to_status='finished',
).order_by('-event_time').values('event_time')[:1]
# Date filters (same logic as interventions_geojson but without effective_begin annotation) # Date filters (same logic as interventions_geojson but without effective_begin annotation)
if date_type == 'planned': if date_type == 'planned':
eff_begin = Coalesce('begin_time', 'planned_begin_time', output_field=DateTimeField()) eff_begin = Coalesce('begin_time', 'planned_begin_time', output_field=DateTimeField())
eff_end = Coalesce('end_time', 'planned_end_time', output_field=DateTimeField()) eff_end = Coalesce(Subquery(_latest_finished_sq), 'end_time', 'planned_end_time', output_field=DateTimeField())
else: else:
eff_begin = Coalesce('begin_time', 'planned_begin_time', 'expected_begin_time', output_field=DateTimeField()) eff_begin = Coalesce('begin_time', 'planned_begin_time', 'expected_begin_time', output_field=DateTimeField())
eff_end = Coalesce('end_time', 'planned_end_time', 'expected_end_time', output_field=DateTimeField()) eff_end = Coalesce(Subquery(_latest_finished_sq), 'end_time', 'planned_end_time', 'expected_end_time', output_field=DateTimeField())
if begin_date_min or begin_date_max or end_date_min or end_date_max: if begin_date_min or begin_date_max or end_date_min or end_date_max:
qs = qs.annotate(_eff_begin=eff_begin, _eff_end=eff_end) qs = qs.annotate(_eff_begin=eff_begin, _eff_end=eff_end)

View file

@ -926,10 +926,15 @@ def intervention_detail_mobile(request, intervention_id):
.distinct() .distinct()
) )
# Récupérer les interventions des 3 derniers mois # Récupérer les interventions des 3 derniers mois
_latest_finished_sq = InterventionTimeLine.objects.filter(
intervention=OuterRef('pk'),
to_status='finished',
).order_by('-event_time').values('event_time')[:1]
related_interventions = ( related_interventions = (
Intervention.objects Intervention.objects
.annotate( .annotate(
effective_end = Coalesce( effective_end = Coalesce(
Subquery(_latest_finished_sq),
'end_time', 'end_time',
'planned_end_time', 'planned_end_time',
'expected_end_time', 'expected_end_time',

View file

@ -510,10 +510,14 @@ def linked_interventions_summary(request):
# Interventions récentes terminées (3 derniers mois) # Interventions récentes terminées (3 derniers mois)
three_months_ago = timezone.now() - timedelta(days=90) three_months_ago = timezone.now() - timedelta(days=90)
_latest_finished_sq = InterventionTimeLine.objects.filter(
intervention=OuterRef('pk'),
to_status='finished',
).order_by('-event_time').values('event_time')[:1]
total_recent = base_qs.filter( total_recent = base_qs.filter(
status__in=LINKED_RECENT_STATUSES status__in=LINKED_RECENT_STATUSES
).annotate( ).annotate(
effective_end=Coalesce('end_time', 'planned_end_time', 'expected_end_time', 'planned_begin_time', 'creation_time', output_field=DateTimeField()) effective_end=Coalesce(Subquery(_latest_finished_sq), 'end_time', 'planned_end_time', 'expected_end_time', 'planned_begin_time', 'creation_time', output_field=DateTimeField())
).filter( ).filter(
effective_end__gte=three_months_ago effective_end__gte=three_months_ago
).count() ).count()
@ -522,9 +526,14 @@ def linked_interventions_summary(request):
def intervention_summary_modal(request, intervention_id): def intervention_summary_modal(request, intervention_id):
_latest_finished_sq = InterventionTimeLine.objects.filter(
intervention=OuterRef('pk'),
to_status='finished',
).order_by('-event_time').values('event_time')[:1]
intervention = get_object_or_404( intervention = get_object_or_404(
Intervention.objects.annotate( Intervention.objects.annotate(
effective_end=Coalesce( effective_end=Coalesce(
Subquery(_latest_finished_sq),
'end_time', 'end_time',
'planned_end_time', 'planned_end_time',
'expected_end_time', 'expected_end_time',
@ -612,10 +621,14 @@ def linked_interventions_list(request):
if list_type == 'recent': if list_type == 'recent':
# Interventions récentes terminées (3 derniers mois) # Interventions récentes terminées (3 derniers mois)
three_months_ago = timezone.now() - timedelta(days=90) three_months_ago = timezone.now() - timedelta(days=90)
_latest_finished_sq = InterventionTimeLine.objects.filter(
intervention=OuterRef('pk'),
to_status='finished',
).order_by('-event_time').values('event_time')[:1]
filtered_qs = base_qs.filter( filtered_qs = base_qs.filter(
status__in=LINKED_RECENT_STATUSES status__in=LINKED_RECENT_STATUSES
).annotate( ).annotate(
effective_end=Coalesce('end_time', 'planned_end_time', 'expected_end_time', 'planned_begin_time', 'creation_time', output_field=DateTimeField()) effective_end=Coalesce(Subquery(_latest_finished_sq), 'end_time', 'planned_end_time', 'expected_end_time', 'planned_begin_time', 'creation_time', output_field=DateTimeField())
).filter( ).filter(
effective_end__gte=three_months_ago effective_end__gte=three_months_ago
) )
@ -722,12 +735,16 @@ def linked_interventions_geojson(request):
).values('id', 'code', 'title', 'status', 'lon', 'lat')[:500] ).values('id', 'code', 'title', 'status', 'lon', 'lat')[:500]
# Recent (terminated, within 3 months) # Recent (terminated, within 3 months)
_latest_finished_sq = InterventionTimeLine.objects.filter(
intervention=OuterRef('pk'),
to_status='finished',
).order_by('-event_time').values('event_time')[:1]
recent_qs = base_qs.filter( recent_qs = base_qs.filter(
status__in=LINKED_RECENT_STATUSES, status__in=LINKED_RECENT_STATUSES,
lon__isnull=False, lon__isnull=False,
lat__isnull=False, lat__isnull=False,
).annotate( ).annotate(
effective_end=Coalesce('end_time', 'planned_end_time', 'expected_end_time', 'planned_begin_time', 'creation_time', output_field=DateTimeField()) effective_end=Coalesce(Subquery(_latest_finished_sq), 'end_time', 'planned_end_time', 'expected_end_time', 'planned_begin_time', 'creation_time', output_field=DateTimeField())
).filter(effective_end__gte=three_months_ago).values('id', 'code', 'title', 'status', 'lon', 'lat')[:500] ).filter(effective_end__gte=three_months_ago).values('id', 'code', 'title', 'status', 'lon', 'lat')[:500]
features = [] features = []