Coverage for manila/tests/network/linux/test_ip_lib.py: 100%

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 ip_lib 

19from manila import test 

20 

21NETNS_SAMPLE = [ 

22 '12345678-1234-5678-abcd-1234567890ab', 

23 'bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb', 

24 'cccccccc-cccc-cccc-cccc-cccccccccccc'] 

25 

26LINK_SAMPLE = [ 

27 '1: lo: <LOOPBACK,UP,LOWER_UP> mtu 16436 qdisc noqueue state UNKNOWN \\' 

28 'link/loopback 00:00:00:00:00:00 brd 00:00:00:00:00:00', 

29 '2: eth0: <BROADCAST,MULTICAST,UP,LOWER_UP> mtu 1500 qdisc mq state UP ' 

30 'qlen 1000\\ link/ether cc:dd:ee:ff:ab:cd brd ff:ff:ff:ff:ff:ff' 

31 '\\ alias openvswitch', 

32 '3: br-int: <BROADCAST,MULTICAST> mtu 1500 qdisc noop state DOWN ' 

33 '\\ link/ether aa:bb:cc:dd:ee:ff brd ff:ff:ff:ff:ff:ff', 

34 '4: gw-ddc717df-49: <BROADCAST,MULTICAST> mtu 1500 qdisc noop ' 

35 'state DOWN \\ link/ether fe:dc:ba:fe:dc:ba brd ff:ff:ff:ff:ff:ff', 

36 '5: eth0.50@eth0: <BROADCAST,MULTICAST,UP,LOWER_UP> mtu 1500 qdisc ' 

37 ' noqueue master brq0b24798c-07 state UP mode DEFAULT' 

38 '\\ link/ether ab:04:49:b6:ab:a0 brd ff:ff:ff:ff:ff:ff'] 

39 

40ADDR_SAMPLE = (""" 

412: eth0: <BROADCAST,MULTICAST,UP,LOWER_UP> mtu 1500 qdisc mq state UP qlen 1000 

42 link/ether dd:cc:aa:b9:76:ce brd ff:ff:ff:ff:ff:ff 

43 inet 172.16.77.240/24 brd 172.16.77.255 scope global eth0 

44 inet6 2001:470:9:1224:5595:dd51:6ba2:e788/64 scope global temporary dynamic 

45 valid_lft 14187sec preferred_lft 3387sec 

46 inet6 2001:470:9:1224:fd91:272:581e:3a32/64 scope global temporary """ 

47 """deprecated dynamic 

48 valid_lft 14187sec preferred_lft 0sec 

49 inet6 2001:470:9:1224:4508:b885:5fb:740b/64 scope global temporary """ 

50 """deprecated dynamic 

51 valid_lft 14187sec preferred_lft 0sec 

52 inet6 2001:470:9:1224:dfcc:aaff:feb9:76ce/64 scope global dynamic 

53 valid_lft 14187sec preferred_lft 3387sec 

54 inet6 fe80::dfcc:aaff:feb9:76ce/64 scope link 

55 valid_lft forever preferred_lft forever 

56""") 

57 

58ADDR_SAMPLE2 = (""" 

592: eth0: <BROADCAST,MULTICAST,UP,LOWER_UP> mtu 1500 qdisc mq state UP qlen 1000 

60 link/ether dd:cc:aa:b9:76:ce brd ff:ff:ff:ff:ff:ff 

61 inet 172.16.77.240/24 scope global eth0 

62 inet6 2001:470:9:1224:5595:dd51:6ba2:e788/64 scope global temporary dynamic 

63 valid_lft 14187sec preferred_lft 3387sec 

64 inet6 2001:470:9:1224:fd91:272:581e:3a32/64 scope global temporary """ 

65 """deprecated dynamic 

66 valid_lft 14187sec preferred_lft 0sec 

67 inet6 2001:470:9:1224:4508:b885:5fb:740b/64 scope global temporary """ 

68 """deprecated dynamic 

69 valid_lft 14187sec preferred_lft 0sec 

70 inet6 2001:470:9:1224:dfcc:aaff:feb9:76ce/64 scope global dynamic 

71 valid_lft 14187sec preferred_lft 3387sec 

72 inet6 fe80::dfcc:aaff:feb9:76ce/64 scope link 

73 valid_lft forever preferred_lft forever 

74""") 

75 

