Tests for db_url

Author: grigi

Diff for: tortoise/backends/base/db_url.py

@@ -3,6 +3,7 @@
 
 from tortoise.backends.asyncpg.client import AsyncpgDBClient
 from tortoise.backends.sqlite.client import SqliteClient
+from tortoise.exceptions import ConfigurationError
 
 urlparse.uses_netloc.append('postgres')
 urlparse.uses_netloc.append('sqlite')
@@ -16,9 +17,6 @@
             'username': 'user',
             'password': 'password',
         },
-        'test_defaults': {
-            'single_connection': True
-        },
     },
     'sqlite': {
         'client': SqliteClient,
@@ -32,19 +30,20 @@
 
 def expand_db_url(db_url: str, testing: bool = False) -> dict:
     url = urlparse.urlparse(db_url)
+    if url.scheme not in DB_LOOKUP:
+        raise ConfigurationError('Unknown DB scheme: {}'.format(url.scheme))
+
     db = DB_LOOKUP[url.scheme]
     if db.get('skip_first_char', True):
         path = url.path[1:]
     else:
         path = url.path
-    if '?' in path and not url.query:
-        path, query = path.split('?', 2)
-    else:
-        path, query = path, url.query
+
+    if not path:
+        raise ConfigurationError('No path specified for DB_URL')
 
     params = {}  # type: dict
-    params.update(db.get('test_defaults', {}))  # type: ignore
-    for key, val in urlparse.parse_qs(query).items():
+    for key, val in urlparse.parse_qs(url.query).items():
         params[key] = val[-1]
 
     if testing:
@@ -53,13 +52,15 @@ def expand_db_url(db_url: str, testing: bool = False) -> dict:
         path = path.format(uuid.uuid4().hex)
 
     vars = {}  # type: dict
-    vars.update(db.get('vars', {}))  # type: ignore
-    if vars.get('path'):
-        params[vars['path']] = path
+    vars.update(db['vars'])  # type: ignore
+    params[vars['path']] = path
     if vars.get('hostname'):
         params[vars['hostname']] = str(url.hostname or '')
-    if vars.get('port'):
-        params[vars['port']] = str(url.port or '')
+    try:
+        if vars.get('port'):
+            params[vars['port']] = str(url.port or '')
+    except ValueError:
+        raise ConfigurationError('Port is not an integer')
     if vars.get('username'):
         params[vars['username']] = str(url.username or '')
     if vars.get('password'):

Diff for: tortoise/tests/test_db_url.py

@@ -0,0 +1,97 @@
+from tortoise.backends.asyncpg.client import AsyncpgDBClient
+from tortoise.backends.base.db_url import expand_db_url
+from tortoise.backends.sqlite.client import SqliteClient
+from tortoise.contrib import test
+from tortoise.exceptions import ConfigurationError
+
+
+class TestDBUrl(test.SimpleTestCase):
+
+    def test_unknown_scheme(self):
+        with self.assertRaises(ConfigurationError):
+            expand_db_url('moo://baa')
+
+    def test_sqlite_basic(self):
+        res = expand_db_url('sqlite:///tmp/test.sqlite')
+        self.assertEqual(res, {
+            'client': SqliteClient,
+            'params': {
+                'filename': '/tmp/test.sqlite',  # nosec
+            }
+        })
+
+    def test_sqlite_testing(self):
+        res = expand_db_url('sqlite:///tmp/test-{}.sqlite', testing=True)
+        self.assertIn('/tmp/test-', res['params']['filename'])  # nosec
+        self.assertIn('.sqlite', res['params']['filename'])
+        self.assertNotEqual('sqlite:///tmp/test-{}.sqlite', res['params']['filename'])
+        self.assertEqual(res, {
+            'client': SqliteClient,
+            'params': {
+                'filename': res['params']['filename'],
+                'single_connection': True,
+            }
+        })
+
+    def test_sqlite_params(self):
+        res = expand_db_url('sqlite:///tmp/test.sqlite?AHA=5&moo=yes')
+        self.assertEqual(res, {
+            'client': SqliteClient,
+            'params': {
+                'filename': '/tmp/test.sqlite',  # nosec
+                'AHA': '5',
+                'moo': 'yes',
+            }
+        })
+
+    def test_sqlite_invalid(self):
+        with self.assertRaises(ConfigurationError):
+            expand_db_url('sqlite://')
+
+    def test_postgres_basic(self):
+        res = expand_db_url('postgres://postgres:@127.0.0.1:5432/test')
+        self.assertEqual(res, {
+            'client': AsyncpgDBClient,
+            'params': {
+                'database': 'test',
+                'host': '127.0.0.1',
+                'password': '',
+                'port': '5432',
+                'user': 'postgres',
+            }
+        })
+
+    def test_postgres_nonint_port(self):
+        with self.assertRaises(ConfigurationError):
+            expand_db_url('postgres://postgres:@127.0.0.1:moo/test')
+
+    def test_postgres_testing(self):
+        res = expand_db_url('postgres://postgres:@127.0.0.1:5432/test_\{\}', testing=True)
+        self.assertIn('test_', res['params']['database'])
+        self.assertNotEqual('test_{}', res['params']['database'])
+        self.assertEqual(res, {
+            'client': AsyncpgDBClient,
+            'params': {
+                'database': res['params']['database'],
+                'host': '127.0.0.1',
+                'password': '',
+                'port': '5432',
+                'user': 'postgres',
+                'single_connection': True,
+            }
+        })
+
+    def test_postgres_params(self):
+        res = expand_db_url('postgres://postgres:@127.0.0.1:5432/test?AHA=5&moo=yes')
+        self.assertEqual(res, {
+            'client': AsyncpgDBClient,
+            'params': {
+                'database': 'test',
+                'host': '127.0.0.1',
+                'password': '',
+                'port': '5432',
+                'user': 'postgres',
+                'AHA': '5',
+                'moo': 'yes',
+            }
+        })