Coverage for manila/tests/network/linux/test_interface.py: 96%

1# Copyright 2014 Mirantis Inc. 

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 

16from unittest import mock 

17 

18from manila.network.linux import interface 

19from manila.network.linux import ip_lib 

20from manila import test 

21from manila.tests import conf_fixture 

22from manila.tests import fake_network 

23from manila import utils 

24 

25 

26class BaseChild(interface.LinuxInterfaceDriver): 

27 def plug(self, *args): 

28 pass 

29 

30 def unplug(self, *args): 

31 pass 

32 

33 

34FakeSubnet = { 

35 'cidr': '192.168.1.1/24', 

36} 

37 

38 

39FakeAllocation = { 

40 'subnet': FakeSubnet, 

41 'ip_address': '192.168.1.2', 

42 'ip_version': 4, 

43} 

44 

45 

46FakePort = { 

47 'id': 'abcdef01-1234-5678-90ab-ba0987654321', 

48 'fixed_ips': [FakeAllocation], 

49 'device_id': 'cccccccc-cccc-cccc-cccc-cccccccccccc', 

50} 

51 

52 

53class TestBase(test.TestCase): 

54 def setUp(self): 

55 super(TestBase, self).setUp() 

56 self.conf = conf_fixture.CONF 

57 self.conf.register_opts(interface.OPTS) 

58 self.ip_dev_p = mock.patch.object(ip_lib, 'IPDevice') 

59 self.ip_dev = self.ip_dev_p.start() 

60 self.ip_p = mock.patch.object(ip_lib, 'IPWrapper') 

61 self.ip = self.ip_p.start() 

62 self.device_exists_p = mock.patch.object(ip_lib, 'device_exists') 

63 self.device_exists = self.device_exists_p.start() 

64 self.addCleanup(self.ip_dev_p.stop) 

65 self.addCleanup(self.ip_p.stop) 

66 self.addCleanup(self.device_exists_p.stop) 

67 

68 

69class TestABCDriver(TestBase): 

70 

71 def test_verify_abs_class_has_abs_methods(self): 

72 

73 class ICanNotBeInstancetiated(interface.LinuxInterfaceDriver): 

74 pass 

75 

76 try: 

77 # pylint: disable=abstract-class-instantiated 

78 ICanNotBeInstancetiated() 

79 except TypeError: 

80 pass 

81 except Exception as e: 

82 self.fail("Unexpected exception thrown: '%s'" % e) 

83 else: 

84 self.fail("ExpectedException 'TypeError' not thrown.") 

85 

86 def test_get_device_name(self): 

87 bc = BaseChild() 

88 device_name = bc.get_device_name(FakePort) 

89 self.assertEqual('tapabcdef01-12', device_name) 

90 

91 def test_l3_init(self): 

92 addresses = [dict(ip_version=4, scope='global', 

93 dynamic=False, cidr='172.16.77.240/24')] 

94 self.ip_dev().addr.list = mock.Mock(return_value=addresses) 

95 

96 bc = BaseChild() 

97 self.mock_object(bc, '_remove_outdated_interfaces') 

98 

99 ns = '12345678-1234-5678-90ab-ba0987654321' 

100 bc.init_l3('tap0', ['192.168.1.2/24'], namespace=ns, 

101 clear_cidrs=['192.168.0.0/16']) 

102 self.ip_dev.assert_has_calls( 

103 [mock.call('tap0', namespace=ns), 

104 mock.call().route.clear_outdated_routes('192.168.0.0/16'), 

105 mock.call().addr.list(scope='global', filters=['permanent']), 

106 mock.call().addr.add(4, '192.168.1.2/24', '192.168.1.255'), 

107 mock.call().addr.delete(4, '172.16.77.240/24'), 

108 mock.call().route.pullup_route('tap0')]) 

109 bc._remove_outdated_interfaces.assert_called_with(self.ip_dev()) 

110 

111 def test__remove_outdated_interfaces(self): 

112 device = fake_network.FakeDevice( 

113 'foobarquuz', [dict(ip_version=4, cidr='1.0.0.0/27')]) 

114 devices = [fake_network.FakeDevice('foobar')] 