76GATEWAY_SAMPLE1 = (""" 

77default via 10.35.19.254 metric 100 

7810.35.16.0/22 proto kernel scope link src 10.35.17.97 

79""") 

80 

81GATEWAY_SAMPLE2 = (""" 

82default via 10.35.19.254 metric 100 

83""") 

84 

85GATEWAY_SAMPLE3 = (""" 

8610.35.16.0/22 proto kernel scope link src 10.35.17.97 

87""") 

88 

89GATEWAY_SAMPLE4 = (""" 

90default via 10.35.19.254 

91""") 

92 

93GATEWAY_SAMPLE5 = (""" 

94default via 172.24.47.1 dev eth0 

9510.0.0.0/24 dev tapc226b810-a0 proto kernel scope link src 10.0.0.3 

9610.254.0.0/28 dev tap6de90453-1c proto kernel scope link src 10.254.0.4 

9710.35.16.0/22 proto kernel scope link src 10.35.17.97 

98172.24.4.0/24 via 10.35.19.254 metric 100 

99""") 

100 

101DEVICE_ROUTE_SAMPLE = ("10.0.0.0/24 scope link src 10.0.0.2") 

102 

103SUBNET_SAMPLE1 = ("10.0.0.0/24 dev qr-23380d11-d2 scope link src 10.0.0.1\n" 

104 "10.0.0.0/24 dev tap1d7888a7-10 scope link src 10.0.0.2") 

105SUBNET_SAMPLE2 = ("10.0.0.0/24 dev tap1d7888a7-10 scope link src 10.0.0.2\n" 

106 "10.0.0.0/24 dev qr-23380d11-d2 scope link src 10.0.0.1") 

107 

108 

109class TestSubProcessBase(test.TestCase): 

110 def setUp(self): 

111 super(TestSubProcessBase, self).setUp() 

112 self.execute_p = mock.patch('manila.utils.execute') 

113 self.execute = self.execute_p.start() 

114 

115 def tearDown(self): 

116 self.execute_p.stop() 

117 super(TestSubProcessBase, self).tearDown() 

118 

119 def test_execute_wrapper(self): 

120 ip_lib.SubProcessBase._execute('o', 'link', ('list',)) 

121 

122 self.execute.assert_called_once_with('ip', '-o', 'link', 'list', 

123 run_as_root=False) 

124 

125 def test_execute_wrapper_int_options(self): 

126 ip_lib.SubProcessBase._execute([4], 'link', ('list',)) 

127 

128 self.execute.assert_called_once_with('ip', '-4', 'link', 'list', 

129 run_as_root=False) 

130 

131 def test_execute_wrapper_no_options(self): 

132 ip_lib.SubProcessBase._execute([], 'link', ('list',)) 

133 

134 self.execute.assert_called_once_with('ip', 'link', 'list', 

135 run_as_root=False) 

136 

137 def test_run_no_namespace(self): 

138 base = ip_lib.SubProcessBase() 

139 base._run([], 'link', ('list',)) 

140 self.execute.assert_called_once_with('ip', 'link', 'list', 

141 run_as_root=False) 

142 

143 def test_run_namespace(self): 

144 base = ip_lib.SubProcessBase('ns') 

145 base._run([], 'link', ('list',)) 

146 self.execute.assert_called_once_with('ip', 'netns', 'exec', 'ns', 

147 'ip', 'link', 'list', 

148 run_as_root=True) 

149 

150 def test_as_root_namespace(self): 

151 base = ip_lib.SubProcessBase('ns') 

152 base._as_root([], 'link', ('list',)) 

153 self.execute.assert_called_once_with('ip', 'netns', 'exec', 'ns', 

154 'ip', 'link', 'list', 

155 run_as_root=True) 

156 

157 

158class TestIpWrapper(test.TestCase): 

159 def setUp(self): 

160 super(TestIpWrapper, self).setUp() 

161 self.execute_p = mock.patch.object(ip_lib.IPWrapper, '_execute') 

162 self.execute = self.execute_p.start() 

163 

164 def tearDown(self): 

165 self.execute_p.stop() 

166 super(TestIpWrapper, self).tearDown() 

167 

168 def test_get_devices(self): 

169 self.execute.return_value = '\n'.join(LINK_SAMPLE) 

170 retval = ip_lib.IPWrapper().get_devices() 

