dovodi
/
Dovodi_Backend
3
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Browse Source

notification templates and schedule models added and send notification service updated

master
Mohsen Taba 1 week ago
parent
e77acb113e
commit
d48b1b054d
8 changed files with 626 additions and 8 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 172
      apps/account/management/commands/seed_notification_templates.py
  2. 53
      apps/account/migrations/0007_notificationtemplate_schedulednotification.py
  3. 122
      apps/account/models/notification.py
  4. 138
      apps/account/notification_service.py
  5. 49
      apps/account/tasks.py
  6. 97
      apps/account/tests/notifications/test_notifications.py
  7. 2
      config/settings/base.py
  8. 1
      requirements.txt

172
apps/account/management/commands/seed_notification_templates.py
View File

@ -0,0 +1,172 @@
from django.core.management.base import BaseCommand
from apps.account.models import NotificationTemplate
class Command(BaseCommand):
help = 'Seeds default NotificationTemplate records in the database.'
def handle(self, *args, **options):
default_templates = [
{
"notification_type": "course_access_granted",
"name": "دسترسی به دوره فعال شد",
"title_fa": "دسترسی به دوره فعال شد",
"title_en": "Course Access Granted",
"body_fa": "دسترسی شما به دوره «{course_title}» فعال شد.",
"body_en": "Your access to the course '{course_title}' has been activated.",
"placeholders_info": "{course_title}, {course_name}, {course_slug}, {student_name}, {fullname}"
},
{
"notification_type": "course_registered",
"name": "ثبت‌نام دوره جدید",
"title_fa": "ثبت‌نام دوره موفقیت‌آمیز بود",
"title_en": "Course Registration Successful",
"body_fa": "ثبت‌نام شما در دوره «{course_title}» با موفقیت انجام شد.",
"body_en": "You have successfully registered for the course '{course_title}'.",
"placeholders_info": "{course_title}, {course_name}, {course_slug}, {student_name}, {fullname}"
},
{
"notification_type": "lesson_completed",
"name": "اتمام درس",
"title_fa": "درس به اتمام رسید",
"title_en": "Lesson Completed",
"body_fa": "تبریک! شما درس «{lesson_title}» را به پایان رساندید.",
"body_en": "Congratulations! You completed the lesson '{lesson_title}'.",
"placeholders_info": "{course_title}, {course_name}, {lesson_title}, {student_name}, {fullname}"
},
{
"notification_type": "course_completed",
"name": "اتمام دوره",
"title_fa": "دوره به اتمام رسید",
"title_en": "Course Completed",
"body_fa": "آفرین! شما دوره «{course_title}» را با موفقیت به پایان رساندید.",
"body_en": "Well done! You have completed the course '{course_title}'.",
"placeholders_info": "{course_title}, {course_name}, {course_slug}, {student_name}, {fullname}"
},
{
"notification_type": "new_course_weekly",
"name": "معرفی دوره جدید هفتگی",
"title_fa": "دوره جدید ثبت شد",
"title_en": "New Course Registered",
"body_fa": "دوره جدید «{course_title}» ثبت شده است. همین حالا می‌توانید ثبت‌نام کنید.",
"body_en": "A new course '{course_title}' has been registered. You can enroll now.",
"placeholders_info": "{course_title}, {course_name}, {course_slug}"
},
{
"notification_type": "low_quiz_result",
"name": "پیشنهاد شرکت مجدد در آزمون",
"title_fa": "پیشنهاد شرکت مجدد در آزمون",
"title_en": "Quiz Retake Suggestion",
"body_fa": "امتیاز شما در آزمون «{quiz_title}» کمتر از حد نصاب شده است. پیشنهاد می‌کنیم مجدداً در این آزمون شرکت کنید.",
"body_en": "You scored below the passing threshold on quiz '{quiz_title}'. We suggest taking it again.",
"placeholders_info": "{quiz_title}, {student_name}, {fullname}"
},
{
"notification_type": "live_class_rescheduled",
"name": "تغییر زمان کلاس زنده",
"title_fa": "زمان کلاس زنده تغییر کرد",
"title_en": "Live Class Rescheduled",
"body_fa": "زمان کلاس زنده برای دوره «{course_title}» تغییر یافته است.",
"body_en": "The live class for '{course_title}' has been rescheduled.",
"placeholders_info": "{course_title}, {course_name}, {session_subject}, {session_start_time}"
},
{
"notification_type": "live_class_cancelled",
"name": "لغو کلاس زنده",
"title_fa": "کلاس زنده لغو شد",
"title_en": "Live Class Cancelled",
"body_fa": "کلاس زنده برای دوره «{course_title}» لغو شده است.",
"body_en": "The live class for '{course_title}' has been cancelled.",
"placeholders_info": "{course_title}, {course_name}, {session_subject}"
},
{
"notification_type": "live_recording_available",
"name": "ویدئو ضبط شده کلاس آماده است",
"title_fa": "ویدئو ضبط شده کلاس آماده است",
"title_en": "Live Recording Available",
"body_fa": "ویدئو ضبط شده کلاس زنده برای دوره «{course_title}» آماده مشاهده است.",
"body_en": "The recorded video of the live class for '{course_title}' is now available.",
"placeholders_info": "{course_title}, {course_name}"
},
{
"notification_type": "missed_live_sessions",
"name": "غیبت متوالی کلاس زنده",
"title_fa": "غیبت متوالی در کلاس‌های زنده",
"title_en": "Missed Live Sessions",
"body_fa": "شما ۲ جلسه زنده متوالی را در دوره «{course_title}» غایب بوده‌اید. توصیه می‌کنیم ویدئوهای ضبط شده را مشاهده کنید.",
"body_en": "You have missed 2 consecutive live sessions in the course '{course_title}'. We recommend watching the recordings.",
"placeholders_info": "{course_title}, {course_name}, {student_name}, {fullname}"
},
{
"notification_type": "payment_successful",
"name": "تراکنش موفق مالی",
"title_fa": "پرداخت موفقیت‌آمیز",
"title_en": "Payment Successful",
"body_fa": "پرداخت شما برای دوره «{course_title}» با موفقیت انجام شد.",
"body_en": "Your payment for course '{course_title}' was successful.",
"placeholders_info": "{course_title}, {course_name}, {student_name}, {fullname}"
},
{
"notification_type": "payment_failed",
"name": "تراکنش ناموفق مالی",
"title_fa": "خطا در پرداخت",
"title_en": "Payment Failed",
"body_fa": "در پردازش پرداخت شما برای دوره «{course_title}» مشکلی رخ داد.",
"body_en": "We encountered an issue processing your payment for '{course_title}'.",
"placeholders_info": "{course_title}, {course_name}, {student_name}, {fullname}"
},
{
"notification_type": "refund_completed",
"name": "بازگشت وجه مالی",
"title_fa": "بازگشت وجه انجام شد",
"title_en": "Refund Completed",
"body_fa": "بازگشت وجه برای دوره «{course_title}» انجام شد.",
"body_en": "A refund has been successfully completed for the course '{course_title}'.",
"placeholders_info": "{course_title}, {course_name}, {student_name}, {fullname}"
},
{
"notification_type": "student_inactivity",
"name": "عدم فعالیت کاربر",
"title_fa": "دلمان برایتان تنگ شده!",
"title_en": "We Miss You!",
"body_fa": "در یک هفته گذشته هیچ درسی را مطالعه نکرده‌اید. منتظرتان هستیم تا مسیر یادگیری خود را ادامه دهید.",
"body_en": "You haven't taken any lessons in the last week. Come back and continue your learning journey!",
"placeholders_info": "{student_name}, {fullname}"
},
{
"notification_type": "teacher_reply",
"name": "پاسخ پیام استاد در چت",
"title_fa": "پاسخ جدید از استاد",
"title_en": "New Message from Teacher",
"body_fa": "استاد به پیام شما پاسخ داده است.",
"body_en": "The teacher has replied to your message.",
"placeholders_info": "{student_name}, {fullname}"
},
{
"notification_type": "support_reply",
"name": "پاسخ پیام پشتیبان در چت",
"title_fa": "پاسخ جدید از پشتیبان",
"title_en": "New Message from Support",
"body_fa": "پشتیبان به پیام شما پاسخ داده است.",
"body_en": "The support agent has replied to your message.",
"placeholders_info": "{student_name}, {fullname}"
}
]
count = 0
for item in default_templates:
obj, created = NotificationTemplate.objects.update_or_create(
notification_type=item['notification_type'],
defaults={
'name': item['name'],
'title_fa': item['title_fa'],
'title_en': item['title_en'],
'body_fa': item['body_fa'],
'body_en': item['body_en'],
'placeholders_info': item['placeholders_info'],
'is_active': True
}
)
if created:
count += 1
self.stdout.write(self.style.SUCCESS(f"Seeding completed! Created {count} new templates, updated existing ones."))

