flows: enum to django TextChoices

passbook/flows/exceptions.py

@@ -1,5 +1,9 @@
 """flow exceptions"""
 
 
-class FlowNonApplicableError(BaseException):
+class FlowNonApplicableException(BaseException):
     """Exception raised when a Flow does not apply to a user."""
+
+
+class EmptyFlowException(BaseException):
+    """Exception raised when a Flow Plan is empty"""

passbook/flows/migrations/0003_auto_20200509_1258.py

@@ -0,0 +1,26 @@
+# Generated by Django 3.0.3 on 2020-05-09 12:58
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("passbook_flows", "0002_default_flows"),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name="flow",
+            name="designation",
+            field=models.CharField(
+                choices=[
+                    ("authentication", "Authentication"),
+                    ("enrollment", "Enrollment"),
+                    ("recovery", "Recovery"),
+                    ("password_change", "Password Change"),
+                ],
+                max_length=100,
+            ),
+        ),
+    ]

passbook/flows/models.py

@@ -11,7 +11,7 @@ from passbook.lib.models import UUIDModel
 from passbook.policies.models import PolicyBindingModel
 
 
-class FlowDesignation(Enum):
+class FlowDesignation(models.TextChoices):
     """Designation of what a Flow should be used for. At a later point, this
     should be replaced by a database entry."""
 
@@ -20,13 +20,6 @@ class FlowDesignation(Enum):
     RECOVERY = "recovery"
     PASSWORD_CHANGE = "password_change"  # nosec # noqa
 
-    @staticmethod
-    def as_choices() -> Tuple[Tuple[str, str]]:
-        """Generate choices of actions used for database"""
-        return tuple(
-            (x, y.value) for x, y in getattr(FlowDesignation, "__members__").items()
-        )
-
 
 class Stage(UUIDModel):
     """Stage is an instance of a component used in a flow. This can verify the user,
@@ -56,7 +49,7 @@ class Flow(PolicyBindingModel, UUIDModel):
     name = models.TextField()
     slug = models.SlugField(unique=True)
 
-    designation = models.CharField(max_length=100, choices=FlowDesignation.as_choices())
+    designation = models.CharField(max_length=100, choices=FlowDesignation.choices)
 
     stages = models.ManyToManyField(Stage, through="FlowStageBinding", blank=True)
 

passbook/flows/planner.py

@@ -6,7 +6,7 @@ from typing import Any, Dict, List, Tuple
 from django.http import HttpRequest
 from structlog import get_logger
 
-from passbook.flows.exceptions import FlowNonApplicableError
+from passbook.flows.exceptions import EmptyFlowException, FlowNonApplicableException
 from passbook.flows.models import Flow, Stage
 from passbook.policies.engine import PolicyEngine
 
@@ -26,8 +26,7 @@ class FlowPlan:
 
     def next(self) -> Stage:
         """Return next pending stage from the bottom of the list"""
-        stage_cls = self.stages.pop(0)
-        return stage_cls
+        return self.stages[0]
 
 
 class FlowPlanner:
@@ -54,7 +53,7 @@ class FlowPlanner:
         # to make sure the user even has access to the flow
         root_passing, root_passing_messages = self._check_flow_root_policies(request)
         if not root_passing:
-            raise FlowNonApplicableError(root_passing_messages)
+            raise FlowNonApplicableException(root_passing_messages)
         # Check Flow policies
         for stage in (
             self.flow.stages.order_by("flowstagebinding__order")
@@ -72,4 +71,6 @@ class FlowPlanner:
         LOGGER.debug(
             "Finished planning", flow=self.flow, duration_s=end_time - start_time
         )
+        if not plan.stages:
+            raise EmptyFlowException()
         return plan

passbook/flows/views.py

@@ -8,8 +8,8 @@ from django.views.generic import View
 from structlog import get_logger
 
 from passbook.core.views.utils import PermissionDeniedView
-from passbook.flows.exceptions import FlowNonApplicableError
-from passbook.flows.models import Flow, Stage
+from passbook.flows.exceptions import EmptyFlowException, FlowNonApplicableException
+from passbook.flows.models import Flow, FlowDesignation, Stage
 from passbook.flows.planner import PLAN_CONTEXT_PENDING_USER, FlowPlan, FlowPlanner
 from passbook.lib.config import CONFIG
 from passbook.lib.utils.reflection import class_to_path, path_to_class
@@ -52,15 +52,15 @@ class FlowExecutorView(View):
             return bad_request_message(self.request, message)
         return None
 
-    def handle_flow_non_applicable(self) -> HttpResponse:
+    def handle_invalid_flow(self, exc: BaseException) -> HttpResponse:
         """When a flow is non-applicable check if user is on the correct domain"""
         if NEXT_ARG_NAME in self.request.GET:
+            LOGGER.debug("Redirecting to next on fail")
             return redirect(self.request.GET.get(NEXT_ARG_NAME))
         incorrect_domain_message = self._check_config_domain()
         if incorrect_domain_message:
             return incorrect_domain_message
-        # TODO: Add message
-        return redirect("passbook_core:index")
+        return bad_request_message(self.request, str(exc))
 
     def dispatch(self, request: HttpRequest, flow_slug: str) -> HttpResponse:
         # Early check if theres an active Plan for the current session
@@ -70,9 +70,12 @@ class FlowExecutorView(View):
             )
             try:
                 self.plan = self._initiate_plan()
-            except FlowNonApplicableError as exc:
+            except FlowNonApplicableException as exc:
                 LOGGER.warning("Flow not applicable to current user", exc=exc)
-                return self.handle_flow_non_applicable()
+                return self.handle_invalid_flow(exc)
+            except EmptyFlowException as exc:
+                LOGGER.warning("Flow is empty", exc=exc)
+                return self.handle_invalid_flow(exc)
         else:
             LOGGER.debug("Continuing existing plan", flow_slug=flow_slug)
             self.plan = self.request.session[SESSION_KEY_PLAN]
@@ -136,6 +139,7 @@ class FlowExecutorView(View):
             stage_class=class_to_path(self.current_stage_view.__class__),
             flow_slug=self.flow.slug,
         )
+        self.plan.stages.pop(0)
         self.request.session[SESSION_KEY_PLAN] = self.plan
         if self.plan.stages:
             LOGGER.debug(
@@ -169,3 +173,16 @@ class FlowExecutorView(View):
 
 class FlowPermissionDeniedView(PermissionDeniedView):
     """User could not be authenticated"""
+
+
+class ToDefaultFlow(View):
+    """Redirect to default flow matching by designation"""
+
+    designation: Optional[FlowDesignation] = None
+
+    def dispatch(self, request: HttpRequest) -> HttpResponse:
+        flow = get_object_or_404(Flow, designation=self.designation)
+        # TODO: Get Flow depending on subdomain?
+        return redirect_with_qs(
+            "passbook_flows:flow-executor", request.GET, flow_slug=flow.slug
+        )