Browse Source

feat: add profiles app, dynamic forms views, serializers, and openapi schema

main
sina_sajjadi 2 weeks ago
parent
commit
95a48860cf
  1. 29
      apps/account/admin/user.py
  2. 39
      apps/dynamic_forms/permissions.py
  3. 9
      apps/dynamic_forms/serializers/form_serializers.py
  4. 156
      apps/dynamic_forms/tests/test_phase9_forms.py
  5. 51
      apps/dynamic_forms/views/form_views.py
  6. 34
      apps/profiles/admin.py
  7. 89
      apps/profiles/management/commands/seed_quick_access_links.py
  8. 38
      apps/profiles/migrations/0003_quickaccesslink.py
  9. 3
      apps/profiles/models/__init__.py
  10. 83
      apps/profiles/models/quick_access_link.py
  11. 8
      apps/profiles/permissions.py
  12. 5
      apps/profiles/serializers/__init__.py
  13. 58
      apps/profiles/serializers/institution_serializers.py
  14. 176
      apps/profiles/tests/test_quick_access_links.py
  15. 9
      apps/profiles/urls.py
  16. 9
      apps/profiles/views/__init__.py
  17. 181
      apps/profiles/views/institution_views.py
  18. BIN
      dump.rdb
  19. 15588
      openapi.json
  20. 1850
      schema.yml

29
apps/account/admin/user.py

