python-neutronclient/neutronclient/tests/unit/test_cli20_securitygroup.py

#!/usr/bin/env python
# Copyright 2012 Red Hat
# All Rights Reserved.
#
#    Licensed under the Apache License, Version 2.0 (the "License"); you may
#    not use this file except in compliance with the License. You may obtain
#    a copy of the License at
#
#         http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
#    License for the specific language governing permissions and limitations
#    under the License.

import sys
from unittest import mock
import urllib.parse as urlparse

from oslo_utils import uuidutils

from neutronclient.common import exceptions
from neutronclient.common import utils
from neutronclient.neutron.v2_0 import securitygroup
from neutronclient.tests.unit import test_cli20


class CLITestV20SecurityGroupsJSON(test_cli20.CLITestV20Base):

    non_admin_status_resources = ['security_group', 'security_group_rule']

    def test_create_security_group(self):
        # Create security group: webservers.
        resource = 'security_group'
        cmd = securitygroup.CreateSecurityGroup(
            test_cli20.MyApp(sys.stdout), None)
        name = 'webservers'
        myid = 'myid'
        args = [name, ]
        position_names = ['name']
        position_values = [name]
        self._test_create_resource(resource, cmd, name, myid, args,
                                   position_names, position_values)

    def test_create_security_group_tenant(self):
        # Create security group: webservers.
        resource = 'security_group'
        cmd = securitygroup.CreateSecurityGroup(
            test_cli20.MyApp(sys.stdout), None)
        name = 'webservers'
        description = 'my webservers'
        myid = 'myid'
        args = ['--tenant_id', 'tenant_id', '--description', description, name]
        position_names = ['name', 'description']
        position_values = [name, description]
        self._test_create_resource(resource, cmd, name, myid, args,
                                   position_names, position_values,
                                   tenant_id='tenant_id')

    def test_create_security_group_with_description(self):
        # Create security group: webservers.
        resource = 'security_group'
        cmd = securitygroup.CreateSecurityGroup(
            test_cli20.MyApp(sys.stdout), None)
        name = 'webservers'
        description = 'my webservers'
        myid = 'myid'
        args = [name, '--description', description]
        position_names = ['name', 'description']
        position_values = [name, description]
        self._test_create_resource(resource, cmd, name, myid, args,
                                   position_names, position_values)

    def test_list_security_groups(self):
        resources = "security_groups"
        cmd = securitygroup.ListSecurityGroup(
            test_cli20.MyApp(sys.stdout), None)
        self._test_list_resources(resources, cmd, True)

    def test_list_security_groups_pagination(self):
        resources = "security_groups"
        cmd = securitygroup.ListSecurityGroup(
            test_cli20.MyApp(sys.stdout), None)
        self._test_list_resources_with_pagination(resources, cmd)

    def test_list_security_groups_sort(self):
        resources = "security_groups"
        cmd = securitygroup.ListSecurityGroup(
            test_cli20.MyApp(sys.stdout), None)
        self._test_list_resources(resources, cmd,
                                  sort_key=["name", "id"],
                                  sort_dir=["asc", "desc"])

    def test_list_security_groups_limit(self):
        resources = "security_groups"
        cmd = securitygroup.ListSecurityGroup(
            test_cli20.MyApp(sys.stdout), None)
        self._test_list_resources(resources, cmd, page_size=1000)

    def test_show_security_group_id(self):
        resource = 'security_group'
        cmd = securitygroup.ShowSecurityGroup(
            test_cli20.MyApp(sys.stdout), None)
        args = ['--fields', 'id', self.test_id]
        self._test_show_resource(resource, cmd, self.test_id,
                                 args, ['id'])

    def test_show_security_group_id_name(self):
        resource = 'security_group'
        cmd = securitygroup.ShowSecurityGroup(
            test_cli20.MyApp(sys.stdout), None)
        args = ['--fields', 'id', '--fields', 'name', self.test_id]
        self._test_show_resource(resource, cmd, self.test_id,
                                 args, ['id', 'name'])

    def test_delete_security_group(self):
        # Delete security group: myid.
        resource = 'security_group'
        cmd = securitygroup.DeleteSecurityGroup(
            test_cli20.MyApp(sys.stdout), None)
        myid = 'myid'
        args = [myid]
        self._test_delete_resource(resource, cmd, myid, args)

    def test_update_security_group(self):
        # Update security group: myid --name myname --description desc.
        resource = 'security_group'
        cmd = securitygroup.UpdateSecurityGroup(
            test_cli20.MyApp(sys.stdout), None)
        self._test_update_resource(resource, cmd, 'myid',
                                   ['myid', '--name', 'myname',
                                    '--description', 'mydescription'],
                                   {'name': 'myname',
                                    'description': 'mydescription'}
                                   )

    def test_update_security_group_with_unicode(self):
        resource = 'security_group'
        cmd = securitygroup.UpdateSecurityGroup(
            test_cli20.MyApp(sys.stdout), None)
        self._test_update_resource(resource, cmd, 'myid',
                                   ['myid', '--name', u'\u7f51\u7edc',
                                    '--description', u'\u7f51\u7edc'],
                                   {'name': u'\u7f51\u7edc',
                                    'description': u'\u7f51\u7edc'}
                                   )

    def test_create_security_group_rule_full(self):
        # Create security group rule.
        resource = 'security_group_rule'
        cmd = securitygroup.CreateSecurityGroupRule(
            test_cli20.MyApp(sys.stdout), None)
        myid = 'myid'
        direction = 'ingress'
        ethertype = 'IPv4'
        protocol = 'tcp'
        port_range_min = '22'
        port_range_max = '22'
        remote_ip_prefix = '10.0.0.0/24'
        security_group_id = '1'
        remote_group_id = '1'
        args = ['--remote_ip_prefix', remote_ip_prefix, '--direction',
                direction, '--ethertype', ethertype, '--protocol', protocol,
                '--port_range_min', port_range_min, '--port_range_max',
                port_range_max, '--remote_group_id', remote_group_id,
                security_group_id, '--description', 'PCI policy 1421912']
        position_names = ['remote_ip_prefix', 'direction', 'ethertype',
                          'protocol', 'port_range_min', 'port_range_max',
                          'remote_group_id', 'security_group_id']
        position_values = [remote_ip_prefix, direction, ethertype, protocol,
                           port_range_min, port_range_max, remote_group_id,
                           security_group_id]
        self._test_create_resource(resource, cmd, None, myid, args,
                                   position_names, position_values,
                                   description='PCI policy 1421912')

    def test_create_security_group_rule_with_integer_protocol_value(self):
        resource = 'security_group_rule'
        cmd = securitygroup.CreateSecurityGroupRule(
            test_cli20.MyApp(sys.stdout), None)
        myid = 'myid'
        direction = 'ingress'
        ethertype = 'IPv4'
        protocol = '2'
        port_range_min = '22'
        port_range_max = '22'
        remote_ip_prefix = '10.0.0.0/24'
        security_group_id = '1'
        remote_group_id = '1'
        args = ['--remote_ip_prefix', remote_ip_prefix, '--direction',
                direction, '--ethertype', ethertype, '--protocol', protocol,
                '--port_range_min', port_range_min, '--port_range_max',
                port_range_max, '--remote_group_id', remote_group_id,
                security_group_id]
        position_names = ['remote_ip_prefix', 'direction', 'ethertype',
                          'protocol', 'port_range_min', 'port_range_max',
                          'remote_group_id', 'security_group_id']
        position_values = [remote_ip_prefix, direction, ethertype, protocol,
                           port_range_min, port_range_max, remote_group_id,
                           security_group_id]
        self._test_create_resource(resource, cmd, None, myid, args,
                                   position_names, position_values)

    def test_delete_security_group_rule(self):
        # Delete security group rule: myid.
        resource = 'security_group_rule'
        cmd = securitygroup.DeleteSecurityGroupRule(
            test_cli20.MyApp(sys.stdout), None)
        myid = 'myid'
        args = [myid]
        self._test_delete_resource(resource, cmd, myid, args)

    @mock.patch.object(securitygroup.ListSecurityGroupRule, "extend_list")
    def test_list_security_group_rules(self, mock_extend_list):
        resources = "security_group_rules"
        cmd = securitygroup.ListSecurityGroupRule(
            test_cli20.MyApp(sys.stdout), None)
        self._test_list_resources(resources, cmd, True)
        mock_extend_list.assert_called_once_with(test_cli20.IsA(list),
                                                 mock.ANY)

    def _build_test_data(self, data, excess=0):
        # Length of a query filter on security group rule id
        # in these testcases, id='secgroupid%02d' (with len(id)=12)
        sec_group_id_filter_len = 12

        response = []
        replace_rules = {'security_group_id': 'security_group',
                         'remote_group_id': 'remote_group'}

        search_opts = {'fields': ['id', 'name']}
        sec_group_ids = set()
        for rule in data:
            for key in replace_rules:
                if rule.get(key):
                    sec_group_ids.add(rule[key])
                    response.append({'id': rule[key], 'name': 'default'})
        sec_group_ids = list(sec_group_ids)

        result = []

        sec_group_count = len(sec_group_ids)
        max_size = ((sec_group_id_filter_len * sec_group_count) - excess)
        chunk_size = max_size // sec_group_id_filter_len

        for i in range(0, sec_group_count, chunk_size):
            search_opts['id'] = sec_group_ids[i: i + chunk_size]
            params = utils.safe_encode_dict(search_opts)
            resp_str = self.client.serialize({'security_groups': response})

            result.append({
                'filter': urlparse.urlencode(params, doseq=1),
                'response': (test_cli20.MyResp(200), resp_str),
            })

        return result

    def test_extend_list(self):
        data = [{'name': 'default',
                 'remote_group_id': 'remgroupid%02d' % i}
                for i in range(10)]
        data.append({'name': 'default', 'remote_group_id': None})
        resources = "security_groups"

        cmd = securitygroup.ListSecurityGroupRule(
            test_cli20.MyApp(sys.stdout), None)

        path = getattr(self.client, resources + '_path')
        responses = self._build_test_data(data)
        known_args, _vs = cmd.get_parser(
            'list' + resources).parse_known_args()
        resp = responses[0]['response']

        with mock.patch.object(cmd, "get_client",
                               return_value=self.client) as mock_get_client, \
                mock.patch.object(self.client.httpclient, "request",
                                  return_value=resp) as mock_request:
            cmd.extend_list(data, known_args)

        mock_get_client.assert_called_once_with()
        mock_request.assert_called_once_with(
            test_cli20.MyUrlComparator(test_cli20.end_url(
                path, responses[0]['filter']), self.client),
            'GET',
            body=None,
            headers=test_cli20.ContainsKeyValue(
                {'X-Auth-Token': test_cli20.TOKEN}))

    def test_extend_list_exceed_max_uri_len(self):
        data = [{'name': 'default',
                 'security_group_id': 'secgroupid%02d' % i,
                 'remote_group_id': 'remgroupid%02d' % i}
                for i in range(10)]
        data.append({'name': 'default',
                     'security_group_id': 'secgroupid10',
                     'remote_group_id': None})
        resources = "security_groups"

        cmd = securitygroup.ListSecurityGroupRule(
            test_cli20.MyApp(sys.stdout), None)
        path = getattr(self.client, resources + '_path')
        responses = self._build_test_data(data, excess=1)

        known_args, _vs = cmd.get_parser(
            'list' + resources).parse_known_args()
        mock_request_side_effects = []
        mock_request_calls = []
        mock_check_uri_side_effects = [exceptions.RequestURITooLong(excess=1)]
        mock_check_uri_calls = [mock.call(mock.ANY)]
        for item in responses:
            mock_request_side_effects.append(item['response'])
            mock_request_calls.append(mock.call(
                test_cli20.MyUrlComparator(
                    test_cli20.end_url(path, item['filter']), self.client),
                'GET',
                body=None,
                headers=test_cli20.ContainsKeyValue(
                    {'X-Auth-Token': test_cli20.TOKEN})))
            mock_check_uri_side_effects.append(None)
            mock_check_uri_calls.append(mock.call(mock.ANY))

        with mock.patch.object(cmd, "get_client",
                               return_value=self.client) as mock_get_client, \
                mock.patch.object(self.client.httpclient,
                                  "request") as mock_request, \
                mock.patch.object(self.client.httpclient,
                                  "_check_uri_length") as mock_check_uri:
            mock_request.side_effect = mock_request_side_effects
            mock_check_uri.side_effect = mock_check_uri_side_effects
            cmd.extend_list(data, known_args)

        mock_get_client.assert_called_once_with()
        mock_request.assert_has_calls(mock_request_calls)
        mock_check_uri.assert_has_calls(mock_check_uri_calls)
        self.assertEqual(len(mock_request_calls), mock_request.call_count)
        self.assertEqual(len(mock_check_uri_calls), mock_check_uri.call_count)

    @mock.patch.object(securitygroup.ListSecurityGroupRule, "extend_list")
    def test_list_security_group_rules_pagination(self, mock_extend_list):
        resources = "security_group_rules"
        cmd = securitygroup.ListSecurityGroupRule(
            test_cli20.MyApp(sys.stdout), None)
        self._test_list_resources_with_pagination(resources, cmd)
        mock_extend_list.assert_called_once_with(test_cli20.IsA(list),
                                                 mock.ANY)

    @mock.patch.object(securitygroup.ListSecurityGroupRule, "extend_list")
    def test_list_security_group_rules_sort(self, mock_extend_list):
        resources = "security_group_rules"
        cmd = securitygroup.ListSecurityGroupRule(
            test_cli20.MyApp(sys.stdout), None)
        self._test_list_resources(resources, cmd,
                                  sort_key=["name", "id"],
                                  sort_dir=["asc", "desc"])
        mock_extend_list.assert_called_once_with(test_cli20.IsA(list),
                                                 mock.ANY)

    @mock.patch.object(securitygroup.ListSecurityGroupRule, "extend_list")
    def test_list_security_group_rules_limit(self, mock_extend_list):
        resources = "security_group_rules"
        cmd = securitygroup.ListSecurityGroupRule(
            test_cli20.MyApp(sys.stdout), None)
        self._test_list_resources(resources, cmd, page_size=1000)
        mock_extend_list.assert_called_once_with(test_cli20.IsA(list),
                                                 mock.ANY)

    def test_show_security_group_rule(self):
        resource = 'security_group_rule'
        cmd = securitygroup.ShowSecurityGroupRule(
            test_cli20.MyApp(sys.stdout), None)
        args = ['--fields', 'id', self.test_id]
        self._test_show_resource(resource, cmd, self.test_id,
                                 args, ['id'])

    def _test_list_security_group_rules_extend(self, api_data, expected,
                                               args=(), conv=True,
                                               query_fields=None):
        def setup_list_stub(resources, data, query, mock_calls, mock_returns):
            reses = {resources: data}
            resstr = self.client.serialize(reses)
            resp = (test_cli20.MyResp(200), resstr)
            path = getattr(self.client, resources + '_path')
            mock_calls.append(mock.call(
                test_cli20.MyUrlComparator(
                    test_cli20.end_url(path, query),
                    self.client),
                'GET',
                body=None,
                headers=test_cli20.ContainsKeyValue(
                    {'X-Auth-Token': test_cli20.TOKEN})))
            mock_returns.append(resp)

        cmd = securitygroup.ListSecurityGroupRule(
            test_cli20.MyApp(sys.stdout), None)
        query = ''
        if query_fields:
            query = '&'.join(['fields=' + f for f in query_fields])
        mock_request_calls = []
        mock_request_returns = []
        setup_list_stub('security_group_rules', api_data, query,
                        mock_request_calls, mock_request_returns)
        if conv:
            sec_ids = set()
            for n in api_data:
                sec_ids.add(n['security_group_id'])
                if n.get('remote_group_id'):
                    sec_ids.add(n['remote_group_id'])
            filters = ''
            for id in sec_ids:
                filters = filters + "&id=%s" % id
            setup_list_stub('security_groups',
                            [{'id': 'myid1', 'name': 'group1'},
                             {'id': 'myid2', 'name': 'group2'},
                             {'id': 'myid3', 'name': 'group3'}],
                            'fields=id&fields=name' + filters,
                            mock_request_calls,
                            mock_request_returns)

        cmd_parser = cmd.get_parser('list_security_group_rules')
        parsed_args = cmd_parser.parse_args(args)

        with mock.patch.object(cmd, "get_client",
                               return_value=self.client) as mock_get_client, \
                mock.patch.object(self.client.httpclient,
                                  "request") as mock_request:
            mock_request.side_effect = mock_request_returns
            result = cmd.take_action(parsed_args)

        self.assert_mock_multiple_calls_with_same_arguments(
            mock_get_client, mock.call(), None)
        mock_request.assert_has_calls(mock_request_calls)
        self.assertEqual(len(mock_request_calls), mock_request.call_count)
        self.assertEqual(expected['cols'], result[0])
        # Check data
        _result = [x for x in result[1]]
        self.assertEqual(len(expected['data']), len(_result))
        for res, exp in zip(_result, expected['data']):
            self.assertEqual(len(exp), len(res))
            self.assertEqual(exp, res)

    def _test_list_security_group_rules_extend_sg_name(
            self, expected_mode=None, args=(), conv=True, query_field=False):
        if query_field:
            field_filters = ['id', 'security_group_id',
                             'remote_ip_prefix', 'remote_group_id']
        else:
            field_filters = None

        data = [self._prepare_rule(rule_id='ruleid1', sg_id='myid1',
                                   remote_group_id='myid1',
                                   filters=field_filters),
                self._prepare_rule(rule_id='ruleid2', sg_id='myid2',
                                   remote_group_id='myid3',
                                   filters=field_filters),
                self._prepare_rule(rule_id='ruleid3', sg_id='myid2',
                                   remote_group_id='myid2',
                                   filters=field_filters),
                ]

        if expected_mode == 'noconv':
            expected = {'cols': ['id', 'security_group_id', 'remote_group_id'],
                        'data': [('ruleid1', 'myid1', 'myid1'),
                                 ('ruleid2', 'myid2', 'myid3'),
                                 ('ruleid3', 'myid2', 'myid2')]}
        elif expected_mode == 'remote_group_id':
            expected = {'cols': ['id', 'security_group', 'remote_group'],
                        'data': [('ruleid1', 'group1', 'group1'),
                                 ('ruleid2', 'group2', 'group3'),
                                 ('ruleid3', 'group2', 'group2')]}
        else:
            expected = {'cols': ['id', 'security_group', 'remote'],
                        'data': [('ruleid1', 'group1', 'group1 (group)'),
                                 ('ruleid2', 'group2', 'group3 (group)'),
                                 ('ruleid3', 'group2', 'group2 (group)')]}

        self._test_list_security_group_rules_extend(
            data, expected, args=args, conv=conv, query_fields=field_filters)

    def test_list_security_group_rules_extend_remote_sg_name(self):
        args = '-c id -c security_group -c remote'.split()
        self._test_list_security_group_rules_extend_sg_name(args=args)

    def test_list_security_group_rules_extend_sg_name_noconv(self):
        args = '--no-nameconv -c id -c security_group_id -c remote_group_id'
        args = args.split()
        self._test_list_security_group_rules_extend_sg_name(
            expected_mode='noconv', args=args, conv=False)

    def test_list_security_group_rules_extend_sg_name_with_columns(self):
        args = '-c id -c security_group_id -c remote_group_id'.split()
        self._test_list_security_group_rules_extend_sg_name(
            expected_mode='remote_group_id', args=args)

    def test_list_security_group_rules_extend_sg_name_with_columns_no_id(self):
        args = '-c id -c security_group -c remote_group'.split()
        self._test_list_security_group_rules_extend_sg_name(
            expected_mode='remote_group_id', args=args)

    def test_list_security_group_rules_extend_sg_name_with_fields(self):
        # NOTE: remote_ip_prefix is required to show "remote" column
        args = ('-F id -F security_group_id '
                '-F remote_ip_prefix -F remote_group_id').split()
        self._test_list_security_group_rules_extend_sg_name(
            args=args, query_field=True)

    def test_list_security_group_rules_extend_sg_name_with_fields_no_id(self):
        # NOTE: remote_ip_prefix is required to show "remote" column
        args = ('-F id -F security_group '
                '-F remote_ip_prefix -F remote_group').split()
        self._test_list_security_group_rules_extend_sg_name(args=args,
                                                            query_field=True)

    def test_list_security_group_rules_extend_remote(self):
        args = '-c id -c security_group -c remote'.split()

        data = [self._prepare_rule(rule_id='ruleid1', sg_id='myid1',
                                   remote_ip_prefix='172.16.18.0/24'),
                self._prepare_rule(rule_id='ruleid2', sg_id='myid2',
                                   remote_ip_prefix='172.16.20.0/24'),
                self._prepare_rule(rule_id='ruleid3', sg_id='myid2',
                                   remote_group_id='myid3')]
        expected = {'cols': ['id', 'security_group', 'remote'],
                    'data': [('ruleid1', 'group1', '172.16.18.0/24 (CIDR)'),
                             ('ruleid2', 'group2', '172.16.20.0/24 (CIDR)'),
                             ('ruleid3', 'group2', 'group3 (group)')]}
        self._test_list_security_group_rules_extend(data, expected, args)

    def test_list_security_group_rules_extend_proto_port(self):
        data = [self._prepare_rule(rule_id='ruleid1', sg_id='myid1',
                                   protocol='tcp',
                                   port_range_min=22, port_range_max=22),
                self._prepare_rule(rule_id='ruleid2', sg_id='myid2',
                                   direction='egress', ethertype='IPv6',
                                   protocol='udp',
                                   port_range_min=80, port_range_max=81),
                self._prepare_rule(rule_id='ruleid3', sg_id='myid2',
                                   protocol='icmp',
                                   remote_ip_prefix='10.2.0.0/16')]
        expected = {
            'cols': ['id', 'security_group', 'direction', 'ethertype',
                     'port/protocol', 'remote'],
            'data': [
                ('ruleid1', 'group1', 'ingress', 'IPv4', '22/tcp', 'any'),
                ('ruleid2', 'group2', 'egress', 'IPv6', '80-81/udp', 'any'),
                ('ruleid3', 'group2', 'ingress', 'IPv4', 'icmp',
                 '10.2.0.0/16 (CIDR)')
            ]}
        self._test_list_security_group_rules_extend(data, expected)

    def _prepare_rule(self, rule_id=None, sg_id=None, tenant_id=None,
                      direction=None, ethertype=None,
                      protocol=None, port_range_min=None, port_range_max=None,
                      remote_ip_prefix=None, remote_group_id=None,
                      filters=None):
        rule = {'id': rule_id or uuidutils.generate_uuid(),
                'tenant_id': tenant_id or uuidutils.generate_uuid(),
                'security_group_id': sg_id or uuidutils.generate_uuid(),
                'direction': direction or 'ingress',
                'ethertype': ethertype or 'IPv4',
                'protocol': protocol,
                'port_range_min': port_range_min,
                'port_range_max': port_range_max,
                'remote_ip_prefix': remote_ip_prefix,
                'remote_group_id': remote_group_id}
        if filters:
            return dict([(k, v) for k, v in rule.items() if k in filters])
        else:
            return rule

    def test__get_remote_both_unspecified(self):
        sg_rule = self._prepare_rule(remote_ip_prefix=None,
                                     remote_group_id=None)
        self.assertIsNone(securitygroup._get_remote(sg_rule))

    def test__get_remote_remote_ip_prefix_specified(self):
        sg_rule = self._prepare_rule(remote_ip_prefix='172.16.18.0/24')
        self.assertEqual('172.16.18.0/24 (CIDR)',
                         securitygroup._get_remote(sg_rule))

    def test__get_remote_remote_group_specified(self):
        sg_rule = self._prepare_rule(remote_group_id='sg_id1')
        self.assertEqual('sg_id1 (group)', securitygroup._get_remote(sg_rule))

    def test__get_protocol_port_all_none(self):
        sg_rule = self._prepare_rule()
        self.assertIsNone(securitygroup._get_protocol_port(sg_rule))

    def test__get_protocol_port_tcp_all_port(self):
        sg_rule = self._prepare_rule(protocol='tcp')
        self.assertEqual('tcp', securitygroup._get_protocol_port(sg_rule))

    def test__get_protocol_port_tcp_one_port(self):
        sg_rule = self._prepare_rule(protocol='tcp',
                                     port_range_min=22, port_range_max=22)
        self.assertEqual('22/tcp', securitygroup._get_protocol_port(sg_rule))

    def test__get_protocol_port_tcp_port_range(self):
        sg_rule = self._prepare_rule(protocol='tcp',
                                     port_range_min=5000, port_range_max=5010)
        self.assertEqual('5000-5010/tcp',
                         securitygroup._get_protocol_port(sg_rule))

    def test__get_protocol_port_udp_all_port(self):
        sg_rule = self._prepare_rule(protocol='udp')
        self.assertEqual('udp', securitygroup._get_protocol_port(sg_rule))

    def test__get_protocol_port_udp_one_port(self):
        sg_rule = self._prepare_rule(protocol='udp',
                                     port_range_min=22, port_range_max=22)
        self.assertEqual('22/udp', securitygroup._get_protocol_port(sg_rule))

    def test__get_protocol_port_udp_port_range(self):
        sg_rule = self._prepare_rule(protocol='udp',
                                     port_range_min=5000, port_range_max=5010)
        self.assertEqual('5000-5010/udp',
                         securitygroup._get_protocol_port(sg_rule))

    def test__get_protocol_port_icmp_all(self):
        sg_rule = self._prepare_rule(protocol='icmp')
        self.assertEqual('icmp', securitygroup._get_protocol_port(sg_rule))

    def test_get_ethertype_for_protocol_icmpv6(self):
        self.assertEqual('IPv6',
                         securitygroup.generate_default_ethertype('icmpv6'))

    def test_get_ethertype_for_protocol_icmp(self):
        self.assertEqual('IPv4',
                         securitygroup.generate_default_ethertype('icmp'))

    def test__get_protocol_port_udp_code_type(self):
        sg_rule = self._prepare_rule(protocol='icmp',
                                     port_range_min=1, port_range_max=8)
        self.assertEqual('icmp (type:1, code:8)',
                         securitygroup._get_protocol_port(sg_rule))

    def test__format_sg_rules(self):
        rules = [self._prepare_rule(),
                 self._prepare_rule(protocol='tcp', port_range_min=80,
                                    port_range_max=80),
                 self._prepare_rule(remote_ip_prefix='192.168.1.0/24'),
                 self._prepare_rule(remote_group_id='group1'),
                 self._prepare_rule(protocol='tcp',
                                    remote_ip_prefix='10.1.1.0/24'),
                 self._prepare_rule(direction='egress'),
                 self._prepare_rule(direction='egress', ethertype='IPv6'),
                 ]
        sg = {'security_group_rules': rules}
        expected_data = ['ingress, IPv4',
                         'ingress, IPv4, 80/tcp',
                         'ingress, IPv4, remote_ip_prefix: 192.168.1.0/24',
                         'ingress, IPv4, remote_group_id: group1',
                         'ingress, IPv4, tcp, remote_ip_prefix: 10.1.1.0/24',
                         'egress, IPv4',
                         'egress, IPv6',
                         ]
        expected = '\n'.join(sorted(expected_data))
        self.assertEqual(expected, securitygroup._format_sg_rules(sg))