Coverage for manila/tests/network/neutron/test_neutron

5# Licensed under the Apache License, Version 2.0 (the "License"); you may

6# not use this file except in compliance with the License. You may obtain

7# a copy of the License at

9# http://www.apache.org/licenses/LICENSE-2.0

10#

11# Unless required by applicable law or agreed to in writing, software

12# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT

13# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the

14# License for the specific language governing permissions and limitations

15# under the License.

17from unittest import mock

19from neutronclient.common import exceptions as neutron_client_exc

20from neutronclient.v2_0 import client as clientv20

21from oslo_config import cfg

23from manila.db import base

24from manila import exception

25from manila.network.neutron import api as neutron_api

26from manila.network.neutron import constants as neutron_constants

27from manila import test

28from manila.tests.db import fakes

29from manila.tests import utils as test_utils

31CONF = cfg.CONF

34class FakeNeutronClient(object):

36 def create_port(self, body):

37 return body

39 def delete_port(self, port_id):

40 pass

42 def show_port(self, port_id):

43 pass

45 def list_ports(self, **search_opts):

46 pass

48 def create_port_binding(self, port_id, body):

49 return body

51 def delete_port_binding(self, port_id, host_id):

52 pass

54 def activate_port_binding(self, port_id, host_id):

55 pass

57 def list_networks(self):

58 pass

60 def show_network(self, network_uuid):

61 pass

63 def show_subnet(self, subnet_uuid):

64 pass

66 def create_router(self, body):

67 return body

69 def list_routers(self):

70 pass

72 def create_network(self, body):

73 return body

75 def create_subnet(self, body):

76 return body

78 def update_port(self, port_id, body):

79 return body

81 def add_interface_router(self, router_id, subnet_id, port_id):

82 pass

84 def update_router(self, router_id, body):

85 return body

87 def show_router(self, router_id):

88 pass

90 def list_extensions(self):

91 pass

94class NeutronclientTestCase(test.TestCase):

96 def test_no_auth_obj(self):

97 mock_client_loader = self.mock_object(

98 neutron_api.client_auth, 'AuthClientLoader')

99 fake_context = 'fake_context'

100 data = {

101 'neutron': {

102 'url': 'http://localhost:9696',

103 'endpoint_type': 'internalURL',

104 'region_name': 'foo_region_name',

105 }

106 }

107

108 self.client = None

109 with test_utils.create_temp_config_with_opts(data):

110 self.client = neutron_api.API()

111 self.client.get_client(fake_context)

112

113 mock_client_loader.assert_called_once_with(

114 client_class=neutron_api.clientv20.Client,

115 cfg_group=neutron_api.NEUTRON_GROUP

116 )

117 mock_client_loader.return_value.get_client.assert_called_once_with(

118 self.client,

119 fake_context,

120 endpoint_type=data['neutron']['endpoint_type'],

121 region_name=data['neutron']['region_name'],

122 endpoint_override=data['neutron']['url'],

123 )

124

125 def test_with_auth_obj(self):

126 fake_context = 'fake_context'

127 data = {

128 'neutron': {

129 'url': 'http://localhost:9696',

130 'endpoint_type': 'internalURL',

131 'region_name': 'foo_region_name',

132 }

133 }

134

135 self.client = None

136 with test_utils.create_temp_config_with_opts(data):

137 self.client = neutron_api.API()

138 self.client.auth_obj = type(

139 'FakeAuthObj', (object, ), {'get_client': mock.Mock()})

140 self.client.get_client(fake_context)

141

142 self.client.auth_obj.get_client.assert_called_once_with(

143 self.client,

144 fake_context,

145 endpoint_type=data['neutron']['endpoint_type'],

146 region_name=data['neutron']['region_name'],

147 endpoint_override=data['neutron']['url'],

148 )

149

150

151class NeutronApiTest(test.TestCase):

152

153 def setUp(self):

154 super(NeutronApiTest, self).setUp()

