From 55ecd4cb15433821c9fec7842cfb273dd743c1c2 Mon Sep 17 00:00:00 2001 From: Endre Karlson Date: Tue, 30 Jun 2015 11:20:16 +0200 Subject: [PATCH] Add support for BETWEEN and add tests Add tests for _apply_criteration applying the correct operators Change-Id: I54f27193e4a8151f28bfcd5322e58257f6ea8091 --- designate/sqlalchemy/base.py | 12 ++- designate/tests/test_sqlalchemy.py | 122 +++++++++++++++++++++++++++++ 2 files changed, 132 insertions(+), 2 deletions(-) create mode 100644 designate/tests/test_sqlalchemy.py diff --git a/designate/sqlalchemy/base.py b/designate/sqlalchemy/base.py index 820d985f..e2971bbb 100644 --- a/designate/sqlalchemy/base.py +++ b/designate/sqlalchemy/base.py @@ -22,7 +22,7 @@ from oslo_db.sqlalchemy import utils as oslodb_utils from oslo_db import exception as oslo_db_exception from oslo_log import log as logging from oslo_utils import timeutils -from sqlalchemy import select, or_ +from sqlalchemy import select, or_, between from designate import exceptions from designate.sqlalchemy import session @@ -97,7 +97,8 @@ class SQLAlchemy(object): def rollback(self): self.session.rollback() - def _apply_criterion(self, table, query, criterion): + @staticmethod + def _apply_criterion(table, query, criterion): if criterion is not None: for name, value in criterion.items(): column = getattr(table.c, name) @@ -131,6 +132,13 @@ class SQLAlchemy(object): queryval = value[1:] 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): query = query.where(column.in_(value)) diff --git a/designate/tests/test_sqlalchemy.py b/designate/tests/test_sqlalchemy.py new file mode 100644 index 00000000..13941d3c --- /dev/null +++ b/designate/tests/test_sqlalchemy.py @@ -0,0 +1,122 @@ +# Copyright 2015 Hewlett-Packard Development Company, L.P. +# +# Author: Endre Karlson +# +# 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.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)