OpenStack Image Management (Glance)
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

test_location_strategy.py 8.3KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185
  1. # Copyright 2014 IBM Corp.
  2. # All Rights Reserved.
  3. #
  4. # Licensed under the Apache License, Version 2.0 (the "License"); you may
  5. # not use this file except in compliance with the License. You may obtain
  6. # a copy of the License at
  7. #
  8. # http://www.apache.org/licenses/LICENSE-2.0
  9. #
  10. # Unless required by applicable law or agreed to in writing, software
  11. # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
  12. # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  13. # License for the specific language governing permissions and limitations
  14. # under the License.
  15. import copy
  16. import stevedore
  17. from glance.common import location_strategy
  18. from glance.common.location_strategy import location_order
  19. from glance.common.location_strategy import store_type
  20. from glance.tests.unit import base
  21. class TestLocationStrategy(base.IsolatedUnitTest):
  22. """Test routines in glance.common.location_strategy"""
  23. def _set_original_strategies(self, original_strategies):
  24. for name in location_strategy._available_strategies.keys():
  25. if name not in original_strategies:
  26. del location_strategy._available_strategies[name]
  27. def setUp(self):
  28. super(TestLocationStrategy, self).setUp()
  29. original_strategies = ['location_order', 'store_type']
  30. self.addCleanup(self._set_original_strategies, original_strategies)
  31. def test_load_strategy_modules(self):
  32. modules = location_strategy._load_strategies()
  33. # By default we have two built-in strategy modules.
  34. self.assertEqual(2, len(modules))
  35. self.assertEqual(set(['location_order', 'store_type']),
  36. set(modules.keys()))
  37. self.assertEqual(location_strategy._available_strategies, modules)
  38. def test_load_strategy_module_with_deduplicating(self):
  39. modules = ['module1', 'module2']
  40. def _fake_stevedore_extension_manager(*args, **kwargs):
  41. ret = lambda: None
  42. ret.names = lambda: modules
  43. return ret
  44. def _fake_stevedore_driver_manager(*args, **kwargs):
  45. ret = lambda: None
  46. ret.driver = lambda: None
  47. ret.driver.__name__ = kwargs['name']
  48. # Module 1 and 2 has a same strategy name
  49. ret.driver.get_strategy_name = lambda: 'module_name'
  50. ret.driver.init = lambda: None
  51. return ret
  52. self.stub = self.stubs.Set(stevedore.extension, "ExtensionManager",
  53. _fake_stevedore_extension_manager)
  54. self.stub = self.stubs.Set(stevedore.driver, "DriverManager",
  55. _fake_stevedore_driver_manager)
  56. loaded_modules = location_strategy._load_strategies()
  57. self.assertEqual(1, len(loaded_modules))
  58. self.assertIn('module_name', loaded_modules)
  59. # Skipped module #2, duplicated one.
  60. self.assertEqual('module1', loaded_modules['module_name'].__name__)
  61. def test_load_strategy_module_with_init_exception(self):
  62. modules = ['module_init_exception', 'module_good']
  63. def _fake_stevedore_extension_manager(*args, **kwargs):
  64. ret = lambda: None
  65. ret.names = lambda: modules
  66. return ret
  67. def _fake_stevedore_driver_manager(*args, **kwargs):
  68. if kwargs['name'] == 'module_init_exception':
  69. raise Exception('strategy module failed to initialize.')
  70. else:
  71. ret = lambda: None
  72. ret.driver = lambda: None
  73. ret.driver.__name__ = kwargs['name']
  74. ret.driver.get_strategy_name = lambda: kwargs['name']
  75. ret.driver.init = lambda: None
  76. return ret
  77. self.stub = self.stubs.Set(stevedore.extension, "ExtensionManager",
  78. _fake_stevedore_extension_manager)
  79. self.stub = self.stubs.Set(stevedore.driver, "DriverManager",
  80. _fake_stevedore_driver_manager)
  81. loaded_modules = location_strategy._load_strategies()
  82. self.assertEqual(1, len(loaded_modules))
  83. self.assertIn('module_good', loaded_modules)
  84. # Skipped module #1, initialize failed one.
  85. self.assertEqual('module_good', loaded_modules['module_good'].__name__)
  86. def test_verify_valid_location_strategy(self):
  87. for strategy_name in ['location_order', 'store_type']:
  88. self.config(location_strategy=strategy_name)
  89. location_strategy.verify_location_strategy()
  90. def test_get_ordered_locations_with_none_or_empty_locations(self):
  91. self.assertEqual([], location_strategy.get_ordered_locations(None))
  92. self.assertEqual([], location_strategy.get_ordered_locations([]))
  93. def test_get_ordered_locations(self):
  94. self.config(location_strategy='location_order')
  95. original_locs = [{'url': 'loc1'}, {'url': 'loc2'}]
  96. ordered_locs = location_strategy.get_ordered_locations(original_locs)
  97. # Original location list should remain unchanged
  98. self.assertNotEqual(id(original_locs), id(ordered_locs))
  99. self.assertEqual(original_locs, ordered_locs)
  100. def test_choose_best_location_with_none_or_empty_locations(self):
  101. self.assertIsNone(location_strategy.choose_best_location(None))
  102. self.assertIsNone(location_strategy.choose_best_location([]))
  103. def test_choose_best_location(self):
  104. self.config(location_strategy='location_order')
  105. original_locs = [{'url': 'loc1'}, {'url': 'loc2'}]
  106. best_loc = location_strategy.choose_best_location(original_locs)
  107. # Deep copy protect original location.
  108. self.assertNotEqual(id(original_locs), id(best_loc))
  109. self.assertEqual(original_locs[0], best_loc)
  110. class TestLocationOrderStrategyModule(base.IsolatedUnitTest):
  111. """Test routines in glance.common.location_strategy.location_order"""
  112. def test_get_ordered_locations(self):
  113. original_locs = [{'url': 'loc1'}, {'url': 'loc2'}]
  114. ordered_locs = location_order.get_ordered_locations(original_locs)
  115. # The result will ordered by original natural order.
  116. self.assertEqual(original_locs, ordered_locs)
  117. class TestStoreTypeStrategyModule(base.IsolatedUnitTest):
  118. """Test routines in glance.common.location_strategy.store_type"""
  119. def test_get_ordered_locations(self):
  120. self.config(store_type_preference=[' rbd', 'sheepdog ', ' file',
  121. 'swift ', ' http ', 'vmware'],
  122. group='store_type_location_strategy')
  123. locs = [{'url': 'file://image0', 'metadata': {'idx': 3}},
  124. {'url': 'rbd://image1', 'metadata': {'idx': 0}},
  125. {'url': 'file://image3', 'metadata': {'idx': 4}},
  126. {'url': 'swift://image4', 'metadata': {'idx': 6}},
  127. {'url': 'cinder://image5', 'metadata': {'idx': 9}},
  128. {'url': 'file://image6', 'metadata': {'idx': 5}},
  129. {'url': 'rbd://image7', 'metadata': {'idx': 1}},
  130. {'url': 'vsphere://image9', 'metadata': {'idx': 8}},
  131. {'url': 'sheepdog://image8', 'metadata': {'idx': 2}}]
  132. ordered_locs = store_type.get_ordered_locations(copy.deepcopy(locs))
  133. locs.sort(key=lambda loc: loc['metadata']['idx'])
  134. # The result will ordered by preferred store type order.
  135. self.assertEqual(locs, ordered_locs)
  136. def test_get_ordered_locations_with_invalid_store_name(self):
  137. self.config(store_type_preference=[' rbd', 'sheepdog ', 'invalid',
  138. 'swift ', ' http '],
  139. group='store_type_location_strategy')
  140. locs = [{'url': 'file://image0', 'metadata': {'idx': 4}},
  141. {'url': 'rbd://image1', 'metadata': {'idx': 0}},
  142. {'url': 'file://image3', 'metadata': {'idx': 5}},
  143. {'url': 'swift://image4', 'metadata': {'idx': 3}},
  144. {'url': 'cinder://image5', 'metadata': {'idx': 6}},
  145. {'url': 'file://image6', 'metadata': {'idx': 7}},
  146. {'url': 'rbd://image7', 'metadata': {'idx': 1}},
  147. {'url': 'sheepdog://image8', 'metadata': {'idx': 2}}]
  148. ordered_locs = store_type.get_ordered_locations(copy.deepcopy(locs))
  149. locs.sort(key=lambda loc: loc['metadata']['idx'])
  150. # The result will ordered by preferred store type order.
  151. self.assertEqual(locs, ordered_locs)