Zareian
1c7c4b8eca
plain_persian() strips kasra/fatha/damma/shadda/sukun/hamza-above and the written «ه‌ی» ezafe, keeping tanwin. Applied at output, not in data: PlainPersianJSONRenderer runs one byte-level pass over every JSON response (~5 ms per MB locally), and to_plain() cleans tutor answers including the stream. Rewriting titles in the DB/seed would have broken question binding by title path and SeedRecord keys. Tests updated to expect the cleaned text; config/tests_plain_persian.py. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
93 lines
4.5 KiB
Python
93 lines
4.5 KiB
Python
"""
|
||
سقفِ جواب‌های هم‌زمانِ دستیار (ai_tutor/concurrency.py): رگبارِ سوال نباید همه‌ی نخ‌های worker را بگیرد.
|
||
هم‌زمانیِ واقعیِ چند نخ با مدلِ واقعی آزموده نشده؛ این‌جا جا را با گرفتنِ دستی پر می‌کنیم.
|
||
"""
|
||
from unittest import mock
|
||
|
||
from django.core.cache import cache
|
||
from django.core.signals import request_finished
|
||
from django.db import close_old_connections
|
||
from django.test import TestCase, override_settings
|
||
|
||
from accounts.authentication import issue_access_token
|
||
from accounts.models import User
|
||
from ai_tutor import concurrency
|
||
from ai_tutor.models import Message
|
||
from content.models import Grade
|
||
|
||
|
||
@override_settings(AI_MAX_CONCURRENT_PER_WORKER=1)
|
||
class AskConcurrencyCapTests(TestCase):
|
||
def setUp(self):
|
||
cache.clear()
|
||
grade = Grade.objects.create(title='پایه نهم', icon='', color='')
|
||
self.user = User.objects.create(phone='09120007801', grade_id=grade.id)
|
||
self.assertEqual(concurrency.in_flight(), 0)
|
||
|
||
def tearDown(self):
|
||
self.assertEqual(concurrency.in_flight(), 0, 'هر تست باید جا را پس بدهد')
|
||
|
||
def _ask(self, **extra):
|
||
return self.client.post(
|
||
'/api/ai/ask', {'userQuestion': 'توان چیست؟', **extra}, content_type='application/json',
|
||
HTTP_AUTHORIZATION=f'Bearer {issue_access_token(self.user.id)}',
|
||
)
|
||
|
||
def _rag(self):
|
||
return (
|
||
mock.patch('rag.services.config.rag_is_configured', return_value=True),
|
||
mock.patch('rag.services.retrieval.retrieve_context',
|
||
return_value=([{'title': 'فصل', 'summary': 'متن'}], {'gradeTitle': 'نهم'})),
|
||
mock.patch('rag.services.generation.generate_answer', return_value=('جواب', [], {})),
|
||
mock.patch('rag.services.generation.stream_answer', side_effect=lambda *a, **k: iter(['جو', 'اب'])),
|
||
)
|
||
|
||
def test_full_means_503_with_retry_after_and_no_quota_used(self):
|
||
self.assertTrue(concurrency.try_acquire()) # یک سوالِ دیگر در حالِ جواب است
|
||
try:
|
||
with self.assertLogs('ai_tutor.views', level='WARNING'):
|
||
response = self._ask()
|
||
finally:
|
||
concurrency.release()
|
||
self.assertEqual(response.status_code, 503)
|
||
self.assertEqual(response.json()['detail'], 'دستیار الان خیلی شلوغه؛ چند ثانیه دیگه دوباره بپرس.')
|
||
self.assertEqual(response['Retry-After'], '10')
|
||
self.assertFalse(Message.objects.exists()) # نه پیامی ساخته شد نه سهمیه‌ای رزرو شد
|
||
|
||
def test_json_answer_gives_the_slot_back(self):
|
||
a, b, c, d = self._rag()
|
||
with a, b, c, d:
|
||
self.assertEqual(self._ask().status_code, 200)
|
||
self.assertEqual(concurrency.in_flight(), 0)
|
||
self.assertEqual(self._ask().status_code, 200) # سقفِ ۱: دومی هم جا گرفت
|
||
|
||
def test_a_failed_answer_gives_the_slot_back(self):
|
||
a, b, _, d = self._rag()
|
||
failing = mock.patch('rag.services.generation.generate_answer', side_effect=RuntimeError('provider'))
|
||
with a, b, failing, d, self.assertLogs('ai_tutor.views', level='ERROR'):
|
||
self.assertEqual(self._ask().status_code, 503)
|
||
|
||
def test_stream_holds_the_slot_until_it_is_read(self):
|
||
a, b, c, d = self._rag()
|
||
with a, b, c, d:
|
||
response = self._ask(stream=True)
|
||
self.assertEqual(concurrency.in_flight(), 1) # هنوز خوانده نشده
|
||
body = b''.join(response.streaming_content).decode() # کلاینتِ تست پایانش close() می‌زند
|
||
self.assertIn('event: done', body)
|
||
|
||
def test_a_stream_that_is_never_read_still_gives_the_slot_back(self):
|
||
a, b, c, d = self._rag()
|
||
with a, b, c, d:
|
||
response = self._ask(stream=True)
|
||
# کاربر پیش از اولین تکه رفت. سیگنالِ request_finished در TestCase اتصالِ تستی را می‌بندد؛ کلاینتِ
|
||
# تست هم برای پاسخِ استریم همین‌طور جدایش می‌کند.
|
||
request_finished.disconnect(close_old_connections)
|
||
try:
|
||
response.close()
|
||
finally:
|
||
request_finished.connect(close_old_connections)
|
||
|
||
def test_release_without_acquire_is_a_loud_bug(self):
|
||
with self.assertRaises(RuntimeError):
|
||
concurrency.release()
|