#    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 copy
from unittest import mock

from keystoneauth1 import exceptions as ks_exc
import netaddr
from neutron_lib.api.definitions import qos
from neutron_lib.callbacks import events
from neutron_lib import constants as lib_constants
from neutron_lib import context
from neutron_lib import exceptions as lib_exc
from neutron_lib.exceptions import placement as pl_exc
from neutron_lib.exceptions import qos as qos_exc
from neutron_lib.objects import utils as obj_utils
from neutron_lib.plugins import constants as plugins_constants
from neutron_lib.plugins import directory
from neutron_lib.services.qos import constants as qos_consts
from neutron_lib.utils import net as net_utils
import os_resource_classes as orc
from oslo_config import cfg
from oslo_utils import uuidutils
import webob.exc

from neutron.exceptions import qos as neutron_qos_exc
from neutron.extensions import qos_pps_minimum_rule_alias
from neutron.extensions import qos_rules_alias
from neutron import manager
from neutron.objects import network as network_object
from neutron.objects import ports as ports_object
from neutron.objects.qos import policy as policy_object
from neutron.objects.qos import rule as rule_object
from neutron.services.qos import qos_plugin
from neutron.tests.unit.db import test_db_base_plugin_v2
from neutron.tests.unit.services.qos import base


DB_PLUGIN_KLASS = 'neutron.db.db_base_plugin_v2.NeutronDbPluginV2'
SERVICE_PLUGIN_KLASS = 'neutron.services.qos.qos_plugin.QoSPlugin'


