<s> import mock <EOL> from oslo_utils import uuidutils <EOL> from neutron import context <EOL> from neutron . db . quota import api as quota_db_api <EOL> from neutron . tests . unit . api import test_extensions <EOL> from neutron . tests . unit . extensions import test_l3 <EOL> from neutron . tests . unit . extensions import test_securitygroup <EOL> from neutron . tests . unit . plugins . ml2 import base as ml2_base <EOL> from neutron . tests . unit . plugins . ml2 import test_plugin <EOL> class SgTestCaseWrapper ( test_securitygroup . SecurityGroupDBTestCase ) : <EOL> def setUp ( self , plugin , ** kwargs ) : <EOL> super ( SgTestCaseWrapper , self ) . setUp ( plugin ) <EOL> class BaseTestTrackedResources ( test_plugin . Ml2PluginV2TestCase , <EOL> SgTestCaseWrapper ) : <EOL> def setUp ( self ) : <EOL> self . ctx = context . get_admin_context ( ) <EOL> def_sec_group_patch = mock . patch ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> def_sec_group_patch . start ( ) <EOL> get_sec_group_port_patch = mock . patch ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> get_sec_group_port_patch . start ( ) <EOL> super ( BaseTestTrackedResources , self ) . setUp ( ) <EOL> self . _tenant_id = uuidutils . generate_uuid ( ) <EOL> def _test_init ( self , resource_name ) : <EOL> quota_db_api . set_quota_usage ( <EOL> self . ctx , resource_name , self . _tenant_id ) <EOL> class BaseTestEventHandler ( object ) : <EOL> def setUp ( self ) : <EOL> handler_patch = mock . patch ( <EOL> '<STR_LIT>' ) <EOL> self . handler_mock = handler_patch . start ( ) <EOL> super ( BaseTestEventHandler , self ) . setUp ( ) <EOL> def _verify_event_handler_calls ( self , data , expected_call_count = <NUM_LIT:1> ) : <EOL> if not hasattr ( data , '<STR_LIT>' ) or isinstance ( data , dict ) : <EOL> data = [ data ] <EOL> self . assertEqual ( expected_call_count , self . handler_mock . call_count ) <EOL> call_idx = - <NUM_LIT:1> <EOL> for item in data : <EOL> if item : <EOL> model = self . handler_mock . call_args_list [ call_idx ] [ <NUM_LIT:0> ] [ - <NUM_LIT:1> ] <EOL> self . assertEqual ( model [ '<STR_LIT:id>' ] , item [ '<STR_LIT:id>' ] ) <EOL> self . assertEqual ( model [ '<STR_LIT>' ] , item [ '<STR_LIT>' ] ) <EOL> call_idx = call_idx - <NUM_LIT:1> <EOL> class TestTrackedResourcesEventHandler ( BaseTestEventHandler , <EOL> BaseTestTrackedResources ) : <EOL> def test_create_delete_network_triggers_event ( self ) : <EOL> self . _test_init ( '<STR_LIT>' ) <EOL> net = self . _make_network ( '<STR_LIT>' , '<STR_LIT>' , True ) [ '<STR_LIT>' ] <EOL> self . _verify_event_handler_calls ( net ) <EOL> self . _delete ( '<STR_LIT>' , net [ '<STR_LIT:id>' ] ) <EOL> self . _verify_event_handler_calls ( net , expected_call_count = <NUM_LIT:2> ) <EOL> def test_create_delete_port_triggers_event ( self ) : <EOL> self . _test_init ( '<STR_LIT:port>' ) <EOL> net = self . _make_network ( '<STR_LIT>' , '<STR_LIT>' , True ) [ '<STR_LIT>' ] <EOL> port = self . _make_port ( '<STR_LIT>' , net [ '<STR_LIT:id>' ] ) [ '<STR_LIT:port>' ] <EOL> self . _verify_event_handler_calls ( port , expected_call_count = <NUM_LIT:2> ) <EOL> self . _delete ( '<STR_LIT>' , port [ '<STR_LIT:id>' ] ) <EOL> self . _verify_event_handler_calls ( port , expected_call_count = <NUM_LIT:3> ) <EOL> def test_create_delete_subnet_triggers_event ( self ) : <EOL> self . _test_init ( '<STR_LIT>' ) <EOL> net = self . _make_network ( '<STR_LIT>' , '<STR_LIT>' , True ) <EOL> subnet = self . _make_subnet ( '<STR_LIT>' , net , '<STR_LIT>' , <EOL> '<STR_LIT>' ) [ '<STR_LIT>' ] <EOL> self . _verify_event_handler_calls ( [ subnet , net [ '<STR_LIT>' ] ] , <EOL> expected_call_count = <NUM_LIT:2> ) <EOL> self . _delete ( '<STR_LIT>' , subnet [ '<STR_LIT:id>' ] ) <EOL> self . _verify_event_handler_calls ( subnet , expected_call_count = <NUM_LIT:3> ) <EOL> def test_create_delete_network_with_subnet_triggers_event ( self ) : <EOL> self . _test_init ( '<STR_LIT>' ) <EOL> self . _test_init ( '<STR_LIT>' ) <EOL> net = self . _make_network ( '<STR_LIT>' , '<STR_LIT>' , True ) <EOL> subnet = self . _make_subnet ( '<STR_LIT>' , net , '<STR_LIT>' , <EOL> '<STR_LIT>' ) [ '<STR_LIT>' ] <EOL> self . _verify_event_handler_calls ( [ subnet , net [ '<STR_LIT>' ] ] , <EOL> expected_call_count = <NUM_LIT:2> ) <EOL> self . _delete ( '<STR_LIT>' , net [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] ) <EOL> self . _verify_event_handler_calls ( [ net [ '<STR_LIT>' ] , subnet ] , <EOL> expected_call_count = <NUM_LIT:4> ) <EOL> def test_create_delete_subnetpool_triggers_event ( self ) : <EOL> self . _test_init ( '<STR_LIT>' ) <EOL> pool = self . _make_subnetpool ( '<STR_LIT>' , [ '<STR_LIT>' ] , <EOL> name = '<STR_LIT>' , <EOL> tenant_id = self . _tenant_id ) [ '<STR_LIT>' ] <EOL> self . _verify_event_handler_calls ( pool ) <EOL> self . _delete ( '<STR_LIT>' , pool [ '<STR_LIT:id>' ] ) <EOL> self . _verify_event_handler_calls ( pool , expected_call_count = <NUM_LIT:2> ) <EOL> def test_create_delete_securitygroup_triggers_event ( self ) : <EOL> self . _test_init ( '<STR_LIT>' ) <EOL> sec_group = self . _make_security_group ( <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , tenant_id = self . _tenant_id ) [ '<STR_LIT>' ] <EOL> self . _verify_event_handler_calls ( [ None , None , sec_group ] , <EOL> expected_call_count = <NUM_LIT:3> ) <EOL> self . _delete ( '<STR_LIT>' , sec_group [ '<STR_LIT:id>' ] ) <EOL> self . _verify_event_handler_calls ( sec_group , expected_call_count = <NUM_LIT:6> ) <EOL> def test_create_delete_securitygrouprule_triggers_event ( self ) : <EOL> self . _test_init ( '<STR_LIT>' ) <EOL> sec_group = self . _make_security_group ( <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , tenant_id = self . _tenant_id ) [ '<STR_LIT>' ] <EOL> rule_req = self . _build_security_group_rule ( <EOL> sec_group [ '<STR_LIT:id>' ] , '<STR_LIT>' , '<STR_LIT>' , tenant_id = self . _tenant_id ) <EOL> sec_group_rule = self . _make_security_group_rule ( <EOL> '<STR_LIT>' , rule_req ) [ '<STR_LIT>' ] <EOL> self . _verify_event_handler_calls ( sec_group_rule , expected_call_count = <NUM_LIT:4> ) <EOL> self . _delete ( '<STR_LIT>' , sec_group_rule [ '<STR_LIT:id>' ] ) <EOL> self . _verify_event_handler_calls ( sec_group_rule , expected_call_count = <NUM_LIT:5> ) <EOL> class TestL3ResourcesEventHandler ( BaseTestEventHandler , <EOL> ml2_base . ML2TestFramework , <EOL> test_l3 . L3NatTestCaseMixin ) : <EOL> def setUp ( self ) : <EOL> super ( TestL3ResourcesEventHandler , self ) . setUp ( ) <EOL> ext_mgr = test_l3 . L3TestExtensionManager ( ) <EOL> self . ext_api = test_extensions . setup_extensions_middleware ( ext_mgr ) <EOL> def test_create_delete_floating_ip_triggers_event ( self ) : <EOL> net = self . _make_network ( '<STR_LIT>' , '<STR_LIT>' , True ) <EOL> subnet = self . _make_subnet ( '<STR_LIT>' , net , '<STR_LIT>' , <EOL> '<STR_LIT>' ) [ '<STR_LIT>' ] <EOL> self . _set_net_external ( subnet [ '<STR_LIT>' ] ) <EOL> floatingip = self . _make_floatingip ( '<STR_LIT>' , subnet [ '<STR_LIT>' ] ) <EOL> internal_port = self . _show ( <EOL> '<STR_LIT>' , floatingip [ '<STR_LIT>' ] [ '<STR_LIT>' ] ) [ '<STR_LIT>' ] [ <NUM_LIT:0> ] <EOL> self . _verify_event_handler_calls ( floatingip [ '<STR_LIT>' ] , <EOL> expected_call_count = <NUM_LIT:4> ) <EOL> self . _delete ( '<STR_LIT>' , floatingip [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] ) <EOL> self . _verify_event_handler_calls ( <EOL> [ internal_port , floatingip [ '<STR_LIT>' ] ] , expected_call_count = <NUM_LIT:6> ) <EOL> class TestTrackedResources ( BaseTestTrackedResources ) : <EOL> def _verify_dirty_bit ( self , resource_name , expected_value = True ) : <EOL> usage = quota_db_api . get_quota_usage_by_resource_and_tenant ( <EOL> self . ctx , resource_name , self . _tenant_id ) <EOL> self . assertEqual ( expected_value , usage . dirty ) <EOL> def test_create_delete_network_marks_dirty ( self ) : <EOL> self . _test_init ( '<STR_LIT>' ) <EOL> net = self . _make_network ( '<STR_LIT>' , '<STR_LIT>' , True ) [ '<STR_LIT>' ] <EOL> self . _verify_dirty_bit ( '<STR_LIT>' ) <EOL> quota_db_api . set_quota_usage_dirty ( <EOL> self . ctx , '<STR_LIT>' , self . _tenant_id , dirty = False ) <EOL> self . _delete ( '<STR_LIT>' , net [ '<STR_LIT:id>' ] ) <EOL> self . _verify_dirty_bit ( '<STR_LIT>' ) <EOL> def test_list_networks_clears_dirty ( self ) : <EOL> self . _test_init ( '<STR_LIT>' ) <EOL> net = self . _make_network ( '<STR_LIT>' , '<STR_LIT>' , True ) [ '<STR_LIT>' ] <EOL> self . ctx . tenant_id = net [ '<STR_LIT>' ] <EOL> self . _list ( '<STR_LIT>' , neutron_context = self . ctx ) <EOL> self . _verify_dirty_bit ( '<STR_LIT>' , expected_value = False ) <EOL> def test_create_delete_port_marks_dirty ( self ) : <EOL> self . _test_init ( '<STR_LIT:port>' ) <EOL> net = self . _make_network ( '<STR_LIT>' , '<STR_LIT>' , True ) [ '<STR_LIT>' ] <EOL> port = self . _make_port ( '<STR_LIT>' , net [ '<STR_LIT:id>' ] ) [ '<STR_LIT:port>' ] <EOL> self . _verify_dirty_bit ( '<STR_LIT:port>' ) <EOL> quota_db_api . set_quota_usage_dirty ( <EOL> self . ctx , '<STR_LIT:port>' , self . _tenant_id , dirty = False ) <EOL> self . _delete ( '<STR_LIT>' , port [ '<STR_LIT:id>' ] ) <EOL> self . _verify_dirty_bit ( '<STR_LIT:port>' ) <EOL> def test_list_ports_clears_dirty ( self ) : <EOL> self . _test_init ( '<STR_LIT:port>' ) <EOL> net = self . _make_network ( '<STR_LIT>' , '<STR_LIT>' , True ) [ '<STR_LIT>' ] <EOL> port = self . _make_port ( '<STR_LIT>' , net [ '<STR_LIT:id>' ] ) [ '<STR_LIT:port>' ] <EOL> self . ctx . tenant_id = port [ '<STR_LIT>' ] <EOL> self . _list ( '<STR_LIT>' , neutron_context = self . ctx ) <EOL> self . _verify_dirty_bit ( '<STR_LIT:port>' , expected_value = False ) <EOL> def test_create_delete_subnet_marks_dirty ( self ) : <EOL> self . _test_init ( '<STR_LIT>' ) <EOL> net = self . _make_network ( '<STR_LIT>' , '<STR_LIT>' , True ) <EOL> subnet = self . _make_subnet ( '<STR_LIT>' , net , '<STR_LIT>' , <EOL> '<STR_LIT>' ) [ '<STR_LIT>' ] <EOL> self . _verify_dirty_bit ( '<STR_LIT>' ) <EOL> quota_db_api . set_quota_usage_dirty ( <EOL> self . ctx , '<STR_LIT>' , self . _tenant_id , dirty = False ) <EOL> self . _delete ( '<STR_LIT>' , subnet [ '<STR_LIT:id>' ] ) <EOL> self . _verify_dirty_bit ( '<STR_LIT>' ) <EOL> def test_create_delete_network_with_subnet_marks_dirty ( self ) : <EOL> self . _test_init ( '<STR_LIT>' ) <EOL> self . _test_init ( '<STR_LIT>' ) <EOL> net = self . _make_network ( '<STR_LIT>' , '<STR_LIT>' , True ) <EOL> self . _make_subnet ( '<STR_LIT>' , net , '<STR_LIT>' , <EOL> '<STR_LIT>' ) [ '<STR_LIT>' ] <EOL> self . _verify_dirty_bit ( '<STR_LIT>' ) <EOL> quota_db_api . set_quota_usage_dirty ( <EOL> self . ctx , '<STR_LIT>' , self . _tenant_id , dirty = False ) <EOL> self . _delete ( '<STR_LIT>' , net [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] ) <EOL> self . _verify_dirty_bit ( '<STR_LIT>' ) <EOL> self . _verify_dirty_bit ( '<STR_LIT>' ) <EOL> def test_list_subnets_clears_dirty ( self ) : <EOL> self . _test_init ( '<STR_LIT>' ) <EOL> net = self . _make_network ( '<STR_LIT>' , '<STR_LIT>' , True ) <EOL> subnet = self . _make_subnet ( '<STR_LIT>' , net , '<STR_LIT>' , <EOL> '<STR_LIT>' ) [ '<STR_LIT>' ] <EOL> self . ctx . tenant_id = subnet [ '<STR_LIT>' ] <EOL> self . _list ( '<STR_LIT>' , neutron_context = self . ctx ) <EOL> self . _verify_dirty_bit ( '<STR_LIT>' , expected_value = False ) <EOL> def test_create_delete_subnetpool_marks_dirty ( self ) : <EOL> self . _test_init ( '<STR_LIT>' ) <EOL> pool = self . _make_subnetpool ( '<STR_LIT>' , [ '<STR_LIT>' ] , <EOL> name = '<STR_LIT>' , <EOL> tenant_id = self . _tenant_id ) [ '<STR_LIT>' ] <EOL> self . _verify_dirty_bit ( '<STR_LIT>' ) <EOL> quota_db_api . set_quota_usage_dirty ( <EOL> self . ctx , '<STR_LIT>' , self . _tenant_id , dirty = False ) <EOL> self . _delete ( '<STR_LIT>' , pool [ '<STR_LIT:id>' ] ) <EOL> self . _verify_dirty_bit ( '<STR_LIT>' ) <EOL> def test_list_subnetpools_clears_dirty ( self ) : <EOL> self . _test_init ( '<STR_LIT>' ) <EOL> pool = self . _make_subnetpool ( '<STR_LIT>' , [ '<STR_LIT>' ] , <EOL> name = '<STR_LIT>' , <EOL> tenant_id = self . _tenant_id ) [ '<STR_LIT>' ] <EOL> self . ctx . tenant_id = pool [ '<STR_LIT>' ] <EOL> self . _list ( '<STR_LIT>' , neutron_context = self . ctx ) <EOL> self . _verify_dirty_bit ( '<STR_LIT>' , expected_value = False ) <EOL> def test_create_delete_securitygroup_marks_dirty ( self ) : <EOL> self . _test_init ( '<STR_LIT>' ) <EOL> sec_group = self . _make_security_group ( <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , tenant_id = self . _tenant_id ) [ '<STR_LIT>' ] <EOL> self . _verify_dirty_bit ( '<STR_LIT>' ) <EOL> quota_db_api . set_quota_usage_dirty ( <EOL> self . ctx , '<STR_LIT>' , self . _tenant_id , dirty = False ) <EOL> self . _delete ( '<STR_LIT>' , sec_group [ '<STR_LIT:id>' ] ) <EOL> self . _verify_dirty_bit ( '<STR_LIT>' ) <EOL> def test_list_securitygroups_clears_dirty ( self ) : <EOL> self . _test_init ( '<STR_LIT>' ) <EOL> self . _make_security_group ( <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , tenant_id = self . _tenant_id ) [ '<STR_LIT>' ] <EOL> self . ctx . tenant_id = self . _tenant_id <EOL> self . _list ( '<STR_LIT>' , neutron_context = self . ctx ) <EOL> self . _verify_dirty_bit ( '<STR_LIT>' , expected_value = False ) <EOL> def test_create_delete_securitygrouprule_marks_dirty ( self ) : <EOL> self . _test_init ( '<STR_LIT>' ) <EOL> sec_group = self . _make_security_group ( <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , tenant_id = self . _tenant_id ) [ '<STR_LIT>' ] <EOL> rule_req = self . _build_security_group_rule ( <EOL> sec_group [ '<STR_LIT:id>' ] , '<STR_LIT>' , '<STR_LIT>' , tenant_id = self . _tenant_id ) <EOL> sec_group_rule = self . _make_security_group_rule ( <EOL> '<STR_LIT>' , rule_req ) [ '<STR_LIT>' ] <EOL> self . _verify_dirty_bit ( '<STR_LIT>' ) <EOL> quota_db_api . set_quota_usage_dirty ( <EOL> self . ctx , '<STR_LIT>' , self . _tenant_id , dirty = False ) <EOL> self . _delete ( '<STR_LIT>' , sec_group_rule [ '<STR_LIT:id>' ] ) <EOL> self . _verify_dirty_bit ( '<STR_LIT>' ) <EOL> def test_list_securitygrouprules_clears_dirty ( self ) : <EOL> self . _test_init ( '<STR_LIT>' ) <EOL> self . _make_security_group ( <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , tenant_id = self . _tenant_id ) [ '<STR_LIT>' ] <EOL> self . ctx . tenant_id = self . _tenant_id <EOL> self . _list ( '<STR_LIT>' , neutron_context = self . ctx ) <EOL> self . _verify_dirty_bit ( '<STR_LIT>' , expected_value = False ) </s>