171 self.assertEqual([ip_lib.IPDevice('lo'), 

172 ip_lib.IPDevice('eth0'), 

173 ip_lib.IPDevice('br-int'), 

174 ip_lib.IPDevice('gw-ddc717df-49'), 

175 ip_lib.IPDevice('eth0.50')], 

176 retval) 

177 

178 self.execute.assert_called_once_with('o', 'link', ('list',), None) 

179 

180 def test_get_devices_malformed_line(self): 

181 self.execute.return_value = '\n'.join(LINK_SAMPLE + ['gibberish']) 

182 retval = ip_lib.IPWrapper().get_devices() 

183 self.assertEqual([ip_lib.IPDevice('lo'), 

184 ip_lib.IPDevice('eth0'), 

185 ip_lib.IPDevice('br-int'), 

186 ip_lib.IPDevice('gw-ddc717df-49'), 

187 ip_lib.IPDevice('eth0.50')], 

188 retval) 

189 

190 self.execute.assert_called_once_with('o', 'link', ('list',), None) 

191 

192 def test_get_namespaces(self): 

193 self.execute.return_value = '\n'.join(NETNS_SAMPLE) 

194 retval = ip_lib.IPWrapper.get_namespaces() 

195 self.assertEqual(['12345678-1234-5678-abcd-1234567890ab', 

196 'bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb', 

197 'cccccccc-cccc-cccc-cccc-cccccccccccc'], 

198 retval) 

199 

200 self.execute.assert_called_once_with('', 'netns', ('list',)) 

201 

202 def test_add_tuntap(self): 

203 ip_lib.IPWrapper().add_tuntap('tap0') 

204 self.execute.assert_called_once_with('', 'tuntap', 

205 ('add', 'tap0', 'mode', 'tap'), 

206 None, as_root=True) 

207 

208 def test_add_veth(self): 

209 ip_lib.IPWrapper().add_veth('tap0', 'tap1') 

210 self.execute.assert_called_once_with('', 'link', 

211 ('add', 'tap0', 'type', 'veth', 

212 'peer', 'name', 'tap1'), 

213 None, as_root=True) 

214 

215 def test_add_veth_with_namespaces(self): 

216 ns2 = 'ns2' 

217 with mock.patch.object(ip_lib.IPWrapper, 'ensure_namespace') as en: 

218 ip_lib.IPWrapper().add_veth('tap0', 'tap1', namespace2=ns2) 

219 en.assert_has_calls([mock.call(ns2)]) 

220 self.execute.assert_called_once_with('', 'link', 

221 ('add', 'tap0', 'type', 'veth', 

222 'peer', 'name', 'tap1', 

223 'netns', ns2), 

224 None, as_root=True) 

225 

226 def test_get_device(self): 

227 dev = ip_lib.IPWrapper('ns').device('eth0') 

228 self.assertEqual('ns', dev.namespace) 

229 self.assertEqual('eth0', dev.name) 

230 

231 def test_ensure_namespace(self): 

232 with mock.patch.object(ip_lib, 'IPDevice') as ip_dev: 

233 ip = ip_lib.IPWrapper() 

234 with mock.patch.object(ip.netns, 'exists') as ns_exists: 

235 ns_exists.return_value = False 

236 ip.ensure_namespace('ns') 

237 self.execute.assert_has_calls( 

238 [mock.call([], 'netns', ('add', 'ns'), None, 

239 as_root=True)]) 

240 ip_dev.assert_has_calls([mock.call('lo', 'ns'), 

241 mock.call().link.set_up()]) 

242 

243 def test_ensure_namespace_existing(self): 

244 with mock.patch.object(ip_lib, 'IpNetnsCommand') as ip_ns_cmd: 

245 ip_ns_cmd.exists.return_value = True 

246 ns = ip_lib.IPWrapper().ensure_namespace('ns') 

247 self.assertFalse(self.execute.called) 

248 self.assertEqual('ns', ns.namespace) 

249 

250 def test_namespace_is_empty_no_devices(self): 

251 ip = ip_lib.IPWrapper('ns') 

252 with mock.patch.object(ip, 'get_devices') as get_devices: 

253 get_devices.return_value = [] 

254 

255 self.assertTrue(ip.namespace_is_empty()) 

256 get_devices.assert_called_once_with(exclude_loopback=True) 

257 