53
apps/account/migrations/0007_notificationtemplate_schedulednotification.py
View File

@ -0,0 +1,53 @@
# Generated by Django 4.2.30 on 2026-07-11 13:28
from django.db import migrations, models
import django.db.models.deletion
class Migration(migrations.Migration):
dependencies = [
('django_celery_beat', '0018_improve_crontab_helptext'),
('account', '0006_notification_action_notification_navigate_to_and_more'),
]
operations = [
migrations.CreateModel(
name='NotificationTemplate',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('notification_type', models.CharField(max_length=50, unique=True, verbose_name='notification type')),
('name', models.CharField(max_length=100, verbose_name='name')),
('is_active', models.BooleanField(default=True, verbose_name='is active')),
('title_fa', models.CharField(max_length=255, verbose_name='title (FA)')),
('title_en', models.CharField(max_length=255, verbose_name='title (EN)')),
('body_fa', models.TextField(max_length=1024, verbose_name='body (FA)')),
('body_en', models.TextField(max_length=1024, verbose_name='body (EN)')),
('placeholders_info', models.CharField(blank=True, max_length=255, null=True, verbose_name='allowed placeholders description')),
],
options={
'verbose_name': 'Notification Template',
'verbose_name_plural': 'Notification Templates',
},
),
migrations.CreateModel(
name='ScheduledNotification',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('name', models.CharField(max_length=100, verbose_name='campaign name')),
('target_group', models.CharField(choices=[('all_students', 'All Students'), ('inactive_students', 'Students Inactive for > 7 Days'), ('all_professors', 'All Professors')], max_length=50, verbose_name='target group')),
('is_active', models.BooleanField(default=True, verbose_name='is active')),
('schedule_type', models.CharField(choices=[('daily', 'Daily'), ('weekly', 'Weekly'), ('monthly', 'Monthly'), ('cron', 'Custom Cron Expression')], default='daily', max_length=20, verbose_name='schedule type')),
('time_of_day', models.TimeField(default='09:00:00', verbose_name='time of day')),
('days_of_week', models.CharField(blank=True, help_text='Comma-separated days (1 for Mon, 6 for Sat, 0 for Sun). E.g. "6,0" for Sat and Sun.', max_length=50, null=True, verbose_name='days of week')),
('days_of_month', models.CharField(blank=True, help_text='Comma-separated days of month (1-31). E.g. "1" for first day of the month.', max_length=50, null=True, verbose_name='days of month')),
('custom_cron', models.CharField(blank=True, help_text='Standard 5-field cron expression: minute hour day_of_month month day_of_week', max_length=100, null=True, verbose_name='custom cron expression')),
('periodic_task', models.OneToOneField(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='scheduled_notification', to='django_celery_beat.periodictask')),
('template', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='account.notificationtemplate', verbose_name='template')),
],
options={
'verbose_name': 'Scheduled Notification',
'verbose_name_plural': 'Scheduled Notifications',
},
),
]

