Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| b92564e219 |
@@ -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 <Team Rocket>"
|
||||
|
||||
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)
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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.
|
||||
@@ -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<RootStore>) =>
|
||||
({
|
||||
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();
|
||||
});
|
||||
});
|
||||
Reference in New Issue
Block a user