<commit_before># Copyright (c) 2015 Eayun, Inc.
# 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.

from oslo import messaging

from neutron.db.qos import qos_db
from neutron import manager
from neutron.extensions import qos as ext_qos
from neutron.openstack.common import log as logging
from neutron.plugins.common import constants as service_constants
from neutron.plugins.ml2 import db  # hacky

LOG = logging.getLogger(__name__)

NIC_NAME_LEN = 14


class QosServerRpcServerMixin(qos_db.QosDbMixin):

    def _get_host_from_qos_id(self, context, qos_id):
        qos = self._get_qos(context, qos_id)
        if qos['router_id'] is not None:
            plugin = manager.NeutronManager.get_service_plugins().get(
                service_constants.L3_ROUTER_NAT)
            adminContext = context if context.is_admin else context.elevated()
            l3_agents = plugin.get_l3_agents_hosting_routers(
                adminContext, [qos['router_id']],
                admin_state_up=True, active=True)
            return l3_agents[0].host
        elif qos['port_id'] is not None:
            """
            plugin = manager.NeutronManager.get_plugin()
            adminContext = context if context.is_admin else context.elevated()
            return plugin.get_port_binding_host(adminContext, qos['port_id'])
            """
            return db.get_port_binding_host(qos['port_id'])
        else:
            return None

    def _get_qos_on_host(self, context, host):
        hosting_qos = dict()
        for qos in self.get_qoss(context):
            if host == self._get_host_from_qos_id(context, qos['id']):
                qos.pop('unattached_filters')
                qos.update(
                    {'devices': self._get_qos_devices(context, qos['id'])})
                hosting_qos[qos['id']] = qos

        return hosting_qos

    def _get_qos_devices(self, context, qos_id):
        ret = []
        qos_db = self._get_qos(context, qos_id)
        if qos_db.router_id is not None:
            plugin = manager.NeutronManager.get_service_plugins().get(
                service_constants.L3_ROUTER_NAT)
            adminContext = context if context.is_admin else context.elevated()
            # A bit hacky here
            router = plugin._get_router(adminContext, qos_db.router_id)
            if qos_db.direction == 'egress':
                if router.gw_port_id:
                    ret.append(("qg-%s" % router.gw_port_id)[:NIC_NAME_LEN])
            else:
                for rp in router.attached_ports:
                    if rp.port_id != router.gw_port_id:
                        ret.append(("qr-%s" % rp.port_id)[:NIC_NAME_LEN])
        elif qos_db.port_id is not None:
            prefix = 'qvo' if qos_db.direction == 'egress' else 'qvb'
            ret.append(("%s%s" % (prefix, qos_db.port_id))[:NIC_NAME_LEN])

        return ret

    def _check_port(self, context, port_id):
        plugin = manager.NeutronManager.get_plugin()
        adminContext = context if context.is_admin else context.elevated()
        port = plugin.get_port(adminContext, port_id)
        if not port['device_owner'].startswith('compute'):
            raise ext_qos.QosInvalidPortType(
                port_id=port_id, port_type=port['device_owner'])

    def _check_router(self, context, router_id):
        plugin = manager.NeutronManager.get_service_plugins().get(
            service_constants.L3_ROUTER_NAT)
        adminContext = context if context.is_admin else context.elevated()
        plugin.get_router(adminContext, router_id)

    def sync_qos(self, context, qos_list, host):
        hosting_qos = self._get_qos_on_host(context, host)
        deleted_qos = filter(lambda qos: qos not in hosting_qos, qos_list)
<commit_msg>Revert "Revert "Fix direction problem when target_type is port""<commit_after># Copyright (c) 2015 Eayun, Inc.
# 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.

from oslo import messaging

from neutron.db.qos import qos_db
from neutron import manager
from neutron.extensions import qos as ext_qos
from neutron.openstack.common import log as logging
from neutron.plugins.common import constants as service_constants
from neutron.plugins.ml2 import db  # hacky

LOG = logging.getLogger(__name__)

NIC_NAME_LEN = 14


class QosServerRpcServerMixin(qos_db.QosDbMixin):

    def _get_host_from_qos_id(self, context, qos_id):
        qos = self._get_qos(context, qos_id)
        if qos['router_id'] is not None:
            plugin = manager.NeutronManager.get_service_plugins().get(
                service_constants.L3_ROUTER_NAT)
            adminContext = context if context.is_admin else context.elevated()
            l3_agents = plugin.get_l3_agents_hosting_routers(
                adminContext, [qos['router_id']],
                admin_state_up=True, active=True)
            return l3_agents[0].host
        elif qos['port_id'] is not None:
            """
            plugin = manager.NeutronManager.get_plugin()
            adminContext = context if context.is_admin else context.elevated()
            return plugin.get_port_binding_host(adminContext, qos['port_id'])
            """
            return db.get_port_binding_host(qos['port_id'])
        else:
            return None

    def _get_qos_on_host(self, context, host):
        hosting_qos = dict()
        for qos in self.get_qoss(context):
            if host == self._get_host_from_qos_id(context, qos['id']):
                qos.pop('unattached_filters')
                qos.update(
                    {'devices': self._get_qos_devices(context, qos['id'])})
                hosting_qos[qos['id']] = qos

        return hosting_qos

    def _get_qos_devices(self, context, qos_id):
        ret = []
        qos_db = self._get_qos(context, qos_id)
        if qos_db.router_id is not None:
            plugin = manager.NeutronManager.get_service_plugins().get(
                service_constants.L3_ROUTER_NAT)
            adminContext = context if context.is_admin else context.elevated()
            # A bit hacky here
            router = plugin._get_router(adminContext, qos_db.router_id)
            if qos_db.direction == 'egress':
                if router.gw_port_id:
                    ret.append(("qg-%s" % router.gw_port_id)[:NIC_NAME_LEN])
            else:
                for rp in router.attached_ports:
                    if rp.port_id != router.gw_port_id:
                        ret.append(("qr-%s" % rp.port_id)[:NIC_NAME_LEN])
        elif qos_db.port_id is not None:
            prefix = 'qvo' if qos_db.direction == 'ingress' else 'qvb'
            ret.append(("%s%s" % (prefix, qos_db.port_id))[:NIC_NAME_LEN])

        return ret

    def _check_port(self, context, port_id):
        plugin = manager.NeutronManager.get_plugin()
        adminContext = context if context.is_admin else context.elevated()
        port = plugin.get_port(adminContext, port_id)
        if not port['device_owner'].startswith('compute'):
            raise ext_qos.QosInvalidPortType(
                port_id=port_id, port_type=port['device_owner'])

    def _check_router(self, context, router_id):
        plugin = manager.NeutronManager.get_service_plugins().get(
            service_constants.L3_ROUTER_NAT)
        adminContext = context if context.is_admin else context.elevated()
        plugin.get_router(adminContext, router_id)

    def sync_qos(self, context, qos_list, host):
        hosting_qos = self._get_qos_on_host(context, host)
        deleted_qos = filter(lambda qos: qos not in hosting_qos, qos_list)

{'fraction_non_alphanumeric': 0.09034051424600417, 'fraction_numerical': 0.004169562195969423, 'mean_word_length': 2.593424885559717, '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': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4201761', 'n_tokens_mistral': 2718, 'n_tokens_neox': 2651, 'n_words': 642}