Fujitsu Neutron Plugin
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.
 
 

287 lines
11 KiB

  1. # Copyright 2015-2016 FUJITSU LIMITED
  2. #
  3. # Licensed under the Apache License, Version 2.0 (the "License");
  4. # you may not use this file except in compliance with the License.
  5. # You may obtain a copy of the License at
  6. #
  7. # http://www.apache.org/licenses/LICENSE-2.0
  8. #
  9. # Unless required by applicable law or agreed to in writing, software
  10. # distributed under the License is distributed on an "AS IS" BASIS,
  11. # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
  12. # implied.
  13. # See the License for the specific language governing permissions and
  14. # limitations under the License.
  15. import os
  16. import mock
  17. from neutron.conf.plugins.ml2 import config
  18. from neutron.plugins.ml2.common import exceptions as ml2_exc
  19. from neutron_lib.api.definitions import portbindings as pb_def
  20. from networking_fujitsu.ml2.cfab import cfabdriver
  21. from networking_fujitsu.ml2.cfab import mech_cfab
  22. from networking_fujitsu.tests.unit.ml2.common import helper
  23. USERNAME = 'admin'
  24. PASSWORD = 'admin'
  25. ADDRESS = '192.168.100.1'
  26. PHYSICAL_NET = ["physnet1:1"]
  27. class TestFujitsuMechDriverV2(helper.FujitsuMechanismHelper):
  28. """Test Fujitsu mechanism driver.
  29. If the environment variable OS_FUJITSU_CFAB_ADDRESS is defined, tests will
  30. be performed using the C-Fabric driver, and a telnet connection to the
  31. address OS_FUJITSU_CFAB_ADDRESS and the port OS_FUJITSU_CFAB_PORT (defaults
  32. to 23) will be established. The username will be taken from
  33. OS_FUJITSU_CFAB_USERNAME (defaults to "admin") and the password will be
  34. taken from OS_FUJITSU_CFAB_PASSWORD (defaults to "password").
  35. If the environment variable OS_FUJITSU_CFAB_ADDRESS is NOT defined, tests
  36. will be performed using the mock driver instead of the C-Fabric driver, and
  37. no real telnet connection will be used.
  38. """
  39. _mechanism_drivers = ['fujitsu_cfab']
  40. # test_create_router_port_and_fail_create_postcommit:
  41. # This one is mocked 'fake_driver' only. So, our plugin's instance
  42. # hasn't mocked and fail definitely. Therefore, skips this test.
  43. _skip = ["test_create_router_port_and_fail_create_postcommit"]
  44. def setUp(self):
  45. config.cfg.CONF.set_override(
  46. 'tenant_network_types', ['vlan'], 'ml2')
  47. config.cfg.CONF.set_override(
  48. 'type_drivers', ['vlan'], 'ml2')
  49. if self._testMethodName in self._skip:
  50. self.skipTest("This test has already verified at neutron's test.")
  51. address = os.environ.get('OS_FUJITSU_CFAB_ADDRESS')
  52. if address:
  53. ml2_fujitsu_opts = {
  54. 'username': os.environ.get('OS_FUJITSU_CFAB_USERNAME') or
  55. "admin",
  56. 'password': os.environ.get('OS_FUJITSU_CFAB_PASSWORD') or
  57. "password",
  58. 'address': address,
  59. 'physical_networks': ["physnet1:1", "physnet2:2"],
  60. 'pprofile_prefix': "test-"}
  61. for opt, val in ml2_fujitsu_opts.items():
  62. config.cfg.CONF.set_override(opt, val, "fujitsu_cfab")
  63. try:
  64. cfabdriver.TELNET_PORT = int(
  65. os.environ['OS_FUJITSU_CFAB_PORT'])
  66. except KeyError:
  67. pass
  68. super(TestFujitsuMechDriverV2, self).setUp()
  69. else:
  70. def mocked_initialize(self):
  71. self._switch = {'address': "", 'username': "", 'password': ""}
  72. self._driver = mock.MagicMock()
  73. self._physical_networks = {'physnet1': "1", 'physnet2': "2"}
  74. with mock.patch.object(mech_cfab.CFABMechanismDriver,
  75. 'initialize', new=mocked_initialize):
  76. super(TestFujitsuMechDriverV2, self).setUp()
  77. class TestMechCFABIsSupported(TestFujitsuMechDriverV2):
  78. def test_is_supported_type_vlan(self):
  79. ctx = self.prepare_dummy_context('network')
  80. network = ctx.current
  81. self.assertTrue(mech_cfab.is_supported(network))
  82. def test_is_supported_not_support_type_flat(self):
  83. ctx = self.prepare_dummy_context('network', net_type='flat')
  84. network = ctx.current
  85. self.assertFalse(mech_cfab.is_supported(network))
  86. def test_is_supported_not_support_type_vxlan(self):
  87. ctx = self.prepare_dummy_context('network', net_type='vxlan')
  88. network = ctx.current
  89. self.assertFalse(mech_cfab.is_supported(network))
  90. def test_is_supported_illegal_seg_id_is_none(self):
  91. ctx = self.prepare_dummy_context('network', net_type='vlan')
  92. network = ctx.current
  93. ctx.current['provider:segmentation_id'] = None
  94. self.assertFalse(mech_cfab.is_supported(network))
  95. def test_is_supported_illegal_seg_id_is_not_defined(self):
  96. ctx = self.prepare_dummy_context('network', net_type='vlan')
  97. network = ctx.current
  98. del network['provider:segmentation_id']
  99. self.assertFalse(mech_cfab.is_supported(network))
  100. class TestFujitsuMechDriverBaremetalPortsV2(helper.FujitsuMechanismHelper):
  101. def setUp(self):
  102. ml2_fujitsu_opts = {
  103. 'username': USERNAME,
  104. 'password': PASSWORD,
  105. 'address': ADDRESS,
  106. 'physical_networks': PHYSICAL_NET,
  107. 'pprofile_prefix': 'test-'
  108. }
  109. for opt, val in ml2_fujitsu_opts.items():
  110. config.cfg.CONF.set_override(opt, val, "fujitsu_cfab")
  111. self.mech = mech_cfab.CFABMechanismDriver()
  112. self.mech._driver = mock.Mock()
  113. super(TestFujitsuMechDriverBaremetalPortsV2, self).setUp()
  114. def test_create_port(self):
  115. ctx = self.prepare_dummy_context()
  116. self.mech.create_port_postcommit(ctx)
  117. self.mech._driver.associate_mac_to_network.assert_not_called()
  118. def test_bind_port_with_single(self):
  119. ctx = self.prepare_dummy_context()
  120. self.mech.bind_port(ctx)
  121. params = params_for_setup_vlan(ctx.current)
  122. self.mech._driver.setup_vlan.assert_called_once_with(
  123. params['address'],
  124. params['username'],
  125. params['password'],
  126. params['vfab_id'],
  127. params['vlanid'],
  128. params['ports'],
  129. params['mac_address'],
  130. )
  131. def test_bind_port_with_lag(self):
  132. ctx = self.prepare_dummy_context(nic='lag')
  133. self.mech.bind_port(ctx)
  134. params = params_for_setup_vlan(ctx.current, lag=True)
  135. self.mech._driver.setup_vlan_with_lag.assert_called_once_with(
  136. params['address'],
  137. params['username'],
  138. params['password'],
  139. params['vfab_id'],
  140. params['vlanid'],
  141. params['ports'],
  142. params['mac_address'],
  143. )
  144. def test_bind_port_with_mlag(self):
  145. ctx = self.prepare_dummy_context(nic='mlag')
  146. self.mech.bind_port(ctx)
  147. params = params_for_setup_vlan(ctx.current, lag=True)
  148. self.mech._driver.setup_vlan_with_lag.assert_called_once_with(
  149. params['address'],
  150. params['username'],
  151. params['password'],
  152. params['vfab_id'],
  153. params['vlanid'],
  154. params['ports'],
  155. params['mac_address'],
  156. )
  157. def test_update_port(self):
  158. ctx = self.prepare_dummy_context(set_original=True)
  159. ctx.current[pb_def.VIF_TYPE] = 'other'
  160. self.mech.update_port_postcommit(ctx)
  161. self.mech._driver.associate_mac_to_network.assert_not_called()
  162. self.mech._driver.clear_vlan.assert_not_called()
  163. def test_update_port_with_unbound(self):
  164. ctx = self.prepare_dummy_context(set_original=True)
  165. params = params_for_setup_vlan(ctx.original)
  166. self.mech.update_port_postcommit(ctx)
  167. self.mech._driver.associate_mac_to_network.assert_not_called()
  168. self.mech._driver.clear_vlan.assert_called_with(
  169. params['address'],
  170. params['username'],
  171. params['password'],
  172. params['vfab_id'],
  173. params['vlanid'],
  174. params['ports'],
  175. params['mac_address'],
  176. )
  177. def test_update_port_network_type_is_not_vlan(self):
  178. ctx = self.prepare_dummy_context(set_original=True, net_type='vxlan')
  179. self.mech.update_port_postcommit(ctx)
  180. self.mech._driver.associate_mac_to_network.assert_not_called()
  181. self.mech._driver.clear_vlan.assert_not_called()
  182. def test_update_port_and_raises_in_clear_vlan(self):
  183. ctx = self.prepare_dummy_context(set_original=True)
  184. self.mech._driver.clear_vlan.side_effect = Exception
  185. self.assertRaises(ml2_exc.MechanismDriverError,
  186. self.mech.update_port_postcommit, ctx)
  187. def test_delete_port_with_single(self):
  188. ctx = self.prepare_dummy_context(nic='single')
  189. self.mech.delete_port_postcommit(ctx)
  190. params = params_for_setup_vlan(ctx.current)
  191. self.mech._driver.clear_vlan.assert_called_with(
  192. params['address'],
  193. params['username'],
  194. params['password'],
  195. params['vfab_id'],
  196. params['vlanid'],
  197. params['ports'],
  198. params['mac_address'],
  199. )
  200. self.mech._driver.dissociate_mac_from_network.assert_not_called()
  201. def test_delete_port_without_local_link_information(self):
  202. ctx = self.prepare_dummy_context(nic='single')
  203. ctx.current['binding:profile'] = {}
  204. self.mech.delete_port_postcommit(ctx)
  205. self.mech._driver.clear_vlan.assert_not_called()
  206. self.mech._driver.dissociate_mac_from_network.assert_not_called()
  207. def test_delete_with_lag(self):
  208. ctx = self.prepare_dummy_context(nic='lag')
  209. self.mech.delete_port_postcommit(ctx)
  210. params = params_for_setup_vlan(ctx.current, lag=True)
  211. self.mech._driver.clear_vlan_with_lag.assert_called_with(
  212. params['address'],
  213. params['username'],
  214. params['password'],
  215. params['vfab_id'],
  216. params['vlanid'],
  217. params['ports'],
  218. params['mac_address'],
  219. )
  220. self.mech._driver.dissociate_mac_from_network.assert_not_called()
  221. def test_delete_with_mlag(self):
  222. ctx = self.prepare_dummy_context(nic='mlag')
  223. self.mech.delete_port_postcommit(ctx)
  224. params = params_for_setup_vlan(ctx.current, lag=True)
  225. self.mech._driver.clear_vlan_with_lag.assert_called_with(
  226. params['address'],
  227. params['username'],
  228. params['password'],
  229. params['vfab_id'],
  230. params['vlanid'],
  231. params['ports'],
  232. params['mac_address'],
  233. )
  234. self.mech._driver.dissociate_mac_from_network.assert_not_called()
  235. def params_for_setup_vlan(port, lag=False):
  236. lli = port[pb_def.PROFILE]['local_link_information']
  237. return {
  238. 'address': ADDRESS,
  239. 'username': USERNAME,
  240. 'password': PASSWORD,
  241. 'vfab_id': '1',
  242. 'vlanid': 1111,
  243. 'ports': ','.join(p['port_id'] for p in lli),
  244. 'mac_address': port['mac_address'],
  245. 'lag': lag
  246. }