program/office security backend

pyproject.toml

@@ -32,9 +32,9 @@ dependencies = [
     "django-stubs-ext",
     "django>=5.1",
     "djangorestframework>=3.15.1",
-    "hope-flex-fields>=0.4.0",
+    "hope-flex-fields>=0.5.0",
     "hope-smart-export>=0.3.0",
-    "hope-smart-import>=0.2.0",
+    "hope-smart-import>=0.3.0",
     "psycopg2-binary>=2.9.9",
     "pytricia>=1.0.2",
     "redis",

pytest.ini

@@ -28,6 +28,7 @@ markers =
     skip_buttons
     select_buttons
     smoke
+    security
     skip_models
 
 

src/country_workspace/models/household.py

@@ -8,6 +8,9 @@
 if TYPE_CHECKING:
     from hope_flex_fields.models import DataChecker
 
+    from .office import Office
+    from .program import Program
+
 
 class Household(Validable, BaseModel):
     system_fields = models.JSONField(default=dict, blank=True)
@@ -23,9 +26,9 @@ def checker(self) -> "DataChecker":
         return self.program.household_checker
 
     @cached_property
-    def program(self) -> "DataChecker":
+    def program(self) -> "Program":
         return self.batch.program
 
     @cached_property
-    def country_office(self) -> "DataChecker":
+    def country_office(self) -> "Office":
         return self.batch.program.country_office

src/country_workspace/models/office.py

@@ -13,5 +13,7 @@ class Office(BaseModel):
     slug = models.SlugField(max_length=100, blank=True, null=True, db_index=True)
     active = models.BooleanField(default=False)
 
+    extra_fields = models.JSONField(default=dict, blank=True, null=False)
+
     def __str__(self) -> str:
         return str(self.name)

src/country_workspace/models/program.py

@@ -61,6 +61,7 @@ class Program(BaseModel):
     individual_search = models.TextField(default="name", help_text="Fields to use for searches")
     household_columns = models.TextField(default="name\nid", help_text="Columns to display ib the Admin table")
     individual_columns = models.TextField(default="name\nid", help_text="Columns to display ib the Admin table")
+    extra_fields = models.JSONField(default=dict, blank=True, null=False)
 
     def __str__(self) -> str:
         return self.name

src/country_workspace/workspaces/admin/batch.py

@@ -3,6 +3,7 @@
 from django.db.models import QuerySet
 from django.http import HttpRequest
 
+from ...state import state
 from ..options import WorkspaceModelAdmin
 
 if TYPE_CHECKING:
@@ -18,4 +19,7 @@ class CountryBatchAdmin(WorkspaceModelAdmin):
     ordering = ("name",)
 
     def get_queryset(self, request: HttpRequest) -> "QuerySet[CountryBatch]":
-        return super().get_queryset(request)
+        return super().get_queryset(request).filter(country_office=state.tenant)
+
+    def has_add_permission(self, request, obj=None):
+        return False

src/country_workspace/workspaces/admin/household.py

@@ -7,6 +7,7 @@
 from admin_extra_buttons.buttons import LinkButton
 from admin_extra_buttons.decorators import link
 
+from ...state import state
 from .hh_ind import CountryHouseholdIndividualBaseAdmin
 
 if TYPE_CHECKING:
@@ -38,7 +39,7 @@ def get_list_display(self, request: HttpRequest) -> list[str]:
         ]
 
     def get_queryset(self, request: HttpRequest) -> "QuerySet[CountryHousehold]":
-        return super().get_queryset(request)
+        return super().get_queryset(request).filter(batch__country_office=state.tenant)
 
     @link(change_list=False)
     def members(self, btn: LinkButton) -> None:

src/country_workspace/workspaces/admin/individual.py

@@ -4,6 +4,7 @@
 from django.db.models import Model
 from django.http import HttpRequest
 
+from ...state import state
 from ..filters import HouseholdFilter, ProgramFilter
 from ..models import CountryHousehold, CountryIndividual, CountryProgram
 from .hh_ind import CountryHouseholdIndividualBaseAdmin
@@ -33,6 +34,9 @@ def __init__(self, model: Model, admin_site: AdminSite):
         self._selected_household = None
         super().__init__(model, admin_site)
 
+    def get_queryset(self, request):
+        return super().get_queryset(request).filter(batch__country_office=state.tenant)
+
     def get_list_display(self, request: HttpRequest) -> list[str]:
         program: CountryProgram | None
         if program := self.get_selected_program(request):

src/country_workspace/workspaces/admin/program.py

