Apply black
This commit is contained in:
Friedrich Lindenberg
+8
-32
@@ -4,38 +4,14 @@ from __future__ import unicode_literals
|
||||
from datetime import datetime
|
||||
|
||||
|
||||
TEST_CITY_1 = 'B€rkeley'
|
||||
TEST_CITY_2 = 'G€lway'
|
||||
TEST_CITY_1 = "B€rkeley"
|
||||
TEST_CITY_2 = "G€lway"
|
||||
|
||||
TEST_DATA = [
|
||||
{
|
||||
'date': datetime(2011, 1, 1),
|
||||
'temperature': 1,
|
||||
'place': TEST_CITY_2
|
||||
},
|
||||
{
|
||||
'date': datetime(2011, 1, 2),
|
||||
'temperature': -1,
|
||||
'place': TEST_CITY_2
|
||||
},
|
||||
{
|
||||
'date': datetime(2011, 1, 3),
|
||||
'temperature': 0,
|
||||
'place': TEST_CITY_2
|
||||
},
|
||||
{
|
||||
'date': datetime(2011, 1, 1),
|
||||
'temperature': 6,
|
||||
'place': TEST_CITY_1
|
||||
},
|
||||
{
|
||||
'date': datetime(2011, 1, 2),
|
||||
'temperature': 8,
|
||||
'place': TEST_CITY_1
|
||||
},
|
||||
{
|
||||
'date': datetime(2011, 1, 3),
|
||||
'temperature': 5,
|
||||
'place': TEST_CITY_1
|
||||
}
|
||||
{"date": datetime(2011, 1, 1), "temperature": 1, "place": TEST_CITY_2},
|
||||
{"date": datetime(2011, 1, 2), "temperature": -1, "place": TEST_CITY_2},
|
||||
{"date": datetime(2011, 1, 3), "temperature": 0, "place": TEST_CITY_2},
|
||||
{"date": datetime(2011, 1, 1), "temperature": 6, "place": TEST_CITY_1},
|
||||
{"date": datetime(2011, 1, 2), "temperature": 8, "place": TEST_CITY_1},
|
||||
{"date": datetime(2011, 1, 3), "temperature": 5, "place": TEST_CITY_1},
|
||||
]
|
||||
|
||||
+239
-261
@@ -11,10 +11,9 @@ from .sample_data import TEST_DATA, TEST_CITY_1
|
||||
|
||||
|
||||
class DatabaseTestCase(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
self.db = connect()
|
||||
self.tbl = self.db['weather']
|
||||
self.tbl = self.db["weather"]
|
||||
self.tbl.insert_many(TEST_DATA)
|
||||
|
||||
def tearDown(self):
|
||||
@@ -22,28 +21,28 @@ class DatabaseTestCase(unittest.TestCase):
|
||||
self.db[table].drop()
|
||||
|
||||
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
|
||||
db = connect("sqlite:///:memory:/?cached_statements=1")
|
||||
assert "cached_statements" in db.url, db.url
|
||||
|
||||
def test_tables(self):
|
||||
assert self.db.tables == ['weather'], self.db.tables
|
||||
assert self.db.tables == ["weather"], self.db.tables
|
||||
|
||||
def test_contains(self):
|
||||
assert 'weather' in self.db, self.db.tables
|
||||
assert "weather" in self.db, self.db.tables
|
||||
|
||||
def test_create_table(self):
|
||||
table = self.db['foo']
|
||||
table = self.db["foo"]
|
||||
assert table.table.exists()
|
||||
assert len(table.table.columns) == 1, table.table.columns
|
||||
assert 'id' in table.table.c, table.table.c
|
||||
assert "id" in table.table.c, table.table.c
|
||||
|
||||
def test_create_table_no_ids(self):
|
||||
if 'mysql' in self.db.engine.dialect.dbapi.__name__:
|
||||
if "mysql" in self.db.engine.dialect.dbapi.__name__:
|
||||
return
|
||||
if 'sqlite' in self.db.engine.dialect.dbapi.__name__:
|
||||
if "sqlite" in self.db.engine.dialect.dbapi.__name__:
|
||||
return
|
||||
table = self.db.create_table("foo_no_id", primary_id=False)
|
||||
assert table.table.exists()
|
||||
@@ -55,8 +54,8 @@ class DatabaseTestCase(unittest.TestCase):
|
||||
assert table.table.exists()
|
||||
assert len(table.table.columns) == 1, table.table.columns
|
||||
assert pid in table.table.c, table.table.c
|
||||
table.insert({pid: 'foobar'})
|
||||
assert table.find_one(string_id='foobar')[pid] == 'foobar'
|
||||
table.insert({pid: "foobar"})
|
||||
assert table.find_one(string_id="foobar")[pid] == "foobar"
|
||||
|
||||
def test_create_table_custom_id2(self):
|
||||
pid = "string_id"
|
||||
@@ -65,8 +64,8 @@ class DatabaseTestCase(unittest.TestCase):
|
||||
assert len(table.table.columns) == 1, table.table.columns
|
||||
assert pid in table.table.c, table.table.c
|
||||
|
||||
table.insert({pid: 'foobar'})
|
||||
assert table.find_one(string_id='foobar')[pid] == 'foobar'
|
||||
table.insert({pid: "foobar"})
|
||||
assert table.find_one(string_id="foobar")[pid] == "foobar"
|
||||
|
||||
def test_create_table_custom_id3(self):
|
||||
pid = "int_id"
|
||||
@@ -83,253 +82,235 @@ class DatabaseTestCase(unittest.TestCase):
|
||||
|
||||
def test_create_table_shorthand1(self):
|
||||
pid = "int_id"
|
||||
table = self.db.get_table('foo5', pid)
|
||||
table = self.db.get_table("foo5", pid)
|
||||
assert table.table.exists
|
||||
assert len(table.table.columns) == 1, table.table.columns
|
||||
assert pid in table.table.c, table.table.c
|
||||
|
||||
table.insert({'int_id': 123})
|
||||
table.insert({'int_id': 124})
|
||||
assert table.find_one(int_id=123)['int_id'] == 123
|
||||
assert table.find_one(int_id=124)['int_id'] == 124
|
||||
self.assertRaises(IntegrityError,
|
||||
lambda: table.insert({'int_id': 123}))
|
||||
table.insert({"int_id": 123})
|
||||
table.insert({"int_id": 124})
|
||||
assert table.find_one(int_id=123)["int_id"] == 123
|
||||
assert table.find_one(int_id=124)["int_id"] == 124
|
||||
self.assertRaises(IntegrityError, lambda: table.insert({"int_id": 123}))
|
||||
|
||||
def test_create_table_shorthand2(self):
|
||||
pid = "string_id"
|
||||
table = self.db.get_table('foo6', primary_id=pid,
|
||||
primary_type=self.db.types.string(255))
|
||||
table = self.db.get_table(
|
||||
"foo6", primary_id=pid, primary_type=self.db.types.string(255)
|
||||
)
|
||||
assert table.table.exists
|
||||
assert len(table.table.columns) == 1, table.table.columns
|
||||
assert pid in table.table.c, table.table.c
|
||||
|
||||
table.insert({
|
||||
'string_id': 'foobar'})
|
||||
assert table.find_one(string_id='foobar')['string_id'] == 'foobar'
|
||||
table.insert({"string_id": "foobar"})
|
||||
assert table.find_one(string_id="foobar")["string_id"] == "foobar"
|
||||
|
||||
def test_with(self):
|
||||
init_length = len(self.db['weather'])
|
||||
init_length = len(self.db["weather"])
|
||||
with self.assertRaises(ValueError):
|
||||
with self.db as tx:
|
||||
tx['weather'].insert({'date': datetime(2011, 1, 1),
|
||||
'temperature': 1,
|
||||
'place': 'tmp_place'})
|
||||
tx["weather"].insert(
|
||||
{
|
||||
"date": datetime(2011, 1, 1),
|
||||
"temperature": 1,
|
||||
"place": "tmp_place",
|
||||
}
|
||||
)
|
||||
raise ValueError()
|
||||
assert len(self.db['weather']) == init_length
|
||||
assert len(self.db["weather"]) == init_length
|
||||
|
||||
def test_invalid_values(self):
|
||||
if 'mysql' in self.db.engine.dialect.dbapi.__name__:
|
||||
if "mysql" in self.db.engine.dialect.dbapi.__name__:
|
||||
# WARNING: mysql seems to be doing some weird type casting
|
||||
# upon insert. The mysql-python driver is not affected but
|
||||
# it isn't compatible with Python 3
|
||||
# Conclusion: use postgresql.
|
||||
return
|
||||
with self.assertRaises(SQLAlchemyError):
|
||||
tbl = self.db['weather']
|
||||
tbl.insert({
|
||||
'date': True,
|
||||
'temperature': 'wrong_value',
|
||||
'place': 'tmp_place'
|
||||
})
|
||||
tbl = self.db["weather"]
|
||||
tbl.insert(
|
||||
{"date": True, "temperature": "wrong_value", "place": "tmp_place"}
|
||||
)
|
||||
|
||||
def test_load_table(self):
|
||||
tbl = self.db.load_table('weather')
|
||||
tbl = self.db.load_table("weather")
|
||||
assert tbl.table.name == self.tbl.table.name
|
||||
|
||||
def test_query(self):
|
||||
r = self.db.query('SELECT COUNT(*) AS num FROM weather').next()
|
||||
assert r['num'] == len(TEST_DATA), r
|
||||
r = self.db.query("SELECT COUNT(*) AS num FROM weather").next()
|
||||
assert r["num"] == len(TEST_DATA), r
|
||||
|
||||
def test_table_cache_updates(self):
|
||||
tbl1 = self.db.get_table('people')
|
||||
data = OrderedDict([('first_name', 'John'), ('last_name', 'Smith')])
|
||||
tbl1 = self.db.get_table("people")
|
||||
data = OrderedDict([("first_name", "John"), ("last_name", "Smith")])
|
||||
tbl1.insert(data)
|
||||
data['id'] = 1
|
||||
tbl2 = self.db.get_table('people')
|
||||
data["id"] = 1
|
||||
tbl2 = self.db.get_table("people")
|
||||
assert dict(tbl2.all().next()) == dict(data), (tbl2.all().next(), data)
|
||||
|
||||
|
||||
class TableTestCase(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
self.db = connect()
|
||||
self.tbl = self.db['weather']
|
||||
self.tbl = self.db["weather"]
|
||||
for row in TEST_DATA:
|
||||
self.tbl.insert(row)
|
||||
|
||||
def test_insert(self):
|
||||
assert len(self.tbl) == len(TEST_DATA), len(self.tbl)
|
||||
last_id = self.tbl.insert({
|
||||
'date': datetime(2011, 1, 2),
|
||||
'temperature': -10,
|
||||
'place': 'Berlin'}
|
||||
last_id = self.tbl.insert(
|
||||
{"date": datetime(2011, 1, 2), "temperature": -10, "place": "Berlin"}
|
||||
)
|
||||
assert len(self.tbl) == len(TEST_DATA) + 1, len(self.tbl)
|
||||
assert self.tbl.find_one(id=last_id)['place'] == 'Berlin'
|
||||
assert self.tbl.find_one(id=last_id)["place"] == "Berlin"
|
||||
|
||||
def test_insert_ignore(self):
|
||||
self.tbl.insert_ignore({
|
||||
'date': datetime(2011, 1, 2),
|
||||
'temperature': -10,
|
||||
'place': 'Berlin'},
|
||||
['place']
|
||||
self.tbl.insert_ignore(
|
||||
{"date": datetime(2011, 1, 2), "temperature": -10, "place": "Berlin"},
|
||||
["place"],
|
||||
)
|
||||
assert len(self.tbl) == len(TEST_DATA) + 1, len(self.tbl)
|
||||
self.tbl.insert_ignore({
|
||||
'date': datetime(2011, 1, 2),
|
||||
'temperature': -10,
|
||||
'place': 'Berlin'},
|
||||
['place']
|
||||
self.tbl.insert_ignore(
|
||||
{"date": datetime(2011, 1, 2), "temperature": -10, "place": "Berlin"},
|
||||
["place"],
|
||||
)
|
||||
assert len(self.tbl) == len(TEST_DATA) + 1, len(self.tbl)
|
||||
|
||||
def test_insert_ignore_all_key(self):
|
||||
for i in range(0, 4):
|
||||
self.tbl.insert_ignore({
|
||||
'date': datetime(2011, 1, 2),
|
||||
'temperature': -10,
|
||||
'place': 'Berlin'},
|
||||
['date', 'temperature', 'place']
|
||||
self.tbl.insert_ignore(
|
||||
{"date": datetime(2011, 1, 2), "temperature": -10, "place": "Berlin"},
|
||||
["date", "temperature", "place"],
|
||||
)
|
||||
assert len(self.tbl) == len(TEST_DATA) + 1, len(self.tbl)
|
||||
|
||||
def test_insert_json(self):
|
||||
last_id = self.tbl.insert({
|
||||
'date': datetime(2011, 1, 2),
|
||||
'temperature': -10,
|
||||
'place': 'Berlin',
|
||||
'info': {
|
||||
'currency': 'EUR',
|
||||
'language': 'German',
|
||||
'population': 3292365
|
||||
last_id = self.tbl.insert(
|
||||
{
|
||||
"date": datetime(2011, 1, 2),
|
||||
"temperature": -10,
|
||||
"place": "Berlin",
|
||||
"info": {
|
||||
"currency": "EUR",
|
||||
"language": "German",
|
||||
"population": 3292365,
|
||||
},
|
||||
}
|
||||
})
|
||||
assert len(self.tbl) == len(TEST_DATA) + 1, len(self.tbl)
|
||||
assert self.tbl.find_one(id=last_id)['place'] == 'Berlin'
|
||||
|
||||
def test_upsert(self):
|
||||
self.tbl.upsert({
|
||||
'date': datetime(2011, 1, 2),
|
||||
'temperature': -10,
|
||||
'place': 'Berlin'},
|
||||
['place']
|
||||
)
|
||||
assert len(self.tbl) == len(TEST_DATA) + 1, len(self.tbl)
|
||||
self.tbl.upsert({
|
||||
'date': datetime(2011, 1, 2),
|
||||
'temperature': -10,
|
||||
'place': 'Berlin'},
|
||||
['place']
|
||||
assert self.tbl.find_one(id=last_id)["place"] == "Berlin"
|
||||
|
||||
def test_upsert(self):
|
||||
self.tbl.upsert(
|
||||
{"date": datetime(2011, 1, 2), "temperature": -10, "place": "Berlin"},
|
||||
["place"],
|
||||
)
|
||||
assert len(self.tbl) == len(TEST_DATA) + 1, len(self.tbl)
|
||||
self.tbl.upsert(
|
||||
{"date": datetime(2011, 1, 2), "temperature": -10, "place": "Berlin"},
|
||||
["place"],
|
||||
)
|
||||
assert len(self.tbl) == len(TEST_DATA) + 1, len(self.tbl)
|
||||
|
||||
def test_upsert_single_column(self):
|
||||
table = self.db['banana_single_col']
|
||||
table.upsert({
|
||||
'color': 'Yellow'},
|
||||
['color']
|
||||
)
|
||||
table = self.db["banana_single_col"]
|
||||
table.upsert({"color": "Yellow"}, ["color"])
|
||||
assert len(table) == 1, len(table)
|
||||
table.upsert({
|
||||
'color': 'Yellow'},
|
||||
['color']
|
||||
)
|
||||
table.upsert({"color": "Yellow"}, ["color"])
|
||||
assert len(table) == 1, len(table)
|
||||
|
||||
def test_upsert_all_key(self):
|
||||
assert len(self.tbl) == len(TEST_DATA), len(self.tbl)
|
||||
for i in range(0, 2):
|
||||
self.tbl.upsert({
|
||||
'date': datetime(2011, 1, 2),
|
||||
'temperature': -10,
|
||||
'place': 'Berlin'},
|
||||
['date', 'temperature', 'place']
|
||||
self.tbl.upsert(
|
||||
{"date": datetime(2011, 1, 2), "temperature": -10, "place": "Berlin"},
|
||||
["date", "temperature", "place"],
|
||||
)
|
||||
assert len(self.tbl) == len(TEST_DATA) + 1, len(self.tbl)
|
||||
|
||||
def test_upsert_id(self):
|
||||
table = self.db['banana_with_id']
|
||||
data = dict(id=10, title='I am a banana!')
|
||||
table.upsert(data, ['id'])
|
||||
table = self.db["banana_with_id"]
|
||||
data = dict(id=10, title="I am a banana!")
|
||||
table.upsert(data, ["id"])
|
||||
assert len(table) == 1, len(table)
|
||||
|
||||
def test_update_while_iter(self):
|
||||
for row in self.tbl:
|
||||
row['foo'] = 'bar'
|
||||
self.tbl.update(row, ['place', 'date'])
|
||||
row["foo"] = "bar"
|
||||
self.tbl.update(row, ["place", "date"])
|
||||
assert len(self.tbl) == len(TEST_DATA), len(self.tbl)
|
||||
|
||||
def test_weird_column_names(self):
|
||||
with self.assertRaises(ValueError):
|
||||
self.tbl.insert({
|
||||
'date': datetime(2011, 1, 2),
|
||||
'temperature': -10,
|
||||
'foo.bar': 'Berlin',
|
||||
'qux.bar': 'Huhu'
|
||||
})
|
||||
self.tbl.insert(
|
||||
{
|
||||
"date": datetime(2011, 1, 2),
|
||||
"temperature": -10,
|
||||
"foo.bar": "Berlin",
|
||||
"qux.bar": "Huhu",
|
||||
}
|
||||
)
|
||||
|
||||
def test_cased_column_names(self):
|
||||
tbl = self.db['cased_column_names']
|
||||
tbl.insert({
|
||||
'place': 'Berlin',
|
||||
})
|
||||
tbl.insert({
|
||||
'Place': 'Berlin',
|
||||
})
|
||||
tbl.insert({
|
||||
'PLACE ': 'Berlin',
|
||||
})
|
||||
tbl = self.db["cased_column_names"]
|
||||
tbl.insert(
|
||||
{"place": "Berlin",}
|
||||
)
|
||||
tbl.insert(
|
||||
{"Place": "Berlin",}
|
||||
)
|
||||
tbl.insert(
|
||||
{"PLACE ": "Berlin",}
|
||||
)
|
||||
assert len(tbl.columns) == 2, tbl.columns
|
||||
assert len(list(tbl.find(Place='Berlin'))) == 3
|
||||
assert len(list(tbl.find(place='Berlin'))) == 3
|
||||
assert len(list(tbl.find(PLACE='Berlin'))) == 3
|
||||
assert len(list(tbl.find(Place="Berlin"))) == 3
|
||||
assert len(list(tbl.find(place="Berlin"))) == 3
|
||||
assert len(list(tbl.find(PLACE="Berlin"))) == 3
|
||||
|
||||
def test_invalid_column_names(self):
|
||||
tbl = self.db['weather']
|
||||
tbl = self.db["weather"]
|
||||
with self.assertRaises(ValueError):
|
||||
tbl.insert({None: 'banana'})
|
||||
tbl.insert({None: "banana"})
|
||||
|
||||
with self.assertRaises(ValueError):
|
||||
tbl.insert({'': 'banana'})
|
||||
tbl.insert({"": "banana"})
|
||||
|
||||
with self.assertRaises(ValueError):
|
||||
tbl.insert({'-': 'banana'})
|
||||
tbl.insert({"-": "banana"})
|
||||
|
||||
def test_delete(self):
|
||||
self.tbl.insert({
|
||||
'date': datetime(2011, 1, 2),
|
||||
'temperature': -10,
|
||||
'place': 'Berlin'}
|
||||
self.tbl.insert(
|
||||
{"date": datetime(2011, 1, 2), "temperature": -10, "place": "Berlin"}
|
||||
)
|
||||
original_count = len(self.tbl)
|
||||
assert len(self.tbl) == len(TEST_DATA) + 1, len(self.tbl)
|
||||
# Test bad use of API
|
||||
with self.assertRaises(ArgumentError):
|
||||
self.tbl.delete({'place': 'Berlin'})
|
||||
self.tbl.delete({"place": "Berlin"})
|
||||
assert len(self.tbl) == original_count, len(self.tbl)
|
||||
|
||||
assert self.tbl.delete(place='Berlin') is True, 'should return 1'
|
||||
assert self.tbl.delete(place="Berlin") is True, "should return 1"
|
||||
assert len(self.tbl) == len(TEST_DATA), len(self.tbl)
|
||||
assert self.tbl.delete() is True, 'should return non zero'
|
||||
assert self.tbl.delete() is True, "should return non zero"
|
||||
assert len(self.tbl) == 0, len(self.tbl)
|
||||
|
||||
def test_repr(self):
|
||||
assert repr(self.tbl) == '<Table(weather)>', \
|
||||
'the representation should be <Table(weather)>'
|
||||
assert (
|
||||
repr(self.tbl) == "<Table(weather)>"
|
||||
), "the representation should be <Table(weather)>"
|
||||
|
||||
def test_delete_nonexist_entry(self):
|
||||
assert self.tbl.delete(place='Berlin') is False, \
|
||||
'entry not exist, should fail to delete'
|
||||
assert (
|
||||
self.tbl.delete(place="Berlin") is False
|
||||
), "entry not exist, should fail to delete"
|
||||
|
||||
def test_find_one(self):
|
||||
self.tbl.insert({
|
||||
'date': datetime(2011, 1, 2),
|
||||
'temperature': -10,
|
||||
'place': 'Berlin'
|
||||
})
|
||||
d = self.tbl.find_one(place='Berlin')
|
||||
assert d['temperature'] == -10, d
|
||||
d = self.tbl.find_one(place='Atlantis')
|
||||
self.tbl.insert(
|
||||
{"date": datetime(2011, 1, 2), "temperature": -10, "place": "Berlin"}
|
||||
)
|
||||
d = self.tbl.find_one(place="Berlin")
|
||||
assert d["temperature"] == -10, d
|
||||
d = self.tbl.find_one(place="Atlantis")
|
||||
assert d is None, d
|
||||
|
||||
def test_count(self):
|
||||
@@ -348,31 +329,31 @@ class TableTestCase(unittest.TestCase):
|
||||
assert len(ds) == 1, ds
|
||||
ds = list(self.tbl.find(_step=2))
|
||||
assert len(ds) == len(TEST_DATA), ds
|
||||
ds = list(self.tbl.find(order_by=['temperature']))
|
||||
assert ds[0]['temperature'] == -1, ds
|
||||
ds = list(self.tbl.find(order_by=['-temperature']))
|
||||
assert ds[0]['temperature'] == 8, ds
|
||||
ds = list(self.tbl.find(order_by=["temperature"]))
|
||||
assert ds[0]["temperature"] == -1, ds
|
||||
ds = list(self.tbl.find(order_by=["-temperature"]))
|
||||
assert ds[0]["temperature"] == 8, ds
|
||||
ds = list(self.tbl.find(self.tbl.table.columns.temperature > 4))
|
||||
assert len(ds) == 3, ds
|
||||
|
||||
def test_find_dsl(self):
|
||||
ds = list(self.tbl.find(place={'like': '%lw%'}))
|
||||
ds = list(self.tbl.find(place={"like": "%lw%"}))
|
||||
assert len(ds) == 3, ds
|
||||
ds = list(self.tbl.find(temperature={'>': 5}))
|
||||
ds = list(self.tbl.find(temperature={">": 5}))
|
||||
assert len(ds) == 2, ds
|
||||
ds = list(self.tbl.find(temperature={'>=': 5}))
|
||||
ds = list(self.tbl.find(temperature={">=": 5}))
|
||||
assert len(ds) == 3, ds
|
||||
ds = list(self.tbl.find(temperature={'<': 0}))
|
||||
ds = list(self.tbl.find(temperature={"<": 0}))
|
||||
assert len(ds) == 1, ds
|
||||
ds = list(self.tbl.find(temperature={'<=': 0}))
|
||||
ds = list(self.tbl.find(temperature={"<=": 0}))
|
||||
assert len(ds) == 2, ds
|
||||
ds = list(self.tbl.find(temperature={'!=': -1}))
|
||||
ds = list(self.tbl.find(temperature={"!=": -1}))
|
||||
assert len(ds) == 5, ds
|
||||
ds = list(self.tbl.find(temperature={'between': [5, 8]}))
|
||||
ds = list(self.tbl.find(temperature={"between": [5, 8]}))
|
||||
assert len(ds) == 3, ds
|
||||
ds = list(self.tbl.find(place={'=': 'G€lway'}))
|
||||
ds = list(self.tbl.find(place={"=": "G€lway"}))
|
||||
assert len(ds) == 3, ds
|
||||
ds = list(self.tbl.find(place={'ilike': '%LwAy'}))
|
||||
ds = list(self.tbl.find(place={"ilike": "%LwAy"}))
|
||||
assert len(ds) == 3, ds
|
||||
|
||||
def test_offset(self):
|
||||
@@ -385,23 +366,26 @@ class TableTestCase(unittest.TestCase):
|
||||
ds = list(self.tbl.find(place=TEST_CITY_1, _streamed=True, _step=1))
|
||||
assert len(ds) == 3, len(ds)
|
||||
for row in self.tbl.find(place=TEST_CITY_1, _streamed=True, _step=1):
|
||||
row['temperature'] = -1
|
||||
self.tbl.update(row, ['id'])
|
||||
row["temperature"] = -1
|
||||
self.tbl.update(row, ["id"])
|
||||
|
||||
def test_distinct(self):
|
||||
x = list(self.tbl.distinct('place'))
|
||||
x = list(self.tbl.distinct("place"))
|
||||
assert len(x) == 2, x
|
||||
x = list(self.tbl.distinct('place', 'date'))
|
||||
x = list(self.tbl.distinct("place", "date"))
|
||||
assert len(x) == 6, x
|
||||
x = list(self.tbl.distinct(
|
||||
'place', 'date',
|
||||
self.tbl.table.columns.date >= datetime(2011, 1, 2, 0, 0)))
|
||||
x = list(
|
||||
self.tbl.distinct(
|
||||
"place",
|
||||
"date",
|
||||
self.tbl.table.columns.date >= datetime(2011, 1, 2, 0, 0),
|
||||
)
|
||||
)
|
||||
assert len(x) == 4, x
|
||||
|
||||
x = list(self.tbl.distinct('temperature', place='B€rkeley'))
|
||||
x = list(self.tbl.distinct("temperature", place="B€rkeley"))
|
||||
assert len(x) == 3, x
|
||||
x = list(self.tbl.distinct('temperature',
|
||||
place=['B€rkeley', 'G€lway']))
|
||||
x = list(self.tbl.distinct("temperature", place=["B€rkeley", "G€lway"]))
|
||||
assert len(x) == 6, x
|
||||
|
||||
def test_insert_many(self):
|
||||
@@ -423,6 +407,7 @@ class TableTestCase(unittest.TestCase):
|
||||
def callback(queue):
|
||||
nonlocal N
|
||||
N += len(queue)
|
||||
|
||||
with chunked.ChunkedInsert(self.tbl, callback=callback) as chunk_tbl:
|
||||
for item in data:
|
||||
chunk_tbl.insert(item)
|
||||
@@ -430,74 +415,73 @@ class TableTestCase(unittest.TestCase):
|
||||
assert len(self.tbl) == len(data) + 6
|
||||
|
||||
def test_update_many(self):
|
||||
tbl = self.db['update_many_test']
|
||||
tbl.insert_many([
|
||||
dict(temp=10), dict(temp=20), dict(temp=30)
|
||||
])
|
||||
tbl.update_many(
|
||||
[dict(id=1, temp=50), dict(id=3, temp=50)], 'id'
|
||||
)
|
||||
tbl = self.db["update_many_test"]
|
||||
tbl.insert_many([dict(temp=10), dict(temp=20), dict(temp=30)])
|
||||
tbl.update_many([dict(id=1, temp=50), dict(id=3, temp=50)], "id")
|
||||
|
||||
# Ensure data has been updated.
|
||||
assert tbl.find_one(id=1)['temp'] == tbl.find_one(id=3)['temp']
|
||||
assert tbl.find_one(id=1)["temp"] == tbl.find_one(id=3)["temp"]
|
||||
|
||||
def test_chunked_update(self):
|
||||
tbl = self.db['update_many_test']
|
||||
tbl.insert_many([
|
||||
dict(temp=10, location='asdf'), dict(temp=20, location='qwer'),
|
||||
dict(temp=30, location='asdf')
|
||||
])
|
||||
tbl = self.db["update_many_test"]
|
||||
tbl.insert_many(
|
||||
[
|
||||
dict(temp=10, location="asdf"),
|
||||
dict(temp=20, location="qwer"),
|
||||
dict(temp=30, location="asdf"),
|
||||
]
|
||||
)
|
||||
|
||||
chunked_tbl = chunked.ChunkedUpdate(tbl, 'id')
|
||||
chunked_tbl = chunked.ChunkedUpdate(tbl, "id")
|
||||
chunked_tbl.update(dict(id=1, temp=50))
|
||||
chunked_tbl.update(dict(id=2, location='asdf'))
|
||||
chunked_tbl.update(dict(id=2, location="asdf"))
|
||||
chunked_tbl.update(dict(id=3, temp=50))
|
||||
chunked_tbl.flush()
|
||||
|
||||
# Ensure data has been updated.
|
||||
assert tbl.find_one(id=1)['temp'] == tbl.find_one(id=3)['temp'] == 50
|
||||
assert tbl.find_one(id=2)['location'] == tbl.find_one(id=3)['location'] == 'asdf' # noqa
|
||||
assert tbl.find_one(id=1)["temp"] == tbl.find_one(id=3)["temp"] == 50
|
||||
assert (
|
||||
tbl.find_one(id=2)["location"] == tbl.find_one(id=3)["location"] == "asdf"
|
||||
) # noqa
|
||||
|
||||
def test_upsert_many(self):
|
||||
# Also tests updating on records with different attributes
|
||||
tbl = self.db['upsert_many_test']
|
||||
tbl = self.db["upsert_many_test"]
|
||||
|
||||
W = 100
|
||||
tbl.upsert_many([dict(age=10), dict(weight=W)], 'id')
|
||||
assert tbl.find_one(id=1)['age'] == 10
|
||||
tbl.upsert_many([dict(age=10), dict(weight=W)], "id")
|
||||
assert tbl.find_one(id=1)["age"] == 10
|
||||
|
||||
tbl.upsert_many([dict(id=1, age=70), dict(id=2, weight=W / 2)], 'id')
|
||||
assert tbl.find_one(id=2)['weight'] == W / 2
|
||||
tbl.upsert_many([dict(id=1, age=70), dict(id=2, weight=W / 2)], "id")
|
||||
assert tbl.find_one(id=2)["weight"] == W / 2
|
||||
|
||||
def test_drop_operations(self):
|
||||
assert self.tbl._table is not None, \
|
||||
'table shouldn\'t be dropped yet'
|
||||
assert self.tbl._table is not None, "table shouldn't be dropped yet"
|
||||
self.tbl.drop()
|
||||
assert self.tbl._table is None, \
|
||||
'table should be dropped now'
|
||||
assert self.tbl._table is None, "table should be dropped now"
|
||||
assert list(self.tbl.all()) == [], self.tbl.all()
|
||||
assert self.tbl.count() == 0, self.tbl.count()
|
||||
|
||||
def test_table_drop(self):
|
||||
assert 'weather' in self.db
|
||||
self.db['weather'].drop()
|
||||
assert 'weather' not in self.db
|
||||
assert "weather" in self.db
|
||||
self.db["weather"].drop()
|
||||
assert "weather" not in self.db
|
||||
|
||||
def test_table_drop_then_create(self):
|
||||
assert 'weather' in self.db
|
||||
self.db['weather'].drop()
|
||||
assert 'weather' not in self.db
|
||||
self.db['weather'].insert({'foo': 'bar'})
|
||||
assert "weather" in self.db
|
||||
self.db["weather"].drop()
|
||||
assert "weather" not in self.db
|
||||
self.db["weather"].insert({"foo": "bar"})
|
||||
|
||||
def test_columns(self):
|
||||
cols = self.tbl.columns
|
||||
assert len(list(cols)) == 4, 'column count mismatch'
|
||||
assert 'date' in cols and 'temperature' in cols and 'place' in cols
|
||||
assert len(list(cols)) == 4, "column count mismatch"
|
||||
assert "date" in cols and "temperature" in cols and "place" in cols
|
||||
|
||||
def test_drop_column(self):
|
||||
try:
|
||||
self.tbl.drop_column('date')
|
||||
assert 'date' not in self.tbl.columns
|
||||
self.tbl.drop_column("date")
|
||||
assert "date" not in self.tbl.columns
|
||||
except RuntimeError:
|
||||
pass
|
||||
|
||||
@@ -509,71 +493,67 @@ class TableTestCase(unittest.TestCase):
|
||||
|
||||
def test_update(self):
|
||||
date = datetime(2011, 1, 2)
|
||||
res = self.tbl.update({
|
||||
'date': date,
|
||||
'temperature': -10,
|
||||
'place': TEST_CITY_1},
|
||||
['place', 'date']
|
||||
res = self.tbl.update(
|
||||
{"date": date, "temperature": -10, "place": TEST_CITY_1}, ["place", "date"]
|
||||
)
|
||||
assert res, 'update should return True'
|
||||
assert res, "update should return True"
|
||||
m = self.tbl.find_one(place=TEST_CITY_1, date=date)
|
||||
assert m['temperature'] == -10, \
|
||||
'new temp. should be -10 but is %d' % m['temperature']
|
||||
assert m["temperature"] == -10, (
|
||||
"new temp. should be -10 but is %d" % m["temperature"]
|
||||
)
|
||||
|
||||
def test_create_column(self):
|
||||
tbl = self.tbl
|
||||
tbl.create_column('foo', FLOAT)
|
||||
assert 'foo' in tbl.table.c, tbl.table.c
|
||||
assert isinstance(tbl.table.c['foo'].type, FLOAT), \
|
||||
tbl.table.c['foo'].type
|
||||
assert 'foo' in tbl.columns, tbl.columns
|
||||
tbl.create_column("foo", FLOAT)
|
||||
assert "foo" in tbl.table.c, tbl.table.c
|
||||
assert isinstance(tbl.table.c["foo"].type, FLOAT), tbl.table.c["foo"].type
|
||||
assert "foo" in tbl.columns, tbl.columns
|
||||
|
||||
def test_ensure_column(self):
|
||||
tbl = self.tbl
|
||||
tbl.create_column_by_example('foo', 0.1)
|
||||
assert 'foo' in tbl.table.c, tbl.table.c
|
||||
assert isinstance(tbl.table.c['foo'].type, FLOAT), \
|
||||
tbl.table.c['bar'].type
|
||||
tbl.create_column_by_example('bar', 1)
|
||||
assert 'bar' in tbl.table.c, tbl.table.c
|
||||
assert isinstance(tbl.table.c['bar'].type, BIGINT), \
|
||||
tbl.table.c['bar'].type
|
||||
tbl.create_column_by_example('pippo', 'test')
|
||||
assert 'pippo' in tbl.table.c, tbl.table.c
|
||||
assert isinstance(tbl.table.c['pippo'].type, TEXT), \
|
||||
tbl.table.c['pippo'].type
|
||||
tbl.create_column_by_example('bigbar', 11111111111)
|
||||
assert 'bigbar' in tbl.table.c, tbl.table.c
|
||||
assert isinstance(tbl.table.c['bigbar'].type, BIGINT), \
|
||||
tbl.table.c['bigbar'].type
|
||||
tbl.create_column_by_example('littlebar', -11111111111)
|
||||
assert 'littlebar' in tbl.table.c, tbl.table.c
|
||||
assert isinstance(tbl.table.c['littlebar'].type, BIGINT), \
|
||||
tbl.table.c['littlebar'].type
|
||||
tbl.create_column_by_example("foo", 0.1)
|
||||
assert "foo" in tbl.table.c, tbl.table.c
|
||||
assert isinstance(tbl.table.c["foo"].type, FLOAT), tbl.table.c["bar"].type
|
||||
tbl.create_column_by_example("bar", 1)
|
||||
assert "bar" in tbl.table.c, tbl.table.c
|
||||
assert isinstance(tbl.table.c["bar"].type, BIGINT), tbl.table.c["bar"].type
|
||||
tbl.create_column_by_example("pippo", "test")
|
||||
assert "pippo" in tbl.table.c, tbl.table.c
|
||||
assert isinstance(tbl.table.c["pippo"].type, TEXT), tbl.table.c["pippo"].type
|
||||
tbl.create_column_by_example("bigbar", 11111111111)
|
||||
assert "bigbar" in tbl.table.c, tbl.table.c
|
||||
assert isinstance(tbl.table.c["bigbar"].type, BIGINT), tbl.table.c[
|
||||
"bigbar"
|
||||
].type
|
||||
tbl.create_column_by_example("littlebar", -11111111111)
|
||||
assert "littlebar" in tbl.table.c, tbl.table.c
|
||||
assert isinstance(tbl.table.c["littlebar"].type, BIGINT), tbl.table.c[
|
||||
"littlebar"
|
||||
].type
|
||||
|
||||
def test_key_order(self):
|
||||
res = self.db.query('SELECT temperature, place FROM weather LIMIT 1')
|
||||
res = self.db.query("SELECT temperature, place FROM weather LIMIT 1")
|
||||
keys = list(res.next().keys())
|
||||
assert keys[0] == 'temperature'
|
||||
assert keys[1] == 'place'
|
||||
assert keys[0] == "temperature"
|
||||
assert keys[1] == "place"
|
||||
|
||||
def test_empty_query(self):
|
||||
empty = list(self.tbl.find(place='not in data'))
|
||||
empty = list(self.tbl.find(place="not in data"))
|
||||
assert len(empty) == 0, empty
|
||||
|
||||
|
||||
class Constructor(dict):
|
||||
""" Very simple low-functionality extension to ``dict`` to
|
||||
provide attribute access to dictionary contents"""
|
||||
|
||||
def __getattr__(self, name):
|
||||
return self[name]
|
||||
|
||||
|
||||
class RowTypeTestCase(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
self.db = connect(row_type=Constructor)
|
||||
self.tbl = self.db['weather']
|
||||
self.tbl = self.db["weather"]
|
||||
for row in TEST_DATA:
|
||||
self.tbl.insert(row)
|
||||
|
||||
@@ -582,15 +562,13 @@ class RowTypeTestCase(unittest.TestCase):
|
||||
self.db[table].drop()
|
||||
|
||||
def test_find_one(self):
|
||||
self.tbl.insert({
|
||||
'date': datetime(2011, 1, 2),
|
||||
'temperature': -10,
|
||||
'place': 'Berlin'}
|
||||
self.tbl.insert(
|
||||
{"date": datetime(2011, 1, 2), "temperature": -10, "place": "Berlin"}
|
||||
)
|
||||
d = self.tbl.find_one(place='Berlin')
|
||||
assert d['temperature'] == -10, d
|
||||
d = self.tbl.find_one(place="Berlin")
|
||||
assert d["temperature"] == -10, d
|
||||
assert d.temperature == -10, d
|
||||
d = self.tbl.find_one(place='Atlantis')
|
||||
d = self.tbl.find_one(place="Atlantis")
|
||||
assert d is None, d
|
||||
|
||||
def test_find(self):
|
||||
@@ -604,11 +582,11 @@ class RowTypeTestCase(unittest.TestCase):
|
||||
assert isinstance(item, Constructor), item
|
||||
|
||||
def test_distinct(self):
|
||||
x = list(self.tbl.distinct('place'))
|
||||
x = list(self.tbl.distinct("place"))
|
||||
assert len(x) == 2, x
|
||||
for item in x:
|
||||
assert isinstance(item, Constructor), item
|
||||
x = list(self.tbl.distinct('place', 'date'))
|
||||
x = list(self.tbl.distinct("place", "date"))
|
||||
assert len(x) == 6, x
|
||||
for item in x:
|
||||
assert isinstance(item, Constructor), item
|
||||
@@ -621,5 +599,5 @@ class RowTypeTestCase(unittest.TestCase):
|
||||
assert c == len(self.tbl)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
Reference in New Issue
Block a user