<commit_before># Copyright 2012 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 mock

from neutron.agent.common import config
from neutron.agent.common import ovs_lib
from neutron.agent.linux import interface
from neutron.agent.linux import ip_lib
from neutron.agent.linux import utils
from neutron.common import constants
from neutron.extensions import flavor
from neutron.openstack.common import uuidutils
from neutron.tests import base


class BaseChild(interface.LinuxInterfaceDriver):
    def plug_new(*args):
        pass

    def unplug(*args):
        pass


class FakeNetwork(object):
    id = '12345678-1234-5678-90ab-ba0987654321'


class FakeSubnet(object):
    cidr = '192.168.1.1/24'


class FakeAllocation(object):
    subnet = FakeSubnet()
    ip_address = '192.168.1.2'
    ip_version = 4


class FakePort(object):
    id = 'abcdef01-1234-5678-90ab-ba0987654321'
    fixed_ips = [FakeAllocation]
    device_id = 'cccccccc-cccc-cccc-cccc-cccccccccccc'
    network = FakeNetwork()
    network_id = network.id


class TestBase(base.BaseTestCase):
    def setUp(self):
        super(TestBase, self).setUp()
        self.conf = config.setup_conf()
        self.conf.register_opts(interface.OPTS)
        self.ip_dev_p = mock.patch.object(ip_lib, 'IPDevice')
        self.ip_dev = self.ip_dev_p.start()
        self.ip_p = mock.patch.object(ip_lib, 'IPWrapper')
        self.ip = self.ip_p.start()
        self.device_exists_p = mock.patch.object(ip_lib, 'device_exists')
        self.device_exists = self.device_exists_p.start()