@@ -131,10 +131,6 @@ def _configure_columns(
                 columns = []
                 for s in form.cleaned_data["columns"]:
                     columns.append(s)
-                    # if s in form.model_core_fields:
-                    #     columns.append(s)
-                    # else:
-                    #     columns.append("flex_fields__%s" % s)
                 setattr(program, context["storage_field"], "\n".join(columns))
                 program.save()
                 return HttpResponseRedirect(reverse("workspace:workspaces_countryprogram_change", args=[program.pk]))

src/country_workspace/workspaces/backend.py

@@ -13,30 +13,63 @@
 
 
 class TenantBackend(BaseBackend):
-    def get_all_permissions(self, user: "User|AnonymousUser", obj: "Model|None" = None) -> set[str]:
-        tenant: "Office|None" = state.tenant
-        if not tenant:
-            return set()
+    def get_group_permissions(self, user: "User|AnonymousUser", obj: "Model|None" = None) -> set[str]:
+        from country_workspace.workspaces.models import (
+            Batch,
+            CountryBatch,
+            CountryHousehold,
+            CountryIndividual,
+            CountryProgram,
+            Household,
+            Individual,
+            Program,
+        )
+
         if user.is_anonymous:
             return set()
-        perm_cache_name = "_tenant_%s_perm_cache" % str(tenant.pk)
-        if not hasattr(user, perm_cache_name):
-            qs = Permission.objects.all()
+        if not obj:
+            obj = get_selected_tenant()
+
+        filters = {}
+        if isinstance(obj, Office):
+            program = None
+            country_office = obj
+            filters = {"group__userrole__country_office": country_office}
+        elif isinstance(obj, (CountryBatch, Batch)):
+            program = obj.program
+            country_office = obj.country_office
+            filters = {"group__userrole__country_office": country_office, "group__userrole__program": program}
+        elif isinstance(obj, (CountryProgram, Program)):
+            program = obj
+            country_office = obj.country_office
+            filters = {
+                "group__userrole__country_office": country_office,
+            }
+        elif isinstance(obj, (CountryHousehold, Household)):
+            program = obj.program
+            country_office = obj.country_office
+        elif isinstance(obj, (CountryIndividual, Individual)):
+            program = obj.program
+            country_office = obj.country_office
+        else:
+            return set()
+        if not hasattr(user, "_tenant_cache"):
+            user._tenant_cache = {}
+        perm_cache_name = "%s_%s" % (str(country_office), str(program))
+        if not user._tenant_cache.get(perm_cache_name):
+            qs = Permission.objects.filter(content_type__app_label="workspaces")
             if not user.is_superuser:
-                qs = qs.filter(
-                    **{
-                        "group__userrole__user": user,
-                        "group__userrole__country_office": tenant,
-                    }
+                qs = qs.filter(group__userrole__user=user).filter(
+                    Q(group__userrole__country_office=country_office, group__userrole__program=None) | Q(**filters)
                 )
             perms = qs.values_list("content_type__app_label", "codename").order_by()
-            setattr(user, perm_cache_name, {f"{ct}.{name}" for ct, name in perms})
-        return getattr(user, perm_cache_name)
+            user._tenant_cache[perm_cache_name] = {f"{ct}.{name}" for ct, name in perms}
+        return user._tenant_cache[perm_cache_name]
 
     def get_available_modules(self, user: "User") -> "set[str]":
-        return {perm[: perm.index(".")] for perm in self.get_all_permissions(user)}
+        return {perm[: perm.index(".")] for perm in self.get_all_permissions(user, state.tenant)}
 
-    def has_perm(self, user_obj: "User", perm: str, obj: Optional[Model] = None):
+    def has_perm(self, user_obj: "User|AnonymousUser", perm: str, obj: Optional[Model] = None):
         if user_obj.is_superuser:
             return True
         return super().has_perm(user_obj, perm, obj)
@@ -47,9 +80,6 @@ def has_module_perms(self, user: "User", app_label: str) -> bool:
         tenant: "Model" = get_selected_tenant()
         if not tenant:
             return False
-
-        if user.is_superuser:
-            return True
         return app_label in self.get_available_modules(user)
 
     def get_allowed_tenants(self, request: "HttpRequest|None" = None) -> "Optional[QuerySet[Model]]":

src/country_workspace/workspaces/migrations/0001_initial.py

@@ -0,0 +1,82 @@
+# Generated by Django 5.1.1 on 2024-10-14 10:37
+
+import django.db.models.deletion
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    initial = True
+
+    dependencies = [
+        ("country_workspace", "0001_initial"),
+        ("hope_flex_fields", "0007_create_default_fields"),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name="CountryBatch",
+            fields=[],
+            options={
+                "verbose_name": "Country Batch",
+                "verbose_name_plural": "Country Batches",
+                "proxy": True,
+                "indexes": [],
+                "constraints": [],
+            },
+            bases=("country_workspace.batch",),
+        ),
+        migrations.CreateModel(
+            name="CountryHousehold",
+            fields=[],
+            options={
+                "verbose_name": "Country Household",
+                "verbose_name_plural": "Country Households",
+                "proxy": True,
+                "indexes": [],
+                "constraints": [],
+            },
+            bases=("country_workspace.household",),
+        ),
+        migrations.CreateModel(
+            name="CountryIndividual",
+            fields=[],
+            options={
+                "proxy": True,
+                "indexes": [],
+                "constraints": [],
+            },
+            bases=("country_workspace.individual",),
+        ),
+        migrations.CreateModel(
+            name="CountryProgram",
+            fields=[],
+            options={
+                "proxy": True,
+                "indexes": [],
+                "constraints": [],
+            },
+            bases=("country_workspace.program",),
+        ),
+        migrations.CreateModel(
+            name="CountryChecker",
+            fields=[
+                (
+                    "datachecker_ptr",
+                    models.OneToOneField(
+                        auto_created=True,
+                        on_delete=django.db.models.deletion.CASCADE,
+                        parent_link=True,
+                        primary_key=True,
+                        serialize=False,
+                        to="hope_flex_fields.datachecker",
+                    ),
+                ),
+                (
+                    "country_office",
+                    models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to="country_workspace.office"),
+                ),
+            ],
+            bases=("hope_flex_fields.datachecker",),
+        ),
+    ]

