1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
|
# Copyright (C) 2016-present the asyncpg authors and contributors
# <see AUTHORS file>
#
# This module is part of asyncpg and is released under
# the Apache 2.0 License: http://www.apache.org/licenses/LICENSE-2.0
import asyncpg
from asyncpg import _testbase as tb
class TestExceptions(tb.ConnectedTestCase):
def test_exceptions_exported(self):
for err in ('PostgresError', 'SubstringError', 'InterfaceError'):
self.assertTrue(hasattr(asyncpg, err))
self.assertIn(err, asyncpg.__all__)
for err in ('PostgresMessage',):
self.assertFalse(hasattr(asyncpg, err))
self.assertNotIn(err, asyncpg.__all__)
self.assertIsNone(asyncpg.PostgresError.schema_name)
async def test_exceptions_unpacking(self):
try:
await self.con.execute('SELECT * FROM _nonexistent_')
except asyncpg.UndefinedTableError as e:
self.assertEqual(e.sqlstate, '42P01')
self.assertEqual(e.position, '15')
self.assertEqual(e.query, 'SELECT * FROM _nonexistent_')
self.assertIsNotNone(e.severity)
else:
self.fail('UndefinedTableError not raised')
async def test_exceptions_str(self):
try:
await self.con.execute('''
CREATE FUNCTION foo() RETURNS bool AS $$ $$ LANGUAGE SQL;
''')
except asyncpg.InvalidFunctionDefinitionError as e:
if self.server_version < (17, 0):
detail = (
"Function's final statement must be SELECT or "
"INSERT/UPDATE/DELETE RETURNING."
)
else:
detail = (
"Function's final statement must be SELECT or "
"INSERT/UPDATE/DELETE/MERGE RETURNING."
)
self.assertEqual(e.detail, detail)
self.assertIn('DETAIL: Function', str(e))
else:
self.fail('InvalidFunctionDefinitionError not raised')
|
