<commit_before># Copyright 2014 OpenStack Foundation
# 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 socket

import netaddr

from tempest.api.network import base
from tempest.api.network import base_security_groups as sec_base
from tempest.common import custom_matchers
from tempest.common.utils import data_utils
from tempest import config
from tempest import exceptions
from tempest import test

CONF = config.CONF


class PortsTestJSON(sec_base.BaseSecGroupTest):
    """Test the following operations for ports:

        port create
        port delete
        port list
        port show
        port update
    """

    @classmethod
    def resource_setup(cls):
        super(PortsTestJSON, cls).resource_setup()
        cls.network = cls.create_network()
        cls.port = cls.create_port(cls.network)

    def _delete_port(self, port_id):
        self.ports_client.delete_port(port_id)
        body = self.ports_client.list_ports()
        ports_list = body['ports']
        self.assertFalse(port_id in [n['id'] for n in ports_list])

    @test.attr(type='smoke')
    @test.idempotent_id('c72c1c0c-2193-4aca-aaa4-b1442640f51c')
    def test_create_update_delete_port(self):
        # Verify port creation
        body = self.ports_client.create_port(network_id=self.network['id'])
        port = body['port']
        # Schedule port deletion with verification upon test completion
        self.addCleanup(self._delete_port, port['id'])
        self.assertTrue(port['admin_state_up'])
        # Verify port update
        new_name = "New_Port"
        body = self.ports_client.update_port(port['id'],
                                             name=new_name,
                                             admin_state_up=False)
        updated_port = body['port']
        self.assertEqual(updated_port['name'], new_name)
        self.assertFalse(updated_port['admin_state_up'])

    @test.idempotent_id('67f1b811-f8db-43e2-86bd-72c074d4a42c')
    def test_create_bulk_port(self):
        network1 = self.network
        name = data_utils.rand_name('network-')
        network2 = self.create_network(network_name=name)
        network_list = [network1['id'], network2['id']]
        port_list = [{'network_id': net_id} for net_id in network_list]
        body = self.ports_client.create_bulk_ports(ports=port_list)
        created_ports = body['ports']
        port1 = created_ports[0]
        port2 = created_ports[1]
        self.addCleanup(self._delete_port, port1['id'])
        self.addCleanup(self._delete_port, port2['id'])
        self.assertEqual(port1['network_id'], network1['id'])
        self.assertEqual(port2['network_id'], network2['id'])
        self.assertTrue(port1['admin_state_up'])
        self.assertTrue(port2['admin_state_up'])

    @classmethod
    def _get_ipaddress_from_tempest_conf(cls):
        """Return subnet with mask bits for configured CIDR """
        if cls._ip_version == 4:
            cidr = netaddr.IPNetwork(CONF.network.tenant_network_cidr)
            cidr.prefixlen = CONF.network.tenant_network_mask_bits

        elif cls._ip_version == 6:
            cidr = netaddr.IPNetwork(CONF.network.tenant_network_v6_cidr)
            cidr.prefixlen = CONF.network.tenant_network_v6_mask_bits

        return cidr

    @test.attr(type='smoke')
    @test.idempotent_id('0435f278-40ae-48cb-a404-b8a087bc09b1')
    def test_create_port_in_allowed_allocation_pools(self):
        network = self.create_network()
        net_id = network['id']
        address = self._get_ipaddress_from_tempest_conf()
        if ((address.version == 4 and address.prefixlen >= 30) or
           (address.version == 6 and address.prefixlen >= 126)):
            msg = ("Subnet %s isn't large enough for the test" % address.cidr)
            raise exceptions.InvalidConfiguration(message=msg)
        allocation_pools = {'allocation_pools': [{'start': str(address[2]),
                                                  'end': str(address[-2])}]}
        subnet = self.create_subnet(network, cidr=address,
                                    mask_bits=address.prefixlen,
                                    **allocation_pools)
        self.addCleanup(self.subnets_client.delete_subnet, subnet['id'])
        body = self.ports_client.create_port(network_id=net_id)
        self.addCleanup(self.ports_client.delete_port, body['port']['id'])
        port = body['port']
        ip_address = port['fixed_ips'][0]['ip_address']
        start_ip_address = allocation_pools['allocation_pools'][0]['start']
        end_ip_address = allocation_pools['allocation_pools'][0]['end']
        ip_range = netaddr.IPRange(start_ip_address, end_ip_address)
        self.assertIn(ip_address, ip_range)

    @test.attr(type='smoke')
    @test.idempotent_id('c9a685bd-e83f-499c-939f-9f7863ca259f')
    def test_show_port(self):
        # Verify the details of port
        body = self.ports_client.show_port(self.port['id'])
        port = body['port']
        self.assertIn('id', port)
        # TODO(Santosh)- This is a temporary workaround to compare create_port
        # and show_port dict elements.Remove this once extra_dhcp_opts issue
        # gets fixed in neutron.( bug - 1365341.)
        self.assertThat(self.port,
                        custom_matchers.MatchesDictExceptForKeys
                        (port, excluded_keys=['extra_dhcp_opts']))

    @test.idempotent_id('45fcdaf2-dab0-4c13-ac6c-fcddfb579dbd')
    def test_show_port_fields(self):
        # Verify specific fields of a port<commit_msg>Ignore created_at/updated_at in test_show_port<commit_after># Copyright 2014 OpenStack Foundation
# 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 socket

import netaddr