115 self.ip().get_devices = mock.Mock(return_value=devices) 

116 

117 bc = BaseChild() 

118 self.mock_object(bc, 'unplug') 

119 

120 bc._remove_outdated_interfaces(device) 

121 bc.unplug.assert_called_once_with('foobar') 

122 

123 def test__get_set_of_device_cidrs(self): 

124 device = fake_network.FakeDevice('foo') 

125 expected = set(('1.0.0.0/27', '2.0.0.0/27')) 

126 

127 bc = BaseChild() 

128 result = bc._get_set_of_device_cidrs(device) 

129 

130 self.assertEqual(expected, result) 

131 

132 def test__get_set_of_device_cidrs_exception(self): 

133 device = fake_network.FakeDevice('foo') 

134 self.mock_object(device.addr, 'list', mock.Mock( 

135 side_effect=Exception('foo does not exist'))) 

136 

137 bc = BaseChild() 

138 result = bc._get_set_of_device_cidrs(device) 

139 

140 self.assertEqual(set(), result) 

141 

142 

143class TestNoopInterfaceDriver(TestBase): 

144 

145 def test_init_l3(self): 

146 self.ip.assert_not_called() 

147 self.ip_dev.assert_not_called() 

148 

149 def test_plug(self): 

150 self.ip.assert_not_called() 

151 self.ip_dev.assert_not_called() 

152 

153 def test_unplug(self): 

154 self.ip.assert_not_called() 

155 self.ip_dev.assert_not_called() 

156 

157 

158class TestOVSInterfaceDriver(TestBase): 

159 

160 def test_get_device_name(self): 

161 br = interface.OVSInterfaceDriver() 

162 device_name = br.get_device_name(FakePort) 

163 self.assertEqual('tapabcdef01-12', device_name) 

164 

165 def test_plug_no_ns(self): 

166 self._test_plug() 

167 

168 def test_plug_with_ns(self): 

169 self._test_plug(namespace='01234567-1234-1234-99') 

170 

171 def test_plug_alt_bridge(self): 

172 self._test_plug(bridge='br-foo') 

173 

174 def _test_plug(self, additional_expectation=None, bridge=None, 

175 namespace=None): 

176 if additional_expectation is None: 176 ↛ 178line 176 didn't jump to line 178 because the condition on line 176 was always true

177 additional_expectation = [] 

178 if not bridge: 

179 bridge = 'br-int' 

180 

181 def device_exists(dev, namespace=None): 

182 return dev == bridge 

183 

184 vsctl_cmd = ['ovs-vsctl', '--', '--may-exist', 'add-port', 

185 bridge, 'tap0', '--', 'set', 'Interface', 'tap0', 

186 'type=internal', '--', 'set', 'Interface', 'tap0', 

187 'external-ids:iface-id=port-1234', '--', 'set', 

188 'Interface', 'tap0', 

189 'external-ids:iface-status=active', '--', 'set', 

190 'Interface', 'tap0', 

191 'external-ids:attached-mac=aa:bb:cc:dd:ee:ff'] 

192 

193 with mock.patch.object(utils, 'execute') as execute: 

194 ovs = interface.OVSInterfaceDriver() 

195 self.device_exists.side_effect = device_exists 

196 ovs.plug('tap0', 

197 'port-1234', 

198 'aa:bb:cc:dd:ee:ff', 

199 bridge=bridge, 

200 namespace=namespace) 

201 execute.assert_called_once_with(*vsctl_cmd, run_as_root=True) 

202 

203 expected = [mock.call(), 

204 mock.call().device('tap0'), 

205 mock.call().device().link.set_address('aa:bb:cc:dd:ee:ff')] 

206 expected.extend(additional_expectation) 

207 if namespace: 

208 expected.extend( 

209 [mock.call().ensure_namespace(namespace), 

210 mock.call().ensure_namespace().add_device_to_namespace( 

211 mock.ANY)]) 

212 expected.extend([mock.call().device().link.set_up()]) 

213 

214 self.ip.assert_has_calls(expected) 

215 

216 def test_plug_reset_mac(self): 

