coverage

.gitignore

@@ -56,6 +56,5 @@ docs/_build/
 .pytest_cache
 coverage.xml
 
-Pipfile
-Pipfile.lock
-celerybeat.pid
+celerybeat.pid
+.cache

.pre-commit-config.yaml

@@ -11,3 +11,20 @@ repos:
       - id: ruff-format
         args:
           - --check
+  - repo: https://github.com/adamchainz/djade-pre-commit
+    rev: 1.3.2
+    hooks:
+      - id: djade
+        args:
+          - --target-version
+          - '5.1'
+#  - repo: https://github.com/saxix/pch
+#    rev: v0.1
+#    hooks:
+#      - id: check-missed-migrations
+#        args:
+#          - src
+#        stages:
+#          - pre-commit
+#        additional_dependencies:
+#          - setuptools

pyproject.toml

@@ -12,7 +12,6 @@ classifiers = [
     "Framework :: Django",
     "Framework :: Django :: 3.2",
     "Framework :: Django :: 4.2",
-    "Framework :: Django :: 5.0",
     "Framework :: Django :: 5.1",
     "Intended Audience :: Developers",
     "Programming Language :: Python :: 3.11",
@@ -56,3 +55,13 @@ dev-dependencies = [
 
 [project.urls]
 Homepage = "https://github.com/unicef/unicef-security"
+
+
+[uv]
+package = true
+
+[tool.nitpick]
+    style = [
+        "github://unicef/hope-code-conventions@main/django/django.toml"
+    ]
+    cache = "1 day"

ruff.toml

@@ -55,7 +55,7 @@ ignore = [
     "D106",  # Missing docstring in public nested class
     "D107",  # Missing docstring in `__init__`
     "D203", # one-blank-line-before-class
-#    "D212", # multi-line-summary-first-line
+    "D212", # multi-line-summary-first-line
     "D213", # multi-line-summary-second-line
     "E731",  # lambda-assignment: lambdas are substential in maintenance of py2/3 codebase
     "ISC001",  # conflicts with ruff format command

src/unicef_security/admin.py

@@ -18,7 +18,7 @@
 logger = logging.getLogger(__name__)
 
 
-def is_superuser(request, *args, **kwargs):
+def is_superuser(request, *args, **kwargs):  # pragma: no cover
     return request.user.is_superuser
 
 
@@ -31,10 +31,20 @@ class UNICEFUserFilter(SimpleListFilter):
     parameter_name = "email"
 
     def lookups(self, request, model_admin):
-        return [
-            ("unicef", "UNICEF"),
-            ("external", "External"),
-        ]
+        return (
+            (
+                "unicef",
+                _(
+                    "UNICEF",
+                ),
+            ),
+            (
+                "external",
+                _(
+                    "External",
+                ),
+            ),
+        )
 
     def queryset(self, request, queryset):
         if self.value() == "unicef":
@@ -114,7 +124,7 @@ def sync_user(self, request, pk):
         try:
             synchronizer = Synchronizer()
             synchronizer.sync_user(obj)
-        except ValueError as e:
+        except ValueError as e:  # pragma: no cover
             self.message_user(request, str(e), messages.ERROR)
 
         self.message_user(request, "User synchronized")
@@ -147,7 +157,7 @@ def link_user_data(self, request, pk):
             ctx["data"] = data
             return TemplateResponse(request, "admin/link_user.html", ctx)
 
-        except ValueError as e:
+        except ValueError as e:  # pragma: no cover
             self.message_user(request, str(e), messages.ERROR)
 
     @button()
@@ -190,7 +200,7 @@ def ad(self, request, pk):
         try:
             synchronizer = Synchronizer()
             context = synchronizer.get_user(obj.username)
-        except ValueError as e:
+        except ValueError as e:  # pragma: no cover
             self.message_user(request, str(e), messages.ERROR)
 
         return TemplateResponse(request, "admin/ad.html", {"ctx": context, "opts": self.model._meta})

src/unicef_security/graph.py

@@ -39,7 +39,7 @@ def default_group(**kwargs):
             user.is_staff = True
             user.is_superuser = True
             user.save()
-        elif group_name := constance.get("DEFAULT_GROUP"):
+        elif group_name := constance.DEFAULT_GROUP:
             group = Group.objects.filter(name=group_name).first()
             if group:
                 user.groups.add(group)
@@ -70,7 +70,7 @@ def get_unicef_user(backend, details, response, *args, **kwargs):
             for k, v in data.items():
                 details[k] = v
 
-        except (ValueError, KeyError) as e:
+        except (ValueError, KeyError) as e:  # pragma: no cover
             logger.error(e)
 
         user, created = User.objects.get_or_create(
@@ -84,7 +84,7 @@ def get_unicef_user(backend, details, response, *args, **kwargs):
                 "azure_id": details.get("id"),
             },
         )
-        social, __ = UserSocialAuth.objects.get_or_create(user=user, provider=backend.name, uid=user.username)
+        social, _ = UserSocialAuth.objects.get_or_create(user=user, provider=backend.name, uid=user.username)
         user.social_user = social
     return {"user": user, "social": social, "uid": details.get("id"), "is_new": created}
 
@@ -147,7 +147,7 @@ def __init__(self, user_model=None, mapping=None, echo=None, identifier=None, se
         self.echo = echo or (lambda lmn: True)
 
     def get_token(self):
-        if not self.id and self.secret:
+        if not self.id and self.secret:  # pragma: no cover
             raise ValueError("Configure AZURE_CLIENT_ID and/or AZURE_CLIENT_SECRET")
         post_dict = {
             "grant_type": "client_credentials",
@@ -175,18 +175,18 @@ def get_page(self, url, single=False):
             headers = {"Authorization": f"Bearer {self.get_token()}"}
             try:
                 response = requests.get(url, headers=headers, timeout=60)
-                if response.status_code == 401:
+                if response.status_code == 401:  # pragma: no cover
                     data = response.json()
                     if data["error"]["message"] == "Access token has expired.":
                         continue
                     raise ConnectionError(f"400: Error processing the response {response.content}")
 
-                if response.status_code != 200:
+                if response.status_code != 200:  # pragma: no cover
                     raise ConnectionError(
-                        f"Code {response.status_code}. " f"Error processing the response {response.content}"
+                        f"Code {response.status_code}. Error processing the response {response.content}"
                     )
                 break
-            except ConnectionError as e:
+            except ConnectionError as e:  # pragma: no cover
                 logger.exception(e)
                 raise
 
@@ -210,9 +210,7 @@ def __iter__(self):
                 values = self.get_page(self.next_link)
                 logger.debug(f"fetched page {pages}")
                 pages += 1
-            except KeyboardInterrupt:
-                break
-            except BaseException as e:
+            except GeneratorExit as e:
                 logger.exception(e)
                 break
 
@@ -221,19 +219,23 @@ def get_record(self, user_info: dict) -> (dict, dict):
         pk = {fieldname: data.pop(fieldname) for fieldname in self.user_pk_fields}
         return pk, data
 
-    def fetch_users(self, filter_params, callback=None):
+    def fetch_users(self, filter_params, max_records=None, callback=None):
         self.startUrl = "%s?$filter=%s" % (self._baseurl, filter_params)
-        return self.synchronize(callback=callback)
+        return self.synchronize(max_records=max_records, callback=callback)
 
     def search_users(self, record):
         url = "%s?$filter=" % self._baseurl
         filters = []
-        if record.email:
-            filters.append("mail eq '%s'" % record.email)
-        if record.last_name:
-            filters.append("surname eq '%s'" % record.last_name)
-        if record.first_name:
-            filters.append("givenName eq '%s'" % record.first_name)
+        field_map = {
+            "email": "mail eq '{value}'",
+            "last_name": "surname eq '{value}'",
+            "first_name": "givenName eq '{value}'",
+        }
+
+        for field, filter_template in field_map.items():
+            value = getattr(record, field, None)
+            if value:
+                filters.append(filter_template.format(value=value))
 
         page = self.get_page(url + " or ".join(filters), single=True)
         return page["value"]
@@ -284,7 +286,7 @@ def synchronize(self, max_records=None, callback=None):
                     results.log(user_info)
                 if max_records and i > max_records:
                     break
-        except BaseException as e:
+        except BaseException as e:  # pragma: no cover
             logger.exception(e)
             raise
         logger.debug(f"End Azure user synchronization: {results}")

tests/.coveragerc

@@ -4,7 +4,7 @@ source = unicef_security
 omit = *migrations*
 
 [report]
-fail_under = 20
+fail_under = 90
 ignore_errors = True
 exclude_lines =
    pragma: no cover

tests/conftest.py

@@ -2,7 +2,7 @@
 
 import pytest
 
-from .factories import UserFactory
+from .factories import UserFactory, SuperUserFactory
 
 
 @pytest.fixture
@@ -14,3 +14,12 @@ def mocked_responses():
 @pytest.fixture
 def auth_user():
     return UserFactory()
+
+
+@pytest.fixture
+def app(django_app_factory, mocked_responses):
+    django_app = django_app_factory(csrf_checks=False)
+    admin_user = SuperUserFactory(username="superuser")
+    django_app.set_user(admin_user)
+    django_app._user = admin_user
+    return django_app

tests/demoproject/demo/settings.py

@@ -90,9 +90,11 @@
 AUTH_USER_MODEL = "demo.User"
 
 CONSTANCE_CONFIG = {
-    "DEFAULT_GROUP": ("", "test_grp"),
+    "DEFAULT_GROUP": ("test_grp", "Test Group"),
 }
 
 CONSTANCE_BACKEND = "constance.backends.database.DatabaseBackend"
 
 LOGOUT_URL = "/"
+
+ADMINS = (("Me", "[email protected]"),)

tests/test_admin.py

@@ -1,7 +1,63 @@
 from django.urls import reverse
+from mock import patch
+import pytest
 
+from demo.admin import UserPlus
+from demo.models import User
+from unicef_security.admin import UNICEFUserFilter
+from .factories import UserFactory
 
-def test_changelist(django_app, admin_user):
-    url = reverse("admin:demo_user_changelist")
-    res = django_app.get(url, user=admin_user)
+
+@pytest.mark.django_db
+@patch("unicef_security.admin.Synchronizer.get_token")
+@patch("unicef_security.admin.Synchronizer.get_user")
+@patch("unicef_security.admin.Synchronizer.sync_user")
+@patch("unicef_security.admin.Synchronizer.search_users")
+def test_link_user_data(patch1, patch2, patch3, patch4, app):
+    user = UserFactory()
+    url = reverse("admin:demo_user_link_user_data", args=[user.pk])
+    res = app.post(url)
+    assert res.status_code == 200
+
+
+@pytest.mark.django_db
+@patch("unicef_security.admin.Synchronizer.get_token")
+@patch("unicef_security.admin.Synchronizer.fetch_users")
+def test_link_load(patch1, patch2, app):
+    url = reverse("admin:demo_user_load")
+    res = app.post(url)
     assert res.status_code == 200
+
+
+@pytest.mark.django_db
+def test_unicef_admin_filter():
+    UserFactory(email="[email protected]", username="[email protected]")
+    UserFactory(email="[email protected]", username="[email protected]")
+    qs = User.objects.all()
+    assert qs.count() == 2
+
+    filters = UNICEFUserFilter(
+        None,
+        {
+            "email": [
+                "unicef",
+            ]
+        },
+        User,
+        UserPlus,
+    )
+    unicef_records = filters.queryset(None, qs)
+    assert unicef_records.count() == 1
+
+    filters = UNICEFUserFilter(
+        None,
+        {
+            "email": [
+                "external",
+            ]
+        },
+        User,
+        UserPlus,
+    )
+    external_records = filters.queryset(None, qs)
+    assert external_records.count() == 1

tests/test_admin_smoke.py

@@ -11,7 +11,7 @@
 import pytest
 from unittest.mock import Mock
 
-from .factories import SuperUserFactory
+from mock import patch
 
 if TYPE_CHECKING:
     from django.db.models.options import Options
@@ -112,15 +112,6 @@ def record(db, request):
     return instance
 
 
-@pytest.fixture
-def app(django_app_factory, mocked_responses):
-    django_app = django_app_factory(csrf_checks=False)
-    admin_user = SuperUserFactory(username="superuser")
-    django_app.set_user(admin_user)
-    django_app._user = admin_user
-    return django_app
-
-
 def test_admin_index(app):
     url = reverse("admin:index")
 
@@ -144,7 +135,7 @@ def test_admin_changelist(app, modeladmin, record):
 def show_error(res):
     errors = []
     for k, v in dict(res.context["adminform"].form.errors).items():
-        errors.append(f'{k}: {"".join(v)}')
+        errors.append(f"{k}: {''.join(v)}")
     return (f"Form submitting failed: {res.status_code}: {errors}",)
 
 
@@ -185,8 +176,11 @@ def test_admin_delete(app, modeladmin, record, monkeypatch):
         pytest.skip("No 'delete' permission")
 
 
-@pytest.mark.skip_buttons("demo.UserPlus:link_user_data", "demo.UserPlus:sync_user", "demo.UserPlus:ad")
-def test_admin_buttons(app, modeladmin, button_handler, record, monkeypatch):
+@patch("unicef_security.admin.Synchronizer.get_token")
+@patch("unicef_security.admin.Synchronizer.get_user")
+@patch("unicef_security.admin.Synchronizer.sync_user")
+@patch("unicef_security.admin.Synchronizer.search_users")
+def test_admin_buttons(patch1, patch2, patch3, patch4, app, modeladmin, button_handler, record, monkeypatch):
     from admin_extra_buttons.handlers import LinkHandler
 
     if isinstance(button_handler, ChoiceHandler):