import importlib import json import logging import os from unittest.mock import patch from django.core.exceptions import ImproperlyConfigured from django.test import RequestFactory, SimpleTestCase, TestCase from rest_framework.test import APIClient from accounts.authentication import issue_access_token from accounts.models import User from config.middleware import APIRequestLoggingMiddleware class SecretKeyFailClosedTests(SimpleTestCase): """ تست‌های متمرکز روی این فیکس: SECRET_KEY دیگه در پروداکشن (DEBUG=False) دیفالتِ هاردکد/قابل‌حدس‌زدن نداره. اگه DJANGO_SECRET_KEY ست نشده باشه، settings.py باید fail-closed بشه (خطای صریح موقعِ import)، نه اینکه با یک fallback شناخته‌شده بالا بیاد. برای DEBUG=True، fallbackِ محلیِ قبلی باید دست‌نخورده بمونه. مثل SecretDefaultsFailClosedTests در payment/tests.py، چون SECRET_KEY فقط یک‌بار موقعِ import محاسبه می‌شه (نه چیزی که override_settings بتونه toggle کنه)، ماژولِ config.settings رو با متغیرهای محیطیِ کنترل‌شده reload می‌کنیم. """ allow_database_queries = True def _reload_settings_with_env(self, env_overrides, env_removed=()): from config import settings as settings_module clean_env = {k: v for k, v in os.environ.items() if k not in env_removed} clean_env.update(env_overrides) with patch.dict(os.environ, clean_env, clear=True): importlib.reload(settings_module) return settings_module def tearDown(self): # همیشه ماژول رو به حالتِ واقعیِ محیطِ تست برمی‌گردونیم تا روی # بقیه‌ی تست‌ها اثر نذاره. from config import settings as settings_module importlib.reload(settings_module) def test_production_missing_secret_key_fails_closed(self): with self.assertRaises(ImproperlyConfigured): self._reload_settings_with_env( env_overrides={'DJANGO_DEBUG': ''}, env_removed={'DJANGO_SECRET_KEY'}, ) def test_production_configured_secret_key_used(self): # ⚠️ اصلاح (باگِ خودِ تست، نه اپلیکیشن): این تست فقط رفتارِ # SECRET_KEY رو می‌سنجه، اما چون CACHE_URL رو ست نمی‌کرد، وقتی # فیکسِ جداگانه‌ی CACHE_URL fail-closed (پایین‌ترِ همین فایل، نگاه # کنید به CacheUrlFailClosedTests) اضافه شد، همین import/reload با # DEBUG=False و بدونِ CACHE_URL به‌جای رسیدن به assertion، با # ImproperlyConfigured مربوط به CACHE_URL (نه SECRET_KEY) fail # می‌شد — یک false negative روی چیزی که این تست اصلاً قرار نیست # بسنجه. یک CACHE_URL معتبر هم ست می‌کنیم تا این تست فقط دقیقاً # همون چیزی که اسمش می‌گه (SECRET_KEY) رو بسنجه. settings_module = self._reload_settings_with_env( env_overrides={ 'DJANGO_DEBUG': '', 'DJANGO_SECRET_KEY': 'prod-secret-xyz', 'CACHE_URL': 'redis://localhost:6379/0', }, ) self.assertFalse(settings_module.DEBUG) self.assertEqual(settings_module.SECRET_KEY, 'prod-secret-xyz') def test_debug_mode_missing_secret_key_uses_local_fallback(self): settings_module = self._reload_settings_with_env( env_overrides={'DJANGO_DEBUG': '1'}, env_removed={'DJANGO_SECRET_KEY'}, ) self.assertTrue(settings_module.DEBUG) self.assertEqual( settings_module.SECRET_KEY, 'django-insecure-jn!s*25lz^+bjm101tzmlz8jlj70i0qka++*h-(!0s^^vk-0ss', ) class CacheUrlFailClosedTests(SimpleTestCase): """ 🔒 تستِ متمرکز روی فیکسِ CACHE_URL: در پروداکشن (DEBUG=False)، اگه CACHE_URL ست نشده باشه، settings.py باید fail-closed بشه (خطای صریح موقعِ import)، نه fallback خاموش به LocMemCache (که بینِ worker/instance ها مشترک نیست و throttle/rate-limit ها رو بی‌اثر می‌کنه). """ allow_database_queries = True def _reload_settings_with_env(self, env_overrides, env_removed=()): from config import settings as settings_module clean_env = {k: v for k, v in os.environ.items() if k not in env_removed} clean_env.update(env_overrides) with patch.dict(os.environ, clean_env, clear=True): importlib.reload(settings_module) return settings_module def tearDown(self): from config import settings as settings_module importlib.reload(settings_module) def test_production_missing_cache_url_fails_closed(self): with self.assertRaises(ImproperlyConfigured): self._reload_settings_with_env( env_overrides={'DJANGO_DEBUG': '', 'DJANGO_SECRET_KEY': 'prod-secret-xyz'}, env_removed={'CACHE_URL'}, ) def test_debug_mode_missing_cache_url_uses_locmem_fallback(self): settings_module = self._reload_settings_with_env( env_overrides={'DJANGO_DEBUG': '1'}, env_removed={'CACHE_URL'}, ) self.assertTrue(settings_module.DEBUG) self.assertEqual( settings_module.CACHES['default']['BACKEND'], 'django.core.cache.backends.locmem.LocMemCache', ) class APIRequestLoggingMiddlewareTests(TestCase): """ 🔒 اثبات میان‌افزارِ جدیدِ لاگِ درخواستِ API (config.middleware. APIRequestLoggingMiddleware): باید برای مسیرهای /api/ یک خطِ لاگِ خلاصه (متد/مسیر/وضعیت/مدت/شناسه‌ی کاربر) ثبت کنه، بدونِ اینکه هیچ‌وقت بدنه/ هدر/سکرت رو لاگ کنه؛ و برای مسیرهای غیرِ /api/ (مثلِ /admin/) اصلاً لاگ نکنه. """ def setUp(self): self.client = APIClient() def test_authenticated_api_request_is_logged_with_user_id_and_no_secrets(self): user = User.objects.create(device_id='device-mw-log-1', is_guest=True, name='') token = issue_access_token(user.id) with self.assertLogs('config.middleware', level='INFO') as captured: response = self.client.post( '/api/user/grade', {'gradeId': 10}, format='json', HTTP_AUTHORIZATION=f'Bearer {token}', ) self.assertEqual(response.status_code, 200) self.assertEqual(len(captured.records), 1) message = captured.records[0].getMessage() self.assertIn('method=POST', message) self.assertIn('path=/api/user/grade', message) self.assertIn('status=200', message) self.assertIn(f'user_id={user.id}', message) # هیچ‌وقت نباید هیچ سکرتی در پیامِ لاگ باشه. self.assertNotIn(token, message) self.assertNotIn('Bearer', message) self.assertNotIn('gradeId', message) def test_anonymous_api_request_is_logged_with_null_user_id(self): with self.assertLogs('config.middleware', level='INFO') as captured: response = self.client.post( '/api/auth/session', {'type': 'guest', 'deviceId': 'device-mw-log-anon-1'}, format='json', ) self.assertEqual(response.status_code, 200) self.assertEqual(len(captured.records), 1) message = captured.records[0].getMessage() self.assertIn('method=POST', message) self.assertIn('path=/api/auth/session', message) self.assertIn('status=200', message) self.assertIn('user_id=None', message) def test_non_api_request_is_not_logged(self): # /admin/ زیرِ /api/ نیست؛ میان‌افزار عمداً این‌جور مسیرها رو لاگ # نمی‌کنه تا لاگِ اضافه/بی‌فایده تولید نشه. logger = logging.getLogger('config.middleware') handler = logging.Handler() records = [] handler.emit = records.append logger.addHandler(handler) try: self.client.get('/admin/login/') finally: logger.removeHandler(handler) self.assertEqual(records, []) def test_unhandled_exception_from_view_is_logged_as_error_and_reraised(self): # 🔒 اثبات فیکسِ این دور: اگه self.get_response(request) خودش یک # exceptionِ دست‌نخورده raise کنه، میان‌افزار باید همچنان یک خطِ # لاگِ سطحِ ERROR (متد/مسیر/status=500/مدت/user_id، بدونِ جزئیاتِ # حساس) ثبت کنه و بعد دقیقاً همون exceptionِ اصلی رو دوباره raise # کنه — بدونِ اینکه هیچ رفتارِ Django/DRF یا exceptionِ دیگه‌ای رو # تغییر بده. مستقیماً خودِ میان‌افزار (نه کلِ چرخه‌ی # exception-handlingِ Django) تست می‌شه، چون همون لایه‌ای‌ست که این # فیکس روش اعمال شده. # # ⚠️ اصلاح (باگِ خودِ تست، نه اپلیکیشن): قبلاً این تست فقط # record.getMessage() رو چک می‌کرد. getMessage() خودش داخلی # msg %% args رو انجام می‌ده، پس اگه شمارِ آرگومان‌ها با # placeholderهای رشته جور نباشه، همون‌جا با یک exception (نه یک # پیامِ نادرست) fail می‌شد و اصلاً به assertion نمی‌رسید. برای اینکه # این تست واقعاً *صحتِ formatting* رو بسنجه (نه فقط اینکه یک‌جوری # fail نشه)، حالا با logging.Formatter().format(record) دقیقاً # همون مسیری که یک handlerِ واقعی (کنسول/فایل) طی می‌کنه رو صدا # می‌زنیم و روی متنِ نهاییِ فرمت‌شده assert می‌کنیم. request = RequestFactory().get('/api/some/failing/path') def failing_get_response(inner_request): raise ValueError('boom') middleware = APIRequestLoggingMiddleware(failing_get_response) with self.assertLogs('config.middleware', level='ERROR') as captured: with self.assertRaises(ValueError): middleware(request) self.assertEqual(len(captured.records), 1) record = captured.records[0] self.assertEqual(record.levelname, 'ERROR') formatted = logging.Formatter().format(record) self.assertIn('method=GET', formatted) self.assertIn('path=/api/some/failing/path', formatted) self.assertIn('status=500', formatted) self.assertIn('user_id=None', formatted) class JSONLogFormatterTests(SimpleTestCase): """ 🔒 اثبات فرمت‌کننده‌ی جدیدِ JSON (config.log_formatters.JSONLogFormatter) که پروژه رو برای جمع‌آوریِ متمرکزِ لاگ (Loki/ELK-style) آماده می‌کنه: خروجی باید یک JSON معتبر با فیلدهای timestamp/level/logger/message باشه، و در صورتِ وجودِ exception، traceback هم به‌عنوانِ فیلدِ "exception" اضافه بشه. """ def _make_record(self, level=logging.INFO, message='hello world', exc_info=None): return logging.LogRecord( name='some.logger', level=level, pathname=__file__, lineno=1, msg=message, args=(), exc_info=exc_info, ) def test_basic_record_is_formatted_as_valid_json_with_expected_fields(self): from config.log_formatters import JSONLogFormatter record = self._make_record(level=logging.INFO, message='api_request method=GET path=/api/x status=200') formatted = JSONLogFormatter().format(record) payload = json.loads(formatted) self.assertEqual(payload['level'], 'INFO') self.assertEqual(payload['logger'], 'some.logger') self.assertEqual(payload['message'], 'api_request method=GET path=/api/x status=200') self.assertIn('timestamp', payload) self.assertNotIn('exception', payload) def test_record_with_exception_includes_traceback_field(self): from config.log_formatters import JSONLogFormatter try: raise ValueError('boom') except ValueError: import sys exc_info = sys.exc_info() record = self._make_record(level=logging.ERROR, message='something failed', exc_info=exc_info) formatted = JSONLogFormatter().format(record) payload = json.loads(formatted) self.assertEqual(payload['level'], 'ERROR') self.assertEqual(payload['message'], 'something failed') self.assertIn('exception', payload) self.assertIn('ValueError: boom', payload['exception'])