<s> from neutron_lib import constants <EOL> from neutron import context <EOL> from neutron . db import models_v2 <EOL> from neutron . extensions import portbindings <EOL> from neutron . plugins . ml2 . drivers . l2pop import db as l2pop_db <EOL> from neutron . plugins . ml2 import models <EOL> from neutron . tests . common import helpers <EOL> from neutron . tests . unit import testlib_api <EOL> class TestL2PopulationDBTestCase ( testlib_api . SqlTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( TestL2PopulationDBTestCase , self ) . setUp ( ) <EOL> self . ctx = context . get_admin_context ( ) <EOL> def test_get_agent_by_host ( self ) : <EOL> helpers . register_l3_agent ( ) <EOL> helpers . register_dhcp_agent ( ) <EOL> helpers . register_ovs_agent ( ) <EOL> agent = l2pop_db . get_agent_by_host ( <EOL> self . ctx . session , helpers . HOST ) <EOL> self . assertEqual ( constants . AGENT_TYPE_OVS , agent . agent_type ) <EOL> def test_get_agent_by_host_no_candidate ( self ) : <EOL> helpers . register_l3_agent ( ) <EOL> helpers . register_dhcp_agent ( ) <EOL> agent = l2pop_db . get_agent_by_host ( <EOL> self . ctx . session , helpers . HOST ) <EOL> self . assertIsNone ( agent ) <EOL> def _setup_port_binding ( self , network_id = '<STR_LIT>' , dvr = True ) : <EOL> with self . ctx . session . begin ( subtransactions = True ) : <EOL> self . ctx . session . add ( models_v2 . Network ( id = network_id ) ) <EOL> device_owner = constants . DEVICE_OWNER_DVR_INTERFACE if dvr else '<STR_LIT>' <EOL> self . ctx . session . add ( models_v2 . Port ( <EOL> id = '<STR_LIT>' , <EOL> network_id = network_id , <EOL> mac_address = '<STR_LIT>' , <EOL> admin_state_up = True , <EOL> status = constants . PORT_STATUS_ACTIVE , <EOL> device_id = '<STR_LIT>' , <EOL> device_owner = device_owner ) ) <EOL> port_binding_cls = ( models . DVRPortBinding if dvr <EOL> else models . PortBinding ) <EOL> binding_kwarg = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:host>' : helpers . HOST , <EOL> '<STR_LIT>' : portbindings . VIF_TYPE_UNBOUND , <EOL> '<STR_LIT>' : portbindings . VNIC_NORMAL <EOL> } <EOL> if dvr : <EOL> binding_kwarg [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> binding_kwarg [ '<STR_LIT:status>' ] = constants . PORT_STATUS_DOWN <EOL> self . ctx . session . add ( port_binding_cls ( ** binding_kwarg ) ) <EOL> def test_get_dvr_active_network_ports ( self ) : <EOL> self . _setup_port_binding ( ) <EOL> helpers . register_l3_agent ( ) <EOL> helpers . register_dhcp_agent ( ) <EOL> helpers . register_ovs_agent ( ) <EOL> tunnel_network_ports = l2pop_db . get_dvr_active_network_ports ( <EOL> self . ctx . session , '<STR_LIT>' ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( tunnel_network_ports ) ) <EOL> _ , agent = tunnel_network_ports [ <NUM_LIT:0> ] <EOL> self . assertEqual ( constants . AGENT_TYPE_OVS , agent . agent_type ) <EOL> def test_get_dvr_active_network_ports_no_candidate ( self ) : <EOL> self . _setup_port_binding ( ) <EOL> helpers . register_l3_agent ( ) <EOL> helpers . register_dhcp_agent ( ) <EOL> tunnel_network_ports = l2pop_db . get_dvr_active_network_ports ( <EOL> self . ctx . session , '<STR_LIT>' ) <EOL> self . assertEqual ( <NUM_LIT:0> , len ( tunnel_network_ports ) ) <EOL> def test_get_nondvr_active_network_ports ( self ) : <EOL> self . _setup_port_binding ( dvr = False ) <EOL> helpers . register_l3_agent ( ) <EOL> helpers . register_dhcp_agent ( ) <EOL> helpers . register_ovs_agent ( ) <EOL> fdb_network_ports = l2pop_db . get_nondvr_active_network_ports ( <EOL> self . ctx . session , '<STR_LIT>' ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( fdb_network_ports ) ) <EOL> _ , agent = fdb_network_ports [ <NUM_LIT:0> ] <EOL> self . assertEqual ( constants . AGENT_TYPE_OVS , agent . agent_type ) <EOL> def test_get_nondvr_active_network_ports_no_candidate ( self ) : <EOL> self . _setup_port_binding ( dvr = False ) <EOL> helpers . register_l3_agent ( ) <EOL> helpers . register_dhcp_agent ( ) <EOL> fdb_network_ports = l2pop_db . get_nondvr_active_network_ports ( <EOL> self . ctx . session , '<STR_LIT>' ) <EOL> self . assertEqual ( <NUM_LIT:0> , len ( fdb_network_ports ) ) </s>
