feature: add interface, write more tests

ckanext/let_me_in/interfaces.py

@@ -0,0 +1,27 @@
+from ckan import model
+from ckan.plugins.interfaces import Interface
+
+
+class ILetMeIn(Interface):
+    """Interface to do something on user login via OTL link"""
+
+    def manage_user(self, user: model.User) -> model.User:
+        """Accept a user object that will be logged. Only Active user could be
+        logged in, so here we could implement user re-activation, for example.
+
+        The user is always exists, otherwise we are not getting here.
+
+        This happend before we are checking for user state and actual login
+
+        Must return a user object
+        """
+        return user
+
+    def before_otl_login(self, user: model.User) -> None:
+        """Allows to do something before we are logging in a user.
+        Happens after all checks and `manage_user` method."""
+        pass
+
+    def after_otl_login(self, user: model.User) -> None:
+        """Allows to do something after we logged in a user."""
+        pass

ckanext/let_me_in/plugin.py

@@ -1,11 +1,22 @@
+import ckan.model as model
 import ckan.plugins as p
 import ckan.plugins.toolkit as tk
 
+import ckanext.let_me_in.utils as lmi_utils
+from ckanext.let_me_in.interfaces import ILetMeIn
+
 
 @tk.blanket.actions
 @tk.blanket.cli
 @tk.blanket.auth_functions
 @tk.blanket.blueprints
 @tk.blanket.validators
 class LetMeInPlugin(p.SingletonPlugin):
-    pass
+    p.implements(ILetMeIn, inherit=True)
+
+    # ILetMeIn
+
+    def after_otl_login(self, user: model.User) -> None:
+        lmi_utils.update_user_last_active(user)
+
+        tk.h.flash_success("You have been logged in.")

ckanext/let_me_in/tests/test_interface.py

@@ -0,0 +1,59 @@
+import pytest
+
+import ckan.model as model
+import ckan.plugins as p
+from ckan.tests.helpers import call_action
+
+from ckanext.let_me_in.interfaces import ILetMeIn
+
+
+class TestOTLPlugin(p.SingletonPlugin):
+    p.implements(p.IConfigurable, inherit=True)
+    p.implements(ILetMeIn)
+
+    def configure(self, _config):
+        self.manage_user_call = 0
+        self.before_otl_login_call = 0
+        self.after_otl_login_call = 0
+
+    def manage_user(self, user: model.User) -> model.User:
+        self.manage_user_call += 1
+
+        return user
+
+    def before_otl_login(self, user: model.User) -> None:
+        self.before_otl_login_call += 1
+
+    def after_otl_login(self, user: model.User) -> None:
+        self.after_otl_login_call += 1
+
+
+@pytest.mark.ckan_config("ckan.plugins", "let_me_in test_otl_plugin")
+@pytest.mark.usefixtures("non_clean_db", "with_plugins")
+class TestOTLInterace(object):
+    def test_xxx(self, app, user, sysadmin):
+        result = call_action("lmi_generate_otl", uid=user["id"])
+
+        result = app.get(result["url"], status=200)
+
+        manage_user_call_total = sum(
+            plugin.manage_user_call
+            for plugin in p.PluginImplementations(ILetMeIn)
+            if plugin.name == "test_otl_plugin"
+        )
+
+        before_otl_login_call_total = sum(
+            plugin.before_otl_login_call
+            for plugin in p.PluginImplementations(ILetMeIn)
+            if plugin.name == "test_otl_plugin"
+        )
+
+        after_otl_login_call_total = sum(
+            plugin.after_otl_login_call
+            for plugin in p.PluginImplementations(ILetMeIn)
+            if plugin.name == "test_otl_plugin"
+        )
+
+        assert manage_user_call_total == 1
+        assert before_otl_login_call_total == 1
+        assert after_otl_login_call_total == 1

ckanext/let_me_in/tests/test_logic.py

@@ -6,6 +6,7 @@
 from ckan.tests.helpers import call_action
 
 
