feat(api): add categories directory

backend/api/category_views.py

@@ -0,0 +1,41 @@
+"""Read-only active-workspace category directory."""
+
+from django.utils.decorators import method_decorator
+
+from categories.services import CategoryAccessDenied, get_categories_for_workspace
+from workspaces.context import WorkspaceContextError, resolve_active_workspace_context
+from workspaces.permissions import WorkspacePermissionDenied, can_perform_operational_work, require_workspace_permission
+
+from .auth_views import JsonMethodView, require_api_auth
+from .http import json_data, json_error
+
+
+READ_FIELDS = ("public_id", "name", "description", "default_deductible", "monthly_budget", "status")
+
+
+def _selectable_only(request):
+    if not request.GET:
+        return False, None
+    if set(request.GET) != {"selectable_only"} or len(request.GET.getlist("selectable_only")) != 1:
+        return None, json_error("invalid_request", status=400)
+    value = request.GET["selectable_only"]
+    if value not in ("true", "false"):
+        return None, json_error("invalid_request", status=400)
+    return value == "true", None
+
+
+class CategoryListView(JsonMethodView):
+    @method_decorator(require_api_auth)
+    def get(self, request):
+        selectable_only, error_response = _selectable_only(request)
+        if error_response is not None:
+            return error_response
+        try:
+            context = resolve_active_workspace_context(request)
+            require_workspace_permission(context.membership, can_perform_operational_work)
+            categories = get_categories_for_workspace(context, selectable_only=selectable_only).order_by("name", "pk")
+        except (CategoryAccessDenied, WorkspacePermissionDenied):
+            return json_error("permission_denied", status=403)
+        except WorkspaceContextError:
+            return json_error("workspace_required", status=400)
+        return json_data({"items": list(categories.values(*READ_FIELDS))})

backend/api/tests.py

@@ -10,6 +10,8 @@
 from django.utils import timezone
 
 from clients.models import Client as ClientModel
+from categories.models import Category
+from categories.services import create_category
 from proposals.models import Proposal
 from projects.models import Project
 from projects.services import archive_project, convert_accepted_proposal, transition_project
