authentik.enterprise.lifecycle.tests.test_models

  1import datetime as dt
  2from datetime import timedelta
  3from unittest.mock import patch
  4
  5from django.contrib.contenttypes.models import ContentType
  6from django.test import RequestFactory, TestCase
  7from django.utils import timezone
  8
  9from authentik.core.models import Application, Group
 10from authentik.core.tests.utils import create_test_user
 11from authentik.enterprise.lifecycle.models import (
 12    LifecycleIteration,
 13    LifecycleRule,
 14    Review,
 15    ReviewState,
 16)
 17from authentik.events.models import (
 18    Event,
 19    EventAction,
 20    NotificationSeverity,
 21    NotificationTransport,
 22)
 23from authentik.lib.generators import generate_id
 24from authentik.rbac.models import Role
 25
 26
 27class TestLifecycleModels(TestCase):
 28
 29    def setUp(self):
 30        self.factory = RequestFactory()
 31
 32    def _get_request(self):
 33        return self.factory.get("/")
 34
 35    def _create_object(self, model):
 36        if model is Application:
 37            return Application.objects.create(name=generate_id(), slug=generate_id())
 38        if model is Role:
 39            return Role.objects.create(name=generate_id())
 40        if model is Group:
 41            return Group.objects.create(name=generate_id())
 42        raise AssertionError(f"Unsupported model {model}")
 43
 44    def _create_rule_for_object(self, obj, **kwargs) -> LifecycleRule:
 45        content_type = ContentType.objects.get_for_model(obj)
 46        return LifecycleRule.objects.create(
 47            name=generate_id(),
 48            content_type=content_type,
 49            object_id=str(obj.pk),
 50            **kwargs,
 51        )
 52
 53    def _create_rule_for_type(self, model, **kwargs) -> LifecycleRule:
 54        content_type = ContentType.objects.get_for_model(model)
 55        return LifecycleRule.objects.create(
 56            name=generate_id(),
 57            content_type=content_type,
 58            object_id=None,
 59            **kwargs,
 60        )
 61
 62    def test_iteration_start_supported_objects(self):
 63        """Ensure iterations are automatically started for applications, roles, and groups."""
 64        for model in (Application, Role, Group):
 65            with self.subTest(model=model.__name__):
 66                obj = self._create_object(model)
 67                content_type = ContentType.objects.get_for_model(obj)
 68
 69                before_events = Event.objects.filter(action=EventAction.REVIEW_INITIATED).count()
 70
 71                rule = self._create_rule_for_object(obj)
 72
 73                # Verify iteration was created automatically
 74                iteration = LifecycleIteration.objects.get(
 75                    content_type=content_type, object_id=str(obj.pk), rule=rule
 76                )
 77                self.assertEqual(iteration.state, ReviewState.PENDING)
 78                self.assertEqual(iteration.object, obj)
 79                self.assertEqual(iteration.rule, rule)
 80                self.assertEqual(
 81                    Event.objects.filter(action=EventAction.REVIEW_INITIATED).count(),
 82                    before_events + 1,
 83                )
 84
 85    def test_review_requires_all_explicit_reviewers(self):
 86        obj = Group.objects.create(name=generate_id())
 87        rule = self._create_rule_for_object(obj)
 88        reviewer_one = create_test_user()
 89        reviewer_two = create_test_user()
 90        rule.reviewers.add(reviewer_one, reviewer_two)
 91
 92        content_type = ContentType.objects.get_for_model(obj)
 93
 94        iteration = LifecycleIteration.objects.get(
 95            content_type=content_type, object_id=str(obj.pk), rule=rule
 96        )
 97        request = self._get_request()
 98
 99        Review.objects.create(iteration=iteration, reviewer=reviewer_one)