src/country_workspace/workspaces/models.py

@@ -1,4 +1,7 @@
+from typing import cast
+
 from django.db import models
+from django.utils.functional import cached_property
 
 from hope_flex_fields.models import DataChecker
 
@@ -20,6 +23,14 @@ class Meta:
         verbose_name = "Country Household"
         verbose_name_plural = "Country Households"
 
+    @cached_property
+    def program(self) -> "CountryProgram":
+        return cast(CountryProgram, self.batch.program)
+
+    @cached_property
+    def country_office(self) -> "DataChecker":
+        return self.batch.program.country_office
+
 
 class CountryIndividual(Individual):
     class Meta:

src/country_workspace/workspaces/sites.py

@@ -142,7 +142,6 @@ def _build_app_dict(self, request: HttpRequest, label=None):
 
         for model, model_admin in models.items():
             app_label = model._meta.app_label
-
             has_module_perms = model_admin.has_module_permission(request)
             if not has_module_perms:
                 continue

src/country_workspace/workspaces/templatetags/workspace_list.py

@@ -124,7 +124,7 @@ def make_qs_param(t, n):
             o_list_primary.insert(0, make_qs_param(new_order_type, i))
 
         yield {
-            "text": text,
+            "text": text.replace("flex_fields__", ""),
             "sortable": True,
             "sorted": is_sorted,
             "ascending": order_type == "asc",

tests/conftest.py

@@ -34,7 +34,10 @@ def pytest_addoption(parser):
 
 def pytest_configure(config):
     if not config.option.enable_selenium and ("selenium" not in getattr(config.option, "markexpr")):
-        setattr(config.option, "markexpr", "not selenium")
+        if config.option.markexpr:
+            config.option.markexpr += " and not selenium"
+        else:
+            config.option.markexpr = "not selenium"
     os.environ.update(DJANGO_SETTINGS_MODULE="country_workspace.config.settings")
     os.environ.setdefault("STATIC_URL", "/static/")
     os.environ.setdefault("MEDIA_ROOT", "/tmp/static/")

tests/extras/testutils/factories/__init__.py

@@ -6,7 +6,7 @@
 from pytest_factoryboy import register
 
 from .base import AutoRegisterModelFactory, TAutoRegisterModelFactory, factories_registry
-from .batch import BatchFactory  # noqa
+from .batch import BatchFactory, CountryBatchFactory  # noqa
 from .django_celery_beat import PeriodicTaskFactory  # noqa
 from .household import CountryHouseholdFactory, HouseholdFactory  # noqa
 from .individual import CountryIndividualFactory, IndividualFactory  # noqa

tests/extras/testutils/factories/batch.py

@@ -3,21 +3,31 @@
 import factory
 
 from country_workspace.models.batch import Batch
+from country_workspace.workspaces.models import CountryBatch
 
 from .base import AutoRegisterModelFactory
+from .program import ProgramFactory
 from .user import UserFactory
 
-# from .office import OfficeFactory
-# from .program import ProgramFactory
-
 
 class BatchFactory(AutoRegisterModelFactory):
     imported_by = factory.SubFactory(UserFactory)
     import_date = factory.LazyFunction(timezone.now)
     name = factory.Sequence(lambda n: f"Batch {n}")
 
-    country_office = factory.SubFactory("testutils.factories.OfficeFactory")
-    program = factory.SubFactory("testutils.factories.ProgramFactory")
+    # country_office = factory.SubFactory("testutils.factories.OfficeFactory")
+    program = factory.SubFactory(ProgramFactory)
 
     class Meta:
         model = Batch
+
+    @classmethod
+    def _create(cls, model_class, *args, **kwargs):
+        kwargs["country_office"] = kwargs["program"].country_office
+        return super()._create(model_class, *args, **kwargs)
+
+
+class CountryBatchFactory(BatchFactory):
+    class Meta:
+        model = CountryBatch
+        django_get_or_create = ("name",)