diff --git a/apps/api/plane/api/serializers/__init__.py b/apps/api/plane/api/serializers/__init__.py index d0278eb1415..2e6a1a8ccf1 100644 --- a/apps/api/plane/api/serializers/__init__.py +++ b/apps/api/plane/api/serializers/__init__.py @@ -68,3 +68,4 @@ ProjectMemberLiteAPISerializer, ) from .sticky import StickySerializer +from .page import PageAPISerializer diff --git a/apps/api/plane/api/serializers/page.py b/apps/api/plane/api/serializers/page.py new file mode 100644 index 00000000000..ea04aa637f7 --- /dev/null +++ b/apps/api/plane/api/serializers/page.py @@ -0,0 +1,114 @@ +# Copyright (c) 2023-present Plane Software, Inc. and contributors +# SPDX-License-Identifier: AGPL-3.0-only +# See the LICENSE file for details. + +from django.db.models import Q +from rest_framework import serializers + +from plane.app.serializers import PageSerializer +from plane.db.models import Page, ProjectPage +from plane.utils.content_validator import validate_html_content + + +class PageAPISerializer(PageSerializer): + """Public API representation for a project Page. + + The app serializer already owns Page creation, project linking, and label + updates. The public API adds the content fields that are otherwise handled + by the browser-only description endpoint and preserves HTML byte-for-byte. + """ + + description_html = serializers.CharField(allow_blank=True, trim_whitespace=False) + description_json = serializers.JSONField(read_only=True) + + class Meta(PageSerializer.Meta): + fields = [ + "id", + "name", + "description_html", + "description_json", + "owned_by", + "access", + "color", + "labels", + "parent", + "is_locked", + "archived_at", + "workspace", + "created_at", + "updated_at", + "created_by", + "updated_by", + "view_props", + "logo_props", + "external_id", + "external_source", + ] + read_only_fields = [ + "id", + "workspace", + "owned_by", + "created_at", + "updated_at", + "created_by", + "updated_by", + "archived_at", + "is_locked", + ] + + def validate_description_html(self, value): + if not value: + return value + + is_valid, error_message, sanitized_html = validate_html_content(value) + if not is_valid: + raise serializers.ValidationError(error_message) + return sanitized_html + + def validate(self, attrs): + attrs = super().validate(attrs) + project_id = self.context["project_id"] + workspace_slug = self.context["workspace_slug"] + + parent = attrs.get("parent") + if parent: + accessible_parent = ProjectPage.objects.filter( + project_id=project_id, + workspace__slug=workspace_slug, + page=parent, + deleted_at__isnull=True, + ).filter(Q(page__owned_by_id=self.context["owned_by_id"]) | Q(page__access=Page.PUBLIC_ACCESS)) + if not accessible_parent.exists(): + raise serializers.ValidationError({"parent": "The parent page is not accessible in this project."}) + if self.instance: + if parent.id == self.instance.id: + raise serializers.ValidationError({"parent": "A page cannot be its own parent."}) + + ancestor_id = parent.parent_id + visited_ids = {parent.id} + while ancestor_id: + if ancestor_id == self.instance.id or ancestor_id in visited_ids: + raise serializers.ValidationError({"parent": "The parent would create a page cycle."}) + visited_ids.add(ancestor_id) + ancestor_id = Page.objects.filter(pk=ancestor_id).values_list("parent_id", flat=True).first() + + labels = attrs.get("labels") + if labels and any(label.project_id != project_id for label in labels): + raise serializers.ValidationError({"labels": "All labels must belong to this project."}) + + return attrs + + def create(self, validated_data): + self.context["description_html"] = validated_data.pop("description_html") + self.context["description_json"] = {} + self.context["description_binary"] = None + return super().create(validated_data) + + def update(self, instance, validated_data): + if "description_html" in validated_data: + # Plane Live treats a non-empty Yjs binary as authoritative. Clear + # it so the next editor connection imports this PAT-authored HTML + # instead of restoring and later persisting stale editor state. + validated_data["description_binary"] = None + validated_data["description_json"] = {} + return super().update(instance, validated_data) diff --git a/apps/api/plane/api/urls/__init__.py b/apps/api/plane/api/urls/__init__.py index 4a202431bc7..ba5e6338d80 100644 --- a/apps/api/plane/api/urls/__init__.py +++ b/apps/api/plane/api/urls/__init__.py @@ -14,6 +14,7 @@ from .work_item import urlpatterns as work_item_patterns from .invite import urlpatterns as invite_patterns from .sticky import urlpatterns as sticky_patterns +from .page import urlpatterns as page_patterns urlpatterns = [ *asset_patterns, @@ -28,4 +29,5 @@ *work_item_patterns, *invite_patterns, *sticky_patterns, + *page_patterns, ] diff --git a/apps/api/plane/api/urls/page.py b/apps/api/plane/api/urls/page.py new file mode 100644 index 00000000000..ccf30bd6538 --- /dev/null +++ b/apps/api/plane/api/urls/page.py @@ -0,0 +1,21 @@ +# Copyright (c) 2023-present Plane Software, Inc. and contributors +# SPDX-License-Identifier: AGPL-3.0-only +# See the LICENSE file for details. + +from django.urls import path + +from plane.api.views import ProjectPageDetailAPIEndpoint, ProjectPageListCreateAPIEndpoint + + +urlpatterns = [ + path( + "workspaces//projects//pages/", + ProjectPageListCreateAPIEndpoint.as_view(http_method_names=["get", "post"]), + name="project-pages", + ), + path( + "workspaces//projects//pages//", + ProjectPageDetailAPIEndpoint.as_view(http_method_names=["get", "patch"]), + name="project-page-detail", + ), +] diff --git a/apps/api/plane/api/views/__init__.py b/apps/api/plane/api/views/__init__.py index 5e4660a7b2b..d05fcff3f7b 100644 --- a/apps/api/plane/api/views/__init__.py +++ b/apps/api/plane/api/views/__init__.py @@ -72,3 +72,4 @@ from .invite import WorkspaceInvitationsViewset from .sticky import StickyViewSet +from .page import ProjectPageDetailAPIEndpoint, ProjectPageListCreateAPIEndpoint diff --git a/apps/api/plane/api/views/issue.py b/apps/api/plane/api/views/issue.py index da9edc66d66..7b740069e06 100644 --- a/apps/api/plane/api/views/issue.py +++ b/apps/api/plane/api/views/issue.py @@ -1118,6 +1118,18 @@ class IssueLinkListCreateAPIEndpoint(BaseAPIView): permission_classes = [ProjectEntityPermission] use_read_replica = True + def get_scoped_issue(self): + return ( + Issue.issue_objects.select_related("project__workspace") + .filter( + workspace__slug=self.kwargs.get("slug"), + project_id=self.kwargs.get("project_id"), + project__archived_at__isnull=True, + pk=self.kwargs.get("issue_id"), + ) + .first() + ) + def get_queryset(self): return ( IssueLink.objects.filter(workspace__slug=self.kwargs.get("slug")) @@ -1159,6 +1171,9 @@ def get(self, request, slug, project_id, issue_id): Retrieve all links associated with a work item. """ + if self.get_scoped_issue() is None: + return Response({"error": "Issue not found"}, status=status.HTTP_404_NOT_FOUND) + return self.paginate( request=request, queryset=(self.get_queryset()), @@ -1193,13 +1208,19 @@ def post(self, request, slug, project_id, issue_id): Add a new external link to a work item with URL, title, and metadata. Automatically tracks link creation activity. """ + issue = self.get_scoped_issue() + if issue is None: + return Response({"error": "Issue not found"}, status=status.HTTP_404_NOT_FOUND) + serializer = IssueLinkCreateSerializer(data=request.data) if serializer.is_valid(): - serializer.save(project_id=project_id, issue_id=issue_id) + serializer.save( + project=issue.project, + issue_id=issue.id, + created_by_id=request.user.id, + ) crawl_work_item_link_title.delay(serializer.instance.id, serializer.instance.url) - link = IssueLink.objects.get(pk=serializer.instance.id) - link.created_by_id = request.data.get("created_by", request.user.id) - link.save(update_fields=["created_by"]) + link = serializer.instance issue_activity.delay( type="link.activity.created", requested_data=json.dumps(serializer.data, cls=DjangoJSONEncoder), diff --git a/apps/api/plane/api/views/page.py b/apps/api/plane/api/views/page.py new file mode 100644 index 00000000000..37a60f37325 --- /dev/null +++ b/apps/api/plane/api/views/page.py @@ -0,0 +1,200 @@ +# Copyright (c) 2023-present Plane Software, Inc. and contributors +# SPDX-License-Identifier: AGPL-3.0-only +# See the LICENSE file for details. + +from django.db import transaction +from django.db.models import Q +from rest_framework import status +from rest_framework.permissions import BasePermission, SAFE_METHODS +from rest_framework.response import Response +from drf_spectacular.types import OpenApiTypes +from drf_spectacular.utils import OpenApiParameter, OpenApiResponse, extend_schema + +from plane.api.serializers import PageAPISerializer +from plane.app.permissions import ROLE +from plane.bgtasks.page_transaction_task import page_transaction +from plane.db.models import Page, Project, ProjectMember +from plane.utils.openapi import CURSOR_PARAMETER, PER_PAGE_PARAMETER, create_paginated_response + +from .base import BaseAPIView + + +class ProjectPageAPIPermission(BasePermission): + """Require project membership and role-gate every PAT Page write.""" + + def has_permission(self, request, view): + role = ( + ProjectMember.objects.filter( + workspace__slug=view.kwargs.get("slug"), + project_id=view.kwargs.get("project_id"), + member=request.user, + is_active=True, + ) + .values_list("role", flat=True) + .first() + ) + if not role: + return False + if request.method in SAFE_METHODS: + return True + return role in [ROLE.ADMIN.value, ROLE.MEMBER.value] + + +class ProjectPageBaseAPIEndpoint(BaseAPIView): + serializer_class = PageAPISerializer + model = Page + permission_classes = [ProjectPageAPIPermission] + + def get_queryset(self): + pages = ( + Page.objects.filter( + workspace__slug=self.kwargs.get("slug"), + project_pages__project_id=self.kwargs.get("project_id"), + project_pages__deleted_at__isnull=True, + project_pages__project__project_projectmember__member=self.request.user, + project_pages__project__project_projectmember__is_active=True, + project_pages__project__archived_at__isnull=True, + ) + .filter(Q(owned_by=self.request.user) | Q(access=Page.PUBLIC_ACCESS)) + .defer("description_binary", "description_stripped") + .distinct() + ) + + if ProjectMember.objects.filter( + workspace__slug=self.workspace_slug, + project_id=self.project_id, + member=self.request.user, + role=ROLE.GUEST.value, + is_active=True, + project__guest_view_all_features=False, + ).exists(): + pages = pages.filter(owned_by=self.request.user) + + return pages + + def get_serializer_context(self): + return { + **super().get_serializer_context(), + "workspace_slug": self.workspace_slug, + "project_id": self.project_id, + "owned_by_id": self.request.user.id, + } + + +class ProjectPageListCreateAPIEndpoint(ProjectPageBaseAPIEndpoint): + """List and create project Pages through PAT authentication.""" + + @extend_schema( + operation_id="list_project_pages", + parameters=[ + CURSOR_PARAMETER, + PER_PAGE_PARAMETER, + OpenApiParameter("search", OpenApiTypes.STR, description="Filter Page names."), + OpenApiParameter( + "type", + OpenApiTypes.STR, + enum=["public", "private"], + description="Filter by Page access.", + ), + ], + responses={200: create_paginated_response(PageAPISerializer, "ProjectPage", "Paginated project Pages")}, + ) + def get(self, request, slug, project_id): + pages = self.get_queryset() + + page_type = request.query_params.get("type") + if page_type == "public": + pages = pages.filter(access=Page.PUBLIC_ACCESS) + elif page_type == "private": + pages = pages.filter(access=Page.PRIVATE_ACCESS) + + search = request.query_params.get("search") + if search: + pages = pages.filter(name__icontains=search) + + pages = pages.order_by("-created_at") + return self.paginate( + request=request, + queryset=pages, + on_results=lambda results: self.get_serializer(results, many=True).data, + ) + + @extend_schema( + operation_id="create_project_page", + request=PageAPISerializer, + responses={201: OpenApiResponse(response=PageAPISerializer, description="Created Page")}, + ) + def post(self, request, slug, project_id): + if not Project.objects.filter( + pk=project_id, + workspace__slug=slug, + archived_at__isnull=True, + ).exists(): + return Response({"error": "Project not found"}, status=status.HTTP_404_NOT_FOUND) + + serializer = self.get_serializer(data=request.data) + if not serializer.is_valid(): + return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) + + with transaction.atomic(): + page = serializer.save(created_by=request.user, updated_by=request.user) + transaction.on_commit( + lambda: page_transaction.delay( + new_description_html=page.description_html, + old_description_html=None, + page_id=page.id, + ), + robust=True, + ) + return Response(self.get_serializer(page).data, status=status.HTTP_201_CREATED) + + +class ProjectPageDetailAPIEndpoint(ProjectPageBaseAPIEndpoint): + """Retrieve and update a project Page through PAT authentication.""" + + @extend_schema( + operation_id="retrieve_project_page", + responses={200: OpenApiResponse(response=PageAPISerializer, description="Project Page")}, + ) + def get(self, request, slug, project_id, page_id): + page = self.get_queryset().get(pk=page_id) + return Response(self.get_serializer(page).data, status=status.HTTP_200_OK) + + @extend_schema( + operation_id="update_project_page", + request=PageAPISerializer, + responses={200: OpenApiResponse(response=PageAPISerializer, description="Updated Page")}, + ) + def patch(self, request, slug, project_id, page_id): + page = self.get_queryset().get(pk=page_id) + if page.is_locked: + return Response({"error": "Page is locked"}, status=status.HTTP_400_BAD_REQUEST) + + requested_access = request.data.get("access", page.access) + if page.access != requested_access and page.owned_by_id != request.user.id: + return Response( + {"error": "Access cannot be updated since this page is owned by someone else"}, + status=status.HTTP_400_BAD_REQUEST, + ) + + previous_description_html = page.description_html + serializer = self.get_serializer( + page, + data=request.data, + partial=True, + ) + if not serializer.is_valid(): + return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) + + with transaction.atomic(): + page = serializer.save(updated_by=request.user) + if "description_html" in request.data and page.description_html != previous_description_html: + transaction.on_commit( + lambda: page_transaction.delay( + new_description_html=page.description_html, + old_description_html=previous_description_html, + page_id=page.id, + ), + robust=True, + ) + return Response(self.get_serializer(page).data, status=status.HTTP_200_OK) diff --git a/apps/api/plane/tests/contract/api/test_pages.py b/apps/api/plane/tests/contract/api/test_pages.py new file mode 100644 index 00000000000..b4ea26317ba --- /dev/null +++ b/apps/api/plane/tests/contract/api/test_pages.py @@ -0,0 +1,608 @@ +# Copyright (c) 2023-present Plane Software, Inc. and contributors +# SPDX-License-Identifier: AGPL-3.0-only +# See the LICENSE file for details. + +from uuid import uuid4 + +import pytest +from django.utils import timezone +from rest_framework import status +from rest_framework.test import APIClient + +from plane.db.models import ( + APIToken, + Issue, + IssueLink, + Page, + Project, + ProjectMember, + ProjectPage, + State, + User, + Workspace, + WorkspaceMember, +) + + +EXACT_PAGE_HTML = """\n