122
apps/account/models/notification.py
View File

@ -31,3 +31,125 @@ class Notification(models.Model):
return self.title
class NotificationTemplate(models.Model):
notification_type = models.CharField(max_length=50, unique=True, verbose_name=_('notification type'))
name = models.CharField(max_length=100, verbose_name=_('name'))
is_active = models.BooleanField(default=True, verbose_name=_('is active'))
title_fa = models.CharField(max_length=255, verbose_name=_('title (FA)'))
title_en = models.CharField(max_length=255, verbose_name=_('title (EN)'))
body_fa = models.TextField(max_length=1024, verbose_name=_('body (FA)'))
body_en = models.TextField(max_length=1024, verbose_name=_('body (EN)'))
placeholders_info = models.CharField(max_length=255, blank=True, null=True, verbose_name=_('allowed placeholders description'))
class Meta:
verbose_name = _('Notification Template')
verbose_name_plural = _('Notification Templates')
def __str__(self):
return f"{self.name} ({self.notification_type})"
class ScheduledNotification(models.Model):
class TargetGroupChoices(models.TextChoices):
ALL_STUDENTS = 'all_students', _('All Students')
INACTIVE_STUDENTS = 'inactive_students', _('Students Inactive for > 7 Days')
ALL_PROFESSORS = 'all_professors', _('All Professors')
class ScheduleTypeChoices(models.TextChoices):
DAILY = 'daily', _('Daily')
WEEKLY = 'weekly', _('Weekly')
MONTHLY = 'monthly', _('Monthly')
CRON = 'cron', _('Custom Cron Expression')
name = models.CharField(max_length=100, verbose_name=_('campaign name'))
template = models.ForeignKey(NotificationTemplate, on_delete=models.CASCADE, verbose_name=_('template'))
target_group = models.CharField(max_length=50, choices=TargetGroupChoices.choices, verbose_name=_('target group'))
is_active = models.BooleanField(default=True, verbose_name=_('is active'))
schedule_type = models.CharField(max_length=20, choices=ScheduleTypeChoices.choices, default=ScheduleTypeChoices.DAILY, verbose_name=_('schedule type'))
time_of_day = models.TimeField(default="09:00:00", verbose_name=_('time of day'))
days_of_week = models.CharField(max_length=50, blank=True, null=True, help_text=_('Comma-separated days (1 for Mon, 6 for Sat, 0 for Sun). E.g. "6,0" for Sat and Sun.'), verbose_name=_('days of week'))
days_of_month = models.CharField(max_length=50, blank=True, null=True, help_text=_('Comma-separated days of month (1-31). E.g. "1" for first day of the month.'), verbose_name=_('days of month'))
custom_cron = models.CharField(max_length=100, blank=True, null=True, help_text=_('Standard 5-field cron expression: minute hour day_of_month month day_of_week'), verbose_name=_('custom cron expression'))
periodic_task = models.OneToOneField(
'django_celery_beat.PeriodicTask',
on_delete=models.SET_NULL,
null=True,
blank=True,
related_name='scheduled_notification'
)
class Meta:
verbose_name = _('Scheduled Notification')
verbose_name_plural = _('Scheduled Notifications')
def __str__(self):
return self.name
def save(self, *args, **kwargs):
from django_celery_beat.models import CrontabSchedule, PeriodicTask
import json
from django.conf import settings
# Calculate cron fields based on schedule_type
minute = str(self.time_of_day.minute)
hour = str(self.time_of_day.hour)
day_of_month = '*'
month_of_year = '*'
day_of_week = '*'
if self.schedule_type == self.ScheduleTypeChoices.DAILY:
pass
elif self.schedule_type == self.ScheduleTypeChoices.WEEKLY:
day_of_week = self.days_of_week or '*'
elif self.schedule_type == self.ScheduleTypeChoices.MONTHLY:
day_of_month = self.days_of_month or '1'
elif self.schedule_type == self.ScheduleTypeChoices.CRON:
if self.custom_cron:
parts = self.custom_cron.split()
if len(parts) >= 5:
minute, hour, day_of_month, month_of_year, day_of_week = parts[:5]
schedule, _ = CrontabSchedule.objects.get_or_create(
minute=minute,
hour=hour,
day_of_week=day_of_week,
day_of_month=day_of_month,
month_of_year=month_of_year,
timezone=getattr(settings, 'CELERY_TIMEZONE', 'Asia/Tehran')
)
task_name = f"Scheduled Notification: {self.name}"
task_kwargs = json.dumps({
'campaign_id': self.id
})
if self.periodic_task:
task = self.periodic_task
task.name = task_name
task.crontab = schedule
task.enabled = self.is_active
task.kwargs = task_kwargs
task.save()
else:
task = PeriodicTask.objects.create(
name=task_name,
task='apps.account.tasks.run_scheduled_notification_task',
crontab=schedule,
enabled=self.is_active,
kwargs=task_kwargs
)
self.periodic_task = task
super().save(*args, **kwargs)
def delete(self, *args, **kwargs):
if self.periodic_task:
self.periodic_task.delete()
super().delete(*args, **kwargs)

