DEV: fix program manager project detail access

projects/permissions.py

@@ -13,6 +13,30 @@
 from projects.models import Project
 
 
+def has_project_read_involvement(user, project: Project) -> bool:
+    """Shared restricted READ visibility; program/admin roles grant no WRITE rights."""
+    if not user or not user.is_authenticated:
+        return False
+
+    if user.is_superuser or user.is_staff or project.leader_id == user.id:
+        return True
+
+    if project.collaborator_set.filter(user_id=user.id).exists():
+        return True
+
+    if project.invite_set.filter(user_id=user.id).exists():
+        return True
+
+    return (
+        PartnerProgramProject.objects.filter(project_id=project.id)
+        .filter(
+            Q(partner_program__managers__id=user.id)
+            | Q(partner_program__experts__user_id=user.id)
+        )
+        .exists()
+    )
+
+
 class ProjectVisibilityPermission(BasePermission):
     """
     Ограничивает доступ к непубличным проектам.
@@ -31,7 +55,10 @@ def has_permission(self, request, view):
         if not project_id and view.__module__.startswith("projects."):
             project_id = view.kwargs.get("id") or view.kwargs.get("pk")
 
-        if not project_id and getattr(getattr(view, "queryset", None), "model", None) is Project:
+        if (
+            not project_id
+            and getattr(getattr(view, "queryset", None), "model", None) is Project
+        ):
             project_id = view.kwargs.get("pk")
 
         if not project_id:
@@ -58,29 +85,7 @@ def has_object_permission(self, request, view, obj):
     def _can_view_project(self, request, project: Project) -> bool:
         if project.is_public:
             return True
-
-        user = getattr(request, "user", None)
-        if not user or not user.is_authenticated:
-            return False
-
-        if user.is_superuser or user.is_staff:
-            return True
-
-        if project.leader_id == user.id:
-            return True
-
-        if project.collaborator_set.filter(user_id=user.id).exists():
-            return True
-
-        if project.invite_set.filter(user_id=user.id).exists():
-            return True
-
-        return PartnerProgramProject.objects.filter(
-            project_id=project.id,
-        ).filter(
-            Q(partner_program__managers__id=user.id)
-            | Q(partner_program__experts__user_id=user.id)
-        ).exists()
+        return has_project_read_involvement(getattr(request, "user", None), project)
 
 
 class IsProjectLeaderOrReadOnlyForNonDrafts(BasePermission):
@@ -123,7 +128,12 @@ def has_object_permission(self, request, view, obj):
 
 class HasInvolvementInProjectOrReadOnly(BasePermission):
     """
