36 lines
1.1 KiB
Python
36 lines
1.1 KiB
Python
import json
|
|
|
|
import pytest
|
|
from django.db import OperationalError
|
|
|
|
from config.api_errors import handle_database_operational_error
|
|
|
|
pytestmark = pytest.mark.unit
|
|
|
|
|
|
class PostgresError(Exception):
|
|
def __init__(self, sqlstate):
|
|
super().__init__(sqlstate)
|
|
self.sqlstate = sqlstate
|
|
|
|
|
|
def test_database_lock_timeout_returns_retryable_service_unavailable():
|
|
error = OperationalError("canceling statement due to lock timeout")
|
|
error.__cause__ = PostgresError("55P03")
|
|
|
|
response = handle_database_operational_error(None, error)
|
|
|
|
assert response.status_code == 503
|
|
assert response["Retry-After"] == "2"
|
|
assert json.loads(response.content) == {
|
|
"detail": "درخواست بهدلیل شلوغی موقت پایگاه داده انجام نشد؛ کمی بعد دوباره تلاش کنید."
|
|
}
|
|
|
|
|
|
def test_unrelated_database_operational_error_is_not_hidden():
|
|
error = OperationalError("database unavailable")
|
|
error.__cause__ = PostgresError("08006")
|
|
|
|
with pytest.raises(OperationalError, match="database unavailable"):
|
|
handle_database_operational_error(None, error)
|