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 = \
  51             self.neutron.list_subnets(network_id=self.net['id'])['subnets']
  52         subnet_param = {'name': net_name, 'network_id': self.net['id'],
  53                         'cidr': self.conf.net_cidr, 'ip_version': 4,
  54                         'enable_dhcp': False}
  55         self.subnet = subnets[0] if subnets \
  56             else self.neutron.create_subnet(
  57             {'subnet': subnet_param})['subnet']
  58         self.log.info('subnet create end.......')
  59
  60     def delete(self):
  61         self.log.info('subnet delete start.......')
  62         if self.subnet:
  63             self.neutron.delete_subnet(self.subnet['id'])
  64         self.log.info('subnet delete end.......')
  65
  66         self.log.info('network delete start.......')
  67         if self.net:
  68             self.neutron.delete_network(self.net['id'])
  69         self.log.info('network delete end.......')