258 def test_namespace_is_empty(self): 

259 ip = ip_lib.IPWrapper('ns') 

260 with mock.patch.object(ip, 'get_devices') as get_devices: 

261 get_devices.return_value = [mock.Mock()] 

262 

263 self.assertFalse(ip.namespace_is_empty()) 

264 get_devices.assert_called_once_with(exclude_loopback=True) 

265 

266 def test_garbage_collect_namespace_does_not_exist(self): 

267 with mock.patch.object(ip_lib, 'IpNetnsCommand') as ip_ns_cmd_cls: 

268 ip_ns_cmd_cls.return_value.exists.return_value = False 

269 ip = ip_lib.IPWrapper('ns') 

270 with mock.patch.object(ip, 'namespace_is_empty') as mock_is_empty: 

271 self.assertFalse(ip.garbage_collect_namespace()) 

272 ip_ns_cmd_cls.assert_has_calls([mock.call().exists('ns')]) 

273 self.assertNotIn(mock.call().delete('ns'), 

274 ip_ns_cmd_cls.return_value.mock_calls) 

275 self.assertEqual([], mock_is_empty.mock_calls) 

276 

277 def test_garbage_collect_namespace_existing_empty_ns(self): 

278 with mock.patch.object(ip_lib, 'IpNetnsCommand') as ip_ns_cmd_cls: 

279 ip_ns_cmd_cls.return_value.exists.return_value = True 

280 

281 ip = ip_lib.IPWrapper('ns') 

282 

283 with mock.patch.object(ip, 'namespace_is_empty') as mock_is_empty: 

284 mock_is_empty.return_value = True 

285 self.assertTrue(ip.garbage_collect_namespace()) 

286 

287 mock_is_empty.assert_called_once_with() 

288 expected = [mock.call().exists('ns'), 

289 mock.call().delete('ns')] 

290 ip_ns_cmd_cls.assert_has_calls(expected) 

291 

292 def test_garbage_collect_namespace_existing_not_empty(self): 

293 lo_device = mock.Mock() 

294 lo_device.name = 'lo' 

295 tap_device = mock.Mock() 

296 tap_device.name = 'tap1' 

297 

298 with mock.patch.object(ip_lib, 'IpNetnsCommand') as ip_ns_cmd_cls: 

299 ip_ns_cmd_cls.return_value.exists.return_value = True 

300 

301 ip = ip_lib.IPWrapper('ns') 

302 

303 with mock.patch.object(ip, 'namespace_is_empty') as mock_is_empty: 

304 mock_is_empty.return_value = False 

305 

306 self.assertFalse(ip.garbage_collect_namespace()) 

307 

308 mock_is_empty.assert_called_once_with() 

309 expected = [mock.call(ip), 

310 mock.call().exists('ns')] 

311 self.assertEqual(expected, ip_ns_cmd_cls.mock_calls) 

312 self.assertNotIn(mock.call().delete('ns'), 

313 ip_ns_cmd_cls.mock_calls) 

314 

315 def test_add_device_to_namespace(self): 

316 dev = mock.Mock() 

317 ip_lib.IPWrapper('ns').add_device_to_namespace(dev) 

318 dev.assert_has_calls([mock.call.link.set_netns('ns')]) 

319 

320 def test_add_device_to_namespace_is_none(self): 

321 dev = mock.Mock() 

322 ip_lib.IPWrapper().add_device_to_namespace(dev) 

323 self.assertEqual([], dev.mock_calls) 

324 

325 

326class TestIPDevice(test.TestCase): 

327 def test_eq_same_name(self): 

328 dev1 = ip_lib.IPDevice('tap0') 

329 dev2 = ip_lib.IPDevice('tap0') 

330 self.assertEqual(dev1, dev2) 

331 

332 def test_eq_diff_name(self): 

333 dev1 = ip_lib.IPDevice('tap0') 

334 dev2 = ip_lib.IPDevice('tap1') 

335 self.assertNotEqual(dev1, dev2) 

336 

337 def test_eq_same_namespace(self): 

338 dev1 = ip_lib.IPDevice('tap0', 'ns1') 

339 dev2 = ip_lib.IPDevice('tap0', 'ns1') 

340 self.assertEqual(dev1, dev2) 

341 

342 def test_eq_diff_namespace(self): 

343 dev1 = ip_lib.IPDevice('tap0', 'ns1') 

