Add a version filtering strategy

The current patch proposes the addition of a new data source strategy
for filtering the strategies by a given list of versions.

Each data source exports a new API, called ``version()``, which should
return the underlying version of the said data source. This implies that
each data source module can export multiple versioned data sources and it's
the duty of an external strategy to select the appropiate data source with
regard to a particular set of versions.

Change-Id: I6ef2fdd6520a53fc600232247f23bd22b848461e
This commit is contained in:
Claudiu Popa 2015-09-01 13:19:32 +03:00
parent a6a6466035
commit dc3efec379
4 changed files with 62 additions and 0 deletions

View File

@ -140,6 +140,10 @@ class BaseDataSource(object):
This should return an instance of :class:`APIResponse`.
"""
@abc.abstractmethod
def version(self):
"""Get the version of the current data source."""
def instance_id(self):
"""Get this instance's id."""

View File

@ -50,6 +50,10 @@ class BaseOpenStackSource(base.BaseDataSource):
def _path_join(self, path, *addons):
"""Join one or more components together."""
def version(self):
"""Get the underlying data source version."""
return self._version
def _working_version(self):
versions = self._available_versions()
# OS_VERSIONS is stored in chronological order, so

View File

@ -77,3 +77,22 @@ class SerialSearchStrategy(BaseSearchStrategy):
for data_source in data_sources:
if self.is_datasource_available(data_source):
yield data_source
class FilterVersionStrategy(BaseSearchStrategy):
"""A strategy for filtering data sources by their version
:param versions:
A list of strings, where each strings is a possible
version that a data source can have.
"""
def __init__(self, versions=None):
if versions is None:
versions = []
self._versions = versions
super(FilterVersionStrategy, self).__init__()
def search_data_sources(self, data_sources):
return (source for source in data_sources
if source.version() in self._versions)

View File

@ -63,3 +63,38 @@ class TestStrategy(tests.TestCase):
valid_sources = list(instance.search_data_sources(sources))
self.assertEqual(available_sources, valid_sources)
def test_filter_version_strategy(self):
class SourceV1(object):
def version(self):
return 'first'
class SourceV2(SourceV1):
def version(self):
return 'second'
class SourceV3(object):
def version(self):
return 'third'
sources = [SourceV1(), SourceV2(), SourceV3()]
instance = strategy.FilterVersionStrategy(['third', 'first'])
filtered_sources = sorted(
source.version()
for source in instance.search_data_sources(sources))
self.assertEqual(len(filtered_sources), 2)
self.assertEqual(filtered_sources, ['first', 'third'])
def test_filter_version_strategy_no_versions_given(self):
class SourceV1(object):
def version(self):
return 'first'
sources = [SourceV1()]
instance = strategy.FilterVersionStrategy()
filtered_sources = list(instance.search_data_sources(sources))
self.assertEqual(len(filtered_sources), 0)