from __future__ import annotations import importlib import os import sys from types import SimpleNamespace from unittest.mock import MagicMock, patch from apps.core import startup_checks from django.test import SimpleTestCase, override_settings def _db_secret() -> str: return "secret" TEST_DATABASES = { "default": { "NAME": "mostovik", "USER": "postgres", "PASSWORD": _db_secret(), "HOST": "db.example.test", "PORT": 5432, "OPTIONS": {"sslmode": "require"}, } } TEST_CACHES = { "default": { "BACKEND": "django.core.cache.backends.redis.RedisCache", "LOCATION": "redis://redis.example.test:6379/1", } } class StartupChecksTest(SimpleTestCase): @override_settings(DATABASES=TEST_DATABASES) @patch("apps.core.startup_checks.psycopg2.connect") def test_check_db_success(self, connect_mock): cursor = MagicMock() connection = MagicMock() connection.cursor.return_value.__enter__.return_value = cursor connect_mock.return_value = connection success, message = startup_checks._check_db(7) self.assertTrue(success) self.assertEqual(message, "OK") connect_mock.assert_called_once_with( dbname="mostovik", user="postgres", password=_db_secret(), host="db.example.test", port=5432, connect_timeout=7, sslmode="require", ) cursor.execute.assert_called_once_with("SELECT 1") cursor.fetchone.assert_called_once_with() connection.close.assert_called_once_with() @override_settings(DATABASES=TEST_DATABASES) @patch( "apps.core.startup_checks.psycopg2.connect", side_effect=RuntimeError("database down"), ) def test_check_db_failure(self, connect_mock): success, message = startup_checks._check_db(5) self.assertFalse(success) self.assertIn("db.example.test:5432/mostovik", message) self.assertIn("database down", message) connect_mock.assert_called_once() @override_settings(CACHES=TEST_CACHES) @patch("apps.core.startup_checks.redis.Redis.from_url") def test_check_redis_success(self, from_url_mock): client = MagicMock() from_url_mock.return_value = client success, message = startup_checks._check_redis(4) self.assertTrue(success) self.assertEqual(message, "OK") from_url_mock.assert_called_once_with( "redis://redis.example.test:6379/1", socket_connect_timeout=4, socket_timeout=4, ) client.ping.assert_called_once_with() @override_settings(CACHES=TEST_CACHES) @patch( "apps.core.startup_checks.redis.Redis.from_url", side_effect=RuntimeError("redis down"), ) def test_check_redis_failure(self, from_url_mock): success, message = startup_checks._check_redis(6) self.assertFalse(success) self.assertIn("redis.example.test:6379/1", message) self.assertIn("redis down", message) from_url_mock.assert_called_once() @override_settings(STARTUP_CHECKS_ENABLED=False) @patch("apps.core.startup_checks._check_db") @patch("apps.core.startup_checks._check_redis") def test_run_startup_checks_skips_when_disabled( self, redis_mock, db_mock, ): startup_checks.run_startup_checks(component="web") db_mock.assert_not_called() redis_mock.assert_not_called() @override_settings( STARTUP_CHECKS_ENABLED=True, STARTUP_DB_TIMEOUT_SECONDS=9, STARTUP_REDIS_TIMEOUT_SECONDS=11, ) @patch("apps.core.startup_checks._check_db", return_value=(True, "OK")) @patch("apps.core.startup_checks._check_redis", return_value=(True, "OK")) def test_run_startup_checks_success(self, redis_mock, db_mock): startup_checks.run_startup_checks(component="worker") db_mock.assert_called_once_with(9) redis_mock.assert_called_once_with(11) @override_settings(STARTUP_CHECKS_ENABLED=True, STARTUP_DB_TIMEOUT_SECONDS=8) @patch("apps.core.startup_checks._check_db", return_value=(False, "db failed")) @patch("apps.core.startup_checks._log") def test_run_startup_checks_exits_on_db_failure(self, log_mock, db_mock): with self.assertRaises(SystemExit) as error: startup_checks.run_startup_checks(component="wsgi") self.assertEqual(error.exception.code, 1) db_mock.assert_called_once_with(8) log_mock.assert_called_once() self.assertIn("[startup:wsgi] DB check failed", log_mock.call_args.args[0]) self.assertIn("db failed", log_mock.call_args.args[0]) @override_settings( STARTUP_CHECKS_ENABLED=True, STARTUP_DB_TIMEOUT_SECONDS=3, STARTUP_REDIS_TIMEOUT_SECONDS=12, ) @patch("apps.core.startup_checks._check_db", return_value=(True, "OK")) @patch( "apps.core.startup_checks._check_redis", return_value=(False, "redis failed") ) @patch("apps.core.startup_checks._log") def test_run_startup_checks_exits_on_redis_failure( self, log_mock, redis_mock, db_mock, ): with self.assertRaises(SystemExit) as error: startup_checks.run_startup_checks(component="asgi") self.assertEqual(error.exception.code, 1) db_mock.assert_called_once_with(3) redis_mock.assert_called_once_with(12) self.assertIn("[startup:asgi] Redis check failed", log_mock.call_args.args[0]) self.assertIn("redis failed", log_mock.call_args.args[0]) class EntryPointImportTest(SimpleTestCase): def _import_fresh(self, module_name: str): sys.modules.pop("core", None) sys.modules.pop(module_name, None) return importlib.import_module(module_name) def test_import_core_asgi_runs_startup_checks_and_sets_default_settings(self): sentinel_application = object() celery_stub = SimpleNamespace(app=object()) with patch.dict(os.environ, {}, clear=False): os.environ.pop("DJANGO_SETTINGS_MODULE", None) with patch.dict(sys.modules, {"core.celery": celery_stub}), patch( "apps.core.startup_checks.run_startup_checks" ) as checks_mock, patch( "django.core.asgi.get_asgi_application", return_value=sentinel_application, ): module = self._import_fresh("core.asgi") self.assertEqual(os.environ["DJANGO_SETTINGS_MODULE"], "settings.test") checks_mock.assert_called_once_with(component="asgi") self.assertIs(module.application, sentinel_application) sys.modules.pop("core.asgi", None) def test_import_core_wsgi_runs_startup_checks_and_sets_default_settings(self): sentinel_application = object() celery_stub = SimpleNamespace(app=object()) with patch.dict(os.environ, {}, clear=False): os.environ.pop("DJANGO_SETTINGS_MODULE", None) with patch.dict(sys.modules, {"core.celery": celery_stub}), patch( "apps.core.startup_checks.run_startup_checks" ) as checks_mock, patch( "django.core.wsgi.get_wsgi_application", return_value=sentinel_application, ): module = self._import_fresh("core.wsgi") self.assertEqual(os.environ["DJANGO_SETTINGS_MODULE"], "settings.test") checks_mock.assert_called_once_with(component="wsgi") self.assertIs(module.application, sentinel_application) sys.modules.pop("core.wsgi", None)