feature: add uid, name, mail args support, write tests

ckanext/__init__.py

@@ -3,7 +3,9 @@
 # this is a namespace package
 try:
     import pkg_resources
+
     pkg_resources.declare_namespace(__name__)
 except ImportError:
     import pkgutil
+
     __path__ = pkgutil.extend_path(__path__, __name__)

ckanext/let_me_in/cli.py

@@ -19,15 +19,18 @@ def letmein():
 
 
 @letmein.command()
-@click.argument("user", required=True)
-def uli(user: str):
-    """Create a one-time login link for a user"""
+@click.option("--uid", "-n", default=None, help="User ID")
+@click.option("--name", "-u", default=None, help="User name")
+@click.option("--mail", "-e", default=None, help="User email")
+def uli(uid: str, name: str, mail: str):
+    """Create a one-time login link for a user by its ID/name/email"""
 
     try:
-        result = tk.get_action("lmi_generate_otl")({"ignore_auth": True}, {"user": user})
+        result = tk.get_action("lmi_generate_otl")(
+            {"ignore_auth": True}, {"uid": uid, "name": name, "mail": mail}
+        )
     except tk.ValidationError as e:
-        return click.secho(e.error_summary, fg="red")
+        return click.secho(e, fg="red", err=True)
 
-    click.echo()
     click.echo("Your one-time login link has been generated")
     click.secho(result["url"], fg="green")

ckanext/let_me_in/logic/action.py

@@ -7,14 +7,12 @@
 import jwt
 
 import ckan.plugins.toolkit as tk
-from ckan import types, model
+from ckan import model, types
 from ckan.logic import validate
 
 import ckanext.let_me_in.logic.schema as schema
 import ckanext.let_me_in.utils as lmi_utils
 