@ -69,6 +69,11 @@ class UserAdminChangeForm(UserChangeForm):
required=False,
label=_("Role & Permissions Overview")
)
can_create_forms = forms.BooleanField(
required=False,
label=_("Can Create Dynamic Forms"),
help_text=_("Toggles permission for creating dynamic forms & surveys for this user.")
)
class Meta(UserChangeForm.Meta):
model = User
@ -79,17 +84,30 @@ class UserAdminChangeForm(UserChangeForm):
self.fields['email'].required = True
if 'custom_permissions' in self.fields:
self.fields['custom_permissions'].widget.instance = self.instance
if 'can_create_forms' in self.fields and self.instance and self.instance.pk:
self.fields['can_create_forms'].initial = self.instance.has_perm_flag('can_create_forms')
def clean_custom_permissions(self):
val = self.cleaned_data.get('custom_permissions')
if isinstance(val, str):
try:
return json.loads(val) if val.strip() else {}
val = json.loads(val) if val.strip() else {}
except Exception:
return {}
elif isinstance(val, dict):
return val
return {}
val = {}
elif not isinstance(val, dict):
val = {}
return val
def save(self, commit=True):
user = super().save(commit=False)
can_create = self.cleaned_data.get('can_create_forms')
perms = dict(user.custom_permissions or {})
perms['can_create_forms'] = bool(can_create)
user.custom_permissions = perms
if commit:
user.save()
self.save_m2m()
return user
class UserAdmin(ModelAdmin, BaseUserAdmin):
@ -158,6 +176,7 @@ class UserAdmin(ModelAdmin, BaseUserAdmin):
"classes": ["tab"],
"fields": (
("is_active", "is_staff", "is_superuser"),
("can_create_forms",),
("two_factor_enabled", "two_factor_method"),
"custom_permissions",
("role", "user_type"),

39
apps/dynamic_forms/permissions.py

@ -0,0 +1,39 @@
from rest_framework.permissions import BasePermission
from django.utils.translation import gettext_lazy as _
class CanCreateDynamicForm(BasePermission):
"""
Permission to create dynamic forms:
1. User must be authenticated and active.
2. Platform staff, superusers, or regional admins are permitted.
3. Regular users must have an associated institution AND
the 'can_create_forms' permission flag (toggled via role or user custom_permissions in panel admin).
"""
message = _("Only users affiliated with an institution and granted form creation permissions can create dynamic forms.")
def has_permission(self, request, view):
user = request.user
if not user or not user.is_authenticated or not user.is_active:
return False
if user.is_staff or getattr(user, 'is_super_admin', False) or getattr(user, 'is_regional_admin', False) or user.is_superuser:
return True
# Check if user has an associated institution
has_institution = bool(
(hasattr(user, 'institution_memberships') and user.institution_memberships.exists()) or
(hasattr(user, 'created_institutions') and user.created_institutions.exists()) or
bool(getattr(user, 'institution_name', None))
)
if not has_institution:
self.message = _("Only users affiliated with an institution can create dynamic forms.")
return False
# Check togglable permission flag from panel admin (Role or custom_permissions)
has_permission_flag = user.has_perm_flag('can_create_forms') if hasattr(user, 'has_perm_flag') else False
if not has_permission_flag:
self.message = _("You do not have permission to create dynamic forms. Please contact an administrator.")
return False
return True

9
apps/dynamic_forms/serializers/form_serializers.py

@ -4,24 +4,31 @@ from apps.dynamic_forms.models import FormDefinition, FormResponse, WorkflowStat
class FormDefinitionListSerializer(serializers.ModelSerializer):
responses_count = serializers.IntegerField(source='responses.count', read_only=True)
creator_name = serializers.CharField(source='created_by.get_full_name', read_only=True)
class Meta:
model = FormDefinition
fields = [
'id', 'title', 'slug', 'description',
'is_active', 'is_public', 'target_role',
'created_by', 'creator_name',
'responses_count', 'created_at',
]
read_only_fields = ['id', 'created_by', 'created_at']
class FormDefinitionDetailSerializer(serializers.ModelSerializer):
creator_name = serializers.CharField(source='created_by.get_full_name', read_only=True)
class Meta:
model = FormDefinition
fields = [
'id', 'title', 'slug', 'description',
'schema', 'is_active', 'is_public',
'target_role', 'created_at', 'updated_at',
'target_role', 'created_by', 'creator_name',
'created_at', 'updated_at',
]
read_only_fields = ['id', 'created_by', 'created_at', 'updated_at']
class FormResponseSerializer(serializers.ModelSerializer):

156
apps/dynamic_forms/tests/test_phase9_forms.py

@ -92,3 +92,159 @@ class Phase9DynamicFormsAPITests(TestCase):
self.assertEqual(review_res.status_code, status.HTTP_200_OK)
self.assertEqual(review_res.data["workflow_status"], WorkflowStatus.APPROVED)
self.assertEqual(review_res.data["admin_notes"], review_payload["admin_notes"])
def test_form_list_filtering_by_user(self):
"""
Verify GET /api/v1/forms/?my_forms=true:
- Authenticated user sees only their own forms.
- Unauthenticated request returns 401.
- Default list returns active forms with created_by and creator_name.
"""
user_a = User.objects.create_user(
email="user_a@shiahub.org",
password="Password123!",
fullname="User A",
user_type=User.UserType.CLIENT,
)
user_b = User.objects.create_user(
email="user_b@shiahub.org",
password="Password123!",
fullname="User B",
user_type=User.UserType.CLIENT,
)
form_a1 = FormDefinition.objects.create(
title="Form A1",
slug="form-a1",
schema=[],
is_active=True,
created_by=user_a,
)
form_a2 = FormDefinition.objects.create(
title="Form A2",
slug="form-a2",
schema=[],
is_active=False,
created_by=user_a,
)
form_b1 = FormDefinition.objects.create(
title="Form B1",
slug="form-b1",
schema=[],
is_active=True,
created_by=user_b,
)
# 1. Unauthenticated request with my_forms=true should return 401
self.client.logout()
res_unauth = self.client.get("/api/v1/forms/?my_forms=true")
self.assertEqual(res_unauth.status_code, status.HTTP_401_UNAUTHORIZED)
# 2. Authenticated as user_a with my_forms=true
self.client.force_authenticate(user=user_a)
res_a = self.client.get("/api/v1/forms/?my_forms=true")
self.assertEqual(res_a.status_code, status.HTTP_200_OK)
self.assertEqual(len(res_a.data), 2)
titles_a = [item["title"] for item in res_a.data]
self.assertIn("Form A1", titles_a)
self.assertIn("Form A2", titles_a)
self.assertEqual(res_a.data[0]["created_by"], user_a.id)
self.assertEqual(res_a.data[0]["creator_name"], "User A")
# 3. Authenticated as user_b with my_forms=true
self.client.force_authenticate(user=user_b)
res_b = self.client.get("/api/v1/forms/?my_forms=true")
self.assertEqual(res_b.status_code, status.HTTP_200_OK)
self.assertEqual(len(res_b.data), 1)
self.assertEqual(res_b.data[0]["title"], "Form B1")
self.assertEqual(res_b.data[0]["created_by"], user_b.id)
self.assertEqual(res_b.data[0]["creator_name"], "User B")
# 4. Standard listing without my_forms returns all active forms
res_all = self.client.get("/api/v1/forms/")
self.assertEqual(res_all.status_code, status.HTTP_200_OK)
# Should include self.form_def, form_a1, form_b1 (form_a2 is inactive)
active_titles = [item["title"] for item in res_all.data]
self.assertIn("Form A1", active_titles)
self.assertIn("Form B1", active_titles)
self.assertNotIn("Form A2", active_titles)
def test_form_creation_permissions(self):
"""
Verify POST /api/v1/forms/ permission constraints:
- Unauthenticated -> 401
- User without institution -> 403
- User with institution but without can_create_forms -> 403
- User with institution AND can_create_forms toggled -> 201
- Staff / Admin user -> 201
"""
# 1. Unauthenticated -> 401
self.client.logout()
payload = {
"title": "Unauthenticated Form",
"slug": "unauth-form",
"schema": [],
}
res = self.client.post("/api/v1/forms/", payload, format="json")
self.assertEqual(res.status_code, status.HTTP_401_UNAUTHORIZED)
# 2. Client without institution -> 403
no_inst_user = User.objects.create_user(
email="no_inst@shiahub.org",
password="Password123!",
fullname="No Institution User",
user_type=User.UserType.CLIENT,
)
self.client.force_authenticate(user=no_inst_user)
payload = {
"title": "No Inst Form",
"slug": "no-inst-form",
"schema": [],
}
res = self.client.post("/api/v1/forms/", payload, format="json")
self.assertEqual(res.status_code, status.HTTP_403_FORBIDDEN)
self.assertIn("institution", str(res.data.get("detail", "")).lower())
# 3. User with institution name but can_create_forms=False -> 403
inst_user = User.objects.create_user(
email="inst_user@shiahub.org",
password="Password123!",
fullname="Institution User",
user_type=User.UserType.CLIENT,
institution_name="Al-Mustafa University",
custom_permissions={"can_create_forms": False},
)
self.client.force_authenticate(user=inst_user)
payload = {
"title": "Disallowed Form",
"slug": "disallowed-form",
"schema": [],
}
res = self.client.post("/api/v1/forms/", payload, format="json")
self.assertEqual(res.status_code, status.HTTP_403_FORBIDDEN)
# 4. User with institution AND can_create_forms=True -> 201
inst_user.custom_permissions["can_create_forms"] = True
inst_user.save()
payload = {
"title": "Permitted Institution Form",
"slug": "permitted-inst-form",
"schema": [
{"id": "proposal", "label": "Proposal Text", "type": "textarea", "required": True}
],
}
res = self.client.post("/api/v1/forms/", payload, format="json")
self.assertEqual(res.status_code, status.HTTP_201_CREATED)
self.assertEqual(res.data["created_by"], inst_user.id)
self.assertEqual(res.data["creator_name"], "Institution User")
# 5. Staff user -> 201
self.client.force_authenticate(user=self.admin_user)
admin_payload = {
"title": "Admin Created Form",
"slug": "admin-created-form",
"schema": [],
}
res = self.client.post("/api/v1/forms/", admin_payload, format="json")
self.assertEqual(res.status_code, status.HTTP_201_CREATED)
self.assertEqual(res.data["created_by"], self.admin_user.id)

51
apps/dynamic_forms/views/form_views.py

@ -8,8 +8,10 @@ from rest_framework.generics import GenericAPIView
from rest_framework.permissions import AllowAny, IsAuthenticated
from rest_framework.response import Response
from rest_framework.parsers import MultiPartParser, FormParser, JSONParser
from drf_spectacular.utils import extend_schema, OpenApiResponse
from drf_spectacular.utils import extend_schema, OpenApiResponse, OpenApiParameter
from drf_spectacular.types import OpenApiTypes
from apps.dynamic_forms.permissions import CanCreateDynamicForm
from apps.dynamic_forms.models import FormDefinition, FormResponse, WorkflowStatus
from apps.dynamic_forms.serializers import (
FormDefinitionListSerializer,
@ -36,31 +38,62 @@ class FormDefinitionListCreateView(GenericAPIView):
queryset = FormDefinition.objects.filter(is_active=True)
@extend_schema(
summary="List active dynamic forms",
description="Retrieves available custom dynamic forms and survey workflows.",
summary="List dynamic forms",
description="Retrieves available forms. Supports filtering to only user's own forms via `my_forms=true`.",
parameters=[
OpenApiParameter(
name="my_forms",
type=OpenApiTypes.BOOL,
location=OpenApiParameter.QUERY,
required=False,
description="If true, returns only forms created by the authenticated user."
),
],
responses={200: FormDefinitionListSerializer(many=True)},
tags=["Dynamic Forms & Surveys"],
)
def get(self, request, *args, **kwargs):
qs = FormDefinition.objects.all()
if not (request.user and request.user.is_authenticated and request.user.is_staff):
qs = qs.filter(is_active=True)
is_my_forms = request.query_params.get('my_forms', '').lower() in ('true', '1')
if is_my_forms:
if not request.user or not request.user.is_authenticated:
return Response(
{'detail': _("Authentication required to view personal forms.")},
status=status.HTTP_401_UNAUTHORIZED
)
qs = FormDefinition.objects.filter(created_by=request.user)
else:
qs = FormDefinition.objects.all()
if not (request.user and request.user.is_authenticated and request.user.is_staff):
qs = qs.filter(is_active=True)
serializer = self.get_serializer(qs, many=True)
return Response(serializer.data, status=status.HTTP_200_OK)
@extend_schema(
summary="Create new dynamic form schema",
description="Defines a new custom form, questions, validation rules, and schema (Staff/Admin only).",
description="Defines a new custom form, questions, validation rules, and schema (Staff or affiliated Institution users with permission).",
request=FormDefinitionDetailSerializer,
responses={
201: FormDefinitionDetailSerializer,
401: OpenApiResponse(description="Authentication required"),
403: OpenApiResponse(description="Permission denied"),
},
tags=["Dynamic Forms & Surveys"],
)
def post(self, request, *args, **kwargs):
if not request.user or not request.user.is_authenticated or not request.user.is_staff:
return Response({'detail': _("Only platform administrators can create form schemas.")}, status=status.HTTP_403_FORBIDDEN)
if not request.user or not request.user.is_authenticated:
return Response(
{'detail': _("Authentication required to create forms.")},
status=status.HTTP_401_UNAUTHORIZED
)
perm = CanCreateDynamicForm()
if not perm.has_permission(request, self):
return Response(
{'detail': getattr(perm, 'message', _("Only users affiliated with an institution and granted form creation permissions can create dynamic forms."))},
status=status.HTTP_403_FORBIDDEN
)
serializer = FormDefinitionDetailSerializer(data=request.data)
serializer.is_valid(raise_exception=True)

34
apps/profiles/admin.py

@ -10,6 +10,7 @@ from apps.profiles.models import (
InstitutionTimeline,
InstitutionMedia,
InstitutionFollower,
QuickAccessLink,
)
from utils.admin import project_admin_site
@ -59,6 +60,12 @@ class InstitutionMemberInline(TabularInline):
readonly_fields = ('joined_at',)
class QuickAccessLinkInline(TabularInline):
model = QuickAccessLink
extra = 0
fields = ('order', 'title', 'url', 'category', 'icon', 'color', 'is_live', 'is_external', 'is_active')
class InstitutionTimelineInline(StackedInline):
model = InstitutionTimeline
extra = 0
@ -91,7 +98,7 @@ class InstitutionAdmin(ModelAdmin):
search_fields = ('name', 'region__name', 'city', 'country', 'description', 'email', 'phone')
prepopulated_fields = {'slug': ('name',)}
readonly_fields = ('follower_count', 'created_at', 'updated_at')
inlines = [InstitutionMemberInline, InstitutionTimelineInline, InstitutionMediaInline]
inlines = [InstitutionMemberInline, QuickAccessLinkInline, InstitutionTimelineInline, InstitutionMediaInline]
actions = ['approve_institutions', 'reject_institutions', 'toggle_featured']
def get_queryset(self, request):
@ -218,3 +225,28 @@ class InstitutionFollowerAdmin(ModelAdmin):
list_display_links = ('user', 'institution')
search_fields = ('user__email', 'user__fullname', 'institution__name')
readonly_fields = ('created_at',)
@admin.register(QuickAccessLink, site=project_admin_site)
class QuickAccessLinkAdmin(ModelAdmin):
list_display = (
'title',
'institution',
'category',
'icon',
'color_badge',
'is_live',
'is_external',
'order',
'is_active',
'created_at',
)
list_display_links = ('title', 'institution')
list_filter = ('is_active', 'is_live', 'color', 'category', 'institution')
search_fields = ('title', 'url', 'description', 'category', 'institution__name')
readonly_fields = ('created_at', 'updated_at')
@display(description=_('Color'))
def color_badge(self, obj):
return obj.get_color_display()

89
apps/profiles/management/commands/seed_quick_access_links.py

@ -0,0 +1,89 @@
from django.core.management.base import BaseCommand
from apps.profiles.models import Institution, QuickAccessLink
PRESETS = [
{
"title": "زیارت مجازی روضه منوره",
"url": "https://tv.razavi.ir/virtual-tour",
"description": "نمای سه‌بعدی و ۳۶۰ درجه از حرم مطهر رضوی، روضه منوره و صحن‌های متبرکه",
"icon": "landmark",
"color": "emerald",
"category": "زیارت مجازی",
"is_live": False,
"is_external": True,
"order": 1,
},
{
"title": "پخش زنده حرم مطهر رضوی",
"url": "https://tv.razavi.ir/live",
"description": "پخش مستقیم و ۲۴ ساعته از صحن انقلاب، گوهرشاد و ضریح مطهر",
"icon": "video",
"color": "rose",
"category": "پخش زنده",
"is_live": True,
"is_external": True,
"order": 2,
},
{
"title": "کتابخانه دیجیتال آستان قدس",
"url": "https://digital.razavi.ir",
"description": "دسترسی و جستجو در صدها هزار نسخه خطی، اسناد تاریخی و کتب معارف",
"icon": "bookOpen",
"color": "blue",
"category": "کتابخانه دیجیتال",
"is_live": False,
"is_external": True,
"order": 3,
},
{
"title": "مرکز تولیدات چندرسانه‌ای",
"url": "https://media.razavi.ir",
"description": "آرشیو تواشیح، نماهنگ‌ها، سخنرانی‌های مذهبی، پادکست‌ها و مستندات چندزبانه",
"icon": "film",
"color": "purple",
"category": "چندرسانه‌ای",
"is_live": False,
"is_external": True,
"order": 4,
},
]
class Command(BaseCommand):
help = "Seed canonical quick access links for primary institution (e.g. Astan Quds / ID 1)"
def add_arguments(self, parser):
parser.add_argument('--institution-id', type=int, default=1, help='Institution ID to attach links to')
def handle(self, *args, **options):
inst_id = options.get('institution_id', 1)
institution = Institution.objects.filter(id=inst_id).first() or Institution.objects.first()
if not institution:
self.stdout.write(self.style.WARNING("No institution found to seed quick access links."))
return
created_count = 0
for item in PRESETS:
link, created = QuickAccessLink.objects.update_or_create(
institution=institution,
title=item['title'],
defaults={
'url': item['url'],
'description': item['description'],
'icon': item['icon'],
'color': item['color'],
'category': item['category'],
'is_live': item['is_live'],
'is_external': item['is_external'],
'order': item['order'],
'is_active': True,
}
)
if created:
created_count += 1
self.stdout.write(
self.style.SUCCESS(
f"Successfully seeded canonical quick access links for '{institution.name}' (ID: {institution.id})."
)
)

38
apps/profiles/migrations/0003_quickaccesslink.py

@ -0,0 +1,38 @@
# Generated by Django 4.2.30 on 2026-10-18 12:18
from django.db import migrations, models
import django.db.models.deletion
class Migration(migrations.Migration):
dependencies = [
('profiles', '0002_region_and_institution_region'),
]
operations = [
migrations.CreateModel(
name='QuickAccessLink',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('title', models.CharField(max_length=255, verbose_name='عنوان خدمت')),
('url', models.CharField(max_length=500, verbose_name='نشانی اینترنتی یا مسیر')),
('description', models.CharField(blank=True, default='', max_length=500, verbose_name='توضیح مختصر')),
('icon', models.CharField(default='link', max_length=50, verbose_name='شناسه آیکون')),
('color', models.CharField(choices=[('emerald', 'سبز رضوی (Emerald)'), ('rose', 'سرخ زنده (Rose)'), ('blue', 'آبی دانش (Blue)'), ('purple', 'بنفش چندرسانه\u200cای (Purple)'), ('amber', 'طلایی ویژه (Amber)'), ('teal', 'فیروزه\u200cای (Teal)')], default='emerald', max_length=20, verbose_name='طیف رنگی')),
('category', models.CharField(blank=True, default='', max_length=100, verbose_name='برچسب یا دسته\u200cبندی')),
('is_live', models.BooleanField(default=False, verbose_name='نشانگر پخش زنده')),
('is_external', models.BooleanField(default=True, verbose_name='باز شدن در برگه جدید')),
('order', models.PositiveIntegerField(default=0, verbose_name='ترتیب نمایش')),
('is_active', models.BooleanField(default=True, verbose_name='فعال بودن')),
('created_at', models.DateTimeField(auto_now_add=True, verbose_name='تاریخ ایجاد')),
('updated_at', models.DateTimeField(auto_now=True, verbose_name='آخرین بروزرسانی')),
('institution', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='quick_access_links', to='profiles.institution', verbose_name='مؤسسه مربوطه')),
],
options={
'verbose_name': 'لینک دسترسی سریع',
'verbose_name_plural': 'لینک\u200cهای دسترسی سریع',
'ordering': ['order', 'id'],
},
),
]

