stages/consent: add unittests for new modes

passbook/flows/planner.py

@@ -52,6 +52,7 @@ class FlowPlan:
         stage = self.stages[0]
         marker = self.markers[0]
 
+        if marker.__class__ is not StageMarker:
             LOGGER.debug("f(plan_inst): stage has marker", stage=stage, marker=marker)
         marked_stage = marker.process(self, stage)
         if not marked_stage:

passbook/stages/consent/models.py

@@ -1,7 +1,7 @@
 """passbook consent stage"""
-from django.db import models
 from typing import Type
 
+from django.db import models
 from django.forms import ModelForm
 from django.utils.translation import gettext_lazy as _
 from django.views import View

passbook/stages/consent/tests.py

@@ -1,14 +1,17 @@
 """consent tests"""
+from time import sleep
+
 from django.shortcuts import reverse
 from django.test import Client, TestCase
 from django.utils.encoding import force_text
 
-from passbook.core.models import User
+from passbook.core.models import Application, User
+from passbook.core.tasks import clean_expired_models
 from passbook.flows.markers import StageMarker
 from passbook.flows.models import Flow, FlowDesignation, FlowStageBinding
-from passbook.flows.planner import FlowPlan
+from passbook.flows.planner import PLAN_CONTEXT_APPLICATION, FlowPlan
 from passbook.flows.views import SESSION_KEY_PLAN
-from passbook.stages.consent.models import ConsentStage
+from passbook.stages.consent.models import ConsentMode, ConsentStage, UserConsent
 
 
 class TestConsentStage(TestCase):
@@ -19,28 +22,29 @@ class TestConsentStage(TestCase):
         self.user = User.objects.create_user(
             username="unittest", email="test@beryju.org"
         )
+        self.application = Application.objects.create(
+            name="test-application", slug="test-application",
+        )
         self.client = Client()
 
-        self.flow = Flow.objects.create(
+    def test_always_required(self):
+        """Test always required consent"""
+        flow = Flow.objects.create(
             name="test-consent",
             slug="test-consent",
             designation=FlowDesignation.AUTHENTICATION,
         )
-        self.stage = ConsentStage.objects.create(name="consent",)
+        stage = ConsentStage.objects.create(
-        FlowStageBinding.objects.create(target=self.flow, stage=self.stage, order=2)
+            name="consent", mode=ConsentMode.ALWAYS_REQUIRE
-
-    def test_valid(self):
-        """Test valid consent"""
-        plan = FlowPlan(
-            flow_pk=self.flow.pk.hex, stages=[self.stage], markers=[StageMarker()]
         )
+        FlowStageBinding.objects.create(target=flow, stage=stage, order=2)
+
+        plan = FlowPlan(flow_pk=flow.pk.hex, stages=[stage], markers=[StageMarker()])
         session = self.client.session
         session[SESSION_KEY_PLAN] = plan
         session.save()
         response = self.client.post(
-            reverse(
+            reverse("passbook_flows:flow-executor", kwargs={"flow_slug": flow.slug}),
-                "passbook_flows:flow-executor", kwargs={"flow_slug": self.flow.slug}
-            ),
             {},
         )
         self.assertEqual(response.status_code, 200)
@@ -48,3 +52,83 @@ class TestConsentStage(TestCase):
             force_text(response.content),
             {"type": "redirect", "to": reverse("passbook_core:overview")},
         )
+        self.assertFalse(UserConsent.objects.filter(user=self.user).exists())
+
+    def test_permanent(self):
+        """Test permanent consent from user"""
+        self.client.force_login(self.user)
+        flow = Flow.objects.create(
+            name="test-consent",
+            slug="test-consent",
+            designation=FlowDesignation.AUTHENTICATION,
+        )
+        stage = ConsentStage.objects.create(name="consent", mode=ConsentMode.PERMANENT)
+        FlowStageBinding.objects.create(target=flow, stage=stage, order=2)
+
+        plan = FlowPlan(
+            flow_pk=flow.pk.hex,
+            stages=[stage],
+            markers=[StageMarker()],
+            context={PLAN_CONTEXT_APPLICATION: self.application},
+        )
+        session = self.client.session
+        session[SESSION_KEY_PLAN] = plan
+        session.save()
+        response = self.client.post(
+            reverse("passbook_flows:flow-executor", kwargs={"flow_slug": flow.slug}),
+            {},
+        )
+        self.assertEqual(response.status_code, 200)
+        self.assertJSONEqual(
+            force_text(response.content),
+            {"type": "redirect", "to": reverse("passbook_core:overview")},
+        )
+        self.assertTrue(
+            UserConsent.objects.filter(
+                user=self.user, application=self.application
+            ).exists()
+        )
+
+    def test_expire(self):
+        """Test expiring consent from user"""
+        self.client.force_login(self.user)
+        flow = Flow.objects.create(
+            name="test-consent",
+            slug="test-consent",
+            designation=FlowDesignation.AUTHENTICATION,
+        )
+        stage = ConsentStage.objects.create(
+            name="consent", mode=ConsentMode.EXPIRING, consent_expire_in="seconds=1"
+        )
+        FlowStageBinding.objects.create(target=flow, stage=stage, order=2)
+
+        plan = FlowPlan(
+            flow_pk=flow.pk.hex,
+            stages=[stage],
+            markers=[StageMarker()],
+            context={PLAN_CONTEXT_APPLICATION: self.application},
+        )
+        session = self.client.session
+        session[SESSION_KEY_PLAN] = plan
+        session.save()
+        response = self.client.post(
+            reverse("passbook_flows:flow-executor", kwargs={"flow_slug": flow.slug}),
+            {},
+        )
+        self.assertEqual(response.status_code, 200)
+        self.assertJSONEqual(
+            force_text(response.content),
+            {"type": "redirect", "to": reverse("passbook_core:overview")},
+        )
+        self.assertTrue(
+            UserConsent.objects.filter(
+                user=self.user, application=self.application
+            ).exists()
+        )
+        sleep(1)
+        clean_expired_models()
+        self.assertFalse(
+            UserConsent.objects.filter(
+                user=self.user, application=self.application
+            ).exists()
+        )