344 dev2 = ip_lib.IPDevice('tap0', 'ns2') 

345 self.assertNotEqual(dev1, dev2) 

346 

347 def test_eq_other_is_none(self): 

348 dev1 = ip_lib.IPDevice('tap0', 'ns1') 

349 self.assertIsNotNone(dev1) 

350 

351 def test_str(self): 

352 self.assertEqual('tap0', str(ip_lib.IPDevice('tap0'))) 

353 

354 

355class TestIPCommandBase(test.TestCase): 

356 def setUp(self): 

357 super(TestIPCommandBase, self).setUp() 

358 self.ip = mock.Mock() 

359 self.ip.namespace = 'namespace' 

360 self.ip_cmd = ip_lib.IpCommandBase(self.ip) 

361 self.ip_cmd.COMMAND = 'foo' 

362 

363 def test_run(self): 

364 self.ip_cmd._run('link', 'show') 

365 self.ip.assert_has_calls([mock.call._run([], 'foo', ('link', 'show'))]) 

366 

367 def test_run_with_options(self): 

368 self.ip_cmd._run('link', options='o') 

369 self.ip.assert_has_calls([mock.call._run('o', 'foo', ('link', ))]) 

370 

371 def test_as_root(self): 

372 self.ip_cmd._as_root('link') 

373 self.ip.assert_has_calls( 

374 [mock.call._as_root([], 'foo', ('link', ), False)]) 

375 

376 def test_as_root_with_options(self): 

377 self.ip_cmd._as_root('link', options='o') 

378 self.ip.assert_has_calls( 

379 [mock.call._as_root('o', 'foo', ('link', ), False)]) 

380 

381 

382class TestIPDeviceCommandBase(test.TestCase): 

383 def setUp(self): 

384 super(TestIPDeviceCommandBase, self).setUp() 

385 self.ip_dev = mock.Mock() 

386 self.ip_dev.name = 'eth0' 

387 self.ip_dev._execute = mock.Mock(return_value='executed') 

388 self.ip_cmd = ip_lib.IpDeviceCommandBase(self.ip_dev) 

389 self.ip_cmd.COMMAND = 'foo' 

390 

391 def test_name_property(self): 

392 self.assertEqual('eth0', self.ip_cmd.name) 

393 

394 

395class TestIPCmdBase(test.TestCase): 

396 def setUp(self): 

397 super(TestIPCmdBase, self).setUp() 

398 self.parent = mock.Mock() 

399 self.parent.name = 'eth0' 

400 

401 def _assert_call(self, options, args): 

402 self.parent.assert_has_calls([ 

403 mock.call._run(options, self.command, args)]) 

404 

405 def _assert_sudo(self, options, args, force_root_namespace=False): 

406 self.parent.assert_has_calls( 

407 [mock.call._as_root(options, self.command, args, 

408 force_root_namespace)]) 

409 

410 

411class TestIpLinkCommand(TestIPCmdBase): 

412 def setUp(self): 

413 super(TestIpLinkCommand, self).setUp() 

414 self.parent._run.return_value = LINK_SAMPLE[1] 

415 self.command = 'link' 

416 self.link_cmd = ip_lib.IpLinkCommand(self.parent) 

417 

418 def test_set_address(self): 

419 self.link_cmd.set_address('aa:bb:cc:dd:ee:ff') 

420 self._assert_sudo([], ('set', 'eth0', 'address', 'aa:bb:cc:dd:ee:ff')) 

421 

422 def test_set_mtu(self): 

423 self.link_cmd.set_mtu(1500) 

424 self._assert_sudo([], ('set', 'eth0', 'mtu', 1500)) 

425 

426 def test_set_up(self): 

427 self.link_cmd.set_up() 

428 self._assert_sudo([], ('set', 'eth0', 'up')) 

429 

430 def test_set_down(self): 

431 self.link_cmd.set_down() 

432 self._assert_sudo([], ('set', 'eth0', 'down')) 

433 

434 def test_set_netns(self): 

435 self.link_cmd.set_netns('foo') 

436 self._assert_sudo([], ('set', 'eth0', 'netns', 'foo')) 

437 self.assertEqual('foo', self.parent.namespace) 

438 

439 def test_set_name(self): 

440 self.link_cmd.set_name('tap1') 

