mirror of
https://github.com/beetbox/beets.git
synced 2025-12-06 08:39:17 +01:00
a57c164348
Remove 'NamedQuery' since it is not being used by any queries any more. This simplifies query parsing logic in 'queryparse.py'. We know that this logic can only receive 'FieldQuery' thus I adjusted types and removed the logic that handles other cases. Effectively, this means that the query parsing logic does not any more care whether the query is named by the corresponding DB field. Instead, queries like 'SingletonQuery' and 'PlaylistQuery' are responsible for translating 'singleton' and 'playlist' to the underlying DB filters.
772 lines
23 KiB
Python
772 lines
23 KiB
Python
# This file is part of beets.
|
|
# Copyright 2016, Adrian Sampson.
|
|
#
|
|
# Permission is hereby granted, free of charge, to any person obtaining
|
|
# a copy of this software and associated documentation files (the
|
|
# "Software"), to deal in the Software without restriction, including
|
|
# without limitation the rights to use, copy, modify, merge, publish,
|
|
# distribute, sublicense, and/or sell copies of the Software, and to
|
|
# permit persons to whom the Software is furnished to do so, subject to
|
|
# the following conditions:
|
|
#
|
|
# The above copyright notice and this permission notice shall be
|
|
# included in all copies or substantial portions of the Software.
|
|
|
|
"""Tests for the DBCore database abstraction."""
|
|
|
|
import os
|
|
import shutil
|
|
import sqlite3
|
|
import unittest
|
|
from tempfile import mkstemp
|
|
|
|
from beets import dbcore
|
|
from beets.test import _common
|
|
|
|
# Fixture: concrete database and model classes. For migration tests, we
|
|
# have multiple models with different numbers of fields.
|
|
|
|
|
|
class SortFixture(dbcore.query.FieldSort):
|
|
pass
|
|
|
|
|
|
class QueryFixture(dbcore.query.FieldQuery):
|
|
def __init__(self, pattern):
|
|
self.pattern = pattern
|
|
|
|
def clause(self):
|
|
return None, ()
|
|
|
|
def match(self):
|
|
return True
|
|
|
|
|
|
class ModelFixture1(dbcore.Model):
|
|
_table = "test"
|
|
_flex_table = "testflex"
|
|
_fields = {
|
|
"id": dbcore.types.PRIMARY_ID,
|
|
"field_one": dbcore.types.INTEGER,
|
|
"field_two": dbcore.types.STRING,
|
|
}
|
|
_types = {
|
|
"some_float_field": dbcore.types.FLOAT,
|
|
}
|
|
_sorts = {
|
|
"some_sort": SortFixture,
|
|
}
|
|
_queries = {
|
|
"some_query": QueryFixture,
|
|
}
|
|
|
|
@classmethod
|
|
def _getters(cls):
|
|
return {}
|
|
|
|
def _template_funcs(self):
|
|
return {}
|
|
|
|
|
|
class DatabaseFixture1(dbcore.Database):
|
|
_models = (ModelFixture1,)
|
|
pass
|
|
|
|
|
|
class ModelFixture2(ModelFixture1):
|
|
_fields = {
|
|
"id": dbcore.types.PRIMARY_ID,
|
|
"field_one": dbcore.types.INTEGER,
|
|
"field_two": dbcore.types.INTEGER,
|
|
}
|
|
|
|
|
|
class DatabaseFixture2(dbcore.Database):
|
|
_models = (ModelFixture2,)
|
|
pass
|
|
|
|
|
|
class ModelFixture3(ModelFixture1):
|
|
_fields = {
|
|
"id": dbcore.types.PRIMARY_ID,
|
|
"field_one": dbcore.types.INTEGER,
|
|
"field_two": dbcore.types.INTEGER,
|
|
"field_three": dbcore.types.INTEGER,
|
|
}
|
|
|
|
|
|
class DatabaseFixture3(dbcore.Database):
|
|
_models = (ModelFixture3,)
|
|
pass
|
|
|
|
|
|
class ModelFixture4(ModelFixture1):
|
|
_fields = {
|
|
"id": dbcore.types.PRIMARY_ID,
|
|
"field_one": dbcore.types.INTEGER,
|
|
"field_two": dbcore.types.INTEGER,
|
|
"field_three": dbcore.types.INTEGER,
|
|
"field_four": dbcore.types.INTEGER,
|
|
}
|
|
|
|
|
|
class DatabaseFixture4(dbcore.Database):
|
|
_models = (ModelFixture4,)
|
|
pass
|
|
|
|
|
|
class AnotherModelFixture(ModelFixture1):
|
|
_table = "another"
|
|
_flex_table = "anotherflex"
|
|
_fields = {
|
|
"id": dbcore.types.PRIMARY_ID,
|
|
"foo": dbcore.types.INTEGER,
|
|
}
|
|
|
|
|
|
class ModelFixture5(ModelFixture1):
|
|
_fields = {
|
|
"some_string_field": dbcore.types.STRING,
|
|
"some_float_field": dbcore.types.FLOAT,
|
|
"some_boolean_field": dbcore.types.BOOLEAN,
|
|
}
|
|
|
|
|
|
class DatabaseFixture5(dbcore.Database):
|
|
_models = (ModelFixture5,)
|
|
pass
|
|
|
|
|
|
class DatabaseFixtureTwoModels(dbcore.Database):
|
|
_models = (ModelFixture2, AnotherModelFixture)
|
|
pass
|
|
|
|
|
|
class ModelFixtureWithGetters(dbcore.Model):
|
|
@classmethod
|
|
def _getters(cls):
|
|
return {"aComputedField": (lambda s: "thing")}
|
|
|
|
def _template_funcs(self):
|
|
return {}
|
|
|
|
|
|
@_common.slow_test()
|
|
class MigrationTest(unittest.TestCase):
|
|
"""Tests the ability to change the database schema between
|
|
versions.
|
|
"""
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
handle, cls.orig_libfile = mkstemp("orig_db")
|
|
os.close(handle)
|
|
# Set up a database with the two-field schema.
|
|
old_lib = DatabaseFixture2(cls.orig_libfile)
|
|
|
|
# Add an item to the old library.
|
|
old_lib._connection().execute(
|
|
"insert into test (field_one, field_two) values (4, 2)"
|
|
)
|
|
old_lib._connection().commit()
|
|
old_lib._connection().close()
|
|
del old_lib
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
os.remove(cls.orig_libfile)
|
|
|
|
def setUp(self):
|
|
handle, self.libfile = mkstemp("db")
|
|
os.close(handle)
|
|
shutil.copyfile(self.orig_libfile, self.libfile)
|
|
|
|
def tearDown(self):
|
|
os.remove(self.libfile)
|
|
|
|
def test_open_with_same_fields_leaves_untouched(self):
|
|
new_lib = DatabaseFixture2(self.libfile)
|
|
c = new_lib._connection().cursor()
|
|
c.execute("select * from test")
|
|
row = c.fetchone()
|
|
c.connection.close()
|
|
self.assertEqual(len(row.keys()), len(ModelFixture2._fields))
|
|
|
|
def test_open_with_new_field_adds_column(self):
|
|
new_lib = DatabaseFixture3(self.libfile)
|
|
c = new_lib._connection().cursor()
|
|
c.execute("select * from test")
|
|
row = c.fetchone()
|
|
c.connection.close()
|
|
self.assertEqual(len(row.keys()), len(ModelFixture3._fields))
|
|
|
|
def test_open_with_fewer_fields_leaves_untouched(self):
|
|
new_lib = DatabaseFixture1(self.libfile)
|
|
c = new_lib._connection().cursor()
|
|
c.execute("select * from test")
|
|
row = c.fetchone()
|
|
c.connection.close()
|
|
self.assertEqual(len(row.keys()), len(ModelFixture2._fields))
|
|
|
|
def test_open_with_multiple_new_fields(self):
|
|
new_lib = DatabaseFixture4(self.libfile)
|
|
c = new_lib._connection().cursor()
|
|
c.execute("select * from test")
|
|
row = c.fetchone()
|
|
c.connection.close()
|
|
self.assertEqual(len(row.keys()), len(ModelFixture4._fields))
|
|
|
|
def test_extra_model_adds_table(self):
|
|
new_lib = DatabaseFixtureTwoModels(self.libfile)
|
|
try:
|
|
c = new_lib._connection()
|
|
c.execute("select * from another")
|
|
c.close()
|
|
except sqlite3.OperationalError:
|
|
self.fail("select failed")
|
|
|
|
|
|
class TransactionTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.db = DatabaseFixture1(":memory:")
|
|
|
|
def tearDown(self):
|
|
self.db._connection().close()
|
|
|
|
def test_mutate_increase_revision(self):
|
|
old_rev = self.db.revision
|
|
with self.db.transaction() as tx:
|
|
tx.mutate(
|
|
"INSERT INTO {} "
|
|
"(field_one) "
|
|
"VALUES (?);".format(ModelFixture1._table),
|
|
(111,),
|
|
)
|
|
self.assertGreater(self.db.revision, old_rev)
|
|
|
|
def test_query_no_increase_revision(self):
|
|
old_rev = self.db.revision
|
|
with self.db.transaction() as tx:
|
|
tx.query("PRAGMA table_info(%s)" % ModelFixture1._table)
|
|
self.assertEqual(self.db.revision, old_rev)
|
|
|
|
|
|
class ModelTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.db = DatabaseFixture1(":memory:")
|
|
|
|
def tearDown(self):
|
|
self.db._connection().close()
|
|
|
|
def test_add_model(self):
|
|
model = ModelFixture1()
|
|
model.add(self.db)
|
|
rows = self.db._connection().execute("select * from test").fetchall()
|
|
self.assertEqual(len(rows), 1)
|
|
|
|
def test_store_fixed_field(self):
|
|
model = ModelFixture1()
|
|
model.add(self.db)
|
|
model.field_one = 123
|
|
model.store()
|
|
row = self.db._connection().execute("select * from test").fetchone()
|
|
self.assertEqual(row["field_one"], 123)
|
|
|
|
def test_revision(self):
|
|
old_rev = self.db.revision
|
|
model = ModelFixture1()
|
|
model.add(self.db)
|
|
model.store()
|
|
self.assertEqual(model._revision, self.db.revision)
|
|
self.assertGreater(self.db.revision, old_rev)
|
|
|
|
mid_rev = self.db.revision
|
|
model2 = ModelFixture1()
|
|
model2.add(self.db)
|
|
model2.store()
|
|
self.assertGreater(model2._revision, mid_rev)
|
|
self.assertGreater(self.db.revision, model._revision)
|
|
|
|
# revision changed, so the model should be re-loaded
|
|
model.load()
|
|
self.assertEqual(model._revision, self.db.revision)
|
|
|
|
# revision did not change, so no reload
|
|
mod2_old_rev = model2._revision
|
|
model2.load()
|
|
self.assertEqual(model2._revision, mod2_old_rev)
|
|
|
|
def test_retrieve_by_id(self):
|
|
model = ModelFixture1()
|
|
model.add(self.db)
|
|
other_model = self.db._get(ModelFixture1, model.id)
|
|
self.assertEqual(model.id, other_model.id)
|
|
|
|
def test_store_and_retrieve_flexattr(self):
|
|
model = ModelFixture1()
|
|
model.add(self.db)
|
|
model.foo = "bar"
|
|
model.store()
|
|
|
|
other_model = self.db._get(ModelFixture1, model.id)
|
|
self.assertEqual(other_model.foo, "bar")
|
|
|
|
def test_delete_flexattr(self):
|
|
model = ModelFixture1()
|
|
model["foo"] = "bar"
|
|
self.assertIn("foo", model)
|
|
del model["foo"]
|
|
self.assertNotIn("foo", model)
|
|
|
|
def test_delete_flexattr_via_dot(self):
|
|
model = ModelFixture1()
|
|
model["foo"] = "bar"
|
|
self.assertIn("foo", model)
|
|
del model.foo
|
|
self.assertNotIn("foo", model)
|
|
|
|
def test_delete_flexattr_persists(self):
|
|
model = ModelFixture1()
|
|
model.add(self.db)
|
|
model.foo = "bar"
|
|
model.store()
|
|
|
|
model = self.db._get(ModelFixture1, model.id)
|
|
del model["foo"]
|
|
model.store()
|
|
|
|
model = self.db._get(ModelFixture1, model.id)
|
|
self.assertNotIn("foo", model)
|
|
|
|
def test_delete_non_existent_attribute(self):
|
|
model = ModelFixture1()
|
|
with self.assertRaises(KeyError):
|
|
del model["foo"]
|
|
|
|
def test_delete_fixed_attribute(self):
|
|
model = ModelFixture5()
|
|
model.some_string_field = "foo"
|
|
model.some_float_field = 1.23
|
|
model.some_boolean_field = True
|
|
|
|
for field, type_ in model._fields.items():
|
|
self.assertNotEqual(model[field], type_.null)
|
|
|
|
for field, type_ in model._fields.items():
|
|
del model[field]
|
|
self.assertEqual(model[field], type_.null)
|
|
|
|
def test_null_value_normalization_by_type(self):
|
|
model = ModelFixture1()
|
|
model.field_one = None
|
|
self.assertEqual(model.field_one, 0)
|
|
|
|
def test_null_value_stays_none_for_untyped_field(self):
|
|
model = ModelFixture1()
|
|
model.foo = None
|
|
self.assertIsNone(model.foo)
|
|
|
|
def test_normalization_for_typed_flex_fields(self):
|
|
model = ModelFixture1()
|
|
model.some_float_field = None
|
|
self.assertEqual(model.some_float_field, 0.0)
|
|
|
|
def test_load_deleted_flex_field(self):
|
|
model1 = ModelFixture1()
|
|
model1["flex_field"] = True
|
|
model1.add(self.db)
|
|
|
|
model2 = self.db._get(ModelFixture1, model1.id)
|
|
self.assertIn("flex_field", model2)
|
|
|
|
del model1["flex_field"]
|
|
model1.store()
|
|
|
|
model2.load()
|
|
self.assertNotIn("flex_field", model2)
|
|
|
|
def test_check_db_fails(self):
|
|
with self.assertRaisesRegex(ValueError, "no database"):
|
|
dbcore.Model()._check_db()
|
|
with self.assertRaisesRegex(ValueError, "no id"):
|
|
ModelFixture1(self.db)._check_db()
|
|
|
|
dbcore.Model(self.db)._check_db(need_id=False)
|
|
|
|
def test_missing_field(self):
|
|
with self.assertRaises(AttributeError):
|
|
ModelFixture1(self.db).nonExistingKey
|
|
|
|
def test_computed_field(self):
|
|
model = ModelFixtureWithGetters()
|
|
self.assertEqual(model.aComputedField, "thing")
|
|
with self.assertRaisesRegex(KeyError, "computed field .+ deleted"):
|
|
del model.aComputedField
|
|
|
|
def test_items(self):
|
|
model = ModelFixture1(self.db)
|
|
model.id = 5
|
|
self.assertEqual(
|
|
{("id", 5), ("field_one", 0), ("field_two", "")}, set(model.items())
|
|
)
|
|
|
|
def test_delete_internal_field(self):
|
|
model = dbcore.Model()
|
|
del model._db
|
|
with self.assertRaises(AttributeError):
|
|
model._db
|
|
|
|
def test_parse_nonstring(self):
|
|
with self.assertRaisesRegex(TypeError, "must be a string"):
|
|
dbcore.Model._parse(None, 42)
|
|
|
|
|
|
class FormatTest(unittest.TestCase):
|
|
def test_format_fixed_field_integer(self):
|
|
model = ModelFixture1()
|
|
model.field_one = 155
|
|
value = model.formatted().get("field_one")
|
|
self.assertEqual(value, "155")
|
|
|
|
def test_format_fixed_field_integer_normalized(self):
|
|
"""The normalize method of the Integer class rounds floats"""
|
|
model = ModelFixture1()
|
|
model.field_one = 142.432
|
|
value = model.formatted().get("field_one")
|
|
self.assertEqual(value, "142")
|
|
|
|
model.field_one = 142.863
|
|
value = model.formatted().get("field_one")
|
|
self.assertEqual(value, "143")
|
|
|
|
def test_format_fixed_field_string(self):
|
|
model = ModelFixture1()
|
|
model.field_two = "caf\xe9"
|
|
value = model.formatted().get("field_two")
|
|
self.assertEqual(value, "caf\xe9")
|
|
|
|
def test_format_flex_field(self):
|
|
model = ModelFixture1()
|
|
model.other_field = "caf\xe9"
|
|
value = model.formatted().get("other_field")
|
|
self.assertEqual(value, "caf\xe9")
|
|
|
|
def test_format_flex_field_bytes(self):
|
|
model = ModelFixture1()
|
|
model.other_field = "caf\xe9".encode()
|
|
value = model.formatted().get("other_field")
|
|
self.assertTrue(isinstance(value, str))
|
|
self.assertEqual(value, "caf\xe9")
|
|
|
|
def test_format_unset_field(self):
|
|
model = ModelFixture1()
|
|
value = model.formatted().get("other_field")
|
|
self.assertEqual(value, "")
|
|
|
|
def test_format_typed_flex_field(self):
|
|
model = ModelFixture1()
|
|
model.some_float_field = 3.14159265358979
|
|
value = model.formatted().get("some_float_field")
|
|
self.assertEqual(value, "3.1")
|
|
|
|
|
|
class FormattedMappingTest(unittest.TestCase):
|
|
def test_keys_equal_model_keys(self):
|
|
model = ModelFixture1()
|
|
formatted = model.formatted()
|
|
self.assertEqual(set(model.keys(True)), set(formatted.keys()))
|
|
|
|
def test_get_unset_field(self):
|
|
model = ModelFixture1()
|
|
formatted = model.formatted()
|
|
with self.assertRaises(KeyError):
|
|
formatted["other_field"]
|
|
|
|
def test_get_method_with_default(self):
|
|
model = ModelFixture1()
|
|
formatted = model.formatted()
|
|
self.assertEqual(formatted.get("other_field"), "")
|
|
|
|
def test_get_method_with_specified_default(self):
|
|
model = ModelFixture1()
|
|
formatted = model.formatted()
|
|
self.assertEqual(formatted.get("other_field", "default"), "default")
|
|
|
|
|
|
class ParseTest(unittest.TestCase):
|
|
def test_parse_fixed_field(self):
|
|
value = ModelFixture1._parse("field_one", "2")
|
|
self.assertIsInstance(value, int)
|
|
self.assertEqual(value, 2)
|
|
|
|
def test_parse_flex_field(self):
|
|
value = ModelFixture1._parse("some_float_field", "2")
|
|
self.assertIsInstance(value, float)
|
|
self.assertEqual(value, 2.0)
|
|
|
|
def test_parse_untyped_field(self):
|
|
value = ModelFixture1._parse("field_nine", "2")
|
|
self.assertEqual(value, "2")
|
|
|
|
|
|
class QueryParseTest(unittest.TestCase):
|
|
def pqp(self, part):
|
|
return dbcore.queryparse.parse_query_part(
|
|
part,
|
|
{"year": dbcore.query.NumericQuery},
|
|
{":": dbcore.query.RegexpQuery},
|
|
)[
|
|
:-1
|
|
] # remove the negate flag
|
|
|
|
def test_one_basic_term(self):
|
|
q = "test"
|
|
r = (None, "test", dbcore.query.SubstringQuery)
|
|
self.assertEqual(self.pqp(q), r)
|
|
|
|
def test_one_keyed_term(self):
|
|
q = "test:val"
|
|
r = ("test", "val", dbcore.query.SubstringQuery)
|
|
self.assertEqual(self.pqp(q), r)
|
|
|
|
def test_colon_at_end(self):
|
|
q = "test:"
|
|
r = ("test", "", dbcore.query.SubstringQuery)
|
|
self.assertEqual(self.pqp(q), r)
|
|
|
|
def test_one_basic_regexp(self):
|
|
q = r":regexp"
|
|
r = (None, "regexp", dbcore.query.RegexpQuery)
|
|
self.assertEqual(self.pqp(q), r)
|
|
|
|
def test_keyed_regexp(self):
|
|
q = r"test::regexp"
|
|
r = ("test", "regexp", dbcore.query.RegexpQuery)
|
|
self.assertEqual(self.pqp(q), r)
|
|
|
|
def test_escaped_colon(self):
|
|
q = r"test\:val"
|
|
r = (None, "test:val", dbcore.query.SubstringQuery)
|
|
self.assertEqual(self.pqp(q), r)
|
|
|
|
def test_escaped_colon_in_regexp(self):
|
|
q = r":test\:regexp"
|
|
r = (None, "test:regexp", dbcore.query.RegexpQuery)
|
|
self.assertEqual(self.pqp(q), r)
|
|
|
|
def test_single_year(self):
|
|
q = "year:1999"
|
|
r = ("year", "1999", dbcore.query.NumericQuery)
|
|
self.assertEqual(self.pqp(q), r)
|
|
|
|
def test_multiple_years(self):
|
|
q = "year:1999..2010"
|
|
r = ("year", "1999..2010", dbcore.query.NumericQuery)
|
|
self.assertEqual(self.pqp(q), r)
|
|
|
|
def test_empty_query_part(self):
|
|
q = ""
|
|
r = (None, "", dbcore.query.SubstringQuery)
|
|
self.assertEqual(self.pqp(q), r)
|
|
|
|
|
|
class QueryFromStringsTest(unittest.TestCase):
|
|
def qfs(self, strings):
|
|
return dbcore.queryparse.query_from_strings(
|
|
dbcore.query.AndQuery,
|
|
ModelFixture1,
|
|
{":": dbcore.query.RegexpQuery},
|
|
strings,
|
|
)
|
|
|
|
def test_zero_parts(self):
|
|
q = self.qfs([])
|
|
self.assertIsInstance(q, dbcore.query.AndQuery)
|
|
self.assertEqual(len(q.subqueries), 1)
|
|
self.assertIsInstance(q.subqueries[0], dbcore.query.TrueQuery)
|
|
|
|
def test_two_parts(self):
|
|
q = self.qfs(["foo", "bar:baz"])
|
|
self.assertIsInstance(q, dbcore.query.AndQuery)
|
|
self.assertEqual(len(q.subqueries), 2)
|
|
self.assertIsInstance(q.subqueries[0], dbcore.query.AnyFieldQuery)
|
|
self.assertIsInstance(q.subqueries[1], dbcore.query.SubstringQuery)
|
|
|
|
def test_parse_fixed_type_query(self):
|
|
q = self.qfs(["field_one:2..3"])
|
|
self.assertIsInstance(q.subqueries[0], dbcore.query.NumericQuery)
|
|
|
|
def test_parse_flex_type_query(self):
|
|
q = self.qfs(["some_float_field:2..3"])
|
|
self.assertIsInstance(q.subqueries[0], dbcore.query.NumericQuery)
|
|
|
|
def test_empty_query_part(self):
|
|
q = self.qfs([""])
|
|
self.assertIsInstance(q.subqueries[0], dbcore.query.TrueQuery)
|
|
|
|
|
|
class SortFromStringsTest(unittest.TestCase):
|
|
def sfs(self, strings):
|
|
return dbcore.queryparse.sort_from_strings(
|
|
ModelFixture1,
|
|
strings,
|
|
)
|
|
|
|
def test_zero_parts(self):
|
|
s = self.sfs([])
|
|
self.assertIsInstance(s, dbcore.query.NullSort)
|
|
self.assertEqual(s, dbcore.query.NullSort())
|
|
|
|
def test_one_parts(self):
|
|
s = self.sfs(["field+"])
|
|
self.assertIsInstance(s, dbcore.query.Sort)
|
|
|
|
def test_two_parts(self):
|
|
s = self.sfs(["field+", "another_field-"])
|
|
self.assertIsInstance(s, dbcore.query.MultipleSort)
|
|
self.assertEqual(len(s.sorts), 2)
|
|
|
|
def test_fixed_field_sort(self):
|
|
s = self.sfs(["field_one+"])
|
|
self.assertIsInstance(s, dbcore.query.FixedFieldSort)
|
|
self.assertEqual(s, dbcore.query.FixedFieldSort("field_one"))
|
|
|
|
def test_flex_field_sort(self):
|
|
s = self.sfs(["flex_field+"])
|
|
self.assertIsInstance(s, dbcore.query.SlowFieldSort)
|
|
self.assertEqual(s, dbcore.query.SlowFieldSort("flex_field"))
|
|
|
|
def test_special_sort(self):
|
|
s = self.sfs(["some_sort+"])
|
|
self.assertIsInstance(s, SortFixture)
|
|
|
|
|
|
class ParseSortedQueryTest(unittest.TestCase):
|
|
def psq(self, parts):
|
|
return dbcore.parse_sorted_query(
|
|
ModelFixture1,
|
|
parts.split(),
|
|
)
|
|
|
|
def test_and_query(self):
|
|
q, s = self.psq("foo bar")
|
|
self.assertIsInstance(q, dbcore.query.AndQuery)
|
|
self.assertIsInstance(s, dbcore.query.NullSort)
|
|
self.assertEqual(len(q.subqueries), 2)
|
|
|
|
def test_or_query(self):
|
|
q, s = self.psq("foo , bar")
|
|
self.assertIsInstance(q, dbcore.query.OrQuery)
|
|
self.assertIsInstance(s, dbcore.query.NullSort)
|
|
self.assertEqual(len(q.subqueries), 2)
|
|
|
|
def test_no_space_before_comma_or_query(self):
|
|
q, s = self.psq("foo, bar")
|
|
self.assertIsInstance(q, dbcore.query.OrQuery)
|
|
self.assertIsInstance(s, dbcore.query.NullSort)
|
|
self.assertEqual(len(q.subqueries), 2)
|
|
|
|
def test_no_spaces_or_query(self):
|
|
q, s = self.psq("foo,bar")
|
|
self.assertIsInstance(q, dbcore.query.AndQuery)
|
|
self.assertIsInstance(s, dbcore.query.NullSort)
|
|
self.assertEqual(len(q.subqueries), 1)
|
|
|
|
def test_trailing_comma_or_query(self):
|
|
q, s = self.psq("foo , bar ,")
|
|
self.assertIsInstance(q, dbcore.query.OrQuery)
|
|
self.assertIsInstance(s, dbcore.query.NullSort)
|
|
self.assertEqual(len(q.subqueries), 3)
|
|
|
|
def test_leading_comma_or_query(self):
|
|
q, s = self.psq(", foo , bar")
|
|
self.assertIsInstance(q, dbcore.query.OrQuery)
|
|
self.assertIsInstance(s, dbcore.query.NullSort)
|
|
self.assertEqual(len(q.subqueries), 3)
|
|
|
|
def test_only_direction(self):
|
|
q, s = self.psq("-")
|
|
self.assertIsInstance(q, dbcore.query.AndQuery)
|
|
self.assertIsInstance(s, dbcore.query.NullSort)
|
|
self.assertEqual(len(q.subqueries), 1)
|
|
|
|
|
|
class ResultsIteratorTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.db = DatabaseFixture1(":memory:")
|
|
model = ModelFixture1()
|
|
model["foo"] = "baz"
|
|
model.add(self.db)
|
|
model = ModelFixture1()
|
|
model["foo"] = "bar"
|
|
model.add(self.db)
|
|
|
|
def tearDown(self):
|
|
self.db._connection().close()
|
|
|
|
def test_iterate_once(self):
|
|
objs = self.db._fetch(ModelFixture1)
|
|
self.assertEqual(len(list(objs)), 2)
|
|
|
|
def test_iterate_twice(self):
|
|
objs = self.db._fetch(ModelFixture1)
|
|
list(objs)
|
|
self.assertEqual(len(list(objs)), 2)
|
|
|
|
def test_concurrent_iterators(self):
|
|
results = self.db._fetch(ModelFixture1)
|
|
it1 = iter(results)
|
|
it2 = iter(results)
|
|
next(it1)
|
|
list(it2)
|
|
self.assertEqual(len(list(it1)), 1)
|
|
|
|
def test_slow_query(self):
|
|
q = dbcore.query.SubstringQuery("foo", "ba", False)
|
|
objs = self.db._fetch(ModelFixture1, q)
|
|
self.assertEqual(len(list(objs)), 2)
|
|
|
|
def test_slow_query_negative(self):
|
|
q = dbcore.query.SubstringQuery("foo", "qux", False)
|
|
objs = self.db._fetch(ModelFixture1, q)
|
|
self.assertEqual(len(list(objs)), 0)
|
|
|
|
def test_iterate_slow_sort(self):
|
|
s = dbcore.query.SlowFieldSort("foo")
|
|
res = self.db._fetch(ModelFixture1, sort=s)
|
|
objs = list(res)
|
|
self.assertEqual(objs[0].foo, "bar")
|
|
self.assertEqual(objs[1].foo, "baz")
|
|
|
|
def test_unsorted_subscript(self):
|
|
objs = self.db._fetch(ModelFixture1)
|
|
self.assertEqual(objs[0].foo, "baz")
|
|
self.assertEqual(objs[1].foo, "bar")
|
|
|
|
def test_slow_sort_subscript(self):
|
|
s = dbcore.query.SlowFieldSort("foo")
|
|
objs = self.db._fetch(ModelFixture1, sort=s)
|
|
self.assertEqual(objs[0].foo, "bar")
|
|
self.assertEqual(objs[1].foo, "baz")
|
|
|
|
def test_length(self):
|
|
objs = self.db._fetch(ModelFixture1)
|
|
self.assertEqual(len(objs), 2)
|
|
|
|
def test_out_of_range(self):
|
|
objs = self.db._fetch(ModelFixture1)
|
|
with self.assertRaises(IndexError):
|
|
objs[100]
|
|
|
|
def test_no_results(self):
|
|
self.assertIsNone(
|
|
self.db._fetch(ModelFixture1, dbcore.query.FalseQuery()).get()
|
|
)
|
|
|
|
|
|
def suite():
|
|
return unittest.TestLoader().loadTestsFromName(__name__)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main(defaultTest="suite")
|