155 self.mock_object(base, 'Base', fakes.FakeModel)

156 self.mock_object(

157 clientv20, 'Client', mock.Mock(return_value=FakeNeutronClient()))

158 self.neutron_api = neutron_api.API()

159

160 def test_create_api_object(self):

161 # instantiate Neutron API object

162 neutron_api_instance = neutron_api.API()

163

164 # Verify results

165 self.assertTrue(hasattr(neutron_api_instance, 'client'))

166 self.assertTrue(hasattr(neutron_api_instance, 'configuration'))

167 self.assertEqual('DEFAULT', neutron_api_instance.config_group_name)

168

169 def test_create_port_with_all_args(self):

170 # Set up test data

171 self.mock_object(self.neutron_api, '_has_port_binding_extension',

172 mock.Mock(return_value=True))

173 port_args = {

174 'tenant_id': 'test tenant', 'network_id': 'test net',

175 'host_id': 'test host', 'subnet_id': 'test subnet',

176 'fixed_ip': 'test ip', 'device_owner': 'test owner',

177 'device_id': 'test device', 'mac_address': 'test mac',

178 'security_group_ids': 'test group',

179 'dhcp_opts': 'test dhcp',

180 }

181

182 # Execute method 'create_port'

183 port = self.neutron_api.create_port(**port_args)

184

185 # Verify results

186 self.assertEqual(port_args['tenant_id'], port['tenant_id'])

187 self.assertEqual(port_args['network_id'], port['network_id'])

188 self.assertEqual(port_args['host_id'], port['binding:host_id'])

189 self.assertEqual(port_args['subnet_id'],

190 port['fixed_ips'][0]['subnet_id'])

191 self.assertEqual(port_args['fixed_ip'],

192 port['fixed_ips'][0]['ip_address'])

193 self.assertEqual(port_args['device_owner'], port['device_owner'])

194 self.assertEqual(port_args['device_id'], port['device_id'])

195 self.assertEqual(port_args['mac_address'], port['mac_address'])

196 self.assertEqual(port_args['security_group_ids'],

197 port['security_groups'])

198 self.assertEqual(port_args['dhcp_opts'], port['extra_dhcp_opts'])

199 self.neutron_api._has_port_binding_extension.assert_called_once_with()

200 self.assertTrue(clientv20.Client.called)

201

202 def test_create_port_with_required_args(self):

203 # Set up test data

204 port_args = {'tenant_id': 'test tenant', 'network_id': 'test net'}

205

206 # Execute method 'create_port'

207 port = self.neutron_api.create_port(**port_args)

208

209 # Verify results

210 self.assertEqual(port_args['tenant_id'], port['tenant_id'])

211 self.assertEqual(port_args['network_id'],

212 port['network_id'])

213 self.assertTrue(clientv20.Client.called)

214

215 def test_create_port_with_additional_kwargs(self):

216 # Set up test data

217 port_args = {'tenant_id': 'test tenant', 'network_id': 'test net',

218 'binding_arg': 'foo'}

219

220 # Execute method 'create_port'

221 port = self.neutron_api.create_port(**port_args)

222

223 # Verify results

224 self.assertEqual(port_args['tenant_id'], port['tenant_id'])

225 self.assertEqual(port_args['network_id'],

226 port['network_id'])

227 self.assertEqual(port_args['binding_arg'],

228 port['binding_arg'])

229 self.assertTrue(clientv20.Client.called)

230

231 def test_create_port_with_host_id_no_binding_ext(self):

232 self.mock_object(self.neutron_api, '_has_port_binding_extension',

233 mock.Mock(return_value=False))

234 port_args = {

235 'tenant_id': 'test tenant',

236 'network_id': 'test net',

237 'host_id': 'foohost'

238 }

239

240 self.assertRaises(exception.NetworkException,

241 self.neutron_api.create_port, **port_args)

242

