7 changed files with 366 additions and 0 deletions
-
64apps/account/tasks.py
-
176apps/account/tests/notifications/test_quiz_notifications.py
-
27apps/hadis/management/commands/detach_edition_from_volume.py
-
27apps/hadis/management/commands/find_orphan_volumes.py
-
64apps/hadis/management/commands/seed_book_excerpts.py
-
4config/settings/base.py
-
4config/settings/production.py
@ -0,0 +1,176 @@ |
|||
import datetime |
|||
from django.utils import timezone |
|||
from django.test import TestCase |
|||
from unittest.mock import patch |
|||
from django.core.files.uploadedfile import SimpleUploadedFile |
|||
|
|||
from dj_language.models import Language |
|||
from apps.account.models import StudentUser, User, Notification |
|||
from apps.course.models.course import Course, CourseCategory |
|||
from apps.quiz.models import Quiz, QuizParticipant |
|||
from apps.account.tasks import check_low_quiz_results_task |
|||
|
|||
class QuizNotificationFlowTests(TestCase): |
|||
def setUp(self): |
|||
# Create languages |
|||
self.lang_fa, _ = Language.objects.update_or_create( |
|||
code='fa', |
|||
defaults={'name': 'Persian', 'status': True, 'countries': []} |
|||
) |
|||
self.lang_en, _ = Language.objects.update_or_create( |
|||
code='en', |
|||
defaults={'name': 'English', 'status': True, 'countries': []} |
|||
) |
|||
|
|||
# Create category |
|||
self.category = CourseCategory.objects.create( |
|||
name=[{"title": "Category", "language_code": "en"}], |
|||
slug=[{"title": "category", "language_code": "en"}] |
|||
) |
|||
|
|||
# Create student user |
|||
self.student = StudentUser.objects.create( |
|||
email='student_quiz@example.com', |
|||
username='student_quiz', |
|||
fullname='Quiz Student', |
|||
language=self.lang_fa, |
|||
fcm='token_quiz_student' |
|||
) |
|||
|
|||
# Create Course |
|||
thumbnail = SimpleUploadedFile('thumb.jpg', b'filecontent', content_type='image/jpeg') |
|||
self.course = Course.objects.create( |
|||
title=[ |
|||
{"title": "Course 1", "language_code": "en"}, |
|||
{"title": "دوره ۱", "language_code": "fa"} |
|||
], |
|||
slug=[{"title": "course-1", "language_code": "en"}], |
|||
category=self.category, |
|||
thumbnail=thumbnail, |
|||
video_type=Course.VedioTypeChoices.YOUTUBE_LINK, |
|||
video_link='https://example.com/video', |
|||
is_online=False, |
|||
level=Course.LevelChoices.BEGINNER, |
|||
duration=10, |
|||
lessons_count=0, |
|||
description='Description', |
|||
short_description='Short description', |
|||
status=[{"title": "ongoing", "language_code": "en"}], |
|||
is_free=True, |
|||
) |
|||
|
|||
# Create Quiz |
|||
self.quiz = Quiz.objects.create( |
|||
course=self.course, |
|||
title=[ |
|||
{"title": "Quiz A", "language_code": "en"}, |
|||
{"title": "آزمون الف", "language_code": "fa"} |
|||
], |
|||
each_question_timing=60, |
|||
status=True |
|||
) |
|||
|
|||
@patch('apps.account.notification_service.send_notification') |
|||
def test_low_quiz_result_triggers_on_intervals(self, mock_send): |
|||
# Event: Low Quiz Result (score < 70) |
|||
# Create failed attempt exactly 2 days ago |
|||
attempt_2_days = QuizParticipant.objects.create( |
|||
quiz=self.quiz, |
|||
user=self.student, |
|||
started_at=timezone.now() - datetime.timedelta(days=2, hours=1), |
|||
ended_at=timezone.now() - datetime.timedelta(days=2), |
|||
total_timing=300, |
|||
question_score=50, |
|||
timing_score=0, |
|||
total_score=50 |
|||
) |
|||
|
|||
# Run periodic task |
|||
check_low_quiz_results_task() |
|||
|
|||
# Verify reminder sent to the student (Persian template check) |
|||
notif = Notification.objects.filter(user=self.student, title="پیشنهاد شرکت مجدد در آزمون").first() |
|||
self.assertIsNotNone(notif) |
|||
self.assertIn("آزمون الف", notif.message) |
|||
self.assertTrue(mock_send.called) |
|||
|
|||
@patch('apps.account.notification_service.send_notification') |
|||
def test_no_reminder_if_passed_initially(self, mock_send): |
|||
# Create successful attempt exactly 2 days ago (score = 80) |
|||
QuizParticipant.objects.create( |
|||
quiz=self.quiz, |
|||
user=self.student, |
|||
started_at=timezone.now() - datetime.timedelta(days=2, hours=1), |
|||
ended_at=timezone.now() - datetime.timedelta(days=2), |
|||
total_timing=300, |
|||
question_score=80, |
|||
timing_score=0, |
|||
total_score=80 |
|||
) |
|||
|
|||
check_low_quiz_results_task() |
|||
|
|||
self.assertFalse(Notification.objects.filter(user=self.student, title="پیشنهاد شرکت مجدد در آزمون").exists()) |
|||
self.assertFalse(mock_send.called) |
|||
|
|||
@patch('apps.account.notification_service.send_notification') |
|||
def test_no_reminder_if_retaken_and_passed(self, mock_send): |
|||
# Create failed attempt exactly 7 days ago |
|||
attempt_failed = QuizParticipant.objects.create( |
|||
quiz=self.quiz, |
|||
user=self.student, |
|||
started_at=timezone.now() - datetime.timedelta(days=7, hours=1), |
|||
ended_at=timezone.now() - datetime.timedelta(days=7), |
|||
total_timing=300, |
|||
question_score=40, |
|||
timing_score=0, |
|||
total_score=40 |
|||
) |
|||
|
|||
# Create a newer successful attempt 3 days ago |
|||
QuizParticipant.objects.create( |
|||
quiz=self.quiz, |
|||
user=self.student, |
|||
started_at=timezone.now() - datetime.timedelta(days=3, hours=1), |
|||
ended_at=timezone.now() - datetime.timedelta(days=3), |
|||
total_timing=300, |
|||
question_score=90, |
|||
timing_score=0, |
|||
total_score=90 |
|||
) |
|||
|
|||
check_low_quiz_results_task() |
|||
|
|||
# Should not suggest retake since they passed subsequently |
|||
self.assertFalse(Notification.objects.filter(user=self.student, title="پیشنهاد شرکت مجدد در آزمون").exists()) |
|||
|
|||
@patch('apps.account.notification_service.send_notification') |
|||
def test_no_reminder_if_already_retaken_even_if_failed(self, mock_send): |
|||
# Create failed attempt exactly 21 days ago |
|||
attempt_failed = QuizParticipant.objects.create( |
|||
quiz=self.quiz, |
|||
user=self.student, |
|||
started_at=timezone.now() - datetime.timedelta(days=21, hours=1), |
|||
ended_at=timezone.now() - datetime.timedelta(days=21), |
|||
total_timing=300, |
|||
question_score=30, |
|||
timing_score=0, |
|||
total_score=30 |
|||
) |
|||
|
|||
# Create a newer attempt 10 days ago (also failed, e.g. score = 50) |
|||
# Since they already attempted a retake 10 days ago, we shouldn't trigger the 21-day reminder of the first failure. |
|||
QuizParticipant.objects.create( |
|||
quiz=self.quiz, |
|||
user=self.student, |
|||
started_at=timezone.now() - datetime.timedelta(days=10, hours=1), |
|||
ended_at=timezone.now() - datetime.timedelta(days=10), |
|||
total_timing=300, |
|||
question_score=50, |
|||
timing_score=0, |
|||
total_score=50 |
|||
) |
|||
|
|||
check_low_quiz_results_task() |
|||
|
|||
self.assertFalse(Notification.objects.filter(user=self.student, title="پیشنهاد شرکت مجدد در آزمون").exists()) |
|||
@ -0,0 +1,27 @@ |
|||
# backend/apps/hadis/management/commands/make_books_independent.py |
|||
|
|||
import random |
|||
from django.core.management.base import BaseCommand |
|||
from django.db import transaction |
|||
from apps.hadis.models import BookReference, BookEdition |
|||
|
|||
class Command(BaseCommand): |
|||
help = 'Makes 50% of BookReferences independent by deleting their BookEditions.' |
|||
|
|||
def handle(self, *args, **options): |
|||
all_books = list(BookReference.objects.all()) |
|||
if not all_books: |
|||
self.stdout.write(self.style.ERROR("No books found.")) |
|||
return |
|||
|
|||
# انتخاب ۵۰ درصد رندوم |
|||
num_to_detach = len(all_books) // 2 |
|||
books_to_detach = random.sample(all_books, num_to_detach) |
|||
|
|||
self.stdout.write(self.style.WARNING(f"Deleting editions from {num_to_detach} books...")) |
|||
|
|||
with transaction.atomic(): |
|||
# حذف تمام ادیشنهای مرتبط با این ۵۰ درصد کتاب |
|||
BookEdition.objects.filter(book_reference__in=books_to_detach).delete() |
|||
|
|||
self.stdout.write(self.style.SUCCESS(f"✅ {num_to_detach} books are now independent (no editions).")) |
|||
@ -0,0 +1,27 @@ |
|||
# backend/apps/hadis/management/commands/find_orphan_volumes.py |
|||
|
|||
from django.core.management.base import BaseCommand |
|||
from apps.hadis.models import BookReference |
|||
|
|||
class Command(BaseCommand): |
|||
help = 'Prints the slugs of BookReferences that have volumes but no associated editions.' |
|||
|
|||
def handle(self, *args, **options): |
|||
# فیلتر کردن رفرنسهایی که تعداد ادیشنهای آنها صفر است اما تعداد ولومهای آنها بیشتر از صفر است |
|||
orphaned_references = BookReference.objects.filter( |
|||
editions__isnull=True, |
|||
volumes__isnull=False |
|||
).distinct() |
|||
|
|||
if not orphaned_references.exists(): |
|||
self.stdout.write(self.style.SUCCESS("✅ No orphaned volumes found. All volumes have associated editions.")) |
|||
return |
|||
|
|||
self.stdout.write(self.style.WARNING("Found the following BookReference slugs (have volumes, no edition):")) |
|||
self.stdout.write("-" * 40) |
|||
|
|||
for book in orphaned_references: |
|||
self.stdout.write(f"Slug: {book.slug} | ID: {book.id}") |
|||
|
|||
self.stdout.write("-" * 40) |
|||
self.stdout.write(self.style.SUCCESS(f"Total found: {orphaned_references.count()}")) |
|||
@ -0,0 +1,64 @@ |
|||
# backend/apps/hadis/management/commands/seed_book_excerpts.py |
|||
|
|||
import random |
|||
from django.core.management.base import BaseCommand |
|||
from django.db import transaction |
|||
from apps.hadis.models import BookReference, Hadis, HadisReference |
|||
|
|||
class Command(BaseCommand): |
|||
help = 'Assigns 4-5 random Hadis excerpts to all BookReferences.' |
|||
|
|||
def handle(self, *args, **options): |
|||
self.stdout.write(self.style.WARNING("Scanning books and hadiths...")) |
|||
|
|||
books = BookReference.objects.all() |
|||
# گرفتن تمام آیدیهای احادیث به صورت یک لیست تخت برای انتخاب رندوم سریع |
|||
all_hadis_ids = list(Hadis.objects.filter(status=True).values_list('id', flat=True)) |
|||
|
|||
if not books.exists(): |
|||
self.stdout.write(self.style.ERROR("❌ No BookReferences found in the database.")) |
|||
return |
|||
|
|||
if not all_hadis_ids: |
|||
self.stdout.write(self.style.ERROR("❌ No active Hadis found to assign as excerpts.")) |
|||
return |
|||
|
|||
new_references = [] |
|||
total_books = books.count() |
|||
self.stdout.write(self.style.WARNING(f"Processing {total_books} books...")) |
|||
|
|||
for book in books: |
|||
# انتخاب تصادفی ۴ یا ۵ اکسرپت |
|||
num_excerpts = random.randint(4, 5) |
|||
|
|||
# در صورتی که تعداد کل احادیث کمتر از ۵ تا باشد |
|||
num_excerpts = min(num_excerpts, len(all_hadis_ids)) |
|||
|
|||
# انتخاب رندوم آیدی احادیث |
|||
selected_hadis_ids = random.sample(all_hadis_ids, num_excerpts) |
|||
|
|||
# استخراج احادیثی که از قبل به این کتاب وصل بودهاند (برای جلوگیری از ثبت تکراری) |
|||
existing_hadis_ids = set(HadisReference.objects.filter( |
|||
book_reference=book, |
|||
hadis_id__in=selected_hadis_ids |
|||
).values_list('hadis_id', flat=True)) |
|||
|
|||
for h_id in selected_hadis_ids: |
|||
if h_id not in existing_hadis_ids: |
|||
new_references.append( |
|||
HadisReference( |
|||
book_reference=book, |
|||
hadis_id=h_id, |
|||
# تولید دیتای دامی برای فیلدهای متنی رفرنس |
|||
volume=f"Vol. {random.randint(1, 10)}", |
|||
pages=str(random.randint(15, 450)), |
|||
hadith_number=str(random.randint(100, 9999)) |
|||
) |
|||
) |
|||
|
|||
if new_references: |
|||
with transaction.atomic(): |
|||
HadisReference.objects.bulk_create(new_references, batch_size=1000) |
|||
self.stdout.write(self.style.SUCCESS(f"🎉 Successfully created {len(new_references)} new excerpts for books!")) |
|||
else: |
|||
self.stdout.write(self.style.SUCCESS("✅ All books already have these excerpts or no new ones needed.")) |
|||
Write
Preview
Loading…
Cancel
Save
Reference in new issue