1702 lines
62 KiB
Python
1702 lines
62 KiB
Python
# Copyright (c) 2013 Boris Pavlovic (boris@pavlovic.me).
|
|
# All Rights Reserved.
|
|
#
|
|
# 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.
|
|
|
|
from unittest import mock
|
|
from urllib import parse
|
|
|
|
import fixtures
|
|
import sqlalchemy
|
|
from sqlalchemy.dialects import mysql
|
|
from sqlalchemy import Boolean, Index, Integer, DateTime, String
|
|
from sqlalchemy import CheckConstraint
|
|
from sqlalchemy import MetaData, Table, Column
|
|
from sqlalchemy import ForeignKey, ForeignKeyConstraint
|
|
from sqlalchemy.dialects.postgresql import psycopg2
|
|
from sqlalchemy.engine import url as sa_url
|
|
from sqlalchemy.exc import OperationalError
|
|
from sqlalchemy.ext.hybrid import hybrid_property
|
|
from sqlalchemy.orm import declarative_base
|
|
from sqlalchemy.orm import mapper
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy import PrimaryKeyConstraint
|
|
from sqlalchemy import sql
|
|
from sqlalchemy.sql.expression import cast
|
|
from sqlalchemy.sql import select
|
|
from sqlalchemy.types import UserDefinedType
|
|
|
|
from oslo_db import exception
|
|
from oslo_db.sqlalchemy import models
|
|
from oslo_db.sqlalchemy import provision
|
|
from oslo_db.sqlalchemy import session
|
|
from oslo_db.sqlalchemy import utils
|
|
from oslo_db.tests import base as test_base
|
|
from oslo_db.tests.sqlalchemy import base as db_test_base
|
|
|
|
Base = declarative_base()
|
|
|
|
|
|
class TestSanitizeDbUrl(test_base.BaseTestCase):
|
|
|
|
def test_url_with_cred(self):
|
|
db_url = 'myproto://johndoe:secret@localhost/myschema'
|
|
expected = 'myproto://****:****@localhost/myschema'
|
|
actual = utils.sanitize_db_url(db_url)
|
|
self.assertEqual(expected, actual)
|
|
|
|
def test_url_with_no_cred(self):
|
|
db_url = 'sqlite:///mysqlitefile'
|
|
actual = utils.sanitize_db_url(db_url)
|
|
self.assertEqual(db_url, actual)
|
|
|
|
|
|
class CustomType(UserDefinedType):
|
|
"""Dummy column type for testing unsupported types."""
|
|
def get_col_spec(self):
|
|
return "CustomType"
|
|
|
|
|
|
class FakeTable(Base):
|
|
__tablename__ = 'fake_table'
|
|
|
|
user_id = Column(String(50), primary_key=True)
|
|
project_id = Column(String(50))
|
|
snapshot_id = Column(String(50))
|
|
updated_at = Column(DateTime, nullable=True)
|
|
enabled = Column(Boolean, default=True)
|
|
|
|
_expr_to_appease_mox = project_id + snapshot_id
|
|
|
|
@hybrid_property
|
|
def some_hybrid(self):
|
|
raise NotImplementedError()
|
|
|
|
@some_hybrid.expression
|
|
def some_hybrid(cls):
|
|
return cls._expr_to_appease_mox
|
|
|
|
def foo(self):
|
|
pass
|
|
|
|
|
|
class FakeTableJoinedInh(FakeTable):
|
|
__tablename__ = 'fake_table_inh'
|
|
|
|
id = Column(String(50), ForeignKey('fake_table.user_id'))
|
|
|
|
|
|
class FakeTableSingleInh(FakeTable):
|
|
__mapper_args__ = {'polymorphic_identity': 'foo'}
|
|
|
|
|
|
class FakeTableWithMultipleKeys(Base):
|
|
__tablename__ = 'fake_table_multiple_keys'
|
|
|
|
key1 = Column(String(50), primary_key=True)
|
|
key2 = Column(String(50), primary_key=True)
|
|
key3 = Column(String(50))
|
|
|
|
|
|
class FakeTableWithIndexes(Base):
|
|
__tablename__ = 'fake_table_unique_index'
|
|
|
|
id = Column(String(50), primary_key=True)
|
|
key1 = Column(String(50))
|
|
key2 = Column(String(50))
|
|
key3 = Column(String(50))
|
|
|
|
__table_args__ = (
|
|
Index('idx_unique', 'key1', 'key2', unique=True),
|
|
Index('idx_unique', 'key1', 'key3', unique=False),
|
|
)
|
|
|
|
|
|
class FakeTableClassicalyMapped(object):
|
|
pass
|
|
|
|
|
|
fake_table = Table(
|
|
'fake_table_classically_mapped',
|
|
Base.metadata,
|
|
Column('id', Integer, primary_key=True),
|
|
Column('key', String(50))
|
|
)
|
|
|
|
mapper(FakeTableClassicalyMapped, fake_table)
|
|
|
|
|
|
class FakeModel(object):
|
|
def __init__(self, values):
|
|
self.values = values
|
|
|
|
def __getattr__(self, name):
|
|
try:
|
|
value = self.values[name]
|
|
except KeyError:
|
|
raise AttributeError(name)
|
|
return value
|
|
|
|
def __getitem__(self, key):
|
|
if key in self.values:
|
|
return self.values[key]
|
|
else:
|
|
raise NotImplementedError()
|
|
|
|
def __repr__(self):
|
|
return '<FakeModel: %s>' % self.values
|
|
|
|
|
|
class TestPaginateQuery(test_base.BaseTestCase):
|
|
def setUp(self):
|
|
super(TestPaginateQuery, self).setUp()
|
|
|
|
self.query = mock.Mock()
|
|
self.mock_asc = self.useFixture(
|
|
fixtures.MockPatchObject(sqlalchemy, 'asc')).mock
|
|
self.mock_desc = self.useFixture(
|
|
fixtures.MockPatchObject(sqlalchemy, 'desc')).mock
|
|
|
|
self.marker = FakeTable(user_id='user',
|
|
project_id='p',
|
|
snapshot_id='s',
|
|
updated_at=None)
|
|
self.model = FakeTable
|
|
|
|
def test_paginate_query_no_pagination_no_sort_dirs(self):
|
|
self.query.order_by.return_value = self.query
|
|
self.mock_asc.side_effect = [
|
|
'asc_3', 'asc_2', 'asc_1'
|
|
]
|
|
|
|
utils.paginate_query(self.query, self.model, 5,
|
|
['user_id', 'project_id', 'snapshot_id'])
|
|
|
|
self.mock_asc.assert_has_calls([
|
|
mock.call(self.model.user_id),
|
|
mock.call(self.model.project_id),
|
|
mock.call(self.model.snapshot_id),
|
|
])
|
|
self.query.order_by.assert_has_calls([
|
|
mock.call('asc_3'),
|
|
mock.call('asc_2'),
|
|
mock.call('asc_1'),
|
|
])
|
|
self.query.limit.assert_called_once_with(5)
|
|
|
|
def test_paginate_query_no_pagination(self):
|
|
self.query.order_by.return_value = self.query
|
|
self.mock_asc.side_effect = ['asc']
|
|
self.mock_desc.side_effect = ['desc']
|
|
|
|
utils.paginate_query(self.query, self.model, 5,
|
|
['user_id', 'project_id'],
|
|
sort_dirs=['asc', 'desc'])
|
|
|
|
self.mock_asc.assert_called_once_with(self.model.user_id)
|
|
self.mock_desc.assert_called_once_with(self.model.project_id)
|
|
self.query.order_by.assert_has_calls([
|
|
mock.call('asc'),
|
|
mock.call('desc'),
|
|
])
|
|
self.query.limit.assert_called_once_with(5)
|
|
|
|
def test_invalid_sort_key_str(self):
|
|
self.assertEqual("Sort key supplied is invalid: None",
|
|
str(exception.InvalidSortKey()))
|
|
self.assertEqual("Sort key supplied is invalid: lol",
|
|
str(exception.InvalidSortKey("lol")))
|
|
|
|
def test_invalid_unicode_paramater_str(self):
|
|
self.assertEqual(
|
|
"Invalid Parameter: Encoding directive wasn't provided.",
|
|
str(exception.DBInvalidUnicodeParameter()))
|
|
|
|
def test_paginate_query_attribute_error(self):
|
|
self.mock_asc.return_value = 'asc'
|
|
|
|
self.assertRaises(exception.InvalidSortKey,
|
|
utils.paginate_query, self.query,
|
|
self.model, 5, ['user_id', 'non-existent key'])
|
|
|
|
self.mock_asc.assert_called_once_with(self.model.user_id)
|
|
self.query.order_by.assert_called_once_with('asc')
|
|
|
|
def test_paginate_query_attribute_error_invalid_sortkey(self):
|
|
self.assertRaises(exception.InvalidSortKey,
|
|
utils.paginate_query, self.query,
|
|
self.model, 5, ['bad_user_id'])
|
|
|
|
def test_paginate_query_attribute_error_invalid_sortkey_2(self):
|
|
self.assertRaises(exception.InvalidSortKey,
|
|
utils.paginate_query, self.query,
|
|
self.model, 5, ['foo'])
|
|
|
|
def test_paginate_query_attribute_error_invalid_sortkey_3(self):
|
|
self.assertRaises(exception.InvalidSortKey,
|
|
utils.paginate_query, self.query,
|
|
self.model, 5, ['asc-nullinvalid'])
|
|
|
|
def test_paginate_query_assertion_error(self):
|
|
self.assertRaises(AssertionError,
|
|
utils.paginate_query, self.query,
|
|
self.model, 5, ['user_id'],
|
|
marker=self.marker,
|
|
sort_dir='asc', sort_dirs=['asc'])
|
|
|
|
def test_paginate_query_assertion_error_2(self):
|
|
self.assertRaises(AssertionError,
|
|
utils.paginate_query, self.query,
|
|
self.model, 5, ['user_id'],
|
|
marker=self.marker,
|
|
sort_dir=None, sort_dirs=['asc', 'desk'])
|
|
|
|
@mock.patch.object(sqlalchemy.sql, 'and_')
|
|
@mock.patch.object(sqlalchemy.sql, 'or_')
|
|
def test_paginate_query(self, mock_or, mock_and):
|
|
self.query.order_by.return_value = self.query
|
|
self.query.filter.return_value = self.query
|
|
self.mock_asc.return_value = 'asc_1'
|
|
self.mock_desc.return_value = 'desc_1'
|
|
mock_and.side_effect = ['some_crit', 'another_crit']
|
|
mock_or.return_value = 'some_f'
|
|
|
|
utils.paginate_query(self.query, self.model, 5,
|
|
['user_id', 'project_id'],
|
|
marker=self.marker,
|
|
sort_dirs=['asc', 'desc'])
|
|
|
|
self.mock_asc.assert_called_once_with(self.model.user_id)
|
|
self.mock_desc.assert_called_once_with(self.model.project_id)
|
|
self.query.order_by.assert_has_calls([
|
|
mock.call('asc_1'),
|
|
mock.call('desc_1'),
|
|
])
|
|
mock_and.assert_has_calls([
|
|
mock.call(mock.ANY),
|
|
mock.call(mock.ANY, mock.ANY)
|
|
])
|
|
mock_or.assert_called_once_with('some_crit', 'another_crit')
|
|
self.query.filter.assert_called_once_with('some_f')
|
|
self.query.limit.assert_called_once_with(5)
|
|
|
|
@mock.patch.object(sqlalchemy.sql, 'and_')
|
|
@mock.patch.object(sqlalchemy.sql, 'or_')
|
|
def test_paginate_query_null(self, mock_or, mock_and):
|
|
self.query.order_by.return_value = self.query
|
|
self.query.filter.return_value = self.query
|
|
self.mock_desc.side_effect = [
|
|
'asc_null_2',
|
|
'desc_null_2',
|
|
'desc_1',
|
|
]
|
|
self.mock_asc.side_effect = [
|
|
'asc_1'
|
|
]
|
|
mock_or.side_effect = [
|
|
'or_1',
|
|
'or_2',
|
|
'some_f',
|
|
]
|
|
mock_and.side_effect = [
|
|
'some_crit',
|
|
'another_crit',
|
|
]
|
|
|
|
with mock.patch.object(self.model.user_id, 'isnot') as mock_isnot, \
|
|
mock.patch.object(self.model.user_id, 'is_') as mock_is_a, \
|
|
mock.patch.object(self.model.project_id, 'is_') as mock_is_b:
|
|
mock_isnot.return_value = 'asc_null_1'
|
|
mock_is_a.side_effect = [
|
|
'desc_null_filter_1',
|
|
'desc_null_filter_2',
|
|
]
|
|
mock_is_b.side_effect = [
|
|
'desc_null_1',
|
|
'asc_null_filter',
|
|
]
|
|
|
|
utils.paginate_query(self.query, self.model, 5,
|
|
['user_id', 'project_id'],
|
|
marker=self.marker,
|
|
sort_dirs=[
|
|
'asc-nullslast', 'desc-nullsfirst'])
|
|
|
|
mock_isnot.assert_called_once_with(None)
|
|
mock_is_a.assert_has_calls([
|
|
mock.call(None),
|
|
mock.call(None),
|
|
])
|
|
mock_is_b.assert_has_calls([
|
|
mock.call(None),
|
|
mock.call(None),
|
|
])
|
|
|
|
self.mock_desc.assert_has_calls([
|
|
mock.call('asc_null_1'),
|
|
mock.call('desc_null_1'),
|
|
mock.call(self.model.project_id),
|
|
])
|
|
self.mock_asc.assert_has_calls([
|
|
mock.call(self.model.user_id),
|
|
])
|
|
mock_or.assert_has_calls([
|
|
mock.call(mock.ANY, 'desc_null_filter_2'),
|
|
mock.call(mock.ANY, 'asc_null_filter'),
|
|
mock.call('some_crit', 'another_crit'),
|
|
])
|
|
mock_and.assert_has_calls([
|
|
mock.call('or_1'),
|
|
mock.call(mock.ANY, 'or_2'),
|
|
])
|
|
|
|
self.query.order_by.assert_has_calls([
|
|
mock.call('asc_null_2'),
|
|
mock.call('asc_1'),
|
|
mock.call('desc_null_2'),
|
|
mock.call('desc_1'),
|
|
])
|
|
self.query.filter.assert_called_once_with('some_f')
|
|
self.query.limit.assert_called_once_with(5)
|
|
|
|
@mock.patch.object(sqlalchemy.sql, 'and_')
|
|
@mock.patch.object(sqlalchemy.sql, 'or_')
|
|
def test_paginate_query_marker_null(self, mock_or, mock_and):
|
|
|
|
self.mock_asc.side_effect = [
|
|
'asc_1'
|
|
]
|
|
self.mock_desc.side_effect = [
|
|
'asc_null_2',
|
|
'desc_null_2',
|
|
'desc_1',
|
|
]
|
|
self.query.order_by.return_value = self.query
|
|
self.query.filter.return_value = self.query
|
|
mock_and.return_value = 'some_crit'
|
|
mock_or.side_effect = ['or_1', 'some_f']
|
|
|
|
with mock.patch.object(self.model.user_id, 'isnot') as mock_isnot, \
|
|
mock.patch.object(self.model.updated_at, 'is_') as mock_is_a, \
|
|
mock.patch.object(self.model.user_id, 'is_') as mock_is_b:
|
|
|
|
mock_isnot.return_value = 'asc_null_1'
|
|
mock_is_a.return_value = 'desc_null_1'
|
|
mock_is_b.side_effect = ['asc_null_filter_1', 'asc_null_filter_2']
|
|
|
|
utils.paginate_query(self.query, self.model, 5,
|
|
['user_id', 'updated_at'],
|
|
marker=self.marker,
|
|
sort_dirs=[
|
|
'asc-nullslast', 'desc-nullsfirst'])
|
|
mock_isnot.assert_called_once_with(None)
|
|
mock_is_a.assert_called_once_with(None)
|
|
mock_is_b.assert_has_calls([mock.call(None), mock.call(None)])
|
|
|
|
self.mock_asc.assert_called_once_with(self.model.user_id)
|
|
self.mock_desc.assert_has_calls([
|
|
mock.call('asc_null_1'),
|
|
mock.call('desc_null_1'),
|
|
mock.call(self.model.updated_at),
|
|
])
|
|
mock_and.assert_called_once_with('or_1')
|
|
mock_or.assert_has_calls([
|
|
mock.call(mock.ANY, 'asc_null_filter_2'),
|
|
mock.call('some_crit'),
|
|
])
|
|
self.query.order_by.assert_has_calls([
|
|
mock.call('asc_null_2'),
|
|
mock.call('asc_1'),
|
|
mock.call('desc_null_2'),
|
|
mock.call('desc_1'),
|
|
])
|
|
self.query.filter.assert_called_once_with('some_f')
|
|
self.query.limit.assert_called_once_with(5)
|
|
|
|
@mock.patch.object(sqlalchemy.sql, 'and_')
|
|
@mock.patch.object(sqlalchemy.sql, 'or_')
|
|
def test_paginate_query_marker_null_with_two_primary_keys(
|
|
self, mock_or, mock_and):
|
|
self.mock_asc.return_value = 'asc_1'
|
|
self.mock_desc.side_effect = [
|
|
'asc_null_2',
|
|
'desc_null_2',
|
|
'desc_1',
|
|
'desc_null_4',
|
|
'desc_4',
|
|
]
|
|
self.query.order_by.return_value = self.query
|
|
mock_or.side_effect = [
|
|
'or_1',
|
|
'or_2',
|
|
'some_f',
|
|
]
|
|
mock_and.side_effect = [
|
|
'some_crit',
|
|
'other_crit',
|
|
]
|
|
self.query.filter.return_value = self.query
|
|
|
|
with mock.patch.object(self.model.user_id, 'isnot') as mock_isnot, \
|
|
mock.patch.object(self.model.updated_at, 'is_') as mock_is_a, \
|
|
mock.patch.object(self.model.user_id, 'is_') as mock_is_b, \
|
|
mock.patch.object(self.model.project_id, 'is_') as mock_is_c:
|
|
|
|
mock_isnot.return_value = 'asc_null_1'
|
|
mock_is_a.return_value = 'desc_null_1'
|
|
mock_is_b.side_effect = ['asc_null_filter_1', 'asc_null_filter_2']
|
|
mock_is_c.side_effect = ['desc_null_3', 'desc_null_filter_3']
|
|
|
|
utils.paginate_query(self.query, self.model, 5,
|
|
['user_id', 'updated_at', 'project_id'],
|
|
marker=self.marker,
|
|
sort_dirs=['asc-nullslast', 'desc-nullsfirst',
|
|
'desc-nullsfirst'])
|
|
|
|
mock_isnot.assert_called_once_with(None)
|
|
mock_is_a.assert_called_once_with(None)
|
|
mock_is_b.assert_has_calls([mock.call(None), mock.call(None)])
|
|
mock_is_c.assert_has_calls([mock.call(None), mock.call(None)])
|
|
|
|
self.mock_asc.assert_called_once_with(self.model.user_id)
|
|
self.mock_desc.assert_has_calls([
|
|
mock.call('asc_null_1'),
|
|
mock.call('desc_null_1'),
|
|
mock.call(self.model.updated_at),
|
|
mock.call('desc_null_3'),
|
|
mock.call(self.model.project_id),
|
|
])
|
|
self.query.order_by.assert_has_calls = [
|
|
mock.call('asc_null_2'),
|
|
mock.call('asc_1'),
|
|
mock.call('desc_null_2'),
|
|
mock.call('desc_1'),
|
|
mock.call('desc_null_4'),
|
|
mock.call('desc_4'),
|
|
]
|
|
mock_or.assert_has_calls([
|
|
mock.call(mock.ANY, 'asc_null_filter_2'),
|
|
mock.call(mock.ANY, 'desc_null_filter_3'),
|
|
mock.call('some_crit', 'other_crit'),
|
|
])
|
|
mock_and.assert_has_calls([
|
|
mock.call('or_1'),
|
|
mock.call(mock.ANY, 'or_2'),
|
|
])
|
|
self.query.filter.assert_called_once_with('some_f')
|
|
self.query.limit.assert_called_once_with(5)
|
|
|
|
def test_paginate_query_value_error(self):
|
|
self.mock_asc.return_value = 'asc_1'
|
|
self.query.order_by.return_value = self.query
|
|
|
|
self.assertRaises(ValueError, utils.paginate_query,
|
|
self.query, self.model, 5, ['user_id', 'project_id'],
|
|
marker=self.marker, sort_dirs=['asc', 'mixed'])
|
|
|
|
self.mock_asc.assert_called_once_with(self.model.user_id)
|
|
self.query.order_by.assert_called_once_with('asc_1')
|
|
|
|
def test_paginate_on_hybrid(self):
|
|
self.mock_asc.return_value = 'asc_1'
|
|
self.mock_desc.return_value = 'desc_1'
|
|
self.query.order_by.return_value = self.query
|
|
|
|
utils.paginate_query(self.query, self.model, 5,
|
|
['user_id', 'some_hybrid'],
|
|
sort_dirs=['asc', 'desc'])
|
|
|
|
self.mock_asc.assert_called_once_with(self.model.user_id)
|
|
self.mock_desc.assert_called_once_with(self.model.some_hybrid)
|
|
self.query.order_by.assert_has_calls([
|
|
mock.call('asc_1'),
|
|
mock.call('desc_1'),
|
|
])
|
|
self.query.limit.assert_called_once_with(5)
|
|
|
|
|
|
class Test_UnstableSortingOrder(test_base.BaseTestCase):
|
|
def test_multiple_primary_keys_stable(self):
|
|
self.assertTrue(
|
|
utils._stable_sorting_order(
|
|
FakeTableWithMultipleKeys, ['key1', 'key2']))
|
|
|
|
def test_classically_mapped_primary_keys_stable(self):
|
|
self.assertTrue(
|
|
utils._stable_sorting_order(FakeTableClassicalyMapped, ['id']))
|
|
|
|
def test_multiple_primary_keys_unstable(self):
|
|
self.assertFalse(
|
|
utils._stable_sorting_order(
|
|
FakeTableWithMultipleKeys, ['key1', 'key3']))
|
|
|
|
def test_joined_inh_stable(self):
|
|
self.assertTrue(
|
|
utils._stable_sorting_order(FakeTableJoinedInh, ['user_id'])
|
|
)
|
|
|
|
def test_single_inh_stable(self):
|
|
self.assertTrue(
|
|
utils._stable_sorting_order(FakeTableSingleInh, ['user_id'])
|
|
)
|
|
|
|
def test_unknown_primary_keys_stable(self):
|
|
self.assertIsNone(
|
|
utils._stable_sorting_order(object, ['key1', 'key2']))
|
|
|
|
def test_unique_index_stable(self):
|
|
self.assertTrue(
|
|
utils._stable_sorting_order(
|
|
FakeTableWithIndexes, ['key1', 'key2']))
|
|
|
|
def test_unique_index_unstable(self):
|
|
self.assertFalse(
|
|
utils._stable_sorting_order(
|
|
FakeTableWithIndexes, ['key1', 'key3']))
|
|
|
|
|
|
class TestGetUniqueKeys(test_base.BaseTestCase):
|
|
def test_multiple_primary_keys(self):
|
|
self.assertEqual(
|
|
[{'key1', 'key2'}],
|
|
utils.get_unique_keys(FakeTableWithMultipleKeys))
|
|
|
|
def test_unique_index(self):
|
|
self.assertEqual(
|
|
[{'id'}, {'key1', 'key2'}],
|
|
utils.get_unique_keys(FakeTableWithIndexes))
|
|
|
|
def test_unknown_primary_keys(self):
|
|
self.assertIsNone(utils.get_unique_keys(object))
|
|
|
|
def test_cache(self):
|
|
|
|
class CacheTable(object):
|
|
info = {}
|
|
constraints_called = 0
|
|
indexes_called = 0
|
|
|
|
@property
|
|
def constraints(self):
|
|
self.constraints_called += 1
|
|
return []
|
|
|
|
@property
|
|
def indexes(self):
|
|
self.indexes_called += 1
|
|
return []
|
|
|
|
class CacheModel(object):
|
|
pass
|
|
|
|
table = CacheTable()
|
|
mapper_mock = mock.Mock(mapped_table=table, local_table=table)
|
|
mapper_mock.base_mapper = mapper_mock
|
|
mock_inspect = mock.Mock(
|
|
return_value=mapper_mock)
|
|
model = CacheModel()
|
|
self.assertNotIn('oslodb_unique_keys', CacheTable.info)
|
|
with mock.patch("oslo_db.sqlalchemy.utils.inspect", mock_inspect):
|
|
utils.get_unique_keys(model)
|
|
|
|
self.assertIn('oslodb_unique_keys', CacheTable.info)
|
|
self.assertEqual(1, table.constraints_called)
|
|
self.assertEqual(1, table.indexes_called)
|
|
|
|
for i in range(10):
|
|
utils.get_unique_keys(model)
|
|
|
|
self.assertEqual(1, table.constraints_called)
|
|
self.assertEqual(1, table.indexes_called)
|
|
|
|
|
|
class TestPaginateQueryActualSQL(test_base.BaseTestCase):
|
|
|
|
def test_paginate_with_boolean_sort(self):
|
|
s = Session()
|
|
q = s.query(FakeTable)
|
|
q = utils.paginate_query(q, FakeTable, 5, ['enabled'],
|
|
sort_dirs=['asc'],
|
|
marker=FakeTable(user_id='hello',
|
|
enabled=False))
|
|
expected_core_sql = (
|
|
select(FakeTable).
|
|
order_by(sqlalchemy.asc(FakeTable.enabled)).
|
|
where(cast(FakeTable.enabled, Integer) > 0).
|
|
limit(5)
|
|
)
|
|
|
|
self.assertEqual(
|
|
str(expected_core_sql.compile()),
|
|
str(q.statement.compile())
|
|
)
|
|
|
|
def test_paginate_on_hybrid_assert_stmt(self):
|
|
s = Session()
|
|
q = s.query(FakeTable)
|
|
q = utils.paginate_query(
|
|
q, FakeTable, 5,
|
|
['user_id', 'some_hybrid'],
|
|
sort_dirs=['asc', 'desc'])
|
|
expected_core_sql = (
|
|
select(FakeTable).
|
|
order_by(sqlalchemy.asc(FakeTable.user_id)).
|
|
order_by(sqlalchemy.desc(FakeTable.some_hybrid)).
|
|
limit(5)
|
|
)
|
|
|
|
self.assertEqual(
|
|
str(expected_core_sql.compile()),
|
|
str(q.statement.compile())
|
|
)
|
|
|
|
|
|
class TestMigrationUtils(db_test_base._DbTestCase):
|
|
|
|
"""Class for testing utils that are used in db migrations."""
|
|
|
|
def setUp(self):
|
|
super(TestMigrationUtils, self).setUp()
|
|
self.meta = MetaData()
|
|
self.conn = self.engine.connect()
|
|
|
|
# self.conn would be better here but does not work right now
|
|
self.addCleanup(self.meta.drop_all, self.engine)
|
|
self.addCleanup(self.conn.close)
|
|
|
|
def _populate_db_for_drop_duplicate_entries(self, engine, meta,
|
|
table_name):
|
|
values = [
|
|
{'id': 11, 'a': 3, 'b': 10, 'c': 'abcdef'},
|
|
{'id': 12, 'a': 5, 'b': 10, 'c': 'abcdef'},
|
|
{'id': 13, 'a': 6, 'b': 10, 'c': 'abcdef'},
|
|
{'id': 14, 'a': 7, 'b': 10, 'c': 'abcdef'},
|
|
{'id': 21, 'a': 1, 'b': 20, 'c': 'aa'},
|
|
{'id': 31, 'a': 1, 'b': 20, 'c': 'bb'},
|
|
{'id': 41, 'a': 1, 'b': 30, 'c': 'aef'},
|
|
{'id': 42, 'a': 2, 'b': 30, 'c': 'aef'},
|
|
{'id': 43, 'a': 3, 'b': 30, 'c': 'aef'}
|
|
]
|
|
|
|
test_table = Table(table_name, meta,
|
|
Column('id', Integer, primary_key=True,
|
|
nullable=False),
|
|
Column('a', Integer),
|
|
Column('b', Integer),
|
|
Column('c', String(255)),
|
|
Column('deleted', Integer, default=0),
|
|
Column('deleted_at', DateTime),
|
|
Column('updated_at', DateTime))
|
|
|
|
test_table.create(engine)
|
|
with engine.connect() as conn:
|
|
conn.execute(test_table.insert(), values)
|
|
return test_table, values
|
|
|
|
def test_drop_old_duplicate_entries_from_table(self):
|
|
table_name = "__test_tmp_table__"
|
|
|
|
test_table, values = self._populate_db_for_drop_duplicate_entries(
|
|
self.engine, self.meta, table_name)
|
|
utils.drop_old_duplicate_entries_from_table(
|
|
self.engine, table_name, False, 'b', 'c')
|
|
|
|
uniq_values = set()
|
|
expected_ids = []
|
|
for value in sorted(values, key=lambda x: x['id'], reverse=True):
|
|
uniq_value = (('b', value['b']), ('c', value['c']))
|
|
if uniq_value in uniq_values:
|
|
continue
|
|
uniq_values.add(uniq_value)
|
|
expected_ids.append(value['id'])
|
|
|
|
with self.engine.connect() as conn:
|
|
real_ids = [
|
|
row[0] for row in
|
|
conn.execute(select(test_table.c.id)).fetchall()
|
|
]
|
|
|
|
self.assertEqual(len(expected_ids), len(real_ids))
|
|
for id_ in expected_ids:
|
|
self.assertIn(id_, real_ids)
|
|
|
|
def test_drop_dup_entries_in_file_conn(self):
|
|
table_name = "__test_tmp_table__"
|
|
tmp_db_file = self.create_tempfiles([['name', '']], ext='.sql')[0]
|
|
in_file_engine = session.EngineFacade(
|
|
'sqlite:///%s' % tmp_db_file).get_engine()
|
|
meta = MetaData()
|
|
test_table, values = self._populate_db_for_drop_duplicate_entries(
|
|
in_file_engine, meta, table_name)
|
|
utils.drop_old_duplicate_entries_from_table(
|
|
in_file_engine, table_name, False, 'b', 'c')
|
|
|
|
def test_drop_old_duplicate_entries_from_table_soft_delete(self):
|
|
table_name = "__test_tmp_table__"
|
|
|
|
table, values = self._populate_db_for_drop_duplicate_entries(
|
|
self.engine, self.meta, table_name)
|
|
utils.drop_old_duplicate_entries_from_table(self.engine, table_name,
|
|
True, 'b', 'c')
|
|
uniq_values = set()
|
|
expected_values = []
|
|
soft_deleted_values = []
|
|
|
|
for value in sorted(values, key=lambda x: x['id'], reverse=True):
|
|
uniq_value = (('b', value['b']), ('c', value['c']))
|
|
if uniq_value in uniq_values:
|
|
soft_deleted_values.append(value)
|
|
continue
|
|
uniq_values.add(uniq_value)
|
|
expected_values.append(value)
|
|
|
|
base_select = table.select()
|
|
|
|
with self.engine.connect() as conn:
|
|
rows_select = base_select.where(table.c.deleted != table.c.id)
|
|
row_ids = [
|
|
row.id for row in conn.execute(rows_select).fetchall()
|
|
]
|
|
self.assertEqual(len(expected_values), len(row_ids))
|
|
for value in expected_values:
|
|
self.assertIn(value['id'], row_ids)
|
|
|
|
deleted_rows_select = base_select.where(
|
|
table.c.deleted == table.c.id)
|
|
deleted_rows_ids = [
|
|
row.id for row in
|
|
conn.execute(deleted_rows_select).fetchall()
|
|
]
|
|
self.assertEqual(len(values) - len(row_ids),
|
|
len(deleted_rows_ids))
|
|
for value in soft_deleted_values:
|
|
self.assertIn(value['id'], deleted_rows_ids)
|
|
|
|
def test_change_deleted_column_type_does_not_drop_index(self):
|
|
table_name = 'abc'
|
|
|
|
indexes = {
|
|
'idx_a_deleted': ['a', 'deleted'],
|
|
'idx_b_deleted': ['b', 'deleted'],
|
|
'idx_a': ['a']
|
|
}
|
|
|
|
index_instances = [Index(name, *columns)
|
|
for name, columns in indexes.items()]
|
|
|
|
table = Table(table_name, self.meta,
|
|
Column('id', Integer, primary_key=True),
|
|
Column('a', String(255)),
|
|
Column('b', String(255)),
|
|
Column('deleted', Boolean),
|
|
*index_instances)
|
|
table.create(self.engine)
|
|
utils.change_deleted_column_type_to_id_type(self.engine, table_name)
|
|
utils.change_deleted_column_type_to_boolean(self.engine, table_name)
|
|
|
|
insp = sqlalchemy.inspect(self.engine)
|
|
real_indexes = insp.get_indexes(table_name)
|
|
self.assertEqual(3, len(real_indexes))
|
|
for index in real_indexes:
|
|
name = index['name']
|
|
self.assertIn(name, indexes)
|
|
self.assertEqual(set(indexes[name]),
|
|
set(index['column_names']))
|
|
|
|
def test_change_deleted_column_type_to_id_type_integer(self):
|
|
table_name = 'abc'
|
|
table = Table(table_name, self.meta,
|
|
Column('id', Integer, primary_key=True),
|
|
Column('deleted', Boolean))
|
|
table.create(self.engine)
|
|
utils.change_deleted_column_type_to_id_type(self.engine, table_name)
|
|
|
|
table = utils.get_table(self.engine, table_name)
|
|
self.assertIsInstance(table.c.deleted.type, Integer)
|
|
|
|
def test_change_deleted_column_type_to_id_type_string(self):
|
|
table_name = 'abc'
|
|
table = Table(table_name, self.meta,
|
|
Column('id', String(255), primary_key=True),
|
|
Column('deleted', Boolean))
|
|
table.create(self.engine)
|
|
utils.change_deleted_column_type_to_id_type(self.engine, table_name)
|
|
|
|
table = utils.get_table(self.engine, table_name)
|
|
self.assertIsInstance(table.c.deleted.type, String)
|
|
|
|
@db_test_base.backend_specific('sqlite')
|
|
def test_change_deleted_column_type_to_id_type_custom(self):
|
|
table_name = 'abc'
|
|
table = Table(table_name, self.meta,
|
|
Column('id', Integer, primary_key=True),
|
|
Column('foo', CustomType),
|
|
Column('deleted', Boolean))
|
|
table.create(self.engine)
|
|
|
|
fooColumn = Column('foo', CustomType())
|
|
utils.change_deleted_column_type_to_id_type(self.engine, table_name,
|
|
foo=fooColumn)
|
|
|
|
table = utils.get_table(self.engine, table_name)
|
|
|
|
self.assertIsInstance(table.c.deleted.type, Integer)
|
|
|
|
def test_change_deleted_column_type_to_boolean(self):
|
|
expected_types = {'mysql': mysql.TINYINT}
|
|
table_name = 'abc'
|
|
table = Table(table_name, self.meta,
|
|
Column('id', Integer, primary_key=True),
|
|
Column('deleted', Integer))
|
|
table.create(self.engine)
|
|
|
|
utils.change_deleted_column_type_to_boolean(self.engine, table_name)
|
|
|
|
table = utils.get_table(self.engine, table_name)
|
|
self.assertIsInstance(table.c.deleted.type,
|
|
expected_types.get(self.engine.name, Boolean))
|
|
|
|
def test_change_deleted_column_type_to_boolean_with_fc(self):
|
|
expected_types = {'mysql': mysql.TINYINT}
|
|
table_name_1 = 'abc'
|
|
table_name_2 = 'bcd'
|
|
|
|
table_1 = Table(table_name_1, self.meta,
|
|
Column('id', Integer, primary_key=True),
|
|
Column('deleted', Integer))
|
|
table_1.create(self.engine)
|
|
|
|
table_2 = Table(table_name_2, self.meta,
|
|
Column('id', Integer, primary_key=True),
|
|
Column('foreign_id', Integer,
|
|
ForeignKey('%s.id' % table_name_1)),
|
|
Column('deleted', Integer))
|
|
table_2.create(self.engine)
|
|
|
|
utils.change_deleted_column_type_to_boolean(self.engine, table_name_2)
|
|
|
|
table = utils.get_table(self.engine, table_name_2)
|
|
self.assertIsInstance(table.c.deleted.type,
|
|
expected_types.get(self.engine.name, Boolean))
|
|
|
|
@db_test_base.backend_specific('sqlite')
|
|
def test_change_deleted_column_type_to_boolean_type_custom(self):
|
|
table_name = 'abc'
|
|
table = Table(table_name, self.meta,
|
|
Column('id', Integer, primary_key=True),
|
|
Column('foo', CustomType),
|
|
Column('deleted', Integer))
|
|
table.create(self.engine)
|
|
|
|
fooColumn = Column('foo', CustomType())
|
|
utils.change_deleted_column_type_to_boolean(self.engine, table_name,
|
|
foo=fooColumn)
|
|
|
|
table = utils.get_table(self.engine, table_name)
|
|
self.assertIsInstance(table.c.deleted.type, Boolean)
|
|
|
|
def test_detect_boolean_deleted_constraint_detection(self):
|
|
table_name = 'abc'
|
|
table = Table(table_name, self.meta,
|
|
Column('id', Integer, primary_key=True),
|
|
Column('deleted', Boolean(create_constraint=True)))
|
|
ck = [
|
|
const for const in table.constraints if
|
|
isinstance(const, CheckConstraint)][0]
|
|
|
|
self.assertTrue(utils._is_deleted_column_constraint(ck))
|
|
|
|
self.assertFalse(
|
|
utils._is_deleted_column_constraint(
|
|
CheckConstraint("deleted > 5")
|
|
)
|
|
)
|
|
|
|
@db_test_base.backend_specific('sqlite')
|
|
def test_change_deleted_column_type_sqlite_drops_check_constraint(self):
|
|
table_name = 'abc'
|
|
table = Table(table_name, self.meta,
|
|
Column('id', Integer, primary_key=True),
|
|
Column('deleted', Boolean))
|
|
table.create(self.engine)
|
|
|
|
utils._change_deleted_column_type_to_id_type_sqlite(self.engine,
|
|
table_name)
|
|
table = Table(table_name, self.meta, autoload_with=self.engine)
|
|
# NOTE(I159): if the CHECK constraint has been dropped (expected
|
|
# behavior), any integer value can be inserted, otherwise only 1 or 0.
|
|
# NOTE(zzzeek): SQLAlchemy 1.2 Boolean type will disallow non 1/0
|
|
# value here, 1.1 also coerces to "1/0" so use raw SQL to test the
|
|
# constraint
|
|
with self.engine.connect() as conn:
|
|
conn.exec_driver_sql(
|
|
"INSERT INTO abc (deleted) VALUES (?)",
|
|
(10, ),
|
|
)
|
|
|
|
self.assertEqual(
|
|
10,
|
|
conn.scalar(sql.text("SELECT deleted FROM abc")),
|
|
)
|
|
|
|
def test_get_foreign_key_constraint_name(self):
|
|
table_1 = Table('table_name_1', self.meta,
|
|
Column('id', Integer, primary_key=True),
|
|
Column('deleted', Integer))
|
|
table_2 = Table('table_name_2', self.meta,
|
|
Column('id', Integer, primary_key=True),
|
|
Column('foreign_id', Integer),
|
|
ForeignKeyConstraint(['foreign_id'],
|
|
['table_name_1.id'],
|
|
name='table_name_2_fk1'),
|
|
Column('deleted', Integer))
|
|
|
|
self.meta.create_all(self.engine, tables=[table_1, table_2])
|
|
fkc = utils.get_foreign_key_constraint_name(self.engine,
|
|
'table_name_2',
|
|
'foreign_id')
|
|
self.assertEqual(fkc, 'table_name_2_fk1')
|
|
|
|
@db_test_base.backend_specific('mysql', 'postgresql')
|
|
def test_suspend_fk_constraints_for_col_alter(self):
|
|
|
|
a = Table(
|
|
'a', self.meta,
|
|
Column('id', Integer, primary_key=True)
|
|
)
|
|
b = Table(
|
|
'b', self.meta,
|
|
Column('key', Integer),
|
|
Column('archive_id', Integer),
|
|
Column('aid', ForeignKey('a.id')),
|
|
PrimaryKeyConstraint("key", "archive_id")
|
|
)
|
|
c = Table(
|
|
'c', self.meta,
|
|
Column('id', Integer, primary_key=True),
|
|
Column('aid', ForeignKey('a.id')),
|
|
Column('key', Integer),
|
|
Column('archive_id', Integer),
|
|
ForeignKeyConstraint(
|
|
['key', 'archive_id'], ['b.key', 'b.archive_id'],
|
|
name="some_composite_fk")
|
|
)
|
|
self.meta.create_all(self.engine, tables=[a, b, c])
|
|
|
|
def get_fk_entries():
|
|
inspector = sqlalchemy.inspect(self.engine)
|
|
return sorted(
|
|
inspector.get_foreign_keys('b') +
|
|
inspector.get_foreign_keys('c'),
|
|
key=lambda fk: fk['referred_table']
|
|
)
|
|
|
|
def normalize_fk_entries(fks):
|
|
return [{
|
|
'name': fk['name'],
|
|
'referred_columns': fk['referred_columns'],
|
|
'referred_table': fk['referred_table'],
|
|
} for fk in fks]
|
|
|
|
existing_foreign_keys = get_fk_entries()
|
|
self.assertEqual(
|
|
[{'name': mock.ANY,
|
|
'referred_columns': ['id'], 'referred_table': 'a'},
|
|
{'name': mock.ANY,
|
|
'referred_columns': ['id'], 'referred_table': 'a'},
|
|
{'name': 'some_composite_fk',
|
|
'referred_columns': ['key', 'archive_id'],
|
|
'referred_table': 'b'}],
|
|
normalize_fk_entries(existing_foreign_keys)
|
|
)
|
|
|
|
with mock.patch("oslo_db.sqlalchemy.ndb.ndb_status",
|
|
mock.Mock(return_value=True)):
|
|
with utils.suspend_fk_constraints_for_col_alter(
|
|
self.engine, 'a', 'id', referents=['b', 'c']):
|
|
no_a_foreign_keys = get_fk_entries()
|
|
self.assertEqual(
|
|
[{'name': 'some_composite_fk',
|
|
'referred_columns': ['key', 'archive_id'],
|
|
'referred_table': 'b'}],
|
|
normalize_fk_entries(no_a_foreign_keys)
|
|
)
|
|
|
|
self.assertEqual(existing_foreign_keys, get_fk_entries())
|
|
|
|
with mock.patch("oslo_db.sqlalchemy.ndb.ndb_status",
|
|
mock.Mock(return_value=True)):
|
|
with utils.suspend_fk_constraints_for_col_alter(
|
|
self.engine, 'b', 'archive_id', referents=['c']):
|
|
self.assertEqual(
|
|
[{'name': mock.ANY,
|
|
'referred_columns': ['id'], 'referred_table': 'a'},
|
|
{'name': mock.ANY,
|
|
'referred_columns': ['id'], 'referred_table': 'a'}],
|
|
normalize_fk_entries(get_fk_entries())
|
|
)
|
|
|
|
self.assertEqual(existing_foreign_keys, get_fk_entries())
|
|
|
|
with utils.suspend_fk_constraints_for_col_alter(
|
|
self.engine, 'a', 'id', referents=['b', 'c']):
|
|
self.assertEqual(existing_foreign_keys, get_fk_entries())
|
|
|
|
if self.engine.name == 'mysql':
|
|
self.engine.dialect._oslodb_enable_ndb_support = True
|
|
|
|
self.addCleanup(
|
|
setattr, self.engine.dialect, "_oslodb_enable_ndb_support",
|
|
False
|
|
)
|
|
|
|
with utils.suspend_fk_constraints_for_col_alter(
|
|
self.engine, 'a', 'id', referents=['b', 'c']):
|
|
self.assertEqual(no_a_foreign_keys, get_fk_entries())
|
|
|
|
|
|
class PostgresqlTestMigrations(TestMigrationUtils,
|
|
db_test_base._PostgreSQLOpportunisticTestCase):
|
|
|
|
"""Test migrations on PostgreSQL."""
|
|
pass
|
|
|
|
|
|
class MySQLTestMigrations(TestMigrationUtils,
|
|
db_test_base._MySQLOpportunisticTestCase):
|
|
|
|
"""Test migrations on MySQL."""
|
|
pass
|
|
|
|
|
|
class TestConnectionUtils(test_base.BaseTestCase):
|
|
|
|
def setUp(self):
|
|
super(TestConnectionUtils, self).setUp()
|
|
|
|
self.full_credentials = {'backend': 'postgresql',
|
|
'database': 'test',
|
|
'user': 'dude',
|
|
'passwd': 'pass'}
|
|
|
|
self.connect_string = 'postgresql://dude:pass@localhost/test'
|
|
|
|
# NOTE(rpodolyaka): mock the dialect parts, so that we don't depend
|
|
# on psycopg2 (or any other DBAPI implementation) in these tests
|
|
|
|
@classmethod
|
|
def fake_dbapi(cls):
|
|
return mock.MagicMock()
|
|
patch_dbapi = mock.patch.object(psycopg2.PGDialect_psycopg2, 'dbapi',
|
|
new=fake_dbapi)
|
|
patch_dbapi.start()
|
|
self.addCleanup(patch_dbapi.stop)
|
|
|
|
patch_onconnect = mock.patch.object(psycopg2.PGDialect_psycopg2,
|
|
'on_connect')
|
|
patch_onconnect.start()
|
|
self.addCleanup(patch_onconnect.stop)
|
|
|
|
def test_ensure_backend_available(self):
|
|
with mock.patch.object(
|
|
sqlalchemy.engine.base.Engine, 'connect') as mock_connect:
|
|
fake_connection = mock.Mock()
|
|
mock_connect.return_value = fake_connection
|
|
|
|
eng = provision.Backend._ensure_backend_available(
|
|
self.connect_string)
|
|
|
|
self.assertIsInstance(eng, sqlalchemy.engine.base.Engine)
|
|
self.assertEqual(self.connect_string, str(eng.url))
|
|
|
|
mock_connect.assert_called_once()
|
|
fake_connection.close.assert_called_once()
|
|
|
|
def test_ensure_backend_available_no_connection_raises(self):
|
|
log = self.useFixture(fixtures.FakeLogger())
|
|
err = OperationalError("Can't connect to database", None, None)
|
|
with mock.patch.object(
|
|
sqlalchemy.engine.base.Engine, 'connect') as mock_connect:
|
|
mock_connect.side_effect = err
|
|
|
|
exc = self.assertRaises(
|
|
exception.BackendNotAvailable,
|
|
provision.Backend._ensure_backend_available,
|
|
self.connect_string)
|
|
self.assertEqual(
|
|
"Backend 'postgresql' is unavailable: "
|
|
"Could not connect", str(exc))
|
|
self.assertEqual(
|
|
"The postgresql backend is unavailable: %s" % err,
|
|
log.output.strip())
|
|
|
|
def test_ensure_backend_available_no_dbapi_raises(self):
|
|
log = self.useFixture(fixtures.FakeLogger())
|
|
with mock.patch.object(sqlalchemy, 'create_engine') as mock_create:
|
|
mock_create.side_effect = ImportError(
|
|
"Can't import DBAPI module foobar")
|
|
|
|
exc = self.assertRaises(
|
|
exception.BackendNotAvailable,
|
|
provision.Backend._ensure_backend_available,
|
|
self.connect_string)
|
|
|
|
mock_create.assert_called_once_with(
|
|
sa_url.make_url(self.connect_string))
|
|
|
|
self.assertEqual(
|
|
"Backend 'postgresql' is unavailable: "
|
|
"No DBAPI installed", str(exc))
|
|
self.assertEqual(
|
|
"The postgresql backend is unavailable: Can't import "
|
|
"DBAPI module foobar", log.output.strip())
|
|
|
|
def test_get_db_connection_info(self):
|
|
conn_pieces = parse.urlparse(self.connect_string)
|
|
self.assertEqual(('dude', 'pass', 'test', 'localhost'),
|
|
utils.get_db_connection_info(conn_pieces))
|
|
|
|
|
|
class MyModelSoftDeletedProjectId(declarative_base(), models.ModelBase,
|
|
models.SoftDeleteMixin):
|
|
__tablename__ = 'soft_deleted_project_id_test_model'
|
|
id = Column(Integer, primary_key=True)
|
|
project_id = Column(Integer)
|
|
|
|
|
|
class MyModel(declarative_base(), models.ModelBase):
|
|
__tablename__ = 'test_model'
|
|
id = Column(Integer, primary_key=True)
|
|
|
|
|
|
class MyModelSoftDeleted(declarative_base(), models.ModelBase,
|
|
models.SoftDeleteMixin):
|
|
__tablename__ = 'soft_deleted_test_model'
|
|
id = Column(Integer, primary_key=True)
|
|
|
|
|
|
class TestModelQuery(test_base.BaseTestCase):
|
|
|
|
def setUp(self):
|
|
super(TestModelQuery, self).setUp()
|
|
|
|
self.session = mock.MagicMock()
|
|
self.session.query.return_value = self.session.query
|
|
self.session.query.filter.return_value = self.session.query
|
|
|
|
def test_wrong_model(self):
|
|
self.assertRaises(TypeError, utils.model_query,
|
|
FakeModel, session=self.session)
|
|
|
|
def test_no_soft_deleted(self):
|
|
self.assertRaises(ValueError, utils.model_query,
|
|
MyModel, session=self.session, deleted=True)
|
|
|
|
def test_deleted_false(self):
|
|
mock_query = utils.model_query(
|
|
MyModelSoftDeleted, session=self.session, deleted=False)
|
|
|
|
deleted_filter = mock_query.filter.call_args[0][0]
|
|
self.assertEqual('soft_deleted_test_model.deleted = :deleted_1',
|
|
str(deleted_filter))
|
|
self.assertEqual(deleted_filter.right.value,
|
|
MyModelSoftDeleted.__mapper__.c.deleted.default.arg)
|
|
|
|
def test_deleted_true(self):
|
|
mock_query = utils.model_query(
|
|
MyModelSoftDeleted, session=self.session, deleted=True)
|
|
|
|
deleted_filter = mock_query.filter.call_args[0][0]
|
|
self.assertEqual(str(deleted_filter),
|
|
'soft_deleted_test_model.deleted != :deleted_1')
|
|
self.assertEqual(deleted_filter.right.value,
|
|
MyModelSoftDeleted.__mapper__.c.deleted.default.arg)
|
|
|
|
@mock.patch.object(utils, "_read_deleted_filter")
|
|
def test_no_deleted_value(self, _read_deleted_filter):
|
|
utils.model_query(MyModelSoftDeleted, session=self.session)
|
|
self.assertEqual(0, _read_deleted_filter.call_count)
|
|
|
|
def test_project_filter(self):
|
|
project_id = 10
|
|
|
|
mock_query = utils.model_query(
|
|
MyModelSoftDeletedProjectId, session=self.session,
|
|
project_only=True, project_id=project_id)
|
|
|
|
deleted_filter = mock_query.filter.call_args[0][0]
|
|
self.assertEqual(
|
|
'soft_deleted_project_id_test_model.project_id = :project_id_1',
|
|
str(deleted_filter))
|
|
self.assertEqual(project_id, deleted_filter.right.value)
|
|
|
|
def test_project_filter_wrong_model(self):
|
|
self.assertRaises(ValueError, utils.model_query,
|
|
MyModelSoftDeleted, session=self.session,
|
|
project_id=10)
|
|
|
|
def test_project_filter_allow_none(self):
|
|
mock_query = utils.model_query(
|
|
MyModelSoftDeletedProjectId,
|
|
session=self.session, project_id=(10, None))
|
|
|
|
self.assertEqual(
|
|
'soft_deleted_project_id_test_model.project_id'
|
|
' IN (:project_id_1, NULL)',
|
|
str(mock_query.filter.call_args[0][0])
|
|
)
|
|
|
|
def test_model_query_common(self):
|
|
utils.model_query(MyModel, args=(MyModel.id,), session=self.session)
|
|
self.session.query.assert_called_with(MyModel.id)
|
|
|
|
|
|
class TestUtils(db_test_base._DbTestCase):
|
|
def setUp(self):
|
|
super(TestUtils, self).setUp()
|
|
meta = MetaData()
|
|
self.test_table = Table(
|
|
'test_table',
|
|
meta,
|
|
Column('a', Integer),
|
|
Column('b', Integer)
|
|
)
|
|
self.test_table.create(self.engine)
|
|
self.addCleanup(meta.drop_all, self.engine)
|
|
|
|
def test_get_indexes(self):
|
|
Index('index_a', self.test_table.c.a).create(self.engine)
|
|
Index('index_b', self.test_table.c.b).create(self.engine)
|
|
indexes = utils.get_indexes(self.engine, "test_table")
|
|
indexes = [(index['name'], index['column_names']) for index in indexes]
|
|
self.assertIn(('index_a', ['a']), indexes)
|
|
self.assertIn(('index_b', ['b']), indexes)
|
|
|
|
def test_index_exists(self):
|
|
self.assertFalse(utils.index_exists(self.engine, 'test_table',
|
|
'new_index'))
|
|
Index('new_index', self.test_table.c.a).create(self.engine)
|
|
self.assertTrue(utils.index_exists(self.engine, 'test_table',
|
|
'new_index'))
|
|
|
|
def test_index_exists_on_columns(self):
|
|
columns = [self.test_table.c.a, self.test_table.c.b]
|
|
Index('new_index', *columns).create(self.engine)
|
|
self.assertTrue(utils.index_exists_on_columns(self.engine,
|
|
'test_table',
|
|
('a', 'b')))
|
|
|
|
def test_add_index(self):
|
|
self.assertFalse(utils.index_exists(self.engine, 'test_table',
|
|
'new_index'))
|
|
utils.add_index(self.engine, 'test_table', 'new_index', ('a',))
|
|
self.assertTrue(utils.index_exists(self.engine, 'test_table',
|
|
'new_index'))
|
|
|
|
def test_add_existing_index(self):
|
|
Index('new_index', self.test_table.c.a).create(self.engine)
|
|
self.assertRaises(ValueError, utils.add_index, self.engine,
|
|
'test_table', 'new_index', ('a',))
|
|
|
|
def test_drop_index(self):
|
|
Index('new_index', self.test_table.c.a).create(self.engine)
|
|
utils.drop_index(self.engine, 'test_table', 'new_index')
|
|
self.assertFalse(utils.index_exists(self.engine, 'test_table',
|
|
'new_index'))
|
|
|
|
def test_drop_unexisting_index(self):
|
|
self.assertRaises(ValueError, utils.drop_index, self.engine,
|
|
'test_table', 'new_index')
|
|
|
|
@mock.patch('oslo_db.sqlalchemy.utils.drop_index')
|
|
@mock.patch('oslo_db.sqlalchemy.utils.add_index')
|
|
def test_change_index_columns(self, add_index, drop_index):
|
|
utils.change_index_columns(self.engine, 'test_table', 'a_index',
|
|
('a',))
|
|
utils.drop_index.assert_called_once_with(self.engine, 'test_table',
|
|
'a_index')
|
|
utils.add_index.assert_called_once_with(self.engine, 'test_table',
|
|
'a_index', ('a',))
|
|
|
|
def test_column_exists(self):
|
|
for col in ['a', 'b']:
|
|
self.assertTrue(utils.column_exists(self.engine, 'test_table',
|
|
col))
|
|
self.assertFalse(utils.column_exists(self.engine, 'test_table',
|
|
'fake_column'))
|
|
|
|
|
|
class TestUtilsMysqlOpportunistically(
|
|
TestUtils, db_test_base._MySQLOpportunisticTestCase):
|
|
pass
|
|
|
|
|
|
class TestUtilsPostgresqlOpportunistically(
|
|
TestUtils, db_test_base._PostgreSQLOpportunisticTestCase):
|
|
pass
|
|
|
|
|
|
class TestDialectFunctionDispatcher(test_base.BaseTestCase):
|
|
def _single_fixture(self):
|
|
callable_fn = mock.Mock()
|
|
|
|
dispatcher = orig = utils.dispatch_for_dialect("*")(
|
|
callable_fn.default)
|
|
dispatcher = dispatcher.dispatch_for("sqlite")(callable_fn.sqlite)
|
|
dispatcher = dispatcher.dispatch_for("mysql+pymysql")(
|
|
callable_fn.mysql_pymysql)
|
|
dispatcher = dispatcher.dispatch_for("mysql")(
|
|
callable_fn.mysql)
|
|
dispatcher = dispatcher.dispatch_for("postgresql")(
|
|
callable_fn.postgresql)
|
|
|
|
self.assertTrue(dispatcher is orig)
|
|
|
|
return dispatcher, callable_fn
|
|
|
|
def _multiple_fixture(self):
|
|
callable_fn = mock.Mock()
|
|
|
|
for targ in [
|
|
callable_fn.default,
|
|
callable_fn.sqlite,
|
|
callable_fn.mysql,
|
|
callable_fn.mysql_pymysql,
|
|
callable_fn.postgresql,
|
|
callable_fn.postgresql_psycopg2,
|
|
callable_fn.pyodbc
|
|
]:
|
|
targ.return_value = None
|
|
|
|
dispatcher = orig = utils.dispatch_for_dialect("*", multiple=True)(
|
|
callable_fn.default)
|
|
dispatcher = dispatcher.dispatch_for("sqlite")(callable_fn.sqlite)
|
|
dispatcher = dispatcher.dispatch_for("mysql+pymysql")(
|
|
callable_fn.mysql_pymysql)
|
|
dispatcher = dispatcher.dispatch_for("mysql")(
|
|
callable_fn.mysql)
|
|
dispatcher = dispatcher.dispatch_for("postgresql+*")(
|
|
callable_fn.postgresql)
|
|
dispatcher = dispatcher.dispatch_for("postgresql+psycopg2")(
|
|
callable_fn.postgresql_psycopg2)
|
|
dispatcher = dispatcher.dispatch_for("*+pyodbc")(
|
|
callable_fn.pyodbc)
|
|
|
|
self.assertTrue(dispatcher is orig)
|
|
|
|
return dispatcher, callable_fn
|
|
|
|
def test_single(self):
|
|
|
|
dispatcher, callable_fn = self._single_fixture()
|
|
dispatcher("sqlite://", 1)
|
|
dispatcher("postgresql+psycopg2://u:p@h/t", 2)
|
|
dispatcher("mysql+pymysql://u:p@h/t", 3)
|
|
dispatcher("mysql://u:p@h/t", 4)
|
|
dispatcher("mysql+mysqlconnector://u:p@h/t", 5)
|
|
|
|
self.assertEqual(
|
|
[
|
|
mock.call.sqlite('sqlite://', 1),
|
|
mock.call.postgresql("postgresql+psycopg2://u:p@h/t", 2),
|
|
mock.call.mysql_pymysql("mysql+pymysql://u:p@h/t", 3),
|
|
mock.call.mysql("mysql://u:p@h/t", 4),
|
|
mock.call.mysql("mysql+mysqlconnector://u:p@h/t", 5),
|
|
],
|
|
callable_fn.mock_calls)
|
|
|
|
def test_single_kwarg(self):
|
|
dispatcher, callable_fn = self._single_fixture()
|
|
dispatcher("sqlite://", foo='bar')
|
|
dispatcher("postgresql+psycopg2://u:p@h/t", 1, x='y')
|
|
|
|
self.assertEqual(
|
|
[
|
|
mock.call.sqlite('sqlite://', foo='bar'),
|
|
mock.call.postgresql(
|
|
"postgresql+psycopg2://u:p@h/t",
|
|
1, x='y'),
|
|
],
|
|
callable_fn.mock_calls)
|
|
|
|
def test_dispatch_on_target(self):
|
|
callable_fn = mock.Mock()
|
|
|
|
@utils.dispatch_for_dialect("*")
|
|
def default_fn(url, x, y):
|
|
callable_fn.default(url, x, y)
|
|
|
|
@default_fn.dispatch_for("sqlite")
|
|
def sqlite_fn(url, x, y):
|
|
callable_fn.sqlite(url, x, y)
|
|
default_fn.dispatch_on_drivername("*")(url, x, y)
|
|
|
|
default_fn("sqlite://", 4, 5)
|
|
self.assertEqual(
|
|
[
|
|
mock.call.sqlite("sqlite://", 4, 5),
|
|
mock.call.default("sqlite://", 4, 5)
|
|
],
|
|
callable_fn.mock_calls
|
|
)
|
|
|
|
def test_single_no_dispatcher(self):
|
|
callable_fn = mock.Mock()
|
|
|
|
dispatcher = utils.dispatch_for_dialect("sqlite")(callable_fn.sqlite)
|
|
dispatcher = dispatcher.dispatch_for("mysql")(callable_fn.mysql)
|
|
exc = self.assertRaises(
|
|
ValueError,
|
|
dispatcher, "postgresql://s:t@localhost/test"
|
|
)
|
|
self.assertEqual(
|
|
"No default function found for driver: 'postgresql+psycopg2'",
|
|
str(exc)
|
|
)
|
|
|
|
def test_multiple_no_dispatcher(self):
|
|
callable_fn = mock.Mock()
|
|
|
|
dispatcher = utils.dispatch_for_dialect("sqlite", multiple=True)(
|
|
callable_fn.sqlite)
|
|
dispatcher = dispatcher.dispatch_for("mysql")(callable_fn.mysql)
|
|
dispatcher("postgresql://s:t@localhost/test")
|
|
self.assertEqual(
|
|
[], callable_fn.mock_calls
|
|
)
|
|
|
|
def test_multiple_no_driver(self):
|
|
callable_fn = mock.Mock(
|
|
default=mock.Mock(return_value=None),
|
|
sqlite=mock.Mock(return_value=None)
|
|
)
|
|
|
|
dispatcher = utils.dispatch_for_dialect("*", multiple=True)(
|
|
callable_fn.default)
|
|
dispatcher = dispatcher.dispatch_for("sqlite")(
|
|
callable_fn.sqlite)
|
|
|
|
dispatcher.dispatch_on_drivername("sqlite")("foo")
|
|
self.assertEqual(
|
|
[mock.call.sqlite("foo"), mock.call.default("foo")],
|
|
callable_fn.mock_calls
|
|
)
|
|
|
|
def test_multiple_nesting(self):
|
|
callable_fn = mock.Mock(
|
|
default=mock.Mock(return_value=None),
|
|
mysql=mock.Mock(return_value=None)
|
|
)
|
|
|
|
dispatcher = utils.dispatch_for_dialect("*", multiple=True)(
|
|
callable_fn.default)
|
|
|
|
dispatcher = dispatcher.dispatch_for("mysql+mysqlconnector")(
|
|
dispatcher.dispatch_for("mysql+mysqldb")(
|
|
callable_fn.mysql
|
|
)
|
|
)
|
|
|
|
mysqldb_url = sqlalchemy.engine.url.make_url("mysql+mysqldb://")
|
|
mysqlconnector_url = sqlalchemy.engine.url.make_url(
|
|
"mysql+mysqlconnector://")
|
|
sqlite_url = sqlalchemy.engine.url.make_url("sqlite://")
|
|
|
|
dispatcher(mysqldb_url, 1)
|
|
dispatcher(mysqlconnector_url, 2)
|
|
dispatcher(sqlite_url, 3)
|
|
|
|
self.assertEqual(
|
|
[
|
|
mock.call.mysql(mysqldb_url, 1),
|
|
mock.call.default(mysqldb_url, 1),
|
|
mock.call.mysql(mysqlconnector_url, 2),
|
|
mock.call.default(mysqlconnector_url, 2),
|
|
mock.call.default(sqlite_url, 3)
|
|
],
|
|
callable_fn.mock_calls
|
|
)
|
|
|
|
def test_single_retval(self):
|
|
dispatcher, callable_fn = self._single_fixture()
|
|
callable_fn.mysql_pymysql.return_value = 5
|
|
|
|
self.assertEqual(
|
|
5, dispatcher("mysql+pymysql://u:p@h/t", 3)
|
|
)
|
|
|
|
def test_engine(self):
|
|
eng = sqlalchemy.create_engine("sqlite:///path/to/my/db.db")
|
|
dispatcher, callable_fn = self._single_fixture()
|
|
|
|
dispatcher(eng)
|
|
self.assertEqual(
|
|
[mock.call.sqlite(eng)],
|
|
callable_fn.mock_calls
|
|
)
|
|
|
|
def test_url_pymysql(self):
|
|
url = sqlalchemy.engine.url.make_url(
|
|
"mysql+pymysql://scott:tiger@localhost/test")
|
|
dispatcher, callable_fn = self._single_fixture()
|
|
|
|
dispatcher(url, 15)
|
|
self.assertEqual(
|
|
[mock.call.mysql_pymysql(url, 15)],
|
|
callable_fn.mock_calls
|
|
)
|
|
|
|
def test_url_mysql_generic(self):
|
|
url = sqlalchemy.engine.url.make_url(
|
|
"mysql://scott:tiger@localhost/test")
|
|
dispatcher, callable_fn = self._single_fixture()
|
|
|
|
dispatcher(url, 15)
|
|
self.assertEqual(
|
|
[mock.call.mysql(url, 15)],
|
|
callable_fn.mock_calls
|
|
)
|
|
|
|
def test_invalid_target(self):
|
|
dispatcher, callable_fn = self._single_fixture()
|
|
|
|
exc = self.assertRaises(
|
|
ValueError,
|
|
dispatcher, 20
|
|
)
|
|
self.assertEqual("Invalid target type: 20", str(exc))
|
|
|
|
def test_invalid_dispatch(self):
|
|
callable_fn = mock.Mock()
|
|
|
|
dispatcher = utils.dispatch_for_dialect("*")(callable_fn.default)
|
|
|
|
exc = self.assertRaises(
|
|
ValueError,
|
|
dispatcher.dispatch_for("+pyodbc"), callable_fn.pyodbc
|
|
)
|
|
self.assertEqual(
|
|
"Couldn't parse database[+driver]: '+pyodbc'",
|
|
str(exc)
|
|
)
|
|
|
|
def test_single_only_one_target(self):
|
|
callable_fn = mock.Mock()
|
|
|
|
dispatcher = utils.dispatch_for_dialect("*")(callable_fn.default)
|
|
dispatcher = dispatcher.dispatch_for("sqlite")(callable_fn.sqlite)
|
|
|
|
exc = self.assertRaises(
|
|
TypeError,
|
|
dispatcher.dispatch_for("sqlite"), callable_fn.sqlite2
|
|
)
|
|
self.assertEqual(
|
|
"Multiple functions for expression 'sqlite'", str(exc)
|
|
)
|
|
|
|
def test_multiple(self):
|
|
dispatcher, callable_fn = self._multiple_fixture()
|
|
|
|
dispatcher("postgresql+pyodbc://", 1)
|
|
dispatcher("mysql+pymysql://", 2)
|
|
dispatcher("postgresql+psycopg2://", 4)
|
|
dispatcher("postgresql://", 5)
|
|
|
|
# TODO(zzzeek): there is a deterministic order here, but we might
|
|
# want to tweak it, or maybe provide options. default first?
|
|
# most specific first? is *+pyodbc or postgresql+* more specific?
|
|
self.assertEqual(
|
|
[
|
|
mock.call.postgresql('postgresql+pyodbc://', 1),
|
|
mock.call.pyodbc('postgresql+pyodbc://', 1),
|
|
mock.call.default('postgresql+pyodbc://', 1),
|
|
mock.call.mysql_pymysql('mysql+pymysql://', 2),
|
|
mock.call.mysql('mysql+pymysql://', 2),
|
|
mock.call.default('mysql+pymysql://', 2),
|
|
mock.call.postgresql_psycopg2('postgresql+psycopg2://', 4),
|
|
mock.call.postgresql('postgresql+psycopg2://', 4),
|
|
mock.call.default('postgresql+psycopg2://', 4),
|
|
# note this is called because we resolve the default
|
|
# DBAPI for the url
|
|
mock.call.postgresql_psycopg2('postgresql://', 5),
|
|
mock.call.postgresql('postgresql://', 5),
|
|
mock.call.default('postgresql://', 5),
|
|
],
|
|
callable_fn.mock_calls
|
|
)
|
|
|
|
def test_multiple_no_return_value(self):
|
|
dispatcher, callable_fn = self._multiple_fixture()
|
|
callable_fn.sqlite.return_value = 5
|
|
|
|
exc = self.assertRaises(
|
|
TypeError,
|
|
dispatcher, "sqlite://"
|
|
)
|
|
self.assertEqual(
|
|
"Return value not allowed for multiple filtered function",
|
|
str(exc)
|
|
)
|
|
|
|
|
|
class TestGetInnoDBTables(db_test_base._MySQLOpportunisticTestCase):
|
|
|
|
def test_all_tables_use_innodb(self):
|
|
with self.engine.connect() as conn:
|
|
conn.execute(
|
|
sql.text(
|
|
"CREATE TABLE customers "
|
|
"(a INT, b CHAR (20), INDEX (a)) ENGINE=InnoDB"))
|
|
self.assertEqual([], utils.get_non_innodb_tables(self.engine))
|
|
|
|
def test_all_tables_use_innodb_false(self):
|
|
with self.engine.connect() as conn:
|
|
conn.execute(
|
|
sql.text("CREATE TABLE employee (i INT) ENGINE=MEMORY"))
|
|
self.assertEqual(['employee'],
|
|
utils.get_non_innodb_tables(self.engine))
|
|
|
|
def test_skip_tables_use_default_value(self):
|
|
with self.engine.connect() as conn:
|
|
conn.execute(
|
|
sql.text("CREATE TABLE migrate_version (i INT) ENGINE=MEMORY"))
|
|
self.assertEqual([],
|
|
utils.get_non_innodb_tables(self.engine))
|
|
|
|
def test_skip_tables_use_passed_value(self):
|
|
with self.engine.connect() as conn:
|
|
conn.execute(
|
|
sql.text("CREATE TABLE some_table (i INT) ENGINE=MEMORY"))
|
|
self.assertEqual([],
|
|
utils.get_non_innodb_tables(
|
|
self.engine, skip_tables=('some_table',)))
|
|
|
|
def test_skip_tables_use_empty_list(self):
|
|
with self.engine.connect() as conn:
|
|
conn.execute(
|
|
sql.text("CREATE TABLE some_table_3 (i INT) ENGINE=MEMORY"))
|
|
self.assertEqual(['some_table_3'],
|
|
utils.get_non_innodb_tables(
|
|
self.engine, skip_tables=()))
|
|
|
|
def test_skip_tables_use_several_values(self):
|
|
with self.engine.connect() as conn:
|
|
conn.execute(
|
|
sql.text("CREATE TABLE some_table_1 (i INT) ENGINE=MEMORY"))
|
|
conn.execute(
|
|
sql.text("CREATE TABLE some_table_2 (i INT) ENGINE=MEMORY"))
|
|
self.assertEqual([],
|
|
utils.get_non_innodb_tables(
|
|
self.engine,
|
|
skip_tables=('some_table_1', 'some_table_2')))
|