class TestQosPlugin(base.BaseQosTestCase):

    def setUp(self):
        super(TestQosPlugin, self).setUp()
        self.setup_coreplugin(load_plugins=False)

        mock.patch('neutron.objects.db.api.create_object').start()
        mock.patch('neutron.objects.db.api.update_object').start()
        mock.patch('neutron.objects.db.api.delete_object').start()
        mock.patch('neutron.objects.db.api.get_object').start()
        _mock_qos_load_attr = mock.patch(
            'neutron.objects.qos.policy.QosPolicy.obj_load_attr')
        self.mock_qos_load_attr = _mock_qos_load_attr.start()
        # We don't use real models as per mocks above. We also need to mock-out
        # methods that work with real data types
        mock.patch(
            'neutron.objects.base.NeutronDbObject.modify_fields_from_db'
        ).start()
        mock.patch.object(policy_object.QosPolicy, 'unset_default').start()
        mock.patch.object(policy_object.QosPolicy, 'set_default').start()

        cfg.CONF.set_override("core_plugin", DB_PLUGIN_KLASS)
        cfg.CONF.set_override("service_plugins", ["qos"])

        manager.init()
        self.qos_plugin = directory.get_plugin(plugins_constants.QOS)

        self.qos_plugin.driver_manager = mock.Mock()

        self.rpc_push = mock.patch('neutron.api.rpc.handlers.resources_rpc'
                                   '.ResourcesPushRpcApi.push').start()

        self.ctxt = context.Context('fake_user', 'fake_tenant')
        self.admin_ctxt = context.get_admin_context()

        self.policy_data = {
            'policy': {'id': uuidutils.generate_uuid(),
                       'project_id': uuidutils.generate_uuid(),
                       'name': 'test-policy',
                       'description': 'Test policy description',
                       'shared': True,
                       'is_default': False}}

        self.rule_data = {
            'bandwidth_limit_rule': {'id': uuidutils.generate_uuid(),
                                     'max_kbps': 100,
                                     'max_burst_kbps': 150},
            'dscp_marking_rule': {'id': uuidutils.generate_uuid(),
                                  'dscp_mark': 16},
            'minimum_bandwidth_rule': {
                'id': uuidutils.generate_uuid(),
                'min_kbps': 10},
            'packet_rate_limit_rule': {
                'id': uuidutils.generate_uuid(),
                'max_kpps': 20,
                'max_burst_kpps': 130},
            'minimum_packet_rate_rule': {
                'id': uuidutils.generate_uuid(),
                'min_kpps': 10,
                'direction': 'any'},
        }

        self.policy = policy_object.QosPolicy(
            self.ctxt, **self.policy_data['policy'])

        self.rule = rule_object.QosBandwidthLimitRule(
            self.ctxt, **self.rule_data['bandwidth_limit_rule'])

        self.dscp_rule = rule_object.QosDscpMarkingRule(
            self.ctxt, **self.rule_data['dscp_marking_rule'])

        self.min_bw_rule = rule_object.QosMinimumBandwidthRule(
            self.ctxt, **self.rule_data['minimum_bandwidth_rule'])

        self.pps_rule = rule_object.QosPacketRateLimitRule(
            self.ctxt, **self.rule_data['packet_rate_limit_rule'])

        self.min_pps_rule = rule_object.QosMinimumPacketRateRule(
            self.ctxt, **self.rule_data['minimum_packet_rate_rule'])

    def _validate_driver_params(self, method_name, ctxt):
        call_args = self.qos_plugin.driver_manager.call.call_args[0]
        self.assertTrue(self.qos_plugin.driver_manager.call.called)
        self.assertEqual(call_args[0], method_name)
        self.assertEqual(call_args[1], ctxt)
        self.assertIsInstance(call_args[2], policy_object.QosPolicy)

    def _create_and_extend_port(self, min_bw_rules, min_pps_rules=None,
                                physical_network='public',
                                has_qos_policy=True, has_net_qos_policy=False,
                                request_groups_uuids=None):
        network_id = uuidutils.generate_uuid()

        self.port_data = {
            'port': {'id': uuidutils.generate_uuid(),
                     'network_id': network_id}
        }

        if has_qos_policy:
            self.port_data['port']['qos_policy_id'] = self.policy.id
        elif has_net_qos_policy:
            self.port_data['port']['qos_network_policy_id'] = self.policy.id

        self.port = ports_object.Port(
            self.ctxt, **self.port_data['port'])

        port_res = {"binding:vnic_type": "normal"}
        segment_mock = mock.MagicMock(network_id=network_id,
                                      physical_network=physical_network)
        min_pps_rules = min_pps_rules if min_pps_rules else []

        with mock.patch('neutron.objects.network.NetworkSegment.get_objects',
                        return_value=[segment_mock]), \
                mock.patch(
                    'neutron.objects.qos.rule.QosMinimumBandwidthRule.'
                    'get_objects',
                    return_value=min_bw_rules), \
                mock.patch(
                    'neutron.objects.qos.rule.QosMinimumPacketRateRule.'
                    'get_objects',
                    return_value=min_pps_rules), \
                mock.patch(
                    'uuid.uuid5',
                    return_value='fake_uuid',
                    side_effect=request_groups_uuids):
            return qos_plugin.QoSPlugin._extend_port_resource_request(
                port_res, self.port)

    def _create_and_extend_ports(self, min_bw_rules, min_pps_rules=None,
                                 physical_network='public',
                                 request_groups_uuids=None):
        network_id = uuidutils.generate_uuid()

        ports_res = [
            {
                "resource_request": {
                    "port_id": uuidutils.generate_uuid(),
                    "qos_id": self.policy.id,
                    "network_id": network_id,
                    "vnic_type": "normal",

                }
            },
            {
                "resource_request": {
                    "port_id": uuidutils.generate_uuid(),
                    "qos_id": self.policy.id,
                    "network_id": network_id,
                    "vnic_type": "normal",
                }
            },
        ]
        segment_mock = mock.MagicMock(network_id=network_id,
                                      physical_network=physical_network)
        min_pps_rules = min_pps_rules if min_pps_rules else []

        with mock.patch('neutron.objects.network.NetworkSegment.get_objects',
                        return_value=[segment_mock]), \
                mock.patch(
                    'neutron.objects.qos.rule.QosMinimumBandwidthRule.'
                    'get_objects',
                    return_value=min_bw_rules), \
                mock.patch(
                    'neutron.objects.qos.rule.QosMinimumPacketRateRule.'
                    'get_objects',
                    return_value=min_pps_rules), \
                mock.patch(
                    'uuid.uuid5',
                    return_value='fake_uuid',
                    side_effect=request_groups_uuids):
            return qos_plugin.QoSPlugin._extend_port_resource_request_bulk(
                ports_res, None)

    def test__extend_port_resource_request_min_bw_rule(self):
        self.min_bw_rule.direction = lib_constants.EGRESS_DIRECTION
        port = self._create_and_extend_port([self.min_bw_rule])

        self.assertEqual(
            1,
            len(port['resource_request']['request_groups'])
        )
        self.assertEqual(
            'fake_uuid',
            port['resource_request']['request_groups'][0]['id']
        )
        self.assertEqual(
            ['CUSTOM_PHYSNET_PUBLIC', 'CUSTOM_VNIC_TYPE_NORMAL'],
            port['resource_request']['request_groups'][0]['required']
        )
        self.assertEqual(
            {orc.NET_BW_EGR_KILOBIT_PER_SEC: 10},
            port['resource_request']['request_groups'][0]['resources'],
        )
        self.assertEqual(
            ['fake_uuid'],
            port['resource_request']['same_subtree'],
        )

    def test__extend_port_resource_request_min_pps_rule(self):
        port = self._create_and_extend_port([], [self.min_pps_rule])

        self.assertEqual(
            1,
            len(port['resource_request']['request_groups'])
        )
        self.assertEqual(
            'fake_uuid',
            port['resource_request']['request_groups'][0]['id']
        )
        self.assertEqual(
            ['CUSTOM_VNIC_TYPE_NORMAL'],
            port['resource_request']['request_groups'][0]['required']
        )
        self.assertEqual(
            {orc.NET_PACKET_RATE_KILOPACKET_PER_SEC: 10},
            port['resource_request']['request_groups'][0]['resources'],
        )
        self.assertEqual(
            ['fake_uuid'],
            port['resource_request']['same_subtree'],
        )

    def test__extend_port_resource_request_min_bw_and_pps_rule(self):
        self.min_bw_rule.direction = lib_constants.EGRESS_DIRECTION
        self.min_pps_rule.direction = lib_constants.EGRESS_DIRECTION
        request_groups_uuids = ['fake_uuid0', 'fake_uuid1']
        min_bw_rule_ingress_data = {
            'id': uuidutils.generate_uuid(),
            'min_kbps': 20,
            'direction': lib_constants.INGRESS_DIRECTION}
        min_pps_rule_ingress_data = {
            'id': uuidutils.generate_uuid(),
            'min_kpps': 20,
            'direction': lib_constants.INGRESS_DIRECTION}
        min_bw_rule_ingress = rule_object.QosMinimumBandwidthRule(
            self.ctxt, **min_bw_rule_ingress_data)
        min_pps_rule_ingress = rule_object.QosMinimumPacketRateRule(
            self.ctxt, **min_pps_rule_ingress_data)
        port = self._create_and_extend_port(
            [self.min_bw_rule, min_bw_rule_ingress],
            [self.min_pps_rule, min_pps_rule_ingress],
            request_groups_uuids=request_groups_uuids)

        self.assertEqual(
            2,
            len(port['resource_request']['request_groups'])
        )
        self.assertIn(
            {
                'id': 'fake_uuid0',
                'required':
                    ['CUSTOM_PHYSNET_PUBLIC', 'CUSTOM_VNIC_TYPE_NORMAL'],
                'resources': {
                    orc.NET_BW_EGR_KILOBIT_PER_SEC: 10,
                    orc.NET_BW_IGR_KILOBIT_PER_SEC: 20},
            },
            port['resource_request']['request_groups']
        )
        self.assertIn(
            {
                'id': 'fake_uuid1',
                'required': ['CUSTOM_VNIC_TYPE_NORMAL'],
                'resources': {
                    orc.NET_PACKET_RATE_EGR_KILOPACKET_PER_SEC: 10,
                    orc.NET_PACKET_RATE_IGR_KILOPACKET_PER_SEC: 20,
                },
            },
            port['resource_request']['request_groups']
        )
        self.assertEqual(
            ['fake_uuid0', 'fake_uuid1'],
            port['resource_request']['same_subtree'],
        )

    def test__extend_port_resource_request_non_min_bw_or_min_pps_rule(self):
        port = self._create_and_extend_port([], [])

        self.assertIsNone(port.get('resource_request'))

    def test__extend_port_resource_request_min_bw_non_provider_net(self):
        self.min_bw_rule.direction = lib_constants.EGRESS_DIRECTION

        port = self._create_and_extend_port([self.min_bw_rule],
                                            physical_network=None)
        self.assertIsNone(port.get('resource_request'))

    def test__extend_port_resource_request_mix_rules_non_provider_net(self):
        self.min_bw_rule.direction = lib_constants.EGRESS_DIRECTION

        port = self._create_and_extend_port([self.min_bw_rule],
                                            [self.min_pps_rule],
                                            physical_network=None)
        self.assertEqual(
            1,
            len(port['resource_request']['request_groups'])
        )
        self.assertEqual(
            'fake_uuid',
            port['resource_request']['request_groups'][0]['id']
        )
        self.assertEqual(
            ['CUSTOM_VNIC_TYPE_NORMAL'],
            port['resource_request']['request_groups'][0]['required']
        )
        self.assertEqual(
            {orc.NET_PACKET_RATE_KILOPACKET_PER_SEC: 10},
            port['resource_request']['request_groups'][0]['resources'],
        )
        self.assertEqual(
            ['fake_uuid'],
            port['resource_request']['same_subtree'],
        )

    def test__extend_port_resource_request_bulk_min_bw_rule(self):
        self.min_bw_rule.direction = lib_constants.EGRESS_DIRECTION
        ports = self._create_and_extend_ports([self.min_bw_rule])

        for port in ports:
            self.assertEqual(
                1,
                len(port['resource_request']['request_groups'])
            )
            self.assertEqual(
                'fake_uuid',
                port['resource_request']['request_groups'][0]['id']
            )
            self.assertEqual(
                ['CUSTOM_PHYSNET_PUBLIC', 'CUSTOM_VNIC_TYPE_NORMAL'],
                port['resource_request']['request_groups'][0]['required']
            )
            self.assertEqual(
                {orc.NET_BW_EGR_KILOBIT_PER_SEC: 10},
                port['resource_request']['request_groups'][0]['resources'],
            )
            self.assertEqual(
                ['fake_uuid'],
                port['resource_request']['same_subtree'],
            )

    def test__extend_port_resource_request_bulk_min_pps_rule(self):
        ports = self._create_and_extend_ports([], [self.min_pps_rule])

        for port in ports:
            self.assertEqual(
                1,
                len(port['resource_request']['request_groups'])
            )
            self.assertEqual(
                'fake_uuid',
                port['resource_request']['request_groups'][0]['id']
            )
            self.assertEqual(
                ['CUSTOM_VNIC_TYPE_NORMAL'],
                port['resource_request']['request_groups'][0]['required']
            )
            self.assertEqual(
                {orc.NET_PACKET_RATE_KILOPACKET_PER_SEC: 10},
                port['resource_request']['request_groups'][0]['resources'],
            )
            self.assertEqual(
                ['fake_uuid'],
                port['resource_request']['same_subtree'],
            )

    def test__extend_port_resource_request_bulk_min_bw_and_pps_rule(self):
        self.min_bw_rule.direction = lib_constants.EGRESS_DIRECTION
        self.min_pps_rule.direction = lib_constants.EGRESS_DIRECTION
        request_groups_uuids = ['fake_uuid0', 'fake_uuid1'] * 2
        min_bw_rule_ingress_data = {
            'id': uuidutils.generate_uuid(),
            'min_kbps': 20,
            'direction': lib_constants.INGRESS_DIRECTION}
        min_pps_rule_ingress_data = {
            'id': uuidutils.generate_uuid(),
            'min_kpps': 20,
            'direction': lib_constants.INGRESS_DIRECTION}
        min_bw_rule_ingress = rule_object.QosMinimumBandwidthRule(
            self.ctxt, **min_bw_rule_ingress_data)
        min_pps_rule_ingress = rule_object.QosMinimumPacketRateRule(
            self.ctxt, **min_pps_rule_ingress_data)
        ports = self._create_and_extend_ports(
            [self.min_bw_rule, min_bw_rule_ingress],
            [self.min_pps_rule, min_pps_rule_ingress],
            request_groups_uuids=request_groups_uuids)

        for port in ports:
            self.assertEqual(
                2,
                len(port['resource_request']['request_groups'])
            )
            self.assertIn(
                {
                    'id': 'fake_uuid0',
                    'required':
                        ['CUSTOM_PHYSNET_PUBLIC', 'CUSTOM_VNIC_TYPE_NORMAL'],
                    'resources': {
                        orc.NET_BW_EGR_KILOBIT_PER_SEC: 10,
                        orc.NET_BW_IGR_KILOBIT_PER_SEC: 20},
                },
                port['resource_request']['request_groups']
            )
            self.assertIn(
                {
                    'id': 'fake_uuid1',
                    'required': ['CUSTOM_VNIC_TYPE_NORMAL'],
                    'resources': {
                        orc.NET_PACKET_RATE_EGR_KILOPACKET_PER_SEC: 10,
                        orc.NET_PACKET_RATE_IGR_KILOPACKET_PER_SEC: 20,
                    },
                },
                port['resource_request']['request_groups']
            )
            self.assertEqual(
                ['fake_uuid0', 'fake_uuid1'],
                port['resource_request']['same_subtree'],
            )

    def test__extend_port_resource_request_no_qos_policy(self):
        port = self._create_and_extend_port([], physical_network='public',
                                            has_qos_policy=False)
        self.assertIsNone(port.get('resource_request'))

    def test__extend_port_resource_request_min_bw_inherited_policy(
            self):
        self.min_bw_rule.direction = lib_constants.EGRESS_DIRECTION
        self.min_bw_rule.qos_policy_id = self.policy.id

        port = self._create_and_extend_port([self.min_bw_rule],
                                            has_net_qos_policy=True)
        self.assertEqual(
            1,
            len(port['resource_request']['request_groups'])
        )
        self.assertEqual(
            'fake_uuid',
            port['resource_request']['request_groups'][0]['id']
        )
        self.assertEqual(
            ['CUSTOM_PHYSNET_PUBLIC', 'CUSTOM_VNIC_TYPE_NORMAL'],
            port['resource_request']['request_groups'][0]['required']
        )
        self.assertEqual(
            {orc.NET_BW_EGR_KILOBIT_PER_SEC: 10},
            port['resource_request']['request_groups'][0]['resources'],
        )
        self.assertEqual(
            ['fake_uuid'],
            port['resource_request']['same_subtree'],
        )

    def test_get_ports_with_policy(self):
        network_ports = [
            mock.MagicMock(qos_policy_id=None),
            mock.MagicMock(qos_policy_id=uuidutils.generate_uuid()),
            mock.MagicMock(qos_policy_id=None)
        ]
        ports = [
            mock.MagicMock(qos_policy_id=self.policy.id),
        ]
        expected_network_ports = [
            port for port in network_ports if port.qos_policy_id is None]
        expected_ports = ports + expected_network_ports
        with mock.patch(
            'neutron.objects.ports.Port.get_objects',
            side_effect=[network_ports, ports]
        ), mock.patch.object(
            self.policy, "get_bound_networks"
        ), mock.patch.object(
            self.policy, "get_bound_ports"
        ):
            policy_ports = self.qos_plugin._get_ports_with_policy(
                self.ctxt, self.policy)
            self.assertEqual(
                len(expected_ports), len(policy_ports))
            for port in expected_ports:
                self.assertIn(port, policy_ports)

    def _test_validate_update_port_callback(self, policy_id=None,
                                            original_policy_id=None):
        port_id = uuidutils.generate_uuid()
        kwargs = {
            "port": {
                "id": port_id,
                qos_consts.QOS_POLICY_ID: policy_id
            },
            "original_port": {
                "id": port_id,
                qos_consts.QOS_POLICY_ID: original_policy_id
            }
        }
        port_mock = mock.MagicMock(id=port_id, qos_policy_id=policy_id)
        policy_mock = mock.MagicMock(id=policy_id)
        admin_ctxt = mock.Mock()
        with mock.patch(
            'neutron.objects.ports.Port.get_object',
            return_value=port_mock
        ) as get_port, mock.patch(
            'neutron.objects.qos.policy.QosPolicy.get_object',
            return_value=policy_mock
        ) as get_policy, mock.patch.object(
            self.qos_plugin, "validate_policy_for_port"
        ) as validate_policy_for_port, mock.patch.object(
            self.ctxt, "elevated", return_value=admin_ctxt
        ):
            self.qos_plugin._validate_update_port_callback(
                "PORT", "precommit_update", "test_plugin",
                payload=events.DBEventPayload(
                    self.ctxt, desired_state=kwargs['port'],
                    states=(kwargs['original_port'],)))
            if policy_id is None or policy_id == original_policy_id:
                get_port.assert_not_called()
                get_policy.assert_not_called()
                validate_policy_for_port.assert_not_called()
            else:
                get_port.assert_called_once_with(self.ctxt, id=port_id)
                get_policy.assert_called_once_with(admin_ctxt, id=policy_id)
                validate_policy_for_port.assert_called_once_with(
                    self.ctxt, policy_mock, port_mock)

    def test_validate_update_port_callback_policy_changed(self):
        self._test_validate_update_port_callback(
            policy_id=uuidutils.generate_uuid())

    def test_validate_update_port_callback_policy_not_changed(self):
        policy_id = uuidutils.generate_uuid()
        self._test_validate_update_port_callback(
            policy_id=policy_id, original_policy_id=policy_id)

    def test_validate_update_port_callback_policy_removed(self):
        self._test_validate_update_port_callback(
            policy_id=None, original_policy_id=uuidutils.generate_uuid())

    def _test_validate_update_network_callback(self, policy_id=None,
                                               original_policy_id=None):
        network_id = uuidutils.generate_uuid()
        kwargs = {
            "context": self.ctxt,
            "network": {
                "id": network_id,
                qos_consts.QOS_POLICY_ID: policy_id
            },
            "original_network": {
                "id": network_id,
                qos_consts.QOS_POLICY_ID: original_policy_id
            }
        }
        port_mock_with_own_policy = mock.MagicMock(
            id=uuidutils.generate_uuid(),
            qos_policy_id=uuidutils.generate_uuid())
        port_mock_without_own_policy = mock.MagicMock(
            id=uuidutils.generate_uuid(), qos_policy_id=None)
        ports = [port_mock_with_own_policy, port_mock_without_own_policy]
        policy_mock = mock.MagicMock(id=policy_id)
        admin_ctxt = mock.Mock()
        with mock.patch(
            'neutron.objects.ports.Port.get_objects',
            return_value=ports
        ) as get_ports, mock.patch(
            'neutron.objects.qos.policy.QosPolicy.get_object',
            return_value=policy_mock
        ) as get_policy, mock.patch.object(
            self.qos_plugin, "validate_policy_for_network"
        ) as validate_policy_for_network, mock.patch.object(
            self.qos_plugin, "validate_policy_for_ports"
        ) as validate_policy_for_ports, mock.patch.object(
            self.ctxt, "elevated", return_value=admin_ctxt
        ):
            self.qos_plugin._validate_update_network_callback(
                "NETWORK", "precommit_update", "test_plugin",
                payload=events.DBEventPayload(
                    self.ctxt, desired_state=kwargs['network'],
                    states=(kwargs['original_network'],)))
            if policy_id is None or policy_id == original_policy_id:
                get_policy.assert_not_called()
                validate_policy_for_network.assert_not_called()
                get_ports.assert_not_called()
                validate_policy_for_ports.assert_not_called()
            else:
                get_policy.assert_called_once_with(admin_ctxt, id=policy_id)
                get_ports.assert_called_once_with(self.ctxt,
                                                  network_id=network_id)
                validate_policy_for_ports.assert_called_once_with(
                    self.ctxt, policy_mock, [port_mock_without_own_policy])

    def test_validate_update_network_callback_policy_changed(self):
        self._test_validate_update_network_callback(
            policy_id=uuidutils.generate_uuid())

    def test_validate_update_network_callback_policy_not_changed(self):
        policy_id = uuidutils.generate_uuid()
        self._test_validate_update_network_callback(
            policy_id=policy_id, original_policy_id=policy_id)

    def test_validate_update_network_callback_policy_removed(self):
        self._test_validate_update_network_callback(
            policy_id=None, original_policy_id=uuidutils.generate_uuid())

    def test_validate_policy_for_port_rule_not_valid(self):
        port = {'id': uuidutils.generate_uuid()}
        with mock.patch.object(
            self.qos_plugin.driver_manager, "validate_rule_for_port",
            return_value=False
        ):
            self.policy.rules = [self.rule]
            self.assertRaises(
                qos_exc.QosRuleNotSupported,
                self.qos_plugin.validate_policy_for_port,
                self.ctxt, self.policy, port)

    def test_validate_policy_for_port_all_rules_valid(self):
        port = {'id': uuidutils.generate_uuid()}
        with mock.patch.object(
            self.qos_plugin.driver_manager, "validate_rule_for_port",
            return_value=True
        ):
            self.policy.rules = [self.rule]
            try:
                self.qos_plugin.validate_policy_for_port(
                    self.ctxt, self.policy, port)
            except qos_exc.QosRuleNotSupported:
                self.fail("QosRuleNotSupported exception unexpectedly raised")

    def test_validate_policy_for_network(self):
        network = uuidutils.generate_uuid()
        with mock.patch.object(
            self.qos_plugin.driver_manager, "validate_rule_for_network",
            return_value=True
        ):
            self.policy.rules = [self.rule]
            try:
                self.qos_plugin.validate_policy_for_network(
                    self.ctxt, self.policy, network_id=network)
            except qos_exc.QosRuleNotSupportedByNetwork:
                self.fail("QosRuleNotSupportedByNetwork "
                          "exception unexpectedly raised")

    def test_create_min_bw_rule_on_bound_port(self):
        policy = self._get_policy()
        policy.rules = [self.min_bw_rule]
        segment = network_object.NetworkSegment(
            physical_network='fake physnet')
        net = network_object.Network(
            self.ctxt,
            segments=[segment])
        port = ports_object.Port(
            self.ctxt,
            id=uuidutils.generate_uuid(),
            network_id=uuidutils.generate_uuid(),
            device_owner='compute:fake-zone')
        with mock.patch(
                'neutron.objects.qos.policy.QosPolicy.get_object',
                return_value=policy), \
            mock.patch(
                'neutron.objects.network.Network.get_object',
                return_value=net), \
            mock.patch.object(
                self.qos_plugin,
                '_get_ports_with_policy',
                return_value=[port]):
            self.assertRaises(
                NotImplementedError,
                self.qos_plugin.create_policy_minimum_bandwidth_rule,
                self.ctxt, policy.id, self.rule_data)

    def test_create_min_bw_rule_on_unbound_port(self):
        policy = self._get_policy()
        policy.rules = [self.min_bw_rule]
        segment = network_object.NetworkSegment(
            physical_network='fake physnet')
        net = network_object.Network(
            self.ctxt,
            segments=[segment])
        port = ports_object.Port(
            self.ctxt,
            id=uuidutils.generate_uuid(),
            network_id=uuidutils.generate_uuid(),
            device_owner='')
        with mock.patch(
                'neutron.objects.qos.policy.QosPolicy.get_object',
                return_value=policy), \
            mock.patch(
                'neutron.objects.network.Network.get_object',
                return_value=net), \
            mock.patch.object(
                self.qos_plugin,
                '_get_ports_with_policy',
                return_value=[port]):
            try:
                self.qos_plugin.create_policy_minimum_bandwidth_rule(
                    self.ctxt, policy.id, self.rule_data)
            except NotImplementedError:
                self.fail()

    @mock.patch(
        'neutron.objects.rbac_db.RbacNeutronDbObjectMixin'
        '.create_rbac_policy')
    @mock.patch('neutron.objects.qos.policy.QosPolicy')
    def test_add_policy(self, mock_qos_policy, mock_create_rbac_policy):
        mock_manager = mock.Mock()
        mock_manager.attach_mock(mock_qos_policy, 'QosPolicy')
        mock_manager.attach_mock(self.qos_plugin.driver_manager, 'driver')
        mock_manager.reset_mock()

        self.qos_plugin.create_policy(self.ctxt, self.policy_data)
        policy_mock_call = mock.call.QosPolicy().create()
        create_precommit_mock_call = mock.call.driver.call(
            'create_policy_precommit', self.ctxt, mock.ANY)
        create_mock_call = mock.call.driver.call(
            'create_policy', self.ctxt, mock.ANY)
        self.assertTrue(
            mock_manager.mock_calls.index(policy_mock_call) <
            mock_manager.mock_calls.index(create_precommit_mock_call) <
            mock_manager.mock_calls.index(create_mock_call))

    def test_add_policy_with_extra_tenant_keyword(self, *mocks):
        policy_id = uuidutils.generate_uuid()
        project_id = uuidutils.generate_uuid()
        tenant_policy = {
            'policy': {'id': policy_id,
                       'project_id': project_id,
                       'tenant_id': project_id,
                       'name': 'test-policy',
                       'description': 'Test policy description',
                       'shared': True,
                       'is_default': False}}

        policy_details = {'id': policy_id,
                          'project_id': project_id,
                          'name': 'test-policy',
                          'description': 'Test policy description',
                          'shared': True,
                          'is_default': False}

        with mock.patch('neutron.objects.qos.policy.QosPolicy') as QosMocked:
            self.qos_plugin.create_policy(self.ctxt, tenant_policy)

        QosMocked.assert_called_once_with(self.ctxt, **policy_details)

    @mock.patch.object(policy_object.QosPolicy, "get_object")
    @mock.patch(
        'neutron.objects.rbac_db.RbacNeutronDbObjectMixin'
        '.create_rbac_policy')
    @mock.patch.object(policy_object.QosPolicy, 'update')
    def test_update_policy(self, mock_qos_policy_update,
                           mock_create_rbac_policy, mock_qos_policy_get):
        mock_qos_policy_get.return_value = self.policy
        mock_manager = mock.Mock()
        mock_manager.attach_mock(mock_qos_policy_update, 'update')
        mock_manager.attach_mock(self.qos_plugin.driver_manager, 'driver')
        mock_manager.reset_mock()

        fields = obj_utils.get_updatable_fields(
            policy_object.QosPolicy, self.policy_data['policy'])
        self.qos_plugin.update_policy(
            self.ctxt, self.policy.id, {'policy': fields})
        self._validate_driver_params('update_policy', self.ctxt)

        policy_update_mock_call = mock.call.update()
        update_precommit_mock_call = mock.call.driver.call(
            'update_policy_precommit', self.ctxt, mock.ANY)
        update_mock_call = mock.call.driver.call(
            'update_policy', self.ctxt, mock.ANY)
        self.assertTrue(
            mock_manager.mock_calls.index(policy_update_mock_call) <
            mock_manager.mock_calls.index(update_precommit_mock_call) <
            mock_manager.mock_calls.index(update_mock_call))

    @mock.patch('neutron.objects.db.api.get_object', return_value=None)
    @mock.patch.object(policy_object.QosPolicy, 'delete')
    def test_delete_policy(self, mock_qos_policy_delete, mock_api_get_policy):
        mock_manager = mock.Mock()
        mock_manager.attach_mock(mock_qos_policy_delete, 'delete')
        mock_manager.attach_mock(self.qos_plugin.driver_manager, 'driver')
        mock_manager.reset_mock()

        self.qos_plugin.delete_policy(self.ctxt, self.policy.id)
        self._validate_driver_params('delete_policy', self.ctxt)

        policy_delete_mock_call = mock.call.delete()
        delete_precommit_mock_call = mock.call.driver.call(
            'delete_policy_precommit', self.ctxt, mock.ANY)
        delete_mock_call = mock.call.driver.call(
            'delete_policy', self.ctxt, mock.ANY)
        self.assertTrue(
            mock_manager.mock_calls.index(policy_delete_mock_call) <
            mock_manager.mock_calls.index(delete_precommit_mock_call) <
            mock_manager.mock_calls.index(delete_mock_call))

    @mock.patch.object(policy_object.QosPolicy, "get_object")
    @mock.patch.object(rule_object.QosBandwidthLimitRule, 'create')
    def test_create_policy_rule(self, mock_qos_rule_create,
                                mock_qos_policy_get):
        _policy = copy.copy(self.policy)
        setattr(_policy, "rules", [])
        mock_qos_policy_get.return_value = _policy
        mock_manager = mock.Mock()
        mock_manager.attach_mock(mock_qos_rule_create, 'create')
        mock_manager.attach_mock(self.qos_plugin.driver_manager, 'driver')
        mock_manager.reset_mock()
        with mock.patch('neutron.objects.qos.qos_policy_validator'
                        '.check_bandwidth_rule_conflict',
                        return_value=None), \
                mock.patch(
                    'neutron.objects.qos.qos_policy_validator'
                    '.check_min_pps_rule_conflict', return_value=None):
            self.qos_plugin.create_policy_bandwidth_limit_rule(
                self.ctxt, self.policy.id, self.rule_data)
            self._validate_driver_params('update_policy', self.ctxt)
            rule_create_mock_call = mock.call.create()
            update_precommit_mock_call = mock.call.driver.call(
                'update_policy_precommit', self.ctxt, mock.ANY)
            update_mock_call = mock.call.driver.call(
                'update_policy', self.ctxt, mock.ANY)
            self.assertTrue(
                mock_manager.mock_calls.index(rule_create_mock_call) <
                mock_manager.mock_calls.index(update_precommit_mock_call) <
                mock_manager.mock_calls.index(update_mock_call))

    def test_create_policy_rule_check_rule_min_less_than_max(self):
        _policy = self._get_policy()
        setattr(_policy, "rules", [self.rule])
        with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
                        return_value=_policy) as mock_qos_get_obj:
            self.qos_plugin.create_policy_minimum_bandwidth_rule(
                self.ctxt, _policy.id, self.rule_data)
            self._validate_driver_params('update_policy', self.ctxt)
            self.mock_qos_load_attr.assert_called_once_with('rules')
            mock_qos_get_obj.assert_called_once_with(self.ctxt, id=_policy.id)

    def test_create_policy_rule_check_rule_max_more_than_min(self):
        _policy = self._get_policy()
        setattr(_policy, "rules", [self.min_bw_rule])
        with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
                        return_value=_policy) as mock_qos_get_obj:
            self.qos_plugin.create_policy_bandwidth_limit_rule(
                self.ctxt, _policy.id, self.rule_data)
            self._validate_driver_params('update_policy', self.ctxt)
            self.mock_qos_load_attr.assert_called_once_with('rules')
            mock_qos_get_obj.assert_called_once_with(self.ctxt, id=_policy.id)

    def test_create_policy_rule_check_rule_bwlimit_less_than_minbw(self):
        _policy = self._get_policy()
        self.rule_data['bandwidth_limit_rule']['max_kbps'] = 1
        setattr(_policy, "rules", [self.min_bw_rule])
        with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
                        return_value=_policy) as mock_qos_get_obj:
            self.assertRaises(qos_exc.QoSRuleParameterConflict,
                self.qos_plugin.create_policy_bandwidth_limit_rule,
                self.ctxt, self.policy.id, self.rule_data)
            mock_qos_get_obj.assert_called_once_with(self.ctxt, id=_policy.id)

    def test_create_policy_rule_check_rule_minbw_gr_than_bwlimit(self):
        _policy = self._get_policy()
        self.rule_data['minimum_bandwidth_rule']['min_kbps'] = 1000000
        setattr(_policy, "rules", [self.rule])
        with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
                        return_value=_policy) as mock_qos_get_obj:
            self.assertRaises(qos_exc.QoSRuleParameterConflict,
                self.qos_plugin.create_policy_minimum_bandwidth_rule,
                self.ctxt, self.policy.id, self.rule_data)
            mock_qos_get_obj.assert_called_once_with(self.ctxt, id=_policy.id)

    def test_create_policy_rule_duplicates(self):
        _policy = self._get_policy()
        setattr(_policy, "rules", [self.rule])
        new_rule_data = {
            'bandwidth_limit_rule': {
                'max_kbps': 5000,
                'direction': self.rule.direction
            }
        }
        with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
                        return_value=_policy) as mock_qos_get_obj:
            self.assertRaises(
                qos_exc.QoSRulesConflict,
                self.qos_plugin.create_policy_bandwidth_limit_rule,
                self.ctxt, _policy.id, new_rule_data)
            mock_qos_get_obj.assert_called_once_with(self.ctxt, id=_policy.id)

    @mock.patch.object(rule_object.QosBandwidthLimitRule, 'update')
    def test_update_policy_rule(self, mock_qos_rule_update):
        mock_manager = mock.Mock()
        mock_manager.attach_mock(mock_qos_rule_update, 'update')
        mock_manager.attach_mock(self.qos_plugin.driver_manager, 'driver')
        mock_manager.reset_mock()

        _policy = policy_object.QosPolicy(
            self.ctxt, **self.policy_data['policy'])
        setattr(_policy, "rules", [self.rule])
        with mock.patch('neutron.objects.qos.rule.get_rules',
                        return_value=[self.rule]), mock.patch(
                'neutron.objects.qos.policy.QosPolicy.get_object',
                return_value=_policy):
            self.rule_data['bandwidth_limit_rule']['max_kbps'] = 1
            self.qos_plugin.update_policy_bandwidth_limit_rule(
                self.ctxt, self.rule.id, self.policy.id, self.rule_data)
            self._validate_driver_params('update_policy', self.ctxt)

            rule_update_mock_call = mock.call.update()
            update_precommit_mock_call = mock.call.driver.call(
                'update_policy_precommit', self.ctxt, mock.ANY)
            update_mock_call = mock.call.driver.call(
                'update_policy', self.ctxt, mock.ANY)
            self.assertTrue(
                mock_manager.mock_calls.index(rule_update_mock_call) <
                mock_manager.mock_calls.index(update_precommit_mock_call) <
                mock_manager.mock_calls.index(update_mock_call))

    def test_update_policy_rule_check_rule_min_less_than_max(self):
        _policy = self._get_policy()
        setattr(_policy, "rules", [self.rule])
        with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
                        return_value=_policy):
            self.qos_plugin.update_policy_bandwidth_limit_rule(
                self.ctxt, self.rule.id, self.policy.id, self.rule_data)
            self.mock_qos_load_attr.assert_called_once_with('rules')
            self._validate_driver_params('update_policy', self.ctxt)

        rules = [self.rule, self.min_bw_rule]
        setattr(_policy, "rules", rules)
        self.mock_qos_load_attr.reset_mock()
        with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
                        return_value=_policy):
            self.qos_plugin.update_policy_minimum_bandwidth_rule(
                self.ctxt, self.min_bw_rule.id,
                self.policy.id, self.rule_data)
            self.mock_qos_load_attr.assert_called_once_with('rules')
            self._validate_driver_params('update_policy', self.ctxt)

    def test_update_policy_rule_check_rule_bwlimit_less_than_minbw(self):
        _policy = self._get_policy()
        setattr(_policy, "rules", [self.rule])
        with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
                        return_value=_policy):
            self.qos_plugin.update_policy_bandwidth_limit_rule(
                self.ctxt, self.rule.id, self.policy.id, self.rule_data)
            self.mock_qos_load_attr.assert_called_once_with('rules')
            self._validate_driver_params('update_policy', self.ctxt)
        self.rule_data['minimum_bandwidth_rule']['min_kbps'] = 1000
        with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
                        return_value=_policy):
            self.assertRaises(
                qos_exc.QoSRuleParameterConflict,
                self.qos_plugin.update_policy_minimum_bandwidth_rule,
                self.ctxt, self.min_bw_rule.id,
                self.policy.id, self.rule_data)

    def test_update_policy_rule_check_rule_minbw_gr_than_bwlimit(self):
        _policy = self._get_policy()
        setattr(_policy, "rules", [self.min_bw_rule])
        with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
                        return_value=_policy):
            self.qos_plugin.update_policy_minimum_bandwidth_rule(
                self.ctxt, self.min_bw_rule.id, self.policy.id,
                self.rule_data)
            self.mock_qos_load_attr.assert_called_once_with('rules')
            self._validate_driver_params('update_policy', self.ctxt)
        self.rule_data['bandwidth_limit_rule']['max_kbps'] = 1
        with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
                        return_value=_policy):
            self.assertRaises(
                qos_exc.QoSRuleParameterConflict,
                self.qos_plugin.update_policy_bandwidth_limit_rule,
                self.ctxt, self.rule.id,
                self.policy.id, self.rule_data)

    def _get_policy(self):
        return policy_object.QosPolicy(
            self.ctxt, **self.policy_data['policy'])

    def test_update_policy_rule_bad_policy(self):
        _policy = policy_object.QosPolicy(
            self.ctxt, **self.policy_data['policy'])
        with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
                        return_value=_policy):
            setattr(_policy, "rules", [])
            self.assertRaises(
                qos_exc.QosRuleNotFound,
                self.qos_plugin.update_policy_bandwidth_limit_rule,
                self.ctxt, self.rule.id, self.policy.id,
                self.rule_data)

    @mock.patch.object(rule_object.QosBandwidthLimitRule, 'delete')
    def test_delete_policy_rule(self, mock_qos_rule_delete):
        mock_manager = mock.Mock()
        mock_manager.attach_mock(mock_qos_rule_delete, 'delete')
        mock_manager.attach_mock(self.qos_plugin.driver_manager, 'driver')
        mock_manager.reset_mock()

        _policy = policy_object.QosPolicy(
            self.ctxt, **self.policy_data['policy'])
        with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
                        return_value=_policy):
            setattr(_policy, "rules", [self.rule])
            self.qos_plugin.delete_policy_bandwidth_limit_rule(
                        self.ctxt, self.rule.id, _policy.id)
            self._validate_driver_params('update_policy', self.ctxt)

            rule_delete_mock_call = mock.call.delete()
            update_precommit_mock_call = mock.call.driver.call(
                'update_policy_precommit', self.ctxt, mock.ANY)
            update_mock_call = mock.call.driver.call(
                'update_policy', self.ctxt, mock.ANY)
            self.assertTrue(
                mock_manager.mock_calls.index(rule_delete_mock_call) <
                mock_manager.mock_calls.index(update_precommit_mock_call) <
                mock_manager.mock_calls.index(update_mock_call))

    def test_delete_policy_rule_bad_policy(self):
        _policy = policy_object.QosPolicy(
            self.ctxt, **self.policy_data['policy'])
        with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
                        return_value=_policy):
            setattr(_policy, "rules", [])
            self.assertRaises(
                qos_exc.QosRuleNotFound,
                self.qos_plugin.delete_policy_bandwidth_limit_rule,
                self.ctxt, self.rule.id, _policy.id)

    def test_get_policy_bandwidth_limit_rule(self):
        with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
                        return_value=self.policy):
            with mock.patch('neutron.objects.qos.rule.'
                            'QosBandwidthLimitRule.'
                            'get_object') as get_object_mock:
                self.qos_plugin.get_policy_bandwidth_limit_rule(
                    self.ctxt, self.rule.id, self.policy.id)
                get_object_mock.assert_called_once_with(self.ctxt,
                    id=self.rule.id)

    def test_get_policy_bandwidth_limit_rules_for_policy(self):
        with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
                        return_value=self.policy):
            with mock.patch('neutron.objects.qos.rule.'
                            'QosBandwidthLimitRule.'
                            'get_objects') as get_objects_mock:
                self.qos_plugin.get_policy_bandwidth_limit_rules(
                    self.ctxt, self.policy.id)
                get_objects_mock.assert_called_once_with(
                    self.ctxt, _pager=mock.ANY, qos_policy_id=self.policy.id)

    def test_get_policy_bandwidth_limit_rules_for_policy_with_filters(self):
        with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
                        return_value=self.policy):
            with mock.patch('neutron.objects.qos.rule.'
                            'QosBandwidthLimitRule.'
                            'get_objects') as get_objects_mock:

                filters = {'filter': 'filter_id'}
                self.qos_plugin.get_policy_bandwidth_limit_rules(
                    self.ctxt, self.policy.id, filters=filters)
                get_objects_mock.assert_called_once_with(
                    self.ctxt, _pager=mock.ANY,
                    qos_policy_id=self.policy.id,
                    filter='filter_id')

    def test_get_policy_for_nonexistent_policy(self):
        with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
                        return_value=None):
            self.assertRaises(
                qos_exc.QosPolicyNotFound,
                self.qos_plugin.get_policy,
                self.ctxt, self.policy.id)

    def test_get_policy_bandwidth_limit_rule_for_nonexistent_policy(self):
        with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
                        return_value=None):
            self.assertRaises(
                qos_exc.QosPolicyNotFound,
                self.qos_plugin.get_policy_bandwidth_limit_rule,
                self.ctxt, self.rule.id, self.policy.id)

    def test_get_policy_bandwidth_limit_rules_for_nonexistent_policy(self):
        with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
                        return_value=None):
            self.assertRaises(
                qos_exc.QosPolicyNotFound,
                self.qos_plugin.get_policy_bandwidth_limit_rules,
                self.ctxt, self.policy.id)

    def test_create_policy_dscp_marking_rule(self):
        _policy = policy_object.QosPolicy(
            self.ctxt, **self.policy_data['policy'])
        with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
                        return_value=_policy):
            setattr(_policy, "rules", [self.dscp_rule])
            self.qos_plugin.create_policy_dscp_marking_rule(
                self.ctxt, self.policy.id, self.rule_data)
            self._validate_driver_params('update_policy', self.ctxt)

    def test_update_policy_dscp_marking_rule(self):
        _policy = policy_object.QosPolicy(
            self.ctxt, **self.policy_data['policy'])
        with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
                        return_value=_policy):
            setattr(_policy, "rules", [self.dscp_rule])
            self.qos_plugin.update_policy_dscp_marking_rule(
                self.ctxt, self.dscp_rule.id, self.policy.id, self.rule_data)
            self._validate_driver_params('update_policy', self.ctxt)

    def test_delete_policy_dscp_marking_rule(self):
        _policy = policy_object.QosPolicy(
            self.ctxt, **self.policy_data['policy'])
        with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
                        return_value=_policy):
            setattr(_policy, "rules", [self.dscp_rule])
            self.qos_plugin.delete_policy_dscp_marking_rule(
                self.ctxt, self.dscp_rule.id, self.policy.id)
            self._validate_driver_params('update_policy', self.ctxt)

    def test_get_policy_dscp_marking_rules(self):
        with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
                        return_value=self.policy):
            with mock.patch('neutron.objects.qos.rule.'
                            'QosDscpMarkingRule.'
                            'get_objects') as get_objects_mock:
                self.qos_plugin.get_policy_dscp_marking_rules(
                    self.ctxt, self.policy.id)
                get_objects_mock.assert_called_once_with(
                    self.ctxt, _pager=mock.ANY, qos_policy_id=self.policy.id)

    def test_get_policy_dscp_marking_rules_for_policy_with_filters(self):
        with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
                        return_value=self.policy):
            with mock.patch('neutron.objects.qos.rule.'
                            'QosDscpMarkingRule.'
                            'get_objects') as get_objects_mock:

                filters = {'filter': 'filter_id'}
                self.qos_plugin.get_policy_dscp_marking_rules(
                    self.ctxt, self.policy.id, filters=filters)
                get_objects_mock.assert_called_once_with(
                    self.ctxt, qos_policy_id=self.policy.id,
                    _pager=mock.ANY, filter='filter_id')

    def test_get_policy_dscp_marking_rule_for_nonexistent_policy(self):
        with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
                        return_value=None):
            self.assertRaises(
                qos_exc.QosPolicyNotFound,
                self.qos_plugin.get_policy_dscp_marking_rule,
                self.ctxt, self.dscp_rule.id, self.policy.id)

    def test_get_policy_dscp_marking_rules_for_nonexistent_policy(self):
        with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
                        return_value=None):
            self.assertRaises(
                qos_exc.QosPolicyNotFound,
                self.qos_plugin.get_policy_dscp_marking_rules,
                self.ctxt, self.policy.id)

    def test_get_policy_minimum_bandwidth_rule(self):
        with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
                        return_value=self.policy):
            with mock.patch('neutron.objects.qos.rule.'
                            'QosMinimumBandwidthRule.'
                            'get_object') as get_object_mock:
                self.qos_plugin.get_policy_minimum_bandwidth_rule(
                    self.ctxt, self.rule.id, self.policy.id)
                get_object_mock.assert_called_once_with(self.ctxt,
                    id=self.rule.id)

    def test_get_policy_minimum_bandwidth_rules_for_policy(self):
        with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
                        return_value=self.policy):
            with mock.patch('neutron.objects.qos.rule.'
                            'QosMinimumBandwidthRule.'
                            'get_objects') as get_objects_mock:
                self.qos_plugin.get_policy_minimum_bandwidth_rules(
                    self.ctxt, self.policy.id)
                get_objects_mock.assert_called_once_with(
                    self.ctxt, _pager=mock.ANY, qos_policy_id=self.policy.id)

    def test_get_policy_minimum_bandwidth_rules_for_policy_with_filters(self):
        with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
                        return_value=self.policy):
            with mock.patch('neutron.objects.qos.rule.'
                            'QosMinimumBandwidthRule.'
                            'get_objects') as get_objects_mock:

                filters = {'filter': 'filter_id'}
                self.qos_plugin.get_policy_minimum_bandwidth_rules(
                    self.ctxt, self.policy.id, filters=filters)
                get_objects_mock.assert_called_once_with(
                    self.ctxt, _pager=mock.ANY,
                    qos_policy_id=self.policy.id,
                    filter='filter_id')

    def test_get_policy_minimum_bandwidth_rule_for_nonexistent_policy(self):
        with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
                        return_value=None):
            self.assertRaises(
                qos_exc.QosPolicyNotFound,
                self.qos_plugin.get_policy_minimum_bandwidth_rule,
                self.ctxt, self.rule.id, self.policy.id)

    def test_get_policy_minimum_bandwidth_rules_for_nonexistent_policy(self):
        with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
                        return_value=None):
            self.assertRaises(
                qos_exc.QosPolicyNotFound,
                self.qos_plugin.get_policy_minimum_bandwidth_rules,
                self.ctxt, self.policy.id)

    def test_create_policy_rule_for_nonexistent_policy(self):
        with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
                        return_value=None):
            self.assertRaises(
                qos_exc.QosPolicyNotFound,
                self.qos_plugin.create_policy_bandwidth_limit_rule,
                self.ctxt, self.policy.id, self.rule_data)

    def test_update_policy_rule_for_nonexistent_policy(self):
        with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
                        return_value=None):
            self.assertRaises(
                qos_exc.QosPolicyNotFound,
                self.qos_plugin.update_policy_bandwidth_limit_rule,
                self.ctxt, self.rule.id, self.policy.id, self.rule_data)

    def test_delete_policy_rule_for_nonexistent_policy(self):
        with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
                        return_value=None):
            self.assertRaises(
                qos_exc.QosPolicyNotFound,
                self.qos_plugin.delete_policy_bandwidth_limit_rule,
                self.ctxt, self.rule.id, self.policy.id)

    def test_verify_bad_method_call(self):
        self.assertRaises(AttributeError, getattr, self.qos_plugin,
                          'create_policy_bandwidth_limit_rules')

    def test_get_rule_type(self):
        admin_ctxt = context.get_admin_context()
        drivers_details = [{
            'name': 'fake-driver',
            'supported_parameters': [{
                'parameter_name': 'max_kbps',
                'parameter_type': lib_constants.VALUES_TYPE_RANGE,
                'parameter_range': {'start': 0, 'end': 100}
            }]
        }]
        with mock.patch.object(
            qos_plugin.QoSPlugin, "supported_rule_type_details",
            return_value=drivers_details
        ):
            rule_type_details = self.qos_plugin.get_rule_type(
                admin_ctxt, qos_consts.RULE_TYPE_BANDWIDTH_LIMIT)
            self.assertEqual(
                qos_consts.RULE_TYPE_BANDWIDTH_LIMIT,
                rule_type_details['type'])
            self.assertEqual(
                drivers_details, rule_type_details['drivers'])

    def test_get_rule_type_as_user(self):
        self.assertRaises(
            lib_exc.NotAuthorized,
            self.qos_plugin.get_rule_type,
            self.ctxt, qos_consts.RULE_TYPE_BANDWIDTH_LIMIT)

    def test_get_rule_types(self):
        filters = {'type': 'type_id'}
        with mock.patch.object(qos_plugin.QoSPlugin, 'supported_rule_types',
                               return_value=qos_consts.VALID_RULE_TYPES):
            types = self.qos_plugin.get_rule_types(self.ctxt, filters=filters)
            self.assertEqual(sorted(qos_consts.VALID_RULE_TYPES),
                             sorted(type_['type'] for type_ in types))

    @mock.patch('neutron.objects.ports.Port')
    @mock.patch('neutron.objects.qos.policy.QosPolicy')
    def test_rule_notification_and_driver_ordering(self, qos_policy_mock,
                                                   port_mock):
        rule_cls_mock = mock.Mock()
        rule_cls_mock.rule_type = 'fake'

        rule_actions = {'create': [self.ctxt, rule_cls_mock,
                                   self.policy.id, {'fake_rule': {}}],
                        'update': [self.ctxt, rule_cls_mock,
                                   self.rule.id,
                                   self.policy.id, {'fake_rule': {}}],
                        'delete': [self.ctxt, rule_cls_mock,
                                   self.rule.id, self.policy.id]}

        mock_manager = mock.Mock()
        mock_manager.attach_mock(qos_policy_mock, 'QosPolicy')
        mock_manager.attach_mock(port_mock, 'Port')
        mock_manager.attach_mock(rule_cls_mock, 'RuleCls')
        mock_manager.attach_mock(self.qos_plugin.driver_manager, 'driver')

        for action, arguments in rule_actions.items():
            mock_manager.reset_mock()
            method = getattr(self.qos_plugin, "%s_policy_rule" % action)
            method(*arguments)

            # some actions get rule from policy
            get_rule_mock_call = getattr(
                mock.call.QosPolicy.get_policy_obj().get_rule_by_id(),
                action)()
            # some actions construct rule from class reference
            rule_mock_call = getattr(mock.call.RuleCls(), action)()

            driver_mock_call = mock.call.driver.call('update_policy',
                                                     self.ctxt, mock.ANY)

            if rule_mock_call in mock_manager.mock_calls:
                action_index = mock_manager.mock_calls.index(rule_mock_call)
            else:
                action_index = mock_manager.mock_calls.index(
                    get_rule_mock_call)

            self.assertLess(
                action_index, mock_manager.mock_calls.index(driver_mock_call))

    def test_create_policy_packet_rate_limit_rule(self):
        _policy = policy_object.QosPolicy(
            self.ctxt, **self.policy_data['policy'])
        with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
                        return_value=_policy):
            setattr(_policy, "rules", [self.pps_rule])
            self.qos_plugin.create_policy_packet_rate_limit_rule(
                self.ctxt, self.policy.id, self.rule_data)
            self._validate_driver_params('update_policy', self.ctxt)

    def test_create_policy_pps_rule_duplicates(self):
        _policy = self._get_policy()
        setattr(_policy, "rules", [self.pps_rule])
        new_rule_data = {
            'packet_rate_limit_rule': {
                'max_kpps': 400,
                'direction': self.pps_rule.direction
            }
        }
        with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
                        return_value=_policy) as mock_qos_get_obj:
            self.assertRaises(
                qos_exc.QoSRulesConflict,
                self.qos_plugin.create_policy_packet_rate_limit_rule,
                self.ctxt, _policy.id, new_rule_data)
            mock_qos_get_obj.assert_called_once_with(self.ctxt, id=_policy.id)

    def test_update_policy_packet_rate_limit_rule(self):
        _policy = policy_object.QosPolicy(
            self.ctxt, **self.policy_data['policy'])
        with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
                        return_value=_policy):
            setattr(_policy, "rules", [self.pps_rule])
            self.qos_plugin.update_policy_packet_rate_limit_rule(
                self.ctxt, self.pps_rule.id, self.policy.id, self.rule_data)
            self._validate_driver_params('update_policy', self.ctxt)

    def test_update_policy_pps_rule_bad_policy(self):
        _policy = policy_object.QosPolicy(
            self.ctxt, **self.policy_data['policy'])
        with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
                        return_value=_policy):
            setattr(_policy, "rules", [])
            self.assertRaises(
                qos_exc.QosRuleNotFound,
                self.qos_plugin.update_policy_packet_rate_limit_rule,
                self.ctxt, self.pps_rule.id, self.policy.id,
                self.rule_data)

    def test_delete_policy_packet_rate_limit_rule(self):
        _policy = policy_object.QosPolicy(
            self.ctxt, **self.policy_data['policy'])
        with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
                        return_value=_policy):
            setattr(_policy, "rules", [self.pps_rule])
            self.qos_plugin.delete_policy_packet_rate_limit_rule(
                self.ctxt, self.pps_rule.id, self.policy.id)
            self._validate_driver_params('update_policy', self.ctxt)

    def test_delete_policy_pps_rule_bad_policy(self):
        _policy = policy_object.QosPolicy(
            self.ctxt, **self.policy_data['policy'])
        with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
                        return_value=_policy):
            setattr(_policy, "rules", [])
            self.assertRaises(
                qos_exc.QosRuleNotFound,
                self.qos_plugin.delete_policy_packet_rate_limit_rule,
                self.ctxt, self.pps_rule.id, _policy.id)

    def test_get_policy_packet_rate_limit_rule(self):
        with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
                        return_value=self.policy):
            with mock.patch('neutron.objects.qos.rule.'
                            'QosPacketRateLimitRule.'
                            'get_object') as get_object_mock:
                self.qos_plugin.get_policy_packet_rate_limit_rule(
                    self.ctxt, self.pps_rule.id, self.policy.id)
                get_object_mock.assert_called_once_with(self.ctxt,
                                                        id=self.pps_rule.id)

    def test_get_policy_packet_rate_limit_rules_for_policy(self):
        with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
                        return_value=self.policy):
            with mock.patch('neutron.objects.qos.rule.'
                            'QosPacketRateLimitRule.'
                            'get_objects') as get_objects_mock:
                self.qos_plugin.get_policy_packet_rate_limit_rules(
                    self.ctxt, self.policy.id)
                get_objects_mock.assert_called_once_with(
                    self.ctxt, _pager=mock.ANY, qos_policy_id=self.policy.id)

    def test_get_policy_packet_rate_limit_rules_for_policy_with_filters(self):
        with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
                        return_value=self.policy):
            with mock.patch('neutron.objects.qos.rule.'
                            'QosPacketRateLimitRule.'
                            'get_objects') as get_objects_mock:
                filters = {'filter': 'filter_id'}
                self.qos_plugin.get_policy_packet_rate_limit_rules(
                    self.ctxt, self.policy.id, filters=filters)
                get_objects_mock.assert_called_once_with(
                    self.ctxt, _pager=mock.ANY,
                    qos_policy_id=self.policy.id,
                    filter='filter_id')

    def test_get_policy_packet_rate_limit_rule_for_nonexistent_policy(self):
        with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
                        return_value=None):
            self.assertRaises(
                qos_exc.QosPolicyNotFound,
                self.qos_plugin.get_policy_packet_rate_limit_rule,
                self.ctxt, self.pps_rule.id, self.policy.id)

    def test_get_policy_packet_rate_limit_rules_for_nonexistent_policy(self):
        with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
                        return_value=None):
            self.assertRaises(
                qos_exc.QosPolicyNotFound,
                self.qos_plugin.get_policy_packet_rate_limit_rules,
                self.ctxt, self.policy.id)

    def test_create_policy_pps_rule_for_nonexistent_policy(self):
        with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
                        return_value=None):
            self.assertRaises(
                qos_exc.QosPolicyNotFound,
                self.qos_plugin.create_policy_packet_rate_limit_rule,
                self.ctxt, self.policy.id, self.rule_data)

    def test_update_policy_pps_rule_for_nonexistent_policy(self):
        with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
                        return_value=None):
            self.assertRaises(
                qos_exc.QosPolicyNotFound,
                self.qos_plugin.update_policy_packet_rate_limit_rule,
                self.ctxt, self.pps_rule.id, self.policy.id, self.rule_data)

    def test_delete_policy_pps_rule_for_nonexistent_policy(self):
        with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
                        return_value=None):
            self.assertRaises(
                qos_exc.QosPolicyNotFound,
                self.qos_plugin.delete_policy_packet_rate_limit_rule,
                self.ctxt, self.pps_rule.id, self.policy.id)

    def test_get_pps_rule_type(self):
        admin_ctxt = context.get_admin_context()
        drivers_details = [{
            'name': 'fake-driver',
            'supported_parameters': [{
                'parameter_name': 'max_kpps',
                'parameter_type': lib_constants.VALUES_TYPE_RANGE,
                'parameter_range': {'start': 0, 'end': 100}
            }]
        }]
        with mock.patch.object(
            qos_plugin.QoSPlugin, "supported_rule_type_details",
            return_value=drivers_details
        ):
            rule_type_details = self.qos_plugin.get_rule_type(
                admin_ctxt, qos_consts.RULE_TYPE_PACKET_RATE_LIMIT)
            self.assertEqual(
                qos_consts.RULE_TYPE_PACKET_RATE_LIMIT,
                rule_type_details['type'])
            self.assertEqual(
                drivers_details, rule_type_details['drivers'])

    def test_get_pps_rule_type_as_user(self):
        self.assertRaises(
            lib_exc.NotAuthorized,
            self.qos_plugin.get_rule_type,
            self.ctxt, qos_consts.RULE_TYPE_PACKET_RATE_LIMIT)

    def test_create_min_pps_rule_on_bound_port(self):
        _policy = self._get_policy()
        setattr(_policy, "rules", [self.min_pps_rule])
        segment = network_object.NetworkSegment(
            physical_network='fake physnet')
        net = network_object.Network(
            self.ctxt,
            segments=[segment])
        port = ports_object.Port(
            self.ctxt,
            id=uuidutils.generate_uuid(),
            network_id=uuidutils.generate_uuid(),
            device_owner='compute:fake-zone')
        with mock.patch(
                'neutron.objects.qos.policy.QosPolicy.get_object',
                return_value=_policy), \
            mock.patch(
                'neutron.objects.network.Network.get_object',
                return_value=net), \
            mock.patch.object(
                self.qos_plugin,
                '_get_ports_with_policy',
                return_value=[port]):
            self.assertRaises(
                NotImplementedError,
                self.qos_plugin.create_policy_minimum_packet_rate_rule,
                self.ctxt, _policy.id, self.rule_data)

    def test_create_min_pps_rule_on_unbound_port(self):
        _policy = self._get_policy()
        setattr(_policy, "rules", [self.min_pps_rule])
        segment = network_object.NetworkSegment(
            physical_network='fake physnet')
        net = network_object.Network(
            self.ctxt,
            segments=[segment])
        port = ports_object.Port(
            self.ctxt,
            id=uuidutils.generate_uuid(),
            network_id=uuidutils.generate_uuid(),
            device_owner='')
        with mock.patch(
                'neutron.objects.qos.policy.QosPolicy.get_object',
                return_value=_policy), \
            mock.patch(
                'neutron.objects.network.Network.get_object',
                return_value=net), \
            mock.patch.object(
                self.qos_plugin,
                '_get_ports_with_policy',
                return_value=[port]):
            try:
                self.qos_plugin.create_policy_minimum_packet_rate_rule(
                    self.ctxt, _policy.id, self.rule_data)
            except NotImplementedError:
                self.fail()

    def test_create_policy_rule_check_rule_min_pps_direction_conflict(self):
        _policy = self._get_policy()
        self.rule_data['minimum_packet_rate_rule']['direction'] = 'any'
        setattr(_policy, "rules", [self.min_pps_rule])
        rules = [
            {
                'minimum_packet_rate_rule': {
                    'id': uuidutils.generate_uuid(),
                    'min_kpps': 10,
                    'direction': 'ingress'
                }
            },
            {
                'minimum_packet_rate_rule': {
                    'id': uuidutils.generate_uuid(),
                    'min_kpps': 10,
                    'direction': 'egress'
                }
            },
        ]
        for new_rule_data in rules:
            with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
                            return_value=_policy) as mock_qos_get_obj:
                self.assertRaises(qos_exc.QoSRuleParameterConflict,
                    self.qos_plugin.create_policy_minimum_packet_rate_rule,
                    self.ctxt, self.policy.id, new_rule_data)
                mock_qos_get_obj.assert_called_once_with(self.ctxt,
                                                         id=_policy.id)

        for rule_data in rules:
            min_pps_rule = rule_object.QosMinimumPacketRateRule(
                self.ctxt, **rule_data['minimum_packet_rate_rule'])
            setattr(_policy, "rules", [min_pps_rule])
            with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
                            return_value=_policy) as mock_qos_get_obj:
                self.assertRaises(qos_exc.QoSRuleParameterConflict,
                    self.qos_plugin.create_policy_minimum_packet_rate_rule,
                    self.ctxt, self.policy.id, self.rule_data)
                mock_qos_get_obj.assert_called_once_with(self.ctxt,
                                                         id=_policy.id)

    def test_create_policy_min_pps_rule(self):
        _policy = self._get_policy()
        setattr(_policy, "rules", [self.min_pps_rule])
        with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
                        return_value=_policy):
            self.qos_plugin.create_policy_minimum_packet_rate_rule(
                self.ctxt, self.policy.id, self.rule_data)
            self._validate_driver_params('update_policy', self.ctxt)

    def test_create_policy_min_pps_rule_duplicates(self):
        _policy = self._get_policy()
        setattr(_policy, "rules", [self.min_pps_rule])
        new_rule_data = {
            'minimum_packet_rate_rule': {
                'id': uuidutils.generate_uuid(),
                'min_kpps': 1234,
                'direction': self.min_pps_rule.direction,
            },
        }

        with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
                        return_value=_policy) as mock_qos_get_obj:
            self.assertRaises(
                qos_exc.QoSRulesConflict,
                self.qos_plugin.create_policy_minimum_packet_rate_rule,
                self.ctxt, _policy.id, new_rule_data)
            mock_qos_get_obj.assert_called_once_with(self.ctxt, id=_policy.id)

    def test_create_policy_min_pps_rule_for_nonexistent_policy(self):
        with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
                        return_value=None):
            self.assertRaises(
                qos_exc.QosPolicyNotFound,
                self.qos_plugin.create_policy_minimum_packet_rate_rule,
                self.ctxt, self.policy.id, self.rule_data)

    def test_update_policy_min_pps_rule(self):
        _policy = self._get_policy()
        setattr(_policy, "rules", [self.min_pps_rule])
        with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
                        return_value=_policy):
            self.qos_plugin.update_policy_minimum_packet_rate_rule(
                self.ctxt, self.min_pps_rule.id, self.policy.id,
                self.rule_data)
            self._validate_driver_params('update_policy', self.ctxt)

    def test_update_policy_rule_check_rule_min_pps_direction_conflict(self):
        _policy = self._get_policy()
        rules_data = [
            {
                'minimum_packet_rate_rule': {
                    'id': uuidutils.generate_uuid(),
                    'min_kpps': 10,
                    'direction': 'ingress'
                }
            },
            {
                'minimum_packet_rate_rule': {
                    'id': uuidutils.generate_uuid(),
                    'min_kpps': 10,
                    'direction': 'egress'
                }
            },
        ]

        self.rule_data['minimum_packet_rate_rule']['direction'] = 'any'
        for rule_data in rules_data:
            rules = [
                rule_object.QosMinimumPacketRateRule(
                    self.ctxt, **rules_data[0]['minimum_packet_rate_rule']),
                rule_object.QosMinimumPacketRateRule(
                    self.ctxt, **rules_data[1]['minimum_packet_rate_rule']),
            ]
            setattr(_policy, 'rules', rules)
            with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
                            return_value=_policy) as mock_qos_get_obj:
                self.assertRaises(qos_exc.QoSRuleParameterConflict,
                    self.qos_plugin.update_policy_minimum_packet_rate_rule,
                    self.ctxt, rule_data['minimum_packet_rate_rule']['id'],
                    self.policy.id, self.rule_data)
                mock_qos_get_obj.assert_called_once_with(self.ctxt,
                                                         id=_policy.id)

    def test_update_policy_min_pps_rule_bad_policy(self):
        _policy = self._get_policy()
        setattr(_policy, "rules", [])
        with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
                        return_value=_policy):
            self.assertRaises(
                qos_exc.QosRuleNotFound,
                self.qos_plugin.update_policy_minimum_packet_rate_rule,
                self.ctxt, self.min_pps_rule.id, self.policy.id,
                self.rule_data)

    def test_update_policy_min_pps_rule_for_nonexistent_policy(self):
        with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
                        return_value=None):
            self.assertRaises(
                qos_exc.QosPolicyNotFound,
                self.qos_plugin.update_policy_minimum_packet_rate_rule,
                self.ctxt, self.min_pps_rule.id, self.policy.id,
                self.rule_data)

    def test_delete_policy_min_pps_rule(self):
        _policy = self._get_policy()
        setattr(_policy, "rules", [self.min_pps_rule])
        with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
                        return_value=_policy):
            self.qos_plugin.delete_policy_minimum_packet_rate_rule(
                self.ctxt, self.min_pps_rule.id, self.policy.id)
            self._validate_driver_params('update_policy', self.ctxt)

    def test_delete_policy_min_pps_rule_bad_policy(self):
        _policy = self._get_policy()
        setattr(_policy, "rules", [])
        with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
                        return_value=_policy):
            self.assertRaises(
                qos_exc.QosRuleNotFound,
                self.qos_plugin.delete_policy_minimum_packet_rate_rule,
                self.ctxt, self.min_pps_rule.id, _policy.id)

    def test_delete_policy_min_pps_rule_for_nonexistent_policy(self):
        with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
                        return_value=None):
            self.assertRaises(
                qos_exc.QosPolicyNotFound,
                self.qos_plugin.delete_policy_minimum_packet_rate_rule,
                self.ctxt, self.min_pps_rule.id, self.policy.id)

    def test_get_policy_min_pps_rule(self):
        with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
                        return_value=self.policy):
            with mock.patch('neutron.objects.qos.rule.'
                            'QosMinimumPacketRateRule.'
                            'get_object') as get_object_mock:
                self.qos_plugin.get_policy_minimum_packet_rate_rule(
                    self.ctxt, self.min_pps_rule.id, self.policy.id)
                get_object_mock.assert_called_once_with(
                    self.ctxt, id=self.min_pps_rule.id)

    def test_get_policy_min_pps_rules_for_policy(self):
        with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
                        return_value=self.policy):
            with mock.patch('neutron.objects.qos.rule.'
                            'QosMinimumPacketRateRule.'
                            'get_objects') as get_objects_mock:
                self.qos_plugin.get_policy_minimum_packet_rate_rules(
                    self.ctxt, self.policy.id)
                get_objects_mock.assert_called_once_with(
                    self.ctxt, _pager=mock.ANY, qos_policy_id=self.policy.id)

    def test_get_policy_min_pps_rules_for_policy_with_filters(self):
        with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
                        return_value=self.policy):
            with mock.patch('neutron.objects.qos.rule.'
                            'QosMinimumPacketRateRule.'
                            'get_objects') as get_objects_mock:
                filters = {'filter': 'filter_id'}
                self.qos_plugin.get_policy_minimum_packet_rate_rules(
                    self.ctxt, self.policy.id, filters=filters)
                get_objects_mock.assert_called_once_with(
                    self.ctxt, _pager=mock.ANY,
                    qos_policy_id=self.policy.id,
                    filter='filter_id')

    def test_get_policy_min_pps_rule_for_nonexistent_policy(self):
        with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
                        return_value=None):
            self.assertRaises(
                qos_exc.QosPolicyNotFound,
                self.qos_plugin.get_policy_minimum_packet_rate_rule,
                self.ctxt, self.min_pps_rule.id, self.policy.id)

    def test_get_policy_min_pps_rules_for_nonexistent_policy(self):
        with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
                        return_value=None):
            self.assertRaises(
                qos_exc.QosPolicyNotFound,
                self.qos_plugin.get_policy_minimum_packet_rate_rules,
                self.ctxt, self.policy.id)

    def test_get_min_pps_rule_type(self):
        admin_ctxt = context.get_admin_context()
        drivers_details = [{
            'name': 'fake-driver',
            'supported_parameters': [{
                'parameter_name': 'min_kpps',
                'parameter_type': lib_constants.VALUES_TYPE_RANGE,
                'parameter_range': {'start': 0, 'end': 100}
            }]
        }]
        with mock.patch.object(
            qos_plugin.QoSPlugin, "supported_rule_type_details",
            return_value=drivers_details
        ):
            rule_type_details = self.qos_plugin.get_rule_type(
                admin_ctxt, qos_consts.RULE_TYPE_MINIMUM_PACKET_RATE)
            self.assertEqual(
                qos_consts.RULE_TYPE_MINIMUM_PACKET_RATE,
                rule_type_details['type'])
            self.assertEqual(
                drivers_details, rule_type_details['drivers'])

    def test_get_min_pps_rule_type_as_user(self):
        self.assertRaises(
            lib_exc.NotAuthorized,
            self.qos_plugin.get_rule_type,
            self.ctxt, qos_consts.RULE_TYPE_MINIMUM_PACKET_RATE)


