<s> import copy <EOL> from datetime import datetime <EOL> import time <EOL> from neutron_lib import constants <EOL> from oslo_config import cfg <EOL> from oslo_utils import uuidutils <EOL> from webob import exc <EOL> from neutron . api . v2 import attributes <EOL> from neutron . common import constants as n_const <EOL> from neutron import context <EOL> from neutron . db import agents_db <EOL> from neutron . db import db_base_plugin_v2 <EOL> from neutron . extensions import agent <EOL> from neutron . tests . common import helpers <EOL> from neutron . tests import tools <EOL> from neutron . tests . unit . api . v2 import test_base <EOL> from neutron . tests . unit . db import test_db_base_plugin_v2 <EOL> _uuid = uuidutils . generate_uuid <EOL> _get_path = test_base . _get_path <EOL> L3_HOSTA = '<STR_LIT>' <EOL> DHCP_HOSTA = '<STR_LIT>' <EOL> L3_HOSTB = '<STR_LIT>' <EOL> DHCP_HOSTC = '<STR_LIT>' <EOL> LBAAS_HOSTA = '<STR_LIT>' <EOL> LBAAS_HOSTB = '<STR_LIT>' <EOL> class AgentTestExtensionManager ( object ) : <EOL> def get_resources ( self ) : <EOL> attributes . RESOURCE_ATTRIBUTE_MAP . update ( <EOL> agent . RESOURCE_ATTRIBUTE_MAP ) <EOL> return agent . Agent . get_resources ( ) <EOL> def get_actions ( self ) : <EOL> return [ ] <EOL> def get_request_extensions ( self ) : <EOL> return [ ] <EOL> class TestAgentPlugin ( db_base_plugin_v2 . NeutronDbPluginV2 , <EOL> agents_db . AgentDbMixin ) : <EOL> supported_extension_aliases = [ "<STR_LIT>" ] <EOL> class AgentDBTestMixIn ( object ) : <EOL> def _list_agents ( self , expected_res_status = None , <EOL> neutron_context = None , <EOL> query_string = None ) : <EOL> agent_res = self . _list ( '<STR_LIT>' , <EOL> neutron_context = neutron_context , <EOL> query_params = query_string ) <EOL> if expected_res_status : <EOL> self . assertEqual ( expected_res_status , agent_res . status_int ) <EOL> return agent_res <EOL> def _register_agent_states ( self , lbaas_agents = False ) : <EOL> """<STR_LIT>""" <EOL> l3_hosta = helpers . _get_l3_agent_dict ( <EOL> L3_HOSTA , n_const . L3_AGENT_MODE_LEGACY ) <EOL> l3_hostb = helpers . _get_l3_agent_dict ( <EOL> L3_HOSTB , n_const . L3_AGENT_MODE_LEGACY ) <EOL> dhcp_hosta = helpers . _get_dhcp_agent_dict ( DHCP_HOSTA ) <EOL> dhcp_hostc = helpers . _get_dhcp_agent_dict ( DHCP_HOSTC ) <EOL> helpers . register_l3_agent ( host = L3_HOSTA ) <EOL> helpers . register_l3_agent ( host = L3_HOSTB ) <EOL> helpers . register_dhcp_agent ( host = DHCP_HOSTA ) <EOL> helpers . register_dhcp_agent ( host = DHCP_HOSTC ) <EOL> res = [ l3_hosta , l3_hostb , dhcp_hosta , dhcp_hostc ] <EOL> if lbaas_agents : <EOL> lbaas_hosta = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:host>' : LBAAS_HOSTA , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : { '<STR_LIT>' : [ '<STR_LIT>' ] } , <EOL> '<STR_LIT>' : constants . AGENT_TYPE_LOADBALANCER } <EOL> lbaas_hostb = copy . deepcopy ( lbaas_hosta ) <EOL> lbaas_hostb [ '<STR_LIT:host>' ] = LBAAS_HOSTB <EOL> callback = agents_db . AgentExtRpcCallback ( ) <EOL> callback . report_state ( <EOL> self . adminContext , <EOL> agent_state = { '<STR_LIT>' : lbaas_hosta } , <EOL> time = datetime . utcnow ( ) . strftime ( constants . ISO8601_TIME_FORMAT ) ) <EOL> callback . report_state ( <EOL> self . adminContext , <EOL> agent_state = { '<STR_LIT>' : lbaas_hostb } , <EOL> time = datetime . utcnow ( ) . strftime ( constants . ISO8601_TIME_FORMAT ) ) <EOL> res += [ lbaas_hosta , lbaas_hostb ] <EOL> return res <EOL> def _register_dvr_agents ( self ) : <EOL> dvr_snat_agent = helpers . register_l3_agent ( <EOL> host = L3_HOSTA , agent_mode = n_const . L3_AGENT_MODE_DVR_SNAT ) <EOL> dvr_agent = helpers . register_l3_agent ( <EOL> host = L3_HOSTB , agent_mode = n_const . L3_AGENT_MODE_DVR ) <EOL> return [ dvr_snat_agent , dvr_agent ] <EOL> def _register_l3_agent ( self , host ) : <EOL> helpers . register_l3_agent ( host ) <EOL> class AgentDBTestCase ( AgentDBTestMixIn , <EOL> test_db_base_plugin_v2 . NeutronDbPluginV2TestCase ) : <EOL> fmt = '<STR_LIT>' <EOL> def setUp ( self ) : <EOL> plugin = '<STR_LIT>' <EOL> cfg . CONF . set_default ( '<STR_LIT>' , True ) <EOL> self . useFixture ( tools . AttributeMapMemento ( ) ) <EOL> ext_mgr = AgentTestExtensionManager ( ) <EOL> super ( AgentDBTestCase , self ) . setUp ( plugin = plugin , ext_mgr = ext_mgr ) <EOL> self . adminContext = context . get_admin_context ( ) <EOL> def test_create_agent ( self ) : <EOL> data = { '<STR_LIT>' : { } } <EOL> _req = self . new_create_request ( '<STR_LIT>' , data , self . fmt ) <EOL> _req . environ [ '<STR_LIT>' ] = context . Context ( <EOL> '<STR_LIT>' , '<STR_LIT>' ) <EOL> res = _req . get_response ( self . ext_api ) <EOL> self . assertEqual ( exc . HTTPBadRequest . code , res . status_int ) <EOL> def test_list_agent ( self ) : <EOL> agents = self . _register_agent_states ( ) <EOL> res = self . _list ( '<STR_LIT>' ) <EOL> self . assertEqual ( len ( agents ) , len ( res [ '<STR_LIT>' ] ) ) <EOL> def test_show_agent ( self ) : <EOL> self . _register_agent_states ( ) <EOL> agents = self . _list_agents ( <EOL> query_string = '<STR_LIT>' ) <EOL> self . assertEqual ( <NUM_LIT:2> , len ( agents [ '<STR_LIT>' ] ) ) <EOL> agent = self . _show ( '<STR_LIT>' , agents [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT:id>' ] ) <EOL> self . assertEqual ( '<STR_LIT>' , agent [ '<STR_LIT>' ] [ '<STR_LIT>' ] ) <EOL> def test_update_agent ( self ) : <EOL> self . _register_agent_states ( ) <EOL> agents = self . _list_agents ( <EOL> query_string = '<STR_LIT>' + L3_HOSTB ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( agents [ '<STR_LIT>' ] ) ) <EOL> com_id = agents [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT:id>' ] <EOL> agent = self . _show ( '<STR_LIT>' , com_id ) <EOL> new_agent = { } <EOL> new_agent [ '<STR_LIT>' ] = { } <EOL> new_agent [ '<STR_LIT>' ] [ '<STR_LIT>' ] = False <EOL> new_agent [ '<STR_LIT>' ] [ '<STR_LIT:description>' ] = '<STR_LIT:description>' <EOL> self . _update ( '<STR_LIT>' , com_id , new_agent ) <EOL> agent = self . _show ( '<STR_LIT>' , com_id ) <EOL> self . assertFalse ( agent [ '<STR_LIT>' ] [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( '<STR_LIT:description>' , agent [ '<STR_LIT>' ] [ '<STR_LIT:description>' ] ) <EOL> def test_dead_agent ( self ) : <EOL> cfg . CONF . set_override ( '<STR_LIT>' , <NUM_LIT:1> ) <EOL> self . _register_agent_states ( ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> agents = self . _list_agents ( <EOL> query_string = '<STR_LIT>' + L3_HOSTB ) <EOL> self . assertFalse ( agents [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) </s>
