-
Notifications
You must be signed in to change notification settings - Fork 1.4k
/
db_url.py
78 lines (65 loc) · 3.01 KB
/
db_url.py
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
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
from peewee import *
from playhouse.db_url import connect
from playhouse.db_url import parse
from .base import BaseTestCase
class TestDBUrl(BaseTestCase):
def test_db_url_parse(self):
cfg = parse('mysql:https://usr:pwd@hst:123/db')
self.assertEqual(cfg['user'], 'usr')
self.assertEqual(cfg['passwd'], 'pwd')
self.assertEqual(cfg['host'], 'hst')
self.assertEqual(cfg['database'], 'db')
self.assertEqual(cfg['port'], 123)
cfg = parse('postgresql:https://usr:pwd@hst/db')
self.assertEqual(cfg['password'], 'pwd')
cfg = parse('mysql+pool:https://usr:pwd@hst:123/db'
'?max_connections=42&stale_timeout=8001.2&zai=&baz=3.4.5'
'&boolz=false')
self.assertEqual(cfg['user'], 'usr')
self.assertEqual(cfg['password'], 'pwd')
self.assertEqual(cfg['host'], 'hst')
self.assertEqual(cfg['database'], 'db')
self.assertEqual(cfg['port'], 123)
self.assertEqual(cfg['max_connections'], 42)
self.assertEqual(cfg['stale_timeout'], 8001.2)
self.assertEqual(cfg['zai'], '')
self.assertEqual(cfg['baz'], '3.4.5')
self.assertEqual(cfg['boolz'], False)
def test_db_url_quoted_password(self):
# By default, the password is not unescaped.
cfg = parse('mysql:https://usr:pwd%23%20@hst:123/db')
self.assertEqual(cfg['user'], 'usr')
self.assertEqual(cfg['passwd'], 'pwd%23%20')
self.assertEqual(cfg['host'], 'hst')
self.assertEqual(cfg['database'], 'db')
self.assertEqual(cfg['port'], 123)
cfg = parse('mysql:https://usr:pwd%23%20@hst:123/db', unquote_password=True)
self.assertEqual(cfg['user'], 'usr')
self.assertEqual(cfg['passwd'], 'pwd# ')
self.assertEqual(cfg['host'], 'hst')
self.assertEqual(cfg['database'], 'db')
self.assertEqual(cfg['port'], 123)
def test_db_url(self):
db = connect('sqlite:https:///:memory:')
self.assertTrue(isinstance(db, SqliteDatabase))
self.assertEqual(db.database, ':memory:')
db = connect('sqlite:https:///:memory:', pragmas=(
('journal_mode', 'MEMORY'),))
self.assertTrue(('journal_mode', 'MEMORY') in db._pragmas)
#db = connect('sqliteext:https:///foo/bar.db')
#self.assertTrue(isinstance(db, SqliteExtDatabase))
#self.assertEqual(db.database, 'foo/bar.db')
db = connect('sqlite:https:////this/is/absolute.path')
self.assertEqual(db.database, '/this/is/absolute.path')
db = connect('sqlite:https://')
self.assertTrue(isinstance(db, SqliteDatabase))
self.assertEqual(db.database, ':memory:')
db = connect('sqlite:https:///test.db?p1=1?a&p2=22&p3=xyz')
self.assertTrue(isinstance(db, SqliteDatabase))
self.assertEqual(db.database, 'test.db')
self.assertEqual(db.connect_params, {
'p1': '1?a', 'p2': 22, 'p3': 'xyz'})
def test_bad_scheme(self):
def _test_scheme():
connect('missing:https:///')
self.assertRaises(RuntimeError, _test_scheme)