@@ -1710,3 +1712,108 @@ def test_cursor_store_prunes_expired_records_and_caps_live_entries(self, mocked_
         self.assertEqual(len(cursors), CURSOR_MAX_ENTRIES)
         self.assertNotIn("expired", cursors)
         self.assertNotIn("0", cursors)
+
+
+class CategoryApiTests(TestCase):
+    def setUp(self):
+        self.user = get_user_model().objects.create_user(
+            email="categories-api@example.com", password="correct-horse-battery-staple"
+        )
+        self.workspace = Workspace.objects.create(name="Category Studio", slug="category-api-studio")
+        self.membership = Membership.objects.create(
+            workspace=self.workspace, user=self.user, role=Membership.Role.OWNER
+        )
+        self.context = ActiveWorkspaceContext(self.workspace, self.membership)
+
+    def authenticate(self, *, expires_at=1_000_100, workspace=None):
+        self.client.force_login(self.user)
+        session = self.client.session
+        session["api.auth_expires_at"] = expires_at
+        if workspace is not None:
+            session["workspaces.active_workspace_public_id"] = str(workspace.public_id)
+        session.save()
+
+    def make_category(self, name, *, context=None, status=Category.Status.ACTIVE, monthly_budget=Decimal("12.50")):
+        return create_category(
+            context or self.context, name=name, description=f"Description for {name}",
+            default_deductible=True, monthly_budget=monthly_budget, status=status,
+        )
+
+    @patch("api.auth_views.time.time", return_value=1_000_000)
+    def test_requires_live_session_and_current_active_membership_context(self, mocked_time):
+        self.assertEqual(self.client.get("/api/v1/categories/").status_code, 401)
+        self.authenticate(expires_at=999_999, workspace=self.workspace)
+        self.assertEqual(self.client.get("/api/v1/categories/").status_code, 401)
+        self.authenticate()
+        self.assertEqual(self.client.get("/api/v1/categories/").json(), {"error": {"code": "workspace_required"}})
+        foreign = Workspace.objects.create(name="Foreign Category", slug="foreign-category-api")
+        self.authenticate(workspace=foreign)
+        self.assertEqual(self.client.get("/api/v1/categories/").json(), {"error": {"code": "workspace_required"}})
+        self.authenticate(workspace=self.workspace)
+        with allow_membership_writes():
+            self.membership.delete()
+        self.assertEqual(self.client.get("/api/v1/categories/").json(), {"error": {"code": "workspace_required"}})
+
+    @patch("api.auth_views.time.time", return_value=1_000_000)
+    def test_only_owner_and_operational_roles_may_read(self, mocked_time):
+        self.authenticate(workspace=self.workspace)
+        for role, expected in ((Membership.Role.ADMINISTRATIVE, 403), (Membership.Role.OPERATIONAL, 200), (Membership.Role.OWNER, 200)):
+            with allow_membership_writes():
+                Membership.objects.filter(pk=self.membership.pk).update(role=role)
+            self.assertEqual(self.client.get("/api/v1/categories/").status_code, expected)
+
+    @patch("api.auth_views.time.time", return_value=1_000_000)
+    def test_projection_is_ordered_tenant_scoped_and_does_not_write(self, mocked_time):
+        self.make_category("Zeta", monthly_budget=None)
+        first = self.make_category("Alpha", monthly_budget=Decimal("125.50"))
+        foreign_workspace = Workspace.objects.create(name="Foreign Category", slug="foreign-category-list")
+        foreign_membership = Membership.objects.create(workspace=foreign_workspace, user=self.user, role=Membership.Role.OWNER)
+        foreign = self.make_category("Foreign", context=ActiveWorkspaceContext(foreign_workspace, foreign_membership))
+        before = list(Category.objects.values_list("pk", "name", "status", "monthly_budget"))
+        self.authenticate(workspace=self.workspace)
+
+        response = self.client.get("/api/v1/categories/")
+
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response["Cache-Control"], "no-store")
+        items = response.json()["data"]["items"]
+        self.assertEqual([item["name"] for item in items], ["Alpha", "Zeta"])
+        self.assertEqual(items[0]["public_id"], str(first.public_id))
+        self.assertNotIn(str(foreign.public_id), {item["public_id"] for item in items})
+        self.assertEqual(set(items[0]), {"public_id", "name", "description", "default_deductible", "monthly_budget", "status"})
+        self.assertEqual(items[0]["monthly_budget"], "125.50")
+        self.assertIsNone(items[1]["monthly_budget"])
+        self.assertEqual(list(Category.objects.values_list("pk", "name", "status", "monthly_budget")), before)
+
+    @patch("api.auth_views.time.time", return_value=1_000_000)
+    def test_selectable_only_exact_values_follow_domain_filter(self, mocked_time):
+        active = self.make_category("Active")
+        inactive = self.make_category("Inactive", status=Category.Status.INACTIVE)
+        self.authenticate(workspace=self.workspace)
+
+        default = self.client.get("/api/v1/categories/")
+        unfiltered = self.client.get("/api/v1/categories/?selectable_only=false")
+        selectable = self.client.get("/api/v1/categories/?selectable_only=true")
+
+        self.assertEqual({item["public_id"] for item in default.json()["data"]["items"]}, {str(active.public_id), str(inactive.public_id)})
+        self.assertEqual({item["public_id"] for item in unfiltered.json()["data"]["items"]}, {str(active.public_id), str(inactive.public_id)})
+        self.assertEqual(selectable.json()["data"]["items"], [
+            {"public_id": str(active.public_id), "name": "Active", "description": "Description for Active", "default_deductible": True, "monthly_budget": "12.50", "status": "ACTIVE"}
+        ])
+
+    @patch("api.auth_views.time.time", return_value=1_000_000)
+    def test_rejects_invalid_query_contract_and_non_get_methods_as_json(self, mocked_time):
+        self.authenticate(workspace=self.workspace)
+        for path in (
+            "/api/v1/categories/?workspace=ignored",
+            "/api/v1/categories/?selectable_only=yes",
+            "/api/v1/categories/?selectable_only=true&selectable_only=false",
+            "/api/v1/categories/?selectable_only=true&extra=value",
+        ):
+            response = self.client.get(path)
+            self.assertEqual(response.status_code, 400)
+            self.assertEqual(response.json(), {"error": {"code": "invalid_request"}})
+        response = self.client.post("/api/v1/categories/")
+        self.assertEqual(response.status_code, 405)
+        self.assertEqual(response.json(), {"error": {"code": "method_not_allowed"}})
+        self.assertEqual(response["Allow"], "GET, HEAD, OPTIONS")

backend/api/urls.py

@@ -2,6 +2,7 @@
 
 from .auth_views import SessionLoginView, SessionLogoutView, SessionView
 from .client_views import ClientListView
+from .category_views import CategoryListView
 from .invoice_views import InvoiceListView
 from .fiscal_views import FiscalConfigurationView
 from .notification_views import NotificationListView
@@ -22,6 +23,7 @@
     path("notifications/", NotificationListView.as_view(), name="notification-list"),
     path("payments/", PaymentListView.as_view(), name="payment-list"),
     path("clients/", ClientListView.as_view(), name="client-list"),
+    path("categories/", CategoryListView.as_view(), name="category-list"),
     path("services/", ServiceListView.as_view(), name="service-list"),
     path("proposals/", ProposalListView.as_view(), name="proposal-list"),
     path("projects/", ProjectListView.as_view(), name="project-list"),