|
import unittest
|
|
from datetime import datetime
|
|
|
|
from sqlaload import *
|
|
|
|
TEST_DATA = [
|
|
{
|
|
'date': datetime(2011, 01, 01),
|
|
'temperature': 1,
|
|
'place': 'Galway'
|
|
},
|
|
{
|
|
'date': datetime(2011, 01, 02),
|
|
'temperature': -1,
|
|
'place': 'Galway'
|
|
},
|
|
{
|
|
'date': datetime(2011, 01, 03),
|
|
'temperature': 0,
|
|
'place': 'Galway'
|
|
},
|
|
{
|
|
'date': datetime(2011, 01, 01),
|
|
'temperature': 6,
|
|
'place': 'Berkeley'
|
|
},
|
|
{
|
|
'date': datetime(2011, 01, 02),
|
|
'temperature': 8,
|
|
'place': 'Berkeley'
|
|
},
|
|
{
|
|
'date': datetime(2011, 01, 03),
|
|
'temperature': 5,
|
|
'place': 'Berkeley'
|
|
}
|
|
]
|
|
|
|
class TableTestCase(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
self.engine = connect('sqlite:///:memory:')
|
|
|
|
def test_create_table(self):
|
|
table = create_table(self.engine, 'foo')
|
|
assert table.exists()
|
|
assert len(table.columns)==1, table.columns
|
|
|
|
|
|
class QueryTestCase(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
self.engine = connect('sqlite:///:memory:')
|
|
self.table = create_table(self.engine, 'weather')
|
|
for entry in TEST_DATA:
|
|
upsert(self.engine, self.table, entry, ['place', 'date'],
|
|
ensure=True)
|
|
|
|
def test_all(self):
|
|
x = list(all(self.engine, self.table))
|
|
assert len(x)==len(TEST_DATA), len(x)
|
|
|
|
def test_distinct(self):
|
|
x = list(distinct(self.engine, self.table, 'place'))
|
|
assert len(x)==2, x
|
|
p = [i['place'] for i in x]
|
|
assert 'Berkeley' in p, p
|
|
assert 'Galway' in p, p
|
|
|
|
def test_all(self):
|
|
x = list(all(self.engine, self.table))
|
|
assert len(x)==len(TEST_DATA), len(x)
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|