class TestABCDriver(TestBase):
    def setUp(self):
        super(TestABCDriver, self).setUp()
        mock_link_addr = mock.PropertyMock(return_value='aa:bb:cc:dd:ee:ff')
        type(self.ip_dev().link).address = mock_link_addr

    def test_get_device_name(self):
        bc = BaseChild(self.conf)
        device_name = bc.get_device_name(FakePort())
        self.assertEqual('tapabcdef01-12', device_name)

    def test_l3_init(self):
        addresses = [dict(scope='global',
                          dynamic=False, cidr='172.16.77.240/24')]
        self.ip_dev().addr.list = mock.Mock(return_value=addresses)
        self.ip_dev().route.list_onlink_routes.return_value = []

        bc = BaseChild(self.conf)
        ns = '12345678-1234-5678-90ab-ba0987654321'
        bc.init_l3('tap0', ['192.168.1.2/24'], namespace=ns,
                   extra_subnets=[{'cidr': '172.20.0.0/24'}])
        self.ip_dev.assert_has_calls(
            [mock.call('tap0', namespace=ns),
             mock.call().addr.list(filters=['permanent']),
             mock.call().addr.add('192.168.1.2/24'),
             mock.call().addr.delete('172.16.77.240/24'),
             mock.call().route.list_onlink_routes(constants.IP_VERSION_4),
             mock.call().route.list_onlink_routes(constants.IP_VERSION_6),
             mock.call().route.add_onlink_route('172.20.0.0/24')])

    def test_l3_init_delete_onlink_routes(self):
        addresses = [dict(scope='global',
                          dynamic=False, cidr='172.16.77.240/24')]
        self.ip_dev().addr.list = mock.Mock(return_value=addresses)
        self.ip_dev().route.list_onlink_routes.return_value = ['172.20.0.0/24']

        bc = BaseChild(self.conf)
        ns = '12345678-1234-5678-90ab-ba0987654321'
        bc.init_l3('tap0', ['192.168.1.2/24'], namespace=ns)
        self.ip_dev.assert_has_calls(
            [mock.call().route.list_onlink_routes(constants.IP_VERSION_4),
             mock.call().route.list_onlink_routes(constants.IP_VERSION_6),
             mock.call().route.delete_onlink_route('172.20.0.0/24')])

    def test_l3_init_with_preserve(self):
        addresses = [dict(scope='global',
                          dynamic=False, cidr='192.168.1.3/32')]
        self.ip_dev().addr.list = mock.Mock(return_value=addresses)

        bc = BaseChild(self.conf)
        ns = '12345678-1234-5678-90ab-ba0987654321'
        bc.init_l3('tap0', ['192.168.1.2/24'], namespace=ns,
                   preserve_ips=['192.168.1.3/32'])
        self.ip_dev.assert_has_calls(
            [mock.call('tap0', namespace=ns),
             mock.call().addr.list(filters=['permanent']),
             mock.call().addr.add('192.168.1.2/24')])
        self.assertFalse(self.ip_dev().addr.delete.called)

    def _test_l3_init_with_ipv6(self, include_gw_ip):
        addresses = [dict(scope='global',
                          dynamic=False,
                          cidr='2001:db8:a::123/64')]
        self.ip_dev().addr.list = mock.Mock(return_value=addresses)
        self.ip_dev().route.list_onlink_routes.return_value = []

        bc = BaseChild(self.conf)
        ns = '12345678-1234-5678-90ab-ba0987654321'
        new_cidr = '2001:db8:a::124/64'
        kwargs = {'namespace': ns,
                  'extra_subnets': [{'cidr': '2001:db8:b::/64'}]}
        if include_gw_ip:
            kwargs['gateway_ips'] = ['2001:db8:a::1']
        bc.init_l3('tap0', [new_cidr], **kwargs)
        expected_calls = (
            [mock.call('tap0', namespace=ns),
             mock.call().addr.list(filters=['permanent']),
             mock.call().addr.add('2fd00:a516:7c1b:17cd:6d81:2137:bd2a:2c5b/64'),
             mock.call().addr.delete('2001:db8:a::123/64')])
        if include_gw_ip:
            expected_calls += (
                [mock.call().route.add_gateway('2001:db8:a::1')])
        expected_calls += (
             [mock.call().route.list_onlink_routes(constants.IP_VERSION_4),
              mock.call().route.list_onlink_routes(constants.IP_VERSION_6),
              mock.call().route.add_onlink_route('2001:db8:b::/64')])
        self.ip_dev.assert_has_calls(expected_calls)

    def test_l3_init_ipv6_with_gw_ip(self):
        self._test_l3_init_with_ipv6(include_gw_ip=True)

    def test_l3_init_ipv6_without_gw_ip(self):
        self._test_l3_init_with_ipv6(include_gw_ip=False)

    def test_l3_init_ext_gw_with_dual_stack(self):
        old_addrs = [dict(ip_version=4, scope='global',
                          dynamic=False, cidr='172.16.77.240/24'),
                     dict(ip_version=6, scope='global',
                          dynamic=False, cidr='2fd00:c2b6:b24b:be67:2827:688d:e6a1:6a3b/64')]
        self.ip_dev().addr.list = mock.Mock(return_value=old_addrs)
        self.ip_dev().route.list_onlink_routes.return_value = []
        bc = BaseChild(self.conf)
        ns = '12345678-1234-5678-90ab-ba0987654321'
        new_cidrs = ['192.168.1.2/24', '2fd00:a516:7c1b:17cd:6d81:2137:bd2a:2c5b/64']
        bc.init_l3('tap0', new_cidrs, namespace=ns,
                   extra_subnets=[{'cidr': '172.20.0.0/24'}])
        self.ip_dev.assert_has_calls(
            [mock.call('tap0', namespace=ns),
             mock.call().addr.list(filters=['permanent']),
             mock.call().addr.add('192.168.1.2/24'),
             mock.call().addr.add('2001:db8:a::124/64'),
             mock.call().addr.delete('172.16.77.240/24'),
             mock.call().addr.delete('2001:db8:a::123/64'),
             mock.call().route.list_onlink_routes(constants.IP_VERSION_4),
             mock.call().route.list_onlink_routes(constants.IP_VERSION_6),
             mock.call().route.add_onlink_route('172.20.0.0/24')])

    def test_l3_init_with_ipv6_delete_onlink_routes(self):
        addresses = [dict(scope='global',
                          dynamic=False, cidr='2001:db8:a::123/64')]
        route = '2001:db8:a::/64'
        self.ip_dev().addr.list = mock.Mock(return_value=addresses)<commit_msg>Do not assume order of dictionary elements in init_l3<commit_after># Copyright 2012 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 mock