100        iteration.on_review(request)
101        iteration.refresh_from_db()
102        self.assertEqual(iteration.state, ReviewState.PENDING)
103
104        Review.objects.create(iteration=iteration, reviewer=reviewer_two)
105        iteration.on_review(request)
106        iteration.refresh_from_db()
107        self.assertEqual(iteration.state, ReviewState.REVIEWED)
108        self.assertTrue(Event.objects.filter(action=EventAction.REVIEW_COMPLETED).exists())
109
110    def test_review_min_reviewers_from_groups(self):
111        """Group-based reviews complete once the minimum number of reviewers review."""
112        obj = Application.objects.create(name=generate_id(), slug=generate_id())
113        rule = self._create_rule_for_object(obj, min_reviewers=2)
114
115        reviewer_group = Group.objects.create(name=generate_id())
116        reviewer_one = create_test_user()
117        reviewer_two = create_test_user()
118        reviewer_group.users.add(reviewer_one, reviewer_two)
119        rule.reviewer_groups.add(reviewer_group)
120
121        content_type = ContentType.objects.get_for_model(obj)
122
123        iteration = LifecycleIteration.objects.get(
124            content_type=content_type, object_id=str(obj.pk), rule=rule
125        )
126        request = self._get_request()
127
128        Review.objects.create(iteration=iteration, reviewer=reviewer_one)
129        iteration.on_review(request)
130        iteration.refresh_from_db()
131        self.assertEqual(iteration.state, ReviewState.PENDING)
132
133        Review.objects.create(iteration=iteration, reviewer=reviewer_two)
134        iteration.on_review(request)
135        iteration.refresh_from_db()
136        self.assertEqual(iteration.state, ReviewState.REVIEWED)
137
138    def test_review_explicit_and_group_reviewers(self):
139        """Reviews require both explicit reviewers AND min_reviewers from groups."""
140        obj = Application.objects.create(name=generate_id(), slug=generate_id())
141        rule = self._create_rule_for_object(obj, min_reviewers=1)
142
143        reviewer_group = Group.objects.create(name=generate_id())
144        group_member = create_test_user()
145        reviewer_group.users.add(group_member)
146        rule.reviewer_groups.add(reviewer_group)
147
148        explicit_reviewer = create_test_user()
149        rule.reviewers.add(explicit_reviewer)
150
151        content_type = ContentType.objects.get_for_model(obj)
152
153        iteration = LifecycleIteration.objects.get(
154            content_type=content_type, object_id=str(obj.pk), rule=rule
155        )
156        request = self._get_request()
157
158        # Only group member reviews - not satisfied (explicit reviewer missing)
159        Review.objects.create(iteration=iteration, reviewer=group_member)
160        iteration.on_review(request)
161        iteration.refresh_from_db()
162        self.assertEqual(iteration.state, ReviewState.PENDING)
163
164        # Explicit reviewer reviews - now satisfied
165        Review.objects.create(iteration=iteration, reviewer=explicit_reviewer)
166        iteration.on_review(request)
167        iteration.refresh_from_db()
168        self.assertEqual(iteration.state, ReviewState.REVIEWED)
169
170    def test_review_min_reviewers_per_group(self):
171        obj = Application.objects.create(name=generate_id(), slug=generate_id())
172        rule = self._create_rule_for_object(obj, min_reviewers=1, min_reviewers_is_per_group=True)
173
174        group_one = Group.objects.create(name=generate_id())
175        group_two = Group.objects.create(name=generate_id())
176        member_group_one = create_test_user()
177        member_group_two = create_test_user()
178        group_one.users.add(member_group_one)
179        group_two.users.add(member_group_two)
180        rule.reviewer_groups.add(group_one, group_two)
181
182        content_type = ContentType.objects.get_for_model(obj)
183
184        iteration = LifecycleIteration.objects.get(
185            content_type=content_type, object_id=str(obj.pk), rule=rule
186        )
187        request = self._get_request()
188
189        # Only member from group_one reviews - not satisfied (need member from each group)
190        Review.objects.create(iteration=iteration, reviewer=member_group_one)
191        iteration.on_review(request)
192        iteration.refresh_from_db()
193        self.assertEqual(iteration.state, ReviewState.PENDING)
194
195        # Member from group_two reviews - now satisfied
196        Review.objects.create(iteration=iteration, reviewer=member_group_two)
197        iteration.on_review(request)
198        iteration.refresh_from_db()
199        self.assertEqual(iteration.state, ReviewState.REVIEWED)
200
201    def test_review_reviewers_from_child_groups(self):
202        obj = Application.objects.create(name=generate_id(), slug=generate_id())
203        rule = self._create_rule_for_object(obj, min_reviewers=1)
204
205        parent_group = Group.objects.create(name=generate_id())
206        child_group = Group.objects.create(name=generate_id())
207        child_group.parents.add(parent_group)
208
209        child_member = create_test_user()
210        child_group.users.add(child_member)
211
212        rule.reviewer_groups.add(parent_group)
213
214        content_type = ContentType.objects.get_for_model(obj)
215
216        iteration = LifecycleIteration.objects.get(
217            content_type=content_type, object_id=str(obj.pk), rule=rule
218        )
219        request = self._get_request()
220
221        # Child group member should be able to review
222        self.assertTrue(iteration.user_can_review(child_member))
223
224        Review.objects.create(iteration=iteration, reviewer=child_member)
225        iteration.on_review(request)
226        iteration.refresh_from_db()
227        self.assertEqual(iteration.state, ReviewState.REVIEWED)
228
229    def test_review_reviewers_from_nested_child_groups(self):
230        obj = Application.objects.create(name=generate_id(), slug=generate_id())
231        rule = self._create_rule_for_object(obj, min_reviewers=2)
232
233        grandparent = Group.objects.create(name=generate_id())
234        parent = Group.objects.create(name=generate_id())
235        child = Group.objects.create(name=generate_id())
236        parent.parents.add(grandparent)
237        child.parents.add(parent)
238
239        parent_member = create_test_user()
240        child_member = create_test_user()
241        parent.users.add(parent_member)
242        child.users.add(child_member)
243
244        rule.reviewer_groups.add(grandparent)
245
246        content_type = ContentType.objects.get_for_model(obj)
247
248        iteration = LifecycleIteration.objects.get(
249            content_type=content_type, object_id=str(obj.pk), rule=rule
250        )
251        request = self._get_request()
252
253        # Both nested members should be able to review
254        self.assertTrue(iteration.user_can_review(parent_member))
255        self.assertTrue(iteration.user_can_review(child_member))
256
257        Review.objects.create(iteration=iteration, reviewer=parent_member)
258        iteration.on_review(request)
259        iteration.refresh_from_db()
260        self.assertEqual(iteration.state, ReviewState.PENDING)
261
262        Review.objects.create(iteration=iteration, reviewer=child_member)
263        iteration.on_review(request)
264        iteration.refresh_from_db()
265        self.assertEqual(iteration.state, ReviewState.REVIEWED)
266
267    def test_notify_reviewers_send_once(self):
268        obj = Group.objects.create(name=generate_id())
269        rule = self._create_rule_for_object(obj)
270
271        reviewer_one = create_test_user()
272        reviewer_two = create_test_user()
273        rule.reviewers.add(reviewer_one, reviewer_two)
274
275        transport_once = NotificationTransport.objects.create(
276            name=generate_id(),
277            send_once=True,
278        )
279        transport_all = NotificationTransport.objects.create(
280            name=generate_id(),
281            send_once=False,
282        )
283        rule.notification_transports.add(transport_once, transport_all)
284
285        event = Event.new(EventAction.REVIEW_INITIATED, target=obj)
286        event.save()
287
288        with patch(
289            "authentik.enterprise.lifecycle.tasks.send_notification.send_with_options"
290        ) as send_with_options:
291            rule.notify_reviewers(event, NotificationSeverity.NOTICE)
292
293            reviewer_pks = {reviewer_one.pk, reviewer_two.pk}
294            self.assertEqual(send_with_options.call_count, len(reviewer_pks) + 1)
295
296            calls = [call.kwargs["args"] for call in send_with_options.call_args_list]
297            once_calls = [args for args in calls if args[0] == transport_once.pk]
298            all_calls = [args for args in calls if args[0] == transport_all.pk]
299
300            self.assertEqual(len(once_calls), 1)
301            self.assertEqual(len(all_calls), len(reviewer_pks))
302            self.assertIn(once_calls[0][2], reviewer_pks)
303            self.assertEqual({args[2] for args in all_calls}, reviewer_pks)
304
305    def test_apply_marks_overdue_and_opens_due_reviews(self):
306        app_one = Application.objects.create(name=generate_id(), slug=generate_id())
307        app_two = Application.objects.create(name=generate_id(), slug=generate_id())
308        content_type = ContentType.objects.get_for_model(Application)
309
310        rule_overdue = LifecycleRule.objects.create(
311            name=generate_id(),
312            content_type=content_type,
313            object_id=str(app_one.pk),
314            interval="days=365",
315            grace_period="days=10",
316        )
317
318        # Get the automatically created iteration and backdate it past the grace period
319        iteration = LifecycleIteration.objects.get(
320            content_type=content_type, object_id=str(app_one.pk), rule=rule_overdue
321        )
322        LifecycleIteration.objects.filter(pk=iteration.pk).update(
323            opened_on=(timezone.now() - timedelta(days=20))
324        )
325
326        # Apply again to trigger overdue logic
327        rule_overdue.apply()
328        iteration.refresh_from_db()
329        self.assertEqual(iteration.state, ReviewState.OVERDUE)
330        self.assertEqual(
331            LifecycleIteration.objects.filter(
332                content_type=content_type, object_id=str(app_one.pk)
333            ).count(),
334            1,
335        )
336
337        LifecycleRule.objects.create(
338            name=generate_id(),
339            content_type=content_type,
340            object_id=str(app_two.pk),
341            interval="days=30",
342            grace_period="days=10",
343        )
344        self.assertEqual(
345            LifecycleIteration.objects.filter(
346                content_type=content_type, object_id=str(app_two.pk)
347            ).count(),
348            1,
349        )
350        new_iteration = LifecycleIteration.objects.get(
351            content_type=content_type, object_id=str(app_two.pk)
352        )
353        self.assertEqual(new_iteration.state, ReviewState.PENDING)
354
355    def test_apply_idempotent(self):
356        app_due = Application.objects.create(name=generate_id(), slug=generate_id())
357        app_overdue = Application.objects.create(name=generate_id(), slug=generate_id())
358        content_type = ContentType.objects.get_for_model(Application)
359
360        initiated_before = Event.objects.filter(action=EventAction.REVIEW_INITIATED).count()
361        overdue_before = Event.objects.filter(action=EventAction.REVIEW_OVERDUE).count()
362
363        rule_due = LifecycleRule.objects.create(
364            name=generate_id(),
365            content_type=content_type,
366            object_id=str(app_due.pk),
367            interval="days=30",
368            grace_period="days=30",
369        )
370        reviewer = create_test_user()
371        rule_due.reviewers.add(reviewer)
372        transport = NotificationTransport.objects.create(name=generate_id())
373        rule_due.notification_transports.add(transport)
374
375        rule_overdue = LifecycleRule.objects.create(
376            name=generate_id(),
377            content_type=content_type,
378            object_id=str(app_overdue.pk),
379            interval="days=365",
380            grace_period="days=10",
381        )
382
383        overdue_iteration = LifecycleIteration.objects.get(
384            content_type=content_type, object_id=str(app_overdue.pk), rule=rule_overdue
385        )
386        LifecycleIteration.objects.filter(pk=overdue_iteration.pk).update(
387            opened_on=(timezone.now() - timedelta(days=20))
388        )
389
390        # Apply overdue rule to mark iteration as overdue
391        rule_overdue.apply()
392
393        due_iteration = LifecycleIteration.objects.get(
394            content_type=content_type, object_id=str(app_due.pk)
395        )
396        overdue_iteration.refresh_from_db()
397        self.assertEqual(due_iteration.state, ReviewState.PENDING)
398        self.assertEqual(overdue_iteration.state, ReviewState.OVERDUE)
399
400        initiated_after_first = Event.objects.filter(action=EventAction.REVIEW_INITIATED).count()
401        overdue_after_first = Event.objects.filter(action=EventAction.REVIEW_OVERDUE).count()
402        # Both rules created iterations on save
403        self.assertEqual(initiated_after_first, initiated_before + 2)
404        self.assertEqual(overdue_after_first, overdue_before + 1)
405
406        # Apply again - should be idempotent
407        rule_due.apply()
408        rule_overdue.apply()
409
410        due_iteration.refresh_from_db()
411        overdue_iteration.refresh_from_db()
412        self.assertEqual(due_iteration.state, ReviewState.PENDING)
413        self.assertEqual(overdue_iteration.state, ReviewState.OVERDUE)
414        self.assertEqual(
415            Event.objects.filter(action=EventAction.REVIEW_INITIATED).count(),
416            initiated_after_first,
417        )
418        self.assertEqual(
419            Event.objects.filter(action=EventAction.REVIEW_OVERDUE).count(),
420            overdue_after_first,
421        )
422
423    def test_rule_matches_entire_type(self):
424        """A rule with object_id=None matches all objects of that type."""
425        app_one = Application.objects.create(name=generate_id(), slug=generate_id())
426        app_two = Application.objects.create(name=generate_id(), slug=generate_id())
427        content_type = ContentType.objects.get_for_model(Application)
428
429        rule = LifecycleRule.objects.create(
430            name=generate_id(),
431            content_type=content_type,
432            object_id=None,
433            interval="days=30",
434            grace_period="days=10",
435        )
436
437        objects = list(rule.get_objects())
438        self.assertIn(app_one, objects)
439        self.assertIn(app_two, objects)
440
441    def test_rule_type_excludes_objects_with_specific_rules(self):
442        app_with_rule = Application.objects.create(name=generate_id(), slug=generate_id())
443        app_without_rule = Application.objects.create(name=generate_id(), slug=generate_id())
444        content_type = ContentType.objects.get_for_model(Application)
445
446        # Create a specific rule for app_with_rule
447        LifecycleRule.objects.create(
448            name=generate_id(),
449            content_type=content_type,
450            object_id=str(app_with_rule.pk),
451            interval="days=30",
452        )
453
454        # Create a type-level rule
455        type_rule = LifecycleRule.objects.create(
456            name=generate_id(),
457            content_type=content_type,
458            object_id=None,
459            interval="days=60",
460        )
461
462        objects = list(type_rule.get_objects())
463        self.assertNotIn(app_with_rule, objects)
464        self.assertIn(app_without_rule, objects)
465
466    def test_rule_type_apply_creates_iterations_for_all_objects(self):
467        app_one = Application.objects.create(name=generate_id(), slug=generate_id())
468        app_two = Application.objects.create(name=generate_id(), slug=generate_id())
469        content_type = ContentType.objects.get_for_model(Application)
470
471        LifecycleRule.objects.create(
472            name=generate_id(),
473            content_type=content_type,
474            object_id=None,
475            interval="days=30",
476            grace_period="days=10",
477        )
478
479        self.assertTrue(
480            LifecycleIteration.objects.filter(
481                content_type=content_type, object_id=str(app_one.pk)
482            ).exists()
483        )
484        self.assertTrue(
485            LifecycleIteration.objects.filter(
486                content_type=content_type, object_id=str(app_two.pk)
487            ).exists()
488        )
489
490    def test_delete_rule_cancels_open_iterations(self):
491        obj = Application.objects.create(name=generate_id(), slug=generate_id())
492
493        rule = self._create_rule_for_object(obj)
494        content_type = ContentType.objects.get_for_model(obj)
495
496        pending_iteration = LifecycleIteration.objects.get(
497            content_type=content_type, object_id=str(obj.pk), rule=rule
498        )
499        self.assertEqual(pending_iteration.state, ReviewState.PENDING)
500
501        overdue_iteration = LifecycleIteration.objects.create(
502            content_type=content_type,
503            object_id=str(obj.pk),
504            rule=rule,
505            state=ReviewState.OVERDUE,
506        )
507        reviewed_iteration = LifecycleIteration.objects.create(
508            content_type=content_type,
509            object_id=str(obj.pk),
510            rule=rule,
511            state=ReviewState.REVIEWED,
512        )
513
514        rule.delete()
515
516        pending_iteration.refresh_from_db()
517        overdue_iteration.refresh_from_db()
518        reviewed_iteration.refresh_from_db()
519
520        self.assertEqual(pending_iteration.state, ReviewState.CANCELED)
521        self.assertEqual(overdue_iteration.state, ReviewState.CANCELED)
522        self.assertEqual(reviewed_iteration.state, ReviewState.REVIEWED)  # Not affected
523
524    def test_update_rule_target_cancels_stale_iterations(self):
525        app_one = Application.objects.create(name=generate_id(), slug=generate_id())
526        app_two = Application.objects.create(name=generate_id(), slug=generate_id())
527        content_type = ContentType.objects.get_for_model(Application)
528
529        rule = LifecycleRule.objects.create(
530            name=generate_id(),
531            content_type=content_type,
532            object_id=str(app_one.pk),
533            interval="days=30",
534        )
535
536        iteration_for_app_one = LifecycleIteration.objects.get(
537            content_type=content_type, object_id=str(app_one.pk), rule=rule
538        )
539        self.assertEqual(iteration_for_app_one.state, ReviewState.PENDING)
540
541        # Change rule target to app_two - save() triggers apply() which cancels stale iterations
542        rule.object_id = str(app_two.pk)
543        rule.save()
544
545        iteration_for_app_one.refresh_from_db()
546        self.assertEqual(iteration_for_app_one.state, ReviewState.CANCELED)
547
548    def test_update_rule_content_type_cancels_stale_iterations(self):
549        app = Application.objects.create(name=generate_id(), slug=generate_id())
550        group = Group.objects.create(name=generate_id())
551        app_content_type = ContentType.objects.get_for_model(Application)
552        group_content_type = ContentType.objects.get_for_model(Group)
553
554        # Creating rule triggers automatic apply() which creates a iteration for app
555        rule = LifecycleRule.objects.create(
556            name=generate_id(),
557            content_type=app_content_type,
558            object_id=str(app.pk),
559            interval="days=30",
560        )
561
562        iteration = LifecycleIteration.objects.get(
563            content_type=app_content_type, object_id=str(app.pk), rule=rule
564        )
565        self.assertEqual(iteration.state, ReviewState.PENDING)
566
567        # Change content type to Group - save() triggers apply() which cancels stale iterations
568        rule.content_type = group_content_type
569        rule.object_id = str(group.pk)
570        rule.save()
571
572        iteration.refresh_from_db()
573        self.assertEqual(iteration.state, ReviewState.CANCELED)
574
575    def test_user_can_review_checks_group_hierarchy(self):
576        obj = Application.objects.create(name=generate_id(), slug=generate_id())
577        rule = self._create_rule_for_object(obj)
578
579        parent_group = Group.objects.create(name=generate_id())
580        child_group = Group.objects.create(name=generate_id())
581        child_group.parents.add(parent_group)
582
583        parent_member = create_test_user()
584        child_member = create_test_user()
585        non_member = create_test_user()
586        parent_group.users.add(parent_member)
587        child_group.users.add(child_member)
588
589        rule.reviewer_groups.add(parent_group)
590
591        content_type = ContentType.objects.get_for_model(obj)
592        # iteration is created automatically when rule is saved
593        iteration = LifecycleIteration.objects.get(
594            content_type=content_type, object_id=str(obj.pk), rule=rule
595        )
596
597        self.assertTrue(iteration.user_can_review(parent_member))
598        self.assertTrue(iteration.user_can_review(child_member))
599        self.assertFalse(iteration.user_can_review(non_member))
600
601    def test_user_cannot_review_twice(self):
602        obj = Application.objects.create(name=generate_id(), slug=generate_id())
603        rule = self._create_rule_for_object(obj)
604        reviewer = create_test_user()
605        rule.reviewers.add(reviewer)
606
607        content_type = ContentType.objects.get_for_model(obj)
608        # iteration is created automatically when rule is saved
609        iteration = LifecycleIteration.objects.get(
610            content_type=content_type, object_id=str(obj.pk), rule=rule
611        )
612
613        self.assertTrue(iteration.user_can_review(reviewer))
614
615        Review.objects.create(iteration=iteration, reviewer=reviewer)
616
617        self.assertFalse(iteration.user_can_review(reviewer))
618
619    def test_user_cannot_review_completed_iteration(self):
620        obj = Application.objects.create(name=generate_id(), slug=generate_id())
621        rule = self._create_rule_for_object(obj)
622        reviewer = create_test_user()
623        rule.reviewers.add(reviewer)
624
625        content_type = ContentType.objects.get_for_model(obj)
626
627        # Get the automatically created pending iteration and test with different states
628        iteration = LifecycleIteration.objects.get(
629            content_type=content_type, object_id=str(obj.pk), rule=rule
630        )
631
632        for state in (ReviewState.REVIEWED, ReviewState.CANCELED):
633            iteration.state = state
634            iteration.save()
635            self.assertFalse(iteration.user_can_review(reviewer))
636
637    def test_get_reviewers_includes_child_group_members(self):
638        obj = Application.objects.create(name=generate_id(), slug=generate_id())
639        rule = self._create_rule_for_object(obj)
640
641        parent_group = Group.objects.create(name=generate_id())
642        child_group = Group.objects.create(name=generate_id())
643        child_group.parents.add(parent_group)
644
645        parent_member = create_test_user()
646        child_member = create_test_user()
647        parent_group.users.add(parent_member)
648        child_group.users.add(child_member)
649
650        rule.reviewer_groups.add(parent_group)
651
652        reviewers = list(rule.get_reviewers())
653        self.assertIn(parent_member, reviewers)
654        self.assertIn(child_member, reviewers)
655
656    def test_get_reviewers_includes_explicit_reviewers(self):
657        obj = Application.objects.create(name=generate_id(), slug=generate_id())
658        rule = self._create_rule_for_object(obj)
659
660        explicit_reviewer = create_test_user()
661        rule.reviewers.add(explicit_reviewer)
662
663        group = Group.objects.create(name=generate_id())
664        group_member = create_test_user()
665        group.users.add(group_member)
666        rule.reviewer_groups.add(group)
667
668        reviewers = list(rule.get_reviewers())
669        self.assertIn(explicit_reviewer, reviewers)
670        self.assertIn(group_member, reviewers)
671
672
673class TestLifecycleDateBoundaries(TestCase):
674    """Verify that start_of_day normalization ensures correct overdue/due
675    detection regardless of exact task execution time within a day.
676
677    The daily task may run at any point during the day. The start_of_day
678    normalization in _get_newly_overdue_iterations and _get_newly_due_objects
679    ensures that the boundary is always at midnight, so millisecond variations
680    in task execution time do not affect results."""
681
682    def _create_rule_and_iteration(self, grace_period="days=1", interval="days=365"):
683        app = Application.objects.create(name=generate_id(), slug=generate_id())
684        content_type = ContentType.objects.get_for_model(Application)
685        rule = LifecycleRule.objects.create(
686            name=generate_id(),
687            content_type=content_type,
688            object_id=str(app.pk),
689            interval=interval,
690            grace_period=grace_period,
691        )
692        iteration = LifecycleIteration.objects.get(
693            content_type=content_type, object_id=str(app.pk), rule=rule
694        )
695        return app, rule, iteration
696
697    def test_overdue_iteration_opened_yesterday(self):
698        """grace_period=1 day: iteration opened yesterday at any time is overdue today."""
699        _, rule, iteration = self._create_rule_and_iteration(grace_period="days=1")
700        fixed_now = dt.datetime(2025, 6, 15, 14, 30, 0, tzinfo=dt.UTC)
701        for opened_on in [
702            dt.datetime(2025, 6, 14, 0, 0, 0, tzinfo=dt.UTC),
703            dt.datetime(2025, 6, 14, 12, 0, 0, tzinfo=dt.UTC),
704            dt.datetime(2025, 6, 14, 23, 59, 59, 999999, tzinfo=dt.UTC),
705        ]:
706            with self.subTest(opened_on=opened_on):
707                LifecycleIteration.objects.filter(pk=iteration.pk).update(
708                    opened_on=opened_on, state=ReviewState.PENDING
709                )
710                with patch("django.utils.timezone.now", return_value=fixed_now):
711                    self.assertIn(iteration, list(rule._get_newly_overdue_iterations()))
712
713    def test_not_overdue_iteration_opened_today(self):
714        """grace_period=1 day: iteration opened today at any time is NOT overdue."""
715        _, rule, iteration = self._create_rule_and_iteration(grace_period="days=1")
716        fixed_now = dt.datetime(2025, 6, 15, 14, 30, 0, tzinfo=dt.UTC)
717        for opened_on in [
718            dt.datetime(2025, 6, 15, 0, 0, 0, tzinfo=dt.UTC),
719            dt.datetime(2025, 6, 15, 14, 30, 0, tzinfo=dt.UTC),
720            dt.datetime(2025, 6, 15, 23, 59, 59, 999999, tzinfo=dt.UTC),
721        ]:
722            with self.subTest(opened_on=opened_on):
723                LifecycleIteration.objects.filter(pk=iteration.pk).update(
724                    opened_on=opened_on, state=ReviewState.PENDING
725                )
726                with patch("django.utils.timezone.now", return_value=fixed_now):
727                    self.assertNotIn(iteration, list(rule._get_newly_overdue_iterations()))
728
729    def test_overdue_independent_of_task_execution_time(self):
730        """Overdue detection gives the same result whether the task runs at 00:00:01 or 23:59:59."""
731        _, rule, iteration = self._create_rule_and_iteration(grace_period="days=1")
732        opened_on = dt.datetime(2025, 6, 14, 18, 0, 0, tzinfo=dt.UTC)
733        LifecycleIteration.objects.filter(pk=iteration.pk).update(
734            opened_on=opened_on, state=ReviewState.PENDING
735        )
736        for task_time in [
737            dt.datetime(2025, 6, 15, 0, 0, 1, tzinfo=dt.UTC),
738            dt.datetime(2025, 6, 15, 12, 0, 0, tzinfo=dt.UTC),
739            dt.datetime(2025, 6, 15, 23, 59, 59, tzinfo=dt.UTC),
740        ]:
741            with self.subTest(task_time=task_time):
742                with patch("django.utils.timezone.now", return_value=task_time):
743                    self.assertIn(iteration, list(rule._get_newly_overdue_iterations()))
744
745    def test_overdue_boundary_multi_day_grace_period(self):
746        """grace_period=30 days: overdue after 30 full days, not after 29."""
747        _, rule, iteration = self._create_rule_and_iteration(grace_period="days=30")
748        fixed_now = dt.datetime(2025, 6, 15, 14, 30, 0, tzinfo=dt.UTC)
749
750        # Opened 30 days ago (May 16), should go overdue
751        LifecycleIteration.objects.filter(pk=iteration.pk).update(
752            opened_on=dt.datetime(2025, 5, 16, 12, 0, 0, tzinfo=dt.UTC),
753            state=ReviewState.PENDING,
754        )
755        with patch("django.utils.timezone.now", return_value=fixed_now):
756            self.assertIn(iteration, list(rule._get_newly_overdue_iterations()))
757
758        # Opened 29 days ago (May 17), should NOT go overdue
759        LifecycleIteration.objects.filter(pk=iteration.pk).update(
760            opened_on=dt.datetime(2025, 5, 17, 12, 0, 0, tzinfo=dt.UTC),
761            state=ReviewState.PENDING,
762        )
763        with patch("django.utils.timezone.now", return_value=fixed_now):
764            self.assertNotIn(iteration, list(rule._get_newly_overdue_iterations()))
765
766    def test_due_object_iteration_opened_yesterday(self):
767        """interval=1 day: object with iteration opened yesterday is due for a new review."""
768        app, rule, iteration = self._create_rule_and_iteration(interval="days=1")
769        fixed_now = dt.datetime(2025, 6, 15, 14, 30, 0, tzinfo=dt.UTC)
770        for opened_on in [
771            dt.datetime(2025, 6, 14, 0, 0, 0, tzinfo=dt.UTC),
772            dt.datetime(2025, 6, 14, 12, 0, 0, tzinfo=dt.UTC),
773            dt.datetime(2025, 6, 14, 23, 59, 59, 999999, tzinfo=dt.UTC),
774        ]:
775            with self.subTest(opened_on=opened_on):
776                LifecycleIteration.objects.filter(pk=iteration.pk).update(opened_on=opened_on)
777                with patch("django.utils.timezone.now", return_value=fixed_now):
778                    self.assertIn(app, list(rule._get_newly_due_objects()))
779
780    def test_not_due_object_iteration_opened_today(self):
781        """interval=1 day: object with iteration opened today is NOT due."""
782        app, rule, iteration = self._create_rule_and_iteration(interval="days=1")
783        fixed_now = dt.datetime(2025, 6, 15, 14, 30, 0, tzinfo=dt.UTC)
784        for opened_on in [
785            dt.datetime(2025, 6, 15, 0, 0, 0, tzinfo=dt.UTC),
786            dt.datetime(2025, 6, 15, 14, 30, 0, tzinfo=dt.UTC),
787            dt.datetime(2025, 6, 15, 23, 59, 59, 999999, tzinfo=dt.UTC),
788        ]:
789            with self.subTest(opened_on=opened_on):
790                LifecycleIteration.objects.filter(pk=iteration.pk).update(opened_on=opened_on)
791                with patch("django.utils.timezone.now", return_value=fixed_now):
792                    self.assertNotIn(app, list(rule._get_newly_due_objects()))
793
794    def test_due_independent_of_task_execution_time(self):
795        """Due detection gives the same result whether the task runs at 00:00:01 or 23:59:59."""
796        app, rule, iteration = self._create_rule_and_iteration(interval="days=1")
797        opened_on = dt.datetime(2025, 6, 14, 18, 0, 0, tzinfo=dt.UTC)
798        LifecycleIteration.objects.filter(pk=iteration.pk).update(opened_on=opened_on)
799        for task_time in [
800            dt.datetime(2025, 6, 15, 0, 0, 1, tzinfo=dt.UTC),
801            dt.datetime(2025, 6, 15, 12, 0, 0, tzinfo=dt.UTC),
802            dt.datetime(2025, 6, 15, 23, 59, 59, tzinfo=dt.UTC),
803        ]:
804            with self.subTest(task_time=task_time):
805                with patch("django.utils.timezone.now", return_value=task_time):
806                    self.assertIn(app, list(rule._get_newly_due_objects()))
807
808    def test_due_boundary_multi_day_interval(self):
809        """interval=30 days: due after 30 full days, not after 29."""
810        app, rule, iteration = self._create_rule_and_iteration(interval="days=30")
811        fixed_now = dt.datetime(2025, 6, 15, 14, 30, 0, tzinfo=dt.UTC)
812
813        # Previous review opened 30 days ago (May 16), review is due for the object
814        LifecycleIteration.objects.filter(pk=iteration.pk).update(
815            opened_on=dt.datetime(2025, 5, 16, 12, 0, 0, tzinfo=dt.UTC)
816        )
817        with patch("django.utils.timezone.now", return_value=fixed_now):
818            self.assertIn(app, list(rule._get_newly_due_objects()))
819
820        # Previous review opened 29 days ago (May 17), new review is NOT due
821        LifecycleIteration.objects.filter(pk=iteration.pk).update(
822            opened_on=dt.datetime(2025, 5, 17, 12, 0, 0, tzinfo=dt.UTC)
823        )
824        with patch("django.utils.timezone.now", return_value=fixed_now):
825            self.assertNotIn(app, list(rule._get_newly_due_objects()))
826
827    def test_apply_overdue_at_boundary(self):
828        """apply() marks iteration overdue when grace period just expired,
829        regardless of what time the daily task runs."""
830        _, rule, iteration = self._create_rule_and_iteration(
831            grace_period="days=1", interval="days=365"
832        )
833        opened_on = dt.datetime(2025, 6, 14, 20, 0, 0, tzinfo=dt.UTC)
834        for task_time in [
835            dt.datetime(2025, 6, 15, 0, 0, 1, tzinfo=dt.UTC),
836            dt.datetime(2025, 6, 15, 23, 59, 59, tzinfo=dt.UTC),
837        ]:
838            with self.subTest(task_time=task_time):
839                LifecycleIteration.objects.filter(pk=iteration.pk).update(
840                    opened_on=opened_on, state=ReviewState.PENDING
841                )
842                with patch("django.utils.timezone.now", return_value=task_time):
843                    rule.apply()
844                iteration.refresh_from_db()
845                self.assertEqual(iteration.state, ReviewState.OVERDUE)