<s> import copy <EOL> import datetime <EOL> import mock <EOL> from neutron_lib import constants <EOL> from neutron_lib import exceptions as n_exc <EOL> from oslo_config import cfg <EOL> from oslo_db import exception as exc <EOL> from oslo_utils import timeutils <EOL> import testscenarios <EOL> from neutron import context <EOL> from neutron . db import agents_db <EOL> from neutron . db import db_base_plugin_v2 as base_plugin <EOL> from neutron . tests . unit import testlib_api <EOL> """<STR_LIT>""" <EOL> load_tests = testscenarios . load_tests_apply_scenarios <EOL> TEST_RESOURCE_VERSIONS = { "<STR_LIT:A>" : "<STR_LIT:1.0>" } <EOL> AGENT_STATUS = { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:host>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : TEST_RESOURCE_VERSIONS } <EOL> TEST_TIME = '<STR_LIT>' <EOL> class FakePlugin ( base_plugin . NeutronDbPluginV2 , agents_db . AgentDbMixin ) : <EOL> """<STR_LIT>""" <EOL> class TestAgentsDbBase ( testlib_api . SqlTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( TestAgentsDbBase , self ) . setUp ( ) <EOL> self . context = context . get_admin_context ( ) <EOL> self . plugin = FakePlugin ( ) <EOL> def _get_agents ( self , hosts , agent_type ) : <EOL> return [ <EOL> agents_db . Agent ( <EOL> binary = '<STR_LIT>' , <EOL> host = host , <EOL> agent_type = agent_type , <EOL> topic = '<STR_LIT>' , <EOL> configurations = "<STR_LIT:{}>" , <EOL> resource_versions = "<STR_LIT:{}>" , <EOL> created_at = timeutils . utcnow ( ) , <EOL> started_at = timeutils . utcnow ( ) , <EOL> heartbeat_timestamp = timeutils . utcnow ( ) ) <EOL> for host in hosts <EOL> ] <EOL> def _save_agents ( self , agents ) : <EOL> for agent in agents : <EOL> with self . context . session . begin ( subtransactions = True ) : <EOL> self . context . session . add ( agent ) <EOL> def _create_and_save_agents ( self , hosts , agent_type , down_agents_count = <NUM_LIT:0> , <EOL> down_but_version_considered = <NUM_LIT:0> ) : <EOL> agents = self . _get_agents ( hosts , agent_type ) <EOL> for agent in agents [ : down_agents_count ] : <EOL> agent [ '<STR_LIT>' ] -= datetime . timedelta ( minutes = <NUM_LIT> ) <EOL> for agent in agents [ down_agents_count : ( <EOL> down_but_version_considered + down_agents_count ) ] : <EOL> agent [ '<STR_LIT>' ] -= datetime . timedelta ( <EOL> seconds = ( cfg . CONF . agent_down_time + <NUM_LIT:1> ) ) <EOL> self . _save_agents ( agents ) <EOL> return agents <EOL> class TestAgentsDbMixin ( TestAgentsDbBase ) : <EOL> def setUp ( self ) : <EOL> super ( TestAgentsDbMixin , self ) . setUp ( ) <EOL> self . agent_status = dict ( AGENT_STATUS ) <EOL> def test_get_enabled_agent_on_host_found ( self ) : <EOL> agents = self . _create_and_save_agents ( [ '<STR_LIT>' ] , <EOL> constants . AGENT_TYPE_L3 ) <EOL> expected = self . plugin . get_enabled_agent_on_host ( <EOL> self . context , constants . AGENT_TYPE_L3 , '<STR_LIT>' ) <EOL> self . assertEqual ( expected , agents [ <NUM_LIT:0> ] ) <EOL> def test_get_enabled_agent_on_host_not_found ( self ) : <EOL> with mock . patch . object ( agents_db . LOG , '<STR_LIT>' ) as mock_log : <EOL> agent = self . plugin . get_enabled_agent_on_host ( <EOL> self . context , constants . AGENT_TYPE_L3 , '<STR_LIT>' ) <EOL> self . assertIsNone ( agent ) <EOL> self . assertTrue ( mock_log . called ) <EOL> def _assert_ref_fields_are_equal ( self , reference , result ) : <EOL> """<STR_LIT>""" <EOL> for field , value in reference . items ( ) : <EOL> self . assertEqual ( value , result [ field ] , field ) <EOL> def test_create_or_update_agent_new_entry ( self ) : <EOL> self . plugin . create_or_update_agent ( self . context , self . agent_status ) <EOL> agent = self . plugin . get_agents ( self . context ) [ <NUM_LIT:0> ] <EOL> self . _assert_ref_fields_are_equal ( self . agent_status , agent ) <EOL> def test_create_or_update_agent_existing_entry ( self ) : <EOL> self . plugin . create_or_update_agent ( self . context , self . agent_status ) <EOL> self . plugin . create_or_update_agent ( self . context , self . agent_status ) <EOL> self . plugin . create_or_update_agent ( self . context , self . agent_status ) <EOL> agents = self . plugin . get_agents ( self . context ) <EOL> self . assertEqual ( len ( agents ) , <NUM_LIT:1> ) <EOL> agent = agents [ <NUM_LIT:0> ] <EOL> self . _assert_ref_fields_are_equal ( self . agent_status , agent ) <EOL> def test_create_or_update_agent_logs_heartbeat ( self ) : <EOL> status = self . agent_status . copy ( ) <EOL> status [ '<STR_LIT>' ] = { '<STR_LIT>' : True } <EOL> with mock . patch . object ( agents_db . LOG , '<STR_LIT:info>' ) as info : <EOL> self . plugin . create_or_update_agent ( self . context , status ) <EOL> self . assertTrue ( info . called ) <EOL> status [ '<STR_LIT>' ] = { '<STR_LIT>' : False } <EOL> info . reset_mock ( ) <EOL> self . plugin . create_or_update_agent ( self . context , status ) <EOL> self . assertFalse ( info . called ) <EOL> def test_create_or_update_agent_concurrent_insert ( self ) : <EOL> with mock . patch ( '<STR_LIT>' ) as add_mock : <EOL> add_mock . side_effect = [ <EOL> exc . DBDuplicateEntry ( ) , <EOL> None <EOL> ] <EOL> self . plugin . create_or_update_agent ( self . context , self . agent_status ) <EOL> self . assertEqual ( add_mock . call_count , <NUM_LIT:2> , <EOL> "<STR_LIT>" ) <EOL> def test_create_or_update_agent_disable_new_agents ( self ) : <EOL> cfg . CONF . set_override ( '<STR_LIT>' , False ) <EOL> self . plugin . create_or_update_agent ( self . context , self . agent_status ) <EOL> agent = self . plugin . get_agents ( self . context ) [ <NUM_LIT:0> ] <EOL> self . assertFalse ( agent [ '<STR_LIT>' ] ) <EOL> def test_agent_health_check ( self ) : <EOL> agents = [ { '<STR_LIT>' : "<STR_LIT>" , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:host>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : True } ] <EOL> with mock . patch . object ( self . plugin , '<STR_LIT>' , <EOL> return_value = agents ) , mock . patch . object ( agents_db . LOG , '<STR_LIT>' ) as warn , mock . patch . object ( agents_db . LOG , '<STR_LIT>' ) as debug : <EOL> self . plugin . agent_health_check ( ) <EOL> self . assertTrue ( debug . called ) <EOL> self . assertFalse ( warn . called ) <EOL> agents [ <NUM_LIT:0> ] [ '<STR_LIT>' ] = False <EOL> self . plugin . agent_health_check ( ) <EOL> warn . assert_called_once_with ( <EOL> mock . ANY , <EOL> { '<STR_LIT:count>' : <NUM_LIT:1> , '<STR_LIT>' : <NUM_LIT:1> , <EOL> '<STR_LIT:data>' : "<STR_LIT>" <EOL> "<STR_LIT>" } <EOL> ) <EOL> def test__get_dict ( self ) : <EOL> db_obj = mock . Mock ( conf1 = '<STR_LIT>' ) <EOL> conf1 = self . plugin . _get_dict ( db_obj , '<STR_LIT>' ) <EOL> self . assertIn ( '<STR_LIT:test>' , conf1 ) <EOL> self . assertEqual ( "<STR_LIT>" , conf1 [ '<STR_LIT:test>' ] ) <EOL> def test__get_dict_missing ( self ) : <EOL> with mock . patch . object ( agents_db . LOG , '<STR_LIT>' ) as warn : <EOL> db_obj = mock . Mock ( spec = [ '<STR_LIT>' , '<STR_LIT:host>' ] ) <EOL> self . plugin . _get_dict ( db_obj , '<STR_LIT>' ) <EOL> self . assertTrue ( warn . called ) <EOL> def test__get_dict_ignore_missing ( self ) : <EOL> with mock . patch . object ( agents_db . LOG , '<STR_LIT>' ) as warn : <EOL> db_obj = mock . Mock ( spec = [ '<STR_LIT>' , '<STR_LIT:host>' ] ) <EOL> missing_conf = self . plugin . _get_dict ( db_obj , '<STR_LIT>' , <EOL> ignore_missing = True ) <EOL> self . assertEqual ( { } , missing_conf ) <EOL> warn . assert_not_called ( ) <EOL> def test__get_dict_broken ( self ) : <EOL> with mock . patch . object ( agents_db . LOG , '<STR_LIT>' ) as warn : <EOL> db_obj = mock . Mock ( conf1 = '<STR_LIT>' ) <EOL> conf1 = self . plugin . _get_dict ( db_obj , '<STR_LIT>' , ignore_missing = True ) <EOL> self . assertEqual ( { } , conf1 ) <EOL> self . assertTrue ( warn . called ) <EOL> def get_configurations_dict ( self ) : <EOL> db_obj = mock . Mock ( configurations = '<STR_LIT>' ) <EOL> cfg = self . plugin . get_configuration_dict ( db_obj ) <EOL> self . assertIn ( '<STR_LIT>' , cfg ) <EOL> def test__get_agents_resource_versions ( self ) : <EOL> tracker = mock . Mock ( ) <EOL> self . _create_and_save_agents ( <EOL> [ '<STR_LIT>' % i for i in range ( <NUM_LIT:5> ) ] , <EOL> constants . AGENT_TYPE_L3 , <EOL> down_agents_count = <NUM_LIT:3> , <EOL> down_but_version_considered = <NUM_LIT:2> ) <EOL> self . plugin . _get_agents_resource_versions ( tracker ) <EOL> self . assertEqual ( tracker . set_versions . call_count , <NUM_LIT:2> ) <EOL> class TestAgentsDbGetAgents ( TestAgentsDbBase ) : <EOL> scenarios = [ <EOL> ( '<STR_LIT>' , dict ( agents = <NUM_LIT:5> , down_agents = <NUM_LIT:2> , <EOL> agents_alive = None , <EOL> expected_agents = <NUM_LIT:5> ) ) , <EOL> ( '<STR_LIT>' , dict ( agents = <NUM_LIT:5> , down_agents = <NUM_LIT:2> , <EOL> agents_alive = '<STR_LIT:True>' , <EOL> expected_agents = <NUM_LIT:3> ) ) , <EOL> ( '<STR_LIT>' , dict ( agents = <NUM_LIT:5> , down_agents = <NUM_LIT:2> , <EOL> agents_alive = '<STR_LIT:False>' , <EOL> expected_agents = <NUM_LIT:2> ) ) , <EOL> ( '<STR_LIT>' , dict ( agents = <NUM_LIT:5> , down_agents = <NUM_LIT:2> , <EOL> agents_alive = '<STR_LIT:true>' , <EOL> expected_agents = <NUM_LIT:3> ) ) , <EOL> ( '<STR_LIT>' , dict ( agents = <NUM_LIT:5> , down_agents = <NUM_LIT:2> , <EOL> agents_alive = '<STR_LIT:false>' , <EOL> expected_agents = <NUM_LIT:2> ) ) , <EOL> ( '<STR_LIT>' , dict ( agents = <NUM_LIT:5> , down_agents = <NUM_LIT:2> , <EOL> agents_alive = '<STR_LIT>' , <EOL> expected_agents = None ) ) , <EOL> ] <EOL> def setUp ( self ) : <EOL> if not hasattr ( self , '<STR_LIT>' ) : <EOL> self . __dict__ . update ( self . scenarios [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] ) <EOL> super ( TestAgentsDbGetAgents , self ) . setUp ( ) <EOL> def test_get_agents ( self ) : <EOL> hosts = [ '<STR_LIT>' % i for i in range ( self . agents ) ] <EOL> self . _create_and_save_agents ( hosts , constants . AGENT_TYPE_L3 , <EOL> down_agents_count = self . down_agents ) <EOL> if self . agents_alive == '<STR_LIT>' : <EOL> self . assertRaises ( n_exc . InvalidInput , self . plugin . get_agents , <EOL> self . context , <EOL> filters = { '<STR_LIT>' : [ self . agents_alive ] } ) <EOL> else : <EOL> returned_agents = self . plugin . get_agents ( <EOL> self . context , filters = { '<STR_LIT>' : [ self . agents_alive ] } <EOL> if self . agents_alive else None ) <EOL> self . assertEqual ( self . expected_agents , len ( returned_agents ) ) <EOL> if self . agents_alive : <EOL> alive = ( self . agents_alive == '<STR_LIT:True>' or <EOL> self . agents_alive == '<STR_LIT:true>' ) <EOL> for agent in returned_agents : <EOL> self . assertEqual ( alive , agent [ '<STR_LIT>' ] ) <EOL> class TestAgentExtRpcCallback ( TestAgentsDbBase ) : <EOL> def setUp ( self ) : <EOL> super ( TestAgentExtRpcCallback , self ) . setUp ( ) <EOL> self . callback = agents_db . AgentExtRpcCallback ( self . plugin ) <EOL> self . callback . server_versions_rpc = mock . Mock ( ) <EOL> self . versions_rpc = self . callback . server_versions_rpc <EOL> self . callback . START_TIME = datetime . datetime ( datetime . MINYEAR , <NUM_LIT:1> , <NUM_LIT:1> ) <EOL> self . update_versions = mock . patch ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) . start ( ) <EOL> self . agent_state = { '<STR_LIT>' : dict ( AGENT_STATUS ) } <EOL> def test_create_or_update_agent_updates_version_manager ( self ) : <EOL> self . callback . report_state ( self . context , agent_state = self . agent_state , <EOL> time = TEST_TIME ) <EOL> self . update_versions . assert_called_once_with ( <EOL> mock . ANY , TEST_RESOURCE_VERSIONS ) <EOL> def test_create_or_update_agent_updates_other_servers ( self ) : <EOL> callback = self . callback <EOL> callback . report_state ( self . context , agent_state = self . agent_state , <EOL> time = TEST_TIME ) <EOL> report_agent_resource_versions = ( <EOL> self . versions_rpc . report_agent_resource_versions ) <EOL> report_agent_resource_versions . assert_called_once_with ( <EOL> mock . ANY , mock . ANY , mock . ANY , TEST_RESOURCE_VERSIONS ) <EOL> def test_no_version_updates_on_further_state_reports ( self ) : <EOL> self . test_create_or_update_agent_updates_version_manager ( ) <EOL> second_agent_state = copy . deepcopy ( self . agent_state ) <EOL> second_agent_state [ '<STR_LIT>' ] . pop ( '<STR_LIT>' ) <EOL> self . update_versions . reset_mock ( ) <EOL> report_agent_resource_versions = ( <EOL> self . versions_rpc . report_agent_resource_versions ) <EOL> report_agent_resource_versions . reset_mock ( ) <EOL> self . callback . report_state ( self . context , <EOL> agent_state = second_agent_state , <EOL> time = TEST_TIME ) <EOL> self . assertFalse ( self . update_versions . called ) <EOL> self . assertFalse ( report_agent_resource_versions . called ) <EOL> def test_version_updates_on_agent_revival ( self ) : <EOL> self . test_create_or_update_agent_updates_version_manager ( ) <EOL> second_agent_state = copy . deepcopy ( self . agent_state ) <EOL> second_agent_state [ '<STR_LIT>' ] . pop ( '<STR_LIT>' ) <EOL> self . _take_down_agent ( ) <EOL> self . update_versions . reset_mock ( ) <EOL> report_agent_resource_versions = ( <EOL> self . versions_rpc . report_agent_resource_versions ) <EOL> report_agent_resource_versions . reset_mock ( ) <EOL> self . callback . report_state ( self . context , <EOL> agent_state = second_agent_state , <EOL> time = TEST_TIME ) <EOL> self . update_versions . assert_called_once_with ( <EOL> mock . ANY , TEST_RESOURCE_VERSIONS ) <EOL> report_agent_resource_versions . assert_called_once_with ( <EOL> mock . ANY , mock . ANY , mock . ANY , TEST_RESOURCE_VERSIONS ) <EOL> def _take_down_agent ( self ) : <EOL> with self . context . session . begin ( subtransactions = True ) : <EOL> query = self . context . session . query ( agents_db . Agent ) <EOL> agt = query . first ( ) <EOL> agt . heartbeat_timestamp = ( <EOL> agt . heartbeat_timestamp - datetime . timedelta ( hours = <NUM_LIT:1> ) ) </s>