138
apps/account/notification_service.py
View File

@ -1,10 +1,106 @@
import asyncio
import logging
from apps.account.models import Notification
from apps.account.tasks import send_notification
from apps.account.models import Notification, NotificationTemplate
logger = logging.getLogger(__name__)
class SafeDict(dict):
def __missing__(self, key):
return f"{{{key}}}"
def get_template_context(user, notification_type, data):
"""
Builds a dynamic context dictionary containing user info and target entity details
(course, lesson, quiz, session) to format templates.
"""
context = {
'student_name': getattr(user, 'fullname', user.username) or user.username,
'fullname': getattr(user, 'fullname', user.username) or user.username,
'username': user.username,
}
if not data or not isinstance(data, dict):
return context
# Import models dynamically inside function to avoid circular imports
from apps.course.models.course import Course
from apps.course.models.lesson import CourseLesson
from apps.course.models.live_session import CourseLiveSession
from apps.quiz.models import Quiz
try:
# 1. Course details
course_id = data.get('course_id')
if not course_id and 'session_id' in data:
session = CourseLiveSession.objects.filter(id=data['session_id']).first()
if session:
course_id = session.course_id
if course_id:
course = Course.objects.filter(id=course_id).first()
if course:
title_val = course.title
# course.title might be a JSON list or dict or string
if isinstance(title_val, list) and title_val:
course_title = title_val[0].get('title', '')
elif isinstance(title_val, dict):
course_title = title_val.get('title', '')
else:
course_title = str(title_val)
context['course_title'] = course_title
context['course_name'] = course_title
slug_val = course.slug
if isinstance(slug_val, list) and slug_val:
course_slug = slug_val[0].get('title', '')
elif isinstance(slug_val, dict):
course_slug = slug_val.get('title', '')
else:
course_slug = str(slug_val)
context['course_slug'] = course_slug
# 2. Lesson details
lesson_id = data.get('lesson_id')
if lesson_id:
lesson = CourseLesson.objects.filter(id=lesson_id).first()
if lesson:
title_val = lesson.title
if isinstance(title_val, list) and title_val:
lesson_title = title_val[0].get('title', '')
elif isinstance(title_val, dict):
lesson_title = title_val.get('title', '')
else:
lesson_title = str(title_val)
context['lesson_title'] = lesson_title
# 3. Quiz details
quiz_id = data.get('quiz_id')
if quiz_id:
quiz = Quiz.objects.filter(id=quiz_id).first()
if quiz:
title_val = quiz.title
if isinstance(title_val, list) and title_val:
quiz_title = title_val[0].get('title', '')
elif isinstance(title_val, dict):
quiz_title = title_val.get('title', '')
else:
quiz_title = str(title_val)
context['quiz_title'] = quiz_title
# 4. Live Session details
session_id = data.get('session_id')
if session_id:
session = CourseLiveSession.objects.filter(id=session_id).first()
if session:
context['session_subject'] = session.subject
if session.start_at:
context['session_start_time'] = session.start_at.strftime('%H:%M')
except Exception as e:
logger.error(f"Error building notification template context: {e}")
return context
def resolve_notification_route(notification_type, data):
"""
Resolves standard GoRouter paths based on notification type and payload data.
@ -75,23 +171,51 @@ def create_and_send_notification(user, title_en, body_en, title_fa, body_fa, ser
data=None, notification_type=None, action='navigate', navigate_to=None):
"""
Creates a Notification record in the database and sends a push notification to FCM.
Resolves localized title and body based on user.language.
Loads and renders the NotificationTemplate from the database if present.
If the template is inactive, the notification is discarded.
"""
# Auto-resolve notification type from data payload if not provided
if not notification_type and isinstance(data, dict):
notification_type = data.get('type')
# 1. Check for template settings in the database
if notification_type:
template = NotificationTemplate.objects.filter(notification_type=notification_type).first()
if template:
if not template.is_active:
logger.info(f"Notification type '{notification_type}' is disabled via database template settings.")
return None
# Override templates with database values
title_fa = template.title_fa
title_en = template.title_en
body_fa = template.body_fa
body_en = template.body_en
# 2. Render templates safely with dynamic context
context = get_template_context(user, notification_type, data)
safe_ctx = SafeDict(context)
try:
title_fa = title_fa.format(**safe_ctx) if title_fa else ""
title_en = title_en.format(**safe_ctx) if title_en else ""
body_fa = body_fa.format(**safe_ctx) if body_fa else ""
body_en = body_en.format(**safe_ctx) if body_en else ""
except Exception as e:
logger.error(f"Error formatting templates: {e}")
# Determine user localized text
lang = getattr(user, 'language', None)
lang_code = lang.code if lang and hasattr(lang, 'code') else 'fa'
title = title_fa if lang_code == 'fa' else title_en
message = body_fa if lang_code == 'fa' else body_en
# Auto-resolve notification type from data payload if not provided
if not notification_type and isinstance(data, dict):
notification_type = data.get('type')
# Auto-resolve GoRouter navigation route if not provided
if not navigate_to and notification_type:
navigate_to = resolve_notification_route(notification_type, data)
# Save to database
from apps.account.tasks import send_notification
notif = Notification.objects.create(
user=user,
title=title,

49
apps/account/tasks.py
View File

@ -397,6 +397,55 @@ def check_inactive_students_task():
)
@shared_task
def run_scheduled_notification_task(campaign_id):
"""
Celery task run by Celery Beat to execute a ScheduledNotification campaign.
"""
from apps.account.models import ScheduledNotification
from apps.account.notification_service import create_and_send_notification
from django.contrib.auth import get_user_model
from django.utils import timezone
from datetime import timedelta
campaign = ScheduledNotification.objects.filter(id=campaign_id).first()
if not campaign or not campaign.is_active or not campaign.template.is_active:
logger.info(f"[Scheduled Notification Task] Campaign {campaign_id} not found or disabled.")
return
template = campaign.template
User = get_user_model()
users = User.objects.filter(is_active=True)
# 1. Filter target audience
if campaign.target_group == ScheduledNotification.TargetGroupChoices.INACTIVE_STUDENTS:
seven_days_ago = timezone.now() - timedelta(days=7)
users = users.filter(
user_type='student',
last_login__lt=seven_days_ago
)
elif campaign.target_group == ScheduledNotification.TargetGroupChoices.ALL_STUDENTS:
users = users.filter(user_type='student')
elif campaign.target_group == ScheduledNotification.TargetGroupChoices.ALL_PROFESSORS:
users = users.filter(user_type='professor')
logger.info(f"[Scheduled Notification Task] Sending campaign '{campaign.name}' to {users.count()} users...")
# 2. Iterate and send
for user in users:
create_and_send_notification(
user=user,
title_en=template.title_en,
body_en=template.body_en,
title_fa=template.title_fa,
body_fa=template.body_fa,
service='imam-javad',
data={'type': template.notification_type},
notification_type=template.notification_type
)

97
apps/account/tests/notifications/test_notifications.py
View File

@ -217,3 +217,100 @@ class NotificationFlowTests(TestCase):
notif = Notification.objects.filter(user=student_recipient, title="ثبت‌نام دوره جدید").first()
self.assertIsNotNone(notif)
self.assertIn("دوره جدید الف", notif.message)
@patch('apps.account.notification_service.send_notification')
def test_notification_template_customization_and_deactivation(self, mock_send):
from apps.account.models import NotificationTemplate
# 1. Create a custom template in the database
template = NotificationTemplate.objects.create(
notification_type="custom_test_type",
name="Test Template",
title_fa="سلام {student_name} به دوره {course_title}",
title_en="Hello {student_name} to course {course_title}",
body_fa="توضیحات فارسی {student_name}",
body_en="English description {student_name}",
is_active=True
)
course = self._create_course('test-tpl-course', 'Tpl Course', 'دوره تی‌پی‌ال')
# 2. Trigger notification and verify custom text formatting
from apps.account.notification_service import create_and_send_notification
create_and_send_notification(
user=self.student,
title_en="Default Title EN",
body_en="Default Body EN",
title_fa="Default Title FA",
body_fa="Default Body FA",
service='imam-javad',
data={'type': 'custom_test_type', 'course_id': course.id}
)
notif = Notification.objects.filter(user=self.student).order_by('-id').first()
self.assertIsNotNone(notif)
# Verify it loaded from template and formatted using course name & student name
self.assertEqual(notif.title, f"سلام {self.student.fullname} به دوره دوره تی‌پی‌ال")
self.assertEqual(notif.message, f"توضیحات فارسی {self.student.fullname}")
# 3. Disable template and verify notification is discarded
template.is_active = False
template.save()
count_before = Notification.objects.filter(user=self.student).count()
result = create_and_send_notification(
user=self.student,
title_en="Default Title EN",
body_en="Default Body EN",
title_fa="Default Title FA",
body_fa="Default Body FA",
service='imam-javad',
data={'type': 'custom_test_type', 'course_id': course.id}
)
self.assertIsNone(result)
count_after = Notification.objects.filter(user=self.student).count()
self.assertEqual(count_before, count_after)
def test_scheduled_notification_syncs_periodic_task(self):
from apps.account.models import NotificationTemplate, ScheduledNotification
from django_celery_beat.models import PeriodicTask
template = NotificationTemplate.objects.create(
notification_type="sched_test",
name="Scheduled Test",
title_fa="تست زمان‌بندی",
title_en="Scheduled Test",
body_fa="تست",
body_en="Test"
)
# Create ScheduledNotification
sched_notif = ScheduledNotification.objects.create(
name="My Weekly Campaign",
template=template,
target_group=ScheduledNotification.TargetGroupChoices.ALL_STUDENTS,
schedule_type=ScheduledNotification.ScheduleTypeChoices.WEEKLY,
days_of_week="6,0",
time_of_day="14:30:00",
is_active=True
)
# Verify PeriodicTask and CrontabSchedule are created
self.assertIsNotNone(sched_notif.periodic_task)
pt = sched_notif.periodic_task
self.assertEqual(pt.name, "Scheduled Notification: My Weekly Campaign")
self.assertEqual(pt.crontab.minute, "30")
self.assertEqual(pt.crontab.hour, "14")
self.assertEqual(pt.crontab.day_of_week, "6,0")
self.assertTrue(pt.enabled)
# Deactivate and verify sync
sched_notif.is_active = False
sched_notif.save()
pt.refresh_from_db()
self.assertFalse(pt.enabled)
# Delete and verify dynamic cleanup
task_id = pt.id
sched_notif.delete()
self.assertFalse(PeriodicTask.objects.filter(id=task_id).exists())

2
config/settings/base.py
View File

@ -78,7 +78,7 @@ THIRD_PARTY_APPS = [
'dj_category',
'corsheaders',
'django_filters',
'django_celery_beat',
]
INSTALLED_APPS = [
"unfold",

1
requirements.txt
View File

@ -132,6 +132,7 @@ geoip2==4.7.0
google-auth==2.6.0
pyjwt
cryptography>=41.0.0
django-celery-beat==2.5.0
https://yaghoubi:e07059e0ac6be3b0032ded5f65f03363fbd3811f@git.habibapp.com/NewHorizon/django-limitless-dashboard.git/archive/master.zip
https://yaghoubi:e07059e0ac6be3b0032ded5f65f03363fbd3811f@git.habibapp.com/NewHorizon/ajax-datatable.git/archive/master.zip

Write Preview
Loading…
Cancel
Save