<s> import copy <EOL> import mock <EOL> from oslo_utils import uuidutils <EOL> from neutron . api . rpc . callbacks . consumer import registry as consumer_reg <EOL> from neutron . api . rpc . callbacks import events <EOL> from neutron . api . rpc . callbacks import resources <EOL> from neutron . objects . qos import policy <EOL> from neutron . objects . qos import rule <EOL> from neutron . tests . common . agents import l2_extensions <EOL> from neutron . tests . functional . agent . l2 import base <EOL> TEST_POLICY_ID1 = "<STR_LIT>" <EOL> TEST_POLICY_ID2 = "<STR_LIT>" <EOL> TEST_DSCP_MARK_1 = <NUM_LIT> <EOL> TEST_DSCP_MARK_2 = <NUM_LIT:30> <EOL> TEST_DSCP_MARKING_RULE_1 = rule . QosDscpMarkingRule ( <EOL> context = None , <EOL> qos_policy_id = TEST_POLICY_ID1 , <EOL> id = "<STR_LIT>" , <EOL> dscp_mark = TEST_DSCP_MARK_1 ) <EOL> TEST_DSCP_MARKING_RULE_2 = rule . QosDscpMarkingRule ( <EOL> context = None , <EOL> qos_policy_id = TEST_POLICY_ID2 , <EOL> id = "<STR_LIT>" , <EOL> dscp_mark = TEST_DSCP_MARK_2 ) <EOL> TEST_BW_LIMIT_RULE_1 = rule . QosBandwidthLimitRule ( <EOL> context = None , <EOL> qos_policy_id = TEST_POLICY_ID1 , <EOL> id = "<STR_LIT>" , <EOL> max_kbps = <NUM_LIT:1000> , <EOL> max_burst_kbps = <NUM_LIT:10> ) <EOL> TEST_BW_LIMIT_RULE_2 = rule . QosBandwidthLimitRule ( <EOL> context = None , <EOL> qos_policy_id = TEST_POLICY_ID2 , <EOL> id = "<STR_LIT>" , <EOL> max_kbps = <NUM_LIT> , <EOL> max_burst_kbps = <NUM_LIT:9> ) <EOL> class OVSAgentQoSExtensionTestFramework ( base . OVSAgentTestFramework ) : <EOL> def setUp ( self ) : <EOL> super ( OVSAgentQoSExtensionTestFramework , self ) . setUp ( ) <EOL> self . config . set_override ( '<STR_LIT>' , [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . _set_pull_mock ( ) <EOL> self . set_test_qos_rules ( TEST_POLICY_ID1 , <EOL> [ TEST_BW_LIMIT_RULE_1 , <EOL> TEST_DSCP_MARKING_RULE_1 ] ) <EOL> self . set_test_qos_rules ( TEST_POLICY_ID2 , <EOL> [ TEST_BW_LIMIT_RULE_2 , <EOL> TEST_DSCP_MARKING_RULE_2 ] ) <EOL> def _set_pull_mock ( self ) : <EOL> self . qos_policies = { } <EOL> def _pull_mock ( context , resource_type , resource_id ) : <EOL> return self . qos_policies [ resource_id ] <EOL> self . pull = mock . patch ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) . start ( ) <EOL> self . pull . side_effect = _pull_mock <EOL> def set_test_qos_rules ( self , policy_id , policy_rules ) : <EOL> """<STR_LIT>""" <EOL> qos_policy = policy . QosPolicy ( <EOL> context = None , <EOL> tenant_id = uuidutils . generate_uuid ( ) , <EOL> id = policy_id , <EOL> name = "<STR_LIT>" , <EOL> description = "<STR_LIT>" , <EOL> shared = False , <EOL> rules = policy_rules ) <EOL> qos_policy . obj_reset_changes ( ) <EOL> self . qos_policies [ policy_id ] = qos_policy <EOL> def _create_test_port_dict ( self , policy_id = None ) : <EOL> port_dict = super ( OVSAgentQoSExtensionTestFramework , <EOL> self ) . _create_test_port_dict ( ) <EOL> port_dict [ '<STR_LIT>' ] = policy_id <EOL> port_dict [ '<STR_LIT>' ] = None <EOL> return port_dict <EOL> def _get_device_details ( self , port , network ) : <EOL> dev = super ( OVSAgentQoSExtensionTestFramework , <EOL> self ) . _get_device_details ( port , network ) <EOL> dev [ '<STR_LIT>' ] = port [ '<STR_LIT>' ] <EOL> return dev <EOL> def _assert_bandwidth_limit_rule_is_set ( self , port , rule ) : <EOL> max_rate , burst = ( <EOL> self . agent . int_br . get_egress_bw_limit_for_port ( port [ '<STR_LIT>' ] ) ) <EOL> self . assertEqual ( max_rate , rule . max_kbps ) <EOL> self . assertEqual ( burst , rule . max_burst_kbps ) <EOL> def _assert_bandwidth_limit_rule_not_set ( self , port ) : <EOL> max_rate , burst = ( <EOL> self . agent . int_br . get_egress_bw_limit_for_port ( port [ '<STR_LIT>' ] ) ) <EOL> self . assertIsNone ( max_rate ) <EOL> self . assertIsNone ( burst ) <EOL> def wait_until_bandwidth_limit_rule_applied ( self , port , rule ) : <EOL> l2_extensions . wait_until_bandwidth_limit_rule_applied ( <EOL> self . agent . int_br , port [ '<STR_LIT>' ] , rule ) <EOL> def _assert_dscp_marking_rule_is_set ( self , port , dscp_rule ) : <EOL> port_num = self . agent . int_br . _get_port_ofport ( port [ '<STR_LIT>' ] ) <EOL> flows = self . agent . int_br . dump_flows_for ( table = '<STR_LIT:0>' , <EOL> in_port = str ( port_num ) ) <EOL> tos_mark = l2_extensions . extract_mod_nw_tos_action ( flows ) <EOL> self . assertEqual ( dscp_rule . dscp_mark << <NUM_LIT:2> , tos_mark ) <EOL> def _assert_dscp_marking_rule_not_set ( self , port ) : <EOL> port_num = self . agent . int_br . _get_port_ofport ( port [ '<STR_LIT>' ] ) <EOL> flows = self . agent . int_br . dump_flows_for ( table = '<STR_LIT:0>' , <EOL> in_port = str ( port_num ) ) <EOL> tos_mark = l2_extensions . extract_mod_nw_tos_action ( flows ) <EOL> self . assertIsNone ( tos_mark ) <EOL> def wait_until_dscp_marking_rule_applied ( self , port , dscp_mark ) : <EOL> l2_extensions . wait_until_dscp_marking_rule_applied ( <EOL> self . agent . int_br , port [ '<STR_LIT>' ] , dscp_mark ) <EOL> def _create_port_with_qos ( self ) : <EOL> port_dict = self . _create_test_port_dict ( ) <EOL> port_dict [ '<STR_LIT>' ] = TEST_POLICY_ID1 <EOL> self . setup_agent_and_ports ( [ port_dict ] ) <EOL> self . wait_until_ports_state ( self . ports , up = True ) <EOL> self . wait_until_bandwidth_limit_rule_applied ( port_dict , <EOL> TEST_BW_LIMIT_RULE_1 ) <EOL> return port_dict <EOL> class TestOVSAgentQosExtension ( OVSAgentQoSExtensionTestFramework ) : <EOL> def test_port_creation_with_bandwidth_limit ( self ) : <EOL> """<STR_LIT>""" <EOL> self . setup_agent_and_ports ( <EOL> port_dicts = self . create_test_ports ( amount = <NUM_LIT:1> , <EOL> policy_id = TEST_POLICY_ID1 ) ) <EOL> self . wait_until_ports_state ( self . ports , up = True ) <EOL> for port in self . ports : <EOL> self . _assert_bandwidth_limit_rule_is_set ( <EOL> port , TEST_BW_LIMIT_RULE_1 ) <EOL> def test_port_creation_with_different_bandwidth_limits ( self ) : <EOL> """<STR_LIT>""" <EOL> port_dicts = self . create_test_ports ( amount = <NUM_LIT:3> ) <EOL> port_dicts [ <NUM_LIT:0> ] [ '<STR_LIT>' ] = TEST_POLICY_ID1 <EOL> port_dicts [ <NUM_LIT:1> ] [ '<STR_LIT>' ] = TEST_POLICY_ID2 <EOL> self . setup_agent_and_ports ( port_dicts ) <EOL> self . wait_until_ports_state ( self . ports , up = True ) <EOL> self . _assert_bandwidth_limit_rule_is_set ( self . ports [ <NUM_LIT:0> ] , <EOL> TEST_BW_LIMIT_RULE_1 ) <EOL> self . _assert_bandwidth_limit_rule_is_set ( self . ports [ <NUM_LIT:1> ] , <EOL> TEST_BW_LIMIT_RULE_2 ) <EOL> self . _assert_bandwidth_limit_rule_not_set ( self . ports [ <NUM_LIT:2> ] ) <EOL> def test_port_creation_with_dscp_marking ( self ) : <EOL> """<STR_LIT>""" <EOL> self . setup_agent_and_ports ( <EOL> port_dicts = self . create_test_ports ( amount = <NUM_LIT:1> , <EOL> policy_id = TEST_POLICY_ID1 ) ) <EOL> self . wait_until_ports_state ( self . ports , up = True ) <EOL> for port in self . ports : <EOL> self . _assert_dscp_marking_rule_is_set ( <EOL> port , TEST_DSCP_MARKING_RULE_1 ) <EOL> def test_port_creation_with_different_dscp_markings ( self ) : <EOL> """<STR_LIT>""" <EOL> port_dicts = self . create_test_ports ( amount = <NUM_LIT:3> ) <EOL> port_dicts [ <NUM_LIT:0> ] [ '<STR_LIT>' ] = TEST_POLICY_ID1 <EOL> port_dicts [ <NUM_LIT:1> ] [ '<STR_LIT>' ] = TEST_POLICY_ID2 <EOL> self . setup_agent_and_ports ( port_dicts ) <EOL> self . wait_until_ports_state ( self . ports , up = True ) <EOL> self . _assert_dscp_marking_rule_is_set ( self . ports [ <NUM_LIT:0> ] , <EOL> TEST_DSCP_MARKING_RULE_1 ) <EOL> self . _assert_dscp_marking_rule_is_set ( self . ports [ <NUM_LIT:1> ] , <EOL> TEST_DSCP_MARKING_RULE_2 ) <EOL> self . _assert_dscp_marking_rule_not_set ( self . ports [ <NUM_LIT:2> ] ) <EOL> def test_simple_port_policy_update ( self ) : <EOL> self . setup_agent_and_ports ( <EOL> port_dicts = self . create_test_ports ( amount = <NUM_LIT:1> , <EOL> policy_id = TEST_POLICY_ID1 ) ) <EOL> self . wait_until_ports_state ( self . ports , up = True ) <EOL> self . _assert_dscp_marking_rule_is_set ( self . ports [ <NUM_LIT:0> ] , <EOL> TEST_DSCP_MARKING_RULE_1 ) <EOL> policy_copy = copy . deepcopy ( self . qos_policies [ TEST_POLICY_ID1 ] ) <EOL> policy_copy . rules [ <NUM_LIT:0> ] . max_kbps = <NUM_LIT> <EOL> policy_copy . rules [ <NUM_LIT:0> ] . max_burst_kbps = <NUM_LIT:5> <EOL> policy_copy . rules [ <NUM_LIT:1> ] . dscp_mark = TEST_DSCP_MARK_2 <EOL> consumer_reg . push ( resources . QOS_POLICY , policy_copy , events . UPDATED ) <EOL> self . wait_until_bandwidth_limit_rule_applied ( self . ports [ <NUM_LIT:0> ] , <EOL> policy_copy . rules [ <NUM_LIT:0> ] ) <EOL> self . _assert_bandwidth_limit_rule_is_set ( self . ports [ <NUM_LIT:0> ] , <EOL> policy_copy . rules [ <NUM_LIT:0> ] ) <EOL> self . _assert_dscp_marking_rule_is_set ( self . ports [ <NUM_LIT:0> ] , <EOL> TEST_DSCP_MARKING_RULE_2 ) <EOL> def test_port_qos_disassociation ( self ) : <EOL> """<STR_LIT>""" <EOL> port_dict = self . _create_port_with_qos ( ) <EOL> port_dict [ '<STR_LIT>' ] = None <EOL> self . agent . port_update ( None , port = port_dict ) <EOL> self . wait_until_bandwidth_limit_rule_applied ( port_dict , None ) <EOL> def test_port_qos_update_policy_id ( self ) : <EOL> """<STR_LIT>""" <EOL> port_dict = self . _create_port_with_qos ( ) <EOL> port_dict [ '<STR_LIT>' ] = TEST_POLICY_ID2 <EOL> self . agent . port_update ( None , port = port_dict ) <EOL> self . wait_until_bandwidth_limit_rule_applied ( port_dict , <EOL> TEST_BW_LIMIT_RULE_2 ) <EOL> def test_policy_rule_delete ( self ) : <EOL> port_dict = self . _create_port_with_qos ( ) <EOL> policy_copy = copy . deepcopy ( self . qos_policies [ TEST_POLICY_ID1 ] ) <EOL> policy_copy . rules = list ( ) <EOL> consumer_reg . push ( resources . QOS_POLICY , policy_copy , events . UPDATED ) <EOL> self . wait_until_bandwidth_limit_rule_applied ( port_dict , None ) </s>
