← Back to branch summary

~ubuntu-branches/ubuntu/saucy/quantum/saucy

« back to all changes in this revision

Viewing changes to quantum/tests/unit/ml2/test_security_group.py

  • Committer: Package Import Robot
  • Author(s): Chuck Short
  • Date: 2013-05-31 09:37:25 UTC
  • mfrom: (2.1.22)
  • Revision ID: package-import@ubuntu.com-20130531093725-bf9jom93l7jm57iv
Tags: 1:2013.2~b1-0ubuntu1
* New upstream release.
* debian/patches/fix-quantum-configuration.patch: Refreshed
* debian/control: Add testrepository.
* debian/control: Add subunit.
* debian/control: Add python-d21o1 and python-pbr as build-depends.
* debian/control: Add python-stevedore.

Show diffs side-by-side

added added

removed removed

Lines of Context:
quantum/tests/unit/ml2/test_security_group.py
 
1
# Copyright (c) 2013 OpenStack Foundation
 
2
# Copyright 2013, Nachi Ueno, NTT MCL, Inc.
 
3
# All Rights Reserved.
 
4
#
 
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
 
8
#
 
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.
 
16
 
 
17
import mock
 
18
 
 
19
from quantum.api.v2 import attributes
 
20
from quantum.extensions import securitygroup as ext_sg
 
21
from quantum import manager
 
22
from quantum.tests.unit import test_extension_security_group as test_sg
 
23
from quantum.tests.unit import test_security_groups_rpc as test_sg_rpc
 
24
 
 
25
PLUGIN_NAME = 'quantum.plugins.ml2.plugin.Ml2Plugin'
 
26
NOTIFIER = 'quantum.plugins.ml2.rpc.AgentNotifierApi'
 
27
 
 
28
 
 
29
class Ml2SecurityGroupsTestCase(test_sg.SecurityGroupDBTestCase):
 
30
    _plugin_name = PLUGIN_NAME
 
31
 
 
32
    def setUp(self, plugin=None):
 
33
        test_sg_rpc.set_firewall_driver(test_sg_rpc.FIREWALL_HYBRID_DRIVER)
 
34
        self.addCleanup(mock.patch.stopall)
 
35
        notifier_p = mock.patch(NOTIFIER)
 
36
        notifier_cls = notifier_p.start()
 
37
        self.notifier = mock.Mock()
 
38
        notifier_cls.return_value = self.notifier
 
39
        self._attribute_map_bk_ = {}
 
40
        for item in attributes.RESOURCE_ATTRIBUTE_MAP:
 
41
            self._attribute_map_bk_[item] = (attributes.
 
42
                                             RESOURCE_ATTRIBUTE_MAP[item].
 
43
                                             copy())
 
44
        super(Ml2SecurityGroupsTestCase, self).setUp(PLUGIN_NAME)
 
45
 
 
46
    def tearDown(self):
 
47
        super(Ml2SecurityGroupsTestCase, self).tearDown()
 
48
        attributes.RESOURCE_ATTRIBUTE_MAP = self._attribute_map_bk_
 
49
 
 
50
 
 
51
class TestMl2SecurityGroups(Ml2SecurityGroupsTestCase,
 
52
                            test_sg.TestSecurityGroups,
 
53
                            test_sg_rpc.SGNotificationTestMixin):
 
54
    def test_security_group_get_port_from_device(self):
 
55
        with self.network() as n:
 
56
            with self.subnet(n):
 
57
                with self.security_group() as sg:
 
58
                    security_group_id = sg['security_group']['id']
 
59
                    res = self._create_port(self.fmt, n['network']['id'])
 
60
                    port = self.deserialize(self.fmt, res)
 
61
                    fixed_ips = port['port']['fixed_ips']
 
62
                    data = {'port': {'fixed_ips': fixed_ips,
 
63
                                     'name': port['port']['name'],
 
64
                                     ext_sg.SECURITYGROUPS:
 
65
                                     [security_group_id]}}
 
66
 
 
67
                    req = self.new_update_request('ports', data,
 
68
                                                  port['port']['id'])
 
69
                    res = self.deserialize(self.fmt,
 
70
                                           req.get_response(self.api))
 
71
                    port_id = res['port']['id']
 
72
                    plugin = manager.QuantumManager.get_plugin()
 
73
                    port_dict = plugin.callbacks.get_port_from_device(port_id)
 
74
                    self.assertEqual(port_id, port_dict['id'])
 
75
                    self.assertEqual([security_group_id],
 
76
                                     port_dict[ext_sg.SECURITYGROUPS])
 
77
                    self.assertEqual([], port_dict['security_group_rules'])
 
78
                    self.assertEqual([fixed_ips[0]['ip_address']],
 
79
                                     port_dict['fixed_ips'])
 
80
                    self._delete('ports', port_id)
 
81
 
 
82
    def test_security_group_get_port_from_device_with_no_port(self):
 
83
        plugin = manager.QuantumManager.get_plugin()
 
84
        port_dict = plugin.callbacks.get_port_from_device('bad_device_id')
 
85
        self.assertEqual(None, port_dict)
 
86
 
 
87
 
 
88
class TestMl2SecurityGroupsXML(TestMl2SecurityGroups):
 
89
    fmt = 'xml'