Add support for BETWEEN and add tests
Add tests for _apply_criteration applying the correct operators Change-Id: I54f27193e4a8151f28bfcd5322e58257f6ea8091
This commit is contained in:
parent
ca19abf311
commit
55ecd4cb15
@ -22,7 +22,7 @@ from oslo_db.sqlalchemy import utils as oslodb_utils
|
|||||||
from oslo_db import exception as oslo_db_exception
|
from oslo_db import exception as oslo_db_exception
|
||||||
from oslo_log import log as logging
|
from oslo_log import log as logging
|
||||||
from oslo_utils import timeutils
|
from oslo_utils import timeutils
|
||||||
from sqlalchemy import select, or_
|
from sqlalchemy import select, or_, between
|
||||||
|
|
||||||
from designate import exceptions
|
from designate import exceptions
|
||||||
from designate.sqlalchemy import session
|
from designate.sqlalchemy import session
|
||||||
@ -97,7 +97,8 @@ class SQLAlchemy(object):
|
|||||||
def rollback(self):
|
def rollback(self):
|
||||||
self.session.rollback()
|
self.session.rollback()
|
||||||
|
|
||||||
def _apply_criterion(self, table, query, criterion):
|
@staticmethod
|
||||||
|
def _apply_criterion(table, query, criterion):
|
||||||
if criterion is not None:
|
if criterion is not None:
|
||||||
for name, value in criterion.items():
|
for name, value in criterion.items():
|
||||||
column = getattr(table.c, name)
|
column = getattr(table.c, name)
|
||||||
@ -131,6 +132,13 @@ class SQLAlchemy(object):
|
|||||||
queryval = value[1:]
|
queryval = value[1:]
|
||||||
query = query.where(column > queryval)
|
query = query.where(column > queryval)
|
||||||
|
|
||||||
|
elif (isinstance(value, six.string_types) and
|
||||||
|
value.startswith('BETWEEN')):
|
||||||
|
elements = [i.strip(" ") for i in
|
||||||
|
value.split(" ", 1)[1].strip(" ").split(",")]
|
||||||
|
query = query.where(between(
|
||||||
|
column, elements[0], elements[1]))
|
||||||
|
|
||||||
elif isinstance(value, list):
|
elif isinstance(value, list):
|
||||||
query = query.where(column.in_(value))
|
query = query.where(column.in_(value))
|
||||||
|
|
||||||
|
122
designate/tests/test_sqlalchemy.py
Normal file
122
designate/tests/test_sqlalchemy.py
Normal file
@ -0,0 +1,122 @@
|
|||||||
|
# Copyright 2015 Hewlett-Packard Development Company, L.P.
|
||||||
|
#
|
||||||
|
# Author: Endre Karlson <endre.karlson@hp.com>
|
||||||
|
#
|
||||||
|
# Licensed under the Apache License, Version 2.0 (the "License"); you may
|
||||||
|
# not use this file except in compliance with the License. You may obtain
|
||||||
|
# a copy of the License at
|
||||||
|
#
|
||||||
|
# http://www.apache.org/licenses/LICENSE-2.0
|
||||||
|
#
|
||||||
|
# Unless required by applicable law or agreed to in writing, software
|
||||||
|
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
|
||||||
|
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
|
||||||
|
# License for the specific language governing permissions and limitations
|
||||||
|
# under the License.
|
||||||
|
import operator
|
||||||
|
|
||||||
|
import mock
|
||||||
|
import sqlalchemy as sa
|
||||||
|
from sqlalchemy.sql import operators
|
||||||
|
|
||||||
|
from designate.sqlalchemy import base
|
||||||
|
from designate.tests import TestCase
|
||||||
|
|
||||||
|
metadata = sa.MetaData()
|
||||||
|
|
||||||
|
dummy_table = sa.Table('dummy', metadata,
|
||||||
|
sa.Column('id', sa.String(36)),
|
||||||
|
sa.Column('a', sa.String()),
|
||||||
|
sa.Column('int', sa.Integer()),
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
class SQLAlchemyTestCase(TestCase):
|
||||||
|
def setUp(self):
|
||||||
|
super(SQLAlchemyTestCase, self).setUp()
|
||||||
|
self.query = mock.Mock()
|
||||||
|
|
||||||
|
def test_wildcard(self):
|
||||||
|
criterion = {"a": "%foo%"}
|
||||||
|
|
||||||
|
op = dummy_table.c.a.like("%foo")
|
||||||
|
with mock.patch.object(dummy_table.c.a, 'operate') as func:
|
||||||
|
func.return_value = op
|
||||||
|
|
||||||
|
base.SQLAlchemy._apply_criterion(
|
||||||
|
dummy_table, self.query, criterion)
|
||||||
|
func.assert_called_with(operators.like_op, "%foo%", escape=None)
|
||||||
|
self.query.where.assert_called_with(op)
|
||||||
|
|
||||||
|
def test_ne(self):
|
||||||
|
criterion = {"a": "!foo"}
|
||||||
|
|
||||||
|
op = dummy_table.c.a != "foo"
|
||||||
|
with mock.patch.object(dummy_table.c.a, 'operate') as func:
|
||||||
|
func.return_value = op
|
||||||
|
|
||||||
|
base.SQLAlchemy._apply_criterion(
|
||||||
|
dummy_table, self.query, criterion)
|
||||||
|
func.assert_called_with(operator.ne, "foo")
|
||||||
|
self.query.where.assert_called_with(op)
|
||||||
|
|
||||||
|
def test_le(self):
|
||||||
|
criterion = {"a": "<=foo"}
|
||||||
|
|
||||||
|
op = dummy_table.c.a <= "foo"
|
||||||
|
with mock.patch.object(dummy_table.c.a, 'operate') as func:
|
||||||
|
func.return_value = op
|
||||||
|
|
||||||
|
base.SQLAlchemy._apply_criterion(
|
||||||
|
dummy_table, self.query, criterion)
|
||||||
|
func.assert_called_with(operator.le, "foo")
|
||||||
|
self.query.where.assert_called_with(op)
|
||||||
|
|
||||||
|
def test_lt(self):
|
||||||
|
criterion = {"a": "<foo"}
|
||||||
|
|
||||||
|
op = dummy_table.c.a < "foo"
|
||||||
|
with mock.patch.object(dummy_table.c.a, 'operate') as func:
|
||||||
|
func.return_value = op
|
||||||
|
|
||||||
|
base.SQLAlchemy._apply_criterion(
|
||||||
|
dummy_table, self.query, criterion)
|
||||||
|
func.assert_called_with(operator.lt, "foo")
|
||||||
|
self.query.where.assert_called_with(op)
|
||||||
|
|
||||||
|
def test_ge(self):
|
||||||
|
criterion = {"a": ">=foo"}
|
||||||
|
|
||||||
|
op = dummy_table.c.a >= "foo"
|
||||||
|
with mock.patch.object(dummy_table.c.a, 'operate') as func:
|
||||||
|
func.return_value = op
|
||||||
|
|
||||||
|
base.SQLAlchemy._apply_criterion(
|
||||||
|
dummy_table, self.query, criterion)
|
||||||
|
func.assert_called_with(operator.ge, "foo")
|
||||||
|
self.query.where.assert_called_with(op)
|
||||||
|
|
||||||
|
def test_gt(self):
|
||||||
|
criterion = {"a": ">foo"}
|
||||||
|
|
||||||
|
op = dummy_table.c.a > "foo"
|
||||||
|
with mock.patch.object(dummy_table.c.a, 'operate') as func:
|
||||||
|
func.return_value = op
|
||||||
|
|
||||||
|
base.SQLAlchemy._apply_criterion(
|
||||||
|
dummy_table, self.query, criterion)
|
||||||
|
func.assert_called_with(operator.gt, "foo")
|
||||||
|
self.query.where.assert_called_with(op)
|
||||||
|
|
||||||
|
def test_between(self):
|
||||||
|
criterion = {"a": "BETWEEN 1,3"}
|
||||||
|
|
||||||
|
op = dummy_table.c.a.between(1, 3)
|
||||||
|
with mock.patch.object(dummy_table.c.a, 'operate') as func:
|
||||||
|
func.return_value = op
|
||||||
|
|
||||||
|
base.SQLAlchemy._apply_criterion(
|
||||||
|
dummy_table, self.query, criterion)
|
||||||
|
func.assert_called_with(operators.between_op, '1', '3',
|
||||||
|
symmetric=False)
|
||||||
|
self.query.where.assert_called_with(op)
|
Loading…
Reference in New Issue
Block a user