<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_lib.api.definitions import servicetype as svctype_apidef
from neutron_lib import context
from neutron_lib import exceptions as n_exc
from neutron_lib.plugins import constants
from oslo_config import cfg
from oslo_utils import uuidutils
import webob.exc as webexc
import webtest

from neutron.api import extensions
from neutron.db.models import servicetype as st_model
from neutron.db import servicetype_db as st_db
from neutron.extensions import servicetype
from neutron.services import provider_configuration as provconf
from neutron.tests.unit.api import test_extensions
from neutron.tests.unit.api.v2 import test_base
from neutron.tests.unit.db import test_db_base_plugin_v2
from neutron.tests.unit import dummy_plugin as dp
from neutron.tests.unit import testlib_api


_uuid = test_base._uuid
_get_path = test_base._get_path

PLUGIN_NAME = 'neutron.db.db_base_plugin_v2.NeutronDbPluginV2'


class ServiceTypeManagerTestCase(testlib_api.SqlTestCase):
    def setUp(self):
        self.service_providers = mock.patch.object(
            provconf.NeutronModule, 'service_providers').start()
        super(ServiceTypeManagerTestCase, self).setUp()
        self.ctx = context.get_admin_context()
        self.setup_coreplugin(PLUGIN_NAME)

    def _set_override(self, service_providers):
        self.service_providers.return_value = service_providers
        st_db.ServiceTypeManager._instance = None
        self.manager = st_db.ServiceTypeManager.get_instance()
        for provider in service_providers:
            self.manager.add_provider_configuration(
                provider.split(':')[0], provconf.ProviderConfiguration())

    def test_service_provider_driver_not_unique(self):
        self._set_override([constants.LOADBALANCER + ':lbaas:driver'])
        prov = {'service_type': constants.LOADBALANCER,
                'name': 'name2',
                'driver': 'driver',
                'default': False}
        self.assertRaises(
            n_exc.Invalid,
            self.manager.config['LOADBALANCER'].add_provider, prov)

    def test_get_service_providers(self):
        """Test that get_service_providers filters correctly."""
        self._set_override(
            [constants.LOADBALANCER +
             ':lbaas:driver_path1',
             constants.FIREWALL +
             ':fwaas:driver_path2'])
        ctx = context.get_admin_context()
        res = self.manager.get_service_providers(
            ctx,
            filters=dict(service_type=[constants.LOADBALANCER])
        )
        self.assertEqual(1, len(res))

        res = self.manager.get_service_providers(
            ctx,
            filters=dict(service_type=[constants.FIREWALL])
        )
        self.assertEqual(1, len(res))

    def test_multiple_default_providers_specified_for_service(self):
        self.assertRaises(
            n_exc.Invalid,
            self._set_override,
            [constants.LOADBALANCER +
            ':lbaas1:driver_path:default',
            constants.LOADBALANCER +
            ':lbaas2:driver_path:default'])

    def test_get_default_provider(self):
        self._set_override([constants.LOADBALANCER +
                            ':lbaas1:driver_path:default',
                            dp.DUMMY_SERVICE_TYPE +
                            ':lbaas2:driver_path2'])
        # can pass None as a context
        p = self.manager.get_default_service_provider(None,
                                                      constants.LOADBALANCER)
        self.assertEqual({'service_type': constants.LOADBALANCER,
                          'name': 'lbaas1',
                          'driver': 'driver_path',
                          'default': True}, p)

        self.assertRaises(
            provconf.DefaultServiceProviderNotFound,
            self.manager.get_default_service_provider,
            None, dp.DUMMY_SERVICE_TYPE
        )

    def test_get_provider_names_by_resource_ids(self):
        self._set_override([dp.DUMMY_SERVICE_TYPE +
                            ':dummy1:driver_path',
                            dp.DUMMY_SERVICE_TYPE +
                            ':dummy2:driver_path2'])
        ctx = context.get_admin_context()
        test_data = [{'provider_name': 'dummy1',
                      'resource_id': uuidutils.generate_uuid()},
                     {'provider_name': 'dummy1',
                      'resource_id': uuidutils.generate_uuid()},
                     {'provider_name': 'dummy2',
                      'resource_id': uuidutils.generate_uuid()}]
        self.manager.add_resource_association(ctx,
                                              dp.DUMMY_SERVICE_TYPE,
                                              **test_data[0])
        self.manager.add_resource_association(ctx,
                                              dp.DUMMY_SERVICE_TYPE,
                                              **test_data[1])
        self.manager.add_resource_association(ctx,
                                              dp.DUMMY_SERVICE_TYPE,
                                              **test_data[2])
        names_by_id = self.manager.get_provider_names_by_resource_ids(
            ctx, [td['resource_id'] for td in test_data])
        # unmatched IDs will be excluded from the result
        self.assertEqual({td['resource_id']: td['provider_name']
                          for td in test_data}, names_by_id)

    def test_add_resource_association(self):
        self._set_override([constants.LOADBALANCER +
                            ':lbaas1:driver_path:default',
                            dp.DUMMY_SERVICE_TYPE +
                            ':lbaas2:driver_path2'])
        ctx = context.get_admin_context()
        self.manager.add_resource_association(ctx,
                                              constants.LOADBALANCER,
                                              'lbaas1',
                                              uuidutils.generate_uuid())
        self.assertEqual(ctx.session.
                         query(st_model.ProviderResourceAssociation).count(),
                         1)
        assoc = ctx.session.query(st_model.ProviderResourceAssociation).one()
        ctx.session.delete(assoc)

    def test_invalid_resource_association(self):
        self._set_override([constants.LOADBALANCER +
                            ':lbaas1:driver_path:default',
                            dp.DUMMY_SERVICE_TYPE +
                            ':lbaas2:driver_path2'])
        ctx = context.get_admin_context()
        self.assertRaises(provconf.ServiceProviderNotFound,
                          self.manager.add_resource_association,
                          ctx, 'BLABLA_svc', 'name', '123-123')