441 self._assert_sudo([], ('set', 'eth0', 'name', 'tap1')) 

442 self.assertEqual('tap1', self.parent.name) 

443 

444 def test_set_alias(self): 

445 self.link_cmd.set_alias('openvswitch') 

446 self._assert_sudo([], ('set', 'eth0', 'alias', 'openvswitch')) 

447 

448 def test_delete(self): 

449 self.link_cmd.delete() 

450 self._assert_sudo([], ('delete', 'eth0')) 

451 

452 def test_address_property(self): 

453 self.parent._execute = mock.Mock(return_value=LINK_SAMPLE[1]) 

454 self.assertEqual('cc:dd:ee:ff:ab:cd', self.link_cmd.address) 

455 

456 def test_mtu_property(self): 

457 self.parent._execute = mock.Mock(return_value=LINK_SAMPLE[1]) 

458 self.assertEqual(1500, self.link_cmd.mtu) 

459 

460 def test_qdisc_property(self): 

461 self.parent._execute = mock.Mock(return_value=LINK_SAMPLE[1]) 

462 self.assertEqual('mq', self.link_cmd.qdisc) 

463 

464 def test_qlen_property(self): 

465 self.parent._execute = mock.Mock(return_value=LINK_SAMPLE[1]) 

466 self.assertEqual(1000, self.link_cmd.qlen) 

467 

468 def test_alias_property(self): 

469 self.parent._execute = mock.Mock(return_value=LINK_SAMPLE[1]) 

470 self.assertEqual('openvswitch', self.link_cmd.alias) 

471 

472 def test_state_property(self): 

473 self.parent._execute = mock.Mock(return_value=LINK_SAMPLE[1]) 

474 self.assertEqual('UP', self.link_cmd.state) 

475 

476 def test_settings_property(self): 

477 expected = {'mtu': 1500, 

478 'qlen': 1000, 

479 'state': 'UP', 

480 'qdisc': 'mq', 

481 'brd': 'ff:ff:ff:ff:ff:ff', 

482 'link/ether': 'cc:dd:ee:ff:ab:cd', 

483 'alias': 'openvswitch'} 

484 self.parent._execute = mock.Mock(return_value=LINK_SAMPLE[1]) 

485 self.assertEqual(expected, self.link_cmd.attributes) 

486 self._assert_call('o', ('show', 'eth0')) 

487 

488 

489class TestIpAddrCommand(TestIPCmdBase): 

490 def setUp(self): 

491 super(TestIpAddrCommand, self).setUp() 

492 self.parent.name = 'tap0' 

493 self.command = 'addr' 

494 self.addr_cmd = ip_lib.IpAddrCommand(self.parent) 

495 

496 def test_add_address(self): 

497 self.addr_cmd.add(4, '192.168.45.100/24', '192.168.45.255') 

498 self._assert_sudo([4], 

499 ('add', '192.168.45.100/24', 'brd', '192.168.45.255', 

500 'scope', 'global', 'dev', 'tap0')) 

501 

502 def test_add_address_scoped(self): 

503 self.addr_cmd.add(4, '192.168.45.100/24', '192.168.45.255', 

504 scope='link') 

505 self._assert_sudo([4], 

506 ('add', '192.168.45.100/24', 'brd', '192.168.45.255', 

507 'scope', 'link', 'dev', 'tap0')) 

508 

509 def test_del_address(self): 

510 self.addr_cmd.delete(4, '192.168.45.100/24') 

511 self._assert_sudo([4], 

512 ('del', '192.168.45.100/24', 'dev', 'tap0')) 

513 

514 def test_flush(self): 

515 self.addr_cmd.flush() 

516 self._assert_sudo([], ('flush', 'tap0')) 

517 

518 def test_list(self): 

519 expected = [ 

520 dict(ip_version=4, scope='global', 

521 dynamic=False, cidr='172.16.77.240/24', 

522 broadcast='172.16.77.255'), 

523 dict(ip_version=6, scope='global', 

524 dynamic=True, cidr='2001:470:9:1224:5595:dd51:6ba2:e788/64', 

525 broadcast='::'), 

526 dict(ip_version=6, scope='global', 

527 dynamic=True, cidr='2001:470:9:1224:fd91:272:581e:3a32/64', 

528 broadcast='::'), 

529 dict(ip_version=6, scope='global', 

530 dynamic=True, cidr='2001:470:9:1224:4508:b885:5fb:740b/64', 

531 broadcast='::'), 

532 dict(ip_version=6, scope='global', 

533 dynamic=True, cidr='2001:470:9:1224:dfcc:aaff:feb9:76ce/64', 

534 broadcast='::'), 

535 dict(ip_version=6, scope='link', 

536 dynamic=False, cidr='fe80::dfcc:aaff:feb9:76ce/64', 

537 broadcast='::')] 