from tempest.api.network import base
from tempest.api.network import base_security_groups as sec_base
from tempest.common import custom_matchers
from tempest.common.utils import data_utils
from tempest import config
from tempest import exceptions
from tempest import test

CONF = config.CONF


class PortsTestJSON(sec_base.BaseSecGroupTest):
    """Test the following operations for ports:

        port create
        port delete
        port list
        port show
        port update
    """

    @classmethod
    def resource_setup(cls):
        super(PortsTestJSON, cls).resource_setup()
        cls.network = cls.create_network()
        cls.port = cls.create_port(cls.network)

    def _delete_port(self, port_id):
        self.ports_client.delete_port(port_id)
        body = self.ports_client.list_ports()
        ports_list = body['ports']
        self.assertFalse(port_id in [n['id'] for n in ports_list])

    @test.attr(type='smoke')
    @test.idempotent_id('c72c1c0c-2193-4aca-aaa4-b1442640f51c')
    def test_create_update_delete_port(self):
        # Verify port creation
        body = self.ports_client.create_port(network_id=self.network['id'])
        port = body['port']
        # Schedule port deletion with verification upon test completion
        self.addCleanup(self._delete_port, port['id'])
        self.assertTrue(port['admin_state_up'])
        # Verify port update
        new_name = "New_Port"
        body = self.ports_client.update_port(port['id'],
                                             name=new_name,
                                             admin_state_up=False)
        updated_port = body['port']
        self.assertEqual(updated_port['name'], new_name)
        self.assertFalse(updated_port['admin_state_up'])

    @test.idempotent_id('67f1b811-f8db-43e2-86bd-72c074d4a42c')
    def test_create_bulk_port(self):
        network1 = self.network
        name = data_utils.rand_name('network-')
        network2 = self.create_network(network_name=name)
        network_list = [network1['id'], network2['id']]
        port_list = [{'network_id': net_id} for net_id in network_list]
        body = self.ports_client.create_bulk_ports(ports=port_list)
        created_ports = body['ports']
        port1 = created_ports[0]
        port2 = created_ports[1]
        self.addCleanup(self._delete_port, port1['id'])
        self.addCleanup(self._delete_port, port2['id'])
        self.assertEqual(port1['network_id'], network1['id'])
        self.assertEqual(port2['network_id'], network2['id'])
        self.assertTrue(port1['admin_state_up'])
        self.assertTrue(port2['admin_state_up'])

    @classmethod
    def _get_ipaddress_from_tempest_conf(cls):
        """Return subnet with mask bits for configured CIDR """
        if cls._ip_version == 4:
            cidr = netaddr.IPNetwork(CONF.network.tenant_network_cidr)
            cidr.prefixlen = CONF.network.tenant_network_mask_bits

        elif cls._ip_version == 6:
            cidr = netaddr.IPNetwork(CONF.network.tenant_network_v6_cidr)
            cidr.prefixlen = CONF.network.tenant_network_v6_mask_bits

        return cidr

    @test.attr(type='smoke')
    @test.idempotent_id('0435f278-40ae-48cb-a404-b8a087bc09b1')
    def test_create_port_in_allowed_allocation_pools(self):
        network = self.create_network()
        net_id = network['id']
        address = self._get_ipaddress_from_tempest_conf()
        if ((address.version == 4 and address.prefixlen >= 30) or
           (address.version == 6 and address.prefixlen >= 126)):
            msg = ("Subnet %s isn't large enough for the test" % address.cidr)
            raise exceptions.InvalidConfiguration(message=msg)
        allocation_pools = {'allocation_pools': [{'start': str(address[2]),
                                                  'end': str(address[-2])}]}
        subnet = self.create_subnet(network, cidr=address,
                                    mask_bits=address.prefixlen,
                                    **allocation_pools)
        self.addCleanup(self.subnets_client.delete_subnet, subnet['id'])
        body = self.ports_client.create_port(network_id=net_id)
        self.addCleanup(self.ports_client.delete_port, body['port']['id'])
        port = body['port']
        ip_address = port['fixed_ips'][0]['ip_address']
        start_ip_address = allocation_pools['allocation_pools'][0]['start']
        end_ip_address = allocation_pools['allocation_pools'][0]['end']
        ip_range = netaddr.IPRange(start_ip_address, end_ip_address)
        self.assertIn(ip_address, ip_range)

    @test.attr(type='smoke')
    @test.idempotent_id('c9a685bd-e83f-499c-939f-9f7863ca259f')
    def test_show_port(self):
        # Verify the details of port
        body = self.ports_client.show_port(self.port['id'])
        port = body['port']
        self.assertIn('id', port)
        # NOTE(rfolco): created_at and updated_at may get inconsistent values
        # due to possible delay between POST request and resource creation.
        # TODO(rfolco): Neutron Bug #1365341 is fixed, can remove the key
        # extra_dhcp_opts in the O release (K/L gate jobs still need it).
        self.assertThat(self.port,
                        custom_matchers.MatchesDictExceptForKeys
                        (port, excluded_keys=['extra_dhcp_opts',
                                              'created_at',
                                              'updated_at']))

    @test.idempotent_id('45fcdaf2-dab0-4c13-ac6c-fcddfb579dbd')
    def test_show_port_fields(self):
        # Verify specific fields of a port
{'fraction_non_alphanumeric': 0.10113718848294218, 'fraction_numerical': 0.022259859666102104, 'mean_word_length': 2.765563316125114, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 20, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '799867', 'n_tokens_mistral': 3855, 'n_tokens_neox': 3722, 'n_words': 778}