File: test_db_heartbeat.py

package info (click to toggle)
python-django-health-check 3.20.0-1
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid
  • size: 428 kB
  • sloc: python: 1,886; makefile: 6
file content (27 lines) | stat: -rw-r--r-- 1,103 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
import unittest
from unittest.mock import MagicMock, patch

from health_check.contrib.db_heartbeat.backends import DatabaseHeartBeatCheck
from health_check.exceptions import ServiceUnavailable


class TestDatabaseHeartBeatCheck(unittest.TestCase):
    @patch("health_check.contrib.db_heartbeat.backends.connection")
    def test_check_status_success(self, mock_connection):
        mock_cursor = MagicMock()
        mock_cursor.fetchone.return_value = (1,)
        mock_connection.cursor.return_value.__enter__.return_value = mock_cursor

        health_check = DatabaseHeartBeatCheck()
        try:
            health_check.check_status()
        except Exception as e:
            self.fail(f"check_status() raised an exception unexpectedly: {e}")

    @patch("health_check.contrib.db_heartbeat.backends.connection")
    def test_check_status_service_unavailable(self, mock_connection):
        mock_connection.cursor.side_effect = Exception("Database error")

        health_check = DatabaseHeartBeatCheck()
        with self.assertRaises(ServiceUnavailable):
            health_check.check_status()