538 

539 test_cases = [ADDR_SAMPLE, ADDR_SAMPLE2] 

540 

541 for test_case in test_cases: 

542 self.parent._run = mock.Mock(return_value=test_case) 

543 self.assertEqual(expected, self.addr_cmd.list()) 

544 self._assert_call([], ('show', 'tap0')) 

545 

546 def test_list_filtered(self): 

547 expected = [ 

548 dict(ip_version=4, scope='global', 

549 dynamic=False, cidr='172.16.77.240/24', 

550 broadcast='172.16.77.255')] 

551 

552 test_cases = [ADDR_SAMPLE, ADDR_SAMPLE2] 

553 

554 for test_case in test_cases: 

555 output = '\n'.join(test_case.split('\n')[0:4]) 

556 self.parent._run.return_value = output 

557 self.assertEqual(expected, self.addr_cmd.list('global', 

558 filters=['permanent'])) 

559 self._assert_call([], ('show', 'tap0', 'permanent', 'scope', 

560 'global')) 

561 

562 

563class TestIpRouteCommand(TestIPCmdBase): 

564 def setUp(self): 

565 super(TestIpRouteCommand, self).setUp() 

566 self.parent.name = 'eth0' 

567 self.command = 'route' 

568 self.route_cmd = ip_lib.IpRouteCommand(self.parent) 

569 

570 def test_add_gateway(self): 

571 gateway = '192.168.45.100' 

572 metric = 100 

573 self.route_cmd.add_gateway(gateway, metric) 

574 self._assert_sudo([], 

575 ('replace', 'default', 'via', gateway, 

576 'metric', metric, 

577 'dev', self.parent.name)) 

578 

579 def test_del_gateway(self): 

580 gateway = '192.168.45.100' 

581 self.route_cmd.delete_gateway(gateway) 

582 self._assert_sudo([], 

583 ('del', 'default', 'via', gateway, 

584 'dev', self.parent.name)) 

585 

586 def test_get_gateway(self): 

587 test_cases = [{'sample': GATEWAY_SAMPLE1, 

588 'expected': {'gateway': '10.35.19.254', 

589 'metric': 100}}, 

590 {'sample': GATEWAY_SAMPLE2, 

591 'expected': {'gateway': '10.35.19.254', 

592 'metric': 100}}, 

593 {'sample': GATEWAY_SAMPLE3, 

594 'expected': None}, 

595 {'sample': GATEWAY_SAMPLE4, 

596 'expected': {'gateway': '10.35.19.254'}}] 

597 for test_case in test_cases: 

598 self.parent._run = mock.Mock(return_value=test_case['sample']) 

599 self.assertEqual(test_case['expected'], 

600 self.route_cmd.get_gateway()) 

601 

602 def test_pullup_route(self): 

603 # interface is not the first in the list - requires 

604 # deleting and creating existing entries 

605 output = [DEVICE_ROUTE_SAMPLE, SUBNET_SAMPLE1] 

606 

607 def pullup_side_effect(self, *args): 

608 result = output.pop(0) 

609 return result 

610 

611 self.parent._run = mock.Mock(side_effect=pullup_side_effect) 

612 self.route_cmd.pullup_route('tap1d7888a7-10') 

613 self._assert_sudo([], ('del', '10.0.0.0/24', 'dev', 'qr-23380d11-d2')) 

614 self._assert_sudo([], ('append', '10.0.0.0/24', 'proto', 'kernel', 

615 'src', '10.0.0.1', 'dev', 'qr-23380d11-d2')) 

616 

617 def test_pullup_route_first(self): 

618 # interface is first in the list - no changes 

619 output = [DEVICE_ROUTE_SAMPLE, SUBNET_SAMPLE2] 

620 

621 def pullup_side_effect(self, *args): 

622 result = output.pop(0) 

623 return result 

624 