from neutron.agent.common import config
from neutron.agent.common import ovs_lib
from neutron.agent.linux import interface
from neutron.agent.linux import ip_lib
from neutron.agent.linux import utils
from neutron.common import constants
from neutron.extensions import flavor
from neutron.openstack.common import uuidutils
from neutron.tests import base


class BaseChild(interface.LinuxInterfaceDriver):
    def plug_new(*args):
        pass

    def unplug(*args):
        pass


class FakeNetwork(object):
    id = '12345678-1234-5678-90ab-ba0987654321'


class FakeSubnet(object):
    cidr = '192.168.1.1/24'


class FakeAllocation(object):
    subnet = FakeSubnet()
    ip_address = '192.168.1.2'
    ip_version = 4


class FakePort(object):
    id = 'abcdef01-1234-5678-90ab-ba0987654321'
    fixed_ips = [FakeAllocation]
    device_id = 'cccccccc-cccc-cccc-cccc-cccccccccccc'
    network = FakeNetwork()
    network_id = network.id


class TestBase(base.BaseTestCase):
    def setUp(self):
        super(TestBase, self).setUp()
        self.conf = config.setup_conf()
        self.conf.register_opts(interface.OPTS)
        self.ip_dev_p = mock.patch.object(ip_lib, 'IPDevice')
        self.ip_dev = self.ip_dev_p.start()
        self.ip_p = mock.patch.object(ip_lib, 'IPWrapper')
        self.ip = self.ip_p.start()
        self.device_exists_p = mock.patch.object(ip_lib, 'device_exists')
        self.device_exists = self.device_exists_p.start()


