← Back to branch summary

~ubuntu-branches/ubuntu/raring/quantum/raring-proposed

« back to all changes in this revision

Viewing changes to quantum/tests/unit/linuxbridge/test_lb_security_group.py

  • Committer: Package Import Robot
  • Author(s): Chuck Short, Chuck Short, Yolanda Robla, James Page, Maru Newby
  • Date: 2013-01-11 09:14:35 UTC
  • mfrom: (2.1.17)
  • Revision ID: package-import@ubuntu.com-20130111091435-vaup7dwmtmajy5oe
Tags: 2013.1~g2-0ubuntu1
https://launchpad.net/bugs/1085038
https://launchpad.net/bugs/1076747
[ Chuck Short ]
* New upstream version. 
* debian/patches/fix-quantum-configuration.patch: Refreshed.

[ Yolanda Robla ]
* debian/quantum-l3-agent.quantum-metadata-agent.upstart: Add
  upstart configuration for Metadata Agent.
* debian/quantum-l3-agent.install: Added quantum-ns-metadata-proxy,
  quantum-metadata-agent and metadata_agent.ini.
* debian/patches/fix-quantum-configuration.patch: Update rootwrap
  configuration in metadata_agent.ini file.
* debian/changelog: Updated package version
* d/p/fix-quantum-configuration.patch: refresh patches

