Aleksandr Meshchriakov
25176f31b4
Some checks failed
CI/CD Pipeline / Telegram Notify Success (push) Has been cancelled
CI/CD Pipeline / Run Tests (push) Has been cancelled
CI/CD Pipeline / Code Quality Checks (push) Has been cancelled
CI/CD Pipeline / Code Quality Checks (pull_request) Successful in 1m42s
CI/CD Pipeline / Run Tests (pull_request) Successful in 2m25s
CI/CD Pipeline / Telegram Notify Success (pull_request) Successful in 1m34s
213 lines
7.5 KiB
Python
213 lines
7.5 KiB
Python
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)
|