class QoSRuleAliasTestExtensionManager(object):

    def get_resources(self):
        return qos_rules_alias.Qos_rules_alias.get_resources()

    def get_actions(self):
        return []

    def get_request_extensions(self):
        return []


class QoSRuleAliasMinimumPacketRateTestExtensionManager(object):

    def get_resources(self):
        return qos_pps_minimum_rule_alias.Qos_pps_minimum_rule_alias.\
            get_resources()

    def get_actions(self):
        return []

    def get_request_extensions(self):
        return []


class TestQoSRuleAlias(test_db_base_plugin_v2.NeutronDbPluginV2TestCase):

    def setUp(self):
        # Remove MissingAuthPlugin exception from logs
        self.patch_notifier = mock.patch(
            'neutron.notifiers.batch_notifier.BatchNotifier._notify')
        self.patch_notifier.start()
        plugin = 'ml2'
        service_plugins = {'qos_plugin_name': SERVICE_PLUGIN_KLASS}
        ext_mgr = QoSRuleAliasTestExtensionManager()
        super(TestQoSRuleAlias, self).setUp(plugin=plugin, ext_mgr=ext_mgr,
                                            service_plugins=service_plugins)
        self.qos_plugin = directory.get_plugin(plugins_constants.QOS)

        self.ctxt = context.Context('fake_user', 'fake_tenant')
        self.rule_objects = {
            'bandwidth_limit': rule_object.QosBandwidthLimitRule,
            'dscp_marking': rule_object.QosDscpMarkingRule,
            'minimum_bandwidth': rule_object.QosMinimumBandwidthRule
        }

        self.qos_policy_id = uuidutils.generate_uuid()
        self.rule_data = {
            'bandwidth_limit_rule': {'max_kbps': 100,
                                     'max_burst_kbps': 150},
            'dscp_marking_rule': {'dscp_mark': 16},
            'minimum_bandwidth_rule': {'min_kbps': 10}
        }

    def _update_rule(self, rule_type, rule_id, **kwargs):
        data = {'alias_%s_rule' % rule_type: kwargs}
        resource = '%s/alias-%s-rules' % (qos.ALIAS,
                                          rule_type.replace('_', '-'))
        request = self.new_update_request(resource, data, rule_id, self.fmt)
        res = request.get_response(self.ext_api)
        if res.status_int >= webob.exc.HTTPClientError.code:
            raise webob.exc.HTTPClientError(code=res.status_int)
        return self.deserialize(self.fmt, res)

    def _show_rule(self, rule_type, rule_id):
        resource = '%s/alias-%s-rules' % (qos.ALIAS,
                                          rule_type.replace('_', '-'))
        request = self.new_show_request(resource, rule_id, self.fmt)
        res = request.get_response(self.ext_api)
        if res.status_int >= webob.exc.HTTPClientError.code:
            raise webob.exc.HTTPClientError(code=res.status_int)
        return self.deserialize(self.fmt, res)

    def _delete_rule(self, rule_type, rule_id):
        resource = '%s/alias-%s-rules' % (qos.ALIAS,
                                          rule_type.replace('_', '-'))
        request = self.new_delete_request(resource, rule_id, self.fmt)
        res = request.get_response(self.ext_api)
        if res.status_int >= webob.exc.HTTPClientError.code:
            raise webob.exc.HTTPClientError(code=res.status_int)

    @mock.patch.object(qos_plugin.QoSPlugin, "update_policy_rule")
    def test_update_rule(self, update_policy_rule_mock):
        calls = []
        for rule_type, rule_object_class in self.rule_objects.items():
            rule_id = uuidutils.generate_uuid()
            rule_data_name = '%s_rule' % rule_type
            data = self.rule_data[rule_data_name]
            rule = rule_object_class(self.ctxt, id=rule_id,
                                     qos_policy_id=self.qos_policy_id,
                                     **data)
            with mock.patch(
                'neutron.objects.qos.rule.QosRule.get_object',
                return_value=rule
            ), mock.patch.object(self.qos_plugin, 'get_policy_rule',
                                 return_value=rule.to_dict()):
                self._update_rule(rule_type, rule_id, **data)
            calls.append(mock.call(mock.ANY, rule_object_class, rule_id,
                                   self.qos_policy_id, {rule_data_name: data}))
        update_policy_rule_mock.assert_has_calls(calls, any_order=True)

    @mock.patch.object(qos_plugin.QoSPlugin, "get_policy_rule")
    def test_show_rule(self, get_policy_rule_mock):
        calls = []
        for rule_type, rule_object_class in self.rule_objects.items():
            rule_id = uuidutils.generate_uuid()
            rule_data_name = '%s_rule' % rule_type
            data = self.rule_data[rule_data_name]
            rule = rule_object_class(self.ctxt, id=rule_id,
                                     qos_policy_id=self.qos_policy_id,
                                     **data)
            with mock.patch('neutron.objects.qos.rule.QosRule.get_object',
                            return_value=rule):
                self._show_rule(rule_type, rule_id)
            calls.append(mock.call(mock.ANY, rule_object_class, rule_id,
                                   self.qos_policy_id))
        get_policy_rule_mock.assert_has_calls(calls, any_order=True)

    @mock.patch.object(qos_plugin.QoSPlugin, "delete_policy_rule")
    def test_delete_rule(self, delete_policy_rule_mock):
        calls = []
        for rule_type, rule_object_class in self.rule_objects.items():
            rule_id = uuidutils.generate_uuid()
            rule_data_name = '%s_rule' % rule_type
            data = self.rule_data[rule_data_name]
            rule = rule_object_class(self.ctxt, id=rule_id,
                                     qos_policy_id=self.qos_policy_id,
                                     **data)
            with mock.patch(
                'neutron.objects.qos.rule.QosRule.get_object',
                return_value=rule
            ), mock.patch.object(self.qos_plugin, 'get_policy_rule',
                                 return_value=rule.to_dict()):
                self._delete_rule(rule_type, rule_id)
            calls.append(mock.call(mock.ANY, rule_object_class, rule_id,
                                   self.qos_policy_id))
        delete_policy_rule_mock.assert_has_calls(calls, any_order=True)

    def test_show_non_existing_rule(self):
        for rule_type, rule_object_class in self.rule_objects.items():
            rule_id = uuidutils.generate_uuid()
            with mock.patch('neutron.objects.qos.rule.QosRule.get_object',
                            return_value=None):
                resource = '%s/alias-%s-rules' % (qos.ALIAS,
                                                  rule_type.replace('_', '-'))
                request = self.new_show_request(resource, rule_id, self.fmt)
                res = request.get_response(self.ext_api)
                self.assertEqual(webob.exc.HTTPNotFound.code, res.status_int)


