diff options
| -rw-r--r-- | oslo_db/sqlalchemy/utils.py | 21 | ||||
| -rw-r--r-- | oslo_db/tests/sqlalchemy/test_utils.py | 25 |
2 files changed, 37 insertions, 9 deletions
diff --git a/oslo_db/sqlalchemy/utils.py b/oslo_db/sqlalchemy/utils.py index 016bae7..c5030a0 100644 --- a/oslo_db/sqlalchemy/utils.py +++ b/oslo_db/sqlalchemy/utils.py @@ -80,18 +80,22 @@ def _get_unique_keys(model): except exc.NoInspectionAvailable: return None else: - table = mapper.mapped_table - if table is None: + local_table = mapper.local_table + base_table = mapper.base_mapper.local_table + + if local_table is None: return None # extract result from cache if present - info = table.info - if 'oslodb_unique_keys' in info: - return info['oslodb_unique_keys'] + has_info = hasattr(local_table, 'info') + if has_info: + info = local_table.info + if 'oslodb_unique_keys' in info: + return info['oslodb_unique_keys'] res = [] try: - constraints = table.constraints + constraints = base_table.constraints except AttributeError: constraints = [] for constraint in constraints: @@ -100,14 +104,15 @@ def _get_unique_keys(model): sqlalchemy.PrimaryKeyConstraint)): res.append({c.name for c in constraint.columns}) try: - indexes = table.indexes + indexes = base_table.indexes except AttributeError: indexes = [] for index in indexes: if index.unique: res.append({c.name for c in index.columns}) # cache result for next calls with the same model - info['oslodb_unique_keys'] = res + if has_info: + info['oslodb_unique_keys'] = res return res diff --git a/oslo_db/tests/sqlalchemy/test_utils.py b/oslo_db/tests/sqlalchemy/test_utils.py index 9911b12..31686fe 100644 --- a/oslo_db/tests/sqlalchemy/test_utils.py +++ b/oslo_db/tests/sqlalchemy/test_utils.py @@ -94,6 +94,16 @@ class FakeTable(Base): 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' @@ -328,6 +338,16 @@ class Test_UnstableSortingOrder(test_base.BaseTestCase): 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'])) @@ -378,7 +398,10 @@ class TestGetUniqueKeys(test_base.BaseTestCase): pass table = CacheTable() - mock_inspect = mock.Mock(return_value=mock.Mock(mapped_table=table)) + 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): |