<commit_before># Copyright (c) 2016 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
import requests

from neutron.plugins.ml2 import plugin
from neutron.tests import base
from neutron.tests.unit.db import test_db_base_plugin_v2
from neutron.tests.unit import testlib_api
from neutron_lib.api.definitions import external_net
from neutron_lib.plugins import constants
from neutron_lib.plugins import directory
from oslo_config import cfg
from oslo_config import fixture as config_fixture
from oslo_serialization import jsonutils
from oslo_utils import uuidutils

from networking_odl.common import client
from networking_odl.common import constants as odl_const
from networking_odl.common import filters
from networking_odl.db import db
from networking_odl.journal import journal
from networking_odl.ml2 import mech_driver_v2
from networking_odl.tests import base as odl_base
from networking_odl.tests.unit import test_base_db

EMPTY_DEP = {'gw_port_id': None}
FLOATINGIP_ID = uuidutils.generate_uuid()
NETWORK_ID = uuidutils.generate_uuid()
ROUTER_ID = uuidutils.generate_uuid()
SUBNET_ID = uuidutils.generate_uuid()
PORT_ID = uuidutils.generate_uuid()


class OpenDayLightMechanismConfigTests(testlib_api.SqlTestCase):
    def setUp(self):
        self.useFixture(odl_base.OpenDaylightRestClientFixture())
        self.useFixture(odl_base.OpenDaylightFeaturesFixture())
        self.cfg = self.useFixture(config_fixture.Config())
        self.useFixture(odl_base.OpenDaylightJournalThreadFixture())
        super(OpenDayLightMechanismConfigTests, self).setUp()
        self.cfg.config(mechanism_drivers=[
                        'logger', 'opendaylight_v2'], group='ml2')
        self.cfg.config(
            port_binding_controller='legacy-port-binding', group='ml2_odl')

    def _set_config(self, url='http://127.0.0.1:9999', username='someuser',
                    password='<PASSWORD>'):
        self.cfg.config(url=url, group='ml2_odl')
        self.cfg.config(username=username, group='ml2_odl')
        self.cfg.config(password=password, group='ml2_odl')

    def _test_missing_config(self, **kwargs):
        self._set_config(**kwargs)
        self.assertRaisesRegex(cfg.RequiredOptError,
                               r'value required for option \w+ in group '
                               r'\[ml2_odl\]',
                               plugin.Ml2Plugin)

    def test_valid_config(self):
        self._set_config()
        plugin.Ml2Plugin()

    def test_missing_url_raises_exception(self):
        self._test_missing_config(url=None)

    def test_missing_username_raises_exception(self):
        self._test_missing_config(username=None)

    def test_missing_password_raises_exception(self):
        self._test_missing_config(password=None)


class DataMatcher(object):

    def __init__(self, operation, object_type, object_dict):
        self._data = object_dict.copy()
        self._object_type = object_type
        filters.filter_for_odl(object_type, operation, self._data)

    def __eq__(self, s):
        data = jsonutils.loads(s)
        return self._data == data[self._object_type]

    def __ne__(self, s):
        return not self.__eq__(s)