[ James Page ]
* d/*.install: Install entry points from bin directory instead
  of easy-install ones generated during the package build process
  (LP: #1085038).
* d/control: Drop BD on python-dev-all; its not required.
* d/rules: Install multiple upstart configurations for quantum-l3-agent.
* d/control: Tidy package descriptions.
* d/*.postrm: Drop as debhelper will generate update-rc.d calls in
  maintainer scripts if required.
* d/quantum-common.postinst: Tweak permissions setting so that /etc/quantum
  is not owned/writable by the quantum user, ensure that /etc/quantum/rootwrap*
  is owned by root:root.
* d/*agent*.postinst: Dropped as permissions now correctly set in
  quantum-common.
* d/patches/fix-quantum-configuration.patch: Re-add dropped fixes rootwrap and
  sqlite defaults for all plugins.
* d/control: Added new BD on alembic (>= 0.4.1~), version python-mock >= 1.0b1.

[ Maru Newby ]
* debian/control: Remove unnecessary openvswitch-vswitch dependency
  from quantum-plugin-openvswitch (LP: #1076747).

Show diffs side-by-side

added added

removed removed

Lines of Context:
quantum/tests/unit/linuxbridge/test_lb_security_group.py
 
1
# vim: tabstop=4 shiftwidth=4 softtabstop=4
 
2
#
 
3
# Copyright 2012, Nachi Ueno, NTT MCL, Inc.
 
4
# All Rights Reserved.
 
5
#
 
6
#    Licensed under the Apache License, Version 2.0 (the "License"); you may
 
7
#    not use this file except in compliance with the License. You may obtain
 
8
#    a copy of the License at
 
9
#
 
10
#         http://www.apache.org/licenses/LICENSE-2.0
 
11
#
 
12
#    Unless required by applicable law or agreed to in writing, software
 
13
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 
14
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 
15
#    License for the specific language governing permissions and limitations
 
16
#    under the License.
 
17
 
 
18
import mock
 
19
from mock import call
 
20
 
 
21
from quantum.api.v2 import attributes
 
22
from quantum.extensions import securitygroup as ext_sg
 
23
from quantum.plugins.linuxbridge.db import l2network_db_v2 as lb_db
 
24
from quantum.tests.unit import test_extension_security_group as test_sg
 
25
 
 
26
PLUGIN_NAME = ('quantum.plugins.linuxbridge.'
 
27
               'lb_quantum_plugin.LinuxBridgePluginV2')
 
28
AGENT_NAME = ('quantum.plugins.linuxbridge.'
 
29
              'agent.linuxbridg_quantum_agent.LinuxBridgeQuantumAgentRPC')
 
30
NOTIFIER = ('quantum.plugins.linuxbridge.'
 
31
            'lb_quantum_plugin.AgentNotifierApi')
 
32
 
 
33
 
 
34
class LinuxBridgeSecurityGroupsTestCase(test_sg.SecurityGroupDBTestCase):
 
35
    _plugin_name = PLUGIN_NAME
 
36
 
 
37
    def setUp(self, plugin=None):
 
38
        self.addCleanup(mock.patch.stopall)
 
39
        notifier_p = mock.patch(NOTIFIER)
 
40
        notifier_cls = notifier_p.start()
 
41
        self.notifier = mock.Mock()
 
42
        notifier_cls.return_value = self.notifier
 
43
        self._attribute_map_bk_ = {}
 
44
        for item in attributes.RESOURCE_ATTRIBUTE_MAP:
 
45
            self._attribute_map_bk_[item] = (attributes.
 
46
                                             RESOURCE_ATTRIBUTE_MAP[item].
 
47
                                             copy())
 
48
        super(LinuxBridgeSecurityGroupsTestCase, self).setUp(PLUGIN_NAME)
 
49
 
 
50
    def tearDown(self):
 
51
        super(LinuxBridgeSecurityGroupsTestCase, self).tearDown()
 
52
        attributes.RESOURCE_ATTRIBUTE_MAP = self._attribute_map_bk_
 
53
 
 
54
 
 
55
class TestLinuxBridgeSecurityGroups(LinuxBridgeSecurityGroupsTestCase,
 
56
                                    test_sg.TestSecurityGroups):
 
57
 
 
58
    def test_security_group_rule_updated(self):
 
59
        name = 'webservers'
 
60
        description = 'my webservers'
 
61
        with self.security_group(name, description) as sg:
 
62
            with self.security_group(name, description) as sg2:
 
63
                security_group_id = sg['security_group']['id']
 
64
                direction = "ingress"
 
65
                source_group_id = sg2['security_group']['id']
 
66
                protocol = 'tcp'
 
67
                port_range_min = 88
 
68
                port_range_max = 88
 
69
                with self.security_group_rule(security_group_id, direction,
 
70
                                              protocol, port_range_min,
 
71
                                              port_range_max,
 
72
                                              source_group_id=source_group_id
 
73
                                              ):
 
74
                    pass
 
75
            self.notifier.assert_has_calls(
 
76
                [call.security_groups_rule_updated(mock.ANY,
 
77
                                                   [security_group_id]),
 
78
                 call.security_groups_rule_updated(mock.ANY,
 
79
                                                   [security_group_id])])
 
80
 
 
81
    def test_security_group_member_updated(self):
 
82
        with self.network() as n:
 
83
            with self.subnet(n):
 
84
                with self.security_group() as sg:
 
85
                    security_group_id = sg['security_group']['id']
 
86
                    res = self._create_port('json', n['network']['id'])
 
87
                    port = self.deserialize('json', res)
 
88
 
 
89
                    data = {'port': {'fixed_ips': port['port']['fixed_ips'],
 
90
                                     'name': port['port']['name'],
 
91
                                     ext_sg.SECURITYGROUP:
 
92
                                     [security_group_id]}}
 
93
 
 
94
                    req = self.new_update_request('ports', data,
 
95
                                                  port['port']['id'])
 
96
                    res = self.deserialize('json', req.get_response(self.api))
 
97
                    self.assertEquals(res['port'][ext_sg.SECURITYGROUP][0],
 
98
                                      security_group_id)
 
99
                    self._delete('ports', port['port']['id'])
 
100
                    self.notifier.assert_has_calls(
 
101
                        [call.security_groups_member_updated(
 
102
                            mock.ANY, [mock.ANY]),
 
103
                         call.security_groups_member_updated(
 
104
                             mock.ANY, [security_group_id])])
 
105
 
 
106
 
 
107
class TestLinuxBridgeSecurityGroupsDB(LinuxBridgeSecurityGroupsTestCase):
 
108
    def test_security_group_get_port_from_device(self):
 
109
        with self.network() as n:
 
110
            with self.subnet(n):
 
111
                with self.security_group() as sg:
 
112
                    security_group_id = sg['security_group']['id']
 
113
                    res = self._create_port('json', n['network']['id'])
 
114
                    port = self.deserialize('json', res)
 
115
                    fixed_ips = port['port']['fixed_ips']
 
116
                    data = {'port': {'fixed_ips': fixed_ips,
 
117
                                     'name': port['port']['name'],
 
118
                                     ext_sg.SECURITYGROUP:
 
119
                                     [security_group_id]}}
 
120
 
 
121
                    req = self.new_update_request('ports', data,
 
122
                                                  port['port']['id'])
 
123
                    res = self.deserialize('json', req.get_response(self.api))
 
124
                    port_id = res['port']['id']
 
125
                    device_id = port_id[:8]
 
126
                    port_dict = lb_db.get_port_from_device(device_id)
 
127
                    self.assertEqual(port_id, port_dict['id'])
 
128
                    self.assertEqual([security_group_id],
 
129
                                     port_dict[ext_sg.SECURITYGROUP])
 
130
                    self.assertEqual([], port_dict['security_group_rules'])
 
131
                    self.assertEqual([fixed_ips[0]['ip_address']],
 
132
                                     port_dict['fixed_ips'])
 
133
                    self._delete('ports', port['port']['id'])
 
134
 
 
135
    def test_security_group_get_port_from_device_with_no_port(self):
 
136
        port_dict = lb_db.get_port_from_device('bad_device_id')
 
137
        self.assertEqual(None, port_dict)