243 @mock.patch.object(neutron_api.LOG, 'exception', mock.Mock())

244 def test_create_port_exception(self):

245 self.mock_object(

246 self.neutron_api.client, 'create_port',

247 mock.Mock(side_effect=neutron_client_exc.NeutronClientException))

248 port_args = {'tenant_id': 'test tenant', 'network_id': 'test net'}

249

250 # Execute method 'create_port'

251 self.assertRaises(exception.NetworkException,

252 self.neutron_api.create_port,

253 **port_args)

254

255 # Verify results

256 self.assertTrue(neutron_api.LOG.exception.called)

257 self.assertTrue(clientv20.Client.called)

258 self.assertTrue(self.neutron_api.client.create_port.called)

259

260 @mock.patch.object(neutron_api.LOG, 'exception', mock.Mock())

261 def test_create_port_exception_status_409(self):

262 # Set up test data

263 self.mock_object(

264 self.neutron_api.client, 'create_port',

265 mock.Mock(side_effect=neutron_client_exc.NeutronClientException(

266 status_code=409)))

267 port_args = {'tenant_id': 'test tenant', 'network_id': 'test net'}

268

269 # Execute method 'create_port'

270 self.assertRaises(exception.PortLimitExceeded,

271 self.neutron_api.create_port,

272 **port_args)

273

274 # Verify results

275 self.assertTrue(neutron_api.LOG.exception.called)

276 self.assertTrue(clientv20.Client.called)

277 self.assertTrue(self.neutron_api.client.create_port.called)

278

279 def test_delete_port(self):

280 # Set up test data

281 self.mock_object(self.neutron_api.client, 'delete_port')

282 port_id = 'test port id'

283

284 # Execute method 'delete_port'

285 self.neutron_api.delete_port(port_id)

286

287 # Verify results

288 self.neutron_api.client.delete_port.assert_called_once_with(port_id)

289 self.assertTrue(clientv20.Client.called)

290

291 def test_delete_port_NeutronClientException(self):

292 # Set up test data

293 self.mock_object(

294 self.neutron_api.client, 'delete_port',

295 mock.Mock(side_effect=neutron_client_exc.NeutronClientException()))

296 port_id = 'test port id'

297

298 self.assertRaises(exception.NetworkException,

299 self.neutron_api.delete_port,

300 port_id)

301

302 # Verify results

303 self.neutron_api.client.delete_port.assert_called_once_with(port_id)

304 self.assertTrue(clientv20.Client.called)

305

306 def test_delete_port_PortNotFoundClient(self):

307 # Set up test data

308 self.mock_object(

309 self.neutron_api.client, 'delete_port',

310 mock.Mock(side_effect=neutron_client_exc.PortNotFoundClient()))

311 port_id = 'test port id'

312

313 # Execute method 'delete_port'

314 self.neutron_api.delete_port(port_id)

315

316 # Verify results

317 self.neutron_api.client.delete_port.assert_called_once_with(port_id)

318 self.assertTrue(clientv20.Client.called)

319

320 def test_list_ports(self):

321 # Set up test data

322 search_opts = {'test_option': 'test_value'}

323 fake_ports = [{'fake port': 'fake port info'}]

324 self.mock_object(

325 self.neutron_api.client, 'list_ports',

326 mock.Mock(return_value={'ports': fake_ports}))

327

328 # Execute method 'list_ports'

329 ports = self.neutron_api.list_ports(**search_opts)

330

331 # Verify results

332 self.assertEqual(fake_ports, ports)

333 self.assertTrue(clientv20.Client.called)

334 self.neutron_api.client.list_ports.assert_called_once_with(

335 **search_opts)

336

337 def test_show_port(self):

338 # Set up test data

339 port_id = 'test port id'

340 fake_port = {'fake port': 'fake port info'}

341 self.mock_object(

342 self.neutron_api.client, 'show_port',

343 mock.Mock(return_value={'port': fake_port}))

344