class TestQoSRuleAliasMinimumPacketRate(TestQoSRuleAlias):
    def setUp(self):
        # Remove MissingAuthPlugin exception from logs
        self.patch_notifier = mock.patch(
            'neutron.notifiers.batch_notifier.BatchNotifier._notify')
        self.patch_notifier.start()
        plugin = 'ml2'
        service_plugins = {'qos_plugin_name': SERVICE_PLUGIN_KLASS}
        ext_mgr = QoSRuleAliasMinimumPacketRateTestExtensionManager()
        super(TestQoSRuleAlias, self).setUp(plugin=plugin, ext_mgr=ext_mgr,
                                            service_plugins=service_plugins)
        self.qos_plugin = directory.get_plugin(plugins_constants.QOS)

        self.ctxt = context.Context('fake_user', 'fake_tenant')
        self.rule_objects = {
            'minimum_packet_rate': rule_object.QosMinimumPacketRateRule
        }

        self.qos_policy_id = uuidutils.generate_uuid()
        self.rule_data = {
            'minimum_packet_rate_rule': {'min_kpps': 10, 'direction': 'any'}
        }


class TestQosPluginDB(base.BaseQosTestCase):

    PORT_ID = 'f02f160e-1612-11ec-b2b8-bf60ab98186c'

    QOS_MIN_BW_RULE_ID = '8bf8eb46-160e-11ec-8024-9f96be32099d'
    # uuid -v5 f02f160e-1612-11ec-b2b8-bf60ab98186c
    # 8bf8eb46-160e-11ec-8024-9f96be32099d
    MIN_BW_REQUEST_GROUP_UUID = 'c8bc1b27-59a1-5135-aa33-aeecad6093f4'
    MIN_BW_RP = 'd7bea120-1626-11ec-9148-c32debfcf0f6'

    QOS_MIN_PPS_RULE_ID = '6ac5db7e-1626-11ec-8c7f-0b70dbb8a8eb'
    # uuid -v5 f02f160e-1612-11ec-b2b8-bf60ab98186c
    # 6ac5db7e-1626-11ec-8c7f-0b70dbb8a8eb
    MIN_PPS_REQUEST_GROUP_UUID = '995008f4-f120-547a-b051-428b89076067'
    MIN_PPS_RP = 'e16161f4-1626-11ec-a5a2-1fc9396e27cc'

    def setUp(self):
        super(TestQosPluginDB, self).setUp()
        self.setup_coreplugin(load_plugins=False)
        cfg.CONF.set_override("core_plugin", DB_PLUGIN_KLASS)
        cfg.CONF.set_override("service_plugins", ["qos"])

        manager.init()
        self.qos_plugin = directory.get_plugin(plugins_constants.QOS)
        self.qos_plugin.driver_manager = mock.Mock()
        self.rpc_push = mock.patch('neutron.api.rpc.handlers.resources_rpc'
                                   '.ResourcesPushRpcApi.push').start()
        self.context = context.get_admin_context()
        self.project_id = uuidutils.generate_uuid()

    def _make_qos_policy(self):
        qos_policy = policy_object.QosPolicy(
            self.context, project_id=self.project_id, shared=False,
            is_default=False)
        qos_policy.create()
        return qos_policy

    def _make_qos_minbw_rule(self, policy_id, direction='ingress',
                             min_kbps=1000, rule_id=None):
        rule_id = rule_id if rule_id else uuidutils.generate_uuid()
        qos_rule = rule_object.QosMinimumBandwidthRule(
            self.context, project_id=self.project_id,
            qos_policy_id=policy_id, direction=direction, min_kbps=min_kbps,
            id=rule_id)
        qos_rule.create()
        return qos_rule

    def _make_qos_minpps_rule(self, policy_id, direction='ingress',
                              min_kpps=1000, rule_id=None):
        rule_id = rule_id if rule_id else uuidutils.generate_uuid()
        qos_rule = rule_object.QosMinimumPacketRateRule(
            self.context, project_id=self.project_id,
            qos_policy_id=policy_id, direction=direction, min_kpps=min_kpps,
            id=rule_id)
        qos_rule.create()
        return qos_rule

    def _make_port(self, network_id, qos_policy_id=None, port_id=None,
                   qos_network_policy_id=None, device_owner=None):
        port_id = port_id if port_id else uuidutils.generate_uuid()
        base_mac = ['aa', 'bb', 'cc', 'dd', 'ee', 'ff']
        mac = netaddr.EUI(next(net_utils.random_mac_generator(base_mac)))
        device_owner = device_owner if device_owner else '3'
        port = ports_object.Port(
            self.context, network_id=network_id, device_owner=device_owner,
            project_id=self.project_id, admin_state_up=True, status='DOWN',
            device_id='2', qos_policy_id=qos_policy_id,
            qos_network_policy_id=qos_network_policy_id, mac_address=mac,
            id=port_id)
        port.create()
        return port

    def _make_network(self, qos_policy_id=None):
        network = network_object.Network(self.context,
                                         qos_policy_id=qos_policy_id)
        network.create()
        return network

    def _test_validate_create_network_callback(self, network_qos=False):
        net_qos_obj = self._make_qos_policy()
        net_qos_id = net_qos_obj.id if network_qos else None
        network = self._make_network(qos_policy_id=net_qos_id)
        kwargs = {"context": self.context,
                  "network": network}

        with mock.patch.object(self.qos_plugin,
                               'validate_policy_for_network') \
                as mock_validate_policy:
            self.qos_plugin._validate_create_network_callback(
                "NETWORK", "precommit_create", "test_plugin",
                payload=events.DBEventPayload(
                    self.context, resource_id=kwargs['network']['id'],))
        qos_policy = None
        if network_qos:
            qos_policy = net_qos_obj

        if qos_policy:
            mock_validate_policy.assert_called_once_with(
                self.context, qos_policy, network.id)
        else:
            mock_validate_policy.assert_not_called()

    def test_validate_create_network_callback(self):
        self._test_validate_create_network_callback(network_qos=True)

    def test_validate_create_network_callback_no_qos(self):
        self._test_validate_create_network_callback(network_qos=False)

    def _test_validate_create_port_callback(self, port_qos=False,
                                            network_qos=False):
        net_qos_obj = self._make_qos_policy()
        port_qos_obj = self._make_qos_policy()
        net_qos_id = net_qos_obj.id if network_qos else None
        port_qos_id = port_qos_obj.id if port_qos else None
        network = self._make_network(qos_policy_id=net_qos_id)
        port = self._make_port(network.id, qos_policy_id=port_qos_id)
        kwargs = {"context": self.context,
                  "port": {"id": port.id}}

        with mock.patch.object(self.qos_plugin, 'validate_policy_for_port') \
                as mock_validate_policy:
            self.qos_plugin._validate_create_port_callback(
                "PORT", "precommit_create", "test_plugin",
                payload=events.DBEventPayload(
                    self.context,
                    resource_id=kwargs['port']['id'],))

        qos_policy = None
        if port_qos:
            qos_policy = port_qos_obj
        elif network_qos:
            qos_policy = net_qos_obj

        if qos_policy:
            mock_validate_policy.assert_called_once_with(
                self.context, qos_policy, port)
        else:
            mock_validate_policy.assert_not_called()

    def test_validate_create_port_callback_policy_on_port(self):
        self._test_validate_create_port_callback(port_qos=True)

    def test_validate_create_port_callback_policy_on_port_and_network(self):
        self._test_validate_create_port_callback(port_qos=True,
                                                 network_qos=True)

    def test_validate_create_port_callback_policy_on_network(self):
        self._test_validate_create_port_callback(network_qos=True)

    def test_validate_create_port_callback_no_policy(self):
        self._test_validate_create_port_callback()

    def _prepare_for_port_placement_allocation_change(self, qos1, qos2,
                                                      qos_network_policy=None):
        qos1_id = qos1.id if qos1 else None
        qos2_id = qos2.id if qos2 else None
        qos_network_policy_id = (
            qos_network_policy.id if qos_network_policy else None)

        network = self._make_network(qos_policy_id=qos_network_policy_id)
        port = self._make_port(
            network.id, qos_policy_id=qos1_id, port_id=TestQosPluginDB.PORT_ID)

        return {"context": self.context,
                "original_port": {
                      "id": port.id,
                      "device_owner": "compute:uu:id",
                      "qos_policy_id": qos1_id,
                      "qos_network_policy_id": qos_network_policy_id},
                "port": {"id": port.id, "qos_policy_id": qos2_id}}

    def test_check_port_for_placement_allocation_change_no_qos_change(self):
        qos1_obj = self._make_qos_policy()
        kwargs = self._prepare_for_port_placement_allocation_change(
            qos1=qos1_obj, qos2=qos1_obj)
        context = kwargs['context']
        original_port = kwargs['original_port']
        port = kwargs['port']

        with mock.patch.object(
                self.qos_plugin,
                '_change_placement_allocation') as mock_alloc_change:
            self.qos_plugin._check_port_for_placement_allocation_change(
                'PORT', 'before_update', 'test_plugin',
                payload=events.DBEventPayload(
                    context, states=(original_port, port)))
        mock_alloc_change.assert_not_called()

    def test_check_port_for_placement_allocation_change(self):
        qos1_obj = self._make_qos_policy()
        qos2_obj = self._make_qos_policy()
        kwargs = self._prepare_for_port_placement_allocation_change(
            qos1=qos1_obj, qos2=qos2_obj)
        context = kwargs['context']
        original_port = kwargs['original_port']
        port = kwargs['port']

        with mock.patch.object(
                self.qos_plugin,
                '_change_placement_allocation') as mock_alloc_change:
            self.qos_plugin._check_port_for_placement_allocation_change(
                'PORT', 'before_update', 'test_plugin',
                payload=events.DBEventPayload(
                    context, states=(original_port, port)))
        mock_alloc_change.assert_called_once_with(
            qos1_obj, qos2_obj, kwargs['original_port'], port)

    def test_check_port_for_placement_allocation_change_no_new_policy(self):
        qos1_obj = self._make_qos_policy()
        kwargs = self._prepare_for_port_placement_allocation_change(
            qos1=qos1_obj, qos2=None)
        context = kwargs['context']
        original_port = kwargs['original_port']
        port = kwargs['port']

        with mock.patch.object(
                self.qos_plugin,
                '_change_placement_allocation') as mock_alloc_change:
            self.qos_plugin._check_port_for_placement_allocation_change(
                'PORT', 'before_update', 'test_plugin',
                payload=events.DBEventPayload(
                    context, states=(original_port, port)))
        mock_alloc_change.assert_called_once_with(
            qos1_obj, None, kwargs['original_port'], port)

    def test_check_port_for_placement_allocation_change_no_qos_update(self):
        qos1_obj = self._make_qos_policy()
        kwargs = self._prepare_for_port_placement_allocation_change(
            qos1=qos1_obj, qos2=None)
        kwargs['port'].pop('qos_policy_id')
        context = kwargs['context']
        original_port = kwargs['original_port']
        port = kwargs['port']

        with mock.patch.object(
                self.qos_plugin,
                '_change_placement_allocation') as mock_alloc_change:
            self.qos_plugin._check_port_for_placement_allocation_change(
                'PORT', 'before_update', 'test_plugin',
                payload=events.DBEventPayload(
                    context, states=(original_port, port)))
        mock_alloc_change.assert_not_called()

    def test_check_port_for_placement_allocation_change_qos_network_policy(
            self):
        qos_network = self._make_qos_policy()
        desired_qos = self._make_qos_policy()
        kwargs = self._prepare_for_port_placement_allocation_change(
            qos1=None, qos2=desired_qos, qos_network_policy=qos_network)
        context = kwargs['context']
        original_port = kwargs['original_port']
        port = kwargs['port']

        with mock.patch.object(
                self.qos_plugin,
                '_change_placement_allocation') as mock_alloc_change:
            self.qos_plugin._check_port_for_placement_allocation_change(
                'PORT', 'before_update', 'test_plugin',
                payload=events.DBEventPayload(
                    context, states=(original_port, port)))
        mock_alloc_change.assert_called_once_with(
            qos_network, desired_qos, kwargs['original_port'], port)

    def test_check_network_for_placement_allocation_change_no_qos_change(self):
        qos1 = self._make_qos_policy()
        original_network = self._make_network(qos1.id)
        network = original_network
        ml2plugin_mock = mock.MagicMock()

        with mock.patch.object(
                self.qos_plugin,
                '_change_placement_allocation') as mock_alloc_change:
            self.qos_plugin._check_network_for_placement_allocation_change(
                'network', 'after_update', ml2plugin_mock,
                payload=events.DBEventPayload(
                    self.context, states=(original_network, network)))
        mock_alloc_change.assert_not_called()
        ml2plugin_mock._make_port_dict.assert_not_called()

    def test_check_network_for_placement_allocation_change_no_ports_to_update(
            self):
        original_qos = self._make_qos_policy()
        qos = self._make_qos_policy()
        port_qos = self._make_qos_policy()
        original_network = self._make_network(original_qos.id)
        network = self._make_network(qos.id)
        # Port which is not compute bound
        self._make_port(network_id=network.id, qos_policy_id=None,
                        device_owner='uu:id')
        # Port with overwritten QoS policy
        self._make_port(network_id=network.id, qos_policy_id=port_qos.id,
                        device_owner='compute:uu:id')
        ml2plugin_mock = mock.MagicMock()

        with mock.patch.object(
                self.qos_plugin,
                '_change_placement_allocation') as mock_alloc_change:
            self.qos_plugin._check_network_for_placement_allocation_change(
                'network', 'after_update', ml2plugin_mock,
                payload=events.DBEventPayload(
                    self.context, states=(original_network, network)))
        mock_alloc_change.assert_not_called()
        ml2plugin_mock._make_port_dict.assert_not_called()

    def test_check_network_for_placement_allocation_change_remove_qos(self):
        original_qos = self._make_qos_policy()
        original_network = self._make_network(original_qos.id)
        network = self._make_network()
        ml2plugin_mock = mock.MagicMock()

        def fake_make_port_dict(port):
            return {
                'id': port.id,
                'device_owner': port.device_owner,
                'qos_policy_id': port.qos_policy_id,
                'qos_network_policy_id': port.qos_network_policy_id,
            }
        ml2plugin_mock._make_port_dict.side_effect = fake_make_port_dict

        port1 = self._make_port(
            network_id=network.id, qos_policy_id=None,
            device_owner='compute:uu:id')
        port1_binding = ports_object.PortBinding(
            self.context, port_id=port1.id, host='fake_host1',
            vnic_type='fake_vnic_type', vif_type='fake_vif_type',
            profile={'allocation': 'fake_allocation'})
        port1_binding.create()
        port1.bindings = [port1_binding]
        port1.update()

        with mock.patch.object(
                self.qos_plugin,
                '_change_placement_allocation') as mock_alloc_change:
            def fake_change_placement_allocation(orig_policy, policy,
                                                 orig_port, port):
                port['binding:profile'] = {}
            mock_alloc_change.side_effect = fake_change_placement_allocation

            self.qos_plugin._check_network_for_placement_allocation_change(
                'network', 'after_update', ml2plugin_mock,
                payload=events.DBEventPayload(
                    self.context, states=(original_network, network)))

        self.assertEqual(ml2plugin_mock._make_port_dict.call_count, 1)
        mock_alloc_change_calls = [
            mock.call(
                original_qos,
                None,
                {'id': port1.id,
                    'device_owner': 'compute:uu:id',
                    'qos_policy_id': None,
                    'qos_network_policy_id': None},
                mock.ANY),
        ]
        mock_alloc_change.assert_has_calls(mock_alloc_change_calls,
                                           any_order=True)
        port1.update()
        self.assertDictEqual(port1.bindings[0].profile, {})

    def test_check_network_for_placement_allocation_change(self):
        original_qos = self._make_qos_policy()
        qos = self._make_qos_policy()
        original_network = self._make_network(original_qos.id)
        network = self._make_network(qos.id)
        ml2plugin_mock = mock.MagicMock()

        def fake_make_port_dict(port):
            return {
                'id': port.id,
                'device_owner': port.device_owner,
                'qos_policy_id': port.qos_policy_id,
                'qos_network_policy_id': port.qos_network_policy_id,
            }
        ml2plugin_mock._make_port_dict.side_effect = fake_make_port_dict

        port1 = self._make_port(
            network_id=network.id, qos_policy_id=None,
            device_owner='compute:uu:id')
        port1_binding = ports_object.PortBinding(
            self.context, port_id=port1.id, host='fake_host1',
            vnic_type='fake_vnic_type', vif_type='fake_vif_type', profile={})
        port1_binding.create()
        port1.bindings = [port1_binding]
        port1.update()

        port2 = self._make_port(
            network_id=network.id, qos_policy_id=None,
            device_owner='compute:uu:id')
        port2_binding = ports_object.PortBinding(
            self.context, port_id=port2.id, host='fake_host2',
            vnic_type='fake_vnic_type', vif_type='fake_vif_type', profile={})
        port2_binding.create()
        port2.bindings = [port2_binding]
        port2.update()

        with mock.patch.object(
                self.qos_plugin,
                '_change_placement_allocation') as mock_alloc_change:
            def fake_change_placement_allocation(orig_policy, policy,
                                                 orig_port, port):
                port['binding:profile'] = {'allocation': 'fake_allocation'}
            mock_alloc_change.side_effect = fake_change_placement_allocation

            self.qos_plugin._check_network_for_placement_allocation_change(
                'network', 'after_update', ml2plugin_mock,
                payload=events.DBEventPayload(
                    self.context, states=(original_network, network)))

        self.assertEqual(ml2plugin_mock._make_port_dict.call_count, 2)

        mock_alloc_change_calls = [
            mock.call(
                original_qos,
                qos,
                {'id': port1.id,
                    'device_owner': 'compute:uu:id',
                    'qos_policy_id': None,
                    'qos_network_policy_id': qos.id},
                mock.ANY),
            mock.call(
                original_qos,
                qos,
                {'id': port2.id,
                    'device_owner': 'compute:uu:id',
                    'qos_policy_id': None,
                    'qos_network_policy_id': qos.id},
                mock.ANY)]
        mock_alloc_change.assert_has_calls(mock_alloc_change_calls,
                                           any_order=True)
        port1.update()
        port2.update()
        self.assertDictEqual(
            port1.bindings[0].profile, {'allocation': 'fake_allocation'})
        self.assertDictEqual(
            port2.bindings[0].profile, {'allocation': 'fake_allocation'})

    def _prepare_port_for_placement_allocation(self, original_qos,
                                               desired_qos=None,
                                               qos_network_policy=None,
                                               original_min_kbps=None,
                                               desired_min_kbps=None,
                                               original_min_kpps=None,
                                               desired_min_kpps=None,
                                               is_sriov=False):
        kwargs = self._prepare_for_port_placement_allocation_change(
            original_qos, desired_qos, qos_network_policy=qos_network_policy)
        orig_port = kwargs['original_port']
        qos = original_qos or qos_network_policy
        qos.rules = []
        allocation = {}
        if original_min_kbps:
            qos.rules += [self._make_qos_minbw_rule(
                qos.id, min_kbps=original_min_kbps,
                rule_id=TestQosPluginDB.QOS_MIN_BW_RULE_ID)]
            allocation.update(
                {TestQosPluginDB.MIN_BW_REQUEST_GROUP_UUID:
                    TestQosPluginDB.MIN_BW_RP})
        if original_min_kpps:
            qos.rules += [self._make_qos_minpps_rule(
                qos.id, min_kpps=original_min_kpps,
                rule_id=TestQosPluginDB.QOS_MIN_PPS_RULE_ID)]
            allocation.update(
                {TestQosPluginDB.MIN_PPS_REQUEST_GROUP_UUID:
                    TestQosPluginDB.MIN_PPS_RP})
        if desired_qos:
            desired_qos.rules = []
            if desired_min_kbps:
                desired_qos.rules += [self._make_qos_minbw_rule(
                    desired_qos.id, min_kbps=desired_min_kbps)]
            if desired_min_kpps:
                desired_qos.rules += [self._make_qos_minpps_rule(
                    desired_qos.id, min_kpps=desired_min_kpps)]

        binding_prof = {}
        if is_sriov:
            binding_prof = {
                'pci_slot': '0000:42:41.0',
                'pci_vendor_info': '8086:107ed',
                'physical_network': 'sriov_phy'
            }

        binding_prof.update({'allocation': allocation})
        orig_port.update(
            {'binding:profile': binding_prof,
             'device_id': 'uu:id'}
        )
        return orig_port, kwargs['port']

    def _assert_pci_info(self, port):
        self.assertIn('pci_slot', port['binding:profile'])
        self.assertIn('pci_vendor_info', port['binding:profile'])
        self.assertIn('physical_network', port['binding:profile'])

    def test_change_placement_allocation_increase(self):
        qos1 = self._make_qos_policy()
        qos2 = self._make_qos_policy()
        orig_port, port = self._prepare_port_for_placement_allocation(
            qos1, qos2, original_min_kbps=1000, desired_min_kbps=2000,
            is_sriov=True)
        with mock.patch.object(self.qos_plugin._placement_client,
                'update_qos_allocation') as mock_update_qos_alloc:
            self.qos_plugin._change_placement_allocation(
                qos1, qos2, orig_port, port)
        mock_update_qos_alloc.assert_called_once_with(
            consumer_uuid='uu:id',
            alloc_diff={self.MIN_BW_RP: {'NET_BW_IGR_KILOBIT_PER_SEC': 1000}})
        self._assert_pci_info(port)

    def test_change_placement_allocation_increase_min_pps(self):
        qos1 = self._make_qos_policy()
        qos2 = self._make_qos_policy()
        orig_port, port = self._prepare_port_for_placement_allocation(
            qos1, qos2, original_min_kpps=1000, desired_min_kpps=2000,
            is_sriov=True)
        with mock.patch.object(self.qos_plugin._placement_client,
                'update_qos_allocation') as mock_update_qos_alloc:
            self.qos_plugin._change_placement_allocation(
                qos1, qos2, orig_port, port)
        mock_update_qos_alloc.assert_called_once_with(
            consumer_uuid='uu:id',
            alloc_diff={self.MIN_PPS_RP: {
                'NET_PACKET_RATE_IGR_KILOPACKET_PER_SEC': 1000}})
        self._assert_pci_info(port)

    def test_change_placement_allocation_increase_min_pps_and_min_bw(self):
        qos1 = self._make_qos_policy()
        qos2 = self._make_qos_policy()
        orig_port, port = self._prepare_port_for_placement_allocation(
            qos1, qos2, original_min_kbps=1000, desired_min_kbps=2000,
            original_min_kpps=500, desired_min_kpps=1000)
        with mock.patch.object(self.qos_plugin._placement_client,
                'update_qos_allocation') as mock_update_qos_alloc:
            self.qos_plugin._change_placement_allocation(
                qos1, qos2, orig_port, port)
        mock_update_qos_alloc.assert_called_once_with(
            consumer_uuid='uu:id',
            alloc_diff={
                self.MIN_PPS_RP: {
                    'NET_PACKET_RATE_IGR_KILOPACKET_PER_SEC': 500},
                self.MIN_BW_RP: {'NET_BW_IGR_KILOBIT_PER_SEC': 1000}})

    def test_change_placement_allocation_change_direction_min_pps_and_min_bw(
            self):
        qos1 = self._make_qos_policy()
        qos2 = self._make_qos_policy()
        orig_port, port = self._prepare_port_for_placement_allocation(
            qos1, qos2, original_min_kbps=1000, desired_min_kbps=2000,
            original_min_kpps=500, desired_min_kpps=1000)
        for rule in qos2.rules:
            rule.direction = 'egress'
        with mock.patch.object(self.qos_plugin._placement_client,
                'update_qos_allocation') as mock_update_qos_alloc:
            self.qos_plugin._change_placement_allocation(
                qos1, qos2, orig_port, port)
        mock_update_qos_alloc.assert_called_once_with(
            consumer_uuid='uu:id',
            alloc_diff={
                self.MIN_PPS_RP: {
                    'NET_PACKET_RATE_IGR_KILOPACKET_PER_SEC': -500,
                    'NET_PACKET_RATE_EGR_KILOPACKET_PER_SEC': 1000},
                self.MIN_BW_RP: {
                    'NET_BW_IGR_KILOBIT_PER_SEC': -1000,
                    'NET_BW_EGR_KILOBIT_PER_SEC': 2000}})

    def test_change_placement_allocation_change_dir_min_pps_ingress_to_any(
            self):
        qos1 = self._make_qos_policy()
        qos2 = self._make_qos_policy()
        orig_port, port = self._prepare_port_for_placement_allocation(
            qos1, qos2, original_min_kpps=1000, desired_min_kpps=1000)
        for rule in qos2.rules:
            rule.direction = 'any'
        with mock.patch.object(self.qos_plugin._placement_client,
                'update_qos_allocation') as mock_update_qos_alloc:
            self.assertRaises(NotImplementedError,
                self.qos_plugin._change_placement_allocation, qos1, qos2,
                orig_port, port)
        mock_update_qos_alloc.assert_not_called()

    def test_change_placement_allocation_min_bw_dataplane_enforcement(self):
        qos1 = self._make_qos_policy()
        qos2 = self._make_qos_policy()
        orig_port, port = self._prepare_port_for_placement_allocation(
            qos1, qos2, desired_min_kbps=1000)
        with mock.patch.object(self.qos_plugin._placement_client,
                'update_qos_allocation') as mock_update_qos_alloc:
            self.qos_plugin._change_placement_allocation(qos1, qos2, orig_port,
                port)
        mock_update_qos_alloc.assert_not_called()

    def test_change_placement_allocation_min_bw_dataplane_enforcement_with_pps(
            self):
        qos1 = self._make_qos_policy()
        qos2 = self._make_qos_policy()
        orig_port, port = self._prepare_port_for_placement_allocation(
            qos1, qos2, desired_min_kbps=1000, original_min_kpps=500,
            desired_min_kpps=1000)
        with mock.patch.object(self.qos_plugin._placement_client,
                'update_qos_allocation') as mock_update_qos_alloc:
            self.qos_plugin._change_placement_allocation(qos1, qos2, orig_port,
                port)
        mock_update_qos_alloc.assert_called_once_with(
            consumer_uuid='uu:id',
            alloc_diff={
                self.MIN_PPS_RP: {
                    'NET_PACKET_RATE_IGR_KILOPACKET_PER_SEC': 500}})

    def test_change_placement_allocation_decrease(self):
        original_qos = self._make_qos_policy()
        desired_qos = self._make_qos_policy()
        orig_port, port = self._prepare_port_for_placement_allocation(
            original_qos, desired_qos, original_min_kbps=2000,
            desired_min_kbps=1000, is_sriov=True)
        with mock.patch.object(self.qos_plugin._placement_client,
                'update_qos_allocation') as mock_update_qos_alloc:
            self.qos_plugin._change_placement_allocation(
                original_qos, desired_qos, orig_port, port)
        mock_update_qos_alloc.assert_called_once_with(
            consumer_uuid='uu:id',
            alloc_diff={self.MIN_BW_RP: {'NET_BW_IGR_KILOBIT_PER_SEC': -1000}})
        self._assert_pci_info(port)

    def test_change_placement_allocation_decrease_min_pps(self):
        original_qos = self._make_qos_policy()
        desired_qos = self._make_qos_policy()
        orig_port, port = self._prepare_port_for_placement_allocation(
            original_qos, desired_qos, original_min_kpps=2000,
            desired_min_kpps=1000, is_sriov=True)
        with mock.patch.object(self.qos_plugin._placement_client,
                'update_qos_allocation') as mock_update_qos_alloc:
            self.qos_plugin._change_placement_allocation(
                original_qos, desired_qos, orig_port, port)
        mock_update_qos_alloc.assert_called_once_with(
            consumer_uuid='uu:id',
            alloc_diff={self.MIN_PPS_RP: {
                'NET_PACKET_RATE_IGR_KILOPACKET_PER_SEC': -1000}})
        self._assert_pci_info(port)

    def test_change_placement_allocation_no_original_qos(self):
        qos1 = None
        qos2 = self._make_qos_policy()
        rule2_obj = self._make_qos_minbw_rule(qos2.id, min_kbps=1000)
        qos2.rules = [rule2_obj]
        orig_port = {'id': 'u:u', 'device_id': 'i:d', 'binding:profile': {}}
        port = {}
        with mock.patch.object(self.qos_plugin._placement_client,
                'update_qos_allocation') as mock_update_qos_alloc:
            self.qos_plugin._change_placement_allocation(qos1, qos2, orig_port,
                port)
        mock_update_qos_alloc.assert_not_called()

    def test_change_placement_allocation_no_original_allocation(self):
        qos1 = self._make_qos_policy()
        rule1_obj = self._make_qos_minbw_rule(qos1.id, min_kbps=500)
        qos1.rules = [rule1_obj]
        qos2 = self._make_qos_policy()
        rule2_obj = self._make_qos_minbw_rule(qos2.id, min_kbps=1000)
        qos2.rules = [rule2_obj]
        orig_port = {'id': 'u:u', 'device_id': 'i:d', 'binding:profile': {}}
        port = {}
        with mock.patch.object(self.qos_plugin._placement_client,
                'update_qos_allocation') as mock_update_qos_alloc:
            self.qos_plugin._change_placement_allocation(qos1, qos2, orig_port,
                port)
        mock_update_qos_alloc.assert_not_called()

    def test_change_placement_allocation_new_policy_empty(self):
        qos1 = self._make_qos_policy()
        orig_port, port = self._prepare_port_for_placement_allocation(qos1,
            original_min_kbps=1000, original_min_kpps=2000)
        with mock.patch.object(self.qos_plugin._placement_client,
                'update_qos_allocation') as mock_update_qos_alloc:
            self.qos_plugin._change_placement_allocation(
                qos1, None, orig_port, port)
        mock_update_qos_alloc.assert_called_once_with(
            consumer_uuid='uu:id',
            alloc_diff={
                self.MIN_BW_RP: {'NET_BW_IGR_KILOBIT_PER_SEC': -1000},
                self.MIN_PPS_RP: {
                    'NET_PACKET_RATE_IGR_KILOPACKET_PER_SEC': -2000}})

    def test_change_placement_allocation_no_min_bw(self):
        qos1 = self._make_qos_policy()
        qos2 = self._make_qos_policy()
        bw_limit_rule1 = rule_object.QosDscpMarkingRule(dscp_mark=16)
        bw_limit_rule2 = rule_object.QosDscpMarkingRule(dscp_mark=18)
        qos1.rules = [bw_limit_rule1]
        qos2.rules = [bw_limit_rule2]
        orig_port = {
            'binding:profile': {'allocation': {
                self.MIN_BW_REQUEST_GROUP_UUID: self.MIN_BW_RP}},
            'device_id': 'uu:id',
            'id': '9416c220-160a-11ec-ba3d-474633eb825c',
        }
        port = {}

        with mock.patch.object(self.qos_plugin._placement_client,
                'update_qos_allocation') as mock_update_qos_alloc:
            self.qos_plugin._change_placement_allocation(
                qos1, None, orig_port, port)
        mock_update_qos_alloc.assert_not_called()

    def test_change_placement_allocation_old_rule_not_min_bw(self):
        qos1 = self._make_qos_policy()
        qos2 = self._make_qos_policy()
        bw_limit_rule = rule_object.QosDscpMarkingRule(dscp_mark=16)
        orig_port, port = self._prepare_port_for_placement_allocation(
            qos1, qos2, desired_min_kbps=2000)
        qos1.rules = [bw_limit_rule]

        with mock.patch.object(self.qos_plugin._placement_client,
                'update_qos_allocation') as mock_update_qos_alloc:
            self.qos_plugin._change_placement_allocation(qos1, qos2, orig_port,
                port)
        mock_update_qos_alloc.assert_not_called()

    def test_change_placement_allocation_new_rule_not_min_bw(self):
        qos1 = self._make_qos_policy()
        qos2 = self._make_qos_policy()
        bw_limit_rule = rule_object.QosDscpMarkingRule(dscp_mark=16)
        qos2.rules = [bw_limit_rule]
        orig_port, port = self._prepare_port_for_placement_allocation(qos1,
            original_min_kbps=1000)

        with mock.patch.object(self.qos_plugin._placement_client,
                'update_qos_allocation') as mock_update_qos_alloc:
            self.qos_plugin._change_placement_allocation(
                qos1, qos2, orig_port, port)
        mock_update_qos_alloc.assert_called_once_with(
            consumer_uuid='uu:id',
            alloc_diff={self.MIN_BW_RP: {'NET_BW_IGR_KILOBIT_PER_SEC': -1000}})

    def test_change_placement_allocation_equal_minkbps_and_minkpps(self):
        qos1 = self._make_qos_policy()
        qos2 = self._make_qos_policy()
        orig_port, port = self._prepare_port_for_placement_allocation(
            qos1, qos2, original_min_kbps=1000, desired_min_kbps=1000,
            original_min_kpps=1000, desired_min_kpps=1000)
        with mock.patch.object(self.qos_plugin._placement_client,
                'update_qos_allocation') as mock_update_qos_alloc:
            self.qos_plugin._change_placement_allocation(
                qos1, qos2, orig_port, port)
        mock_update_qos_alloc.assert_not_called()

    def test_change_placement_allocation_update_conflict(self):
        qos1 = self._make_qos_policy()
        qos2 = self._make_qos_policy()
        orig_port, port = self._prepare_port_for_placement_allocation(
            qos1, qos2, original_min_kbps=1000, desired_min_kbps=2000)
        with mock.patch.object(self.qos_plugin._placement_client,
                'update_qos_allocation') as mock_update_qos_alloc:
            mock_update_qos_alloc.side_effect = ks_exc.Conflict(
                response={'errors': [{'code': 'placement.concurrent_update'}]}
            )
            self.assertRaises(
                neutron_qos_exc.QosPlacementAllocationUpdateConflict,
                self.qos_plugin._change_placement_allocation,
                qos1, qos2, orig_port, port)

    def test_change_placement_allocation_update_generation_conflict(self):
        qos1 = self._make_qos_policy()
        qos2 = self._make_qos_policy()
        orig_port, port = self._prepare_port_for_placement_allocation(
            qos1, qos2, original_min_kbps=1000, desired_min_kbps=2000)
        with mock.patch.object(self.qos_plugin._placement_client,
                'update_qos_allocation') as mock_update_qos_alloc:
            mock_update_qos_alloc.side_effect = (
                pl_exc.PlacementAllocationGenerationConflict(
                    consumer=self.MIN_BW_RP))
            self.assertRaises(
                pl_exc.PlacementAllocationGenerationConflict,
                self.qos_plugin._change_placement_allocation,
                qos1, qos2, orig_port, port)

    def test_change_placement_allocation_qos_network_policy(self):
        qos_network = self._make_qos_policy()
        desired_qos = self._make_qos_policy()
        orig_port, port = self._prepare_port_for_placement_allocation(
            None, desired_qos, qos_network_policy=qos_network,
            original_min_kbps=1000, desired_min_kbps=2000)
        with mock.patch.object(self.qos_plugin._placement_client,
                'update_qos_allocation') as mock_update_qos_alloc:
            self.qos_plugin._change_placement_allocation(
                qos_network, desired_qos, orig_port, port)
        mock_update_qos_alloc.assert_called_once_with(
            consumer_uuid='uu:id',
            alloc_diff={self.MIN_BW_RP: {'NET_BW_IGR_KILOBIT_PER_SEC': 1000}})

