ExecuteCampaignRealConcurrencyTests had been written off as a known flake for weeks — about 45% red on a full-suite run. It was not timing noise. It was reporting a real defect, and the defect is in production code. execute_campaign looped with .iterator(), which keeps the read cursor on the user table open for the whole loop — a loop that writes a CampaignDelivery per user and, on the sms branch, makes a network call to Kavenegar in between. On SQLite that open cursor locks the table against any concurrent writer, and the lock is not released until the loop ends. So _retry_on_sqlite_lock could never recover from it no matter how long it waited: widening its window from 0.5s to 2s (also done here, since the linear backoff was too tight regardless) moved the failure rate not at all on its own. Both branches now materialise the ids before the first write. After the change: 20 consecutive runs of the test itself and 3 full-suite runs, all green. This is not only a test fix. While we are on SQLite the same pattern produces the same lock in production, where it surfaces as "database is locked" on unrelated requests that happen to land during a campaign run. Also in this pass ----------------- - CLAUDE.md carried three sections each claiming to be the current state, with three different question counts (1,441 / 1,783 / reality 2,463). There is now one current-state section and the older two are labelled as history, so whoever picks this up next is not reading stale numbers as fact. Three other figures that had drifted are corrected against the database: the verdict split, the figure-gap share (881/~42% is now 508/~14% after the five books were re-read), and the node coverage. - Dropped four genuinely unused imports. The remaining pyflakes hits are the signal registrations in apps.py, which are imported for effect. - .env.example suggested gpt-4o for vision; gpt-4.1-mini measured both cheaper and more accurate and is what the server actually runs. Audited all 2,463 questions across every grade: no question with the wrong number of options, none with zero or two correct answers, no empty or duplicated option, no orphan lesson or node, no duplicate question inside a lesson. 458 tests pass. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
349 lines
15 KiB
Python
349 lines
15 KiB
Python
"""
|
||
تستهای وارد/خارج کردنِ سوالها.
|
||
|
||
نکتهی محوری: فایلِ بد نباید چیزی بنویسد. وارد کردنِ نیمهکاره بدترین حالت است
|
||
چون کسی که فایل را داده فکر میکند کار شده، ولی درس نصفه پر است.
|
||
"""
|
||
import json
|
||
|
||
from django.test import TestCase
|
||
|
||
from content.models import Grade, Lesson, Node, Question, Unit
|
||
from content.question_io import ImportError_, export_questions, import_questions, parse
|
||
|
||
|
||
class QuestionImportTests(TestCase):
|
||
def setUp(self):
|
||
grade = Grade.objects.create(title='پایه نهم', icon='', color='')
|
||
unit = Unit.objects.create(grade=grade, title='فصل ۱', order=1)
|
||
node = Node.objects.create(unit=unit, title='درس ۱', order=1)
|
||
self.lesson = Lesson.objects.create(node=node, title='نوع ۱')
|
||
|
||
def _payload(self, **over):
|
||
row = {
|
||
'lessonId': self.lesson.pk,
|
||
'prompt': 'حاصل ۲+۳ چند است؟',
|
||
'type': 'multiple_choice',
|
||
'choices': ['۴', '۵', '۶', '۷'],
|
||
'correct': 1,
|
||
}
|
||
row.update(over)
|
||
return json.dumps([row], ensure_ascii=False).encode()
|
||
|
||
def test_a_valid_file_creates_the_question_and_marks_one_choice_correct(self):
|
||
result = import_questions(parse(self._payload()))
|
||
self.assertEqual(result['created'], 1)
|
||
question = Question.objects.get()
|
||
self.assertEqual(question.choices.count(), 4)
|
||
self.assertEqual(question.choices.filter(is_correct=True).get().value, '۵')
|
||
|
||
def test_an_unknown_lesson_is_rejected_before_anything_is_written(self):
|
||
with self.assertRaises(ImportError_):
|
||
import_questions(parse(self._payload(lessonId=999999)))
|
||
self.assertEqual(Question.objects.count(), 0)
|
||
|
||
def test_a_bad_row_in_the_middle_rolls_the_whole_file_back(self):
|
||
good = json.loads(self._payload())[0]
|
||
bad = dict(good, lessonId=999999)
|
||
raw = json.dumps([good, bad], ensure_ascii=False).encode()
|
||
with self.assertRaises(ImportError_):
|
||
import_questions(parse(raw))
|
||
self.assertEqual(Question.objects.count(), 0, 'نباید نیمهکاره بنویسد')
|
||
|
||
def test_correct_out_of_range_is_rejected(self):
|
||
with self.assertRaises(ImportError_):
|
||
parse(self._payload(correct=9))
|
||
|
||
def test_duplicate_choices_are_rejected(self):
|
||
with self.assertRaises(ImportError_):
|
||
parse(self._payload(choices=['۵', '۵', '۶', '۷']))
|
||
|
||
def test_broken_json_says_where_it_broke(self):
|
||
with self.assertRaises(ImportError_) as ctx:
|
||
parse(b'[{"lessonId": 1,,}]')
|
||
self.assertIn('JSON', str(ctx.exception))
|
||
|
||
def test_replace_clears_the_previous_questions_of_that_lesson(self):
|
||
import_questions(parse(self._payload()))
|
||
import_questions(parse(self._payload(prompt='سوال تازه')), replace=True)
|
||
self.assertEqual(Question.objects.count(), 1)
|
||
self.assertEqual(Question.objects.get().prompt, 'سوال تازه')
|
||
|
||
def test_export_round_trips_back_through_import(self):
|
||
import_questions(parse(self._payload()))
|
||
exported = export_questions(Question.objects.all())
|
||
Question.objects.all().delete()
|
||
import_questions(parse(json.dumps(exported, ensure_ascii=False).encode()))
|
||
self.assertEqual(Question.objects.get().choices.filter(is_correct=True).get().value, '۵')
|
||
|
||
|
||
class HealthEndpointTests(TestCase):
|
||
def test_reports_ok_and_counts_content(self):
|
||
response = self.client.get('/api/health')
|
||
self.assertEqual(response.status_code, 200)
|
||
body = response.json()
|
||
self.assertEqual(body['status'], 'ok')
|
||
self.assertTrue(body['checks']['database']['ok'])
|
||
self.assertTrue(body['checks']['migrations']['ok'])
|
||
self.assertIn('questions', body['checks']['content'])
|
||
|
||
def test_never_leaks_a_secret(self):
|
||
raw = self.client.get('/api/health').content.decode()
|
||
for needle in ('sk-', 'SECRET', 'API_KEY', 'password'):
|
||
self.assertNotIn(needle, raw)
|
||
|
||
def test_needs_no_authentication(self):
|
||
self.assertEqual(self.client.get('/api/health').status_code, 200)
|
||
|
||
|
||
class CurriculumImportTests(TestCase):
|
||
"""
|
||
برنامهی درسی باید بدونِ Qdrant و بدونِ ترمینال به سرور برسد.
|
||
|
||
مهمترین خاصیتش این است که چیزی پاک نمیکند: یک فایلِ ناقص نباید محتوای
|
||
سرور را از بین ببرد.
|
||
"""
|
||
|
||
SAMPLE = {
|
||
'grades': [{
|
||
'title': 'پایه نهم', 'icon': 'grade-9', 'color': '#FF9800',
|
||
'units': [{
|
||
'title': 'فصل ۴: توان و ریشه', 'order': 4,
|
||
'nodes': [{
|
||
'title': 'توان صحیح', 'order': 1,
|
||
'lessons': [{'title': 'ضرب توانها', 'introText': 'پایهها برابرند.'}],
|
||
}],
|
||
}],
|
||
}],
|
||
}
|
||
|
||
def _raw(self, payload=None):
|
||
return json.dumps(payload or self.SAMPLE, ensure_ascii=False).encode()
|
||
|
||
def test_builds_the_whole_tree(self):
|
||
from content.curriculum_io import import_curriculum
|
||
from content.curriculum_io import parse as parse_curriculum
|
||
|
||
counts = import_curriculum(parse_curriculum(self._raw()))
|
||
self.assertEqual((counts['grades'], counts['units'], counts['nodes'], counts['lessons']),
|
||
(1, 1, 1, 1))
|
||
self.assertEqual(Lesson.objects.get().intro_text, 'پایهها برابرند.')
|
||
|
||
def test_running_it_twice_creates_nothing_new(self):
|
||
from content.curriculum_io import import_curriculum
|
||
from content.curriculum_io import parse as parse_curriculum
|
||
|
||
import_curriculum(parse_curriculum(self._raw()))
|
||
counts = import_curriculum(parse_curriculum(self._raw()))
|
||
self.assertEqual(counts['lessons'], 0)
|
||
self.assertEqual(Lesson.objects.count(), 1)
|
||
|
||
def test_a_partial_file_never_deletes_what_is_already_there(self):
|
||
from content.curriculum_io import import_curriculum
|
||
from content.curriculum_io import parse as parse_curriculum
|
||
|
||
import_curriculum(parse_curriculum(self._raw()))
|
||
smaller = json.loads(json.dumps(self.SAMPLE))
|
||
smaller['grades'][0]['units'] = []
|
||
import_curriculum(parse_curriculum(self._raw(smaller)))
|
||
self.assertEqual(Lesson.objects.count(), 1, 'محتوای موجود نباید پاک شود')
|
||
|
||
def test_a_missing_title_is_rejected_with_a_readable_message(self):
|
||
from content.curriculum_io import CurriculumError
|
||
from content.curriculum_io import parse as parse_curriculum
|
||
|
||
broken = json.loads(json.dumps(self.SAMPLE))
|
||
broken['grades'][0]['units'][0]['nodes'][0]['title'] = ''
|
||
with self.assertRaises(CurriculumError) as ctx:
|
||
parse_curriculum(self._raw(broken))
|
||
self.assertIn('درس', str(ctx.exception))
|
||
|
||
def test_export_round_trips_back_through_import(self):
|
||
from content.curriculum_io import export_curriculum, import_curriculum
|
||
from content.curriculum_io import parse as parse_curriculum
|
||
|
||
import_curriculum(parse_curriculum(self._raw()))
|
||
exported = export_curriculum(Grade.objects.all())
|
||
Grade.objects.all().delete()
|
||
import_curriculum(parse_curriculum(json.dumps(exported, ensure_ascii=False).encode()))
|
||
self.assertEqual(Lesson.objects.get().title, 'ضرب توانها')
|
||
|
||
|
||
class HealthWarningTests(TestCase):
|
||
def test_debug_mode_is_reported_as_a_warning(self):
|
||
with self.settings(DEBUG=True, DEV_MASTER_OTP='0000'):
|
||
body = self.client.get('/api/health').json()
|
||
joined = ' '.join(body['warnings'])
|
||
self.assertIn('DEBUG', joined)
|
||
self.assertIn('0000', joined)
|
||
|
||
def test_allowed_origins_include_the_deployed_frontend(self):
|
||
body = self.client.get('/api/health').json()
|
||
self.assertIn('https://math.tadrisino2.ir', body['allowedOrigins'])
|
||
|
||
|
||
class MigrationCheckTests(TestCase):
|
||
"""
|
||
این بررسی مهمترین چیزیست که /api/health میگوید، و تنها وقتی اهمیت دارد
|
||
که واقعاً مهاجرتِ اعمالنشده باشد — یعنی همان حالتی که تستش سخت است و
|
||
برای همین اولین بار غلط نوشته شد (Migration فیلدِ key ندارد).
|
||
"""
|
||
|
||
def test_pending_migrations_are_listed_by_name(self):
|
||
from unittest.mock import patch
|
||
|
||
from django.db.migrations import Migration
|
||
|
||
from config import health
|
||
|
||
fake = Migration('0009_something', 'content')
|
||
with patch.object(health.MigrationExecutor, 'migration_plan', return_value=[(fake, False)]):
|
||
result = health._check_migrations()
|
||
|
||
self.assertFalse(result['ok'])
|
||
self.assertEqual(result['pending'], ['content.0009_something'])
|
||
self.assertNotIn('error', result, 'نباید به استثنا بیفتد')
|
||
|
||
def test_a_clean_database_reports_nothing_pending(self):
|
||
from config import health
|
||
|
||
result = health._check_migrations()
|
||
self.assertTrue(result['ok'])
|
||
self.assertEqual(result['pending'], [])
|
||
|
||
|
||
class ReplaceCountTests(TestCase):
|
||
"""عددی که به کاربر گزارش میشود باید تعدادِ سوال باشد، نه مجموعِ ردیفهای آبشاری."""
|
||
|
||
def test_replace_reports_the_number_of_questions_not_of_rows(self):
|
||
grade = Grade.objects.create(title='پایه نهم', icon='', color='')
|
||
unit = Unit.objects.create(grade=grade, title='فصل ۱', order=1)
|
||
node = Node.objects.create(unit=unit, title='درس ۱', order=1)
|
||
lesson = Lesson.objects.create(node=node, title='نوع ۱')
|
||
|
||
payload = json.dumps([{
|
||
'lessonId': lesson.pk, 'prompt': 'سوال یک', 'type': 'multiple_choice',
|
||
'choices': ['۱', '۲', '۳', '۴'], 'correct': 0,
|
||
}], ensure_ascii=False).encode()
|
||
import_questions(parse(payload)) # ۱ سوال + ۴ گزینه = ۵ ردیف
|
||
|
||
result = import_questions(parse(payload), replace=True)
|
||
self.assertEqual(result['removed'], 1, 'باید ۱ باشد نه ۵')
|
||
|
||
|
||
class HealthTrafficTests(TestCase):
|
||
"""
|
||
سیگنالِ «کسی وصل شده یا نه» — تنها راهی که از بیرون میشود فهمید فرانت
|
||
واقعاً به سرور رسیده است.
|
||
"""
|
||
|
||
def test_reports_counts_and_last_activity(self):
|
||
from django.utils import timezone
|
||
|
||
from accounts.models import User
|
||
|
||
User.objects.create(phone='09120000009', last_activity_at=timezone.now())
|
||
traffic = self.client.get('/api/health').json()['traffic']
|
||
|
||
self.assertEqual(traffic['users'], 1)
|
||
self.assertEqual(traffic['activeLast24h'], 1)
|
||
self.assertIsNotNone(traffic['lastActivityAt'])
|
||
|
||
def test_an_idle_user_is_not_counted_as_active(self):
|
||
from datetime import timedelta
|
||
|
||
from django.utils import timezone
|
||
|
||
from accounts.models import User
|
||
|
||
User.objects.create(phone='09120000010',
|
||
last_activity_at=timezone.now() - timedelta(days=3))
|
||
traffic = self.client.get('/api/health').json()['traffic']
|
||
self.assertEqual(traffic['users'], 1)
|
||
self.assertEqual(traffic['activeLast24h'], 0)
|
||
|
||
def test_it_leaks_no_personal_data(self):
|
||
from accounts.models import User
|
||
|
||
User.objects.create(phone='09121234567', name='علی')
|
||
raw = self.client.get('/api/health').content.decode()
|
||
self.assertNotIn('09121234567', raw)
|
||
self.assertNotIn('علی', raw)
|
||
|
||
|
||
|
||
class QuestionGenerationPageBridgeTests(TestCase):
|
||
"""
|
||
پلِ بینِ chunkهای کتاب و درسهای مسیرِ بازی.
|
||
|
||
تطبیق از روی عنوان شکننده است: عنوانِ درس از یک استخراجِ قدیمی میآید و هر
|
||
بازخوانیِ کتاب عنوانِ تازه میسازد. بعد از بازخوانیِ ۱۴۰۵/۰۶/۲۹ فقط ۵ تا ۸
|
||
درصدِ عنوانها میخوردند و ۹۵٪ درسها **بیهیچ خطایی** بدونِ نامزد میماندند.
|
||
لنگرِ پایدار شمارهی صفحه است: هر Node یک بخشِ فهرستِ کتاب با بازهی مشخص.
|
||
"""
|
||
|
||
def setUp(self):
|
||
from content.models import Grade, Lesson, Node, Unit
|
||
from rag.models import LessonChunk
|
||
from rag.services.textbooks import TEXTBOOKS
|
||
|
||
self.book = TEXTBOOKS['ریاضی نهم']
|
||
chapter = self.book.chapters[0]
|
||
section = chapter.resolved_sections()[0]
|
||
self.section = section
|
||
|
||
grade = Grade.objects.create(title='پایه نهم')
|
||
unit = Unit.objects.create(grade=grade, title=chapter.title, order=chapter.number)
|
||
node = Node.objects.create(unit=unit, title=section.title, order=1)
|
||
# عنوانِ درس عمداً با هیچ chunkی یکی نیست — همان حالتی که بعد از
|
||
# بازخوانی پیش آمد.
|
||
self.lesson = Lesson.objects.create(
|
||
node=node, title='عنوانی که دیگر وجود ندارد', intro_text='x')
|
||
|
||
self.chunk = LessonChunk.objects.create(
|
||
collection='ریاضی نهم', subject='ریاضی', lesson_number=1,
|
||
chapter=chapter.number, chapter_title=chapter.title, order=1,
|
||
title=chapter.title, original_text='x', summary='x',
|
||
problem_type='نوعِ تازهی استخراجشده',
|
||
problem='۲ + ۲ چند است؟', solution='۴',
|
||
source_page=section.first_page + self.book.page_offset,
|
||
)
|
||
|
||
def _targets(self, **over):
|
||
from content.management.commands.generate_questions_from_chunks import Command
|
||
from content.models import Grade
|
||
|
||
opts = {'chapters': None, 'per_lesson': 1, 'limit': None,
|
||
'verified_only': False, 'by_page': True}
|
||
opts.update(over)
|
||
cmd = Command()
|
||
cmd._book = self.book
|
||
return cmd._targets(Grade.objects.get(title='پایه نهم'), 'ریاضی نهم', opts)
|
||
|
||
def test_a_chunk_in_the_nodes_page_range_is_used_even_if_the_title_differs(self):
|
||
targets = self._targets()
|
||
self.assertEqual([(l.pk, c.pk) for l, c in targets],
|
||
[(self.lesson.pk, self.chunk.pk)])
|
||
|
||
def test_strict_title_mode_finds_nothing(self):
|
||
# رفتارِ قدیم، نگه داشته شده تا بشود مقایسه کرد.
|
||
self.assertEqual(self._targets(by_page=False), [])
|
||
|
||
def test_a_chunk_outside_the_page_range_is_not_borrowed(self):
|
||
self.chunk.source_page = self.section.last_page + self.book.page_offset + 50
|
||
self.chunk.save(update_fields=['source_page'])
|
||
self.assertEqual(self._targets(), [])
|
||
|
||
def test_unverified_chunks_are_skipped_by_default(self):
|
||
# ساختنِ سوال از راهحلی که تأیید نشده، خطای کتاب را به دانشآموز میبرد.
|
||
self.assertEqual(self._targets(verified_only=True), [])
|
||
|
||
def test_one_chunk_is_never_given_to_two_lessons(self):
|
||
from content.models import Lesson
|
||
|
||
second = Lesson.objects.create(
|
||
node=self.lesson.node, title='درسِ دوم', intro_text='x')
|
||
targets = self._targets()
|
||
self.assertEqual(len(targets), 1)
|
||
self.assertNotIn(second.pk, [l.pk for l, _ in targets])
|