#    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}})

[((30625, 30711), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.rbac_db.RbacNeutronDbObjectMixin.create_rbac_policy"""'], {}), "(\n    'neutron.objects.rbac_db.RbacNeutronDbObjectMixin.create_rbac_policy')\n", (30635, 30711), False, 'from unittest import mock\n'), ((30732, 30782), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy"""'], {}), "('neutron.objects.qos.policy.QosPolicy')\n", (30742, 30782), False, 'from unittest import mock\n'), ((32673, 32729), 'unittest.mock.patch.object', 'mock.patch.object', (['policy_object.QosPolicy', '"""get_object"""'], {}), "(policy_object.QosPolicy, 'get_object')\n", (32690, 32729), False, 'from unittest import mock\n'), ((32735, 32821), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.rbac_db.RbacNeutronDbObjectMixin.create_rbac_policy"""'], {}), "(\n    'neutron.objects.rbac_db.RbacNeutronDbObjectMixin.create_rbac_policy')\n", (32745, 32821), False, 'from unittest import mock\n'), ((32842, 32894), 'unittest.mock.patch.object', 'mock.patch.object', (['policy_object.QosPolicy', '"""update"""'], {}), "(policy_object.QosPolicy, 'update')\n", (32859, 32894), False, 'from unittest import mock\n'), ((34077, 34143), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.db.api.get_object"""'], {'return_value': 'None'}), "('neutron.objects.db.api.get_object', return_value=None)\n", (34087, 34143), False, 'from unittest import mock\n'), ((34149, 34201), 'unittest.mock.patch.object', 'mock.patch.object', (['policy_object.QosPolicy', '"""delete"""'], {}), "(policy_object.QosPolicy, 'delete')\n", (34166, 34201), False, 'from unittest import mock\n'), ((35130, 35186), 'unittest.mock.patch.object', 'mock.patch.object', (['policy_object.QosPolicy', '"""get_object"""'], {}), "(policy_object.QosPolicy, 'get_object')\n", (35147, 35186), False, 'from unittest import mock\n'), ((35192, 35254), 'unittest.mock.patch.object', 'mock.patch.object', (['rule_object.QosBandwidthLimitRule', '"""create"""'], {}), "(rule_object.QosBandwidthLimitRule, 'create')\n", (35209, 35254), False, 'from unittest import mock\n'), ((40048, 40110), 'unittest.mock.patch.object', 'mock.patch.object', (['rule_object.QosBandwidthLimitRule', '"""update"""'], {}), "(rule_object.QosBandwidthLimitRule, 'update')\n", (40065, 40110), False, 'from unittest import mock\n'), ((45336, 45398), 'unittest.mock.patch.object', 'mock.patch.object', (['rule_object.QosBandwidthLimitRule', '"""delete"""'], {}), "(rule_object.QosBandwidthLimitRule, 'delete')\n", (45353, 45398), False, 'from unittest import mock\n'), ((59494, 59534), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.ports.Port"""'], {}), "('neutron.objects.ports.Port')\n", (59504, 59534), False, 'from unittest import mock\n'), ((59540, 59590), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy"""'], {}), "('neutron.objects.qos.policy.QosPolicy')\n", (59550, 59590), False, 'from unittest import mock\n'), ((87381, 87442), 'unittest.mock.patch.object', 'mock.patch.object', (['qos_plugin.QoSPlugin', '"""update_policy_rule"""'], {}), "(qos_plugin.QoSPlugin, 'update_policy_rule')\n", (87398, 87442), False, 'from unittest import mock\n'), ((88466, 88524), 'unittest.mock.patch.object', 'mock.patch.object', (['qos_plugin.QoSPlugin', '"""get_policy_rule"""'], {}), "(qos_plugin.QoSPlugin, 'get_policy_rule')\n", (88483, 88524), False, 'from unittest import mock\n'), ((89371, 89432), 'unittest.mock.patch.object', 'mock.patch.object', (['qos_plugin.QoSPlugin', '"""delete_policy_rule"""'], {}), "(qos_plugin.QoSPlugin, 'delete_policy_rule')\n", (89388, 89432), False, 'from unittest import mock\n'), ((2511, 2575), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.obj_load_attr"""'], {}), "('neutron.objects.qos.policy.QosPolicy.obj_load_attr')\n", (2521, 2575), False, 'from unittest import mock\n'), ((3050, 3103), 'oslo_config.cfg.CONF.set_override', 'cfg.CONF.set_override', (['"""core_plugin"""', 'DB_PLUGIN_KLASS'], {}), "('core_plugin', DB_PLUGIN_KLASS)\n", (3071, 3103), False, 'from oslo_config import cfg\n'), ((3112, 3161), 'oslo_config.cfg.CONF.set_override', 'cfg.CONF.set_override', (['"""service_plugins"""', "['qos']"], {}), "('service_plugins', ['qos'])\n", (3133, 3161), False, 'from oslo_config import cfg\n'), ((3171, 3185), 'neutron.manager.init', 'manager.init', ([], {}), '()\n', (3183, 3185), False, 'from neutron import manager\n'), ((3212, 3255), 'neutron_lib.plugins.directory.get_plugin', 'directory.get_plugin', (['plugins_constants.QOS'], {}), '(plugins_constants.QOS)\n', (3232, 3255), False, 'from neutron_lib.plugins import directory\n'), ((3298, 3309), 'unittest.mock.Mock', 'mock.Mock', ([], {}), '()\n', (3307, 3309), False, 'from unittest import mock\n'), ((3480, 3523), 'neutron_lib.context.Context', 'context.Context', (['"""fake_user"""', '"""fake_tenant"""'], {}), "('fake_user', 'fake_tenant')\n", (3495, 3523), False, 'from neutron_lib import context\n'), ((3550, 3577), 'neutron_lib.context.get_admin_context', 'context.get_admin_context', ([], {}), '()\n', (3575, 3577), False, 'from neutron_lib import context\n'), ((4731, 4795), 'neutron.objects.qos.policy.QosPolicy', 'policy_object.QosPolicy', (['self.ctxt'], {}), "(self.ctxt, **self.policy_data['policy'])\n", (4754, 4795), True, 'from neutron.objects.qos import policy as policy_object\n'), ((4830, 4921), 'neutron.objects.qos.rule.QosBandwidthLimitRule', 'rule_object.QosBandwidthLimitRule', (['self.ctxt'], {}), "(self.ctxt, **self.rule_data[\n    'bandwidth_limit_rule'])\n", (4863, 4921), True, 'from neutron.objects.qos import rule as rule_object\n'), ((4956, 5041), 'neutron.objects.qos.rule.QosDscpMarkingRule', 'rule_object.QosDscpMarkingRule', (['self.ctxt'], {}), "(self.ctxt, **self.rule_data['dscp_marking_rule']\n    )\n", (4986, 5041), True, 'from neutron.objects.qos import rule as rule_object\n'), ((5078, 5173), 'neutron.objects.qos.rule.QosMinimumBandwidthRule', 'rule_object.QosMinimumBandwidthRule', (['self.ctxt'], {}), "(self.ctxt, **self.rule_data[\n    'minimum_bandwidth_rule'])\n", (5113, 5173), True, 'from neutron.objects.qos import rule as rule_object\n'), ((5207, 5301), 'neutron.objects.qos.rule.QosPacketRateLimitRule', 'rule_object.QosPacketRateLimitRule', (['self.ctxt'], {}), "(self.ctxt, **self.rule_data[\n    'packet_rate_limit_rule'])\n", (5241, 5301), True, 'from neutron.objects.qos import rule as rule_object\n'), ((5339, 5437), 'neutron.objects.qos.rule.QosMinimumPacketRateRule', 'rule_object.QosMinimumPacketRateRule', (['self.ctxt'], {}), "(self.ctxt, **self.rule_data[\n    'minimum_packet_rate_rule'])\n", (5375, 5437), True, 'from neutron.objects.qos import rule as rule_object\n'), ((6100, 6125), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ([], {}), '()\n', (6123, 6125), False, 'from oslo_utils import uuidutils\n'), ((6493, 6547), 'neutron.objects.ports.Port', 'ports_object.Port', (['self.ctxt'], {}), "(self.ctxt, **self.port_data['port'])\n", (6510, 6547), True, 'from neutron.objects import ports as ports_object\n'), ((6636, 6708), 'unittest.mock.MagicMock', 'mock.MagicMock', ([], {'network_id': 'network_id', 'physical_network': 'physical_network'}), '(network_id=network_id, physical_network=physical_network)\n', (6650, 6708), False, 'from unittest import mock\n'), ((7804, 7829), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ([], {}), '()\n', (7827, 7829), False, 'from oslo_utils import uuidutils\n'), ((8443, 8515), 'unittest.mock.MagicMock', 'mock.MagicMock', ([], {'network_id': 'network_id', 'physical_network': 'physical_network'}), '(network_id=network_id, physical_network=physical_network)\n', (8457, 8515), False, 'from unittest import mock\n'), ((11710, 11784), 'neutron.objects.qos.rule.QosMinimumBandwidthRule', 'rule_object.QosMinimumBandwidthRule', (['self.ctxt'], {}), '(self.ctxt, **min_bw_rule_ingress_data)\n', (11745, 11784), True, 'from neutron.objects.qos import rule as rule_object\n'), ((11829, 11905), 'neutron.objects.qos.rule.QosMinimumPacketRateRule', 'rule_object.QosMinimumPacketRateRule', (['self.ctxt'], {}), '(self.ctxt, **min_pps_rule_ingress_data)\n', (11865, 11905), True, 'from neutron.objects.qos import rule as rule_object\n'), ((17244, 17318), 'neutron.objects.qos.rule.QosMinimumBandwidthRule', 'rule_object.QosMinimumBandwidthRule', (['self.ctxt'], {}), '(self.ctxt, **min_bw_rule_ingress_data)\n', (17279, 17318), True, 'from neutron.objects.qos import rule as rule_object\n'), ((17363, 17439), 'neutron.objects.qos.rule.QosMinimumPacketRateRule', 'rule_object.QosMinimumPacketRateRule', (['self.ctxt'], {}), '(self.ctxt, **min_pps_rule_ingress_data)\n', (17399, 17439), True, 'from neutron.objects.qos import rule as rule_object\n'), ((21358, 21383), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ([], {}), '()\n', (21381, 21383), False, 'from oslo_utils import uuidutils\n'), ((21690, 21741), 'unittest.mock.MagicMock', 'mock.MagicMock', ([], {'id': 'port_id', 'qos_policy_id': 'policy_id'}), '(id=port_id, qos_policy_id=policy_id)\n', (21704, 21741), False, 'from unittest import mock\n'), ((21764, 21792), 'unittest.mock.MagicMock', 'mock.MagicMock', ([], {'id': 'policy_id'}), '(id=policy_id)\n', (21778, 21792), False, 'from unittest import mock\n'), ((21814, 21825), 'unittest.mock.Mock', 'mock.Mock', ([], {}), '()\n', (21823, 21825), False, 'from unittest import mock\n'), ((23350, 23375), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ([], {}), '()\n', (23373, 23375), False, 'from oslo_utils import uuidutils\n'), ((23843, 23868), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ([], {}), '()\n', (23866, 23868), False, 'from oslo_utils import uuidutils\n'), ((24561, 24589), 'unittest.mock.MagicMock', 'mock.MagicMock', ([], {'id': 'policy_id'}), '(id=policy_id)\n', (24575, 24589), False, 'from unittest import mock\n'), ((24611, 24622), 'unittest.mock.Mock', 'mock.Mock', ([], {}), '()\n', (24620, 24622), False, 'from unittest import mock\n'), ((26438, 26463), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ([], {}), '()\n', (26461, 26463), False, 'from oslo_utils import uuidutils\n'), ((27871, 27896), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ([], {}), '()\n', (27894, 27896), False, 'from oslo_utils import uuidutils\n'), ((28553, 28615), 'neutron.objects.network.NetworkSegment', 'network_object.NetworkSegment', ([], {'physical_network': '"""fake physnet"""'}), "(physical_network='fake physnet')\n", (28582, 28615), True, 'from neutron.objects import network as network_object\n'), ((28643, 28696), 'neutron.objects.network.Network', 'network_object.Network', (['self.ctxt'], {'segments': '[segment]'}), '(self.ctxt, segments=[segment])\n', (28665, 28696), True, 'from neutron.objects import network as network_object\n'), ((29660, 29722), 'neutron.objects.network.NetworkSegment', 'network_object.NetworkSegment', ([], {'physical_network': '"""fake physnet"""'}), "(physical_network='fake physnet')\n", (29689, 29722), True, 'from neutron.objects import network as network_object\n'), ((29750, 29803), 'neutron.objects.network.Network', 'network_object.Network', (['self.ctxt'], {'segments': '[segment]'}), '(self.ctxt, segments=[segment])\n', (29772, 29803), True, 'from neutron.objects import network as network_object\n'), ((30879, 30890), 'unittest.mock.Mock', 'mock.Mock', ([], {}), '()\n', (30888, 30890), False, 'from unittest import mock\n'), ((31226, 31295), 'unittest.mock.call.driver.call', 'mock.call.driver.call', (['"""create_policy_precommit"""', 'self.ctxt', 'mock.ANY'], {}), "('create_policy_precommit', self.ctxt, mock.ANY)\n", (31247, 31295), False, 'from unittest import mock\n'), ((31336, 31395), 'unittest.mock.call.driver.call', 'mock.call.driver.call', (['"""create_policy"""', 'self.ctxt', 'mock.ANY'], {}), "('create_policy', self.ctxt, mock.ANY)\n", (31357, 31395), False, 'from unittest import mock\n'), ((31715, 31740), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ([], {}), '()\n', (31738, 31740), False, 'from oslo_utils import uuidutils\n'), ((31762, 31787), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ([], {}), '()\n', (31785, 31787), False, 'from oslo_utils import uuidutils\n'), ((33104, 33115), 'unittest.mock.Mock', 'mock.Mock', ([], {}), '()\n', (33113, 33115), False, 'from unittest import mock\n'), ((33310, 33398), 'neutron_lib.objects.utils.get_updatable_fields', 'obj_utils.get_updatable_fields', (['policy_object.QosPolicy', "self.policy_data['policy']"], {}), "(policy_object.QosPolicy, self.policy_data[\n    'policy'])\n", (33340, 33398), True, 'from neutron_lib.objects import utils as obj_utils\n'), ((33605, 33623), 'unittest.mock.call.update', 'mock.call.update', ([], {}), '()\n', (33621, 33623), False, 'from unittest import mock\n'), ((33661, 33730), 'unittest.mock.call.driver.call', 'mock.call.driver.call', (['"""update_policy_precommit"""', 'self.ctxt', 'mock.ANY'], {}), "('update_policy_precommit', self.ctxt, mock.ANY)\n", (33682, 33730), False, 'from unittest import mock\n'), ((33771, 33830), 'unittest.mock.call.driver.call', 'mock.call.driver.call', (['"""update_policy"""', 'self.ctxt', 'mock.ANY'], {}), "('update_policy', self.ctxt, mock.ANY)\n", (33792, 33830), False, 'from unittest import mock\n'), ((34304, 34315), 'unittest.mock.Mock', 'mock.Mock', ([], {}), '()\n', (34313, 34315), False, 'from unittest import mock\n'), ((34658, 34676), 'unittest.mock.call.delete', 'mock.call.delete', ([], {}), '()\n', (34674, 34676), False, 'from unittest import mock\n'), ((34714, 34783), 'unittest.mock.call.driver.call', 'mock.call.driver.call', (['"""delete_policy_precommit"""', 'self.ctxt', 'mock.ANY'], {}), "('delete_policy_precommit', self.ctxt, mock.ANY)\n", (34735, 34783), False, 'from unittest import mock\n'), ((34824, 34883), 'unittest.mock.call.driver.call', 'mock.call.driver.call', (['"""delete_policy"""', 'self.ctxt', 'mock.ANY'], {}), "('delete_policy', self.ctxt, mock.ANY)\n", (34845, 34883), False, 'from unittest import mock\n'), ((35387, 35409), 'copy.copy', 'copy.copy', (['self.policy'], {}), '(self.policy)\n', (35396, 35409), False, 'import copy\n'), ((35522, 35533), 'unittest.mock.Mock', 'mock.Mock', ([], {}), '()\n', (35531, 35533), False, 'from unittest import mock\n'), ((40195, 40206), 'unittest.mock.Mock', 'mock.Mock', ([], {}), '()\n', (40204, 40206), False, 'from unittest import mock\n'), ((40400, 40464), 'neutron.objects.qos.policy.QosPolicy', 'policy_object.QosPolicy', (['self.ctxt'], {}), "(self.ctxt, **self.policy_data['policy'])\n", (40423, 40464), True, 'from neutron.objects.qos import policy as policy_object\n'), ((44712, 44776), 'neutron.objects.qos.policy.QosPolicy', 'policy_object.QosPolicy', (['self.ctxt'], {}), "(self.ctxt, **self.policy_data['policy'])\n", (44735, 44776), True, 'from neutron.objects.qos import policy as policy_object\n'), ((44859, 44923), 'neutron.objects.qos.policy.QosPolicy', 'policy_object.QosPolicy', (['self.ctxt'], {}), "(self.ctxt, **self.policy_data['policy'])\n", (44882, 44923), True, 'from neutron.objects.qos import policy as policy_object\n'), ((45483, 45494), 'unittest.mock.Mock', 'mock.Mock', ([], {}), '()\n', (45492, 45494), False, 'from unittest import mock\n'), ((45688, 45752), 'neutron.objects.qos.policy.QosPolicy', 'policy_object.QosPolicy', (['self.ctxt'], {}), "(self.ctxt, **self.policy_data['policy'])\n", (45711, 45752), True, 'from neutron.objects.qos import policy as policy_object\n'), ((46735, 46799), 'neutron.objects.qos.policy.QosPolicy', 'policy_object.QosPolicy', (['self.ctxt'], {}), "(self.ctxt, **self.policy_data['policy'])\n", (46758, 46799), True, 'from neutron.objects.qos import policy as policy_object\n'), ((50293, 50357), 'neutron.objects.qos.policy.QosPolicy', 'policy_object.QosPolicy', (['self.ctxt'], {}), "(self.ctxt, **self.policy_data['policy'])\n", (50316, 50357), True, 'from neutron.objects.qos import policy as policy_object\n'), ((50809, 50873), 'neutron.objects.qos.policy.QosPolicy', 'policy_object.QosPolicy', (['self.ctxt'], {}), "(self.ctxt, **self.policy_data['policy'])\n", (50832, 50873), True, 'from neutron.objects.qos import policy as policy_object\n'), ((51344, 51408), 'neutron.objects.qos.policy.QosPolicy', 'policy_object.QosPolicy', (['self.ctxt'], {}), "(self.ctxt, **self.policy_data['policy'])\n", (51367, 51408), True, 'from neutron.objects.qos import policy as policy_object\n'), ((58010, 58037), 'neutron_lib.context.get_admin_context', 'context.get_admin_context', ([], {}), '()\n', (58035, 58037), False, 'from neutron_lib import context\n'), ((59752, 59763), 'unittest.mock.Mock', 'mock.Mock', ([], {}), '()\n', (59761, 59763), False, 'from unittest import mock\n'), ((60270, 60281), 'unittest.mock.Mock', 'mock.Mock', ([], {}), '()\n', (60279, 60281), False, 'from unittest import mock\n'), ((61634, 61698), 'neutron.objects.qos.policy.QosPolicy', 'policy_object.QosPolicy', (['self.ctxt'], {}), "(self.ctxt, **self.policy_data['policy'])\n", (61657, 61698), True, 'from neutron.objects.qos import policy as policy_object\n'), ((62896, 62960), 'neutron.objects.qos.policy.QosPolicy', 'policy_object.QosPolicy', (['self.ctxt'], {}), "(self.ctxt, **self.policy_data['policy'])\n", (62919, 62960), True, 'from neutron.objects.qos import policy as policy_object\n'), ((63436, 63500), 'neutron.objects.qos.policy.QosPolicy', 'policy_object.QosPolicy', (['self.ctxt'], {}), "(self.ctxt, **self.policy_data['policy'])\n", (63459, 63500), True, 'from neutron.objects.qos import policy as policy_object\n'), ((63989, 64053), 'neutron.objects.qos.policy.QosPolicy', 'policy_object.QosPolicy', (['self.ctxt'], {}), "(self.ctxt, **self.policy_data['policy'])\n", (64012, 64053), True, 'from neutron.objects.qos import policy as policy_object\n'), ((64513, 64577), 'neutron.objects.qos.policy.QosPolicy', 'policy_object.QosPolicy', (['self.ctxt'], {}), "(self.ctxt, **self.policy_data['policy'])\n", (64536, 64577), True, 'from neutron.objects.qos import policy as policy_object\n'), ((68988, 69015), 'neutron_lib.context.get_admin_context', 'context.get_admin_context', ([], {}), '()\n', (69013, 69015), False, 'from neutron_lib import context\n'), ((70200, 70262), 'neutron.objects.network.NetworkSegment', 'network_object.NetworkSegment', ([], {'physical_network': '"""fake physnet"""'}), "(physical_network='fake physnet')\n", (70229, 70262), True, 'from neutron.objects import network as network_object\n'), ((70290, 70343), 'neutron.objects.network.Network', 'network_object.Network', (['self.ctxt'], {'segments': '[segment]'}), '(self.ctxt, segments=[segment])\n', (70312, 70343), True, 'from neutron.objects import network as network_object\n'), ((71326, 71388), 'neutron.objects.network.NetworkSegment', 'network_object.NetworkSegment', ([], {'physical_network': '"""fake physnet"""'}), "(physical_network='fake physnet')\n", (71355, 71388), True, 'from neutron.objects import network as network_object\n'), ((71416, 71469), 'neutron.objects.network.Network', 'network_object.Network', (['self.ctxt'], {'segments': '[segment]'}), '(self.ctxt, segments=[segment])\n', (71438, 71469), True, 'from neutron.objects import network as network_object\n'), ((83074, 83101), 'neutron_lib.context.get_admin_context', 'context.get_admin_context', ([], {}), '()\n', (83099, 83101), False, 'from neutron_lib import context\n'), ((84226, 84273), 'neutron.extensions.qos_rules_alias.Qos_rules_alias.get_resources', 'qos_rules_alias.Qos_rules_alias.get_resources', ([], {}), '()\n', (84271, 84273), False, 'from neutron.extensions import qos_rules_alias\n'), ((84489, 84558), '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', ([], {}), '()\n', (84556, 84558), False, 'from neutron.extensions import qos_pps_minimum_rule_alias\n'), ((84859, 84927), 'unittest.mock.patch', 'mock.patch', (['"""neutron.notifiers.batch_notifier.BatchNotifier._notify"""'], {}), "('neutron.notifiers.batch_notifier.BatchNotifier._notify')\n", (84869, 84927), False, 'from unittest import mock\n'), ((85300, 85343), 'neutron_lib.plugins.directory.get_plugin', 'directory.get_plugin', (['plugins_constants.QOS'], {}), '(plugins_constants.QOS)\n', (85320, 85343), False, 'from neutron_lib.plugins import directory\n'), ((85365, 85408), 'neutron_lib.context.Context', 'context.Context', (['"""fake_user"""', '"""fake_tenant"""'], {}), "('fake_user', 'fake_tenant')\n", (85380, 85408), False, 'from neutron_lib import context\n'), ((85674, 85699), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ([], {}), '()\n', (85697, 85699), False, 'from oslo_utils import uuidutils\n'), ((91223, 91291), 'unittest.mock.patch', 'mock.patch', (['"""neutron.notifiers.batch_notifier.BatchNotifier._notify"""'], {}), "('neutron.notifiers.batch_notifier.BatchNotifier._notify')\n", (91233, 91291), False, 'from unittest import mock\n'), ((91681, 91724), 'neutron_lib.plugins.directory.get_plugin', 'directory.get_plugin', (['plugins_constants.QOS'], {}), '(plugins_constants.QOS)\n', (91701, 91724), False, 'from neutron_lib.plugins import directory\n'), ((91746, 91789), 'neutron_lib.context.Context', 'context.Context', (['"""fake_user"""', '"""fake_tenant"""'], {}), "('fake_user', 'fake_tenant')\n", (91761, 91789), False, 'from neutron_lib import context\n'), ((91932, 91957), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ([], {}), '()\n', (91955, 91957), False, 'from oslo_utils import uuidutils\n'), ((92873, 92926), 'oslo_config.cfg.CONF.set_override', 'cfg.CONF.set_override', (['"""core_plugin"""', 'DB_PLUGIN_KLASS'], {}), "('core_plugin', DB_PLUGIN_KLASS)\n", (92894, 92926), False, 'from oslo_config import cfg\n'), ((92935, 92984), 'oslo_config.cfg.CONF.set_override', 'cfg.CONF.set_override', (['"""service_plugins"""', "['qos']"], {}), "('service_plugins', ['qos'])\n", (92956, 92984), False, 'from oslo_config import cfg\n'), ((92994, 93008), 'neutron.manager.init', 'manager.init', ([], {}), '()\n', (93006, 93008), False, 'from neutron import manager\n'), ((93035, 93078), 'neutron_lib.plugins.directory.get_plugin', 'directory.get_plugin', (['plugins_constants.QOS'], {}), '(plugins_constants.QOS)\n', (93055, 93078), False, 'from neutron_lib.plugins import directory\n'), ((93120, 93131), 'unittest.mock.Mock', 'mock.Mock', ([], {}), '()\n', (93129, 93131), False, 'from unittest import mock\n'), ((93303, 93330), 'neutron_lib.context.get_admin_context', 'context.get_admin_context', ([], {}), '()\n', (93328, 93330), False, 'from neutron_lib import context\n'), ((93357, 93382), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ([], {}), '()\n', (93380, 93382), False, 'from oslo_utils import uuidutils\n'), ((93437, 93539), 'neutron.objects.qos.policy.QosPolicy', 'policy_object.QosPolicy', (['self.context'], {'project_id': 'self.project_id', 'shared': '(False)', 'is_default': '(False)'}), '(self.context, project_id=self.project_id, shared=\n    False, is_default=False)\n', (93460, 93539), True, 'from neutron.objects.qos import policy as policy_object\n'), ((93828, 93992), 'neutron.objects.qos.rule.QosMinimumBandwidthRule', 'rule_object.QosMinimumBandwidthRule', (['self.context'], {'project_id': 'self.project_id', 'qos_policy_id': 'policy_id', 'direction': 'direction', 'min_kbps': 'min_kbps', 'id': 'rule_id'}), '(self.context, project_id=self.\n    project_id, qos_policy_id=policy_id, direction=direction, min_kbps=\n    min_kbps, id=rule_id)\n', (93863, 93992), True, 'from neutron.objects.qos import rule as rule_object\n'), ((94286, 94451), 'neutron.objects.qos.rule.QosMinimumPacketRateRule', 'rule_object.QosMinimumPacketRateRule', (['self.context'], {'project_id': 'self.project_id', 'qos_policy_id': 'policy_id', 'direction': 'direction', 'min_kpps': 'min_kpps', 'id': 'rule_id'}), '(self.context, project_id=self.\n    project_id, qos_policy_id=policy_id, direction=direction, min_kpps=\n    min_kpps, id=rule_id)\n', (94322, 94451), True, 'from neutron.objects.qos import rule as rule_object\n'), ((94942, 95219), 'neutron.objects.ports.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'}), "(self.context, network_id=network_id, device_owner=\n    device_owner, project_id=self.project_id, admin_state_up=True, status=\n    'DOWN', device_id='2', qos_policy_id=qos_policy_id,\n    qos_network_policy_id=qos_network_policy_id, mac_address=mac, id=port_id)\n", (94959, 95219), True, 'from neutron.objects import ports as ports_object\n'), ((95377, 95442), 'neutron.objects.network.Network', 'network_object.Network', (['self.context'], {'qos_policy_id': 'qos_policy_id'}), '(self.context, qos_policy_id=qos_policy_id)\n', (95399, 95442), True, 'from neutron.objects import network as network_object\n'), ((103983, 103999), 'unittest.mock.MagicMock', 'mock.MagicMock', ([], {}), '()\n', (103997, 103999), False, 'from unittest import mock\n'), ((105180, 105196), 'unittest.mock.MagicMock', 'mock.MagicMock', ([], {}), '()\n', (105194, 105196), False, 'from unittest import mock\n'), ((105943, 105959), 'unittest.mock.MagicMock', 'mock.MagicMock', ([], {}), '()\n', (105957, 105959), False, 'from unittest import mock\n'), ((106467, 106648), 'neutron.objects.ports.PortBinding', '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'}"}), "(self.context, port_id=port1.id, host='fake_host1',\n    vnic_type='fake_vnic_type', vif_type='fake_vif_type', profile={\n    'allocation': 'fake_allocation'})\n", (106491, 106648), True, 'from neutron.objects import ports as ports_object\n'), ((108317, 108333), 'unittest.mock.MagicMock', 'mock.MagicMock', ([], {}), '()\n', (108331, 108333), False, 'from unittest import mock\n'), ((108841, 108986), 'neutron.objects.ports.PortBinding', 'ports_object.PortBinding', (['self.context'], {'port_id': 'port1.id', 'host': '"""fake_host1"""', 'vnic_type': '"""fake_vnic_type"""', 'vif_type': '"""fake_vif_type"""', 'profile': '{}'}), "(self.context, port_id=port1.id, host='fake_host1',\n    vnic_type='fake_vnic_type', vif_type='fake_vif_type', profile={})\n", (108865, 108986), True, 'from neutron.objects import ports as ports_object\n'), ((109258, 109403), 'neutron.objects.ports.PortBinding', 'ports_object.PortBinding', (['self.context'], {'port_id': 'port2.id', 'host': '"""fake_host2"""', 'vnic_type': '"""fake_vnic_type"""', 'vif_type': '"""fake_vif_type"""', 'profile': '{}'}), "(self.context, port_id=port2.id, host='fake_host2',\n    vnic_type='fake_vnic_type', vif_type='fake_vif_type', profile={})\n", (109282, 109403), True, 'from neutron.objects import ports as ports_object\n'), ((123346, 123390), 'neutron.objects.qos.rule.QosDscpMarkingRule', 'rule_object.QosDscpMarkingRule', ([], {'dscp_mark': '(16)'}), '(dscp_mark=16)\n', (123376, 123390), True, 'from neutron.objects.qos import rule as rule_object\n'), ((123416, 123460), 'neutron.objects.qos.rule.QosDscpMarkingRule', 'rule_object.QosDscpMarkingRule', ([], {'dscp_mark': '(18)'}), '(dscp_mark=18)\n', (123446, 123460), True, 'from neutron.objects.qos import rule as rule_object\n'), ((124251, 124295), 'neutron.objects.qos.rule.QosDscpMarkingRule', 'rule_object.QosDscpMarkingRule', ([], {'dscp_mark': '(16)'}), '(dscp_mark=16)\n', (124281, 124295), True, 'from neutron.objects.qos import rule as rule_object\n'), ((124908, 124952), 'neutron.objects.qos.rule.QosDscpMarkingRule', 'rule_object.QosDscpMarkingRule', ([], {'dscp_mark': '(16)'}), '(dscp_mark=16)\n', (124938, 124952), True, 'from neutron.objects.qos import rule as rule_object\n'), ((6824, 6921), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.network.NetworkSegment.get_objects"""'], {'return_value': '[segment_mock]'}), "('neutron.objects.network.NetworkSegment.get_objects',\n    return_value=[segment_mock])\n", (6834, 6921), False, 'from unittest import mock\n'), ((6961, 7066), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.rule.QosMinimumBandwidthRule.get_objects"""'], {'return_value': 'min_bw_rules'}), "('neutron.objects.qos.rule.QosMinimumBandwidthRule.get_objects',\n    return_value=min_bw_rules)\n", (6971, 7066), False, 'from unittest import mock\n'), ((7146, 7253), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.rule.QosMinimumPacketRateRule.get_objects"""'], {'return_value': 'min_pps_rules'}), "('neutron.objects.qos.rule.QosMinimumPacketRateRule.get_objects',\n    return_value=min_pps_rules)\n", (7156, 7253), False, 'from unittest import mock\n'), ((7333, 7422), 'unittest.mock.patch', 'mock.patch', (['"""uuid.uuid5"""'], {'return_value': '"""fake_uuid"""', 'side_effect': 'request_groups_uuids'}), "('uuid.uuid5', return_value='fake_uuid', side_effect=\n    request_groups_uuids)\n", (7343, 7422), False, 'from unittest import mock\n'), ((7499, 7570), 'neutron.services.qos.qos_plugin.QoSPlugin._extend_port_resource_request', 'qos_plugin.QoSPlugin._extend_port_resource_request', (['port_res', 'self.port'], {}), '(port_res, self.port)\n', (7549, 7570), False, 'from neutron.services.qos import qos_plugin\n'), ((8631, 8728), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.network.NetworkSegment.get_objects"""'], {'return_value': '[segment_mock]'}), "('neutron.objects.network.NetworkSegment.get_objects',\n    return_value=[segment_mock])\n", (8641, 8728), False, 'from unittest import mock\n'), ((8768, 8873), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.rule.QosMinimumBandwidthRule.get_objects"""'], {'return_value': 'min_bw_rules'}), "('neutron.objects.qos.rule.QosMinimumBandwidthRule.get_objects',\n    return_value=min_bw_rules)\n", (8778, 8873), False, 'from unittest import mock\n'), ((8953, 9060), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.rule.QosMinimumPacketRateRule.get_objects"""'], {'return_value': 'min_pps_rules'}), "('neutron.objects.qos.rule.QosMinimumPacketRateRule.get_objects',\n    return_value=min_pps_rules)\n", (8963, 9060), False, 'from unittest import mock\n'), ((9140, 9229), 'unittest.mock.patch', 'mock.patch', (['"""uuid.uuid5"""'], {'return_value': '"""fake_uuid"""', 'side_effect': 'request_groups_uuids'}), "('uuid.uuid5', return_value='fake_uuid', side_effect=\n    request_groups_uuids)\n", (9150, 9229), False, 'from unittest import mock\n'), ((9306, 9378), 'neutron.services.qos.qos_plugin.QoSPlugin._extend_port_resource_request_bulk', 'qos_plugin.QoSPlugin._extend_port_resource_request_bulk', (['ports_res', 'None'], {}), '(ports_res, None)\n', (9361, 9378), False, 'from neutron.services.qos import qos_plugin\n'), ((11398, 11423), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ([], {}), '()\n', (11421, 11423), False, 'from oslo_utils import uuidutils\n'), ((11567, 11592), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ([], {}), '()\n', (11590, 11592), False, 'from oslo_utils import uuidutils\n'), ((16932, 16957), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ([], {}), '()\n', (16955, 16957), False, 'from oslo_utils import uuidutils\n'), ((17101, 17126), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ([], {}), '()\n', (17124, 17126), False, 'from oslo_utils import uuidutils\n'), ((20221, 20255), 'unittest.mock.MagicMock', 'mock.MagicMock', ([], {'qos_policy_id': 'None'}), '(qos_policy_id=None)\n', (20235, 20255), False, 'from unittest import mock\n'), ((20338, 20372), 'unittest.mock.MagicMock', 'mock.MagicMock', ([], {'qos_policy_id': 'None'}), '(qos_policy_id=None)\n', (20352, 20372), False, 'from unittest import mock\n'), ((20413, 20457), 'unittest.mock.MagicMock', 'mock.MagicMock', ([], {'qos_policy_id': 'self.policy.id'}), '(qos_policy_id=self.policy.id)\n', (20427, 20457), False, 'from unittest import mock\n'), ((20647, 20740), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.ports.Port.get_objects"""'], {'side_effect': '[network_ports, ports]'}), "('neutron.objects.ports.Port.get_objects', side_effect=[\n    network_ports, ports])\n", (20657, 20740), False, 'from unittest import mock\n'), ((20771, 20823), 'unittest.mock.patch.object', 'mock.patch.object', (['self.policy', '"""get_bound_networks"""'], {}), "(self.policy, 'get_bound_networks')\n", (20788, 20823), False, 'from unittest import mock\n'), ((20847, 20896), 'unittest.mock.patch.object', 'mock.patch.object', (['self.policy', '"""get_bound_ports"""'], {}), "(self.policy, 'get_bound_ports')\n", (20864, 20896), False, 'from unittest import mock\n'), ((21839, 21914), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.ports.Port.get_object"""'], {'return_value': 'port_mock'}), "('neutron.objects.ports.Port.get_object', return_value=port_mock)\n", (21849, 21914), False, 'from unittest import mock\n'), ((21962, 22054), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': 'policy_mock'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=\n    policy_mock)\n", (21972, 22054), False, 'from unittest import mock\n'), ((22099, 22161), 'unittest.mock.patch.object', 'mock.patch.object', (['self.qos_plugin', '"""validate_policy_for_port"""'], {}), "(self.qos_plugin, 'validate_policy_for_port')\n", (22116, 22161), False, 'from unittest import mock\n'), ((22213, 22278), 'unittest.mock.patch.object', 'mock.patch.object', (['self.ctxt', '"""elevated"""'], {'return_value': 'admin_ctxt'}), "(self.ctxt, 'elevated', return_value=admin_ctxt)\n", (22230, 22278), False, 'from unittest import mock\n'), ((24636, 24708), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.ports.Port.get_objects"""'], {'return_value': 'ports'}), "('neutron.objects.ports.Port.get_objects', return_value=ports)\n", (24646, 24708), False, 'from unittest import mock\n'), ((24757, 24849), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': 'policy_mock'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=\n    policy_mock)\n", (24767, 24849), False, 'from unittest import mock\n'), ((24894, 24959), 'unittest.mock.patch.object', 'mock.patch.object', (['self.qos_plugin', '"""validate_policy_for_network"""'], {}), "(self.qos_plugin, 'validate_policy_for_network')\n", (24911, 24959), False, 'from unittest import mock\n'), ((25014, 25077), 'unittest.mock.patch.object', 'mock.patch.object', (['self.qos_plugin', '"""validate_policy_for_ports"""'], {}), "(self.qos_plugin, 'validate_policy_for_ports')\n", (25031, 25077), False, 'from unittest import mock\n'), ((25130, 25195), 'unittest.mock.patch.object', 'mock.patch.object', (['self.ctxt', '"""elevated"""'], {'return_value': 'admin_ctxt'}), "(self.ctxt, 'elevated', return_value=admin_ctxt)\n", (25147, 25195), False, 'from unittest import mock\n'), ((26859, 26884), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ([], {}), '()\n', (26882, 26884), False, 'from oslo_utils import uuidutils\n'), ((26899, 26998), 'unittest.mock.patch.object', 'mock.patch.object', (['self.qos_plugin.driver_manager', '"""validate_rule_for_port"""'], {'return_value': '(False)'}), "(self.qos_plugin.driver_manager, 'validate_rule_for_port',\n    return_value=False)\n", (26916, 26998), False, 'from unittest import mock\n'), ((27338, 27363), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ([], {}), '()\n', (27361, 27363), False, 'from oslo_utils import uuidutils\n'), ((27378, 27476), 'unittest.mock.patch.object', 'mock.patch.object', (['self.qos_plugin.driver_manager', '"""validate_rule_for_port"""'], {'return_value': '(True)'}), "(self.qos_plugin.driver_manager, 'validate_rule_for_port',\n    return_value=True)\n", (27395, 27476), False, 'from unittest import mock\n'), ((27910, 28011), 'unittest.mock.patch.object', 'mock.patch.object', (['self.qos_plugin.driver_manager', '"""validate_rule_for_network"""'], {'return_value': '(True)'}), "(self.qos_plugin.driver_manager,\n    'validate_rule_for_network', return_value=True)\n", (27927, 28011), False, 'from unittest import mock\n'), ((28930, 29017), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': 'policy'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=\n    policy)\n", (28940, 29017), False, 'from unittest import mock\n'), ((29061, 29135), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.network.Network.get_object"""'], {'return_value': 'net'}), "('neutron.objects.network.Network.get_object', return_value=net)\n", (29071, 29135), False, 'from unittest import mock\n'), ((29184, 29270), 'unittest.mock.patch.object', 'mock.patch.object', (['self.qos_plugin', '"""_get_ports_with_policy"""'], {'return_value': '[port]'}), "(self.qos_plugin, '_get_ports_with_policy', return_value=[\n    port])\n", (29201, 29270), False, 'from unittest import mock\n'), ((30020, 30107), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': 'policy'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=\n    policy)\n", (30030, 30107), False, 'from unittest import mock\n'), ((30151, 30225), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.network.Network.get_object"""'], {'return_value': 'net'}), "('neutron.objects.network.Network.get_object', return_value=net)\n", (30161, 30225), False, 'from unittest import mock\n'), ((30274, 30360), 'unittest.mock.patch.object', 'mock.patch.object', (['self.qos_plugin', '"""_get_ports_with_policy"""'], {'return_value': '[port]'}), "(self.qos_plugin, '_get_ports_with_policy', return_value=[\n    port])\n", (30291, 30360), False, 'from unittest import mock\n'), ((32462, 32512), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy"""'], {}), "('neutron.objects.qos.policy.QosPolicy')\n", (32472, 32512), False, 'from unittest import mock\n'), ((35721, 35833), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.qos_policy_validator.check_bandwidth_rule_conflict"""'], {'return_value': 'None'}), "(\n    'neutron.objects.qos.qos_policy_validator.check_bandwidth_rule_conflict',\n    return_value=None)\n", (35731, 35833), False, 'from unittest import mock\n'), ((35895, 36005), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.qos_policy_validator.check_min_pps_rule_conflict"""'], {'return_value': 'None'}), "(\n    'neutron.objects.qos.qos_policy_validator.check_min_pps_rule_conflict',\n    return_value=None)\n", (35905, 36005), False, 'from unittest import mock\n'), ((36270, 36288), 'unittest.mock.call.create', 'mock.call.create', ([], {}), '()\n', (36286, 36288), False, 'from unittest import mock\n'), ((36330, 36399), 'unittest.mock.call.driver.call', 'mock.call.driver.call', (['"""update_policy_precommit"""', 'self.ctxt', 'mock.ANY'], {}), "('update_policy_precommit', self.ctxt, mock.ANY)\n", (36351, 36399), False, 'from unittest import mock\n'), ((36448, 36507), 'unittest.mock.call.driver.call', 'mock.call.driver.call', (['"""update_policy"""', 'self.ctxt', 'mock.ANY'], {}), "('update_policy', self.ctxt, mock.ANY)\n", (36469, 36507), False, 'from unittest import mock\n'), ((36932, 37020), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': '_policy'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=\n    _policy)\n", (36942, 37020), False, 'from unittest import mock\n'), ((37572, 37660), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': '_policy'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=\n    _policy)\n", (37582, 37660), False, 'from unittest import mock\n'), ((38279, 38367), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': '_policy'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=\n    _policy)\n", (38289, 38367), False, 'from unittest import mock\n'), ((38919, 39007), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': '_policy'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=\n    _policy)\n", (38929, 39007), False, 'from unittest import mock\n'), ((39639, 39727), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': '_policy'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=\n    _policy)\n", (39649, 39727), False, 'from unittest import mock\n'), ((40538, 40612), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.rule.get_rules"""'], {'return_value': '[self.rule]'}), "('neutron.objects.qos.rule.get_rules', return_value=[self.rule])\n", (40548, 40612), False, 'from unittest import mock\n'), ((40638, 40726), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': '_policy'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=\n    _policy)\n", (40648, 40726), False, 'from unittest import mock\n'), ((41066, 41084), 'unittest.mock.call.update', 'mock.call.update', ([], {}), '()\n', (41082, 41084), False, 'from unittest import mock\n'), ((41126, 41195), 'unittest.mock.call.driver.call', 'mock.call.driver.call', (['"""update_policy_precommit"""', 'self.ctxt', 'mock.ANY'], {}), "('update_policy_precommit', self.ctxt, mock.ANY)\n", (41147, 41195), False, 'from unittest import mock\n'), ((41244, 41303), 'unittest.mock.call.driver.call', 'mock.call.driver.call', (['"""update_policy"""', 'self.ctxt', 'mock.ANY'], {}), "('update_policy', self.ctxt, mock.ANY)\n", (41265, 41303), False, 'from unittest import mock\n'), ((41728, 41816), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': '_policy'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=\n    _policy)\n", (41738, 41816), False, 'from unittest import mock\n'), ((42258, 42346), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': '_policy'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=\n    _policy)\n", (42268, 42346), False, 'from unittest import mock\n'), ((42839, 42927), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': '_policy'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=\n    _policy)\n", (42849, 42927), False, 'from unittest import mock\n'), ((43304, 43392), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': '_policy'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=\n    _policy)\n", (43314, 43392), False, 'from unittest import mock\n'), ((43837, 43925), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': '_policy'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=\n    _policy)\n", (43847, 43925), False, 'from unittest import mock\n'), ((44322, 44410), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': '_policy'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=\n    _policy)\n", (44332, 44410), False, 'from unittest import mock\n'), ((44950, 45038), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': '_policy'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=\n    _policy)\n", (44960, 45038), False, 'from unittest import mock\n'), ((45779, 45867), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': '_policy'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=\n    _policy)\n", (45789, 45867), False, 'from unittest import mock\n'), ((46170, 46188), 'unittest.mock.call.delete', 'mock.call.delete', ([], {}), '()\n', (46186, 46188), False, 'from unittest import mock\n'), ((46230, 46299), 'unittest.mock.call.driver.call', 'mock.call.driver.call', (['"""update_policy_precommit"""', 'self.ctxt', 'mock.ANY'], {}), "('update_policy_precommit', self.ctxt, mock.ANY)\n", (46251, 46299), False, 'from unittest import mock\n'), ((46348, 46407), 'unittest.mock.call.driver.call', 'mock.call.driver.call', (['"""update_policy"""', 'self.ctxt', 'mock.ANY'], {}), "('update_policy', self.ctxt, mock.ANY)\n", (46369, 46407), False, 'from unittest import mock\n'), ((46826, 46914), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': '_policy'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=\n    _policy)\n", (46836, 46914), False, 'from unittest import mock\n'), ((47236, 47328), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': 'self.policy'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=\n    self.policy)\n", (47246, 47328), False, 'from unittest import mock\n'), ((47828, 47920), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': 'self.policy'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=\n    self.policy)\n", (47838, 47920), False, 'from unittest import mock\n'), ((48454, 48546), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': 'self.policy'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=\n    self.policy)\n", (48464, 48546), False, 'from unittest import mock\n'), ((49185, 49270), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': 'None'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=None\n    )\n", (49195, 49270), False, 'from unittest import mock\n'), ((49541, 49626), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': 'None'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=None\n    )\n", (49551, 49626), False, 'from unittest import mock\n'), ((49933, 50018), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': 'None'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=None\n    )\n", (49943, 50018), False, 'from unittest import mock\n'), ((50384, 50472), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': '_policy'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=\n    _policy)\n", (50394, 50472), False, 'from unittest import mock\n'), ((50900, 50988), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': '_policy'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=\n    _policy)\n", (50910, 50988), False, 'from unittest import mock\n'), ((51435, 51523), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': '_policy'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=\n    _policy)\n", (51445, 51523), False, 'from unittest import mock\n'), ((51856, 51948), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': 'self.policy'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=\n    self.policy)\n", (51866, 51948), False, 'from unittest import mock\n'), ((52473, 52565), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': 'self.policy'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=\n    self.policy)\n", (52483, 52565), False, 'from unittest import mock\n'), ((53196, 53281), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': 'None'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=None\n    )\n", (53206, 53281), False, 'from unittest import mock\n'), ((53587, 53672), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': 'None'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=None\n    )\n", (53597, 53672), False, 'from unittest import mock\n'), ((53941, 54033), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': 'self.policy'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=\n    self.policy)\n", (53951, 54033), False, 'from unittest import mock\n'), ((54539, 54631), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': 'self.policy'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=\n    self.policy)\n", (54549, 54631), False, 'from unittest import mock\n'), ((55171, 55263), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': 'self.policy'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=\n    self.policy)\n", (55181, 55263), False, 'from unittest import mock\n'), ((55929, 56014), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': 'None'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=None\n    )\n", (55939, 56014), False, 'from unittest import mock\n'), ((56325, 56410), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': 'None'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=None\n    )\n", (56335, 56410), False, 'from unittest import mock\n'), ((56692, 56777), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': 'None'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=None\n    )\n", (56702, 56777), False, 'from unittest import mock\n'), ((57075, 57160), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': 'None'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=None\n    )\n", (57085, 57160), False, 'from unittest import mock\n'), ((57472, 57557), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': 'None'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=None\n    )\n", (57482, 57557), False, 'from unittest import mock\n'), ((58353, 58457), 'unittest.mock.patch.object', 'mock.patch.object', (['qos_plugin.QoSPlugin', '"""supported_rule_type_details"""'], {'return_value': 'drivers_details'}), "(qos_plugin.QoSPlugin, 'supported_rule_type_details',\n    return_value=drivers_details)\n", (58370, 58457), False, 'from unittest import mock\n'), ((59134, 59243), 'unittest.mock.patch.object', 'mock.patch.object', (['qos_plugin.QoSPlugin', '"""supported_rule_types"""'], {'return_value': 'qos_consts.VALID_RULE_TYPES'}), "(qos_plugin.QoSPlugin, 'supported_rule_types',\n    return_value=qos_consts.VALID_RULE_TYPES)\n", (59151, 59243), False, 'from unittest import mock\n'), ((61080, 61139), 'unittest.mock.call.driver.call', 'mock.call.driver.call', (['"""update_policy"""', 'self.ctxt', 'mock.ANY'], {}), "('update_policy', self.ctxt, mock.ANY)\n", (61101, 61139), False, 'from unittest import mock\n'), ((61725, 61813), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': '_policy'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=\n    _policy)\n", (61735, 61813), False, 'from unittest import mock\n'), ((62415, 62503), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': '_policy'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=\n    _policy)\n", (62425, 62503), False, 'from unittest import mock\n'), ((62987, 63075), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': '_policy'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=\n    _policy)\n", (62997, 63075), False, 'from unittest import mock\n'), ((63527, 63615), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': '_policy'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=\n    _policy)\n", (63537, 63615), False, 'from unittest import mock\n'), ((64080, 64168), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': '_policy'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=\n    _policy)\n", (64090, 64168), False, 'from unittest import mock\n'), ((64604, 64692), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': '_policy'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=\n    _policy)\n", (64614, 64692), False, 'from unittest import mock\n'), ((65022, 65114), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': 'self.policy'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=\n    self.policy)\n", (65032, 65114), False, 'from unittest import mock\n'), ((65663, 65755), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': 'self.policy'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=\n    self.policy)\n", (65673, 65755), False, 'from unittest import mock\n'), ((66294, 66386), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': 'self.policy'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=\n    self.policy)\n", (66304, 66386), False, 'from unittest import mock\n'), ((67050, 67135), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': 'None'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=None\n    )\n", (67060, 67135), False, 'from unittest import mock\n'), ((67450, 67535), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': 'None'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=None\n    )\n", (67460, 67535), False, 'from unittest import mock\n'), ((67821, 67906), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': 'None'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=None\n    )\n", (67831, 67906), False, 'from unittest import mock\n'), ((68210, 68295), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': 'None'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=None\n    )\n", (68220, 68295), False, 'from unittest import mock\n'), ((68617, 68702), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': 'None'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=None\n    )\n", (68627, 68702), False, 'from unittest import mock\n'), ((69331, 69435), 'unittest.mock.patch.object', 'mock.patch.object', (['qos_plugin.QoSPlugin', '"""supported_rule_type_details"""'], {'return_value': 'drivers_details'}), "(qos_plugin.QoSPlugin, 'supported_rule_type_details',\n    return_value=drivers_details)\n", (69348, 69435), False, 'from unittest import mock\n'), ((70577, 70665), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': '_policy'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=\n    _policy)\n", (70587, 70665), False, 'from unittest import mock\n'), ((70709, 70783), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.network.Network.get_object"""'], {'return_value': 'net'}), "('neutron.objects.network.Network.get_object', return_value=net)\n", (70719, 70783), False, 'from unittest import mock\n'), ((70832, 70918), 'unittest.mock.patch.object', 'mock.patch.object', (['self.qos_plugin', '"""_get_ports_with_policy"""'], {'return_value': '[port]'}), "(self.qos_plugin, '_get_ports_with_policy', return_value=[\n    port])\n", (70849, 70918), False, 'from unittest import mock\n'), ((71686, 71774), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': '_policy'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=\n    _policy)\n", (71696, 71774), False, 'from unittest import mock\n'), ((71818, 71892), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.network.Network.get_object"""'], {'return_value': 'net'}), "('neutron.objects.network.Network.get_object', return_value=net)\n", (71828, 71892), False, 'from unittest import mock\n'), ((71941, 72027), 'unittest.mock.patch.object', 'mock.patch.object', (['self.qos_plugin', '"""_get_ports_with_policy"""'], {'return_value': '[port]'}), "(self.qos_plugin, '_get_ports_with_policy', return_value=[\n    port])\n", (71958, 72027), False, 'from unittest import mock\n'), ((73600, 73693), 'neutron.objects.qos.rule.QosMinimumPacketRateRule', 'rule_object.QosMinimumPacketRateRule', (['self.ctxt'], {}), "(self.ctxt, **rule_data[\n    'minimum_packet_rate_rule'])\n", (73636, 73693), True, 'from neutron.objects.qos import rule as rule_object\n'), ((74410, 74498), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': '_policy'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=\n    _policy)\n", (74420, 74498), False, 'from unittest import mock\n'), ((75114, 75202), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': '_policy'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=\n    _policy)\n", (75124, 75202), False, 'from unittest import mock\n'), ((75605, 75690), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': 'None'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=None\n    )\n", (75615, 75690), False, 'from unittest import mock\n'), ((76069, 76157), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': '_policy'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=\n    _policy)\n", (76079, 76157), False, 'from unittest import mock\n'), ((78171, 78259), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': '_policy'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=\n    _policy)\n", (78181, 78259), False, 'from unittest import mock\n'), ((78605, 78690), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': 'None'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=None\n    )\n", (78615, 78690), False, 'from unittest import mock\n'), ((79107, 79195), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': '_policy'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=\n    _policy)\n", (79117, 79195), False, 'from unittest import mock\n'), ((79565, 79653), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': '_policy'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=\n    _policy)\n", (79575, 79653), False, 'from unittest import mock\n'), ((79963, 80048), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': 'None'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=None\n    )\n", (79973, 80048), False, 'from unittest import mock\n'), ((80338, 80430), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': 'self.policy'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=\n    self.policy)\n", (80348, 80430), False, 'from unittest import mock\n'), ((80946, 81038), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': 'self.policy'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=\n    self.policy)\n", (80956, 81038), False, 'from unittest import mock\n'), ((81571, 81663), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': 'self.policy'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=\n    self.policy)\n", (81581, 81663), False, 'from unittest import mock\n'), ((82321, 82406), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': 'None'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=None\n    )\n", (82331, 82406), False, 'from unittest import mock\n'), ((82717, 82802), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': 'None'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=None\n    )\n", (82727, 82802), False, 'from unittest import mock\n'), ((83417, 83521), 'unittest.mock.patch.object', 'mock.patch.object', (['qos_plugin.QoSPlugin', '"""supported_rule_type_details"""'], {'return_value': 'drivers_details'}), "(qos_plugin.QoSPlugin, 'supported_rule_type_details',\n    return_value=drivers_details)\n", (83434, 83521), False, 'from unittest import mock\n'), ((87612, 87637), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ([], {}), '()\n', (87635, 87637), False, 'from oslo_utils import uuidutils\n'), ((88689, 88714), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ([], {}), '()\n', (88712, 88714), False, 'from oslo_utils import uuidutils\n'), ((89602, 89627), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ([], {}), '()\n', (89625, 89627), False, 'from oslo_utils import uuidutils\n'), ((90555, 90580), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ([], {}), '()\n', (90578, 90580), False, 'from oslo_utils import uuidutils\n'), ((93783, 93808), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ([], {}), '()\n', (93806, 93808), False, 'from oslo_utils import uuidutils\n'), ((94241, 94266), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ([], {}), '()\n', (94264, 94266), False, 'from oslo_utils import uuidutils\n'), ((94710, 94735), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ([], {}), '()\n', (94733, 94735), False, 'from oslo_utils import uuidutils\n'), ((95871, 95936), 'unittest.mock.patch.object', 'mock.patch.object', (['self.qos_plugin', '"""validate_policy_for_network"""'], {}), "(self.qos_plugin, 'validate_policy_for_network')\n", (95888, 95936), False, 'from unittest import mock\n'), ((97380, 97442), 'unittest.mock.patch.object', 'mock.patch.object', (['self.qos_plugin', '"""validate_policy_for_port"""'], {}), "(self.qos_plugin, 'validate_policy_for_port')\n", (97397, 97442), False, 'from unittest import mock\n'), ((99919, 99985), 'unittest.mock.patch.object', 'mock.patch.object', (['self.qos_plugin', '"""_change_placement_allocation"""'], {}), "(self.qos_plugin, '_change_placement_allocation')\n", (99936, 99985), False, 'from unittest import mock\n'), ((100712, 100778), 'unittest.mock.patch.object', 'mock.patch.object', (['self.qos_plugin', '"""_change_placement_allocation"""'], {}), "(self.qos_plugin, '_change_placement_allocation')\n", (100729, 100778), False, 'from unittest import mock\n'), ((101540, 101606), 'unittest.mock.patch.object', 'mock.patch.object', (['self.qos_plugin', '"""_change_placement_allocation"""'], {}), "(self.qos_plugin, '_change_placement_allocation')\n", (101557, 101606), False, 'from unittest import mock\n'), ((102408, 102474), 'unittest.mock.patch.object', 'mock.patch.object', (['self.qos_plugin', '"""_change_placement_allocation"""'], {}), "(self.qos_plugin, '_change_placement_allocation')\n", (102425, 102474), False, 'from unittest import mock\n'), ((103270, 103336), 'unittest.mock.patch.object', 'mock.patch.object', (['self.qos_plugin', '"""_change_placement_allocation"""'], {}), "(self.qos_plugin, '_change_placement_allocation')\n", (103287, 103336), False, 'from unittest import mock\n'), ((104014, 104080), 'unittest.mock.patch.object', 'mock.patch.object', (['self.qos_plugin', '"""_change_placement_allocation"""'], {}), "(self.qos_plugin, '_change_placement_allocation')\n", (104031, 104080), False, 'from unittest import mock\n'), ((105211, 105277), 'unittest.mock.patch.object', 'mock.patch.object', (['self.qos_plugin', '"""_change_placement_allocation"""'], {}), "(self.qos_plugin, '_change_placement_allocation')\n", (105228, 105277), False, 'from unittest import mock\n'), ((106786, 106852), 'unittest.mock.patch.object', 'mock.patch.object', (['self.qos_plugin', '"""_change_placement_allocation"""'], {}), "(self.qos_plugin, '_change_placement_allocation')\n", (106803, 106852), False, 'from unittest import mock\n'), ((107541, 107693), 'unittest.mock.call', 'mock.call', (['original_qos', 'None', "{'id': port1.id, 'device_owner': 'compute:uu:id', 'qos_policy_id': None,\n    'qos_network_policy_id': None}", '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)\n", (107550, 107693), False, 'from unittest import mock\n'), ((109534, 109600), 'unittest.mock.patch.object', 'mock.patch.object', (['self.qos_plugin', '"""_change_placement_allocation"""'], {}), "(self.qos_plugin, '_change_placement_allocation')\n", (109551, 109600), False, 'from unittest import mock\n'), ((110321, 110475), 'unittest.mock.call', 'mock.call', (['original_qos', 'qos', "{'id': port1.id, 'device_owner': 'compute:uu:id', 'qos_policy_id': None,\n    'qos_network_policy_id': qos.id}", '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)\n", (110330, 110475), False, 'from unittest import mock\n'), ((110605, 110759), 'unittest.mock.call', 'mock.call', (['original_qos', 'qos', "{'id': port2.id, 'device_owner': 'compute:uu:id', 'qos_policy_id': None,\n    'qos_network_policy_id': qos.id}", '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)\n", (110614, 110759), False, 'from unittest import mock\n'), ((114150, 114227), 'unittest.mock.patch.object', 'mock.patch.object', (['self.qos_plugin._placement_client', '"""update_qos_allocation"""'], {}), "(self.qos_plugin._placement_client, 'update_qos_allocation')\n", (114167, 114227), False, 'from unittest import mock\n'), ((114904, 114981), 'unittest.mock.patch.object', 'mock.patch.object', (['self.qos_plugin._placement_client', '"""update_qos_allocation"""'], {}), "(self.qos_plugin._placement_client, 'update_qos_allocation')\n", (114921, 114981), False, 'from unittest import mock\n'), ((115730, 115807), 'unittest.mock.patch.object', 'mock.patch.object', (['self.qos_plugin._placement_client', '"""update_qos_allocation"""'], {}), "(self.qos_plugin._placement_client, 'update_qos_allocation')\n", (115747, 115807), False, 'from unittest import mock\n'), ((116701, 116778), 'unittest.mock.patch.object', 'mock.patch.object', (['self.qos_plugin._placement_client', '"""update_qos_allocation"""'], {}), "(self.qos_plugin._placement_client, 'update_qos_allocation')\n", (116718, 116778), False, 'from unittest import mock\n'), ((117756, 117833), 'unittest.mock.patch.object', 'mock.patch.object', (['self.qos_plugin._placement_client', '"""update_qos_allocation"""'], {}), "(self.qos_plugin._placement_client, 'update_qos_allocation')\n", (117773, 117833), False, 'from unittest import mock\n'), ((118371, 118448), 'unittest.mock.patch.object', 'mock.patch.object', (['self.qos_plugin._placement_client', '"""update_qos_allocation"""'], {}), "(self.qos_plugin._placement_client, 'update_qos_allocation')\n", (118388, 118448), False, 'from unittest import mock\n'), ((119010, 119087), 'unittest.mock.patch.object', 'mock.patch.object', (['self.qos_plugin._placement_client', '"""update_qos_allocation"""'], {}), "(self.qos_plugin._placement_client, 'update_qos_allocation')\n", (119027, 119087), False, 'from unittest import mock\n'), ((119799, 119876), 'unittest.mock.patch.object', 'mock.patch.object', (['self.qos_plugin._placement_client', '"""update_qos_allocation"""'], {}), "(self.qos_plugin._placement_client, 'update_qos_allocation')\n", (119816, 119876), False, 'from unittest import mock\n'), ((120599, 120676), 'unittest.mock.patch.object', 'mock.patch.object', (['self.qos_plugin._placement_client', '"""update_qos_allocation"""'], {}), "(self.qos_plugin._placement_client, 'update_qos_allocation')\n", (120616, 120676), False, 'from unittest import mock\n'), ((121408, 121485), 'unittest.mock.patch.object', 'mock.patch.object', (['self.qos_plugin._placement_client', '"""update_qos_allocation"""'], {}), "(self.qos_plugin._placement_client, 'update_qos_allocation')\n", (121425, 121485), False, 'from unittest import mock\n'), ((122143, 122220), 'unittest.mock.patch.object', 'mock.patch.object', (['self.qos_plugin._placement_client', '"""update_qos_allocation"""'], {}), "(self.qos_plugin._placement_client, 'update_qos_allocation')\n", (122160, 122220), False, 'from unittest import mock\n'), ((122669, 122746), 'unittest.mock.patch.object', 'mock.patch.object', (['self.qos_plugin._placement_client', '"""update_qos_allocation"""'], {}), "(self.qos_plugin._placement_client, 'update_qos_allocation')\n", (122686, 122746), False, 'from unittest import mock\n'), ((123807, 123884), 'unittest.mock.patch.object', 'mock.patch.object', (['self.qos_plugin._placement_client', '"""update_qos_allocation"""'], {}), "(self.qos_plugin._placement_client, 'update_qos_allocation')\n", (123824, 123884), False, 'from unittest import mock\n'), ((124465, 124542), 'unittest.mock.patch.object', 'mock.patch.object', (['self.qos_plugin._placement_client', '"""update_qos_allocation"""'], {}), "(self.qos_plugin._placement_client, 'update_qos_allocation')\n", (124482, 124542), False, 'from unittest import mock\n'), ((125116, 125193), 'unittest.mock.patch.object', 'mock.patch.object', (['self.qos_plugin._placement_client', '"""update_qos_allocation"""'], {}), "(self.qos_plugin._placement_client, 'update_qos_allocation')\n", (125133, 125193), False, 'from unittest import mock\n'), ((125876, 125953), 'unittest.mock.patch.object', 'mock.patch.object', (['self.qos_plugin._placement_client', '"""update_qos_allocation"""'], {}), "(self.qos_plugin._placement_client, 'update_qos_allocation')\n", (125893, 125953), False, 'from unittest import mock\n'), ((126447, 126524), 'unittest.mock.patch.object', 'mock.patch.object', (['self.qos_plugin._placement_client', '"""update_qos_allocation"""'], {}), "(self.qos_plugin._placement_client, 'update_qos_allocation')\n", (126464, 126524), False, 'from unittest import mock\n'), ((126615, 126694), 'keystoneauth1.exceptions.Conflict', 'ks_exc.Conflict', ([], {'response': "{'errors': [{'code': 'placement.concurrent_update'}]}"}), "(response={'errors': [{'code': 'placement.concurrent_update'}]})\n", (126630, 126694), True, 'from keystoneauth1 import exceptions as ks_exc\n'), ((127242, 127319), 'unittest.mock.patch.object', 'mock.patch.object', (['self.qos_plugin._placement_client', '"""update_qos_allocation"""'], {}), "(self.qos_plugin._placement_client, 'update_qos_allocation')\n", (127259, 127319), False, 'from unittest import mock\n'), ((127428, 127497), 'neutron_lib.exceptions.placement.PlacementAllocationGenerationConflict', 'pl_exc.PlacementAllocationGenerationConflict', ([], {'consumer': 'self.MIN_BW_RP'}), '(consumer=self.MIN_BW_RP)\n', (127472, 127497), True, 'from neutron_lib.exceptions import placement as pl_exc\n'), ((128086, 128163), 'unittest.mock.patch.object', 'mock.patch.object', (['self.qos_plugin._placement_client', '"""update_qos_allocation"""'], {}), "(self.qos_plugin._placement_client, 'update_qos_allocation')\n", (128103, 128163), False, 'from unittest import mock\n'), ((2224, 2274), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.db.api.create_object"""'], {}), "('neutron.objects.db.api.create_object')\n", (2234, 2274), False, 'from unittest import mock\n'), ((2291, 2341), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.db.api.update_object"""'], {}), "('neutron.objects.db.api.update_object')\n", (2301, 2341), False, 'from unittest import mock\n'), ((2358, 2408), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.db.api.delete_object"""'], {}), "('neutron.objects.db.api.delete_object')\n", (2368, 2408), False, 'from unittest import mock\n'), ((2425, 2472), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.db.api.get_object"""'], {}), "('neutron.objects.db.api.get_object')\n", (2435, 2472), False, 'from unittest import mock\n'), ((2788, 2860), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.base.NeutronDbObject.modify_fields_from_db"""'], {}), "('neutron.objects.base.NeutronDbObject.modify_fields_from_db')\n", (2798, 2860), False, 'from unittest import mock\n'), ((2899, 2958), 'unittest.mock.patch.object', 'mock.patch.object', (['policy_object.QosPolicy', '"""unset_default"""'], {}), "(policy_object.QosPolicy, 'unset_default')\n", (2916, 2958), False, 'from unittest import mock\n'), ((2975, 3032), 'unittest.mock.patch.object', 'mock.patch.object', (['policy_object.QosPolicy', '"""set_default"""'], {}), "(policy_object.QosPolicy, 'set_default')\n", (2992, 3032), False, 'from unittest import mock\n'), ((3335, 3412), 'unittest.mock.patch', 'mock.patch', (['"""neutron.api.rpc.handlers.resources_rpc.ResourcesPushRpcApi.push"""'], {}), "('neutron.api.rpc.handlers.resources_rpc.ResourcesPushRpcApi.push')\n", (3345, 3412), False, 'from unittest import mock\n'), ((3637, 3662), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ([], {}), '()\n', (3660, 3662), False, 'from oslo_utils import uuidutils\n'), ((3701, 3726), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ([], {}), '()\n', (3724, 3726), False, 'from oslo_utils import uuidutils\n'), ((3994, 4019), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ([], {}), '()\n', (4017, 4019), False, 'from oslo_utils import uuidutils\n'), ((4176, 4201), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ([], {}), '()\n', (4199, 4201), False, 'from oslo_utils import uuidutils\n'), ((4317, 4342), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ([], {}), '()\n', (4340, 4342), False, 'from oslo_utils import uuidutils\n'), ((4439, 4464), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ([], {}), '()\n', (4462, 4464), False, 'from oslo_utils import uuidutils\n'), ((4602, 4627), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ([], {}), '()\n', (4625, 4627), False, 'from oslo_utils import uuidutils\n'), ((6181, 6206), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ([], {}), '()\n', (6204, 6206), False, 'from oslo_utils import uuidutils\n'), ((23233, 23258), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ([], {}), '()\n', (23256, 23258), False, 'from oslo_utils import uuidutils\n'), ((23652, 23677), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ([], {}), '()\n', (23675, 23677), False, 'from oslo_utils import uuidutils\n'), ((24268, 24293), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ([], {}), '()\n', (24291, 24293), False, 'from oslo_utils import uuidutils\n'), ((24321, 24346), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ([], {}), '()\n', (24344, 24346), False, 'from oslo_utils import uuidutils\n'), ((24418, 24443), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ([], {}), '()\n', (24441, 24443), False, 'from oslo_utils import uuidutils\n'), ((26318, 26343), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ([], {}), '()\n', (26341, 26343), False, 'from oslo_utils import uuidutils\n'), ((26749, 26774), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ([], {}), '()\n', (26772, 26774), False, 'from oslo_utils import uuidutils\n'), ((28794, 28819), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ([], {}), '()\n', (28817, 28819), False, 'from oslo_utils import uuidutils\n'), ((28844, 28869), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ([], {}), '()\n', (28867, 28869), False, 'from oslo_utils import uuidutils\n'), ((29901, 29926), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ([], {}), '()\n', (29924, 29926), False, 'from oslo_utils import uuidutils\n'), ((29951, 29976), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ([], {}), '()\n', (29974, 29976), False, 'from oslo_utils import uuidutils\n'), ((31158, 31179), 'unittest.mock.call.QosPolicy', 'mock.call.QosPolicy', ([], {}), '()\n', (31177, 31179), False, 'from unittest import mock\n'), ((47366, 47437), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.rule.QosBandwidthLimitRule.get_object"""'], {}), "('neutron.objects.qos.rule.QosBandwidthLimitRule.get_object')\n", (47376, 47437), False, 'from unittest import mock\n'), ((47958, 48030), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.rule.QosBandwidthLimitRule.get_objects"""'], {}), "('neutron.objects.qos.rule.QosBandwidthLimitRule.get_objects')\n", (47968, 48030), False, 'from unittest import mock\n'), ((48584, 48656), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.rule.QosBandwidthLimitRule.get_objects"""'], {}), "('neutron.objects.qos.rule.QosBandwidthLimitRule.get_objects')\n", (48594, 48656), False, 'from unittest import mock\n'), ((51986, 52055), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.rule.QosDscpMarkingRule.get_objects"""'], {}), "('neutron.objects.qos.rule.QosDscpMarkingRule.get_objects')\n", (51996, 52055), False, 'from unittest import mock\n'), ((52603, 52672), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.rule.QosDscpMarkingRule.get_objects"""'], {}), "('neutron.objects.qos.rule.QosDscpMarkingRule.get_objects')\n", (52613, 52672), False, 'from unittest import mock\n'), ((54071, 54144), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.rule.QosMinimumBandwidthRule.get_object"""'], {}), "('neutron.objects.qos.rule.QosMinimumBandwidthRule.get_object')\n", (54081, 54144), False, 'from unittest import mock\n'), ((54669, 54743), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.rule.QosMinimumBandwidthRule.get_objects"""'], {}), "('neutron.objects.qos.rule.QosMinimumBandwidthRule.get_objects')\n", (54679, 54743), False, 'from unittest import mock\n'), ((55301, 55375), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.rule.QosMinimumBandwidthRule.get_objects"""'], {}), "('neutron.objects.qos.rule.QosMinimumBandwidthRule.get_objects')\n", (55311, 55375), False, 'from unittest import mock\n'), ((65152, 65224), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.rule.QosPacketRateLimitRule.get_object"""'], {}), "('neutron.objects.qos.rule.QosPacketRateLimitRule.get_object')\n", (65162, 65224), False, 'from unittest import mock\n'), ((65793, 65866), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.rule.QosPacketRateLimitRule.get_objects"""'], {}), "('neutron.objects.qos.rule.QosPacketRateLimitRule.get_objects')\n", (65803, 65866), False, 'from unittest import mock\n'), ((66424, 66497), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.rule.QosPacketRateLimitRule.get_objects"""'], {}), "('neutron.objects.qos.rule.QosPacketRateLimitRule.get_objects')\n", (66434, 66497), False, 'from unittest import mock\n'), ((70441, 70466), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ([], {}), '()\n', (70464, 70466), False, 'from oslo_utils import uuidutils\n'), ((70491, 70516), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ([], {}), '()\n', (70514, 70516), False, 'from oslo_utils import uuidutils\n'), ((71567, 71592), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ([], {}), '()\n', (71590, 71592), False, 'from oslo_utils import uuidutils\n'), ((71617, 71642), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ([], {}), '()\n', (71640, 71642), False, 'from oslo_utils import uuidutils\n'), ((73061, 73149), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': '_policy'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=\n    _policy)\n", (73071, 73149), False, 'from unittest import mock\n'), ((73777, 73865), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': '_policy'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=\n    _policy)\n", (73787, 73865), False, 'from unittest import mock\n'), ((74956, 74981), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ([], {}), '()\n', (74979, 74981), False, 'from oslo_utils import uuidutils\n'), ((77157, 77254), 'neutron.objects.qos.rule.QosMinimumPacketRateRule', 'rule_object.QosMinimumPacketRateRule', (['self.ctxt'], {}), "(self.ctxt, **rules_data[0][\n    'minimum_packet_rate_rule'])\n", (77193, 77254), True, 'from neutron.objects.qos import rule as rule_object\n'), ((77288, 77385), 'neutron.objects.qos.rule.QosMinimumPacketRateRule', 'rule_object.QosMinimumPacketRateRule', (['self.ctxt'], {}), "(self.ctxt, **rules_data[1][\n    'minimum_packet_rate_rule'])\n", (77324, 77385), True, 'from neutron.objects.qos import rule as rule_object\n'), ((77479, 77567), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.policy.QosPolicy.get_object"""'], {'return_value': '_policy'}), "('neutron.objects.qos.policy.QosPolicy.get_object', return_value=\n    _policy)\n", (77489, 77567), False, 'from unittest import mock\n'), ((80468, 80542), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.rule.QosMinimumPacketRateRule.get_object"""'], {}), "('neutron.objects.qos.rule.QosMinimumPacketRateRule.get_object')\n", (80478, 80542), False, 'from unittest import mock\n'), ((81076, 81151), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.rule.QosMinimumPacketRateRule.get_objects"""'], {}), "('neutron.objects.qos.rule.QosMinimumPacketRateRule.get_objects')\n", (81086, 81151), False, 'from unittest import mock\n'), ((81701, 81776), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.rule.QosMinimumPacketRateRule.get_objects"""'], {}), "('neutron.objects.qos.rule.QosMinimumPacketRateRule.get_objects')\n", (81711, 81776), False, 'from unittest import mock\n'), ((87932, 88008), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.rule.QosRule.get_object"""'], {'return_value': 'rule'}), "('neutron.objects.qos.rule.QosRule.get_object', return_value=rule)\n", (87942, 88008), False, 'from unittest import mock\n'), ((88260, 88356), 'unittest.mock.call', 'mock.call', (['mock.ANY', 'rule_object_class', 'rule_id', 'self.qos_policy_id', '{rule_data_name: data}'], {}), '(mock.ANY, rule_object_class, rule_id, self.qos_policy_id, {\n    rule_data_name: data})\n', (88269, 88356), False, 'from unittest import mock\n'), ((89009, 89085), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.rule.QosRule.get_object"""'], {'return_value': 'rule'}), "('neutron.objects.qos.rule.QosRule.get_object', return_value=rule)\n", (89019, 89085), False, 'from unittest import mock\n'), ((89192, 89259), 'unittest.mock.call', 'mock.call', (['mock.ANY', 'rule_object_class', 'rule_id', 'self.qos_policy_id'], {}), '(mock.ANY, rule_object_class, rule_id, self.qos_policy_id)\n', (89201, 89259), False, 'from unittest import mock\n'), ((89922, 89998), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.rule.QosRule.get_object"""'], {'return_value': 'rule'}), "('neutron.objects.qos.rule.QosRule.get_object', return_value=rule)\n", (89932, 89998), False, 'from unittest import mock\n'), ((90242, 90309), 'unittest.mock.call', 'mock.call', (['mock.ANY', 'rule_object_class', 'rule_id', 'self.qos_policy_id'], {}), '(mock.ANY, rule_object_class, rule_id, self.qos_policy_id)\n', (90251, 90309), False, 'from unittest import mock\n'), ((90598, 90674), 'unittest.mock.patch', 'mock.patch', (['"""neutron.objects.qos.rule.QosRule.get_object"""'], {'return_value': 'None'}), "('neutron.objects.qos.rule.QosRule.get_object', return_value=None)\n", (90608, 90674), False, 'from unittest import mock\n'), ((93156, 93233), 'unittest.mock.patch', 'mock.patch', (['"""neutron.api.rpc.handlers.resources_rpc.ResourcesPushRpcApi.push"""'], {}), "('neutron.api.rpc.handlers.resources_rpc.ResourcesPushRpcApi.push')\n", (93166, 93233), False, 'from unittest import mock\n'), ((94823, 94863), 'neutron_lib.utils.net.random_mac_generator', 'net_utils.random_mac_generator', (['base_mac'], {}), '(base_mac)\n', (94853, 94863), True, 'from neutron_lib.utils import net as net_utils\n'), ((7936, 7961), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ([], {}), '()\n', (7959, 7961), False, 'from oslo_utils import uuidutils\n'), ((8215, 8240), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ([], {}), '()\n', (8238, 8240), False, 'from oslo_utils import uuidutils\n'), ((20298, 20323), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ([], {}), '()\n', (20321, 20323), False, 'from oslo_utils import uuidutils\n'), ((22445, 22547), 'neutron_lib.callbacks.events.DBEventPayload', 'events.DBEventPayload', (['self.ctxt'], {'desired_state': "kwargs['port']", 'states': "(kwargs['original_port'],)"}), "(self.ctxt, desired_state=kwargs['port'], states=(\n    kwargs['original_port'],))\n", (22466, 22547), False, 'from neutron_lib.callbacks import events\n'), ((25368, 25476), 'neutron_lib.callbacks.events.DBEventPayload', 'events.DBEventPayload', (['self.ctxt'], {'desired_state': "kwargs['network']", 'states': "(kwargs['original_network'],)"}), "(self.ctxt, desired_state=kwargs['network'], states=(\n    kwargs['original_network'],))\n", (25389, 25476), False, 'from neutron_lib.callbacks import events\n'), ((61017, 61036), 'unittest.mock.call.RuleCls', 'mock.call.RuleCls', ([], {}), '()\n', (61034, 61036), False, 'from unittest import mock\n'), ((72635, 72660), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ([], {}), '()\n', (72658, 72660), False, 'from oslo_utils import uuidutils\n'), ((72860, 72885), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ([], {}), '()\n', (72883, 72885), False, 'from oslo_utils import uuidutils\n'), ((76636, 76661), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ([], {}), '()\n', (76659, 76661), False, 'from oslo_utils import uuidutils\n'), ((76861, 76886), 'oslo_utils.uuidutils.generate_uuid', 'uuidutils.generate_uuid', ([], {}), '()\n', (76884, 76886), False, 'from oslo_utils import uuidutils\n'), ((96160, 96232), 'neutron_lib.callbacks.events.DBEventPayload', 'events.DBEventPayload', (['self.context'], {'resource_id': "kwargs['network']['id']"}), "(self.context, resource_id=kwargs['network']['id'])\n", (96181, 96232), False, 'from neutron_lib.callbacks import events\n'), ((97629, 97698), 'neutron_lib.callbacks.events.DBEventPayload', 'events.DBEventPayload', (['self.context'], {'resource_id': "kwargs['port']['id']"}), "(self.context, resource_id=kwargs['port']['id'])\n", (97650, 97698), False, 'from neutron_lib.callbacks import events\n'), ((100194, 100254), 'neutron_lib.callbacks.events.DBEventPayload', 'events.DBEventPayload', (['context'], {'states': '(original_port, port)'}), '(context, states=(original_port, port))\n', (100215, 100254), False, 'from neutron_lib.callbacks import events\n'), ((100987, 101047), 'neutron_lib.callbacks.events.DBEventPayload', 'events.DBEventPayload', (['context'], {'states': '(original_port, port)'}), '(context, states=(original_port, port))\n', (101008, 101047), False, 'from neutron_lib.callbacks import events\n'), ((101815, 101875), 'neutron_lib.callbacks.events.DBEventPayload', 'events.DBEventPayload', (['context'], {'states': '(original_port, port)'}), '(context, states=(original_port, port))\n', (101836, 101875), False, 'from neutron_lib.callbacks import events\n'), ((102683, 102743), 'neutron_lib.callbacks.events.DBEventPayload', 'events.DBEventPayload', (['context'], {'states': '(original_port, port)'}), '(context, states=(original_port, port))\n', (102704, 102743), False, 'from neutron_lib.callbacks import events\n'), ((103545, 103605), 'neutron_lib.callbacks.events.DBEventPayload', 'events.DBEventPayload', (['context'], {'states': '(original_port, port)'}), '(context, states=(original_port, port))\n', (103566, 103605), False, 'from neutron_lib.callbacks import events\n'), ((104295, 104366), 'neutron_lib.callbacks.events.DBEventPayload', 'events.DBEventPayload', (['self.context'], {'states': '(original_network, network)'}), '(self.context, states=(original_network, network))\n', (104316, 104366), False, 'from neutron_lib.callbacks import events\n'), ((105492, 105563), 'neutron_lib.callbacks.events.DBEventPayload', 'events.DBEventPayload', (['self.context'], {'states': '(original_network, network)'}), '(self.context, states=(original_network, network))\n', (105513, 105563), False, 'from neutron_lib.callbacks import events\n'), ((107327, 107398), 'neutron_lib.callbacks.events.DBEventPayload', 'events.DBEventPayload', (['self.context'], {'states': '(original_network, network)'}), '(self.context, states=(original_network, network))\n', (107348, 107398), False, 'from neutron_lib.callbacks import events\n'), ((110106, 110177), 'neutron_lib.callbacks.events.DBEventPayload', 'events.DBEventPayload', (['self.context'], {'states': '(original_network, network)'}), '(self.context, states=(original_network, network))\n', (110127, 110177), False, 'from neutron_lib.callbacks import events\n'), ((60836, 60872), 'unittest.mock.call.QosPolicy.get_policy_obj', 'mock.call.QosPolicy.get_policy_obj', ([], {}), '()\n', (60870, 60872), False, 'from unittest import mock\n')]