class TestServiceTypeExtensionManager(object):<commit_msg>test_servicetype: Service Type model to OVO<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_lib.api.definitions import servicetype as svctype_apidef
from neutron_lib import context
from neutron_lib import exceptions as n_exc
from neutron_lib.plugins import constants
from oslo_config import cfg
from oslo_utils import uuidutils
import webob.exc as webexc
import webtest

from neutron.api import extensions
from neutron.db import servicetype_db as st_db
from neutron.extensions import servicetype
from neutron.objects import servicetype as servicetype_obj
from neutron.services import provider_configuration as provconf
from neutron.tests.unit.api import test_extensions
from neutron.tests.unit.api.v2 import test_base
from neutron.tests.unit.db import test_db_base_plugin_v2
from neutron.tests.unit import dummy_plugin as dp
from neutron.tests.unit import testlib_api


_uuid = test_base._uuid
_get_path = test_base._get_path

PLUGIN_NAME = 'neutron.db.db_base_plugin_v2.NeutronDbPluginV2'


class ServiceTypeManagerTestCase(testlib_api.SqlTestCase):
    def setUp(self):
        self.service_providers = mock.patch.object(
            provconf.NeutronModule, 'service_providers').start()
        super(ServiceTypeManagerTestCase, self).setUp()
        self.ctx = context.get_admin_context()
        self.setup_coreplugin(PLUGIN_NAME)

    def _set_override(self, service_providers):
        self.service_providers.return_value = service_providers
        st_db.ServiceTypeManager._instance = None
        self.manager = st_db.ServiceTypeManager.get_instance()
        for provider in service_providers:
            self.manager.add_provider_configuration(
                provider.split(':')[0], provconf.ProviderConfiguration())

    def test_service_provider_driver_not_unique(self):
        self._set_override([constants.LOADBALANCER + ':lbaas:driver'])
        prov = {'service_type': constants.LOADBALANCER,
                'name': 'name2',
                'driver': 'driver',
                'default': False}
        self.assertRaises(
            n_exc.Invalid,
            self.manager.config['LOADBALANCER'].add_provider, prov)

    def test_get_service_providers(self):
        """Test that get_service_providers filters correctly."""
        self._set_override(
            [constants.LOADBALANCER +
             ':lbaas:driver_path1',
             constants.FIREWALL +
             ':fwaas:driver_path2'])
        ctx = context.get_admin_context()
        res = self.manager.get_service_providers(
            ctx,
            filters=dict(service_type=[constants.LOADBALANCER])
        )
        self.assertEqual(1, len(res))

        res = self.manager.get_service_providers(
            ctx,
            filters=dict(service_type=[constants.FIREWALL])
        )
        self.assertEqual(1, len(res))

    def test_multiple_default_providers_specified_for_service(self):
        self.assertRaises(
            n_exc.Invalid,
            self._set_override,
            [constants.LOADBALANCER +
            ':lbaas1:driver_path:default',
            constants.LOADBALANCER +
            ':lbaas2:driver_path:default'])

    def test_get_default_provider(self):
        self._set_override([constants.LOADBALANCER +
                            ':lbaas1:driver_path:default',
                            dp.DUMMY_SERVICE_TYPE +
                            ':lbaas2:driver_path2'])
        # can pass None as a context
        p = self.manager.get_default_service_provider(None,
                                                      constants.LOADBALANCER)
        self.assertEqual({'service_type': constants.LOADBALANCER,
                          'name': 'lbaas1',
                          'driver': 'driver_path',
                          'default': True}, p)

        self.assertRaises(
            provconf.DefaultServiceProviderNotFound,
            self.manager.get_default_service_provider,
            None, dp.DUMMY_SERVICE_TYPE
        )

    def test_get_provider_names_by_resource_ids(self):
        self._set_override([dp.DUMMY_SERVICE_TYPE +
                            ':dummy1:driver_path',
                            dp.DUMMY_SERVICE_TYPE +
                            ':dummy2:driver_path2'])
        ctx = context.get_admin_context()
        test_data = [{'provider_name': 'dummy1',
                      'resource_id': uuidutils.generate_uuid()},
                     {'provider_name': 'dummy1',
                      'resource_id': uuidutils.generate_uuid()},
                     {'provider_name': 'dummy2',
                      'resource_id': uuidutils.generate_uuid()}]
        self.manager.add_resource_association(ctx,
                                              dp.DUMMY_SERVICE_TYPE,
                                              **test_data[0])
        self.manager.add_resource_association(ctx,
                                              dp.DUMMY_SERVICE_TYPE,
                                              **test_data[1])
        self.manager.add_resource_association(ctx,
                                              dp.DUMMY_SERVICE_TYPE,
                                              **test_data[2])
        names_by_id = self.manager.get_provider_names_by_resource_ids(
            ctx, [td['resource_id'] for td in test_data])
        # unmatched IDs will be excluded from the result
        self.assertEqual({td['resource_id']: td['provider_name']
                          for td in test_data}, names_by_id)

    def test_add_resource_association(self):
        self._set_override([constants.LOADBALANCER +
                            ':lbaas1:driver_path:default',
                            dp.DUMMY_SERVICE_TYPE +
                            ':lbaas2:driver_path2'])
        ctx = context.get_admin_context()
        self.manager.add_resource_association(ctx,
                                              constants.LOADBALANCER,
                                              'lbaas1',
                                              uuidutils.generate_uuid())
        self.assertEqual(
            1, servicetype_obj.ProviderResourceAssociation.count(ctx))
        servicetype_obj.ProviderResourceAssociation.delete_objects(ctx)

    def test_invalid_resource_association(self):
        self._set_override([constants.LOADBALANCER +
                            ':lbaas1:driver_path:default',
                            dp.DUMMY_SERVICE_TYPE +
                            ':lbaas2:driver_path2'])
        ctx = context.get_admin_context()
        self.assertRaises(provconf.ServiceProviderNotFound,
                          self.manager.add_resource_association,
                          ctx, 'BLABLA_svc', 'name', '123-123')


class TestServiceTypeExtensionManager(object):
{'fraction_non_alphanumeric': 0.08042478354978355, 'fraction_numerical': 0.006358225108225108, 'mean_word_length': 2.025992632009824, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 40, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2134539', 'n_tokens_mistral': 4020, 'n_tokens_neox': 3875, 'n_words': 748}