skip to Main Content

I am taking a online course about Django. In this courseI am connecting my project with a Postgresql database. For a case that my Django app start before the Postgresql database start, we write a command that delays the app start until Postgresql starts. Here is the command code:

from psycopg2 import OperationalError as Psycopg2OpError

from django.db.utils import OperationalError
from django.core.management.base import BaseCommand

class Command(BaseCommand):

    def handle(self, *args, **options):
        """ Entrypoint for command """
        self.stdout.write('Waiting for database...')
        db_up = False
        while db_up is False:
            try:
                self.check(databases=['default'])
                db_up = True
            except(Psycopg2OpError, OperationalError):
                self.stdout.write('Database unavailable, waiting for 1 second...')
                time.sleep(1)

        self.stdout.write(self.style.SUCCESS('Database available!'))

But I didn’t understand the test codes that testing if this commands work. Can you explain these codes logic?

from unittest.mock import patch

from psycopg2 import OperationalError as Psycopg2Error

from django.core.management import call_command
from django.db.utils import OperationalError
from django.test import SimpleTestCase

@patch('core.management.commands.wait_for_db.Command.check')
class CommandTests(SimpleTestCase):
    """ Test commands """

    def test_wait_for_db_ready(self, patched_check):
        """ Test waiting for database if database is ready"""
        patched_check.return_value = True

        call_command('wait_for_db')

        patched_check.assert_called_once_with(databases=['default'])


    @patch('time.sleep')
    def test_wait_for_db_delay(self, patched_sleep, patched_check):
        """ Test waiting for database when getting OperationalError"""

        patched_check.side_effect = [Psycopg2Error] * 2 + 
            [OperationalError] * 3 + [True]

        call_command('wait_for_db')

        self.assertEqual(patched_check.call_count, 6)
        patched_check.assert_called_with(databases=['default'])

2

Answers


  1. Background for test utils

    So you are creating a custom Command that you want to execute. Inside your command, you call the function check (see docs).

    Regarding your test case. You are performing 2 test scenarios.
    But before you run your tests you provide a patch. (See docs)

    The patch will – if used as decorator – be injected either at constructor or function parameter. For your case patched_check. Your patched_check now acts as a mock of your check function.

    What are the test cases doing?

    test_wait_for_db_ready(self, patched_check) validates that self.check is called with the param databases=['default'] exactly once.

    test_wait_for_db_delay validates that if the self.check throws an exception then time.sleep(1) is called 6 times.

    Login or Signup to reply.
  2. I was facing the same issue, please checkout your database environment variables in docker-compose.yml file. You have misconfigured any of database environment variable.

    Login or Signup to reply.
Please signup or login to give your own answer.
Back To Top
Search