← Back to branch summary

~ubuntu-branches/ubuntu/vivid/neutron/vivid-updates

« back to all changes in this revision

Viewing changes to neutron/tests/unit/cisco/l3/device_handling_test_support.py

  • Committer: Package Import Robot
  • Author(s): James Page
  • Date: 2015-03-30 11:17:19 UTC
  • mfrom: (1.1.21)
  • Revision ID: package-import@ubuntu.com-20150330111719-h0gx7233p4jkkgfh
Tags: 1:2015.1~b3-0ubuntu1
https://launchpad.net/bugs/1263539
* New upstream milestone release:
  - d/control: Align version requirements with upstream.
  - d/control: Add new dependency on oslo-log.
  - d/p/*: Rebase.
  - d/control,d/neutron-plugin-hyperv*: Dropped, decomposed into
    separate project upstream.
  - d/control,d/neutron-plugin-openflow*: Dropped, decomposed into
    separate project upstream.
  - d/neutron-common.install: Add neutron-rootwrap-daemon and 
    neutron-keepalived-state-change binaries.
  - d/rules: Ignore neutron-hyperv-agent when installing; only for Windows.
  - d/neutron-plugin-cisco.install: Drop neutron-cisco-cfg-agent as
    decomposed into separate project upstream.
  - d/neutron-plugin-vmware.install: Drop neutron-check-nsx-config and
    neutron-nsx-manage as decomposed into separate project upstream.
  - d/control: Add dependency on python-neutron-fwaas to neutron-l3-agent.
* d/pydist-overrides: Add overrides for oslo packages.
* d/control: Fixup type in package description (LP: #1263539).
* d/p/fixup-driver-test-execution.patch: Cherry pick fix from upstream VCS
  to support unit test exection in out-of-tree vendor drivers.
* d/neutron-common.postinst: Allow general access to /etc/neutron but limit
  access to root/neutron to /etc/neutron/neutron.conf to support execution
  of unit tests in decomposed vendor drivers.
* d/control: Add dependency on python-neutron-fwaas to neutron-l3-agent
  package.

Show diffs side-by-side

added added

removed removed

Lines of Context:
neutron/tests/unit/cisco/l3/device_handling_test_support.py
1
 
# Copyright 2014 Cisco Systems, Inc.  All rights reserved.
2
 
#
3
 
#    Licensed under the Apache License, Version 2.0 (the "License"); you may
4
 
#    not use this file except in compliance with the License. You may obtain
5
 
#    a copy of the License at
6
 
#
7
 
#         http://www.apache.org/licenses/LICENSE-2.0
8
 
#
9
 
#    Unless required by applicable law or agreed to in writing, software
10
 
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
11
 
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
12
 
#    License for the specific language governing permissions and limitations
13
 
#    under the License.
14
 
 
15
 
import mock
16
 
from novaclient import exceptions as nova_exc
17
 
from oslo.config import cfg
18
 
from oslo.utils import excutils
19
 
 
20
 
from neutron import context as n_context
21
 
from neutron.i18n import _LE
22
 
from neutron import manager
23
 
from neutron.openstack.common import log as logging
24
 
from neutron.openstack.common import uuidutils
25
 
from neutron.plugins.common import constants
26
 
 
27
 
LOG = logging.getLogger(__name__)
28
 
 
29
 
 
30
 
_uuid = uuidutils.generate_uuid
31
 
 
32
 
 
33
 
class DeviceHandlingTestSupportMixin(object):
34
 
 
35
 
    @property
36
 
    def _core_plugin(self):
37
 
        return manager.NeutronManager.get_plugin()
38
 
 
39
 
    def _mock_l3_admin_tenant(self):
40
 
        # Mock l3 admin tenant
41
 
        self.tenant_id_fcn_p = mock.patch(
42
 
            'neutron.plugins.cisco.db.l3.device_handling_db.'
43
 
            'DeviceHandlingMixin.l3_tenant_id')
44
 
        self.tenant_id_fcn = self.tenant_id_fcn_p.start()
45
 
        self.tenant_id_fcn.return_value = "L3AdminTenantId"
46
 
 
47
 
    def _create_mgmt_nw_for_tests(self, fmt):
48
 
        self._mgmt_nw = self._make_network(fmt,
49
 
                                           cfg.CONF.general.management_network,
50
 
                                           True, tenant_id="L3AdminTenantId",
51
 
                                           shared=False)
52
 
        self._mgmt_subnet = self._make_subnet(fmt, self._mgmt_nw,
53
 
                                              "10.0.100.1", "10.0.100.0/24",
54
 
                                              ip_version=4)
55
 
 
56
 
    def _remove_mgmt_nw_for_tests(self):
57
 
        q_p = "network_id=%s" % self._mgmt_nw['network']['id']
58
 
        subnets = self._list('subnets', query_params=q_p)
59
 
        if subnets:
60
 
            for p in self._list('ports', query_params=q_p).get('ports'):
61
 
                self._delete('ports', p['id'])
62
 
            self._delete('subnets', self._mgmt_subnet['subnet']['id'])
63
 
            self._delete('networks', self._mgmt_nw['network']['id'])
64
 
 
65
 
    # Function used to mock novaclient services list
66
 
    def _novaclient_services_list(self, all=True):
67
 
        services = set(['nova-conductor', 'nova-cert', 'nova-scheduler',
68
 
                        'nova-compute', 'nova-consoleauth'])
69
 
        full_list = [FakeResource(binary=res) for res in services]
70
 
        _all = all
71
 
 
72
 
        def response():
73
 
            if _all:
74
 
                return full_list
75
 
            else:
76
 
                return full_list[2:]
77
 
        return response
78
 
 
79
 
    # Function used to mock novaclient servers create
80
 
    def _novaclient_servers_create(self, instance_name, image_id, flavor_id,
81
 
                                   nics, files, config_drive):
82
 
        fake_vm = FakeResource()
83
 
        for nic in nics:
84
 
            p_dict = {'port': {'device_id': fake_vm.id,
85
 
                               'device_owner': 'nova'}}
86
 
            self._core_plugin.update_port(n_context.get_admin_context(),
87
 
                                          nic['port-id'], p_dict)
88
 
        return fake_vm
89
 
 
90
 
    # Function used to mock novaclient servers delete
91
 
    def _novaclient_servers_delete(self, vm_id):
92
 
        q_p = "device_id=%s" % vm_id
93
 
        ports = self._list('ports', query_params=q_p)
94
 
        for port in ports.get('ports', []):
95
 
            try:
96
 
                self._delete('ports', port['id'])
97
 
            except Exception as e:
98
 
                with excutils.save_and_reraise_exception(reraise=False):
99
 
                    LOG.error(_LE('Failed to delete port %(p_id)s for vm '
100
 
                                  'instance %(v_id)s due to %(err)s'),
101
 
                              {'p_id': port['id'], 'v_id': vm_id, 'err': e})
102
 
                    raise nova_exc.InternalServerError()
103
 
 
104
 
    def _mock_svc_vm_create_delete(self, plugin):
105
 
        # Mock novaclient methods for creation/deletion of service VMs
106
 
        mock.patch(
107
 
            'neutron.plugins.cisco.l3.service_vm_lib.n_utils.find_resource',
108
 
            lambda *args, **kw: FakeResource()).start()
109
 
        self._nclient_services_mock = mock.MagicMock()
110
 
        self._nclient_services_mock.list = self._novaclient_services_list()
111
 
        mock.patch.object(plugin._svc_vm_mgr._nclient, 'services',
112
 
                          self._nclient_services_mock).start()
113
 
        nclient_servers_mock = mock.MagicMock()
114
 
        nclient_servers_mock.create = self._novaclient_servers_create
115
 
        nclient_servers_mock.delete = self._novaclient_servers_delete
116
 
        mock.patch.object(plugin._svc_vm_mgr._nclient, 'servers',
117
 
                          nclient_servers_mock).start()
118
 
 
119
 
    def _mock_io_file_ops(self):
120
 
        # Mock library functions for config drive file operations
121
 
        cfg_template = '\n'.join(['interface GigabitEthernet1',
122
 
                                  'ip address <ip> <mask>',
123
 
                                  'no shutdown'])
124
 
        m = mock.mock_open(read_data=cfg_template)
125
 
        m.return_value.__iter__.return_value = cfg_template.splitlines()
126
 
        mock.patch('neutron.plugins.cisco.l3.hosting_device_drivers.'
127
 
                   'csr1kv_hd_driver.open', m, create=True).start()
128
 
 
129
 
    def _test_remove_all_hosting_devices(self):
130
 
        """Removes all hosting devices created during a test."""
131
 
        plugin = manager.NeutronManager.get_service_plugins()[
132
 
            constants.L3_ROUTER_NAT]
133
 
        context = n_context.get_admin_context()
134
 
        plugin.delete_all_hosting_devices(context, True)
135
 
 
136
 
    def _get_fake_resource(self, tenant_id=None, id=None):
137
 
        return {'id': id or _uuid(),
138
 
                'tenant_id': tenant_id or _uuid()}
139
 
 
140
 
    def _get_test_context(self, user_id=None, tenant_id=None, is_admin=False):
141
 
        return n_context.Context(user_id, tenant_id, is_admin,
142
 
                                 load_admin_roles=True)
143
 
 
144
 
 
145
 
# Used to fake Glance images, Nova VMs and Nova services
146
 
class FakeResource(object):
147
 
    def __init__(self, id=None, enabled='enabled', state='up', binary=None):
148
 
        self.id = id or _uuid()
149
 
        self.status = enabled
150
 
        self.state = state
151
 
        self.binary = binary