option: "update_neutron_port_dp_status"
[doctor.git] / doctor_tests / network.py
1 ##############################################################################
2 # Copyright (c) 2017 ZTE Corporation and others.
3 #
4 # All rights reserved. This program and the accompanying materials
5 # are made available under the terms of the Apache License, Version 2.0
6 # which accompanies this distribution, and is available at
7 # http://www.apache.org/licenses/LICENSE-2.0
8 ##############################################################################
9 from oslo_config import cfg
10
11 from doctor_tests.identity_auth import get_identity_auth
12 from doctor_tests.identity_auth import get_session
13 from doctor_tests.os_clients import neutron_client
14
15
16 OPTS = [
17     cfg.StrOpt('net_name',
18                default='doctor_net',
19                help='the name of test net',
20                required=True),
21     cfg.StrOpt('net_cidr',
22                default='192.168.168.0/24',
23                help='the cidr of test subnet',
24                required=True),
25 ]
26
27
28 class Network(object):
29
30     def __init__(self, conf, log):
31         self.conf = conf
32         self.log = log
33         self.auth = get_identity_auth(username=self.conf.doctor_user,
34                                       password=self.conf.doctor_passwd,
35                                       project=self.conf.doctor_project)
36         self.neutron = neutron_client(get_session(auth=self.auth))
37         self.net = None
38         self.subnet = None
39
40     def create(self):
41         self.log.info('network create start.......')
42         net_name = self.conf.net_name
43         networks = self.neutron.list_networks(name=net_name)['networks']
44         self.net = networks[0] if networks \
45             else self.neutron.create_network(
46             {'network': {'name': net_name}})['network']
47         self.log.info('network create end.......')
48
49         self.log.info('subnet create start.......')
50         subnets = self.neutron.list_subnets(network_id=self.net['id'])['subnets']
51         subnet_param = {'name': net_name, 'network_id': self.net['id'],
52                         'cidr': self.conf.net_cidr, 'ip_version': 4,
53                         'enable_dhcp': False}
54         self.subnet = subnets[0] if subnets \
55             else self.neutron.create_subnet(
56             {'subnet': subnet_param})['subnet']
57         self.log.info('subnet create end.......')
58
59     def delete(self):
60         self.log.info('subnet delete start.......')
61         if self.subnet:
62             self.neutron.delete_subnet(self.subnet['id'])
63         self.log.info('subnet delete end.......')
64
65         self.log.info('network delete start.......')
66         if self.net:
67             self.neutron.delete_network(self.net['id'])
68         self.log.info('network delete end.......')