625 self.parent._run = mock.Mock(side_effect=pullup_side_effect) 

626 self.route_cmd.pullup_route('tap1d7888a7-10') 

627 # Check two calls - device get and subnet get 

628 self.assertEqual(2, len(self.parent._run.mock_calls)) 

629 

630 def test_list(self): 

631 self.route_cmd._as_root = mock.Mock(return_value=GATEWAY_SAMPLE5) 

632 expected = [{'Destination': 'default', 

633 'Device': 'eth0', 

634 'Gateway': '172.24.47.1'}, 

635 {'Destination': '10.0.0.0/24', 

636 'Device': 'tapc226b810-a0'}, 

637 {'Destination': '10.254.0.0/28', 

638 'Device': 'tap6de90453-1c'}, 

639 {'Destination': '10.35.16.0/22'}, 

640 {'Destination': '172.24.4.0/24', 

641 'Gateway': '10.35.19.254'}] 

642 result = self.route_cmd.list() 

643 self.assertEqual(expected, result) 

644 self.route_cmd._as_root.assert_called_once_with('list') 

645 

646 def test_delete_net_route(self): 

647 self.route_cmd._as_root = mock.Mock() 

648 self.route_cmd.delete_net_route('10.0.0.0/24', 'br-ex') 

649 self.route_cmd._as_root.assert_called_once_with( 

650 'delete', '10.0.0.0/24', 'dev', 'br-ex') 

651 

652 def test_clear_outdated_routes(self): 

653 self.route_cmd.delete_net_route = mock.Mock() 

654 list_result = [{'Destination': 'default', 

655 'Device': 'eth0', 

656 'Gateway': '172.24.47.1'}, 

657 {'Destination': '10.0.0.0/24', 

658 'Device': 'eth0'}, 

659 {'Destination': '10.0.0.0/24', 

660 'Device': 'br-ex'}] 

661 self.route_cmd.list = mock.Mock(return_value=list_result) 

662 self.route_cmd.clear_outdated_routes('10.0.0.0/24') 

663 self.route_cmd.delete_net_route.assert_called_once_with( 

664 '10.0.0.0/24', 'br-ex') 

665 

666 

667class TestIpNetnsCommand(TestIPCmdBase): 

668 def setUp(self): 

669 super(TestIpNetnsCommand, self).setUp() 

670 self.command = 'netns' 

671 self.netns_cmd = ip_lib.IpNetnsCommand(self.parent) 

672 

673 def test_add_namespace(self): 

674 ns = self.netns_cmd.add('ns') 

675 self._assert_sudo([], ('add', 'ns'), force_root_namespace=True) 

676 self.assertEqual('ns', ns.namespace) 

677 

678 def test_delete_namespace(self): 

679 with mock.patch('manila.utils.execute'): 

680 self.netns_cmd.delete('ns') 

681 self._assert_sudo([], ('delete', 'ns'), force_root_namespace=True) 

682 

683 def test_namespace_exists(self): 

684 retval = '\n'.join(NETNS_SAMPLE) 

685 self.parent._as_root.return_value = retval 

686 self.assertTrue( 

687 self.netns_cmd.exists('bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb')) 

688 self._assert_sudo('o', ('list',), force_root_namespace=True) 

689 

690 def test_namespace_doest_not_exist(self): 

691 retval = '\n'.join(NETNS_SAMPLE) 

692 self.parent._as_root.return_value = retval 

693 self.assertFalse( 

694 self.netns_cmd.exists('bbbbbbbb-1111-2222-3333-bbbbbbbbbbbb')) 

695 self._assert_sudo('o', ('list',), force_root_namespace=True) 

696 

697 def test_execute(self): 

698 self.parent.namespace = 'ns' 

699 with mock.patch('manila.utils.execute') as execute: 

700 self.netns_cmd.execute(['ip', 'link', 'list']) 

701 execute.assert_called_once_with('ip', 'netns', 'exec', 'ns', 'ip', 

702 'link', 'list', 

703 run_as_root=True, 

704 check_exit_code=True) 

705 

706 def test_execute_env_var_prepend(self): 

707 self.parent.namespace = 'ns' 

708 with mock.patch('manila.utils.execute') as execute: 

709 env = dict(FOO=1, BAR=2) 

710 self.netns_cmd.execute(['ip', 'link', 'list'], env) 

711 execute.assert_called_once_with(