345 # Execute method 'show_port'

346 port = self.neutron_api.show_port(port_id)

347

348 # Verify results

349 self.assertEqual(fake_port, port)

350 self.assertTrue(clientv20.Client.called)

351 self.neutron_api.client.show_port.assert_called_once_with(port_id)

352

353 def test_get_network(self):

354 # Set up test data

355 network_id = 'test network id'

356 fake_network = {'fake network': 'fake network info'}

357 self.mock_object(

358 self.neutron_api.client, 'show_network',

359 mock.Mock(return_value={'network': fake_network}))

360

361 # Execute method 'get_network'

362 network = self.neutron_api.get_network(network_id)

363

364 # Verify results

365 self.assertEqual(fake_network, network)

366 self.assertTrue(clientv20.Client.called)

367 self.neutron_api.client.show_network.assert_called_once_with(

368 network_id)

369

370 def test_get_subnet(self):

371 # Set up test data

372 subnet_id = 'fake subnet id'

373 self.mock_object(

374 self.neutron_api.client, 'show_subnet',

375 mock.Mock(return_value={'subnet': {}}))

376

377 # Execute method 'get_subnet'

378 subnet = self.neutron_api.get_subnet(subnet_id)

379

380 # Verify results

381 self.assertEqual({}, subnet)

382 self.assertTrue(clientv20.Client.called)

383 self.neutron_api.client.show_subnet.assert_called_once_with(

384 subnet_id)

385

386 def test_get_all_network(self):

387 # Set up test data

388 fake_networks = [{'fake network': 'fake network info'}]

389 self.mock_object(

390 self.neutron_api.client, 'list_networks',

391 mock.Mock(return_value={'networks': fake_networks}))

392

393 # Execute method 'get_all_networks'

394 networks = self.neutron_api.get_all_networks()

395

396 # Verify results

397 self.assertEqual(fake_networks, networks)

398 self.assertTrue(clientv20.Client.called)

399 self.neutron_api.client.list_networks.assert_called_once_with()

400

401 def test_list_extensions(self):

402 # Set up test data

403 extensions = [

404 {'name': neutron_constants.PORTBINDING_EXT},

405 {'name': neutron_constants.PROVIDER_NW_EXT},

406 ]

407 self.mock_object(

408 self.neutron_api.client, 'list_extensions',

409 mock.Mock(return_value={'extensions': extensions}))

410

411 # Execute method 'list_extensions'

412 result = self.neutron_api.list_extensions()

413

414 # Verify results

415 self.assertTrue(clientv20.Client.called)

416 self.neutron_api.client.list_extensions.assert_called_once_with()

417 self.assertIn(neutron_constants.PORTBINDING_EXT, result)

418 self.assertIn(neutron_constants.PROVIDER_NW_EXT, result)

419 self.assertEqual(

420 extensions[0], result[neutron_constants.PORTBINDING_EXT])

421 self.assertEqual(

422 extensions[1], result[neutron_constants.PROVIDER_NW_EXT])

423

424 def test_create_network(self):

425 # Set up test data

426 net_args = {'tenant_id': 'test tenant', 'name': 'test name'}

427

428 # Execute method 'network_create'

429 network = self.neutron_api.network_create(**net_args)

430

431 # Verify results

432 self.assertEqual(net_args['tenant_id'], network['tenant_id'])

433 self.assertEqual(net_args['name'], network['name'])

434 self.assertTrue(clientv20.Client.called)

435

436 def test_create_subnet(self):

437 # Set up test data

438 subnet_args = {

439 'tenant_id': 'test tenant',

440 'name': 'test name',

441 'net_id': 'test net id',

442 'cidr': '10.0.0.0/24',

443 }

444

445 # Execute method 'subnet_create'

446 subnet = self.neutron_api.subnet_create(**subnet_args)

447

448 # Verify results

449 self.assertEqual(subnet_args['tenant_id'], subnet['tenant_id'])