3
apps/profiles/models/__init__.py

@ -6,6 +6,7 @@ from .institution import (
InstitutionMedia,
InstitutionFollower,
)
from .quick_access_link import QuickAccessLink
__all__ = [
'Region',
@ -15,4 +16,6 @@ __all__ = [
'InstitutionTimeline',
'InstitutionMedia',
'InstitutionFollower',
'QuickAccessLink',
]

83
apps/profiles/models/quick_access_link.py

@ -0,0 +1,83 @@
from django.db import models
from django.utils.translation import gettext_lazy as _
class QuickAccessLink(models.Model):
class ColorChoice(models.TextChoices):
EMERALD = "emerald", _("سبز رضوی (Emerald)")
ROSE = "rose", _("سرخ زنده (Rose)")
BLUE = "blue", _("آبی دانش (Blue)")
PURPLE = "purple", _("بنفش چندرسانه‌ای (Purple)")
AMBER = "amber", _("طلایی ویژه (Amber)")
TEAL = "teal", _("فیروزه‌ای (Teal)")
institution = models.ForeignKey(
"profiles.Institution",
on_delete=models.CASCADE,
related_name="quick_access_links",
verbose_name=_("مؤسسه مربوطه"),
)
title = models.CharField(
max_length=255,
verbose_name=_("عنوان خدمت")
)
url = models.CharField(
max_length=500,
verbose_name=_("نشانی اینترنتی یا مسیر")
)
description = models.CharField(
max_length=500,
blank=True,
default="",
verbose_name=_("توضیح مختصر")
)
icon = models.CharField(
max_length=50,
default="link",
verbose_name=_("شناسه آیکون")
)
color = models.CharField(
max_length=20,
choices=ColorChoice.choices,
default=ColorChoice.EMERALD,
verbose_name=_("طیف رنگی"),
)
category = models.CharField(
max_length=100,
blank=True,
default="",
verbose_name=_("برچسب یا دسته‌بندی")
)
is_live = models.BooleanField(
default=False,
verbose_name=_("نشانگر پخش زنده")
)
is_external = models.BooleanField(
default=True,
verbose_name=_("باز شدن در برگه جدید")
)
order = models.PositiveIntegerField(
default=0,
verbose_name=_("ترتیب نمایش")
)
is_active = models.BooleanField(
default=True,
verbose_name=_("فعال بودن")
)
created_at = models.DateTimeField(
auto_now_add=True,
verbose_name=_("تاریخ ایجاد")
)
updated_at = models.DateTimeField(
auto_now=True,
verbose_name=_("آخرین بروزرسانی")
)
class Meta:
verbose_name = _("لینک دسترسی سریع")
verbose_name_plural = _("لینک‌های دسترسی سریع")
ordering = ["order", "id"]
def __str__(self):
return f"{self.title} ({self.institution.name})"

8
apps/profiles/permissions.py

@ -19,8 +19,8 @@ class IsInstitutionAdminOrReadOnly(BasePermission):
if not request.user or not request.user.is_authenticated or not request.user.is_active:
return False
# Super admin has global write authority
if getattr(request.user, 'is_super_admin', False) or request.user.is_superuser:
# Super admin or staff has global write authority
if getattr(request.user, 'is_super_admin', False) or request.user.is_superuser or request.user.is_staff:
return True
# Check if obj is Institution or a related model with an institution foreign key
@ -28,6 +28,10 @@ class IsInstitutionAdminOrReadOnly(BasePermission):
if not institution:
return False
user_inst_id = getattr(request.user, 'institution_id', None)
if user_inst_id and user_inst_id == institution.id:
return True
return institution.is_admin(request.user)

5
apps/profiles/serializers/__init__.py

@ -9,6 +9,8 @@ from .institution_serializers import (
InstitutionDetailSerializer,
InstitutionCreateUpdateSerializer,
FollowResponseSerializer,
QuickAccessLinkSerializer,
QuickAccessLinkReorderSerializer,
)
__all__ = [
@ -22,5 +24,8 @@ __all__ = [
'InstitutionDetailSerializer',
'InstitutionCreateUpdateSerializer',
'FollowResponseSerializer',
'QuickAccessLinkSerializer',
'QuickAccessLinkReorderSerializer',
]

58
apps/profiles/serializers/institution_serializers.py

@ -1,4 +1,5 @@
from django.contrib.auth import get_user_model
from django.utils.translation import gettext_lazy as _
from rest_framework import serializers
from drf_spectacular.utils import extend_schema_field
from django.db.models import Q
@ -9,6 +10,7 @@ from apps.profiles.models import (
InstitutionTimeline,
InstitutionMedia,
InstitutionFollower,
QuickAccessLink,
)
User = get_user_model()
@ -74,6 +76,45 @@ class InstitutionMediaSerializer(serializers.ModelSerializer):
read_only_fields = ['id', 'institution', 'uploaded_at', 'media_type_display']
class QuickAccessLinkSerializer(serializers.ModelSerializer):
institution_id = serializers.IntegerField(source='institution.id', read_only=True)
class Meta:
model = QuickAccessLink
fields = [
'id',
'institution',
'institution_id',
'title',
'url',
'description',
'icon',
'color',
'category',
'is_live',
'is_external',
'order',
'is_active',
'created_at',
'updated_at',
]
read_only_fields = ['id', 'institution', 'institution_id', 'created_at', 'updated_at']
def validate_url(self, value):
val = value.strip()
if not val.startswith(('http://', 'https://', '/', '#')):
raise serializers.ValidationError(_("URL must start with http://, https:// or /"))
return val
class QuickAccessLinkReorderSerializer(serializers.Serializer):
order = serializers.ListField(
child=serializers.IntegerField(),
required=True,
help_text="Ordered list of QuickAccessLink IDs."
)
class MemberUserMiniSerializer(serializers.ModelSerializer):
class Meta:
model = User
@ -189,6 +230,7 @@ class InstitutionDetailSerializer(serializers.ModelSerializer):
region_name = serializers.CharField(source='region.name', read_only=True)
timeline_entries = InstitutionTimelineSerializer(many=True, read_only=True)
media_items = InstitutionMediaSerializer(many=True, read_only=True)
quick_access_links = serializers.SerializerMethodField()
projects = serializers.SerializerMethodField()
posts = serializers.SerializerMethodField()
members_count = serializers.SerializerMethodField()
@ -235,6 +277,7 @@ class InstitutionDetailSerializer(serializers.ModelSerializer):
'user_role',
'timeline_entries',
'media_items',
'quick_access_links',
'projects',
'posts',
'created_at',
@ -257,12 +300,27 @@ class InstitutionDetailSerializer(serializers.ModelSerializer):
'user_role',
'timeline_entries',
'media_items',
'quick_access_links',
'projects',
'posts',
'created_at',
'updated_at',
]
@extend_schema_field(QuickAccessLinkSerializer(many=True))
def get_quick_access_links(self, obj):
links = obj.quick_access_links.all()
request = self.context.get('request')
is_privileged = False
if request and request.user and request.user.is_authenticated:
if request.user.is_staff or getattr(request.user, 'is_super_admin', False) or request.user.is_superuser:
is_privileged = True
elif hasattr(obj, 'is_admin') and obj.is_admin(request.user):
is_privileged = True
if not is_privileged:
links = links.filter(is_active=True)
return QuickAccessLinkSerializer(links.order_by('order', 'id'), many=True, context=self.context).data
@extend_schema_field(serializers.IntegerField)
def get_members_count(self, obj) -> int:
return obj.members.count()

176
apps/profiles/tests/test_quick_access_links.py

@ -0,0 +1,176 @@
from django.test import TestCase
from django.contrib.auth import get_user_model
from rest_framework.test import APIClient
from rest_framework import status
from apps.profiles.models import Region, Institution, InstitutionMember, QuickAccessLink
User = get_user_model()
class QuickAccessLinkAPITests(TestCase):
def setUp(self):
self.client = APIClient()
self.region = Region.objects.create(name="Middle East", code="ME", slug="middle-east")
self.admin_user = User.objects.create_user(
email="admin@razavi.ir",
password="SecurePassword123!",
fullname="Admin User",
user_type=User.UserType.INSTITUTION_ADMIN,
is_staff=True,
)
self.other_user = User.objects.create_user(
email="other@example.com",
password="SecurePassword123!",
fullname="Regular User",
user_type=User.UserType.CLIENT,
)
self.institution = Institution.objects.create(
name="Astan Quds Razavi",
slug="astan-quds",
region=self.region,
country="Iran",
city="Mashhad",
created_by=self.admin_user,
)
InstitutionMember.objects.create(
institution=self.institution,
user=self.admin_user,
role=InstitutionMember.MemberRole.ADMIN,
)
self.link1 = QuickAccessLink.objects.create(
institution=self.institution,
title="زیارت مجازی روضه منوره",
url="https://tv.razavi.ir/virtual-tour",
description="نمای سه‌بعدی و ۳۶۰ درجه از حرم مطهر رضوی",
icon="landmark",
color=QuickAccessLink.ColorChoice.EMERALD,
category="زیارت مجازی",
is_live=False,
is_external=True,
order=1,
is_active=True,
)
self.link2 = QuickAccessLink.objects.create(
institution=self.institution,
title="پخش زنده حرم مطهر رضوی",
url="https://tv.razavi.ir/live",
description="پخش مستقیم و ۲۴ ساعته",
icon="video",
color=QuickAccessLink.ColorChoice.ROSE,
category="پخش زنده",
is_live=True,
is_external=True,
order=2,
is_active=False, # Inactive link to test filtering
)
def test_embedded_quick_links_in_institution_detail(self):
# Public user should only see active link
response = self.client.get(f"/api/v1/profiles/{self.institution.id}/")
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertIn("quick_access_links", response.data)
links = response.data["quick_access_links"]
self.assertEqual(len(links), 1)
self.assertEqual(links[0]["id"], self.link1.id)
# Admin user should see all links
self.client.force_authenticate(user=self.admin_user)
response = self.client.get(f"/api/v1/profiles/{self.institution.id}/")
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(len(response.data["quick_access_links"]), 2)
def test_list_quick_access_links_public_and_admin(self):
# Public user
response = self.client.get(f"/api/v1/profiles/{self.institution.slug}/quick-links/")
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(len(response.data), 1)
self.assertEqual(response.data[0]["title"], self.link1.title)
# Admin user
self.client.force_authenticate(user=self.admin_user)
response = self.client.get(f"/api/v1/profiles/{self.institution.id}/quick-links/")
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(len(response.data), 2)
def test_create_quick_access_link_permission(self):
new_data = {
"title": "کتابخانه دیجیتال",
"url": "https://digital.razavi.ir",
"description": "دسترسی به کتب",
"icon": "bookOpen",
"color": "blue",
"category": "کتابخانه",
"order": 3,
"is_live": False,
"is_external": True,
}
# Unauthenticated create should fail (401)
response = self.client.post(f"/api/v1/profiles/{self.institution.id}/quick-links/", new_data, format="json")
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
# Unauthorized user create should fail (403)
self.client.force_authenticate(user=self.other_user)
response = self.client.post(f"/api/v1/profiles/{self.institution.id}/quick-links/", new_data, format="json")
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
# Admin create should succeed (201)
self.client.force_authenticate(user=self.admin_user)
response = self.client.post(f"/api/v1/profiles/{self.institution.slug}/quick-links/", new_data, format="json")
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
self.assertEqual(response.data["title"], "کتابخانه دیجیتال")
self.assertEqual(response.data["institution_id"], self.institution.id)
def test_url_validation(self):
self.client.force_authenticate(user=self.admin_user)
invalid_data = {
"title": "Invalid Link",
"url": "javascript:alert('xss')",
}
response = self.client.post(f"/api/v1/profiles/{self.institution.id}/quick-links/", invalid_data, format="json")
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
error_fields = [e.get('field') for e in response.data.get('errors', [])] if isinstance(response.data, dict) and 'errors' in response.data else response.data
self.assertIn("url", error_fields)
def test_retrieve_update_and_delete_quick_link(self):
# Retrieve detail
response = self.client.get(f"/api/v1/profiles/{self.institution.id}/quick-links/{self.link1.id}/")
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data["id"], self.link1.id)
# Update link (patch)
self.client.force_authenticate(user=self.admin_user)
patch_response = self.client.patch(
f"/api/v1/profiles/{self.institution.id}/quick-links/{self.link1.id}/",
{"title": "تور مجازی ۳۶۰ درجه به روز رسانی شده"},
format="json",
)
self.assertEqual(patch_response.status_code, status.HTTP_200_OK)
self.assertEqual(patch_response.data["title"], "تور مجازی ۳۶۰ درجه به روز رسانی شده")
# Delete link
del_response = self.client.delete(f"/api/v1/profiles/{self.institution.id}/quick-links/{self.link1.id}/")
self.assertEqual(del_response.status_code, status.HTTP_204_NO_CONTENT)
self.assertFalse(QuickAccessLink.objects.filter(id=self.link1.id).exists())
def test_reorder_quick_links(self):
self.client.force_authenticate(user=self.admin_user)
reorder_data = {"order": [self.link2.id, self.link1.id]}
response = self.client.post(
f"/api/v1/profiles/{self.institution.id}/quick-links/reorder/",
reorder_data,
format="json",
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.link1.refresh_from_db()
self.link2.refresh_from_db()
self.assertEqual(self.link2.order, 0)
self.assertEqual(self.link1.order, 1)

9
apps/profiles/urls.py

@ -11,6 +11,9 @@ from apps.profiles.views import (
InstitutionFollowToggleView,
InstitutionMembersView,
InstitutionMemberDetailView,
QuickAccessLinkListCreateView,
QuickAccessLinkDetailView,
QuickAccessLinkReorderView,
)
urlpatterns = [
@ -22,6 +25,11 @@ urlpatterns = [
path('', InstitutionListCreateView.as_view(), name='institution_list_create'),
path('<str:pk_or_slug>/', InstitutionDetailView.as_view(), name='institution_detail'),
# Quick Access Links to Services (F-CMS-04)
path('<str:pk_or_slug>/quick-links/', QuickAccessLinkListCreateView.as_view(), name='institution_quick_links'),
path('<str:pk_or_slug>/quick-links/<int:link_id>/', QuickAccessLinkDetailView.as_view(), name='institution_quick_link_detail'),
path('<str:pk_or_slug>/quick-links/reorder/', QuickAccessLinkReorderView.as_view(), name='institution_quick_links_reorder'),
# Historical Timeline Milestones
path('<str:pk_or_slug>/timeline/', InstitutionTimelineView.as_view(), name='institution_timeline'),
path('<str:pk_or_slug>/timeline/<int:entry_id>/', InstitutionTimelineDetailView.as_view(), name='institution_timeline_detail'),
@ -37,3 +45,4 @@ urlpatterns = [
path('<str:pk_or_slug>/members/', InstitutionMembersView.as_view(), name='institution_members'),
path('<str:pk_or_slug>/members/<int:member_id>/', InstitutionMemberDetailView.as_view(), name='institution_member_detail'),
]

9
apps/profiles/views/__init__.py

@ -10,6 +10,10 @@ from .institution_views import (
InstitutionFollowToggleView,
InstitutionMembersView,
InstitutionMemberDetailView,
QuickAccessLinkListCreateView,
QuickAccessLinkDetailView,
QuickAccessLinkReorderView,
QuickAccessLinkViewSet,
)
__all__ = [
@ -24,5 +28,10 @@ __all__ = [
'InstitutionFollowToggleView',
'InstitutionMembersView',
'InstitutionMemberDetailView',
'QuickAccessLinkListCreateView',
'QuickAccessLinkDetailView',
'QuickAccessLinkReorderView',
'QuickAccessLinkViewSet',
]

181
apps/profiles/views/institution_views.py

@ -3,7 +3,7 @@ from django.contrib.auth import get_user_model
from django.db.models import Q, F
from django.shortcuts import get_object_or_404
from django.utils.translation import gettext_lazy as _
from rest_framework import status
from rest_framework import status, viewsets
from rest_framework.generics import GenericAPIView
from rest_framework.parsers import MultiPartParser, FormParser, JSONParser
from rest_framework.permissions import AllowAny, IsAuthenticated
@ -17,6 +17,7 @@ from apps.profiles.models import (
InstitutionTimeline,
InstitutionMedia,
InstitutionFollower,
QuickAccessLink,
)
from apps.profiles.permissions import (
IsInstitutionAdminOrReadOnly,
@ -34,6 +35,8 @@ from apps.profiles.serializers import (
InstitutionMemberSerializer,
InstitutionMemberAddSerializer,
FollowResponseSerializer,
QuickAccessLinkSerializer,
QuickAccessLinkReorderSerializer,
)
from utils.pagination import StandardResultsSetPagination, InstitutionPagination
@ -589,3 +592,179 @@ class InstitutionMemberDetailView(GenericAPIView):
member.delete()
return Response(status=status.HTTP_204_NO_CONTENT)
class QuickAccessLinkListCreateView(GenericAPIView):
serializer_class = QuickAccessLinkSerializer
queryset = QuickAccessLink.objects.all()
def get_permissions(self):
if self.request.method == 'POST':
return [IsAuthenticated(), IsInstitutionAdminOrReadOnly()]
return [AllowAny()]
@extend_schema(
summary="List quick access links",
description="Returns administrative quick access service links (virtual tour, live broadcast, digital library, etc.) for an institution.",
responses={200: QuickAccessLinkSerializer(many=True)},
tags=["Institutions & Profiles"],
)
def get(self, request, pk_or_slug, *args, **kwargs):
institution = get_institution_by_id_or_slug(pk_or_slug)
links = institution.quick_access_links.all().order_by('order', 'id')
is_admin_user = (
request.user and request.user.is_authenticated and
(request.user.is_staff or request.user.is_superuser or getattr(request.user, 'is_super_admin', False) or institution.is_admin(request.user))
)
if not is_admin_user:
links = links.filter(is_active=True)
serializer = QuickAccessLinkSerializer(links, many=True, context={'request': request})
return Response(serializer.data, status=status.HTTP_200_OK)
@extend_schema(
summary="Create quick access link",
description="Creates a new quick access service link for an institution.",
request=QuickAccessLinkSerializer,
responses={201: QuickAccessLinkSerializer},
tags=["Institutions & Profiles"],
)
def post(self, request, pk_or_slug, *args, **kwargs):
institution = get_institution_by_id_or_slug(pk_or_slug)
self.check_object_permissions(request, institution)
serializer = QuickAccessLinkSerializer(data=request.data, context={'request': request})
serializer.is_valid(raise_exception=True)
link = serializer.save(institution=institution)
return Response(QuickAccessLinkSerializer(link, context={'request': request}).data, status=status.HTTP_201_CREATED)
class QuickAccessLinkDetailView(GenericAPIView):
serializer_class = QuickAccessLinkSerializer
queryset = QuickAccessLink.objects.all()
def get_permissions(self):
if self.request.method in ['PUT', 'PATCH', 'DELETE']:
return [IsAuthenticated(), IsInstitutionAdminOrReadOnly()]
return [AllowAny()]
@extend_schema(
summary="Retrieve quick access link",
description="Gets details of a specific quick access service link.",
responses={200: QuickAccessLinkSerializer},
tags=["Institutions & Profiles"],
)
def get(self, request, pk_or_slug, link_id, *args, **kwargs):
institution = get_institution_by_id_or_slug(pk_or_slug)
link = get_object_or_404(QuickAccessLink, id=link_id, institution=institution)
serializer = QuickAccessLinkSerializer(link, context={'request': request})
return Response(serializer.data, status=status.HTTP_200_OK)
@extend_schema(
summary="Update quick access link",
description="Fully updates an existing quick access service link.",
request=QuickAccessLinkSerializer,
responses={200: QuickAccessLinkSerializer},
tags=["Institutions & Profiles"],
)
def put(self, request, pk_or_slug, link_id, *args, **kwargs):
institution = get_institution_by_id_or_slug(pk_or_slug)
link = get_object_or_404(QuickAccessLink, id=link_id, institution=institution)
self.check_object_permissions(request, link)
serializer = QuickAccessLinkSerializer(link, data=request.data, partial=False, context={'request': request})
serializer.is_valid(raise_exception=True)
serializer.save()
return Response(serializer.data, status=status.HTTP_200_OK)
@extend_schema(
summary="Partial update quick access link",
description="Partially updates an existing quick access service link.",
request=QuickAccessLinkSerializer,
responses={200: QuickAccessLinkSerializer},
tags=["Institutions & Profiles"],
)
def patch(self, request, pk_or_slug, link_id, *args, **kwargs):
institution = get_institution_by_id_or_slug(pk_or_slug)
link = get_object_or_404(QuickAccessLink, id=link_id, institution=institution)
self.check_object_permissions(request, link)
serializer = QuickAccessLinkSerializer(link, data=request.data, partial=True, context={'request': request})
serializer.is_valid(raise_exception=True)
serializer.save()
return Response(serializer.data, status=status.HTTP_200_OK)
@extend_schema(
summary="Delete quick access link",
description="Removes a quick access service link from an institution.",
responses={204: OpenApiResponse(description="Link deleted")},
tags=["Institutions & Profiles"],
)
def delete(self, request, pk_or_slug, link_id, *args, **kwargs):
institution = get_institution_by_id_or_slug(pk_or_slug)
link = get_object_or_404(QuickAccessLink, id=link_id, institution=institution)
self.check_object_permissions(request, link)
link.delete()
return Response(status=status.HTTP_204_NO_CONTENT)
class QuickAccessLinkReorderView(GenericAPIView):
serializer_class = QuickAccessLinkReorderSerializer
permission_classes = [IsAuthenticated, IsInstitutionAdminOrReadOnly]
queryset = QuickAccessLink.objects.all()
@extend_schema(
summary="Reorder quick access links",
description="Reorders the display sequence of quick access links. Accepts either a list of IDs in order `{\"order\": [1, 2, 3]}` or an array of objects `[{\"id\": 1, \"order\": 0}, ...]`.",
request=QuickAccessLinkReorderSerializer,
responses={200: QuickAccessLinkSerializer(many=True)},
tags=["Institutions & Profiles"],
)
def post(self, request, pk_or_slug, *args, **kwargs):
institution = get_institution_by_id_or_slug(pk_or_slug)
self.check_object_permissions(request, institution)
data = request.data
if isinstance(data, list):
for item in data:
if isinstance(item, dict) and 'id' in item and 'order' in item:
QuickAccessLink.objects.filter(id=item['id'], institution=institution).update(order=item['order'])
elif isinstance(data, dict):
if 'order' in data and isinstance(data['order'], list):
for idx, link_id in enumerate(data['order']):
QuickAccessLink.objects.filter(id=link_id, institution=institution).update(order=idx)
elif 'items' in data and isinstance(data['items'], list):
for item in data['items']:
if isinstance(item, dict) and 'id' in item and 'order' in item:
QuickAccessLink.objects.filter(id=item['id'], institution=institution).update(order=item['order'])
links = institution.quick_access_links.all().order_by('order', 'id')
return Response(QuickAccessLinkSerializer(links, many=True, context={'request': request}).data, status=status.HTTP_200_OK)
class QuickAccessLinkViewSet(viewsets.ModelViewSet):
serializer_class = QuickAccessLinkSerializer
permission_classes = [IsInstitutionAdminOrReadOnly]
queryset = QuickAccessLink.objects.all()
def get_queryset(self):
inst_lookup = self.kwargs.get("institution_pk") or self.kwargs.get("pk_or_slug")
qs = QuickAccessLink.objects.all()
if inst_lookup:
if str(inst_lookup).isdigit():
qs = qs.filter(institution_id=int(inst_lookup))
else:
qs = qs.filter(institution__slug=inst_lookup)
if not (self.request.user and self.request.user.is_authenticated and (self.request.user.is_staff or self.request.user.is_superuser)):
qs = qs.filter(is_active=True)
return qs.order_by('order', 'id')
def perform_create(self, serializer):
inst_lookup = self.kwargs.get("institution_pk") or self.kwargs.get("pk_or_slug")
institution = get_institution_by_id_or_slug(inst_lookup) if inst_lookup else None
if institution:
self.check_object_permissions(self.request, institution)
serializer.save(institution=institution)
else:
serializer.save()

BIN
dump.rdb

15588
openapi.json
File diff suppressed because it is too large
View File

1850
schema.yml
File diff suppressed because it is too large
View File

Loading…
Cancel
Save