Fix tests; pass url query args to create_engine. Fixes #40

Browse Source
This commit is contained in:
Friedrich Lindenberg 2013-11-15 23:23:03 +02:00
parent 8ec494c515
commit b7bbde45fa
2 changed files with 13 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
dataset/persistence/database.py
View File

@ -1,6 +1,7 @@
import logging import logging
import threading import threading
from urlparse import parse_qs from urlparse import parse_qs
from urllib import urlencode
from sqlalchemy import create_engine from sqlalchemy import create_engine
from migrate.versioning.util import construct_engine from migrate.versioning.util import construct_engine
@ -32,6 +33,8 @@ class Database(object):
schema_qs = query.pop('schema', query.pop('searchpath', [])) schema_qs = query.pop('schema', query.pop('searchpath', []))
if len(schema_qs): if len(schema_qs):
schema = schema_qs.pop() schema = schema_qs.pop()
if len(query):
url = url + '?' + urlencode(query, doseq=True)
self.schema = schema self.schema = schema
engine = create_engine(url, **kw) engine = create_engine(url, **kw)
self.url = url self.url = url

16
test/test_persistence.py
View File

@ -24,6 +24,10 @@ class DatabaseTestCase(unittest.TestCase):
def test_valid_database_url(self): def test_valid_database_url(self):
assert self.db.url, os.environ['DATABASE_URL'] assert self.db.url, os.environ['DATABASE_URL']
def test_database_url_query_string(self):
db = connect('sqlite:///:memory:/?cached_statements=1')
assert 'cached_statements' in db.url, db.url
def test_tables(self): def test_tables(self):
assert self.db.tables == ['weather'], self.db.tables assert self.db.tables == ['weather'], self.db.tables
@ -42,7 +46,7 @@ class DatabaseTestCase(unittest.TestCase):
table.insert({ table.insert({
'string_id': 'foobar'}) 'string_id': 'foobar'})
assert table.find_one(string_id = 'foobar')[0] == 'foobar' assert table.find_one(string_id = 'foobar')['string_id'] == 'foobar'
def test_create_table_custom_id2(self): def test_create_table_custom_id2(self):
pid = "int_id" pid = "int_id"
@ -53,8 +57,8 @@ class DatabaseTestCase(unittest.TestCase):
table.insert({'int_id': 123}) table.insert({'int_id': 123})
table.insert({'int_id': 124}) table.insert({'int_id': 124})
assert table.find_one(int_id = 123)[0] == 123 assert table.find_one(int_id = 123)['int_id'] == 123
assert table.find_one(int_id = 124)[0] == 124 assert table.find_one(int_id = 124)['int_id'] == 124
with self.assertRaises(IntegrityError): with self.assertRaises(IntegrityError):
table.insert({'int_id': 123}) table.insert({'int_id': 123})
@ -67,8 +71,8 @@ class DatabaseTestCase(unittest.TestCase):
table.insert({'int_id': 123}) table.insert({'int_id': 123})
table.insert({'int_id': 124}) table.insert({'int_id': 124})
assert table.find_one(int_id = 123)[0] == 123 assert table.find_one(int_id = 123)['int_id'] == 123
assert table.find_one(int_id = 124)[0] == 124 assert table.find_one(int_id = 124)['int_id'] == 124
with self.assertRaises(IntegrityError): with self.assertRaises(IntegrityError):
table.insert({'int_id': 123}) table.insert({'int_id': 123})
@ -81,7 +85,7 @@ class DatabaseTestCase(unittest.TestCase):
table.insert({ table.insert({
'string_id': 'foobar'}) 'string_id': 'foobar'})
assert table.find_one(string_id = 'foobar')[0] == 'foobar' assert table.find_one(string_id = 'foobar')['string_id'] == 'foobar'
def test_load_table(self): def test_load_table(self):
tbl = self.db.load_table('weather') tbl = self.db.load_table('weather')