450 self.assertEqual(subnet_args['name'], subnet['name'])

451 self.assertTrue(clientv20.Client.called)

452

453 def test_create_router(self):

454 # Set up test data

455 router_args = {'tenant_id': 'test tenant', 'name': 'test name'}

456

457 # Execute method 'router_create'

458 router = self.neutron_api.router_create(**router_args)

459

460 # Verify results

461 self.assertEqual(router_args['tenant_id'], router['tenant_id'])

462 self.assertEqual(router_args['name'], router['name'])

463 self.assertTrue(clientv20.Client.called)

464

465 def test_list_routers(self):

466 # Set up test data

467 fake_routers = [{'fake router': 'fake router info'}]

468 self.mock_object(

469 self.neutron_api.client, 'list_routers',

470 mock.Mock(return_value={'routers': fake_routers}))

471

472 # Execute method 'router_list'

473 networks = self.neutron_api.router_list()

474

475 # Verify results

476 self.assertEqual(fake_routers, networks)

477 self.assertTrue(clientv20.Client.called)

478 self.neutron_api.client.list_routers.assert_called_once_with()

479

480 def test_create_network_exception(self):

481 # Set up test data

482 net_args = {'tenant_id': 'test tenant', 'name': 'test name'}

483 self.mock_object(

484 self.neutron_api.client, 'create_network',

485 mock.Mock(side_effect=neutron_client_exc.NeutronClientException))

486

487 # Execute method 'network_create'

488 self.assertRaises(

489 exception.NetworkException,

490 self.neutron_api.network_create,

491 **net_args)

492

493 # Verify results

494 self.neutron_api.client.create_network.assert_called_once_with(

495 {'network': net_args})

496 self.assertTrue(clientv20.Client.called)

497

498 def test_create_subnet_exception(self):

499 # Set up test data

500 subnet_args = {

501 'tenant_id': 'test tenant',

502 'name': 'test name',

503 'net_id': 'test net id',

504 'cidr': '10.0.0.0/24',

505 }

506 self.mock_object(

507 self.neutron_api.client, 'create_subnet',

508 mock.Mock(side_effect=neutron_client_exc.NeutronClientException))

509

510 # Execute method 'subnet_create'

511 self.assertRaises(

512 exception.NetworkException,

513 self.neutron_api.subnet_create,

514 **subnet_args)

515

516 # Verify results

517 expected_data = {

518 'network_id': subnet_args['net_id'],

519 'tenant_id': subnet_args['tenant_id'],

520 'cidr': subnet_args['cidr'],

521 'name': subnet_args['name'],

522 'ip_version': 4,

523 }

524 self.neutron_api.client.create_subnet.assert_called_once_with(

525 {'subnet': expected_data})

526 self.assertTrue(clientv20.Client.called)

527

528 def test_create_router_exception(self):

529 # Set up test data

530 router_args = {'tenant_id': 'test tenant', 'name': 'test name'}

531 self.mock_object(

532 self.neutron_api.client, 'create_router',

533 mock.Mock(side_effect=neutron_client_exc.NeutronClientException))

534

535 # Execute method 'router_create'

536 self.assertRaises(

537 exception.NetworkException,

538 self.neutron_api.router_create,

539 **router_args)

540

541 # Verify results

542 self.neutron_api.client.create_router.assert_called_once_with(

543 {'router': router_args})

544 self.assertTrue(clientv20.Client.called)

545

546 def test_update_port_fixed_ips(self):

547 # Set up test data

548 port_id = 'test_port'

549 fixed_ips = {'fixed_ips': [{'subnet_id': 'test subnet'}]}

550

551 # Execute method 'update_port_fixed_ips'

552 port = self.neutron_api.update_port_fixed_ips(port_id, fixed_ips)

553

554 # Verify results

555 self.assertEqual(fixed_ips, port)

556 self.assertTrue(clientv20.Client.called)

557