class OpenDaylightL3TestCase(test_db_base_plugin_v2.NeutronDbPluginV2TestCase,
                             test_base_db.ODLBaseDbTestCase,
                             base.BaseTestCase):
    def setUp(self):
        self.cfg = self.useFixture(config_fixture.Config())
        self.cfg.config(core_plugin='neutron.plugins.ml2.plugin.Ml2Plugin')
        self.cfg.config(mechanism_drivers=[
                        'logger', 'opendaylight_v2'], group='ml2')
        self.useFixture(odl_base.OpenDaylightRestClientFixture())
        self.cfg.config(service_plugins=['odl-router_v2'])
        core_plugin = cfg.CONF.core_plugin
        service_plugins = {'l3_plugin_name': 'odl-router_v2'}
        self.useFixture(odl_base.OpenDaylightJournalThreadFixture())
        mock.patch.object(mech_driver_v2.OpenDaylightMechanismDriver,
                          '_record_in_journal').start()
        mock.patch.object(mech_driver_v2.OpenDaylightMechanismDriver,
                          'sync_from_callback_precommit').start()
        mock.patch.object(mech_driver_v2.OpenDaylightMechanismDriver,
                          'sync_from_callback_postcommit').start()
        self.useFixture(odl_base.OpenDaylightPeriodicTaskFixture())
        self.useFixture(odl_base.OpenDaylightFeaturesFixture())
        self.useFixture(odl_base.OpenDaylightPseudoAgentPrePopulateFixture())
        super(OpenDaylightL3TestCase, self).setUp(
            plugin=core_plugin, service_plugins=service_plugins)
        self.plugin = directory.get_plugin()
        self.plugin._network_is_external = mock.Mock(return_value=True)
        self.driver = directory.get_plugin(constants.L3)
        self.thread = journal.OpenDaylightJournalThread()
        self.driver.get_floatingip = mock.Mock(
            return_value={'router_id': ROUTER_ID,
                          'floating_network_id': NETWORK_ID})

    @staticmethod
    def _get_mock_router_operation_info(network, subnet):
        router = {odl_const.ODL_ROUTER:
                  {'name': 'router1',
                   'admin_state_up': True,
                   'tenant_id': network['network']['tenant_id'],
                   'external_gateway_info': {'network_id':
                                             network['network']['id']}}}
        return router

    @staticmethod
    def _get_mock_floatingip_operation_info(network, subnet):
        floatingip = {odl_const.ODL_FLOATINGIP:
                      {'floating_network_id': network['network']['id'],
                       'tenant_id': network['network']['tenant_id'],
                       'subnet_id': None,
                       'floating_ip_address': None}}
        return floatingip

    @staticmethod
    def _get_mock_router_interface_operation_info(network, subnet):
        router_intf_dict = {'subnet_id': subnet['subnet']['id'],
                            'id': network['network']['id']}
        return router_intf_dict

    @classmethod
    def _get_mock_operation_info(cls, object_type, *args):
        getter = getattr(cls, '_get_mock_' + object_type + '_operation_info')
        return getter(*args)

    @classmethod
    def _get_mock_request_response(cls, status_code):
        response = mock.Mock(status_code=status_code)
        response.raise_for_status = mock.Mock() if status_code < 400 else (
            mock.Mock(side_effect=requests.exceptions.HTTPError(
                cls._status_code_msgs[status_code])))
        return response

    def _test_operation(self, status_code, expected_calls, *args, **kwargs):
        request_response = self._get_mock_request_response(status_code)
        with mock.patch('requests.sessions.Session.request',
                        return_value=request_response) as mock_method:
            self.thread.sync_pending_entries()

        if expected_calls:
            mock_method.assert_called_with(
                headers={'Content-Type': 'application/json'},
                timeout=cfg.CONF.ml2_odl.timeout, *args, **kwargs)
        self.assertEqual(expected_calls, mock_method.call_count)

    def _call_operation_object(self, operation, object_type, object_id,
                               network, subnet):
        object_dict = self._get_mock_operation_info(
            object_type, network, subnet)
        method = getattr(self.driver, operation + '_' + object_type)

        if operation == odl_const.ODL_CREATE:
            new_object_dict = method(self.db_context, object_dict)
        elif operation == odl_const.ODL_UPDATE:
            new_object_dict = method(self.db_context, object_id, object_dict)
        else:
            new_object_dict = method(self.db_context, object_id)

        return new_object_dict<commit_msg>Removing unused variable<commit_after># Copyright (c) 2016 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
import requests

from neutron.plugins.ml2 import plugin
from neutron.tests import base
from neutron.tests.unit.db import test_db_base_plugin_v2
from neutron.tests.unit import testlib_api
from neutron_lib.api.definitions import external_net
from neutron_lib.plugins import constants
from neutron_lib.plugins import directory
from oslo_config import cfg
from oslo_config import fixture as config_fixture
from oslo_serialization import jsonutils
from oslo_utils import uuidutils

from networking_odl.common import client
from networking_odl.common import constants as odl_const
from networking_odl.common import filters
from networking_odl.db import db
from networking_odl.journal import journal
from networking_odl.ml2 import mech_driver_v2
from networking_odl.tests import base as odl_base
from networking_odl.tests.unit import test_base_db

EMPTY_DEP = {'gw_port_id': None}
FLOATINGIP_ID = uuidutils.generate_uuid()
NETWORK_ID = uuidutils.generate_uuid()
ROUTER_ID = uuidutils.generate_uuid()
SUBNET_ID = uuidutils.generate_uuid()
PORT_ID = uuidutils.generate_uuid()


class OpenDayLightMechanismConfigTests(testlib_api.SqlTestCase):
    def setUp(self):
        self.useFixture(odl_base.OpenDaylightRestClientFixture())
        self.useFixture(odl_base.OpenDaylightFeaturesFixture())
        self.cfg = self.useFixture(config_fixture.Config())
        self.useFixture(odl_base.OpenDaylightJournalThreadFixture())
        super(OpenDayLightMechanismConfigTests, self).setUp()
        self.cfg.config(mechanism_drivers=[
                        'logger', 'opendaylight_v2'], group='ml2')
        self.cfg.config(
            port_binding_controller='legacy-port-binding', group='ml2_odl')

    def _set_config(self, url='http://127.0.0.1:9999', username='someuser',
                    password='<PASSWORD>'):
        self.cfg.config(url=url, group='ml2_odl')
        self.cfg.config(username=username, group='ml2_odl')
        self.cfg.config(password=password, group='ml2_odl')

    def _test_missing_config(self, **kwargs):
        self._set_config(**kwargs)
        self.assertRaisesRegex(cfg.RequiredOptError,
                               r'value required for option \w+ in group '
                               r'\[ml2_odl\]',
                               plugin.Ml2Plugin)

    def test_valid_config(self):
        self._set_config()
        plugin.Ml2Plugin()

    def test_missing_url_raises_exception(self):
        self._test_missing_config(url=None)

    def test_missing_username_raises_exception(self):
        self._test_missing_config(username=None)

    def test_missing_password_raises_exception(self):
        self._test_missing_config(password=None)