217 fake_mac_addr = 'aa:bb:cc:dd:ee:ff' 

218 self.device_exists.return_value = True 

219 

220 self.ip().device().link.address = mock.Mock(return_value=fake_mac_addr) 

221 ovs = interface.OVSInterfaceDriver() 

222 ovs.plug('tap0', 

223 'port-1234', 

224 'ff:ee:dd:cc:bb:aa', 

225 bridge='br-int') 

226 expected = [mock.call(), 

227 mock.call().device('tap0'), 

228 mock.call().device().link.set_address('ff:ee:dd:cc:bb:aa'), 

229 mock.call().device().link.set_up()] 

230 self.ip.assert_has_calls(expected) 

231 

232 def test_unplug(self, bridge=None): 

233 if not bridge: 233 ↛ 235line 233 didn't jump to line 235 because the condition on line 233 was always true

234 bridge = 'br-int' 

235 with mock.patch('manila.network.linux.ovs_lib.OVSBridge') as ovs_br: 

236 ovs = interface.OVSInterfaceDriver() 

237 ovs.unplug('tap0') 

238 ovs_br.assert_has_calls([mock.call(bridge), 

239 mock.call().delete_port('tap0')]) 

240 

241 

242class TestBridgeInterfaceDriver(TestBase): 

243 def test_get_device_name(self): 

244 br = interface.BridgeInterfaceDriver() 

245 device_name = br.get_device_name(FakePort) 

246 self.assertEqual('ns-abcdef01-12', device_name) 

247 

248 def test_plug_no_ns(self): 

249 self._test_plug() 

250 

251 def test_plug_with_ns(self): 

252 self._test_plug(namespace='01234567-1234-1234-99') 

253 

254 def _test_plug(self, namespace=None, mtu=None): 

255 def device_exists(device, root_helper=None, namespace=None): 

256 return device.startswith('brq') 

257 

258 root_veth = mock.Mock() 

259 ns_veth = mock.Mock() 

260 

261 self.ip().add_veth = mock.Mock(return_value=(root_veth, ns_veth)) 

262 

263 self.device_exists.side_effect = device_exists 

264 br = interface.BridgeInterfaceDriver() 

265 mac_address = 'aa:bb:cc:dd:ee:ff' 

266 br.plug('ns-0', 

267 'port-1234', 

268 mac_address, 

269 namespace=namespace) 

270 

271 ip_calls = [mock.call(), 

272 mock.call().add_veth('tap0', 'ns-0', namespace2=namespace)] 

273 ns_veth.assert_has_calls([mock.call.link.set_address(mac_address)]) 

274 

275 self.ip.assert_has_calls(ip_calls) 

276 

277 root_veth.assert_has_calls([mock.call.link.set_up()]) 

278 ns_veth.assert_has_calls([mock.call.link.set_up()]) 

279 

280 def test_plug_dev_exists(self): 

281 self.device_exists.return_value = True 

282 with mock.patch('manila.network.linux.interface.LOG.warning') as log: 

283 br = interface.BridgeInterfaceDriver() 

284 br.plug('port-1234', 

285 'tap0', 

286 'aa:bb:cc:dd:ee:ff') 

287 self.ip_dev.assert_has_calls([]) 

288 self.assertEqual(1, log.call_count) 

289 

290 def test_unplug_no_device(self): 

291 self.device_exists.return_value = False 

292 self.ip_dev().link.delete.side_effect = RuntimeError 

293 with mock.patch('manila.network.linux.interface.LOG') as log: 

294 br = interface.BridgeInterfaceDriver() 

295 br.unplug('tap0') 

296 [mock.call(), mock.call('tap0'), mock.call().link.delete()] 

297 self.assertEqual(1, log.error.call_count) 

298 

299 def test_unplug(self): 

300 self.device_exists.return_value = True 

301 with mock.patch('manila.network.linux.interface.LOG.debug') as log: 

302 br = interface.BridgeInterfaceDriver() 

303 br.unplug('tap0') 

304 self.assertTrue(log.called) 

305 self.ip_dev.assert_has_calls([mock.call('tap0', None), 

306 mock.call().link.delete()])