+@pytest.mark.usefixtures("non_clean_db", "with_plugins")
 class TestGenerateOTL:
     def test_generate_no_params(self):
         with pytest.raises(

ckanext/let_me_in/tests/test_validators.py

@@ -10,7 +10,7 @@
 
 class TestEmailExistValidator:
     def test_no_user(self):
-        with pytest.raises(tk.Invalid, match="User not found"):
+        with pytest.raises(tk.Invalid, match="Not found: User"):
             user_email_exists("test", {"model": model})
 
     @pytest.mark.usefixtures("clean_db")

ckanext/let_me_in/tests/test_views.py

@@ -0,0 +1,38 @@
+from __future__ import annotations
+
+from typing import cast
+
+import pytest
+
+import ckan.model as model
+from ckan.tests.helpers import call_action
+
+
+@pytest.mark.usefixtures("non_clean_db")
+class TestOTLViews(object):
+    def test_login_user_with_otl(self, app, user):
+        otl = call_action("lmi_generate_otl", uid=user["id"])
+
+        result = app.get(otl["url"], status=200)
+        assert "You have been logged in" in result.text
+
+        result = app.get(otl["url"], status=200)
+        assert (
+            "You have tried to use a one-time login link that has expired"
+            in result.text
+        )
+
+    def test_user_login_expires_the_otl(self, app, user):
+        """We are not creating any entity for OTL. It expires right after the
+        user it was created for is logged in. This triggers the update of
+        `last_active` field and if the OTL is older than this, it will be invalidated"""
+        otl = call_action("lmi_generate_otl", uid=user["id"])
+
+        user = cast(model.User, model.User.get(user["id"]))
+        user.set_user_last_active()
+
+        result = app.get(otl["url"], status=200)
+        assert (
+            "You have tried to use a one-time login link that has expired"
+            in result.text
+        )

ckanext/let_me_in/utils.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from datetime import datetime as dt
 from typing import cast
 
 from ckan import model
@@ -17,3 +18,9 @@ def get_secret(encode: bool) -> str:
 def get_user(user_id: str) -> model.User:
     """Get a user by its ID/name"""
     return cast(model.User, model.User.get(user_id))
+
+
+def update_user_last_active(user: model.User) -> None:
+    """Update a last_active for a user after we logged him in."""
+    user.last_active = dt.utcnow()
+    model.Session.commit()

ckanext/let_me_in/views.py

@@ -7,9 +7,11 @@
 from flask import Blueprint
 
 import ckan.model as model
+import ckan.plugins as p
 from ckan.plugins import toolkit as tk
 
 import ckanext.let_me_in.utils as lmi_utils
+from ckanext.let_me_in.interfaces import ILetMeIn
 
 log = logging.getLogger(__name__)
 lmi = Blueprint("lmi", __name__)
@@ -26,27 +28,29 @@ def login_with_token(token):
     else:
         user = lmi_utils.get_user(token["user_id"])
 
+        for plugin in p.PluginImplementations(ILetMeIn):
+            user = plugin.manage_user(user)
+
         if user.state != model.State.ACTIVE:
             tk.h.flash_error(tk._("User is not active. Can't login"))
             return tk.h.redirect_to("user.login")
 
-        if user.last_active > dt.fromtimestamp(token["created_at"]):
+        if user.last_active and user.last_active > dt.fromtimestamp(
+            token["created_at"]
+        ):
             tk.h.flash_error(
                 tk._("You have tried to use a one-time login link that has expired.")
             )
             return tk.h.redirect_to("user.login")
 
+        for plugin in p.PluginImplementations(ILetMeIn):
+            plugin.before_otl_login(user)
+
         tk.login_user(user)
-        _update_user_last_active(user)
 
-        tk.h.flash_success("You have been logged in.")
+        for plugin in p.PluginImplementations(ILetMeIn):
+            plugin.after_otl_login(user)
 
         return tk.h.redirect_to("user.me")
 
     return tk.h.redirect_to("user.login")
-
-
-def _update_user_last_active(user: model.User) -> None:
-    """Update a last_active for a user after we logged him in."""
-    user.last_active = dt.utcnow()
-    model.Session.commit()

setup.cfg

@@ -25,6 +25,7 @@ include_package_data = True
 [options.entry_points]
 ckan.plugins =
              let_me_in = ckanext.let_me_in.plugin:LetMeInPlugin
+             test_otl_plugin = ckanext.let_me_in.tests.test_interface:TestOTLPlugin
 
 babel.extractors =
                  ckan = ckan.lib.extract:extract_ckan