[((745, 5, 747, 30), 'unittest.mock.patch', 'mock.patch', ({(746, 8, 747, 29): '"""neutron.objects.rbac_db.RbacNeutronDbObjectMixin.create_rbac_policy"""'}, {}), "(\n    'neutron.objects.rbac_db.RbacNeutronDbObjectMixin.create_rbac_policy')", False, 'from unittest import mock\n'), ((748, 5, 748, 55), 'unittest.mock.patch', 'mock.patch', ({(748, 16, 748, 54): '"""neutron.objects.qos.policy.QosPolicy"""'}, {}), "('neutron.objects.qos.policy.QosPolicy')", False, 'from unittest import mock\n'), ((790, 5, 790, 61), 'unittest.mock.patch.object', 'mock.patch.object', ({(790, 23, 790, 46): 'policy_object.QosPolicy', (790, 48, 790, 60): '"""get_object"""'}, {}), "(policy_object.QosPolicy, 'get_object')", False, 'from unittest import mock\n'), ((791, 5, 793, 30), 'unittest.mock.patch', 'mock.patch', ({(792, 8, 793, 29): '"""neutron.objects.rbac_db.RbacNeutronDbObjectMixin.create_rbac_policy"""'}, {}), "(\n    'neutron.objects.rbac_db.RbacNeutronDbObjectMixin.create_rbac_policy')", False, 'from unittest import mock\n'), ((794, 5, 794, 57), 'unittest.mock.patch.object', 'mock.patch.object', ({(794, 23, 794, 46): 'policy_object.QosPolicy', (794, 48, 794, 56): '"""update"""'}, {}), "(policy_object.QosPolicy, 'update')", False, 'from unittest import mock\n'), ((819, 5, 819, 71), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((820, 5, 820, 57), 'unittest.mock.patch.object', 'mock.patch.object', ({(820, 23, 820, 46): 'policy_object.QosPolicy', (820, 48, 820, 56): '"""delete"""'}, {}), "(policy_object.QosPolicy, 'delete')", False, 'from unittest import mock\n'), ((840, 5, 840, 61), 'unittest.mock.patch.object', 'mock.patch.object', ({(840, 23, 840, 46): 'policy_object.QosPolicy', (840, 48, 840, 60): '"""get_object"""'}, {}), "(policy_object.QosPolicy, 'get_object')", False, 'from unittest import mock\n'), ((841, 5, 841, 67), 'unittest.mock.patch.object', 'mock.patch.object', ({(841, 23, 841, 56): 'rule_object.QosBandwidthLimitRule', (841, 58, 841, 66): '"""create"""'}, {}), "(rule_object.QosBandwidthLimitRule, 'create')", False, 'from unittest import mock\n'), ((931, 5, 931, 67), 'unittest.mock.patch.object', 'mock.patch.object', ({(931, 23, 931, 56): 'rule_object.QosBandwidthLimitRule', (931, 58, 931, 66): '"""update"""'}, {}), "(rule_object.QosBandwidthLimitRule, 'update')", False, 'from unittest import mock\n'), ((1034, 5, 1034, 67), 'unittest.mock.patch.object', 'mock.patch.object', ({(1034, 23, 1034, 56): 'rule_object.QosBandwidthLimitRule', (1034, 58, 1034, 66): '"""delete"""'}, {}), "(rule_object.QosBandwidthLimitRule, 'delete')", False, 'from unittest import mock\n'), ((1320, 5, 1320, 45), 'unittest.mock.patch', 'mock.patch', ({(1320, 16, 1320, 44): '"""neutron.objects.ports.Port"""'}, {}), "('neutron.objects.ports.Port')", False, 'from unittest import mock\n'), ((1321, 5, 1321, 55), 'unittest.mock.patch', 'mock.patch', ({(1321, 16, 1321, 54): '"""neutron.objects.qos.policy.QosPolicy"""'}, {}), "('neutron.objects.qos.policy.QosPolicy')", False, 'from unittest import mock\n'), ((1929, 5, 1929, 66), 'unittest.mock.patch.object', 'mock.patch.object', ({(1929, 23, 1929, 43): 'qos_plugin.QoSPlugin', (1929, 45, 1929, 65): '"""update_policy_rule"""'}, {}), "(qos_plugin.QoSPlugin, 'update_policy_rule')", False, 'from unittest import mock\n'), ((1949, 5, 1949, 63), 'unittest.mock.patch.object', 'mock.patch.object', ({(1949, 23, 1949, 43): 'qos_plugin.QoSPlugin', (1949, 45, 1949, 62): '"""get_policy_rule"""'}, {}), "(qos_plugin.QoSPlugin, 'get_policy_rule')", False, 'from unittest import mock\n'), ((1966, 5, 1966, 66), 'unittest.mock.patch.object', 'mock.patch.object', ({(1966, 23, 1966, 43): 'qos_plugin.QoSPlugin', (1966, 45, 1966, 65): '"""delete_policy_rule"""'}, {}), "(qos_plugin.QoSPlugin, 'delete_policy_rule')", False, 'from unittest import mock\n'), ((62, 30, 63, 65), 'unittest.mock.patch', 'mock.patch', ({(63, 12, 63, 64): '"""neutron.objects.qos.policy.QosPolicy.obj_load_attr"""'}, {}), "('neutron.objects.qos.policy.QosPolicy.obj_load_attr')", False, 'from unittest import mock\n'), ((73, 8, 73, 61), 'oslo_config.cfg.CONF.set_override', 'cfg.CONF.set_override', ({(73, 30, 73, 43): '"""core_plugin"""', (73, 45, 73, 60): 'DB_PLUGIN_KLASS'}, {}), "('core_plugin', DB_PLUGIN_KLASS)", False, 'from oslo_config import cfg\n'), ((74, 8, 74, 57), 'oslo_config.cfg.CONF.set_override', 'cfg.CONF.set_override', ({(74, 30, 74, 47): '"""service_plugins"""', (74, 49, 74, 56): "['qos']"}, {}), "('service_plugins', ['qos'])", False, 'from oslo_config import cfg\n'), ((76, 8, 76, 22), 'neutron.manager.init', 'manager.init', ({}, {}), '()', False, 'from neutron import manager\n'), ((77, 26, 77, 69), 'neutron_lib.plugins.directory.get_plugin', 'directory.get_plugin', ({(77, 47, 77, 68): 'plugins_constants.QOS'}, {}), '(plugins_constants.QOS)', False, 'from neutron_lib.plugins import directory\n'), ((79, 41, 79, 52), 'unittest.mock.Mock', 'mock.Mock', ({}, {}), '()', False, 'from unittest import mock\n'), ((84, 20, 84, 63), 'neutron_lib.context.Context', 'context.Context', ({(84, 36, 84, 47): '"""fake_user"""', (84, 49, 84, 62): '"""fake_tenant"""'}, {}), "('fake_user', 'fake_tenant')", False, 'from neutron_lib import context\n'), ((85, 26, 85, 53), 'neutron_lib.context.get_admin_context', 'context.get_admin_context', ({}, {}), '()', False, 'from neutron_lib import context\n'), ((114, 22, 115, 52), 'neutron.objects.qos.policy.QosPolicy', 'policy_object.QosPolicy', ({(115, 12, 115, 21): 'self.ctxt'}, {}), "(self.ctxt, **self.policy_data['policy'])", True, 'from neutron.objects.qos import policy as policy_object\n'), ((117, 20, 118, 64), 'neutron.objects.qos.rule.QosBandwidthLimitRule', 'rule_object.QosBandwidthLimitRule', ({(118, 12, 118, 21): 'self.ctxt'}, {}), "(self.ctxt, **self.rule_data[\n    'bandwidth_limit_rule'])", True, 'from neutron.objects.qos import rule as rule_object\n'), ((120, 25, 121, 61), 'neutron.objects.qos.rule.QosDscpMarkingRule', 'rule_object.QosDscpMarkingRule', ({(121, 12, 121, 21): 'self.ctxt'}, {}), "(self.ctxt, **self.rule_data['dscp_marking_rule']\n    )", True, 'from neutron.objects.qos import rule as rule_object\n'), ((123, 27, 124, 66), 'neutron.objects.qos.rule.QosMinimumBandwidthRule', 'rule_object.QosMinimumBandwidthRule', ({(124, 12, 124, 21): 'self.ctxt'}, {}), "(self.ctxt, **self.rule_data[\n    'minimum_bandwidth_rule'])", True, 'from neutron.objects.qos import rule as rule_object\n'), ((126, 24, 127, 66), 'neutron.objects.qos.rule.QosPacketRateLimitRule', 'rule_object.QosPacketRateLimitRule', ({(127, 12, 127, 21): 'self.ctxt'}, {}), "(self.ctxt, **self.rule_data[\n    'packet_rate_limit_rule'])", True, 'from neutron.objects.qos import rule as rule_object\n'), ((129, 28, 130, 68), 'neutron.objects.qos.rule.QosMinimumPacketRateRule', 'rule_object.QosMinimumPacketRateRule', ({(130, 12, 130, 21): 'self.ctxt'}, {}), "(self.ctxt, **self.rule_data[\n    'minimum_packet_rate_rule'])", True, 'from neutron.objects.qos import rule as rule_object\n'), ((143, 21, 143, 46), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ({}, {}), '()', False, 'from oslo_utils import uuidutils\n'), ((155, 20, 156, 48), 'neutron.objects.ports.Port', 'ports_object.Port', ({(156, 12, 156, 21): 'self.ctxt'}, {}), "(self.ctxt, **self.port_data['port'])", True, 'from neutron.objects import ports as ports_object\n'), ((159, 23, 160, 72), 'unittest.mock.MagicMock', 'mock.MagicMock', (), '', False, 'from unittest import mock\n'), ((183, 21, 183, 46), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ({}, {}), '()', False, 'from oslo_utils import uuidutils\n'), ((204, 23, 205, 72), 'unittest.mock.MagicMock', 'mock.MagicMock', (), '', False, 'from unittest import mock\n'), ((286, 30, 287, 50), 'neutron.objects.qos.rule.QosMinimumBandwidthRule', 'rule_object.QosMinimumBandwidthRule', ({(287, 12, 287, 21): 'self.ctxt'}, {}), '(self.ctxt, **min_bw_rule_ingress_data)', True, 'from neutron.objects.qos import rule as rule_object\n'), ((288, 31, 289, 51), 'neutron.objects.qos.rule.QosMinimumPacketRateRule', 'rule_object.QosMinimumPacketRateRule', ({(289, 12, 289, 21): 'self.ctxt'}, {}), '(self.ctxt, **min_pps_rule_ingress_data)', True, 'from neutron.objects.qos import rule as rule_object\n'), ((428, 30, 429, 50), 'neutron.objects.qos.rule.QosMinimumBandwidthRule', 'rule_object.QosMinimumBandwidthRule', ({(429, 12, 429, 21): 'self.ctxt'}, {}), '(self.ctxt, **min_bw_rule_ingress_data)', True, 'from neutron.objects.qos import rule as rule_object\n'), ((430, 31, 431, 51), 'neutron.objects.qos.rule.QosMinimumPacketRateRule', 'rule_object.QosMinimumPacketRateRule', ({(431, 12, 431, 21): 'self.ctxt'}, {}), '(self.ctxt, **min_pps_rule_ingress_data)', True, 'from neutron.objects.qos import rule as rule_object\n'), ((531, 18, 531, 43), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ({}, {}), '()', False, 'from oslo_utils import uuidutils\n'), ((542, 20, 542, 71), 'unittest.mock.MagicMock', 'mock.MagicMock', (), '', False, 'from unittest import mock\n'), ((543, 22, 543, 50), 'unittest.mock.MagicMock', 'mock.MagicMock', (), '', False, 'from unittest import mock\n'), ((544, 21, 544, 32), 'unittest.mock.Mock', 'mock.Mock', ({}, {}), '()', False, 'from unittest import mock\n'), ((576, 20, 576, 45), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ({}, {}), '()', False, 'from oslo_utils import uuidutils\n'), ((586, 21, 586, 46), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ({}, {}), '()', False, 'from oslo_utils import uuidutils\n'), ((604, 22, 604, 50), 'unittest.mock.MagicMock', 'mock.MagicMock', (), '', False, 'from unittest import mock\n'), ((605, 21, 605, 32), 'unittest.mock.Mock', 'mock.Mock', ({}, {}), '()', False, 'from unittest import mock\n'), ((641, 20, 641, 45), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ({}, {}), '()', False, 'from oslo_utils import uuidutils\n'), ((675, 18, 675, 43), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ({}, {}), '()', False, 'from oslo_utils import uuidutils\n'), ((691, 18, 692, 44), 'neutron.objects.network.NetworkSegment', 'network_object.NetworkSegment', (), '', True, 'from neutron.objects import network as network_object\n'), ((693, 14, 695, 31), 'neutron.objects.network.Network', 'network_object.Network', (), '', True, 'from neutron.objects import network as network_object\n'), ((719, 18, 720, 44), 'neutron.objects.network.NetworkSegment', 'network_object.NetworkSegment', (), '', True, 'from neutron.objects import network as network_object\n'), ((721, 14, 723, 31), 'neutron.objects.network.Network', 'network_object.Network', (), '', True, 'from neutron.objects import network as network_object\n'), ((750, 23, 750, 34), 'unittest.mock.Mock', 'mock.Mock', ({}, {}), '()', False, 'from unittest import mock\n'), ((757, 37, 758, 59), 'unittest.mock.call.driver.call', 'mock.call.driver.call', ({(758, 12, 758, 37): '"""create_policy_precommit"""', (758, 39, 758, 48): 'self.ctxt', (758, 50, 758, 58): 'mock.ANY'}, {}), "('create_policy_precommit', self.ctxt, mock.ANY)", False, 'from unittest import mock\n'), ((759, 27, 760, 49), 'unittest.mock.call.driver.call', 'mock.call.driver.call', ({(760, 12, 760, 27): '"""create_policy"""', (760, 29, 760, 38): 'self.ctxt', (760, 40, 760, 48): 'mock.ANY'}, {}), "('create_policy', self.ctxt, mock.ANY)", False, 'from unittest import mock\n'), ((767, 20, 767, 45), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ({}, {}), '()', False, 'from oslo_utils import uuidutils\n'), ((768, 21, 768, 46), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ({}, {}), '()', False, 'from oslo_utils import uuidutils\n'), ((798, 23, 798, 34), 'unittest.mock.Mock', 'mock.Mock', ({}, {}), '()', False, 'from unittest import mock\n'), ((803, 17, 804, 64), 'neutron_lib.objects.utils.get_updatable_fields', 'obj_utils.get_updatable_fields', ({(804, 12, 804, 35): 'policy_object.QosPolicy', (804, 37, 804, 63): "self.policy_data['policy']"}, {}), "(policy_object.QosPolicy, self.policy_data[\n    'policy'])", True, 'from neutron_lib.objects import utils as obj_utils\n'), ((809, 34, 809, 52), 'unittest.mock.call.update', 'mock.call.update', ({}, {}), '()', False, 'from unittest import mock\n'), ((810, 37, 811, 59), 'unittest.mock.call.driver.call', 'mock.call.driver.call', ({(811, 12, 811, 37): '"""update_policy_precommit"""', (811, 39, 811, 48): 'self.ctxt', (811, 50, 811, 58): 'mock.ANY'}, {}), "('update_policy_precommit', self.ctxt, mock.ANY)", False, 'from unittest import mock\n'), ((812, 27, 813, 49), 'unittest.mock.call.driver.call', 'mock.call.driver.call', ({(813, 12, 813, 27): '"""update_policy"""', (813, 29, 813, 38): 'self.ctxt', (813, 40, 813, 48): 'mock.ANY'}, {}), "('update_policy', self.ctxt, mock.ANY)", False, 'from unittest import mock\n'), ((822, 23, 822, 34), 'unittest.mock.Mock', 'mock.Mock', ({}, {}), '()', False, 'from unittest import mock\n'), ((830, 34, 830, 52), 'unittest.mock.call.delete', 'mock.call.delete', ({}, {}), '()', False, 'from unittest import mock\n'), ((831, 37, 832, 59), 'unittest.mock.call.driver.call', 'mock.call.driver.call', ({(832, 12, 832, 37): '"""delete_policy_precommit"""', (832, 39, 832, 48): 'self.ctxt', (832, 50, 832, 58): 'mock.ANY'}, {}), "('delete_policy_precommit', self.ctxt, mock.ANY)", False, 'from unittest import mock\n'), ((833, 27, 834, 49), 'unittest.mock.call.driver.call', 'mock.call.driver.call', ({(834, 12, 834, 27): '"""delete_policy"""', (834, 29, 834, 38): 'self.ctxt', (834, 40, 834, 48): 'mock.ANY'}, {}), "('delete_policy', self.ctxt, mock.ANY)", False, 'from unittest import mock\n'), ((844, 18, 844, 40), 'copy.copy', 'copy.copy', ({(844, 28, 844, 39): 'self.policy'}, {}), '(self.policy)', False, 'import copy\n'), ((847, 23, 847, 34), 'unittest.mock.Mock', 'mock.Mock', ({}, {}), '()', False, 'from unittest import mock\n'), ((933, 23, 933, 34), 'unittest.mock.Mock', 'mock.Mock', ({}, {}), '()', False, 'from unittest import mock\n'), ((938, 18, 939, 52), 'neutron.objects.qos.policy.QosPolicy', 'policy_object.QosPolicy', ({(939, 12, 939, 21): 'self.ctxt'}, {}), "(self.ctxt, **self.policy_data['policy'])", True, 'from neutron.objects.qos import policy as policy_object\n'), ((1019, 15, 1020, 52), 'neutron.objects.qos.policy.QosPolicy', 'policy_object.QosPolicy', ({(1020, 12, 1020, 21): 'self.ctxt'}, {}), "(self.ctxt, **self.policy_data['policy'])", True, 'from neutron.objects.qos import policy as policy_object\n'), ((1023, 18, 1024, 52), 'neutron.objects.qos.policy.QosPolicy', 'policy_object.QosPolicy', ({(1024, 12, 1024, 21): 'self.ctxt'}, {}), "(self.ctxt, **self.policy_data['policy'])", True, 'from neutron.objects.qos import policy as policy_object\n'), ((1036, 23, 1036, 34), 'unittest.mock.Mock', 'mock.Mock', ({}, {}), '()', False, 'from unittest import mock\n'), ((1041, 18, 1042, 52), 'neutron.objects.qos.policy.QosPolicy', 'policy_object.QosPolicy', ({(1042, 12, 1042, 21): 'self.ctxt'}, {}), "(self.ctxt, **self.policy_data['policy'])", True, 'from neutron.objects.qos import policy as policy_object\n'), ((1061, 18, 1062, 52), 'neutron.objects.qos.policy.QosPolicy', 'policy_object.QosPolicy', ({(1062, 12, 1062, 21): 'self.ctxt'}, {}), "(self.ctxt, **self.policy_data['policy'])", True, 'from neutron.objects.qos import policy as policy_object\n'), ((1133, 18, 1134, 52), 'neutron.objects.qos.policy.QosPolicy', 'policy_object.QosPolicy', ({(1134, 12, 1134, 21): 'self.ctxt'}, {}), "(self.ctxt, **self.policy_data['policy'])", True, 'from neutron.objects.qos import policy as policy_object\n'), ((1143, 18, 1144, 52), 'neutron.objects.qos.policy.QosPolicy', 'policy_object.QosPolicy', ({(1144, 12, 1144, 21): 'self.ctxt'}, {}), "(self.ctxt, **self.policy_data['policy'])", True, 'from neutron.objects.qos import policy as policy_object\n'), ((1153, 18, 1154, 52), 'neutron.objects.qos.policy.QosPolicy', 'policy_object.QosPolicy', ({(1154, 12, 1154, 21): 'self.ctxt'}, {}), "(self.ctxt, **self.policy_data['policy'])", True, 'from neutron.objects.qos import policy as policy_object\n'), ((1285, 21, 1285, 48), 'neutron_lib.context.get_admin_context', 'context.get_admin_context', ({}, {}), '()', False, 'from neutron_lib import context\n'), ((1324, 24, 1324, 35), 'unittest.mock.Mock', 'mock.Mock', ({}, {}), '()', False, 'from unittest import mock\n'), ((1335, 23, 1335, 34), 'unittest.mock.Mock', 'mock.Mock', ({}, {}), '()', False, 'from unittest import mock\n'), ((1366, 18, 1367, 52), 'neutron.objects.qos.policy.QosPolicy', 'policy_object.QosPolicy', ({(1367, 12, 1367, 21): 'self.ctxt'}, {}), "(self.ctxt, **self.policy_data['policy'])", True, 'from neutron.objects.qos import policy as policy_object\n'), ((1393, 18, 1394, 52), 'neutron.objects.qos.policy.QosPolicy', 'policy_object.QosPolicy', ({(1394, 12, 1394, 21): 'self.ctxt'}, {}), "(self.ctxt, **self.policy_data['policy'])", True, 'from neutron.objects.qos import policy as policy_object\n'), ((1403, 18, 1404, 52), 'neutron.objects.qos.policy.QosPolicy', 'policy_object.QosPolicy', ({(1404, 12, 1404, 21): 'self.ctxt'}, {}), "(self.ctxt, **self.policy_data['policy'])", True, 'from neutron.objects.qos import policy as policy_object\n'), ((1415, 18, 1416, 52), 'neutron.objects.qos.policy.QosPolicy', 'policy_object.QosPolicy', ({(1416, 12, 1416, 21): 'self.ctxt'}, {}), "(self.ctxt, **self.policy_data['policy'])", True, 'from neutron.objects.qos import policy as policy_object\n'), ((1425, 18, 1426, 52), 'neutron.objects.qos.policy.QosPolicy', 'policy_object.QosPolicy', ({(1426, 12, 1426, 21): 'self.ctxt'}, {}), "(self.ctxt, **self.policy_data['policy'])", True, 'from neutron.objects.qos import policy as policy_object\n'), ((1512, 21, 1512, 48), 'neutron_lib.context.get_admin_context', 'context.get_admin_context', ({}, {}), '()', False, 'from neutron_lib import context\n'), ((1542, 18, 1543, 44), 'neutron.objects.network.NetworkSegment', 'network_object.NetworkSegment', (), '', True, 'from neutron.objects import network as network_object\n'), ((1544, 14, 1546, 31), 'neutron.objects.network.Network', 'network_object.Network', (), '', True, 'from neutron.objects import network as network_object\n'), ((1570, 18, 1571, 44), 'neutron.objects.network.NetworkSegment', 'network_object.NetworkSegment', (), '', True, 'from neutron.objects import network as network_object\n'), ((1572, 14, 1574, 31), 'neutron.objects.network.Network', 'network_object.Network', (), '', True, 'from neutron.objects import network as network_object\n'), ((1820, 21, 1820, 48), 'neutron_lib.context.get_admin_context', 'context.get_admin_context', ({}, {}), '()', False, 'from neutron_lib import context\n'), ((1851, 15, 1851, 62), 'neutron.extensions.qos_rules_alias.Qos_rules_alias.get_resources', 'qos_rules_alias.Qos_rules_alias.get_resources', ({}, {}), '()', False, 'from neutron.extensions import qos_rules_alias\n'), ((1863, 15, 1864, 27), 'neutron.extensions.qos_pps_minimum_rule_alias.Qos_pps_minimum_rule_alias.get_resources', 'qos_pps_minimum_rule_alias.Qos_pps_minimum_rule_alias.get_resources', ({}, {}), '()', False, 'from neutron.extensions import qos_pps_minimum_rule_alias\n'), ((1877, 30, 1878, 69), 'unittest.mock.patch', 'mock.patch', ({(1878, 12, 1878, 68): '"""neutron.notifiers.batch_notifier.BatchNotifier._notify"""'}, {}), "('neutron.notifiers.batch_notifier.BatchNotifier._notify')", False, 'from unittest import mock\n'), ((1885, 26, 1885, 69), 'neutron_lib.plugins.directory.get_plugin', 'directory.get_plugin', ({(1885, 47, 1885, 68): 'plugins_constants.QOS'}, {}), '(plugins_constants.QOS)', False, 'from neutron_lib.plugins import directory\n'), ((1887, 20, 1887, 63), 'neutron_lib.context.Context', 'context.Context', ({(1887, 36, 1887, 47): '"""fake_user"""', (1887, 49, 1887, 62): '"""fake_tenant"""'}, {}), "('fake_user', 'fake_tenant')", False, 'from neutron_lib import context\n'), ((1894, 29, 1894, 54), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ({}, {}), '()', False, 'from oslo_utils import uuidutils\n'), ((2001, 30, 2002, 69), 'unittest.mock.patch', 'mock.patch', ({(2002, 12, 2002, 68): '"""neutron.notifiers.batch_notifier.BatchNotifier._notify"""'}, {}), "('neutron.notifiers.batch_notifier.BatchNotifier._notify')", False, 'from unittest import mock\n'), ((2009, 26, 2009, 69), 'neutron_lib.plugins.directory.get_plugin', 'directory.get_plugin', ({(2009, 47, 2009, 68): 'plugins_constants.QOS'}, {}), '(plugins_constants.QOS)', False, 'from neutron_lib.plugins import directory\n'), ((2011, 20, 2011, 63), 'neutron_lib.context.Context', 'context.Context', ({(2011, 36, 2011, 47): '"""fake_user"""', (2011, 49, 2011, 62): '"""fake_tenant"""'}, {}), "('fake_user', 'fake_tenant')", False, 'from neutron_lib import context\n'), ((2016, 29, 2016, 54), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ({}, {}), '()', False, 'from oslo_utils import uuidutils\n'), ((2041, 8, 2041, 61), 'oslo_config.cfg.CONF.set_override', 'cfg.CONF.set_override', ({(2041, 30, 2041, 43): '"""core_plugin"""', (2041, 45, 2041, 60): 'DB_PLUGIN_KLASS'}, {}), "('core_plugin', DB_PLUGIN_KLASS)", False, 'from oslo_config import cfg\n'), ((2042, 8, 2042, 57), 'oslo_config.cfg.CONF.set_override', 'cfg.CONF.set_override', ({(2042, 30, 2042, 47): '"""service_plugins"""', (2042, 49, 2042, 56): "['qos']"}, {}), "('service_plugins', ['qos'])", False, 'from oslo_config import cfg\n'), ((2044, 8, 2044, 22), 'neutron.manager.init', 'manager.init', ({}, {}), '()', False, 'from neutron import manager\n'), ((2045, 26, 2045, 69), 'neutron_lib.plugins.directory.get_plugin', 'directory.get_plugin', ({(2045, 47, 2045, 68): 'plugins_constants.QOS'}, {}), '(plugins_constants.QOS)', False, 'from neutron_lib.plugins import directory\n'), ((2046, 41, 2046, 52), 'unittest.mock.Mock', 'mock.Mock', ({}, {}), '()', False, 'from unittest import mock\n'), ((2049, 23, 2049, 50), 'neutron_lib.context.get_admin_context', 'context.get_admin_context', ({}, {}), '()', False, 'from neutron_lib import context\n'), ((2050, 26, 2050, 51), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ({}, {}), '()', False, 'from oslo_utils import uuidutils\n'), ((2053, 21, 2055, 29), 'neutron.objects.qos.policy.QosPolicy', 'policy_object.QosPolicy', (), '', True, 'from neutron.objects.qos import policy as policy_object\n'), ((2062, 19, 2065, 23), 'neutron.objects.qos.rule.QosMinimumBandwidthRule', 'rule_object.QosMinimumBandwidthRule', (), '', True, 'from neutron.objects.qos import rule as rule_object\n'), ((2072, 19, 2075, 23), 'neutron.objects.qos.rule.QosMinimumPacketRateRule', 'rule_object.QosMinimumPacketRateRule', (), '', True, 'from neutron.objects.qos import rule as rule_object\n'), ((2085, 15, 2090, 23), 'neutron.objects.ports.Port', 'ports_object.Port', (), '', True, 'from neutron.objects import ports as ports_object\n'), ((2095, 18, 2096, 69), 'neutron.objects.network.Network', 'network_object.Network', (), '', True, 'from neutron.objects import network as network_object\n'), ((2289, 25, 2289, 41), 'unittest.mock.MagicMock', 'mock.MagicMock', ({}, {}), '()', False, 'from unittest import mock\n'), ((2314, 25, 2314, 41), 'unittest.mock.MagicMock', 'mock.MagicMock', ({}, {}), '()', False, 'from unittest import mock\n'), ((2330, 25, 2330, 41), 'unittest.mock.MagicMock', 'mock.MagicMock', ({}, {}), '()', False, 'from unittest import mock\n'), ((2344, 24, 2347, 54), 'neutron.objects.ports.PortBinding', 'ports_object.PortBinding', (), '', True, 'from neutron.objects import ports as ports_object\n'), ((2386, 25, 2386, 41), 'unittest.mock.MagicMock', 'mock.MagicMock', ({}, {}), '()', False, 'from unittest import mock\n'), ((2400, 24, 2402, 77), 'neutron.objects.ports.PortBinding', 'ports_object.PortBinding', (), '', True, 'from neutron.objects import ports as ports_object\n'), ((2410, 24, 2412, 77), 'neutron.objects.ports.PortBinding', 'ports_object.PortBinding', (), '', True, 'from neutron.objects import ports as ports_object\n'), ((2706, 25, 2706, 69), 'neutron.objects.qos.rule.QosDscpMarkingRule', 'rule_object.QosDscpMarkingRule', (), '', True, 'from neutron.objects.qos import rule as rule_object\n'), ((2707, 25, 2707, 69), 'neutron.objects.qos.rule.QosDscpMarkingRule', 'rule_object.QosDscpMarkingRule', (), '', True, 'from neutron.objects.qos import rule as rule_object\n'), ((2727, 24, 2727, 68), 'neutron.objects.qos.rule.QosDscpMarkingRule', 'rule_object.QosDscpMarkingRule', (), '', True, 'from neutron.objects.qos import rule as rule_object\n'), ((2741, 24, 2741, 68), 'neutron.objects.qos.rule.QosDscpMarkingRule', 'rule_object.QosDscpMarkingRule', (), '', True, 'from neutron.objects.qos import rule as rule_object\n'), ((163, 13, 164, 52), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((165, 16, 168, 46), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((169, 16, 172, 47), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((173, 16, 176, 53), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((177, 19, 178, 36), 'neutron.services.qos.qos_plugin.QoSPlugin._extend_port_resource_request', 'qos_plugin.QoSPlugin._extend_port_resource_request', ({(178, 16, 178, 24): 'port_res', (178, 26, 178, 35): 'self.port'}, {}), '(port_res, self.port)', False, 'from neutron.services.qos import qos_plugin\n'), ((208, 13, 209, 52), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((210, 16, 213, 46), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((214, 16, 217, 47), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((218, 16, 221, 53), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((222, 19, 223, 32), 'neutron.services.qos.qos_plugin.QoSPlugin._extend_port_resource_request_bulk', 'qos_plugin.QoSPlugin._extend_port_resource_request_bulk', ({(223, 16, 223, 25): 'ports_res', (223, 27, 223, 31): 'None'}, {}), '(ports_res, None)', False, 'from neutron.services.qos import qos_plugin\n'), ((279, 18, 279, 43), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ({}, {}), '()', False, 'from oslo_utils import uuidutils\n'), ((283, 18, 283, 43), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ({}, {}), '()', False, 'from oslo_utils import uuidutils\n'), ((421, 18, 421, 43), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ({}, {}), '()', False, 'from oslo_utils import uuidutils\n'), ((425, 18, 425, 43), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ({}, {}), '()', False, 'from oslo_utils import uuidutils\n'), ((504, 12, 504, 46), 'unittest.mock.MagicMock', 'mock.MagicMock', (), '', False, 'from unittest import mock\n'), ((506, 12, 506, 46), 'unittest.mock.MagicMock', 'mock.MagicMock', (), '', False, 'from unittest import mock\n'), ((509, 12, 509, 56), 'unittest.mock.MagicMock', 'mock.MagicMock', (), '', False, 'from unittest import mock\n'), ((514, 13, 517, 9), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((517, 11, 519, 9), 'unittest.mock.patch.object', 'mock.patch.object', ({(518, 12, 518, 23): 'self.policy', (518, 25, 518, 45): '"""get_bound_networks"""'}, {}), "(self.policy, 'get_bound_networks')", False, 'from unittest import mock\n'), ((519, 11, 521, 9), 'unittest.mock.patch.object', 'mock.patch.object', ({(520, 12, 520, 23): 'self.policy', (520, 25, 520, 42): '"""get_bound_ports"""'}, {}), "(self.policy, 'get_bound_ports')", False, 'from unittest import mock\n'), ((545, 13, 548, 9), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((548, 23, 551, 9), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((551, 25, 553, 9), 'unittest.mock.patch.object', 'mock.patch.object', ({(552, 12, 552, 27): 'self.qos_plugin', (552, 29, 552, 55): '"""validate_policy_for_port"""'}, {}), "(self.qos_plugin, 'validate_policy_for_port')", False, 'from unittest import mock\n'), ((553, 39, 555, 9), 'unittest.mock.patch.object', 'mock.patch.object', (), '', False, 'from unittest import mock\n'), ((606, 13, 609, 9), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((609, 24, 612, 9), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((612, 25, 614, 9), 'unittest.mock.patch.object', 'mock.patch.object', ({(613, 12, 613, 27): 'self.qos_plugin', (613, 29, 613, 58): '"""validate_policy_for_network"""'}, {}), "(self.qos_plugin, 'validate_policy_for_network')", False, 'from unittest import mock\n'), ((614, 42, 616, 9), 'unittest.mock.patch.object', 'mock.patch.object', ({(615, 12, 615, 27): 'self.qos_plugin', (615, 29, 615, 56): '"""validate_policy_for_ports"""'}, {}), "(self.qos_plugin, 'validate_policy_for_ports')", False, 'from unittest import mock\n'), ((616, 40, 618, 9), 'unittest.mock.patch.object', 'mock.patch.object', (), '', False, 'from unittest import mock\n'), ((650, 22, 650, 47), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ({}, {}), '()', False, 'from oslo_utils import uuidutils\n'), ((651, 13, 654, 9), 'unittest.mock.patch.object', 'mock.patch.object', (), '', False, 'from unittest import mock\n'), ((662, 22, 662, 47), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ({}, {}), '()', False, 'from oslo_utils import uuidutils\n'), ((663, 13, 666, 9), 'unittest.mock.patch.object', 'mock.patch.object', (), '', False, 'from unittest import mock\n'), ((676, 13, 679, 9), 'unittest.mock.patch.object', 'mock.patch.object', (), '', False, 'from unittest import mock\n'), ((701, 13, 703, 36), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((704, 12, 706, 33), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((707, 12, 710, 36), 'unittest.mock.patch.object', 'mock.patch.object', (), '', False, 'from unittest import mock\n'), ((729, 13, 731, 36), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((732, 12, 734, 33), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((735, 12, 738, 36), 'unittest.mock.patch.object', 'mock.patch.object', (), '', False, 'from unittest import mock\n'), ((785, 13, 785, 63), 'unittest.mock.patch', 'mock.patch', ({(785, 24, 785, 62): '"""neutron.objects.qos.policy.QosPolicy"""'}, {}), "('neutron.objects.qos.policy.QosPolicy')", False, 'from unittest import mock\n'), ((851, 13, 853, 42), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((854, 16, 856, 70), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((860, 36, 860, 54), 'unittest.mock.call.create', 'mock.call.create', ({}, {}), '()', False, 'from unittest import mock\n'), ((861, 41, 862, 63), 'unittest.mock.call.driver.call', 'mock.call.driver.call', ({(862, 16, 862, 41): '"""update_policy_precommit"""', (862, 43, 862, 52): 'self.ctxt', (862, 54, 862, 62): 'mock.ANY'}, {}), "('update_policy_precommit', self.ctxt, mock.ANY)", False, 'from unittest import mock\n'), ((863, 31, 864, 53), 'unittest.mock.call.driver.call', 'mock.call.driver.call', ({(864, 16, 864, 31): '"""update_policy"""', (864, 33, 864, 42): 'self.ctxt', (864, 44, 864, 52): 'mock.ANY'}, {}), "('update_policy', self.ctxt, mock.ANY)", False, 'from unittest import mock\n'), ((873, 13, 874, 45), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((884, 13, 885, 45), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((896, 13, 897, 45), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((907, 13, 908, 45), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((923, 13, 924, 45), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((941, 13, 942, 49), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((942, 51, 944, 37), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((950, 36, 950, 54), 'unittest.mock.call.update', 'mock.call.update', ({}, {}), '()', False, 'from unittest import mock\n'), ((951, 41, 952, 63), 'unittest.mock.call.driver.call', 'mock.call.driver.call', ({(952, 16, 952, 41): '"""update_policy_precommit"""', (952, 43, 952, 52): 'self.ctxt', (952, 54, 952, 62): 'mock.ANY'}, {}), "('update_policy_precommit', self.ctxt, mock.ANY)", False, 'from unittest import mock\n'), ((953, 31, 954, 53), 'unittest.mock.call.driver.call', 'mock.call.driver.call', ({(954, 16, 954, 31): '"""update_policy"""', (954, 33, 954, 42): 'self.ctxt', (954, 44, 954, 52): 'mock.ANY'}, {}), "('update_policy', self.ctxt, mock.ANY)", False, 'from unittest import mock\n'), ((963, 13, 964, 45), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((973, 13, 974, 45), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((984, 13, 985, 45), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((991, 13, 992, 45), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((1002, 13, 1003, 45), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((1010, 13, 1011, 45), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((1025, 13, 1026, 45), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((1043, 13, 1044, 45), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((1050, 36, 1050, 54), 'unittest.mock.call.delete', 'mock.call.delete', ({}, {}), '()', False, 'from unittest import mock\n'), ((1051, 41, 1052, 63), 'unittest.mock.call.driver.call', 'mock.call.driver.call', ({(1052, 16, 1052, 41): '"""update_policy_precommit"""', (1052, 43, 1052, 52): 'self.ctxt', (1052, 54, 1052, 62): 'mock.ANY'}, {}), "('update_policy_precommit', self.ctxt, mock.ANY)", False, 'from unittest import mock\n'), ((1053, 31, 1054, 53), 'unittest.mock.call.driver.call', 'mock.call.driver.call', ({(1054, 16, 1054, 31): '"""update_policy"""', (1054, 33, 1054, 42): 'self.ctxt', (1054, 44, 1054, 52): 'mock.ANY'}, {}), "('update_policy', self.ctxt, mock.ANY)", False, 'from unittest import mock\n'), ((1063, 13, 1064, 45), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((1072, 13, 1073, 49), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((1083, 13, 1084, 49), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((1094, 13, 1095, 49), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((1109, 13, 1110, 42), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((1117, 13, 1118, 42), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((1125, 13, 1126, 42), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((1135, 13, 1136, 45), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((1145, 13, 1146, 45), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((1155, 13, 1156, 45), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((1163, 13, 1164, 49), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((1174, 13, 1175, 49), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((1188, 13, 1189, 42), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((1196, 13, 1197, 42), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((1204, 13, 1205, 49), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((1215, 13, 1216, 49), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((1226, 13, 1227, 49), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((1241, 13, 1242, 42), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((1249, 13, 1250, 42), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((1257, 13, 1258, 42), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((1265, 13, 1266, 42), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((1273, 13, 1274, 42), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((1294, 13, 1297, 9), 'unittest.mock.patch.object', 'mock.patch.object', (), '', False, 'from unittest import mock\n'), ((1314, 13, 1315, 72), 'unittest.mock.patch.object', 'mock.patch.object', (), '', False, 'from unittest import mock\n'), ((1353, 31, 1354, 73), 'unittest.mock.call.driver.call', 'mock.call.driver.call', ({(1353, 53, 1353, 68): '"""update_policy"""', (1354, 53, 1354, 62): 'self.ctxt', (1354, 64, 1354, 72): 'mock.ANY'}, {}), "('update_policy', self.ctxt, mock.ANY)", False, 'from unittest import mock\n'), ((1368, 13, 1369, 45), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((1384, 13, 1385, 45), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((1395, 13, 1396, 45), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((1405, 13, 1406, 45), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((1417, 13, 1418, 45), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((1427, 13, 1428, 45), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((1436, 13, 1437, 49), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((1447, 13, 1448, 49), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((1458, 13, 1459, 49), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((1472, 13, 1473, 42), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((1480, 13, 1481, 42), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((1488, 13, 1489, 42), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((1496, 13, 1497, 42), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((1504, 13, 1505, 42), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((1521, 13, 1524, 9), 'unittest.mock.patch.object', 'mock.patch.object', (), '', False, 'from unittest import mock\n'), ((1552, 13, 1554, 37), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((1555, 12, 1557, 33), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((1558, 12, 1561, 36), 'unittest.mock.patch.object', 'mock.patch.object', (), '', False, 'from unittest import mock\n'), ((1580, 13, 1582, 37), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((1583, 12, 1585, 33), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((1586, 12, 1589, 36), 'unittest.mock.patch.object', 'mock.patch.object', (), '', False, 'from unittest import mock\n'), ((1626, 27, 1627, 67), 'neutron.objects.qos.rule.QosMinimumPacketRateRule', 'rule_object.QosMinimumPacketRateRule', ({(1627, 16, 1627, 25): 'self.ctxt'}, {}), "(self.ctxt, **rule_data[\n    'minimum_packet_rate_rule'])", True, 'from neutron.objects.qos import rule as rule_object\n'), ((1640, 13, 1641, 45), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((1657, 13, 1658, 45), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((1666, 13, 1667, 42), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((1676, 13, 1677, 45), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((1723, 13, 1724, 45), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((1732, 13, 1733, 42), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((1743, 13, 1744, 45), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((1752, 13, 1753, 45), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((1760, 13, 1761, 42), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((1768, 13, 1769, 49), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((1779, 13, 1780, 49), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((1790, 13, 1791, 49), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((1804, 13, 1805, 42), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((1812, 13, 1813, 42), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((1829, 13, 1832, 9), 'unittest.mock.patch.object', 'mock.patch.object', (), '', False, 'from unittest import mock\n'), ((1933, 22, 1933, 47), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ({}, {}), '()', False, 'from oslo_utils import uuidutils\n'), ((1953, 22, 1953, 47), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ({}, {}), '()', False, 'from oslo_utils import uuidutils\n'), ((1970, 22, 1970, 47), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ({}, {}), '()', False, 'from oslo_utils import uuidutils\n'), ((1988, 22, 1988, 47), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ({}, {}), '()', False, 'from oslo_utils import uuidutils\n'), ((2061, 42, 2061, 67), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ({}, {}), '()', False, 'from oslo_utils import uuidutils\n'), ((2071, 42, 2071, 67), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ({}, {}), '()', False, 'from oslo_utils import uuidutils\n'), ((2081, 42, 2081, 67), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ({}, {}), '()', False, 'from oslo_utils import uuidutils\n'), ((2107, 13, 2108, 61), 'unittest.mock.patch.object', 'mock.patch.object', ({(2107, 31, 2107, 46): 'self.qos_plugin', (2108, 31, 2108, 60): '"""validate_policy_for_network"""'}, {}), "(self.qos_plugin, 'validate_policy_for_network')", False, 'from unittest import mock\n'), ((2141, 13, 2141, 75), 'unittest.mock.patch.object', 'mock.patch.object', ({(2141, 31, 2141, 46): 'self.qos_plugin', (2141, 48, 2141, 74): '"""validate_policy_for_port"""'}, {}), "(self.qos_plugin, 'validate_policy_for_port')", False, 'from unittest import mock\n'), ((2201, 13, 2203, 47), 'unittest.mock.patch.object', 'mock.patch.object', ({(2202, 16, 2202, 31): 'self.qos_plugin', (2203, 16, 2203, 46): '"""_change_placement_allocation"""'}, {}), "(self.qos_plugin, '_change_placement_allocation')", False, 'from unittest import mock\n'), ((2219, 13, 2221, 47), 'unittest.mock.patch.object', 'mock.patch.object', ({(2220, 16, 2220, 31): 'self.qos_plugin', (2221, 16, 2221, 46): '"""_change_placement_allocation"""'}, {}), "(self.qos_plugin, '_change_placement_allocation')", False, 'from unittest import mock\n'), ((2237, 13, 2239, 47), 'unittest.mock.patch.object', 'mock.patch.object', ({(2238, 16, 2238, 31): 'self.qos_plugin', (2239, 16, 2239, 46): '"""_change_placement_allocation"""'}, {}), "(self.qos_plugin, '_change_placement_allocation')", False, 'from unittest import mock\n'), ((2256, 13, 2258, 47), 'unittest.mock.patch.object', 'mock.patch.object', ({(2257, 16, 2257, 31): 'self.qos_plugin', (2258, 16, 2258, 46): '"""_change_placement_allocation"""'}, {}), "(self.qos_plugin, '_change_placement_allocation')", False, 'from unittest import mock\n'), ((2275, 13, 2277, 47), 'unittest.mock.patch.object', 'mock.patch.object', ({(2276, 16, 2276, 31): 'self.qos_plugin', (2277, 16, 2277, 46): '"""_change_placement_allocation"""'}, {}), "(self.qos_plugin, '_change_placement_allocation')", False, 'from unittest import mock\n'), ((2291, 13, 2293, 47), 'unittest.mock.patch.object', 'mock.patch.object', ({(2292, 16, 2292, 31): 'self.qos_plugin', (2293, 16, 2293, 46): '"""_change_placement_allocation"""'}, {}), "(self.qos_plugin, '_change_placement_allocation')", False, 'from unittest import mock\n'), ((2316, 13, 2318, 47), 'unittest.mock.patch.object', 'mock.patch.object', ({(2317, 16, 2317, 31): 'self.qos_plugin', (2318, 16, 2318, 46): '"""_change_placement_allocation"""'}, {}), "(self.qos_plugin, '_change_placement_allocation')", False, 'from unittest import mock\n'), ((2352, 13, 2354, 47), 'unittest.mock.patch.object', 'mock.patch.object', ({(2353, 16, 2353, 31): 'self.qos_plugin', (2354, 16, 2354, 46): '"""_change_placement_allocation"""'}, {}), "(self.qos_plugin, '_change_placement_allocation')", False, 'from unittest import mock\n'), ((2367, 12, 2374, 25), 'unittest.mock.call', 'mock.call', ({(2368, 16, 2368, 28): 'original_qos', (2369, 16, 2369, 20): 'None', (2370, 16, 2373, 50): "{'id': port1.id, 'device_owner': 'compute:uu:id', 'qos_policy_id': None,\n    'qos_network_policy_id': None}", (2374, 16, 2374, 24): 'mock.ANY'}, {}), "(original_qos, None, {'id': port1.id, 'device_owner':\n    'compute:uu:id', 'qos_policy_id': None, 'qos_network_policy_id': None},\n    mock.ANY)", False, 'from unittest import mock\n'), ((2417, 13, 2419, 47), 'unittest.mock.patch.object', 'mock.patch.object', ({(2418, 16, 2418, 31): 'self.qos_plugin', (2419, 16, 2419, 46): '"""_change_placement_allocation"""'}, {}), "(self.qos_plugin, '_change_placement_allocation')", False, 'from unittest import mock\n'), ((2433, 12, 2440, 25), 'unittest.mock.call', 'mock.call', ({(2434, 16, 2434, 28): 'original_qos', (2435, 16, 2435, 19): 'qos', (2436, 16, 2439, 52): "{'id': port1.id, 'device_owner': 'compute:uu:id', 'qos_policy_id': None,\n    'qos_network_policy_id': qos.id}", (2440, 16, 2440, 24): 'mock.ANY'}, {}), "(original_qos, qos, {'id': port1.id, 'device_owner':\n    'compute:uu:id', 'qos_policy_id': None, 'qos_network_policy_id': qos.id\n    }, mock.ANY)", False, 'from unittest import mock\n'), ((2441, 12, 2448, 25), 'unittest.mock.call', 'mock.call', ({(2442, 16, 2442, 28): 'original_qos', (2443, 16, 2443, 19): 'qos', (2444, 16, 2447, 52): "{'id': port2.id, 'device_owner': 'compute:uu:id', 'qos_policy_id': None,\n    'qos_network_policy_id': qos.id}", (2448, 16, 2448, 24): 'mock.ANY'}, {}), "(original_qos, qos, {'id': port2.id, 'device_owner':\n    'compute:uu:id', 'qos_policy_id': None, 'qos_network_policy_id': qos.id\n    }, mock.ANY)", False, 'from unittest import mock\n'), ((2521, 13, 2522, 40), 'unittest.mock.patch.object', 'mock.patch.object', ({(2521, 31, 2521, 64): 'self.qos_plugin._placement_client', (2522, 16, 2522, 39): '"""update_qos_allocation"""'}, {}), "(self.qos_plugin._placement_client, 'update_qos_allocation')", False, 'from unittest import mock\n'), ((2536, 13, 2537, 40), 'unittest.mock.patch.object', 'mock.patch.object', ({(2536, 31, 2536, 64): 'self.qos_plugin._placement_client', (2537, 16, 2537, 39): '"""update_qos_allocation"""'}, {}), "(self.qos_plugin._placement_client, 'update_qos_allocation')", False, 'from unittest import mock\n'), ((2552, 13, 2553, 40), 'unittest.mock.patch.object', 'mock.patch.object', ({(2552, 31, 2552, 64): 'self.qos_plugin._placement_client', (2553, 16, 2553, 39): '"""update_qos_allocation"""'}, {}), "(self.qos_plugin._placement_client, 'update_qos_allocation')", False, 'from unittest import mock\n'), ((2572, 13, 2573, 40), 'unittest.mock.patch.object', 'mock.patch.object', ({(2572, 31, 2572, 64): 'self.qos_plugin._placement_client', (2573, 16, 2573, 39): '"""update_qos_allocation"""'}, {}), "(self.qos_plugin._placement_client, 'update_qos_allocation')", False, 'from unittest import mock\n'), ((2594, 13, 2595, 40), 'unittest.mock.patch.object', 'mock.patch.object', ({(2594, 31, 2594, 64): 'self.qos_plugin._placement_client', (2595, 16, 2595, 39): '"""update_qos_allocation"""'}, {}), "(self.qos_plugin._placement_client, 'update_qos_allocation')", False, 'from unittest import mock\n'), ((2606, 13, 2607, 40), 'unittest.mock.patch.object', 'mock.patch.object', ({(2606, 31, 2606, 64): 'self.qos_plugin._placement_client', (2607, 16, 2607, 39): '"""update_qos_allocation"""'}, {}), "(self.qos_plugin._placement_client, 'update_qos_allocation')", False, 'from unittest import mock\n'), ((2619, 13, 2620, 40), 'unittest.mock.patch.object', 'mock.patch.object', ({(2619, 31, 2619, 64): 'self.qos_plugin._placement_client', (2620, 16, 2620, 39): '"""update_qos_allocation"""'}, {}), "(self.qos_plugin._placement_client, 'update_qos_allocation')", False, 'from unittest import mock\n'), ((2635, 13, 2636, 40), 'unittest.mock.patch.object', 'mock.patch.object', ({(2635, 31, 2635, 64): 'self.qos_plugin._placement_client', (2636, 16, 2636, 39): '"""update_qos_allocation"""'}, {}), "(self.qos_plugin._placement_client, 'update_qos_allocation')", False, 'from unittest import mock\n'), ((2650, 13, 2651, 40), 'unittest.mock.patch.object', 'mock.patch.object', ({(2650, 31, 2650, 64): 'self.qos_plugin._placement_client', (2651, 16, 2651, 39): '"""update_qos_allocation"""'}, {}), "(self.qos_plugin._placement_client, 'update_qos_allocation')", False, 'from unittest import mock\n'), ((2667, 13, 2668, 40), 'unittest.mock.patch.object', 'mock.patch.object', ({(2667, 31, 2667, 64): 'self.qos_plugin._placement_client', (2668, 16, 2668, 39): '"""update_qos_allocation"""'}, {}), "(self.qos_plugin._placement_client, 'update_qos_allocation')", False, 'from unittest import mock\n'), ((2682, 13, 2683, 40), 'unittest.mock.patch.object', 'mock.patch.object', ({(2682, 31, 2682, 64): 'self.qos_plugin._placement_client', (2683, 16, 2683, 39): '"""update_qos_allocation"""'}, {}), "(self.qos_plugin._placement_client, 'update_qos_allocation')", False, 'from unittest import mock\n'), ((2692, 13, 2693, 40), 'unittest.mock.patch.object', 'mock.patch.object', ({(2692, 31, 2692, 64): 'self.qos_plugin._placement_client', (2693, 16, 2693, 39): '"""update_qos_allocation"""'}, {}), "(self.qos_plugin._placement_client, 'update_qos_allocation')", False, 'from unittest import mock\n'), ((2718, 13, 2719, 40), 'unittest.mock.patch.object', 'mock.patch.object', ({(2718, 31, 2718, 64): 'self.qos_plugin._placement_client', (2719, 16, 2719, 39): '"""update_qos_allocation"""'}, {}), "(self.qos_plugin._placement_client, 'update_qos_allocation')", False, 'from unittest import mock\n'), ((2732, 13, 2733, 40), 'unittest.mock.patch.object', 'mock.patch.object', ({(2732, 31, 2732, 64): 'self.qos_plugin._placement_client', (2733, 16, 2733, 39): '"""update_qos_allocation"""'}, {}), "(self.qos_plugin._placement_client, 'update_qos_allocation')", False, 'from unittest import mock\n'), ((2746, 13, 2747, 40), 'unittest.mock.patch.object', 'mock.patch.object', ({(2746, 31, 2746, 64): 'self.qos_plugin._placement_client', (2747, 16, 2747, 39): '"""update_qos_allocation"""'}, {}), "(self.qos_plugin._placement_client, 'update_qos_allocation')", False, 'from unittest import mock\n'), ((2760, 13, 2761, 40), 'unittest.mock.patch.object', 'mock.patch.object', ({(2760, 31, 2760, 64): 'self.qos_plugin._placement_client', (2761, 16, 2761, 39): '"""update_qos_allocation"""'}, {}), "(self.qos_plugin._placement_client, 'update_qos_allocation')", False, 'from unittest import mock\n'), ((2771, 13, 2772, 40), 'unittest.mock.patch.object', 'mock.patch.object', ({(2771, 31, 2771, 64): 'self.qos_plugin._placement_client', (2772, 16, 2772, 39): '"""update_qos_allocation"""'}, {}), "(self.qos_plugin._placement_client, 'update_qos_allocation')", False, 'from unittest import mock\n'), ((2773, 48, 2775, 13), 'keystoneauth1.exceptions.Conflict', 'ks_exc.Conflict', (), '', True, 'from keystoneauth1 import exceptions as ks_exc\n'), ((2786, 13, 2787, 40), 'unittest.mock.patch.object', 'mock.patch.object', ({(2786, 31, 2786, 64): 'self.qos_plugin._placement_client', (2787, 16, 2787, 39): '"""update_qos_allocation"""'}, {}), "(self.qos_plugin._placement_client, 'update_qos_allocation')", False, 'from unittest import mock\n'), ((2789, 16, 2790, 44), 'neutron_lib.exceptions.placement.PlacementAllocationGenerationConflict', 'pl_exc.PlacementAllocationGenerationConflict', (), '', True, 'from neutron_lib.exceptions import placement as pl_exc\n'), ((2802, 13, 2803, 40), 'unittest.mock.patch.object', 'mock.patch.object', ({(2802, 31, 2802, 64): 'self.qos_plugin._placement_client', (2803, 16, 2803, 39): '"""update_qos_allocation"""'}, {}), "(self.qos_plugin._placement_client, 'update_qos_allocation')", False, 'from unittest import mock\n'), ((58, 8, 58, 58), 'unittest.mock.patch', 'mock.patch', ({(58, 19, 58, 57): '"""neutron.objects.db.api.create_object"""'}, {}), "('neutron.objects.db.api.create_object')", False, 'from unittest import mock\n'), ((59, 8, 59, 58), 'unittest.mock.patch', 'mock.patch', ({(59, 19, 59, 57): '"""neutron.objects.db.api.update_object"""'}, {}), "('neutron.objects.db.api.update_object')", False, 'from unittest import mock\n'), ((60, 8, 60, 58), 'unittest.mock.patch', 'mock.patch', ({(60, 19, 60, 57): '"""neutron.objects.db.api.delete_object"""'}, {}), "('neutron.objects.db.api.delete_object')", False, 'from unittest import mock\n'), ((61, 8, 61, 55), 'unittest.mock.patch', 'mock.patch', ({(61, 19, 61, 54): '"""neutron.objects.db.api.get_object"""'}, {}), "('neutron.objects.db.api.get_object')", False, 'from unittest import mock\n'), ((67, 8, 69, 9), 'unittest.mock.patch', 'mock.patch', ({(68, 12, 68, 72): '"""neutron.objects.base.NeutronDbObject.modify_fields_from_db"""'}, {}), "('neutron.objects.base.NeutronDbObject.modify_fields_from_db')", False, 'from unittest import mock\n'), ((70, 8, 70, 67), 'unittest.mock.patch.object', 'mock.patch.object', ({(70, 26, 70, 49): 'policy_object.QosPolicy', (70, 51, 70, 66): '"""unset_default"""'}, {}), "(policy_object.QosPolicy, 'unset_default')", False, 'from unittest import mock\n'), ((71, 8, 71, 65), 'unittest.mock.patch.object', 'mock.patch.object', ({(71, 26, 71, 49): 'policy_object.QosPolicy', (71, 51, 71, 64): '"""set_default"""'}, {}), "(policy_object.QosPolicy, 'set_default')", False, 'from unittest import mock\n'), ((81, 24, 82, 63), 'unittest.mock.patch', 'mock.patch', ({(81, 35, 82, 62): '"""neutron.api.rpc.handlers.resources_rpc.ResourcesPushRpcApi.push"""'}, {}), "('neutron.api.rpc.handlers.resources_rpc.ResourcesPushRpcApi.push')", False, 'from unittest import mock\n'), ((88, 29, 88, 54), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ({}, {}), '()', False, 'from oslo_utils import uuidutils\n'), ((89, 37, 89, 62), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ({}, {}), '()', False, 'from oslo_utils import uuidutils\n'), ((96, 43, 96, 68), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ({}, {}), '()', False, 'from oslo_utils import uuidutils\n'), ((99, 40, 99, 65), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ({}, {}), '()', False, 'from oslo_utils import uuidutils\n'), ((102, 22, 102, 47), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ({}, {}), '()', False, 'from oslo_utils import uuidutils\n'), ((105, 22, 105, 47), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ({}, {}), '()', False, 'from oslo_utils import uuidutils\n'), ((109, 22, 109, 47), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ({}, {}), '()', False, 'from oslo_utils import uuidutils\n'), ((146, 27, 146, 52), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ({}, {}), '()', False, 'from oslo_utils import uuidutils\n'), ((573, 22, 573, 47), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ({}, {}), '()', False, 'from oslo_utils import uuidutils\n'), ((582, 47, 582, 72), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ({}, {}), '()', False, 'from oslo_utils import uuidutils\n'), ((599, 15, 599, 40), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ({}, {}), '()', False, 'from oslo_utils import uuidutils\n'), ((600, 26, 600, 51), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ({}, {}), '()', False, 'from oslo_utils import uuidutils\n'), ((602, 15, 602, 40), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ({}, {}), '()', False, 'from oslo_utils import uuidutils\n'), ((638, 22, 638, 47), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ({}, {}), '()', False, 'from oslo_utils import uuidutils\n'), ((647, 47, 647, 72), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ({}, {}), '()', False, 'from oslo_utils import uuidutils\n'), ((698, 15, 698, 40), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ({}, {}), '()', False, 'from oslo_utils import uuidutils\n'), ((699, 23, 699, 48), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ({}, {}), '()', False, 'from oslo_utils import uuidutils\n'), ((726, 15, 726, 40), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ({}, {}), '()', False, 'from oslo_utils import uuidutils\n'), ((727, 23, 727, 48), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ({}, {}), '()', False, 'from oslo_utils import uuidutils\n'), ((756, 27, 756, 48), 'unittest.mock.call.QosPolicy', 'mock.call.QosPolicy', ({}, {}), '()', False, 'from unittest import mock\n'), ((1074, 17, 1076, 41), 'unittest.mock.patch', 'mock.patch', ({(1074, 28, 1076, 40): '"""neutron.objects.qos.rule.QosBandwidthLimitRule.get_object"""'}, {}), "('neutron.objects.qos.rule.QosBandwidthLimitRule.get_object')", False, 'from unittest import mock\n'), ((1085, 17, 1087, 42), 'unittest.mock.patch', 'mock.patch', ({(1085, 28, 1087, 41): '"""neutron.objects.qos.rule.QosBandwidthLimitRule.get_objects"""'}, {}), "('neutron.objects.qos.rule.QosBandwidthLimitRule.get_objects')", False, 'from unittest import mock\n'), ((1096, 17, 1098, 42), 'unittest.mock.patch', 'mock.patch', ({(1096, 28, 1098, 41): '"""neutron.objects.qos.rule.QosBandwidthLimitRule.get_objects"""'}, {}), "('neutron.objects.qos.rule.QosBandwidthLimitRule.get_objects')", False, 'from unittest import mock\n'), ((1165, 17, 1167, 42), 'unittest.mock.patch', 'mock.patch', ({(1165, 28, 1167, 41): '"""neutron.objects.qos.rule.QosDscpMarkingRule.get_objects"""'}, {}), "('neutron.objects.qos.rule.QosDscpMarkingRule.get_objects')", False, 'from unittest import mock\n'), ((1176, 17, 1178, 42), 'unittest.mock.patch', 'mock.patch', ({(1176, 28, 1178, 41): '"""neutron.objects.qos.rule.QosDscpMarkingRule.get_objects"""'}, {}), "('neutron.objects.qos.rule.QosDscpMarkingRule.get_objects')", False, 'from unittest import mock\n'), ((1206, 17, 1208, 41), 'unittest.mock.patch', 'mock.patch', ({(1206, 28, 1208, 40): '"""neutron.objects.qos.rule.QosMinimumBandwidthRule.get_object"""'}, {}), "('neutron.objects.qos.rule.QosMinimumBandwidthRule.get_object')", False, 'from unittest import mock\n'), ((1217, 17, 1219, 42), 'unittest.mock.patch', 'mock.patch', ({(1217, 28, 1219, 41): '"""neutron.objects.qos.rule.QosMinimumBandwidthRule.get_objects"""'}, {}), "('neutron.objects.qos.rule.QosMinimumBandwidthRule.get_objects')", False, 'from unittest import mock\n'), ((1228, 17, 1230, 42), 'unittest.mock.patch', 'mock.patch', ({(1228, 28, 1230, 41): '"""neutron.objects.qos.rule.QosMinimumBandwidthRule.get_objects"""'}, {}), "('neutron.objects.qos.rule.QosMinimumBandwidthRule.get_objects')", False, 'from unittest import mock\n'), ((1438, 17, 1440, 41), 'unittest.mock.patch', 'mock.patch', ({(1438, 28, 1440, 40): '"""neutron.objects.qos.rule.QosPacketRateLimitRule.get_object"""'}, {}), "('neutron.objects.qos.rule.QosPacketRateLimitRule.get_object')", False, 'from unittest import mock\n'), ((1449, 17, 1451, 42), 'unittest.mock.patch', 'mock.patch', ({(1449, 28, 1451, 41): '"""neutron.objects.qos.rule.QosPacketRateLimitRule.get_objects"""'}, {}), "('neutron.objects.qos.rule.QosPacketRateLimitRule.get_objects')", False, 'from unittest import mock\n'), ((1460, 17, 1462, 42), 'unittest.mock.patch', 'mock.patch', ({(1460, 28, 1462, 41): '"""neutron.objects.qos.rule.QosPacketRateLimitRule.get_objects"""'}, {}), "('neutron.objects.qos.rule.QosPacketRateLimitRule.get_objects')", False, 'from unittest import mock\n'), ((1549, 15, 1549, 40), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ({}, {}), '()', False, 'from oslo_utils import uuidutils\n'), ((1550, 23, 1550, 48), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ({}, {}), '()', False, 'from oslo_utils import uuidutils\n'), ((1577, 15, 1577, 40), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ({}, {}), '()', False, 'from oslo_utils import uuidutils\n'), ((1578, 23, 1578, 48), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ({}, {}), '()', False, 'from oslo_utils import uuidutils\n'), ((1617, 17, 1618, 49), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((1629, 17, 1630, 49), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((1651, 22, 1651, 47), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ({}, {}), '()', False, 'from oslo_utils import uuidutils\n'), ((1705, 16, 1706, 75), 'neutron.objects.qos.rule.QosMinimumPacketRateRule', 'rule_object.QosMinimumPacketRateRule', ({(1706, 20, 1706, 29): 'self.ctxt'}, {}), "(self.ctxt, **rules_data[0][\n    'minimum_packet_rate_rule'])", True, 'from neutron.objects.qos import rule as rule_object\n'), ((1707, 16, 1708, 75), 'neutron.objects.qos.rule.QosMinimumPacketRateRule', 'rule_object.QosMinimumPacketRateRule', ({(1708, 20, 1708, 29): 'self.ctxt'}, {}), "(self.ctxt, **rules_data[1][\n    'minimum_packet_rate_rule'])", True, 'from neutron.objects.qos import rule as rule_object\n'), ((1711, 17, 1712, 49), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((1770, 17, 1772, 41), 'unittest.mock.patch', 'mock.patch', ({(1770, 28, 1772, 40): '"""neutron.objects.qos.rule.QosMinimumPacketRateRule.get_object"""'}, {}), "('neutron.objects.qos.rule.QosMinimumPacketRateRule.get_object')", False, 'from unittest import mock\n'), ((1781, 17, 1783, 42), 'unittest.mock.patch', 'mock.patch', ({(1781, 28, 1783, 41): '"""neutron.objects.qos.rule.QosMinimumPacketRateRule.get_objects"""'}, {}), "('neutron.objects.qos.rule.QosMinimumPacketRateRule.get_objects')", False, 'from unittest import mock\n'), ((1792, 17, 1794, 42), 'unittest.mock.patch', 'mock.patch', ({(1792, 28, 1794, 41): '"""neutron.objects.qos.rule.QosMinimumPacketRateRule.get_objects"""'}, {}), "('neutron.objects.qos.rule.QosMinimumPacketRateRule.get_objects')", False, 'from unittest import mock\n'), ((1939, 17, 1942, 13), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((1945, 25, 1946, 78), 'unittest.mock.call', 'mock.call', ({(1945, 35, 1945, 43): 'mock.ANY', (1945, 45, 1945, 62): 'rule_object_class', (1945, 64, 1945, 71): 'rule_id', (1946, 35, 1946, 53): 'self.qos_policy_id', (1946, 55, 1946, 77): '{rule_data_name: data}'}, {}), '(mock.ANY, rule_object_class, rule_id, self.qos_policy_id, {\n    rule_data_name: data})', False, 'from unittest import mock\n'), ((1959, 17, 1960, 46), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((1962, 25, 1963, 54), 'unittest.mock.call', 'mock.call', ({(1962, 35, 1962, 43): 'mock.ANY', (1962, 45, 1962, 62): 'rule_object_class', (1962, 64, 1962, 71): 'rule_id', (1963, 35, 1963, 53): 'self.qos_policy_id'}, {}), '(mock.ANY, rule_object_class, rule_id, self.qos_policy_id)', False, 'from unittest import mock\n'), ((1976, 17, 1979, 13), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((1982, 25, 1983, 54), 'unittest.mock.call', 'mock.call', ({(1982, 35, 1982, 43): 'mock.ANY', (1982, 45, 1982, 62): 'rule_object_class', (1982, 64, 1982, 71): 'rule_id', (1983, 35, 1983, 53): 'self.qos_policy_id'}, {}), '(mock.ANY, rule_object_class, rule_id, self.qos_policy_id)', False, 'from unittest import mock\n'), ((1989, 17, 1990, 46), 'unittest.mock.patch', 'mock.patch', (), '', False, 'from unittest import mock\n'), ((2047, 24, 2048, 63), 'unittest.mock.patch', 'mock.patch', ({(2047, 35, 2048, 62): '"""neutron.api.rpc.handlers.resources_rpc.ResourcesPushRpcApi.push"""'}, {}), "('neutron.api.rpc.handlers.resources_rpc.ResourcesPushRpcApi.push')", False, 'from unittest import mock\n'), ((2083, 31, 2083, 71), 'neutron_lib.utils.net.random_mac_generator', 'net_utils.random_mac_generator', ({(2083, 62, 2083, 70): 'base_mac'}, {}), '(base_mac)', True, 'from neutron_lib.utils import net as net_utils\n'), ((188, 31, 188, 56), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ({}, {}), '()', False, 'from oslo_utils import uuidutils\n'), ((197, 31, 197, 56), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ({}, {}), '()', False, 'from oslo_utils import uuidutils\n'), ((505, 41, 505, 66), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ({}, {}), '()', False, 'from oslo_utils import uuidutils\n'), ((558, 24, 560, 54), 'neutron_lib.callbacks.events.DBEventPayload', 'events.DBEventPayload', (), '', False, 'from neutron_lib.callbacks import events\n'), ((621, 24, 623, 57), 'neutron_lib.callbacks.events.DBEventPayload', 'events.DBEventPayload', (), '', False, 'from neutron_lib.callbacks import events\n'), ((1351, 37, 1351, 56), 'unittest.mock.call.RuleCls', 'mock.call.RuleCls', ({}, {}), '()', False, 'from unittest import mock\n'), ((1603, 26, 1603, 51), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ({}, {}), '()', False, 'from oslo_utils import uuidutils\n'), ((1610, 26, 1610, 51), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ({}, {}), '()', False, 'from oslo_utils import uuidutils\n'), ((1688, 26, 1688, 51), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ({}, {}), '()', False, 'from oslo_utils import uuidutils\n'), ((1695, 26, 1695, 51), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ({}, {}), '()', False, 'from oslo_utils import uuidutils\n'), ((2112, 24, 2113, 71), 'neutron_lib.callbacks.events.DBEventPayload', 'events.DBEventPayload', (), '', False, 'from neutron_lib.callbacks import events\n'), ((2145, 24, 2147, 54), 'neutron_lib.callbacks.events.DBEventPayload', 'events.DBEventPayload', (), '', False, 'from neutron_lib.callbacks import events\n'), ((2206, 24, 2207, 58), 'neutron_lib.callbacks.events.DBEventPayload', 'events.DBEventPayload', (), '', False, 'from neutron_lib.callbacks import events\n'), ((2224, 24, 2225, 58), 'neutron_lib.callbacks.events.DBEventPayload', 'events.DBEventPayload', (), '', False, 'from neutron_lib.callbacks import events\n'), ((2242, 24, 2243, 58), 'neutron_lib.callbacks.events.DBEventPayload', 'events.DBEventPayload', (), '', False, 'from neutron_lib.callbacks import events\n'), ((2261, 24, 2262, 58), 'neutron_lib.callbacks.events.DBEventPayload', 'events.DBEventPayload', (), '', False, 'from neutron_lib.callbacks import events\n'), ((2280, 24, 2281, 58), 'neutron_lib.callbacks.events.DBEventPayload', 'events.DBEventPayload', (), '', False, 'from neutron_lib.callbacks import events\n'), ((2296, 24, 2297, 69), 'neutron_lib.callbacks.events.DBEventPayload', 'events.DBEventPayload', (), '', False, 'from neutron_lib.callbacks import events\n'), ((2321, 24, 2322, 69), 'neutron_lib.callbacks.events.DBEventPayload', 'events.DBEventPayload', (), '', False, 'from neutron_lib.callbacks import events\n'), ((2362, 24, 2363, 69), 'neutron_lib.callbacks.events.DBEventPayload', 'events.DBEventPayload', (), '', False, 'from neutron_lib.callbacks import events\n'), ((2427, 24, 2428, 69), 'neutron_lib.callbacks.events.DBEventPayload', 'events.DBEventPayload', (), '', False, 'from neutron_lib.callbacks import events\n'), ((1348, 16, 1348, 52), 'unittest.mock.call.QosPolicy.get_policy_obj', 'mock.call.QosPolicy.get_policy_obj', ({}, {}), '()', False, 'from unittest import mock\n')]