class DataMatcher(object):

    def __init__(self, operation, object_type, object_dict):
        self._data = object_dict.copy()
        self._object_type = object_type
        filters.filter_for_odl(object_type, operation, self._data)

    def __eq__(self, s):
        data = jsonutils.loads(s)
        return self._data == data[self._object_type]

    def __ne__(self, s):
        return not self.__eq__(s)


class OpenDaylightL3TestCase(test_db_base_plugin_v2.NeutronDbPluginV2TestCase,
                             test_base_db.ODLBaseDbTestCase,
                             base.BaseTestCase):
    def setUp(self):
        self.cfg = self.useFixture(config_fixture.Config())
        self.cfg.config(core_plugin='neutron.plugins.ml2.plugin.Ml2Plugin')
        self.cfg.config(mechanism_drivers=[
                        'logger', 'opendaylight_v2'], group='ml2')
        self.useFixture(odl_base.OpenDaylightRestClientFixture())
        self.cfg.config(service_plugins=['odl-router_v2'])
        core_plugin = cfg.CONF.core_plugin
        service_plugins = {'l3_plugin_name': 'odl-router_v2'}
        self.useFixture(odl_base.OpenDaylightJournalThreadFixture())
        mock.patch.object(mech_driver_v2.OpenDaylightMechanismDriver,
                          '_record_in_journal').start()
        mock.patch.object(mech_driver_v2.OpenDaylightMechanismDriver,
                          'sync_from_callback_precommit').start()
        mock.patch.object(mech_driver_v2.OpenDaylightMechanismDriver,
                          'sync_from_callback_postcommit').start()
        self.useFixture(odl_base.OpenDaylightPeriodicTaskFixture())
        self.useFixture(odl_base.OpenDaylightFeaturesFixture())
        self.useFixture(odl_base.OpenDaylightPseudoAgentPrePopulateFixture())
        super(OpenDaylightL3TestCase, self).setUp(
            plugin=core_plugin, service_plugins=service_plugins)
        self.plugin = directory.get_plugin()
        self.plugin._network_is_external = mock.Mock(return_value=True)
        self.driver = directory.get_plugin(constants.L3)
        self.thread = journal.OpenDaylightJournalThread()
        self.driver.get_floatingip = mock.Mock(
            return_value={'router_id': ROUTER_ID,
                          'floating_network_id': NETWORK_ID})

    @staticmethod
    def _get_mock_router_operation_info(network, subnet):
        router = {odl_const.ODL_ROUTER:
                  {'name': 'router1',
                   'admin_state_up': True,
                   'tenant_id': network['network']['tenant_id'],
                   'external_gateway_info': {'network_id':
                                             network['network']['id']}}}
        return router

    @staticmethod
    def _get_mock_floatingip_operation_info(network, subnet):
        floatingip = {odl_const.ODL_FLOATINGIP:
                      {'floating_network_id': network['network']['id'],
                       'tenant_id': network['network']['tenant_id'],
                       'subnet_id': None,
                       'floating_ip_address': None}}
        return floatingip

    @staticmethod
    def _get_mock_router_interface_operation_info(network, subnet):
        router_intf_dict = {'subnet_id': subnet['subnet']['id'],
                            'id': network['network']['id']}
        return router_intf_dict

    @classmethod
    def _get_mock_operation_info(cls, object_type, *args):
        getter = getattr(cls, '_get_mock_' + object_type + '_operation_info')
        return getter(*args)

    @classmethod
    def _get_mock_request_response(cls, status_code):
        response = mock.Mock(status_code=status_code)
        response.raise_for_status = mock.Mock() if status_code < 400 else (
            mock.Mock(side_effect=requests.exceptions.HTTPError(status_code)))
        return response

    def _test_operation(self, status_code, expected_calls, *args, **kwargs):
        request_response = self._get_mock_request_response(status_code)
        with mock.patch('requests.sessions.Session.request',
                        return_value=request_response) as mock_method:
            self.thread.sync_pending_entries()

        if expected_calls:
            mock_method.assert_called_with(
                headers={'Content-Type': 'application/json'},
                timeout=cfg.CONF.ml2_odl.timeout, *args, **kwargs)
        self.assertEqual(expected_calls, mock_method.call_count)

    def _call_operation_object(self, operation, object_type, object_id,
                               network, subnet):
        object_dict = self._get_mock_operation_info(
            object_type, network, subnet)
        method = getattr(self.driver, operation + '_' + object_type)

        if operation == odl_const.ODL_CREATE:
            new_object_dict = method(self.db_context, object_dict)
        elif operation == odl_const.ODL_UPDATE:
            new_object_dict = method(self.db_context, object_id, object_dict)
        else:
            new_object_dict = method(self.db_context, object_id)

        return new_object_dict
{'fraction_non_alphanumeric': 0.09097374503644877, 'fraction_numerical': 0.006045160907959462, 'mean_word_length': 2.9806558150507194, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 2, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3740334', 'n_tokens_mistral': 4989, 'n_tokens_neox': 4895, 'n_words': 898}