class TestABCDriver(TestBase):
    def setUp(self):
        super(TestABCDriver, self).setUp()
        mock_link_addr = mock.PropertyMock(return_value='aa:bb:cc:dd:ee:ff')
        type(self.ip_dev().link).address = mock_link_addr

    def test_get_device_name(self):
        bc = BaseChild(self.conf)
        device_name = bc.get_device_name(FakePort())
        self.assertEqual('tapabcdef01-12', device_name)

    def test_l3_init(self):
        addresses = [dict(scope='global',
                          dynamic=False, cidr='172.16.77.240/24')]
        self.ip_dev().addr.list = mock.Mock(return_value=addresses)
        self.ip_dev().route.list_onlink_routes.return_value = []

        bc = BaseChild(self.conf)
        ns = '12345678-1234-5678-90ab-ba0987654321'
        bc.init_l3('tap0', ['192.168.1.2/24'], namespace=ns,
                   extra_subnets=[{'cidr': '172.20.0.0/24'}])
        self.ip_dev.assert_has_calls(
            [mock.call('tap0', namespace=ns),
             mock.call().addr.list(filters=['permanent']),
             mock.call().addr.add('192.168.1.2/24'),
             mock.call().addr.delete('172.16.77.240/24'),
             mock.call().route.list_onlink_routes(constants.IP_VERSION_4),
             mock.call().route.list_onlink_routes(constants.IP_VERSION_6),
             mock.call().route.add_onlink_route('172.20.0.0/24')])

    def test_l3_init_delete_onlink_routes(self):
        addresses = [dict(scope='global',
                          dynamic=False, cidr='172.16.77.240/24')]
        self.ip_dev().addr.list = mock.Mock(return_value=addresses)
        self.ip_dev().route.list_onlink_routes.return_value = ['172.20.0.0/24']

        bc = BaseChild(self.conf)
        ns = '12345678-1234-5678-90ab-ba0987654321'
        bc.init_l3('tap0', ['192.168.1.2/24'], namespace=ns)
        self.ip_dev.assert_has_calls(
            [mock.call().route.list_onlink_routes(constants.IP_VERSION_4),
             mock.call().route.list_onlink_routes(constants.IP_VERSION_6),
             mock.call().route.delete_onlink_route('172.20.0.0/24')])

    def test_l3_init_with_preserve(self):
        addresses = [dict(scope='global',
                          dynamic=False, cidr='192.168.1.3/32')]
        self.ip_dev().addr.list = mock.Mock(return_value=addresses)

        bc = BaseChild(self.conf)
        ns = '12345678-1234-5678-90ab-ba0987654321'
        bc.init_l3('tap0', ['192.168.1.2/24'], namespace=ns,
                   preserve_ips=['192.168.1.3/32'])
        self.ip_dev.assert_has_calls(
            [mock.call('tap0', namespace=ns),
             mock.call().addr.list(filters=['permanent']),
             mock.call().addr.add('192.168.1.2/24')])
        self.assertFalse(self.ip_dev().addr.delete.called)

    def _test_l3_init_with_ipv6(self, include_gw_ip):
        addresses = [dict(scope='global',
                          dynamic=False,
                          cidr='2001:db8:a::123/64')]
        self.ip_dev().addr.list = mock.Mock(return_value=addresses)
        self.ip_dev().route.list_onlink_routes.return_value = []

        bc = BaseChild(self.conf)
        ns = '12345678-1234-5678-90ab-ba0987654321'
        new_cidr = '2001:db8:a::124/64'
        kwargs = {'namespace': ns,
                  'extra_subnets': [{'cidr': '2001:db8:b::/64'}]}
        if include_gw_ip:
            kwargs['gateway_ips'] = ['2001:db8:a::1']
        bc.init_l3('tap0', [new_cidr], **kwargs)
        expected_calls = (
            [mock.call('tap0', namespace=ns),
             mock.call().addr.list(filters=['permanent']),
             mock.call().addr.add('2001:db8:a::124/64'),
             mock.call().addr.delete('2001:db8:a::123/64')])
        if include_gw_ip:
            expected_calls += (
                [mock.call().route.add_gateway('2001:db8:a::1')])
        expected_calls += (
             [mock.call().route.list_onlink_routes(constants.IP_VERSION_4),
              mock.call().route.list_onlink_routes(constants.IP_VERSION_6),
              mock.call().route.add_onlink_route('2001:db8:b::/64')])
        self.ip_dev.assert_has_calls(expected_calls)

    def test_l3_init_ipv6_with_gw_ip(self):
        self._test_l3_init_with_ipv6(include_gw_ip=True)

    def test_l3_init_ipv6_without_gw_ip(self):
        self._test_l3_init_with_ipv6(include_gw_ip=False)

    def test_l3_init_ext_gw_with_dual_stack(self):
        old_addrs = [dict(ip_version=4, scope='global',
                          dynamic=False, cidr='172.16.77.240/24'),
                     dict(ip_version=6, scope='global',
                          dynamic=False, cidr='2001:db8:a::123/64')]
        self.ip_dev().addr.list = mock.Mock(return_value=old_addrs)
        self.ip_dev().route.list_onlink_routes.return_value = []
        bc = BaseChild(self.conf)
        ns = '12345678-1234-5678-90ab-ba0987654321'
        new_cidrs = ['192.168.1.2/24', '20fdf8:f53e:61e4::18/64']
        bc.init_l3('tap0', new_cidrs, namespace=ns,
                   extra_subnets=[{'cidr': '172.20.0.0/24'}])
        self.ip_dev.assert_has_calls(
            [mock.call('tap0', namespace=ns),
             mock.call().addr.list(filters=['permanent']),
             mock.call().addr.add('192.168.1.2/24'),
             mock.call().addr.add('2001:fd00:c2b6:b24b:be67:2827:688d:e6a1:6a3b/64'),
             mock.call().addr.delete('172.16.77.240/24'),
             mock.call().addr.delete('2001:fd00:c2b6:b24b:be67:2827:688d:e6a1:6a3b/64'),
             mock.call().route.list_onlink_routes(constants.IP_VERSION_4),
             mock.call().route.list_onlink_routes(constants.IP_VERSION_6),
             mock.call().route.add_onlink_route('172.20.0.0/24')],
            any_order=True)

    def test_l3_init_with_ipv6_delete_onlink_routes(self):
        addresses = [dict(scope='global',
                          dynamic=False, cidr='2001:db8:a::123/64')]
        route = '2001:db8:a::/64'
        self.ip_dev().addr.list = mock.Mock(return_value=addresses)
{'fraction_non_alphanumeric': 0.1431655563779916, 'fraction_numerical': 0.07679496669133974, 'mean_word_length': 2.7032891731384194, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 44, 'substrings_counts': 4, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '668467', 'n_tokens_mistral': 6306, 'n_tokens_neox': 5667, 'n_words': 746}