558 def test_update_port_fixed_ips_exception(self):

559 # Set up test data

560 port_id = 'test_port'

561 fixed_ips = {'fixed_ips': [{'subnet_id': 'test subnet'}]}

562 self.mock_object(

563 self.neutron_api.client, 'update_port',

564 mock.Mock(side_effect=neutron_client_exc.NeutronClientException))

565

566 # Execute method 'update_port_fixed_ips'

567 self.assertRaises(

568 exception.NetworkException,

569 self.neutron_api.update_port_fixed_ips,

570 port_id, fixed_ips)

571

572 # Verify results

573 self.neutron_api.client.update_port.assert_called_once_with(

574 port_id, {'port': fixed_ips})

575 self.assertTrue(clientv20.Client.called)

576

577 def test_bind_port_to_host(self):

578 port_id = 'test_port'

579 host = 'test_host'

580 vnic_type = 'test_vnic_type'

581

582 port = self.neutron_api.bind_port_to_host(port_id, host, vnic_type)

583

584 self.assertEqual(host, port['host'])

585 self.assertTrue(clientv20.Client.called)

586

587 def test_delete_port_binding(self):

588 port_id = 'test_port'

589 host = 'test_host'

590 self.neutron_api.delete_port_binding(port_id, host)

591 self.assertTrue(clientv20.Client.called)

592

593 def test_activate_port_binding(self):

594 port_id = 'test_port'

595 host = 'test_host'

596

597 self.neutron_api.activate_port_binding(port_id, host)

598

599 self.assertTrue(clientv20.Client.called)

600

601 def test_router_update_routes(self):

602 # Set up test data

603 router_id = 'test_router'

604 routes = {

605 'routes': [

606 {'destination': '0.0.0.0/0', 'nexthop': '8.8.8.8', },

607 ],

608 }

609

610 # Execute method 'router_update_routes'

611 router = self.neutron_api.router_update_routes(router_id, routes)

612

613 # Verify results

614 self.assertEqual(routes, router)

615 self.assertTrue(clientv20.Client.called)

616

617 def test_router_update_routes_exception(self):

618 # Set up test data

619 router_id = 'test_router'

620 routes = {

621 'routes': [

622 {'destination': '0.0.0.0/0', 'nexthop': '8.8.8.8', },

623 ],

624 }

625 self.mock_object(

626 self.neutron_api.client, 'update_router',

627 mock.Mock(side_effect=neutron_client_exc.NeutronClientException))

628

629 # Execute method 'router_update_routes'

630 self.assertRaises(

631 exception.NetworkException,

632 self.neutron_api.router_update_routes,

633 router_id, routes)

634

635 # Verify results

636 self.neutron_api.client.update_router.assert_called_once_with(

637 router_id, {'router': routes})

638 self.assertTrue(clientv20.Client.called)

639

640 def test_show_router(self):

641 # Set up test data

642 router_id = 'test router id'

643 fake_router = {'fake router': 'fake router info'}

644 self.mock_object(

645 self.neutron_api.client, 'show_router',

646 mock.Mock(return_value={'router': fake_router}))

647

648 # Execute method 'show_router'

649 port = self.neutron_api.show_router(router_id)

650

651 # Verify results

652 self.assertEqual(fake_router, port)

653 self.assertTrue(clientv20.Client.called)

654 self.neutron_api.client.show_router.assert_called_once_with(router_id)

655

656 def test_router_add_interface(self):

657 # Set up test data

658 router_id = 'test port id'

659 subnet_id = 'test subnet id'

660 port_id = 'test port id'

661 self.mock_object(self.neutron_api.client, 'add_interface_router')

662

663 # Execute method 'router_add_interface'

664 self.neutron_api.router_add_interface(router_id, subnet_id, port_id)

665

666 # Verify results

667 self.neutron_api.client.add_interface_router.assert_called_once_with(

668 port_id, {'subnet_id': subnet_id, 'port_id': port_id})