-# import ckanext.let_me_in.model as lmi_model
-
 
 @validate(schema.lmi_generate_otl)
 def lmi_generate_otl(
@@ -27,7 +25,25 @@ def lmi_generate_otl(
     """
     tk.check_access("lmi_generate_otl", context, data_dict)
 
-    user = cast(model.User, model.User.get(data_dict["user"]))
+    uid = data_dict.get("uid", "")
+    name = data_dict.get("name", "")
+    mail = data_dict.get("mail", "")
+
+    if not any([uid, name, mail]):
+        raise tk.ValidationError(
+            tk._(
+                "Please, provide uid, name or mail option",
+            )
+        )
+
+    if sum([1 for x in (uid, name, mail) if x]) > 1:
+        raise tk.ValidationError(
+            tk._(
+                "One param could be used at a time: uid, name or mail",
+            )
+        )
+
+    user = cast(model.User, lmi_utils.get_user(uid or name or mail))
     now = dt.utcnow()
     expires_at = now + td(hours=24)
 
@@ -37,8 +53,4 @@ def lmi_generate_otl(
         algorithm="HS256",
     )
 
-    # lmi_model.OneTimeLoginToken.create(
-    #     {"token": token, "user_id": user.id, "expires_at": expires_at}
-    # )
-
     return {"url": tk.url_for("lmi.login_with_token", token=token, _external=True)}

ckanext/let_me_in/logic/auth.py

@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-
 from ckan import types
 
 

ckanext/let_me_in/logic/schema.py

@@ -4,10 +4,19 @@
 
 from ckan.logic.schema import validator_args
 
-
 Schema = Dict[str, Any]
 
 
 @validator_args
-def lmi_generate_otl(not_empty, unicode_safe, user_id_or_name_exists) -> Schema:
-    return {"user": [not_empty, unicode_safe, user_id_or_name_exists]}
+def lmi_generate_otl(
+    ignore_missing,
+    unicode_safe,
+    user_id_exists,
+    user_name_exists,
+    user_email_exists,
+) -> Schema:
+    return {
+        "uid": [ignore_missing, unicode_safe, user_id_exists],
+        "name": [ignore_missing, unicode_safe, user_name_exists],
+        "mail": [ignore_missing, unicode_safe, user_email_exists],
+    }

ckanext/let_me_in/logic/validators.py

@@ -0,0 +1,17 @@
+from __future__ import annotations
+
+from typing import Any
+
+import ckan.plugins.toolkit as tk
+from ckan import types
+
+
+def user_email_exists(email: str, context: types.Context) -> Any:
+    """Ensures that user with a specific email exists.
+    Transform the email to user ID"""
+    result = context["model"].User.by_email(email)
+
+    if not result:
+        raise tk.Invalid(tk._("Not found: User"))
+
+    return result.id

ckanext/let_me_in/plugin.py

@@ -6,10 +6,6 @@
 @tk.blanket.cli
 @tk.blanket.auth_functions
 @tk.blanket.blueprints
+@tk.blanket.validators
 class LetMeInPlugin(p.SingletonPlugin):
-    p.implements(p.IConfigurer)
-
-    # IConfigurer
-
-    def update_config(self, config_):
-        pass
+    pass

ckanext/let_me_in/tests/test_auth.py

@@ -0,0 +1,26 @@
+from __future__ import annotations
+
+import pytest
+
+import ckan.model as model
+import ckan.plugins.toolkit as tk
+from ckan.tests.helpers import call_auth
+
+
+class TestGenerateOTLAuth:
+    def test_anon(self):
+        with pytest.raises(tk.NotAuthorized):
+            call_auth("lmi_generate_otl", context={"user": None, "model": model})
+
+    @pytest.mark.usefixtures("clean_db")
+    def test_regular_user(self, user):
+        with pytest.raises(tk.NotAuthorized):
+            call_auth(
+                "lmi_generate_otl", context={"user": user["name"], "model": model}
+            )
+
+    @pytest.mark.usefixtures("clean_db")
+    def test_sysadmin(self, sysadmin):
+        call_auth(
+            "lmi_generate_otl", context={"user": sysadmin["name"], "model": model}
+        )

ckanext/let_me_in/tests/test_logic.py

@@ -0,0 +1,46 @@
+from __future__ import annotations
+
+import pytest
+
+import ckan.plugins.toolkit as tk
+from ckan.tests.helpers import call_action
+
+
+class TestGenerateOTL:
+    def test_generate_no_params(self):
+        with pytest.raises(
+            tk.ValidationError, match="Please, provide uid, name or mail option"
+        ):
+            call_action("lmi_generate_otl")
+
+    @pytest.mark.usefixtures("clean_db")
+    def test_more_than_one_param(self, user):
+        with pytest.raises(
+            tk.ValidationError,
+            match="One param could be used at a time: uid, name or mail",
+        ):
+            call_action("lmi_generate_otl", uid=user["id"], name=user["name"])
+
+    def test_uid_not_exist(self):
+        with pytest.raises(tk.ValidationError, match="Not found: User"):
+            call_action("lmi_generate_otl", uid="xxx")
+
+    def test_name_not_exist(self):
+        with pytest.raises(tk.ValidationError, match="Not found: User"):
+            call_action("lmi_generate_otl", name="xxx")
+
+    def test_mail_not_exist(self):
+        with pytest.raises(tk.ValidationError, match="Not found: User"):
+            call_action("lmi_generate_otl", mail="xxx")
+
+    @pytest.mark.usefixtures("clean_db")
+    def test_by_uid(self, user):
+        call_action("lmi_generate_otl", uid=user["id"])
+
+    @pytest.mark.usefixtures("clean_db")
+    def test_by_name(self, user):
+        call_action("lmi_generate_otl", name=user["name"])
+
+    @pytest.mark.usefixtures("clean_db")
+    def test_by_male(self, user):
+        call_action("lmi_generate_otl", mail=user["email"])

ckanext/let_me_in/tests/test_validators.py

@@ -0,0 +1,18 @@
+from __future__ import annotations
+
+import pytest
+
+import ckan.plugins.toolkit as tk
+from ckan import model
+
+from ckanext.let_me_in.logic.validators import user_email_exists
+
+
+class TestEmailExistValidator:
+    def test_no_user(self):
+        with pytest.raises(tk.Invalid, match="User not found"):
+            user_email_exists("test", {"model": model})
+
+    @pytest.mark.usefixtures("clean_db")
+    def test_user_exists(self, user):
+        assert user_email_exists(user["email"], {"model": model})

ckanext/let_me_in/utils.py

@@ -2,8 +2,8 @@
 
 from typing import cast
 
-from ckan.lib.api_token import _get_secret
 from ckan import model
+from ckan.lib.api_token import _get_secret
 
 
 def get_secret(encode: bool) -> str:
@@ -15,5 +15,5 @@ def get_secret(encode: bool) -> str:
 
 
 def get_user(user_id: str) -> model.User:
-    """Get a user by its ID"""
+    """Get a user by its ID/name"""
     return cast(model.User, model.User.get(user_id))

ckanext/let_me_in/views.py

@@ -4,10 +4,10 @@
 from datetime import datetime as dt
 
 import jwt
+from flask import Blueprint
 
 import ckan.model as model
 from ckan.plugins import toolkit as tk
-from flask import Blueprint
 
 import ckanext.let_me_in.utils as lmi_utils
 
@@ -47,5 +47,6 @@ def login_with_token(token):
 
 
 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.py

@@ -7,10 +7,10 @@
     # message extraction at
     # http://babel.pocoo.org/docs/messages/#extraction-method-mapping-and-configuration
     message_extractors={
-        'ckanext': [
-            ('**.py', 'python', None),
-            ('**.js', 'javascript', None),
-            ('**/templates/**.html', 'ckan', None),
+        "ckanext": [
+            ("**.py", "python", None),
+            ("**.js", "javascript", None),
+            ("**/templates/**.html", "ckan", None),
         ],
     }
 )