diff --git a/apps/api/plane/tests/test_models_project.py b/apps/api/plane/tests/test_models_project.py new file mode 100644 index 00000000000..0f09e9f7c0d --- /dev/null +++ b/apps/api/plane/tests/test_models_project.py @@ -0,0 +1,205 @@ +# Testing library/framework: pytest + pytest-django + factory_boy +import pytest +import pytz +from django.core.exceptions import ValidationError +from django.db import IntegrityError +from django.utils import timezone + +# Prefer existing FactoryBoy factories defined in the repo +from plane.tests.factories import WorkspaceFactory, UserFactory, ProjectFactory + +# Import models and helpers under test +from plane.db.models.project import ( + Project, + ProjectMember, + ProjectMemberInvite, + ProjectIdentifier, + ProjectDeployBoard, + ProjectPublicMember, + ROLE, + ROLE_CHOICES, + ProjectNetwork, + get_default_props, + get_default_preferences, + get_default_views, +) + +# FileAsset is optional in some environments; guard import +try: + from plane.db.models.asset import FileAsset +except Exception: + FileAsset = None + + +@pytest.mark.django_db +class TestProjectModel: + def test_identifier_is_stripped_and_uppercased_on_save(self): + ws = WorkspaceFactory() + proj = Project.objects.create( + name="Alpha", + identifier=" al-1 ", + workspace=ws, + ) + proj.refresh_from_db() + assert proj.identifier == "AL-1" + + def test_cover_image_url_resolution_and_precedence(self): + ws = WorkspaceFactory() + proj = ProjectFactory(workspace=ws) + + # No cover data + proj.cover_image = None + proj.cover_image_asset = None + assert proj.cover_image_url is None + + # Text cover image + proj.cover_image = "https://cdn.example.com/cover.jpg" + assert proj.cover_image_url == "https://cdn.example.com/cover.jpg" + + # FileAsset should take precedence when present + if FileAsset: + asset = FileAsset.objects.create(asset_url="https://assets.example.com/a.png") + proj.cover_image_asset = asset + assert proj.cover_image_url == "https://assets.example.com/a.png" + + def test___str___format(self): + ws = WorkspaceFactory(name="Team Rocket") + proj = ProjectFactory(name="Gamma", identifier="GMM", workspace=ws) + assert str(proj) == "Gamma " + + def test_timezone_default_and_choices_include_utc(self): + ws = WorkspaceFactory() + proj = ProjectFactory(workspace=ws, timezone="UTC") + assert proj.timezone == "UTC" + assert ("UTC", "UTC") in Project.TIMEZONE_CHOICES + assert "UTC" in pytz.common_timezones + + def test_archive_close_in_validators_bounds(self): + ws = WorkspaceFactory() + proj = Project(name="Delta", identifier="DLT", workspace=ws, archive_in=0, close_in=12) + # Valid + proj.full_clean() + + # Invalid negative + proj.archive_in = -1 + with pytest.raises(ValidationError): + proj.full_clean() + + # Invalid > 12 + proj.archive_in = 0 + proj.close_in = 13 + with pytest.raises(ValidationError): + proj.full_clean() + + def test_unique_identifier_and_name_per_workspace_with_soft_delete(self): + ws = WorkspaceFactory() + p1 = Project.objects.create(name="Echo", identifier="ECH", workspace=ws) + + # Duplicates while not soft-deleted -> IntegrityError + with pytest.raises(IntegrityError): + Project.objects.create(name="Echo", identifier="ECH2", workspace=ws) + with pytest.raises(IntegrityError): + Project.objects.create(name="Foxtrot", identifier="ECH", workspace=ws) + + # Soft-delete first, then allow duplicates for the constrained fields + p1.deleted_at = timezone.now() + p1.save(update_fields=["deleted_at"]) + + Project.objects.create(name="Echo", identifier="ECHX", workspace=ws) + Project.objects.create(name="Zulu", identifier="ECH", workspace=ws) + + def test_network_choices_and_enum(self): + assert Project.NETWORK_CHOICES == ((0, "Secret"), (2, "Public")) + assert ProjectNetwork.choices() == [(0, "Secret"), (2, "Public")] + + +@pytest.mark.django_db +class TestProjectBaseAndMembers: + def test_projectbasemodel_sets_workspace_on_save_for_invite(self): + ws = WorkspaceFactory() + proj = ProjectFactory(workspace=ws) + inv = ProjectMemberInvite.objects.create(project=proj, email="invitee@example.com", token="tok") + inv.refresh_from_db() + assert inv.workspace_id == proj.workspace_id + assert str(inv) == f"{proj.name} invitee@example.com {inv.accepted}" + + def test_projectmember_sort_order_initialization(self): + ws = WorkspaceFactory() + user = UserFactory() + proj_a = ProjectFactory(workspace=ws) + proj_b = ProjectFactory(workspace=ws) + + # First membership => default 65535 + m1 = ProjectMember.objects.create(project=proj_a, member=user) + assert pytest.approx(m1.sort_order) == 65535 + + # Second membership for same user in same workspace => smallest - 10000 + m2 = ProjectMember.objects.create(project=proj_b, member=user) + assert m2.sort_order == m1.sort_order - 10000 + + def test_projectmember_defaults_role_props_and_str(self): + ws = WorkspaceFactory() + user = UserFactory() + proj = ProjectFactory(workspace=ws) + + member = ProjectMember.objects.create(project=proj, member=user) + + # Role defaults + assert member.role == ROLE.GUEST.value == 5 + assert ROLE_CHOICES == ((20, "Admin"), (15, "Member"), (5, "Guest")) + + # Default props/preferences + assert member.view_props == get_default_props() + assert member.default_props == get_default_props() + assert member.preferences == get_default_preferences() + + assert str(member) == f"{user.email} <{proj.name}>" + + +@pytest.mark.django_db +class TestProjectIdentifierAndDeployBoard: + def test_project_identifier_unique_with_soft_delete(self): + ws = WorkspaceFactory() + proj = ProjectFactory(workspace=ws, identifier="MU", name="Mu") + pid1 = ProjectIdentifier.objects.create(project=proj, workspace=ws, name="MU") + + with pytest.raises(IntegrityError): + ProjectIdentifier.objects.create(project=proj, workspace=ws, name="MU") + + pid1.deleted_at = timezone.now() + pid1.save(update_fields=["deleted_at"]) + + # Now allowed after soft-delete + ProjectIdentifier.objects.create(project=proj, workspace=ws, name="MU") + + def test_deploy_board_defaults_and_str(self): + ws = WorkspaceFactory() + proj = ProjectFactory(workspace=ws) + board = ProjectDeployBoard.objects.create(project=proj) + + assert isinstance(board.anchor, str) and len(board.anchor) >= 32 + assert board.views == get_default_views() + assert str(board) == f"{board.anchor} <{proj.name}>" + + def test_get_default_views_contents(self): + assert get_default_views() == { + "list": True, + "kanban": True, + "calendar": True, + "gantt": True, + "spreadsheet": True, + } + + +@pytest.mark.django_db +class TestProjectPublicMember: + def test_unique_constraint_on_public_member(self): + ws = WorkspaceFactory() + user = UserFactory() + proj = ProjectFactory(workspace=ws) + + pm1 = ProjectPublicMember.objects.create(project=proj, member=user) + assert pm1.pk is not None + + with pytest.raises(IntegrityError): + ProjectPublicMember.objects.create(project=proj, member=user) \ No newline at end of file diff --git a/apps/api/plane/tests/test_permissions_base_decorator_behaviors.py b/apps/api/plane/tests/test_permissions_base_decorator_behaviors.py new file mode 100644 index 00000000000..82e7d2917ad --- /dev/null +++ b/apps/api/plane/tests/test_permissions_base_decorator_behaviors.py @@ -0,0 +1,213 @@ +import builtins +from types import SimpleNamespace +from unittest.mock import patch, MagicMock + +import pytest +from rest_framework.response import Response +from rest_framework import status + +# Import the decorator and ROLE enum from the provided module +from apps.api.plane.tests.test_permissions_base import allow_permission, ROLE + +# Helper: minimal request/user and dummy view +class DummyUser(SimpleNamespace): + pass + +class DummyRequest(SimpleNamespace): + pass + +def dummy_view(instance, request, *args, **kwargs): + return Response({"ok": True}, status=status.HTTP_200_OK) + +@pytest.fixture +def user(): + return DummyUser(id=1, username="u1") + +@pytest.fixture +def request(user): + return DummyRequest(user=user) + +@pytest.fixture +def view_instance(): + # For function-based views 'instance' is unused; for method-decorated views could be 'self' + return object() + +def _wrap_and_call(deco, request, view_instance=None, **kwargs): + wrapped = deco(dummy_view) + return wrapped(view_instance or object(), request, **kwargs) + +def make_exists_mock(result: bool): + # Returns a mock for model.objects.filter(...).exists() chain + exists_mock = MagicMock(return_value=result) + filter_mock = MagicMock(return_value=SimpleNamespace(exists=exists_mock)) + objects_mock = SimpleNamespace(filter=filter_mock) + return objects_mock, filter_mock, exists_mock + +# --- Tests --- + +def test_allows_when_creator_matches_model_pk_and_created_by(request, view_instance): + # If creator=True and model is supplied, creator match should short-circuit and allow regardless of roles + class DummyModel: + pass + + objects_mock, filter_mock, exists_mock = make_exists_mock(True) + with patch.object(DummyModel, "objects", objects_mock): + deco = allow_permission(allowed_roles=[ROLE.GUEST], level="PROJECT", creator=True, model=DummyModel) + resp = _wrap_and_call(deco, request, view_instance, pk=123, slug="ws", project_id=456) + + filter_mock.assert_called_once() + # Ensure query filters by id and created_by + assert filter_mock.call_args.kwargs.get("id") == 123 + assert filter_mock.call_args.kwargs.get("created_by") is request.user + exists_mock.assert_called_once() + assert isinstance(resp, Response) and resp.status_code == 200 and resp.data == {"ok": True} + +@pytest.mark.parametrize( + "allowed_roles_input", + [ + [ROLE.ADMIN, ROLE.MEMBER], # enum members + [20, 15], # numeric roles + [ROLE.ADMIN.value, ROLE.MEMBER.value], # explicit ints + ], +) +def test_workspace_level_allows_when_workspace_member_has_allowed_role(request, view_instance, allowed_roles_input): + # WorkspaceMember.exists() -> True should allow + WM_objects, WM_filter, WM_exists = make_exists_mock(True) + with patch("apps.api.plane.tests.test_permissions_base.WorkspaceMember") as WM, \ + patch("apps.api.plane.tests.test_permissions_base.ProjectMember") as PM: + WM.objects = WM_objects + # ProjectMember shouldn't be called at WORKSPACE level when allowed + PM.objects = SimpleNamespace(filter=MagicMock()) + + deco = allow_permission(allowed_roles=allowed_roles_input, level="WORKSPACE") + resp = _wrap_and_call(deco, request, view_instance, slug="acme", project_id=999, pk=1) + + WM_filter.assert_called_once() + assert WM_exists.called + assert resp.status_code == 200 + +def test_workspace_level_denies_when_no_allowed_role_or_inactive(request, view_instance): + # WorkspaceMember.exists() -> False should deny (no project fallback at WORKSPACE level) + WM_objects, WM_filter, WM_exists = make_exists_mock(False) + with patch("apps.api.plane.tests.test_permissions_base.WorkspaceMember") as WM: + WM.objects = WM_objects + deco = allow_permission(allowed_roles=[ROLE.ADMIN], level="WORKSPACE") + resp = _wrap_and_call(deco, request, view_instance, slug="acme", project_id=1, pk=1) + + assert resp.status_code == status.HTTP_403_FORBIDDEN + assert resp.data == {"error": "You don't have the required permissions."} + +def test_project_level_allows_when_project_member_has_allowed_role(request, view_instance): + PM_objects, PM_filter, PM_exists = make_exists_mock(True) + with patch("apps.api.plane.tests.test_permissions_base.ProjectMember") as PM, \ + patch("apps.api.plane.tests.test_permissions_base.WorkspaceMember") as WM: + PM.objects = PM_objects + WM.objects = SimpleNamespace(filter=MagicMock()) # not used + deco = allow_permission(allowed_roles=[ROLE.MEMBER], level="PROJECT") + resp = _wrap_and_call(deco, request, view_instance, slug="acme", project_id=42, pk=7) + + PM_filter.assert_called_once() + assert PM_exists.called + assert resp.status_code == 200 + +def test_project_level_allows_when_workspace_admin_even_if_project_role_not_allowed(request, view_instance): + # First project role check -> False + PM_allowed_objects, PM_allowed_filter, PM_allowed_exists = make_exists_mock(False) + # Then project membership check (any role) -> True + PM_member_objects, PM_member_filter, PM_member_exists = make_exists_mock(True) + # Workspace admin check -> True + WM_admin_objects, WM_admin_filter, WM_admin_exists = make_exists_mock(True) + + def pm_filter_side_effect(*args, **kwargs): + # For role__in present, use allowed False; otherwise for membership True + if "role__in" in kwargs: + return PM_allowed_objects.filter(**kwargs) + return PM_member_objects.filter(**kwargs) + + with patch("apps.api.plane.tests.test_permissions_base.ProjectMember") as PM, \ + patch("apps.api.plane.tests.test_permissions_base.WorkspaceMember") as WM: + PM.objects = SimpleNamespace(filter=MagicMock(side_effect=pm_filter_side_effect)) + WM.objects = WM_admin_objects + + deco = allow_permission(allowed_roles=[ROLE.GUEST], level="PROJECT") + resp = _wrap_and_call(deco, request, view_instance, slug="acme", project_id=123, pk=9) + + # Confirm both the membership and admin checks participated + assert resp.status_code == 200 + assert WM_admin_filter.called + assert PM_allowed_filter.called or PM.objects.filter.called + +def test_project_level_denies_when_not_allowed_role_not_member_or_not_workspace_admin(request, view_instance): + # All checks -> False + PM_allowed_objects, PM_allowed_filter, PM_allowed_exists = make_exists_mock(False) + PM_member_objects, PM_member_filter, PM_member_exists = make_exists_mock(False) + WM_admin_objects, WM_admin_filter, WM_admin_exists = make_exists_mock(False) + + def pm_filter_side_effect(*args, **kwargs): + if "role__in" in kwargs: + return PM_allowed_objects.filter(**kwargs) + return PM_member_objects.filter(**kwargs) + + with patch("apps.api.plane.tests.test_permissions_base.ProjectMember") as PM, \ + patch("apps.api.plane.tests.test_permissions_base.WorkspaceMember") as WM: + PM.objects = SimpleNamespace(filter=MagicMock(side_effect=pm_filter_side_effect)) + WM.objects = WM_admin_objects + + deco = allow_permission(allowed_roles=[ROLE.ADMIN], level="PROJECT") + resp = _wrap_and_call(deco, request, view_instance, slug="acme", project_id=99, pk=3) + + assert resp.status_code == status.HTTP_403_FORBIDDEN + assert resp.data == {"error": "You don't have the required permissions."} + +def test_roles_accepts_mixed_enum_and_int_inputs(request, view_instance): + # Ensure conversion to values works: mix enum and ints + PM_objects, PM_filter, PM_exists = make_exists_mock(True) + with patch("apps.api.plane.tests.test_permissions_base.ProjectMember") as PM, \ + patch("apps.api.plane.tests.test_permissions_base.WorkspaceMember") as WM: + PM.objects = PM_objects + WM.objects = SimpleNamespace(filter=MagicMock()) + deco = allow_permission(allowed_roles=[ROLE.MEMBER, 5], level="PROJECT") + resp = _wrap_and_call(deco, request, view_instance, slug="acme", project_id=1, pk=1) + + assert resp.status_code == 200 + # Verify role__in passed as ints + passed_roles = PM_filter.call_args.kwargs.get("role__in") + assert isinstance(passed_roles, list) + assert all(isinstance(r, int) for r in passed_roles) + +def test_creator_flag_without_model_does_not_error_and_falls_back_to_role_checks(request, view_instance): + # creator=True but no model should not crash; proceed to role checks which we set to allow + PM_objects, PM_filter, PM_exists = make_exists_mock(True) + with patch("apps.api.plane.tests.test_permissions_base.ProjectMember") as PM, \ + patch("apps.api.plane.tests.test_permissions_base.WorkspaceMember") as WM: + PM.objects = PM_objects + WM.objects = SimpleNamespace(filter=MagicMock()) + deco = allow_permission(allowed_roles=[ROLE.ADMIN], level="PROJECT", creator=True, model=None) + resp = _wrap_and_call(deco, request, view_instance, slug="acme", project_id=11, pk=22) + + assert resp.status_code == 200 + +def test_missing_kwargs_results_in_403_in_project_level(request, view_instance): + # If required kwargs like slug or project_id are missing, filter will get KeyError before queries; + # The decorator accesses kwargs[...] so simulate missing and expect KeyError -> treat as failure handled by pytest + PM_objects, PM_filter, PM_exists = make_exists_mock(False) + with patch("apps.api.plane.tests.test_permissions_base.ProjectMember") as PM, \ + patch("apps.api.plane.tests.test_permissions_base.WorkspaceMember") as WM: + PM.objects = PM_objects + WM.objects = SimpleNamespace(filter=MagicMock()) + + deco = allow_permission(allowed_roles=[ROLE.ADMIN], level="PROJECT") + with pytest.raises(KeyError): + _ = _wrap_and_call(deco, request, view_instance, pk=1) # slug/project_id missing + +def test_inactive_members_are_not_considered(request, view_instance): + # Ensure is_active=True is part of filters; we validate by inspecting call kwargs + PM_objects, PM_filter, PM_exists = make_exists_mock(True) + with patch("apps.api.plane.tests.test_permissions_base.ProjectMember") as PM, \ + patch("apps.api.plane.tests.test_permissions_base.WorkspaceMember") as WM: + PM.objects = PM_objects + WM.objects = SimpleNamespace(filter=MagicMock()) + deco = allow_permission(allowed_roles=[ROLE.MEMBER], level="PROJECT") + _ = _wrap_and_call(deco, request, view_instance, slug="acme", project_id=1, pk=1) + + assert PM_filter.call_args.kwargs.get("is_active") is True \ No newline at end of file diff --git a/apps/api/plane/tests/test_permissions_project_additional.py b/apps/api/plane/tests/test_permissions_project_additional.py new file mode 100644 index 00000000000..7fdc76bb67b --- /dev/null +++ b/apps/api/plane/tests/test_permissions_project_additional.py @@ -0,0 +1,278 @@ +import types +from unittest.mock import MagicMock, patch + +import pytest + +# Subject under test: permission classes defined in the provided file +from apps.api.plane.tests import test_permissions_project as perms + + +def _req(user_is_anonymous: bool, method: str = "GET"): + return types.SimpleNamespace( + user=types.SimpleNamespace(is_anonymous=user_is_anonymous), + method=method, + ) + + +def _view(workspace_slug="ws-1", project_id=1, project_identifier=None): + v = types.SimpleNamespace( + workspace_slug=workspace_slug, + project_id=project_id, + ) + if project_identifier is not None: + v.project_identifier = project_identifier + return v + + +def _qs_with_exists(result: bool): + qs = MagicMock() + qs.exists.return_value = result + # Any nested filter(...) should return the same qs unless overridden in a test + qs.filter.return_value = qs + return qs + + +@pytest.mark.parametrize("method", ["GET", "HEAD", "OPTIONS"]) +def test_project_base_permission_safe_methods_allowed_for_active_workspace_member(method): + request = _req(False, method) + view = _view() + + with patch.object(perms.WorkspaceMember.objects, "filter") as wm_filter: + wm_filter.return_value = _qs_with_exists(True) + + assert perms.ProjectBasePermission().has_permission(request, view) is True + + wm_filter.assert_called_with( + workspace__slug=view.workspace_slug, + member=request.user, + is_active=True, + ) + + +@pytest.mark.parametrize("method", ["GET", "HEAD", "OPTIONS"]) +def test_project_base_permission_safe_methods_denied_for_non_member(method): + request = _req(False, method) + view = _view() + + with patch.object(perms.WorkspaceMember.objects, "filter") as wm_filter: + wm_filter.return_value = _qs_with_exists(False) + + assert perms.ProjectBasePermission().has_permission(request, view) is False + + +def test_project_base_permission_post_requires_workspace_admin_or_member(): + request = _req(False, "POST") + view = _view() + + with patch.object(perms.WorkspaceMember.objects, "filter") as wm_filter: + qs = _qs_with_exists(True) + wm_filter.return_value = qs + + assert perms.ProjectBasePermission().has_permission(request, view) is True + + # Validate role__in and other filters passed + called_kwargs = wm_filter.call_args.kwargs + assert called_kwargs["workspace__slug"] == view.workspace_slug + assert called_kwargs["member"] == request.user + assert called_kwargs["is_active"] is True + assert "role__in" in called_kwargs + assert isinstance(called_kwargs["role__in"], (list, tuple)) + assert len(called_kwargs["role__in"]) == 2 # [ROLE.ADMIN.value, ROLE.MEMBER.value] + + +def test_project_base_permission_post_denied_when_not_in_workspace(): + request = _req(False, "POST") + view = _view() + + with patch.object(perms.WorkspaceMember.objects, "filter") as wm_filter: + wm_filter.return_value = _qs_with_exists(False) + + assert perms.ProjectBasePermission().has_permission(request, view) is False + + +@pytest.mark.parametrize("method", ["PUT", "PATCH", "DELETE"]) +def test_project_base_permission_non_safe_granted_if_project_admin(method): + request = _req(False, method) + view = _view() + + with patch.object(perms.ProjectMember.objects, "filter") as pm_filter: + base_qs = _qs_with_exists(False) + admin_qs = _qs_with_exists(True) # admin exists + base_qs.filter.return_value = admin_qs + pm_filter.return_value = base_qs + + assert perms.ProjectBasePermission().has_permission(request, view) is True + + # Ensure admin role filter applied + admin_call = base_qs.filter.call_args + assert admin_call.kwargs == {"role": perms.ROLE.ADMIN.value} + + +@pytest.mark.parametrize("method", ["PUT", "PATCH", "DELETE"]) +def test_project_base_permission_non_safe_granted_if_project_member_and_workspace_admin(method): + request = _req(False, method) + view = _view() + + with patch.object(perms.ProjectMember.objects, "filter") as pm_filter, \ + patch.object(perms.WorkspaceMember.objects, "filter") as wm_filter: + + base_qs = _qs_with_exists(True) # user is a project member + admin_qs = _qs_with_exists(False) # not a project admin + base_qs.filter.return_value = admin_qs + pm_filter.return_value = base_qs + + wm_filter.return_value = _qs_with_exists(True) # workspace admin + + assert perms.ProjectBasePermission().has_permission(request, view) is True + + # Validate workspace admin filter call includes role=ROLE.ADMIN.value + called_kwargs = wm_filter.call_args.kwargs + assert called_kwargs["workspace__slug"] == view.workspace_slug + assert called_kwargs["member"] == request.user + assert called_kwargs["role"] == perms.ROLE.ADMIN.value + assert called_kwargs["is_active"] is True + + +@pytest.mark.parametrize("method", ["PUT", "PATCH", "DELETE"]) +def test_project_base_permission_non_safe_denied_if_not_admin_and_not_workspace_admin(method): + request = _req(False, method) + view = _view() + + with patch.object(perms.ProjectMember.objects, "filter") as pm_filter, \ + patch.object(perms.WorkspaceMember.objects, "filter") as wm_filter: + + base_qs = _qs_with_exists(False) # not a project member (or inactive) + admin_qs = _qs_with_exists(False) + base_qs.filter.return_value = admin_qs + pm_filter.return_value = base_qs + + wm_filter.return_value = _qs_with_exists(False) + + assert perms.ProjectBasePermission().has_permission(request, view) is False + + +def test_project_base_permission_denies_anonymous_for_all_methods(): + for method in ["GET", "POST", "PATCH"]: + assert perms.ProjectBasePermission().has_permission(_req(True, method), _view()) is False + + +# ---- ProjectMemberPermission tests ---- + +@pytest.mark.parametrize("method", ["GET", "HEAD", "OPTIONS"]) +def test_project_member_permission_safe_methods_require_project_membership(method): + request = _req(False, method) + view = _view() + + with patch.object(perms.ProjectMember.objects, "filter") as pm_filter: + pm_filter.return_value = _qs_with_exists(True) + assert perms.ProjectMemberPermission().has_permission(request, view) is True + + pm_filter.return_value = _qs_with_exists(False) + assert perms.ProjectMemberPermission().has_permission(request, view) is False + + +def test_project_member_permission_post_requires_workspace_admin_or_member(): + request = _req(False, "POST") + view = _view() + + with patch.object(perms.WorkspaceMember.objects, "filter") as wm_filter: + wm_filter.return_value = _qs_with_exists(True) + assert perms.ProjectMemberPermission().has_permission(request, view) is True + + called_kwargs = wm_filter.call_args.kwargs + assert "role__in" in called_kwargs + assert len(called_kwargs["role__in"]) == 2 + + +@pytest.mark.parametrize("is_active", [True, False]) +def test_project_member_permission_write_requires_admin_or_member_and_active(is_active): + request = _req(False, "PATCH") + view = _view() + + with patch.object(perms.ProjectMember.objects, "filter") as pm_filter: + qs = _qs_with_exists(is_active) + pm_filter.return_value = qs + + result = perms.ProjectMemberPermission().has_permission(request, view) + assert result is is_active + + # Validate role__in and scoping filters + called_kwargs = pm_filter.call_args.kwargs + assert called_kwargs["workspace__slug"] == view.workspace_slug + assert called_kwargs["member"] == request.user + assert called_kwargs["project_id"] == view.project_id + assert "role__in" in called_kwargs + assert called_kwargs["is_active"] is True + + +def test_project_member_permission_denies_anonymous(): + assert perms.ProjectMemberPermission().has_permission(_req(True, "GET"), _view()) is False + + +# ---- ProjectEntityPermission tests ---- + +def test_project_entity_permission_safe_with_project_identifier_scopes_by_identifier(): + request = _req(False, "GET") + view = _view(project_identifier="PRJ-123") + + with patch.object(perms.ProjectMember.objects, "filter") as pm_filter: + pm_filter.return_value = _qs_with_exists(True) + assert perms.ProjectEntityPermission().has_permission(request, view) is True + + # Ensure identifier used (not project_id) + called_kwargs = pm_filter.call_args.kwargs + assert called_kwargs["project__identifier"] == "PRJ-123" + assert "project_id" not in called_kwargs + + +def test_project_entity_permission_safe_without_identifier_scopes_by_project_id(): + request = _req(False, "GET") + view = _view(project_identifier=None) + + with patch.object(perms.ProjectMember.objects, "filter") as pm_filter: + pm_filter.return_value = _qs_with_exists(True) + assert perms.ProjectEntityPermission().has_permission(request, view) is True + + called_kwargs = pm_filter.call_args.kwargs + assert called_kwargs["project_id"] == view.project_id + + +@pytest.mark.parametrize("method", ["POST", "PUT", "PATCH", "DELETE"]) +def test_project_entity_permission_write_requires_admin_or_member(method): + request = _req(False, method) + view = _view() + + with patch.object(perms.ProjectMember.objects, "filter") as pm_filter: + pm_filter.return_value = _qs_with_exists(True) + assert perms.ProjectEntityPermission().has_permission(request, view) is True + + called_kwargs = pm_filter.call_args.kwargs + assert "role__in" in called_kwargs + assert len(called_kwargs["role__in"]) == 2 + assert called_kwargs["is_active"] is True + + +def test_project_entity_permission_denies_anonymous(): + assert perms.ProjectEntityPermission().has_permission(_req(True, "GET"), _view()) is False + + +# ---- ProjectLitePermission tests ---- + +def test_project_lite_permission_denies_anonymous(): + assert perms.ProjectLitePermission().has_permission(_req(True, "GET"), _view()) is False + + +@pytest.mark.parametrize("exists", [True, False]) +def test_project_lite_permission_requires_project_membership(exists): + request = _req(False, "GET") + view = _view() + + with patch.object(perms.ProjectMember.objects, "filter") as pm_filter: + pm_filter.return_value = _qs_with_exists(exists) + assert perms.ProjectLitePermission().has_permission(request, view) is exists + + called_kwargs = pm_filter.call_args.kwargs + assert called_kwargs["workspace__slug"] == view.workspace_slug + assert called_kwargs["member"] == request.user + assert called_kwargs["project_id"] == view.project_id + assert called_kwargs["is_active"] is True \ No newline at end of file diff --git a/apps/api/plane/tests/test_views_project_base.py b/apps/api/plane/tests/test_views_project_base.py new file mode 100644 index 00000000000..96abab0fb04 --- /dev/null +++ b/apps/api/plane/tests/test_views_project_base.py @@ -0,0 +1,475 @@ +import json +from unittest import mock + +import pytest +from django.urls import reverse +from django.utils import timezone +from rest_framework import status +from rest_framework.test import APIClient, APIRequestFactory, force_authenticate + +# NOTE: Import paths may need adjustment based on the repo's layout. +# We bias for action by importing via common app paths; adapt as needed if tests fail. +from plane.app.models import ( + Project, + Workspace, + WorkspaceMember, + ProjectMember, + DeployBoard, + UserFavorite, + Intake, + State, + ProjectIdentifier, + User, +) +from plane.app.views import ( + ProjectViewSet, + ProjectArchiveUnarchiveEndpoint, + ProjectIdentifierEndpoint, + ProjectUserViewsEndpoint, + ProjectFavoritesViewSet, + ProjectPublicCoverImagesEndpoint, + DeployBoardViewSet, +) +from plane.app.permissions import ROLE + + +@pytest.mark.django_db +class TestProjectViewSetListDetail: + def setup_method(self): + self.factory = APIRequestFactory() + self.client = APIClient() + # Users + self.owner = User.objects.create_user(email="owner@example.com", password="x") + self.member = User.objects.create_user(email="member@example.com", password="x") + self.guest = User.objects.create_user(email="guest@example.com", password="x") + # Workspace and memberships + self.ws = Workspace.objects.create(name="Acme", slug="acme") + WorkspaceMember.objects.create( + workspace=self.ws, member=self.owner, role=ROLE.ADMIN.value, is_active=True + ) + WorkspaceMember.objects.create( + workspace=self.ws, member=self.member, role=ROLE.MEMBER.value, is_active=True + ) + WorkspaceMember.objects.create( + workspace=self.ws, member=self.guest, role=ROLE.GUEST.value, is_active=True + ) + # Projects + self.p1 = Project.objects.create(name="P1", identifier="P1", workspace=self.ws) + self.p2 = Project.objects.create(name="P2", identifier="P2", workspace=self.ws, network=2) + self.p3 = Project.objects.create(name="P3", identifier="P3", workspace=self.ws, archived_at=None) + # Project memberships + ProjectMember.objects.create(project=self.p1, workspace=self.ws, member=self.owner, role=ROLE.ADMIN.value, is_active=True) + ProjectMember.objects.create(project=self.p1, workspace=self.ws, member=self.member, role=ROLE.MEMBER.value, is_active=True) + ProjectMember.objects.create(project=self.p2, workspace=self.ws, member=self.owner, role=ROLE.ADMIN.value, is_active=True) + ProjectMember.objects.create(project=self.p3, workspace=self.ws, member=self.owner, role=ROLE.ADMIN.value, is_active=True) + # Favorites and DeployBoard anchor annotations coverage + UserFavorite.objects.create(user=self.owner, entity_type="project", entity_identifier=self.p1.id, project=self.p1, workspace=self.ws) + DeployBoard.objects.create(entity_name="project", entity_identifier=self.p1.id, project=self.p1, workspace=self.ws, anchor="p1-anchor") + + def _view(self, action: str): + return ProjectViewSet.as_view({ "get": action }) + + def test_list_detail_admin_sees_all_sorted(self): + request = self.factory.get(f"/workspaces/{self.ws.slug}/projects/list-detail") + force_authenticate(request, user=self.owner) + response = self._view("list_detail")(request, slug=self.ws.slug) + assert response.status_code == status.HTTP_200_OK + # Should include at least p1, p2, p3 + ids = [p["id"] for p in response.data] + assert self.p1.id in ids and self.p2.id in ids and self.p3.id in ids + + def test_list_detail_guest_only_memberships(self): + request = self.factory.get(f"/workspaces/{self.ws.slug}/projects/list-detail") + force_authenticate(request, user=self.guest) + response = self._view("list_detail")(request, slug=self.ws.slug) + assert response.status_code == status.HTTP_200_OK + # guest not a member of any project -> sees none + assert response.data == [] + + # Add guest to p1, ensure visibility + ProjectMember.objects.create(project=self.p1, workspace=self.ws, member=self.guest, role=ROLE.GUEST.value, is_active=True) + request = self.factory.get(f"/workspaces/{self.ws.slug}/projects/list-detail") + force_authenticate(request, user=self.guest) + response = self._view("list_detail")(request, slug=self.ws.slug) + ids = [p["id"] for p in response.data] + assert self.p1.id in ids + assert self.p2.id not in ids # guest not member of p2 + + def test_list_detail_member_sees_memberships_plus_network_2(self): + request = self.factory.get(f"/workspaces/{self.ws.slug}/projects/list-detail") + force_authenticate(request, user=self.member) + response = self._view("list_detail")(request, slug=self.ws.slug) + assert response.status_code == status.HTTP_200_OK + ids = [p["id"] for p in response.data] + # Member of p1, should also see p2 because network=2 + assert self.p1.id in ids and self.p2.id in ids + + def test_list_detail_fields_filtering(self): + request = self.factory.get(f"/workspaces/{self.ws.slug}/projects/list-detail?fields=id,name") + force_authenticate(request, user=self.owner) + response = self._view("list_detail")(request, slug=self.ws.slug) + assert response.status_code == status.HTTP_200_OK + assert set(response.data[0].keys()).issubset({"id","name"}) + + def test_list_detail_cursor_pagination_path(self, monkeypatch): + # Force paginate branch by providing per_page and cursor + request = self.factory.get(f"/workspaces/{self.ws.slug}/projects/list-detail?per_page=2&cursor=abc&order_by=-created_at") + force_authenticate(request, user=self.owner) + # Mock paginate to ensure it is invoked + with mock.patch.object(ProjectViewSet, "paginate", return_value=mock.sentinel.PAGED) as m: + response = self._view("list_detail")(request, slug=self.ws.slug) + assert response == mock.sentinel.PAGED + assert m.called + + +@pytest.mark.django_db +class TestProjectViewSetList: + def setup_method(self): + self.factory = APIRequestFactory() + self.ws = Workspace.objects.create(name="Acme", slug="acme") + self.admin = User.objects.create_user(email="admin@example.com", password="x") + WorkspaceMember.objects.create(workspace=self.ws, member=self.admin, role=ROLE.ADMIN.value, is_active=True) + self.p1 = Project.objects.create(name="P1", identifier="P1", workspace=self.ws) + self.p2 = Project.objects.create(name="P2", identifier="P2", workspace=self.ws, network=2) + ProjectMember.objects.create(project=self.p1, workspace=self.ws, member=self.admin, role=ROLE.ADMIN.value, is_active=True) + + def _view(self, action): + return ProjectViewSet.as_view({ "get": action }) + + def test_list_values_shape_and_inbox_view(self): + request = self.factory.get(f"/workspaces/{self.ws.slug}/projects") + force_authenticate(request, user=self.admin) + response = self._view("list")(request, slug=self.ws.slug) + assert response.status_code == status.HTTP_200_OK + assert isinstance(response.data, list) + item = response.data[0] + # Expect selected fields only + expected_subset = {"id","name","identifier","sort_order","member_role","inbox_view","network","created_at","updated_at"} + assert expected_subset.issubset(set(item.keys())) + + def test_list_member_filtering_includes_network_two(self): + member = User.objects.create_user(email="member@example.com", password="x") + WorkspaceMember.objects.create(workspace=self.ws, member=member, role=ROLE.MEMBER.value, is_active=True) + ProjectMember.objects.create(project=self.p1, workspace=self.ws, member=member, role=ROLE.MEMBER.value, is_active=True) + + request = self.factory.get(f"/workspaces/{self.ws.slug}/projects") + force_authenticate(request, user=member) + response = self._view("list")(request, slug=self.ws.slug) + ids = [p["id"] for p in response.data] + assert self.p1.id in ids + assert self.p2.id in ids # network=2 visible to members + + +@pytest.mark.django_db +class TestProjectViewSetRetrieveCreateUpdateDestroy: + def setup_method(self): + self.factory = APIRequestFactory() + self.ws = Workspace.objects.create(name="Acme", slug="acme") + self.admin = User.objects.create_user(email="admin@example.com", password="x") + self.member = User.objects.create_user(email="user@example.com", password="x") + WorkspaceMember.objects.create(workspace=self.ws, member=self.admin, role=ROLE.ADMIN.value, is_active=True) + WorkspaceMember.objects.create(workspace=self.ws, member=self.member, role=ROLE.MEMBER.value, is_active=True) + self.project = Project.objects.create(name="Proj", identifier="PROJ", workspace=self.ws) + ProjectMember.objects.create(project=self.project, workspace=self.ws, member=self.admin, role=ROLE.ADMIN.value, is_active=True) + + def _view(self, method_name, http_method="get"): + return ProjectViewSet.as_view({ http_method: method_name }) + + @mock.patch("plane.app.views.recent_visited_task.delay") + def test_retrieve_happy_path_and_404(self, mock_task): + # make member of project + ProjectMember.objects.create(project=self.project, workspace=self.ws, member=self.member, role=ROLE.MEMBER.value, is_active=True) + # OK + req = self.factory.get("/") + force_authenticate(req, user=self.member) + resp = self._view("retrieve")(req, slug=self.ws.slug, pk=str(self.project.id)) + assert resp.status_code == status.HTTP_200_OK + mock_task.assert_called_once() + # 404 when not a member + outsider = User.objects.create_user(email="out@example.com", password="x") + WorkspaceMember.objects.create(workspace=self.ws, member=outsider, role=ROLE.MEMBER.value, is_active=True) + req = self.factory.get("/") + force_authenticate(req, user=outsider) + resp = self._view("retrieve")(req, slug=self.ws.slug, pk=str(self.project.id)) + assert resp.status_code == status.HTTP_404_NOT_FOUND + + @mock.patch("plane.app.views.model_activity.delay") + def test_create_creates_default_states_and_members(self, mock_activity): + req = self.factory.post("/", data={"name": "NewProj", "identifier": "NP"}, format="json") + force_authenticate(req, user=self.admin) + resp = self._view("create", http_method="post")(req, slug=self.ws.slug) + assert resp.status_code == status.HTTP_201_CREATED + created_id = resp.data["id"] + # Defaults: states created + assert State.objects.filter(project_id=created_id, workspace=self.ws).count() >= 5 + # Memberships: admin added + assert ProjectMember.objects.filter(project_id=created_id, member=self.admin, role=ROLE.ADMIN.value).exists() + mock_activity.assert_called_once() + + @mock.patch("plane.app.views.model_activity.delay") + def test_partial_update_requires_permission_and_handles_archived(self, mock_activity): + # Non-admin project member should be forbidden + non_admin = self.member + ProjectMember.objects.create(project=self.project, workspace=self.ws, member=non_admin, role=ROLE.MEMBER.value, is_active=True) + req = self.factory.patch("/", data={"name": "X"}, format="json") + force_authenticate(req, user=non_admin) + resp = self._view("partial_update", http_method="patch")(req, slug=self.ws.slug, pk=str(self.project.id)) + assert resp.status_code == status.HTTP_403_FORBIDDEN + + # Archive project then attempt update -> 400 + self.project.archived_at = timezone.now() + self.project.save() + req = self.factory.patch("/", data={"name": "Y"}, format="json") + force_authenticate(req, user=self.admin) + resp = self._view("partial_update", http_method="patch")(req, slug=self.ws.slug, pk=str(self.project.id)) + assert resp.status_code == status.HTTP_400_BAD_REQUEST + + # Unarchive and enable inbox_view -> ensure Intake default created + self.project.archived_at = None + self.project.save() + req = self.factory.patch("/", data={"inbox_view": True}, format="json") + force_authenticate(req, user=self.admin) + resp = self._view("partial_update", http_method="patch")(req, slug=self.ws.slug, pk=str(self.project.id)) + assert resp.status_code == status.HTTP_200_OK + assert Intake.objects.filter(project=self.project, is_default=True).exists() + assert mock_activity.called + + @mock.patch("plane.app.views.webhook_activity.delay") + def test_destroy_deletes_related_and_respects_permission(self, mock_webhook): + # Non-admin cannot delete + req = self.factory.delete("/") + force_authenticate(req, user=self.member) + resp = self._view("destroy", http_method="delete")(req, slug=self.ws.slug, pk=str(self.project.id)) + assert resp.status_code == status.HTTP_403_FORBIDDEN + + # Admin delete removes deploy board and favorites + DeployBoard.objects.create(entity_name="project", entity_identifier=self.project.id, project=self.project, workspace=self.ws) + UserFavorite.objects.create(user=self.admin, entity_type="project", entity_identifier=self.project.id, project=self.project, workspace=self.ws) + req = self.factory.delete("/") + force_authenticate(req, user=self.admin) + resp = self._view("destroy", http_method="delete")(req, slug=self.ws.slug, pk=str(self.project.id)) + assert resp.status_code == status.HTTP_204_NO_CONTENT + assert not DeployBoard.objects.filter(project_id=self.project.id, workspace=self.ws).exists() + assert not UserFavorite.objects.filter(project_id=self.project.id, workspace=self.ws).exists() + mock_webhook.assert_called_once() + + +@pytest.mark.django_db +class TestProjectArchiveUnarchiveEndpoint: + def setup_method(self): + self.factory = APIRequestFactory() + self.ws = Workspace.objects.create(name="Acme", slug="acme") + self.admin = User.objects.create_user(email="admin@example.com", password="x") + WorkspaceMember.objects.create(workspace=self.ws, member=self.admin, role=ROLE.ADMIN.value, is_active=True) + self.project = Project.objects.create(name="Proj", identifier="PROJ", workspace=self.ws) + + def test_archive_sets_archived_at_and_removes_favorites(self): + fav = UserFavorite.objects.create(user=self.admin, entity_type="project", entity_identifier=self.project.id, project=self.project, workspace=self.ws) + view = ProjectArchiveUnarchiveEndpoint.as_view({"post": "post"}) + req = self.factory.post("/") + force_authenticate(req, user=self.admin) + resp = view(req, slug=self.ws.slug, project_id=str(self.project.id)) + assert resp.status_code == status.HTTP_200_OK + self.project.refresh_from_db() + assert self.project.archived_at is not None + assert not UserFavorite.objects.filter(pk=fav.pk).exists() + + def test_unarchive_clears_archived_at(self): + self.project.archived_at = timezone.now() + self.project.save() + view = ProjectArchiveUnarchiveEndpoint.as_view({"delete": "delete"}) + req = self.factory.delete("/") + force_authenticate(req, user=self.admin) + resp = view(req, slug=self.ws.slug, project_id=str(self.project.id)) + assert resp.status_code == status.HTTP_204_NO_CONTENT + self.project.refresh_from_db() + assert self.project.archived_at is None + + +@pytest.mark.django_db +class TestProjectIdentifierEndpoint: + def setup_method(self): + self.factory = APIRequestFactory() + self.ws = Workspace.objects.create(name="Acme", slug="acme") + self.admin = User.objects.create_user(email="admin@example.com", password="x") + WorkspaceMember.objects.create(workspace=self.ws, member=self.admin, role=ROLE.ADMIN.value, is_active=True) + + def test_get_requires_name(self): + view = ProjectIdentifierEndpoint.as_view({"get": "get"}) + req = self.factory.get("/", data={"name": ""}) + force_authenticate(req, user=self.admin) + resp = view(req, slug=self.ws.slug) + assert resp.status_code == status.HTTP_400_BAD_REQUEST + + def test_get_returns_existing_identifiers(self): + ProjectIdentifier.objects.create(name="API", workspace=self.ws) + view = ProjectIdentifierEndpoint.as_view({"get": "get"}) + req = self.factory.get("/", data={"name": "api"}) + force_authenticate(req, user=self.admin) + resp = view(req, slug=self.ws.slug) + assert resp.status_code == status.HTTP_200_OK + assert resp.data["exists"] >= 1 + assert any(i["name"] == "API" for i in resp.data["identifiers"]) + + def test_delete_validation_and_success(self): + view = ProjectIdentifierEndpoint.as_view({"delete": "delete"}) + # Missing name + req = self.factory.delete("/", data={"name": ""}, format="json") + force_authenticate(req, user=self.admin) + resp = view(req, slug=self.ws.slug) + assert resp.status_code == status.HTTP_400_BAD_REQUEST + + # Cannot delete if used by a project + proj = Project.objects.create(name="Proj", identifier="USED", workspace=self.ws) + req = self.factory.delete("/", data={"name": "used"}, format="json") + force_authenticate(req, user=self.admin) + resp = view(req, slug=self.ws.slug) + assert resp.status_code == status.HTTP_400_BAD_REQUEST + + # Can delete free identifier + ProjectIdentifier.objects.create(name="FREE", workspace=self.ws) + req = self.factory.delete("/", data={"name": "free"}, format="json") + force_authenticate(req, user=self.admin) + resp = view(req, slug=self.ws.slug) + assert resp.status_code == status.HTTP_204_NO_CONTENT + assert not ProjectIdentifier.objects.filter(name="FREE", workspace=self.ws).exists() + + +@pytest.mark.django_db +class TestProjectUserViewsEndpoint: + def setup_method(self): + self.factory = APIRequestFactory() + self.ws = Workspace.objects.create(name="Acme", slug="acme") + self.user = User.objects.create_user(email="u@example.com", password="x") + self.project = Project.objects.create(name="Proj", identifier="PROJ", workspace=self.ws) + + def test_post_forbidden_if_not_project_member(self): + view = ProjectUserViewsEndpoint.as_view({"post": "post"}) + req = self.factory.post("/", data={"view_props": {"x": 1}}, format="json") + force_authenticate(req, user=self.user) + resp = view(req, slug=self.ws.slug, project_id=str(self.project.id)) + assert resp.status_code == status.HTTP_403_FORBIDDEN + + def test_post_updates_member_props(self): + # Make user a member + ProjectMember.objects.create(project=self.project, workspace=self.ws, member=self.user, role=ROLE.MEMBER.value, is_active=True, view_props={"a": 1}, default_props={"b": 2}, preferences={"c": 3}, sort_order=5) + view = ProjectUserViewsEndpoint.as_view({"post": "post"}) + payload = { + "view_props": {"a": 9}, + "default_props": {"b": 8}, + "preferences": {"c": 7}, + "sort_order": 42, + } + req = self.factory.post("/", data=payload, format="json") + force_authenticate(req, user=self.user) + resp = view(req, slug=self.ws.slug, project_id=str(self.project.id)) + assert resp.status_code == status.HTTP_204_NO_CONTENT + pm = ProjectMember.objects.get(project=self.project, member=self.user) + assert pm.view_props == payload["view_props"] + assert pm.default_props == payload["default_props"] + assert pm.preferences == payload["preferences"] + assert pm.sort_order == 42 + + +@pytest.mark.django_db +class TestProjectFavoritesViewSet: + def setup_method(self): + self.factory = APIRequestFactory() + self.ws = Workspace.objects.create(name="Acme", slug="acme") + self.user = User.objects.create_user(email="u@example.com", password="x") + WorkspaceMember.objects.create(workspace=self.ws, member=self.user, role=ROLE.MEMBER.value, is_active=True) + self.project = Project.objects.create(name="Proj", identifier="PROJ", workspace=self.ws) + + def _view(self, method, http="post"): + return ProjectFavoritesViewSet.as_view({ http: method }) + + def test_create_and_destroy_favorite(self): + # Create + req = self.factory.post("/", data={"project": self.project.id}, format="json") + force_authenticate(req, user=self.user) + resp = self._view("create", "post")(req, slug=self.ws.slug) + assert resp.status_code == status.HTTP_204_NO_CONTENT + fav = UserFavorite.objects.get(user=self.user, project=self.project, entity_type="project") + + # Destroy + req = self.factory.delete("/") + force_authenticate(req, user=self.user) + resp = self._view("destroy", "delete")(req, slug=self.ws.slug, project_id=str(self.project.id)) + assert resp.status_code == status.HTTP_204_NO_CONTENT + assert not UserFavorite.objects.filter(pk=fav.pk).exists() + + +@pytest.mark.django_db +class TestProjectPublicCoverImagesEndpoint: + @mock.patch("plane.app.views.boto3.client") + def test_get_lists_public_images_and_handles_errors(self, mock_client): + # Mock S3 response + s3 = mock_client.return_value + s3.list_objects_v2.return_value = { + "Contents": [ + {"Key": "static/project-cover/img1.png"}, + {"Key": "static/project-cover/nested/"}, + {"Key": "static/project-cover/img2.jpg"}, + ] + } + req = APIRequestFactory().get("/public/project-covers") + resp = ProjectPublicCoverImagesEndpoint.as_view()(req) + assert resp.status_code == status.HTTP_200_OK + # Only files, ignore folders + assert len(resp.data) == 2 + assert all(k.endswith(("img1.png","img2.jpg")) for k in resp.data[0:2] or []) + + # Error path returns empty list + s3.list_objects_v2.side_effect = Exception("boom") + req = APIRequestFactory().get("/public/project-covers") + resp = ProjectPublicCoverImagesEndpoint.as_view()(req) + assert resp.status_code == status.HTTP_200_OK + assert resp.data == [] + + +@pytest.mark.django_db +class TestDeployBoardViewSet: + def setup_method(self): + self.factory = APIRequestFactory() + self.client = APIClient() + self.ws = Workspace.objects.create(name="Acme", slug="acme") + self.user = User.objects.create_user(email="u@example.com", password="x") + WorkspaceMember.objects.create(workspace=self.ws, member=self.user, role=ROLE.MEMBER.value, is_active=True) + self.project = Project.objects.create(name="Proj", identifier="PROJ", workspace=self.ws) + ProjectMember.objects.create(project=self.project, workspace=self.ws, member=self.user, role=ROLE.MEMBER.value, is_active=True) + + def _view(self, method, http="get"): + return DeployBoardViewSet.as_view({ http: method }) + + def test_list_returns_existing_board(self): + db = DeployBoard.objects.create(entity_name="project", entity_identifier=self.project.id, project=self.project, workspace=self.ws, view_props={"list": True}) + req = self.factory.get("/") + force_authenticate(req, user=self.user) + resp = self._view("list", "get")(req, slug=self.ws.slug, project_id=str(self.project.id)) + assert resp.status_code == status.HTTP_200_OK + assert resp.data["id"] == str(db.id) or resp.data["id"] == db.id + + def test_create_upserts_and_sets_flags(self): + payload = { + "is_comments_enabled": True, + "is_reactions_enabled": True, + "is_votes_enabled": True, + "intake": {"enabled": True}, + "views": {"list": True, "kanban": False, "calendar": True, "gantt": False, "spreadsheet": True}, + } + req = self.factory.post("/", data=payload, format="json") + force_authenticate(req, user=self.user) + resp = self._view("create", "post")(req, slug=self.ws.slug, project_id=str(self.project.id)) + assert resp.status_code == status.HTTP_200_OK + + db = DeployBoard.objects.get(entity_name="project", entity_identifier=self.project.id) + assert db.is_comments_enabled is True + assert db.is_reactions_enabled is True + assert db.is_votes_enabled is True + assert db.view_props == payload["views"] + assert db.intake == payload["intake"] + + +# Testing framework note: +# These tests are written for pytest with pytest-django and Django REST Framework's testing utilities (APIRequestFactory/APIClient). +# They follow repository conventions by mocking external dependencies (boto3, celery tasks) and exercising public interfaces of view classes. diff --git a/apps/web/core/store/user/base-permissions.store.spec.ts b/apps/web/core/store/user/base-permissions.store.spec.ts new file mode 100644 index 00000000000..e07c81cf4de --- /dev/null +++ b/apps/web/core/store/user/base-permissions.store.spec.ts @@ -0,0 +1,350 @@ +/** + * Tests for BaseUserPermissionStore + * Framework: Jest (TypeScript). We mock external services and constants. + * + * Focus: Validate computed helpers, allowPermissions logic, and async actions + * (fetch/leave workspace/project, join/leave project, permissions fetch). + * We create a minimal concrete subclass for the abstract method. + */ + +/* eslint-disable @typescript-eslint/no-explicit-any */ + +import { runInAction } from "mobx"; + +// Mock @plane/constants to control enums and navigation links deterministically +jest.mock("@plane/constants", () => { + enum EUserPermissions { + VIEWER = 1, + MEMBER = 2, + ADMIN = 3, + } + enum EUserPermissionsLevel { + WORKSPACE = "WORKSPACE", + PROJECT = "PROJECT", + } + const WORKSPACE_SIDEBAR_DYNAMIC_NAVIGATION_ITEMS_LINKS = [ + { key: "overview", access: [EUserPermissions.VIEWER, EUserPermissions.MEMBER, EUserPermissions.ADMIN] }, + { key: "projects", access: [EUserPermissions.MEMBER, EUserPermissions.ADMIN] }, + { key: "settings", access: [EUserPermissions.ADMIN] }, + ]; + return { + EUserPermissions, + EUserPermissionsLevel, + WORKSPACE_SIDEBAR_DYNAMIC_NAVIGATION_ITEMS_LINKS, + }; +}, { virtual: true }); + +// Provide minimal runtime enum for workspace roles used by the store +jest.mock("@plane/types", () => ({ + EUserWorkspaceRoles: { ADMIN: 3 }, +}), { virtual: true }); + +// Pull mocked constants/types into scope +import { + EUserPermissions, + EUserPermissionsLevel, + WORKSPACE_SIDEBAR_DYNAMIC_NAVIGATION_ITEMS_LINKS, +} from "@plane/constants"; +import { EUserWorkspaceRoles } from "@plane/types"; + +// Mock services that the store uses (as virtual modules to avoid resolver config) +const mockWorkspaceMemberMe = jest.fn(); +const mockGetWorkspaceUserProjectsRole = jest.fn(); + +jest.mock("@/plane-web/services/workspace.service", () => { + return { + WorkspaceService: jest.fn().mockImplementation(() => ({ + workspaceMemberMe: (...args: any[]) => mockWorkspaceMemberMe(...args), + getWorkspaceUserProjectsRole: (...args: any[]) => mockGetWorkspaceUserProjectsRole(...args), + })), + }; +}, { virtual: true }); + +const mockJoinProject = jest.fn(); +const mockLeaveProject = jest.fn(); +const mockLeaveWorkspace = jest.fn(); + +jest.mock("@/services/user.service", () => ({ + __esModule: true, + default: { + joinProject: (...args: any[]) => mockJoinProject(...args), + leaveProject: (...args: any[]) => mockLeaveProject(...args), + leaveWorkspace: (...args: any[]) => mockLeaveWorkspace(...args), + }, +}), { virtual: true }); + +const mockProjectMemberMe = jest.fn(); +jest.mock("@/services/project/project-member.service", () => ({ + __esModule: true, + default: { + projectMemberMe: (...args: any[]) => mockProjectMemberMe(...args), + }, +}), { virtual: true }); + +// Import after setting up jest.mocks +import type { RootStore } from "@/plane-web/store/root.store"; +import { BaseUserPermissionStore } from "./base-permissions.store"; + +// Minimal concrete subclass exposing the protected getProjectRole via the abstract API +class TestUserPermissionStore extends BaseUserPermissionStore { + getProjectRoleByWorkspaceSlugAndProjectId = (ws: string, pid: string) => this["getProjectRole"](ws, pid); +} + +const makeStore = (overrides?: Partial) => + ({ + router: { + workspaceSlug: "ws-1", + projectId: "p-1", + ...(overrides?.router as any), + }, + projectRoot: { + project: { + projectMap: {}, + }, + }, + ...(overrides as any), + } as RootStore); + +describe("BaseUserPermissionStore - computed helpers", () => { + let store: TestUserPermissionStore; + + beforeEach(() => { + jest.clearAllMocks(); + store = new TestUserPermissionStore(makeStore()); + }); + + test("workspaceInfoBySlug returns undefined for empty slug and for missing entry", () => { + expect(store.workspaceInfoBySlug("")).toBeUndefined(); + expect(store.workspaceInfoBySlug("ws-missing")).toBeUndefined(); + }); + + test("workspaceInfoBySlug returns stored member info", () => { + const member = { id: "u1", role: EUserPermissions.ADMIN } as any; + runInAction(() => { + store.workspaceUserInfo["ws-1"] = member; + }); + expect(store.workspaceInfoBySlug("ws-1")).toBe(member); + }); + + test("getWorkspaceRoleByWorkspaceSlug returns role or undefined", () => { + expect(store.getWorkspaceRoleByWorkspaceSlug("ws-NA")).toBeUndefined(); + runInAction(() => { + store.workspaceUserInfo["ws-1"] = { role: EUserPermissions.MEMBER } as any; + }); + expect(store.getWorkspaceRoleByWorkspaceSlug("ws-1")).toBe(EUserPermissions.MEMBER); + }); + + test("getProjectRoleByWorkspaceSlugAndProjectId: admin workspace role maps to ADMIN for any project", () => { + runInAction(() => { + store.workspaceUserInfo["ws-1"] = { role: EUserWorkspaceRoles.ADMIN } as any; + store.workspaceProjectsPermissions["ws-1"] = { "p-1": EUserPermissions.VIEWER }; + }); + expect(store.getProjectRoleByWorkspaceSlugAndProjectId("ws-1", "p-1")).toBe(EUserPermissions.ADMIN); + }); + + test("getProjectRoleByWorkspaceSlugAndProjectId: returns project role when not admin", () => { + runInAction(() => { + store.workspaceUserInfo["ws-1"] = { role: EUserPermissions.MEMBER } as any; + store.workspaceProjectsPermissions["ws-1"] = { "p-2": EUserPermissions.VIEWER }; + }); + expect(store.getProjectRoleByWorkspaceSlugAndProjectId("ws-1", "p-2")).toBe(EUserPermissions.VIEWER); + }); + + test("getProjectRolesByWorkspaceSlug reduces to available roles using internal computed", () => { + runInAction(() => { + store.workspaceUserInfo["ws-1"] = { role: EUserPermissions.MEMBER } as any; + store.workspaceProjectsPermissions["ws-1"] = { + "p-1": EUserPermissions.VIEWER, + "p-2": EUserPermissions.MEMBER, + }; + }); + const result = store.getProjectRolesByWorkspaceSlug("ws-1"); + expect(result).toEqual({ + "p-1": EUserPermissions.VIEWER, + "p-2": EUserPermissions.MEMBER, + }); + }); +}); + +describe("BaseUserPermissionStore - hasPageAccess", () => { + let store: TestUserPermissionStore; + + beforeEach(() => { + jest.clearAllMocks(); + store = new TestUserPermissionStore(makeStore()); + runInAction(() => { + store.workspaceUserInfo["ws-1"] = { role: EUserPermissions.MEMBER } as any; + }); + }); + + test("returns false for missing slug or key", () => { + expect(store.hasPageAccess("", "settings")).toBe(false); + expect(store.hasPageAccess("ws-1", "")).toBe(false); + }); + + test("returns false when key not found in navigation items", () => { + expect(store.hasPageAccess("ws-1", "unknown")).toBe(false); + }); + + test("grants access based on allowed roles", () => { + // overview allows VIEWER+ + expect(store.hasPageAccess("ws-1", "overview")).toBe(true); + // settings allows only ADMIN + expect(store.hasPageAccess("ws-1", "settings")).toBe(false); + runInAction(() => { + store.workspaceUserInfo["ws-1"] = { role: EUserPermissions.ADMIN } as any; + }); + expect(store.hasPageAccess("ws-1", "settings")).toBe(true); + }); + + test("uses allowPermissions under the hood with workspace-level", () => { + const entry = WORKSPACE_SIDEBAR_DYNAMIC_NAVIGATION_ITEMS_LINKS.find((i) => i.key === "projects") as any; + expect(entry.access).toContain(EUserPermissions.MEMBER); + expect(store.hasPageAccess("ws-1", "projects")).toBe(true); + }); +}); + +describe("BaseUserPermissionStore - allowPermissions", () => { + let store: TestUserPermissionStore; + + beforeEach(() => { + jest.clearAllMocks(); + store = new TestUserPermissionStore(makeStore()); + }); + + test("returns false when current role not available", () => { + expect( + store.allowPermissions([EUserPermissions.ADMIN], EUserPermissionsLevel.WORKSPACE, "ws-1") + ).toBe(false); + }); + + test("accepts workspace role when included", () => { + runInAction(() => { + store.workspaceUserInfo["ws-1"] = { role: EUserPermissions.ADMIN } as any; + }); + expect( + store.allowPermissions([EUserPermissions.ADMIN], EUserPermissionsLevel.WORKSPACE, "ws-1") + ).toBe(true); + }); + + test("accepts project role when included", () => { + runInAction(() => { + store.workspaceUserInfo["ws-1"] = { role: EUserPermissions.MEMBER } as any; + store.workspaceProjectsPermissions["ws-1"] = { "p-1": EUserPermissions.VIEWER }; + }); + expect( + store.allowPermissions([EUserPermissions.VIEWER], EUserPermissionsLevel.PROJECT, "ws-1", "p-1") + ).toBe(true); + }); + + test("uses router fallback when workspaceSlug/projectId not provided", () => { + runInAction(() => { + store.workspaceUserInfo["ws-1"] = { role: EUserPermissions.ADMIN } as any; + }); + expect(store.allowPermissions([EUserPermissions.ADMIN], EUserPermissionsLevel.WORKSPACE)).toBe(true); + }); + + test("invokes onPermissionAllowed callback when provided", () => { + runInAction(() => { + store.workspaceUserInfo["ws-1"] = { role: EUserPermissions.ADMIN } as any; + }); + const cb = jest.fn().mockReturnValue(true); + expect( + store.allowPermissions([EUserPermissions.ADMIN], EUserPermissionsLevel.WORKSPACE, "ws-1", undefined, cb) + ).toBe(true); + expect(cb).toHaveBeenCalledTimes(1); + }); + + test("handles string role by parsing to number", () => { + runInAction(() => { + store.workspaceUserInfo["ws-1"] = { role: "2" as any } as any; // MEMBER + }); + expect( + store.allowPermissions([EUserPermissions.MEMBER], EUserPermissionsLevel.WORKSPACE, "ws-1") + ).toBe(true); + }); +}); + +describe("BaseUserPermissionStore - actions", () => { + let store: TestUserPermissionStore; + + beforeEach(() => { + jest.clearAllMocks(); + store = new TestUserPermissionStore(makeStore()); + }); + + test("fetchUserWorkspaceInfo stores response and toggles loader", async () => { + const resp = { id: "me", role: EUserPermissions.MEMBER } as any; + mockWorkspaceMemberMe.mockResolvedValueOnce(resp); + + const p = store.fetchUserWorkspaceInfo("ws-1"); + expect(store.loader).toBe(true); + await expect(p).resolves.toBe(resp); + expect(store.workspaceUserInfo["ws-1"]).toEqual(resp); + expect(store.loader).toBe(false); + }); + + test("fetchUserWorkspaceInfo propagates error and resets loader", async () => { + mockWorkspaceMemberMe.mockRejectedValueOnce(new Error("boom")); + await expect(store.fetchUserWorkspaceInfo("ws-1")).rejects.toThrow("boom"); + expect(store.loader).toBe(false); + }); + + test("leaveWorkspace clears related maps", async () => { + runInAction(() => { + store.workspaceUserInfo["ws-1"] = { role: EUserPermissions.ADMIN } as any; + store.projectUserInfo["ws-1"] = { "p-1": { id: "pm" } as any }; + store.workspaceProjectsPermissions["ws-1"] = { "p-1": EUserPermissions.ADMIN }; + }); + mockLeaveWorkspace.mockResolvedValueOnce(undefined); + await store.leaveWorkspace("ws-1"); + expect(store.workspaceUserInfo["ws-1"]).toBeUndefined(); + expect(store.projectUserInfo["ws-1"]).toBeUndefined(); + expect(store.workspaceProjectsPermissions["ws-1"]).toBeUndefined(); + }); + + test("fetchUserProjectInfo stores membership and permission", async () => { + const membership = { id: "pm-1", role: EUserPermissions.MEMBER } as any; + mockProjectMemberMe.mockResolvedValueOnce(membership); + await store.fetchUserProjectInfo("ws-1", "p-1"); + expect(store.projectUserInfo["ws-1"]["p-1"]).toEqual(membership); + expect(store.workspaceProjectsPermissions["ws-1"]["p-1"]).toBe(EUserPermissions.MEMBER); + }); + + test("fetchUserProjectPermissions stores entire map", async () => { + const map = { "p-1": EUserPermissions.VIEWER, "p-2": EUserPermissions.ADMIN } as any; + mockGetWorkspaceUserProjectsRole.mockResolvedValueOnce(map); + await store.fetchUserProjectPermissions("ws-1"); + expect(store.workspaceProjectsPermissions["ws-1"]).toEqual(map); + }); + + test("joinProject stores role equal to workspace member role or MEMBER by default", async () => { + mockJoinProject.mockResolvedValueOnce({ ok: true }); + // no workspace role -> default MEMBER + await store.joinProject("ws-1", "p-1"); + expect(store.workspaceProjectsPermissions["ws-1"]["p-1"]).toBe(EUserPermissions.MEMBER); + + // with workspace role ADMIN + runInAction(() => { + store.workspaceUserInfo["ws-1"] = { role: EUserPermissions.ADMIN } as any; + }); + mockJoinProject.mockResolvedValueOnce({ ok: true }); + await store.joinProject("ws-1", "p-2"); + expect(store.workspaceProjectsPermissions["ws-1"]["p-2"]).toBe(EUserPermissions.ADMIN); + }); + + test("leaveProject removes entries from permissions, membership and project map", async () => { + const rs = makeStore(); + (rs as any).projectRoot.project.projectMap["p-1"] = { id: "p-1" }; + store = new TestUserPermissionStore(rs); + runInAction(() => { + store.workspaceProjectsPermissions["ws-1"] = { "p-1": EUserPermissions.VIEWER }; + store.projectUserInfo["ws-1"] = { "p-1": { id: "pm-1" } as any }; + }); + mockLeaveProject.mockResolvedValueOnce(undefined); + await store.leaveProject("ws-1", "p-1"); + expect(store.workspaceProjectsPermissions["ws-1"]?.["p-1"]).toBeUndefined(); + expect(store.projectUserInfo["ws-1"]?.["p-1"]).toBeUndefined(); + expect((rs as any).projectRoot.project.projectMap["p-1"]).toBeUndefined(); + }); +}); \ No newline at end of file