1

I have added both connect and statement timeouts for a postgres database in my django service. So the relevant django setting looks like;

_CONN_TIMEOUT = 5
_STATEMENT_TIMEOUT = 3000 # millisecond
DATABASES = {
    "default": {
        "ENGINE": "django.db.backends.postgresql",
        "NAME": "some_database",
        # ...,
        "OPTIONS": {
            "connect_timeout": _CONN_TIMEOUT,
            "options": "-c statement_timeout={0}ms".format(_STATEMENT_TIMEOUT),
        },
    }
}

And then, I'm writing tests like this;

class DbTimeoutTest(TestCase):
    def test_db_statement_timeout(self):
        """
        test carrying out an SQL query that takes longer than the configured
        postgres `statement_timeout` value
        """
        # set statement_timeout to 1 millisecond
        mock_s_timeout = 1
        with self.settings(_CONN_TIMEOUT=5, _STATEMENT_TIMEOUT=mock_s_timeout):
            self.assertEqual(
                settings.DATABASES["default"]["OPTIONS"]["options"],
                "-c statement_timeout={0}ms".format(mock_s_timeout),
            )
            Book.objects.create(name="Hello")

However, that test is not working.

  • The assert self.assertEqual does not pass, meaning that the setting override did not work.
  • I would expect the Book.objects.create statement to fail with timeout but it does not.

So questions;

  1. How do I test for postgres statement_timeout?(been able to also test for connect timeout would also be a plus)
  2. How do I catch, the statement timeout error in my code?
1
  • try running the django server with this command python manage.py runserver --http_timeout=1800 Commented Aug 19, 2019 at 13:50

2 Answers 2

3

Have a look the at django project's test case

They set the timeout to 1ms and assertRaises(OperationalError)

Full code:

def test_statement_timeout(self):
    databases = copy.deepcopy(settings.DATABASES)
    # Set timeout to 1ms and execute a 1s query.
    databases[DEFAULT_DB_ALIAS]['STATEMENT_TIMEOUT'] = 1
    new_connections = ConnectionHandler(databases)
    new_connection = new_connections[DEFAULT_DB_ALIAS]
    try:
        with new_connection.cursor() as cursor:
            with self.assertRaises(OperationalError):
                cursor.execute("SELECT pg_sleep(1)")
    finally:
        new_connection.close()
Sign up to request clarification or add additional context in comments.

Comments

2

This is what I ended up doing;

In your settings.py file:

CONNECTION_TIMEOUT = 5 # seconds
STATEMENT_TIMEOUT = 3000 # milliseconds
DATABASES = {
    "default": {
        "ENGINE": "django.db.backends.postgresql",
        "NAME": "some_database",
        # ...,
        "OPTIONS": {
            "connect_timeout": CONNECTION_TIMEOUT,
            "options": "-c statement_timeout={0}ms".format(STATEMENT_TIMEOUT),
        },
    }
}

And in your test_database.py file:

from django.conf import settings
from django.test import TestCase
from django.db import connection
from django.db.utils import OperationalError


class DatabaseTest(TestCase):
    def test_timeout(self):
        statement_timeout_seconds = settings.STATEMENT_TIMEOUT / 1000
        cursor = connection.cursor()
        with self.assertRaises(OperationalError) as raised_exception:
            cursor.execute(
                # sql query that takes 3 times longer than the configured statement_timeout
                "SELECT pg_sleep({0})".format((statement_timeout_seconds) * 3)
            )
            cursor.fetchall()
        cursor.close()

        self.assertIn(
            "canceling statement due to statement timeout", str(raised_exception.exception)
        )

Comments

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Start asking to get answers

Find the answer to your question by asking.

Ask question

Explore related questions

See similar questions with these tags.