669 self.assertTrue(clientv20.Client.called)

670

671 def test_router_add_interface_exception(self):

672 # Set up test data

673 router_id = 'test port id'

674 subnet_id = 'test subnet id'

675 port_id = 'test port id'

676 self.mock_object(

677 self.neutron_api.client, 'add_interface_router',

678 mock.Mock(side_effect=neutron_client_exc.NeutronClientException))

679

680 # Execute method 'router_add_interface'

681 self.assertRaises(

682 exception.NetworkException,

683 self.neutron_api.router_add_interface,

684 router_id, subnet_id, port_id)

685

686 # Verify results

687 self.neutron_api.client.add_interface_router.assert_called_once_with(

688 router_id, {'subnet_id': subnet_id, 'port_id': port_id})

689 self.assertTrue(clientv20.Client.called)

690

691 def test_admin_project_id_exist(self):

692 fake_admin_project_id = 'fake_admin_project_id_value'

693 self.neutron_api.client.httpclient = mock.Mock()

694 self.neutron_api.client.httpclient.auth_token = mock.Mock()

695 self.neutron_api.client.httpclient.get_project_id = mock.Mock(

696 return_value=fake_admin_project_id)

697

698 admin_project_id = self.neutron_api.admin_project_id

699

700 self.assertEqual(fake_admin_project_id, admin_project_id)

701 self.neutron_api.client.httpclient.auth_token.called

702

703 def test_admin_project_id_not_exist(self):

704 fake_admin_project_id = 'fake_admin_project_id_value'

705 self.neutron_api.client.httpclient = mock.Mock()

706 self.neutron_api.client.httpclient.auth_token = mock.Mock(

707 return_value=None)

708 self.neutron_api.client.httpclient.authenticate = mock.Mock()

709 self.neutron_api.client.httpclient.get_project_id = mock.Mock(

710 return_value=fake_admin_project_id)

711

712 admin_project_id = self.neutron_api.admin_project_id

713

714 self.assertEqual(fake_admin_project_id, admin_project_id)

715 self.neutron_api.client.httpclient.auth_token.called

716 self.neutron_api.client.httpclient.authenticate.called

717

718 def test_admin_project_id_not_exist_with_failure(self):

719 self.neutron_api.client.httpclient = mock.Mock()

720 self.neutron_api.client.httpclient.auth_token = None

721 self.neutron_api.client.httpclient.authenticate = mock.Mock(

722 side_effect=neutron_client_exc.NeutronClientException)

723 self.neutron_api.client.httpclient.auth_tenant_id = mock.Mock()

724

725 try:

726 self.neutron_api.admin_project_id

727 except exception.NetworkException:

728 pass

729 else:

730 raise Exception('Expected error was not raised')

731

732 self.assertTrue(self.neutron_api.client.httpclient.authenticate.called)

733 self.assertFalse(

734 self.neutron_api.client.httpclient.auth_tenant_id.called)

735

736 def test_get_all_admin_project_networks(self):

737 fake_networks = {'networks': ['fake_net_1', 'fake_net_2']}

738 self.mock_object(

739 self.neutron_api.client, 'list_networks',

740 mock.Mock(return_value=fake_networks))

741 self.neutron_api.client.httpclient = mock.Mock()

742 self.neutron_api.client.httpclient.auth_token = mock.Mock()

743 self.neutron_api.client.httpclient.auth_tenant_id = mock.Mock()

744

745 networks = self.neutron_api.get_all_admin_project_networks()

746

747 self.assertEqual(fake_networks['networks'], networks)

748 self.neutron_api.client.httpclient.auth_token.called

749 self.neutron_api.client.httpclient.auth_tenant_id.called

750 self.neutron_api.client.list_networks.assert_called_once_with(

751 tenant_id=self.neutron_api.admin_project_id, shared=False)

Coverage for manila/tests/network/neutron/test_neutron_api.py: 97%

349 statements