如何在django中测试postgres statement_timeout
如何在django中测试postgres statement_timeout
提问于 2019-08-19 10:30:16
我在django服务中为postgres数据库添加了连接和语句超时。因此，相关的django设置如下；
_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),
},
}
}然后，我写这样的测试；
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")然而，这个测试不起作用。
- 断言
self.assertEqual不传递，这意味着设置覆盖不起作用。 - 我希望
Book.objects.create语句在超时的情况下失败，但它没有。
所以问题；
- 如何测试postgres statement_timeout？(也可以测试连接超时)
- 如何捕获代码中的语句超时错误？
回答 2
Stack Overflow用户
回答已采纳
发布于 2020-05-27 11:41:34
查看django项目的测试用例
他们将超时设置为1ms和assertRaises(OperationalError)
完整代码：
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()Stack Overflow用户
发布于 2021-01-22 20:41:38
这就是我最后所做的；
在您的settings.py文件中：
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),
},
}
}在您的test_database.py文件中：
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)
)页面原文内容由Stack Overflow提供。腾讯云小微IT领域专用引擎提供翻译支持
原文链接：
https://stackoverflow.com/questions/57554951
复制相关文章
相似问题
腾讯云开发者
