mirror of
https://github.com/sqlalchemy/sqlalchemy.git
synced 2026-05-23 09:01:23 -04:00
Khairi Hafsham
772374735d
tested using pycodestyle version 2.2.0 Fixes: #3885 Change-Id: I5df43adc3aefe318f9eeab72a078247a548ec566 Pull-request: https://github.com/zzzeek/sqlalchemy/pull/343
240 lines
8.2 KiB
Python
240 lines
8.2 KiB
Python
"""Tests exceptions and DB-API exception wrapping."""
|
|
|
|
|
|
from sqlalchemy import exc as sa_exceptions
|
|
from sqlalchemy.testing import fixtures
|
|
from sqlalchemy.testing import eq_
|
|
from sqlalchemy.engine import default
|
|
|
|
|
|
class Error(Exception):
|
|
pass
|
|
|
|
|
|
class DatabaseError(Error):
|
|
pass
|
|
|
|
|
|
class OperationalError(DatabaseError):
|
|
pass
|
|
|
|
|
|
class ProgrammingError(DatabaseError):
|
|
|
|
def __str__(self):
|
|
return '<%s>' % self.bogus
|
|
|
|
|
|
class OutOfSpec(DatabaseError):
|
|
pass
|
|
|
|
|
|
# exception with a totally different name...
|
|
class WrongNameError(DatabaseError):
|
|
pass
|
|
|
|
|
|
# but they're going to call it their "IntegrityError"
|
|
IntegrityError = WrongNameError
|
|
|
|
|
|
# and they're going to subclass it!
|
|
class SpecificIntegrityError(WrongNameError):
|
|
pass
|
|
|
|
|
|
class WrapTest(fixtures.TestBase):
|
|
|
|
def _translating_dialect_fixture(self):
|
|
d = default.DefaultDialect()
|
|
d.dbapi_exception_translation_map = {
|
|
"WrongNameError": "IntegrityError"
|
|
}
|
|
return d
|
|
|
|
def test_db_error_normal(self):
|
|
try:
|
|
raise sa_exceptions.DBAPIError.instance(
|
|
'', [],
|
|
OperationalError(), DatabaseError)
|
|
except sa_exceptions.DBAPIError:
|
|
self.assert_(True)
|
|
|
|
def test_tostring(self):
|
|
try:
|
|
raise sa_exceptions.DBAPIError.instance(
|
|
'this is a message',
|
|
None, OperationalError(), DatabaseError)
|
|
except sa_exceptions.DBAPIError as exc:
|
|
eq_(
|
|
str(exc),
|
|
"(test.base.test_except.OperationalError) "
|
|
"[SQL: 'this is a message']")
|
|
|
|
def test_tostring_large_dict(self):
|
|
try:
|
|
raise sa_exceptions.DBAPIError.instance(
|
|
'this is a message',
|
|
{
|
|
'a': 1, 'b': 2, 'c': 3, 'd': 4, 'e': 5, 'f': 6, 'g': 7,
|
|
'h': 8, 'i': 9, 'j': 10, 'k': 11
|
|
},
|
|
OperationalError(), DatabaseError)
|
|
except sa_exceptions.DBAPIError as exc:
|
|
assert str(exc).startswith(
|
|
"(test.base.test_except.OperationalError) "
|
|
"[SQL: 'this is a message'] [parameters: {")
|
|
|
|
def test_tostring_large_list(self):
|
|
try:
|
|
raise sa_exceptions.DBAPIError.instance(
|
|
'this is a message',
|
|
[1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11],
|
|
OperationalError(), DatabaseError)
|
|
except sa_exceptions.DBAPIError as exc:
|
|
assert str(exc).startswith(
|
|
"(test.base.test_except.OperationalError) "
|
|
"[SQL: 'this is a message'] [parameters: "
|
|
"[1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11]]")
|
|
|
|
def test_tostring_large_executemany(self):
|
|
try:
|
|
raise sa_exceptions.DBAPIError.instance(
|
|
'this is a message',
|
|
[{1: 1}, {1: 1}, {1: 1}, {1: 1}, {1: 1}, {1: 1},
|
|
{1: 1}, {1: 1}, {1: 1}, {1: 1}, ],
|
|
OperationalError(), DatabaseError)
|
|
except sa_exceptions.DBAPIError as exc:
|
|
eq_(
|
|
str(exc),
|
|
"(test.base.test_except.OperationalError) "
|
|
"[SQL: 'this is a message'] [parameters: [{1: 1}, "
|
|
"{1: 1}, {1: 1}, {1: 1}, {1: 1}, {1: 1}, {1: 1}, {1: "
|
|
"1}, {1: 1}, {1: 1}]]"
|
|
)
|
|
try:
|
|
raise sa_exceptions.DBAPIError.instance('this is a message', [
|
|
{1: 1}, {1: 1}, {1: 1}, {1: 1}, {1: 1}, {1: 1}, {1: 1},
|
|
{1: 1}, {1: 1}, {1: 1}, {1: 1},
|
|
], OperationalError(), DatabaseError)
|
|
except sa_exceptions.DBAPIError as exc:
|
|
eq_(str(exc),
|
|
"(test.base.test_except.OperationalError) "
|
|
"[SQL: 'this is a message'] [parameters: [{1: 1}, "
|
|
"{1: 1}, {1: 1}, {1: 1}, {1: 1}, {1: 1}, "
|
|
"{1: 1}, {1: 1} ... displaying 10 of 11 total "
|
|
"bound parameter sets ... {1: 1}, {1: 1}]]"
|
|
)
|
|
try:
|
|
raise sa_exceptions.DBAPIError.instance(
|
|
'this is a message',
|
|
[
|
|
(1, ), (1, ), (1, ), (1, ), (1, ), (1, ),
|
|
(1, ), (1, ), (1, ), (1, ),
|
|
], OperationalError(), DatabaseError)
|
|
|
|
except sa_exceptions.DBAPIError as exc:
|
|
eq_(
|
|
str(exc),
|
|
"(test.base.test_except.OperationalError) "
|
|
"[SQL: 'this is a message'] [parameters: [(1,), "
|
|
"(1,), (1,), (1,), (1,), (1,), (1,), (1,), (1,), (1,)]]")
|
|
try:
|
|
raise sa_exceptions.DBAPIError.instance('this is a message', [
|
|
(1, ), (1, ), (1, ), (1, ), (1, ), (1, ), (1, ), (1, ), (1, ),
|
|
(1, ), (1, ),
|
|
], OperationalError(), DatabaseError)
|
|
except sa_exceptions.DBAPIError as exc:
|
|
eq_(str(exc),
|
|
"(test.base.test_except.OperationalError) "
|
|
"[SQL: 'this is a message'] [parameters: [(1,), "
|
|
"(1,), (1,), (1,), (1,), (1,), (1,), (1,) "
|
|
"... displaying 10 of 11 total bound "
|
|
"parameter sets ... (1,), (1,)]]"
|
|
)
|
|
|
|
def test_db_error_busted_dbapi(self):
|
|
try:
|
|
raise sa_exceptions.DBAPIError.instance(
|
|
'', [],
|
|
ProgrammingError(), DatabaseError)
|
|
except sa_exceptions.DBAPIError as e:
|
|
self.assert_(True)
|
|
self.assert_('Error in str() of DB-API' in e.args[0])
|
|
|
|
def test_db_error_noncompliant_dbapi(self):
|
|
try:
|
|
raise sa_exceptions.DBAPIError.instance(
|
|
'', [], OutOfSpec(),
|
|
DatabaseError)
|
|
except sa_exceptions.DBAPIError as e:
|
|
# OutOfSpec subclasses DatabaseError
|
|
self.assert_(e.__class__ is sa_exceptions.DatabaseError)
|
|
except OutOfSpec:
|
|
self.assert_(False)
|
|
|
|
try:
|
|
raise sa_exceptions.DBAPIError.instance(
|
|
'', [],
|
|
sa_exceptions.ArgumentError(), DatabaseError)
|
|
except sa_exceptions.DBAPIError as e:
|
|
self.assert_(e.__class__ is sa_exceptions.DBAPIError)
|
|
except sa_exceptions.ArgumentError:
|
|
self.assert_(False)
|
|
|
|
dialect = self._translating_dialect_fixture()
|
|
try:
|
|
raise sa_exceptions.DBAPIError.instance(
|
|
'', [],
|
|
sa_exceptions.ArgumentError(), DatabaseError,
|
|
dialect=dialect)
|
|
except sa_exceptions.DBAPIError as e:
|
|
self.assert_(e.__class__ is sa_exceptions.DBAPIError)
|
|
except sa_exceptions.ArgumentError:
|
|
self.assert_(False)
|
|
|
|
def test_db_error_dbapi_uses_wrong_names(self):
|
|
dialect = self._translating_dialect_fixture()
|
|
|
|
try:
|
|
raise sa_exceptions.DBAPIError.instance(
|
|
'', [], IntegrityError(),
|
|
DatabaseError, dialect=dialect)
|
|
except sa_exceptions.DBAPIError as e:
|
|
self.assert_(e.__class__ is sa_exceptions.IntegrityError)
|
|
|
|
try:
|
|
raise sa_exceptions.DBAPIError.instance(
|
|
'', [], SpecificIntegrityError(),
|
|
DatabaseError, dialect=dialect)
|
|
except sa_exceptions.DBAPIError as e:
|
|
self.assert_(e.__class__ is sa_exceptions.IntegrityError)
|
|
|
|
try:
|
|
raise sa_exceptions.DBAPIError.instance(
|
|
'', [], SpecificIntegrityError(),
|
|
DatabaseError)
|
|
except sa_exceptions.DBAPIError as e:
|
|
# doesn't work without a dialect
|
|
self.assert_(e.__class__ is not sa_exceptions.IntegrityError)
|
|
|
|
def test_db_error_keyboard_interrupt(self):
|
|
try:
|
|
raise sa_exceptions.DBAPIError.instance(
|
|
'', [],
|
|
KeyboardInterrupt(), DatabaseError)
|
|
except sa_exceptions.DBAPIError:
|
|
self.assert_(False)
|
|
except KeyboardInterrupt:
|
|
self.assert_(True)
|
|
|
|
def test_db_error_system_exit(self):
|
|
try:
|
|
raise sa_exceptions.DBAPIError.instance(
|
|
'', [],
|
|
SystemExit(), DatabaseError)
|
|
except sa_exceptions.DBAPIError:
|
|
self.assert_(False)
|
|
except SystemExit:
|
|
self.assert_(True)
|