authentik.enterprise.lifecycle.tests.test_models

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