-    Allows access to read to everyone involved in the project, and to update only to project leader.
+    READ: published projects pass this gate; drafts use shared restricted visibility.
+
+    WRITE: preserve existing involvement rules (published: leader; draft: leader,
+    collaborator or invite). Program managers/experts and admins are read-only
+    unless they independently satisfy those existing write rules.
+    ProjectVisibilityPermission separately guards private projects.
     """
 
     def has_permission(self, request, view) -> bool:
@@ -132,6 +142,9 @@ def has_permission(self, request, view) -> bool:
         return False
 
     def has_object_permission(self, request, view, obj):
+        if request.method in SAFE_METHODS:
+            return not obj.draft or has_project_read_involvement(request.user, obj)
+
         if not obj.draft:
             # not a draft, read-only for everyone, write only for leader
             if (request.method in SAFE_METHODS) or (request.user == obj.leader):
@@ -195,9 +208,7 @@ def _prepare_exception_detail(
             datetime_finished + timedelta(seconds=self._SECONDS_AFTER_CANT_EDIT)
         )
         days_until_resolution: int = (
-            int(self._SECONDS_AFTER_CANT_EDIT / 60 / 60 / 24)
-            - days_from_end_program
-            - 1
+            int(self._SECONDS_AFTER_CANT_EDIT / 60 / 60 / 24) - days_from_end_program - 1
         )
         return {
             "program_name": program_profile.partner_program.name,
@@ -222,7 +233,9 @@ def has_permission(self, request, view):
             return False
 
         project_pk = view.kwargs.get("project_pk")
-        project_id = project_pk or view.kwargs.get("project_id") or request.data.get("project")
+        project_id = (
+            project_pk or view.kwargs.get("project_id") or request.data.get("project")
+        )
         if not project_id:
             return False
 
@@ -259,10 +272,14 @@ def has_permission(self, request, view):
 
         submission_deadline = program.get_project_submission_deadline()
         if submission_deadline and submission_deadline < timezone.now():
-            raise ValidationError({"partner_program_id": "Срок подачи проектов в программу завершён."})
+            raise ValidationError(
+                {"partner_program_id": "Срок подачи проектов в программу завершён."}
+            )
 
         if program.datetime_finished < timezone.now():
-            raise ValidationError({"partner_program_id": "Нельзя выбрать завершённую программу."})
+            raise ValidationError(
+                {"partner_program_id": "Нельзя выбрать завершённую программу."}
+            )
 
         if program.is_manager(request.user):
             return True

projects/serializers.py

@@ -261,11 +261,11 @@ class ProjectDetailSerializer(serializers.ModelSerializer):
     trl = serializers.IntegerField(required=False, allow_null=True)
 
     def get_partner_program(self, project):
-        try:
-            link = project.program_links.select_related("partner_program").get()
-            return PartnerProgramProjectSerializer(link).data
-        except PartnerProgramProject.DoesNotExist:
-            return None
+        """Keep the legacy singular contract: choose the earliest link by pk."""
+        link = (
+            project.program_links.select_related("partner_program").order_by("pk").first()
+        )
+        return PartnerProgramProjectSerializer(link).data if link is not None else None
 
     @classmethod
     def get_partner_program_tags(cls, project):

projects/tests/test_program_project_detail_access.py

@@ -0,0 +1,211 @@
+"""Program analytics links must open project detail without granting write access."""
+
+from django.contrib.contenttypes.models import ContentType
+from django.core.cache import cache
+from django.db import connection
+from django.test import TestCase
+from django.test.utils import CaptureQueriesContext
+from rest_framework.test import APIClient
+
+from invites.models import Invite
+from project_rates.tests.helpers import create_rate_expert
+from projects.serializers import ProjectDetailSerializer, PartnerProgramProjectSerializer
+
+from .helpers import (
+    create_collaborator,
+    create_partner_program,
+    create_project,
+    create_user,
+    link_project_to_program,
+)
+
+
+class ProgramProjectDetailAccessTests(TestCase):
+    @classmethod
+    def setUpTestData(cls):
+        cls.program = create_partner_program(is_competitive=True)
+        cls.other_program = create_partner_program(is_competitive=True)
+        cls.project = create_project(draft=True, is_public=False)
+        cls.collaborator = create_collaborator(cls.project).user
+        cls.link = link_project_to_program(cls.project, cls.program)
+        cls.manager = create_user(prefix="detail-manager")
+        cls.program.managers.add(cls.manager)
+        cls.expert = create_rate_expert(program=cls.program)
+        cls.other_manager = create_user(prefix="other-manager")
+        cls.other_program.managers.add(cls.other_manager)
+        cls.other_expert = create_rate_expert(program=cls.other_program)
+        cls.staff = create_user(prefix="detail-staff")
+        cls.staff.is_staff = True
+        cls.staff.save(update_fields=["is_staff"])
+        cls.superuser = create_user(prefix="detail-superuser")
+        cls.superuser.is_superuser = True
+        cls.superuser.save(update_fields=["is_superuser"])
+        cls.invited = create_user(prefix="detail-invited")
+        Invite.objects.create(project=cls.project, user=cls.invited)
+        cls.outsider = create_user(prefix="detail-outsider")
+
+    def setUp(self):
+        self.client = APIClient()
+        self.url = f"/projects/{self.project.pk}/"
+
+    def assert_readable(self, user):
+        self.client.force_authenticate(user)
+        response = self.client.get(self.url)
+        self.assertEqual(response.status_code, 200, response.data)
+        self.assertEqual(response.data["id"], self.project.pk)
+
+    def test_linked_manager_can_read_private_draft(self):
+        self.assert_readable(self.manager)
+
+    def test_linked_expert_can_read_private_draft(self):
+        self.assert_readable(self.expert)
+
+    def test_linked_manager_safe_methods_remain_read_only(self):
+        self.client.force_authenticate(self.manager)
+        for method in ("head", "options"):
+            with self.subTest(method=method):
+                self.assertEqual(getattr(self.client, method)(self.url).status_code, 200)
+
+    def test_staff_without_superuser_can_read_private_draft(self):
+        self.assertFalse(self.staff.is_superuser)
+        self.assert_readable(self.staff)
+
+    def test_superuser_without_staff_can_read_private_draft(self):
+        self.assertFalse(self.superuser.is_staff)
+        self.assert_readable(self.superuser)
+
+    def test_existing_project_roles_can_read_private_draft(self):
+        for user in (self.project.leader, self.collaborator, self.invited):
+            with self.subTest(user=user.pk):
+                self.assert_readable(user)
+
+    def test_unrelated_and_other_program_roles_cannot_read_private_draft(self):
+        for user in (self.outsider, self.other_manager, self.other_expert):
+            with self.subTest(user=user.pk):
+                self.client.force_authenticate(user)
+                self.assertEqual(self.client.get(self.url).status_code, 403)
+
+    def test_anonymous_cannot_read_private_draft(self):
+        self.assertIn(self.client.get(self.url).status_code, (401, 403))
+
+    def test_public_flag_does_not_expose_draft_to_outsiders(self):
+        self.project.is_public = True
+        self.project.save(update_fields=["is_public"])
+        for user in (None, self.outsider, self.other_manager, self.other_expert):
+            with self.subTest(user=user):
+                self.client.force_authenticate(user)
+                self.assertIn(self.client.get(self.url).status_code, (401, 403))
+        self.assert_readable(self.manager)
+        self.assert_readable(self.expert)
+
+    def test_published_private_and_public_visibility_is_preserved(self):
+        self.project.draft = False
+        self.project.save(update_fields=["draft"])
+        self.assert_readable(self.manager)
+        self.client.force_authenticate(self.outsider)
+        self.assertEqual(self.client.get(self.url).status_code, 403)
+        self.project.is_public = True
+        self.project.save(update_fields=["is_public"])
+        self.assert_readable(self.outsider)
+        self.assert_readable(None)
+
+    def test_read_roles_do_not_gain_write_access_to_drafts_or_published_projects(self):
+        original_name = self.project.name
+        for draft in (True, False):
+            self.project.draft = draft
+            self.project.save(update_fields=["draft"])
+            for user in (self.manager, self.expert, self.staff, self.superuser):
+                self.client.force_authenticate(user)
+                for method in ("patch", "put", "delete"):
+                    with self.subTest(draft=draft, user=user.pk, method=method):
+                        response = getattr(self.client, method)(
+                            self.url, {"name": "Unauthorized change"}, format="json"
+                        )
+                        self.assertEqual(response.status_code, 403, response.data)
+                        self.project.refresh_from_db()
+                        self.assertEqual(self.project.name, original_name)
+
+    def test_manager_opens_exact_project_from_attention_with_one_or_two_links(self):
+        self.client.force_authenticate(self.manager)
+        self.assertNotEqual(self.project.leader_id, self.manager.pk)
+        self.assertFalse(self.project.collaborator_set.filter(user=self.manager).exists())
+        self.assertFalse(self.project.invite_set.filter(user=self.manager).exists())
+        for count in (1, 2):
+            with self.subTest(program_links=count):
+                if count == 2:
+                    link_project_to_program(self.project, self.other_program)
+                attention = self.client.get(
+                    f"/programs/{self.program.pk}/manager-overview/projects-not-submitted/"
+                )
+                self.assertEqual(attention.status_code, 200, attention.data)
+                self.assertEqual(attention.data["count"], 1)
+                row = attention.data["results"][0]
+                self.assertEqual(row["program_project_id"], self.link.pk)
+                detail = self.client.get(f"/projects/{row['project']['id']}/")
+                self.assertEqual(detail.status_code, 200, detail.data)
+                self.assertEqual(detail.data["id"], self.project.pk)
+                self.assertEqual(
+                    detail.data["partner_program"]["program_link_id"], self.link.pk
+                )
+        self.assertFalse(self.project.collaborator_set.filter(user=self.manager).exists())
+        self.assertFalse(self.project.invite_set.filter(user=self.manager).exists())
+
+
+class ProjectDetailProgramLinkTests(TestCase):
+    @classmethod
+    def setUpTestData(cls):
+        cls.project = create_project(draft=False, is_public=True)
+        # Create programs in the opposite order to links: selection is by link pk.
+        cls.later_link_program = create_partner_program(name="Created first")
+        cls.first_link_program = create_partner_program(name="Linked first")
+
+    def setUp(self):
+        self.client = APIClient()
+        self.client.force_authenticate(self.project.leader)
+        self.url = f"/projects/{self.project.pk}/"
+
+    def detail(self):
+        response = self.client.get(self.url)
+        self.assertEqual(response.status_code, 200, response.data)
+        return response.data
+
+    def test_no_program_links_returns_null(self):
+        self.assertIsNone(self.detail()["partner_program"])
+
+    def test_single_program_link_preserves_legacy_contract(self):
+        link = link_project_to_program(self.project, self.first_link_program)
+        self.assertEqual(
+            self.detail()["partner_program"], PartnerProgramProjectSerializer(link).data
+        )
+
+    def test_multiple_links_return_earliest_link_deterministically(self):
+        first = link_project_to_program(self.project, self.first_link_program)
+        link_project_to_program(self.project, self.later_link_program, submitted=True)
+        for _ in range(2):
+            data = self.detail()["partner_program"]
+            self.assertEqual(data, PartnerProgramProjectSerializer(first).data)
+            self.assertEqual(data["program_id"], self.first_link_program.pk)
+            self.assertFalse(data["is_submitted"])
+
+    def test_single_link_serializer_uses_one_join_and_existing_field_queries(self):
+        link_project_to_program(self.project, self.first_link_program)
+        # Existing contract: joined link/program, program fields, field values.
+        with self.assertNumQueries(3):
+            ProjectDetailSerializer().get_partner_program(self.project)
+
+    def test_detail_queries_do_not_grow_with_number_of_program_links(self):
+        link_project_to_program(self.project, self.first_link_program)
+        self.detail()  # Warm only the existing View record and ContentType cache.
+        ContentType.objects.get_for_model(self.project)
+        cache.clear()
+        with CaptureQueriesContext(connection) as single:
+            self.detail()
+        for _ in range(7):
+            link_project_to_program(self.project, create_partner_program())
+        cache.clear()
+        with CaptureQueriesContext(connection) as multiple:
+            self.detail()
+        self.assertEqual(len(multiple), len(single))
+        # Only one selected program is serialized, irrespective of link count.
+        with self.assertNumQueries(3):
+            ProjectDetailSerializer().get_partner_program(self.project)