Design notes

+
  • MUST preserve these bytes.
+
AB
+
{\"enabled\": true}
+""" + +UPDATED_PAGE_HTML = """

Updated

+

Leading and trailing whitespace are intentional.

""" + + +@pytest.fixture +def project(db, workspace, create_user): + project = Project.objects.create( + name="Pages Project", + identifier="PAGE", + workspace=workspace, + created_by=create_user, + ) + ProjectMember.objects.create( + project=project, + workspace=workspace, + member=create_user, + role=20, + is_active=True, + ) + return project + + +def _user_with_project_role(workspace, project, role): + unique_id = uuid4().hex[:8] + user = User.objects.create( + email=f"page-user-{unique_id}@plane.so", + username=f"page_user_{unique_id}", + first_name="Page", + last_name="User", + ) + WorkspaceMember.objects.create(workspace=workspace, member=user, role=role, is_active=True) + ProjectMember.objects.create( + workspace=workspace, + project=project, + member=user, + role=role, + is_active=True, + ) + token = APIToken.objects.create(user=user, label=f"Page API {unique_id}", token=f"page-token-{unique_id}") + client = APIClient() + client.credentials(HTTP_X_API_KEY=token.token) + return user, client + + +def _create_page(workspace, project, owner, *, name="Existing page", access=Page.PUBLIC_ACCESS, html="

x

"): + page = Page.objects.create( + workspace=workspace, + owned_by=owner, + name=name, + access=access, + description_html=html, + created_by=owner, + updated_by=owner, + ) + ProjectPage.objects.create( + workspace=workspace, + project=project, + page=page, + created_by=owner, + updated_by=owner, + ) + return page + + +def _list_url(workspace, project): + return f"/api/v1/workspaces/{workspace.slug}/projects/{project.id}/pages/" + + +def _detail_url(workspace, project, page): + return f"{_list_url(workspace, project)}{page.id}/" + + +def _create_issue(workspace, project, owner, *, name="Page review"): + state = State.objects.create( + name="Backlog", + workspace=workspace, + project=project, + group="backlog", + default=True, + ) + return Issue.objects.create( + name=name, + workspace=workspace, + project=project, + state=state, + created_by=owner, + ) + + +def _issue_links_url(workspace, project, issue): + return f"/api/v1/workspaces/{workspace.slug}/projects/{project.id}/work-items/{issue.id}/links/" + + +@pytest.mark.contract +class TestProjectPageAPI: + @pytest.mark.django_db + def test_create_get_update_preserves_exact_content(self, api_key_client, workspace, project, create_user): + create_response = api_key_client.post( + _list_url(workspace, project), + { + "name": "Design notes", + "description_html": EXACT_PAGE_HTML, + "description_json": {"untrusted": "ignored"}, + "access": Page.PUBLIC_ACCESS, + }, + format="json", + ) + + assert create_response.status_code == status.HTTP_201_CREATED + assert create_response.data["description_html"] == EXACT_PAGE_HTML + assert create_response.data["description_json"] == {} + page = Page.objects.get(pk=create_response.data["id"]) + assert page.description_html == EXACT_PAGE_HTML + assert page.owned_by_id == create_user.id + + get_response = api_key_client.get(_detail_url(workspace, project, page)) + assert get_response.status_code == status.HTTP_200_OK + assert get_response.data["description_html"] == EXACT_PAGE_HTML + assert get_response.data["description_json"] == {} + + update_response = api_key_client.patch( + _detail_url(workspace, project, page), + {"name": "Updated design notes", "description_html": UPDATED_PAGE_HTML}, + format="json", + ) + assert update_response.status_code == status.HTTP_200_OK, update_response.data + assert update_response.data["description_html"] == UPDATED_PAGE_HTML + page.refresh_from_db() + assert page.description_html == UPDATED_PAGE_HTML + + @pytest.mark.django_db + def test_list_is_project_scoped_and_searchable(self, api_key_client, workspace, project, create_user): + matching_page = _create_page(workspace, project, create_user, name="Agent handbook") + _create_page(workspace, project, create_user, name="Unrelated page") + + other_project = Project.objects.create( + name="Other Project", + identifier="OTHER", + workspace=workspace, + created_by=create_user, + ) + ProjectMember.objects.create( + workspace=workspace, + project=other_project, + member=create_user, + role=20, + is_active=True, + ) + _create_page(workspace, other_project, create_user, name="Agent handbook in another project") + + response = api_key_client.get(_list_url(workspace, project), {"search": "handbook"}) + + assert response.status_code == status.HTTP_200_OK + assert [item["id"] for item in response.data["results"]] == [matching_page.id] + + @pytest.mark.django_db + def test_page_id_cannot_cross_project_or_workspace_boundaries( + self, api_key_client, workspace, project, create_user + ): + other_project = Project.objects.create( + name="Other Project", + identifier="OTHER", + workspace=workspace, + created_by=create_user, + ) + ProjectMember.objects.create( + workspace=workspace, + project=other_project, + member=create_user, + role=20, + is_active=True, + ) + other_page = _create_page(workspace, other_project, create_user) + + project_response = api_key_client.get(_detail_url(workspace, project, other_page)) + assert project_response.status_code in (status.HTTP_403_FORBIDDEN, status.HTTP_404_NOT_FOUND) + + other_workspace = Workspace.objects.create( + name="Other Workspace", + slug=f"other-{uuid4().hex[:8]}", + owner=create_user, + ) + WorkspaceMember.objects.create(workspace=other_workspace, member=create_user, role=20, is_active=True) + other_workspace_project = Project.objects.create( + name="Other Workspace Project", + identifier="OWP", + workspace=other_workspace, + created_by=create_user, + ) + ProjectMember.objects.create( + workspace=other_workspace, + project=other_workspace_project, + member=create_user, + role=20, + is_active=True, + ) + other_workspace_page = _create_page(other_workspace, other_workspace_project, create_user) + + workspace_response = api_key_client.get(_detail_url(workspace, project, other_workspace_page)) + assert workspace_response.status_code in (status.HTTP_403_FORBIDDEN, status.HTTP_404_NOT_FOUND) + + @pytest.mark.django_db + def test_create_rejects_parent_from_another_project(self, api_key_client, workspace, project, create_user): + other_project = Project.objects.create( + name="Other Project", + identifier="OTHER", + workspace=workspace, + created_by=create_user, + ) + ProjectMember.objects.create( + workspace=workspace, + project=other_project, + member=create_user, + role=20, + is_active=True, + ) + other_page = _create_page(workspace, other_project, create_user) + + response = api_key_client.post( + _list_url(workspace, project), + { + "name": "Cross-project child", + "description_html": "

child

", + "parent": str(other_page.id), + }, + format="json", + ) + + assert response.status_code == status.HTTP_400_BAD_REQUEST + assert not Page.objects.filter(name="Cross-project child").exists() + + @pytest.mark.django_db + def test_member_can_write_but_guest_is_read_only(self, workspace, project, create_user): + page = _create_page(workspace, project, create_user) + _, member_client = _user_with_project_role(workspace, project, role=15) + _, guest_client = _user_with_project_role(workspace, project, role=5) + + member_create = member_client.post( + _list_url(workspace, project), + {"name": "Member page", "description_html": "

member

"}, + format="json", + ) + assert member_create.status_code == status.HTTP_201_CREATED, member_create.data + member_page = Page.objects.get(pk=member_create.data["id"]) + member_update = member_client.patch( + _detail_url(workspace, project, member_page), + {"name": "Member-updated page"}, + format="json", + ) + assert member_update.status_code == status.HTTP_200_OK, member_update.data + + Project.objects.filter(pk=project.id).update(guest_view_all_features=True) + guest_read = guest_client.get(_detail_url(workspace, project, page)) + assert guest_read.status_code == status.HTTP_200_OK + private_page = _create_page( + workspace, + project, + create_user, + name="Private page", + access=Page.PRIVATE_ACCESS, + ) + guest_private_read = guest_client.get(_detail_url(workspace, project, private_page)) + assert guest_private_read.status_code == status.HTTP_404_NOT_FOUND + guest_create = guest_client.post( + _list_url(workspace, project), + {"name": "Guest page", "description_html": "

guest

"}, + format="json", + ) + assert guest_create.status_code == status.HTTP_403_FORBIDDEN + guest_update = guest_client.patch( + _detail_url(workspace, project, page), + {"name": "Guest edit"}, + format="json", + ) + assert guest_update.status_code == status.HTTP_403_FORBIDDEN + + @pytest.mark.django_db + def test_guest_visibility_honors_project_feature_setting(self, workspace, project, create_user): + other_page = _create_page(workspace, project, create_user, name="Admin page") + guest, guest_client = _user_with_project_role(workspace, project, role=5) + own_page = _create_page(workspace, project, guest, name="Guest page") + + restricted_list = guest_client.get(_list_url(workspace, project)) + assert restricted_list.status_code == status.HTTP_200_OK + assert [item["id"] for item in restricted_list.data["results"]] == [own_page.id] + restricted_detail = guest_client.get(_detail_url(workspace, project, other_page)) + assert restricted_detail.status_code in ( + status.HTTP_403_FORBIDDEN, + status.HTTP_404_NOT_FOUND, + ) + + Project.objects.filter(pk=project.id).update(guest_view_all_features=True) + enabled_list = guest_client.get(_list_url(workspace, project)) + assert enabled_list.status_code == status.HTTP_200_OK + assert {item["id"] for item in enabled_list.data["results"]} == { + own_page.id, + other_page.id, + } + enabled_detail = guest_client.get(_detail_url(workspace, project, other_page)) + assert enabled_detail.status_code == status.HTTP_200_OK + + @pytest.mark.django_db + def test_soft_deleted_project_link_denies_page_access(self, api_key_client, workspace, project, create_user): + page = _create_page(workspace, project, create_user) + ProjectPage.objects.filter(project=project, page=page).update(deleted_at=timezone.now()) + + response = api_key_client.get(_detail_url(workspace, project, page)) + + assert response.status_code in (status.HTTP_403_FORBIDDEN, status.HTTP_404_NOT_FOUND) + + @pytest.mark.django_db + def test_invalid_pat_is_rejected(self, api_client, workspace, project): + api_client.credentials(HTTP_X_API_KEY="not-a-valid-token") + + response = api_client.get(_list_url(workspace, project)) + + assert response.status_code in (status.HTTP_401_UNAUTHORIZED, status.HTTP_403_FORBIDDEN) + + @pytest.mark.django_db + def test_html_is_sanitized_and_external_links_remain_usable(self, api_key_client, workspace, project, create_user): + response = api_key_client.post( + _list_url(workspace, project), + { + "name": "Linked page", + "description_html": ( + '

link

' + ), + }, + format="json", + ) + + assert response.status_code == status.HTTP_201_CREATED, response.data + stored_html = Page.objects.get(pk=response.data["id"]).description_html + assert 'href="https://example.com"' in stored_html + assert "onclick" not in stored_html + assert "agent update

"}, + format="json", + ) + + assert response.status_code == status.HTTP_200_OK, response.data + page.refresh_from_db() + assert page.description_html == "

agent update

" + assert page.description_binary is None + assert page.description_json == {} + + @pytest.mark.django_db + def test_page_lifecycle_fields_are_not_mass_assignable(self, api_key_client, workspace, project, create_user): + page = _create_page(workspace, project, create_user, name="Active page") + + response = api_key_client.patch( + _detail_url(workspace, project, page), + {"archived_at": "2026-09-07T12:00:00Z", "is_locked": True}, + format="json", + ) + + assert response.status_code == status.HTTP_200_OK + page.refresh_from_db() + assert page.archived_at is None + assert page.is_locked is False + + @pytest.mark.django_db + def test_guest_cannot_update_a_page_they_own(self, workspace, project): + guest, guest_client = _user_with_project_role(workspace, project, role=5) + page = _create_page(workspace, project, guest, name="Guest-owned page") + + response = guest_client.patch( + _detail_url(workspace, project, page), + {"name": "Guest edit"}, + format="json", + ) + + assert response.status_code == status.HTTP_403_FORBIDDEN + page.refresh_from_db() + assert page.name == "Guest-owned page" + + @pytest.mark.django_db + def test_create_rejects_an_archived_project(self, api_key_client, workspace, project): + Project.objects.filter(pk=project.id).update(archived_at=timezone.now()) + + response = api_key_client.post( + _list_url(workspace, project), + {"name": "Invisible page", "description_html": "

hidden

"}, + format="json", + ) + + assert response.status_code == status.HTTP_404_NOT_FOUND + assert not Page.objects.filter(name="Invisible page").exists() + + @pytest.mark.django_db + def test_update_rejects_parent_cycles(self, api_key_client, workspace, project, create_user): + parent = _create_page(workspace, project, create_user, name="Parent") + child = _create_page(workspace, project, create_user, name="Child") + Page.objects.filter(pk=child.id).update(parent=parent) + + self_parent = api_key_client.patch( + _detail_url(workspace, project, parent), + {"parent": str(parent.id)}, + format="json", + ) + assert self_parent.status_code == status.HTTP_400_BAD_REQUEST + + two_page_cycle = api_key_client.patch( + _detail_url(workspace, project, parent), + {"parent": str(child.id)}, + format="json", + ) + assert two_page_cycle.status_code == status.HTTP_400_BAD_REQUEST + parent.refresh_from_db() + assert parent.parent_id is None + + @pytest.mark.django_db + def test_delete_is_not_exposed(self, api_key_client, workspace, project, create_user): + page = _create_page(workspace, project, create_user) + + response = api_key_client.delete(_detail_url(workspace, project, page)) + + assert response.status_code == status.HTTP_405_METHOD_NOT_ALLOWED + assert Page.objects.filter(pk=page.id).exists() + + +@pytest.mark.contract +class TestProjectPageWorkItemLinks: + @pytest.mark.django_db + def test_create_and_list_link_preserves_exact_page_url(self, api_key_client, workspace, project, create_user): + page = _create_page(workspace, project, create_user, name="Linked design") + issue = _create_issue(workspace, project, create_user) + page_url = f"https://plane.example.test/{workspace.slug}/projects/{project.id}/pages/{page.id}" + + create_response = api_key_client.post( + _issue_links_url(workspace, project, issue), + {"title": page.name, "url": page_url}, + format="json", + ) + + assert create_response.status_code == status.HTTP_201_CREATED, create_response.data + assert create_response.data["title"] == page.name + assert create_response.data["url"] == page_url + link = IssueLink.objects.get(pk=create_response.data["id"]) + assert link.issue_id == issue.id + assert link.project_id == project.id + assert link.workspace_id == workspace.id + assert link.url == page_url + + list_response = api_key_client.get(_issue_links_url(workspace, project, issue)) + + assert list_response.status_code == status.HTTP_200_OK + assert [(item["title"], item["url"]) for item in list_response.data["results"]] == [(page.name, page_url)] + + collection_delete = api_key_client.delete(_issue_links_url(workspace, project, issue)) + assert collection_delete.status_code == status.HTTP_405_METHOD_NOT_ALLOWED + assert IssueLink.objects.filter(pk=link.id).exists() + + @pytest.mark.django_db + def test_archived_issue_rejects_link_list_and_create(self, api_key_client, workspace, project, create_user): + issue = _create_issue(workspace, project, create_user) + url = _issue_links_url(workspace, project, issue) + initial_response = api_key_client.post( + url, + {"title": "Existing link", "url": "https://plane.example.test/page/existing"}, + format="json", + ) + assert initial_response.status_code == status.HTTP_201_CREATED + + Issue.objects.filter(pk=issue.id).update(archived_at=timezone.now().date()) + + list_response = api_key_client.get(url) + create_response = api_key_client.post( + url, + {"title": "Archived link", "url": "https://plane.example.test/page/archived"}, + format="json", + ) + + assert list_response.status_code == status.HTTP_404_NOT_FOUND + assert create_response.status_code == status.HTTP_404_NOT_FOUND + assert IssueLink.objects.filter(issue=issue).count() == 1 + + @pytest.mark.django_db + def test_create_rejects_issue_from_another_project(self, api_key_client, workspace, project, create_user): + other_project = Project.objects.create( + name="Other Project", + identifier="OTHER", + workspace=workspace, + created_by=create_user, + ) + ProjectMember.objects.create( + workspace=workspace, + project=other_project, + member=create_user, + role=20, + is_active=True, + ) + other_issue = _create_issue(workspace, other_project, create_user) + + response = api_key_client.post( + _issue_links_url(workspace, project, other_issue), + {"title": "Cross-project page", "url": "https://plane.example.test/page/cross-project"}, + format="json", + ) + + assert response.status_code in (status.HTTP_400_BAD_REQUEST, status.HTTP_404_NOT_FOUND) + assert not IssueLink.objects.filter(issue=other_issue, project=project).exists() + list_response = api_key_client.get(_issue_links_url(workspace, project, other_issue)) + assert list_response.status_code == status.HTTP_404_NOT_FOUND + + @pytest.mark.django_db + def test_create_rejects_issue_from_another_workspace(self, api_key_client, workspace, project, create_user): + other_workspace = Workspace.objects.create( + name="Other Workspace", + slug=f"other-{uuid4().hex[:8]}", + owner=create_user, + ) + WorkspaceMember.objects.create( + workspace=other_workspace, + member=create_user, + role=20, + is_active=True, + ) + other_project = Project.objects.create( + name="Other Workspace Project", + identifier="OWP", + workspace=other_workspace, + created_by=create_user, + ) + ProjectMember.objects.create( + workspace=other_workspace, + project=other_project, + member=create_user, + role=20, + is_active=True, + ) + other_issue = _create_issue(other_workspace, other_project, create_user) + + response = api_key_client.post( + _issue_links_url(workspace, project, other_issue), + {"title": "Cross-workspace page", "url": "https://plane.example.test/page/cross-workspace"}, + format="json", + ) + + assert response.status_code in (status.HTTP_400_BAD_REQUEST, status.HTTP_404_NOT_FOUND) + assert not IssueLink.objects.filter(issue=other_issue, project=project).exists() + list_response = api_key_client.get(_issue_links_url(workspace, project, other_issue)) + assert list_response.status_code == status.HTTP_404_NOT_FOUND + + @pytest.mark.django_db + def test_link_authentication_and_roles(self, api_client, workspace, project, create_user): + issue = _create_issue(workspace, project, create_user) + url = _issue_links_url(workspace, project, issue) + + api_client.credentials(HTTP_X_API_KEY="not-a-valid-token") + invalid_pat_response = api_client.get(url) + assert invalid_pat_response.status_code in ( + status.HTTP_401_UNAUTHORIZED, + status.HTTP_403_FORBIDDEN, + ) + + _, guest_client = _user_with_project_role(workspace, project, role=5) + guest_read = guest_client.get(url) + assert guest_read.status_code == status.HTTP_200_OK + guest_write = guest_client.post( + url, + {"title": "Guest page", "url": "https://plane.example.test/page/guest"}, + format="json", + ) + assert guest_write.status_code == status.HTTP_403_FORBIDDEN