<s> """<STR_LIT>""" <EOL> import datetime <EOL> import mock <EOL> from oslo_utils import timeutils <EOL> from ironic . common import exception <EOL> from ironic . tests . unit . db import base <EOL> from ironic . tests . unit . db import utils <EOL> class DbConductorTestCase ( base . DbTestCase ) : <EOL> def test_register_conductor_existing_fails ( self ) : <EOL> c = utils . get_test_conductor ( ) <EOL> self . dbapi . register_conductor ( c ) <EOL> self . assertRaises ( <EOL> exception . ConductorAlreadyRegistered , <EOL> self . dbapi . register_conductor , <EOL> c ) <EOL> def test_register_conductor_override ( self ) : <EOL> c = utils . get_test_conductor ( ) <EOL> self . dbapi . register_conductor ( c ) <EOL> self . dbapi . register_conductor ( c , update_existing = True ) <EOL> def _create_test_cdr ( self , ** kwargs ) : <EOL> c = utils . get_test_conductor ( ** kwargs ) <EOL> return self . dbapi . register_conductor ( c ) <EOL> def test_get_conductor ( self ) : <EOL> c1 = self . _create_test_cdr ( ) <EOL> c2 = self . dbapi . get_conductor ( c1 . hostname ) <EOL> self . assertEqual ( c1 . id , c2 . id ) <EOL> def test_get_conductor_not_found ( self ) : <EOL> self . _create_test_cdr ( ) <EOL> self . assertRaises ( <EOL> exception . ConductorNotFound , <EOL> self . dbapi . get_conductor , <EOL> '<STR_LIT>' ) <EOL> def test_unregister_conductor ( self ) : <EOL> c = self . _create_test_cdr ( ) <EOL> self . dbapi . unregister_conductor ( c . hostname ) <EOL> self . assertRaises ( <EOL> exception . ConductorNotFound , <EOL> self . dbapi . unregister_conductor , <EOL> c . hostname ) <EOL> @ mock . patch . object ( timeutils , '<STR_LIT>' , autospec = True ) <EOL> def test_touch_conductor ( self , mock_utcnow ) : <EOL> test_time = datetime . datetime ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:0> , <NUM_LIT:0> ) <EOL> mock_utcnow . return_value = test_time <EOL> c = self . _create_test_cdr ( ) <EOL> self . assertEqual ( test_time , timeutils . normalize_time ( c . updated_at ) ) <EOL> test_time = datetime . datetime ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:0> , <NUM_LIT:1> ) <EOL> mock_utcnow . return_value = test_time <EOL> self . dbapi . touch_conductor ( c . hostname ) <EOL> c = self . dbapi . get_conductor ( c . hostname ) <EOL> self . assertEqual ( test_time , timeutils . normalize_time ( c . updated_at ) ) <EOL> def test_touch_conductor_not_found ( self ) : <EOL> self . _create_test_cdr ( ) <EOL> self . assertRaises ( <EOL> exception . ConductorNotFound , <EOL> self . dbapi . touch_conductor , <EOL> '<STR_LIT>' ) <EOL> def test_touch_offline_conductor ( self ) : <EOL> c = self . _create_test_cdr ( ) <EOL> self . dbapi . unregister_conductor ( c . hostname ) <EOL> self . assertRaises ( <EOL> exception . ConductorNotFound , <EOL> self . dbapi . get_conductor , <EOL> c . hostname ) <EOL> self . dbapi . touch_conductor ( c . hostname ) <EOL> self . dbapi . get_conductor ( c . hostname ) <EOL> def test_clear_node_reservations_for_conductor ( self ) : <EOL> node1 = self . dbapi . create_node ( { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> node2 = self . dbapi . create_node ( { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> node3 = self . dbapi . create_node ( { '<STR_LIT>' : None } ) <EOL> self . dbapi . clear_node_reservations_for_conductor ( '<STR_LIT>' ) <EOL> node1 = self . dbapi . get_node_by_id ( node1 . id ) <EOL> node2 = self . dbapi . get_node_by_id ( node2 . id ) <EOL> node3 = self . dbapi . get_node_by_id ( node3 . id ) <EOL> self . assertIsNone ( node1 . reservation ) <EOL> self . assertEqual ( '<STR_LIT>' , node2 . reservation ) <EOL> self . assertIsNone ( node3 . reservation ) <EOL> @ mock . patch . object ( timeutils , '<STR_LIT>' , autospec = True ) <EOL> def test_get_active_driver_dict_one_host_no_driver ( self , mock_utcnow ) : <EOL> h = '<STR_LIT>' <EOL> expected = { } <EOL> mock_utcnow . return_value = datetime . datetime . utcnow ( ) <EOL> self . _create_test_cdr ( hostname = h , drivers = [ ] ) <EOL> result = self . dbapi . get_active_driver_dict ( ) <EOL> self . assertEqual ( expected , result ) <EOL> @ mock . patch . object ( timeutils , '<STR_LIT>' , autospec = True ) <EOL> def test_get_active_driver_dict_one_host_one_driver ( self , mock_utcnow ) : <EOL> h = '<STR_LIT>' <EOL> d = '<STR_LIT>' <EOL> expected = { d : set ( [ h ] ) } <EOL> mock_utcnow . return_value = datetime . datetime . utcnow ( ) <EOL> self . _create_test_cdr ( hostname = h , drivers = [ d ] ) <EOL> result = self . dbapi . get_active_driver_dict ( ) <EOL> self . assertEqual ( expected , result ) <EOL> @ mock . patch . object ( timeutils , '<STR_LIT>' , autospec = True ) <EOL> def test_get_active_driver_dict_one_host_many_drivers ( self , mock_utcnow ) : <EOL> h = '<STR_LIT>' <EOL> d1 = '<STR_LIT>' <EOL> d2 = '<STR_LIT>' <EOL> expected = { d1 : set ( [ h ] ) , d2 : set ( [ h ] ) } <EOL> mock_utcnow . return_value = datetime . datetime . utcnow ( ) <EOL> self . _create_test_cdr ( hostname = h , drivers = [ d1 , d2 ] ) <EOL> result = self . dbapi . get_active_driver_dict ( ) <EOL> self . assertEqual ( expected , result ) <EOL> @ mock . patch . object ( timeutils , '<STR_LIT>' , autospec = True ) <EOL> def test_get_active_driver_dict_many_hosts_one_driver ( self , mock_utcnow ) : <EOL> h1 = '<STR_LIT>' <EOL> h2 = '<STR_LIT>' <EOL> d = '<STR_LIT>' <EOL> expected = { d : set ( [ h1 , h2 ] ) } <EOL> mock_utcnow . return_value = datetime . datetime . utcnow ( ) <EOL> self . _create_test_cdr ( id = <NUM_LIT:1> , hostname = h1 , drivers = [ d ] ) <EOL> self . _create_test_cdr ( id = <NUM_LIT:2> , hostname = h2 , drivers = [ d ] ) <EOL> result = self . dbapi . get_active_driver_dict ( ) <EOL> self . assertEqual ( expected , result ) <EOL> @ mock . patch . object ( timeutils , '<STR_LIT>' , autospec = True ) <EOL> def test_get_active_driver_dict_many_hosts_and_drivers ( self , mock_utcnow ) : <EOL> h1 = '<STR_LIT>' <EOL> h2 = '<STR_LIT>' <EOL> h3 = '<STR_LIT>' <EOL> d1 = '<STR_LIT>' <EOL> d2 = '<STR_LIT>' <EOL> expected = { d1 : set ( [ h1 , h2 ] ) , d2 : set ( [ h2 , h3 ] ) } <EOL> mock_utcnow . return_value = datetime . datetime . utcnow ( ) <EOL> self . _create_test_cdr ( id = <NUM_LIT:1> , hostname = h1 , drivers = [ d1 ] ) <EOL> self . _create_test_cdr ( id = <NUM_LIT:2> , hostname = h2 , drivers = [ d1 , d2 ] ) <EOL> self . _create_test_cdr ( id = <NUM_LIT:3> , hostname = h3 , drivers = [ d2 ] ) <EOL> result = self . dbapi . get_active_driver_dict ( ) <EOL> self . assertEqual ( expected , result ) <EOL> @ mock . patch . object ( timeutils , '<STR_LIT>' , autospec = True ) <EOL> def test_get_active_driver_dict_with_old_conductor ( self , mock_utcnow ) : <EOL> past = datetime . datetime ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:0> , <NUM_LIT:0> ) <EOL> present = past + datetime . timedelta ( minutes = <NUM_LIT:2> ) <EOL> d = '<STR_LIT>' <EOL> h1 = '<STR_LIT>' <EOL> d1 = '<STR_LIT>' <EOL> mock_utcnow . return_value = past <EOL> self . _create_test_cdr ( id = <NUM_LIT:1> , hostname = h1 , drivers = [ d , d1 ] ) <EOL> h2 = '<STR_LIT>' <EOL> d2 = '<STR_LIT>' <EOL> mock_utcnow . return_value = present <EOL> self . _create_test_cdr ( id = <NUM_LIT:2> , hostname = h2 , drivers = [ d , d2 ] ) <EOL> one_minute = <NUM_LIT> <EOL> expected = { d : set ( [ h2 ] ) , d2 : set ( [ h2 ] ) } <EOL> result = self . dbapi . get_active_driver_dict ( interval = one_minute ) <EOL> self . assertEqual ( expected , result ) <EOL> two_minute = one_minute * <NUM_LIT:2> <EOL> expected = { d : set ( [ h1 , h2 ] ) , d1 : set ( [ h1 ] ) , d2 : set ( [ h2 ] ) } <EOL> result = self . dbapi . get_active_driver_dict ( interval = two_minute ) <EOL> self . assertEqual ( expected , result ) <EOL> @ mock . patch . object ( timeutils , '<STR_LIT>' , autospec = True ) <EOL> def test_get_offline_conductors ( self , mock_utcnow ) : <EOL> self . config ( heartbeat_timeout = <NUM_LIT> , group = '<STR_LIT>' ) <EOL> time_ = datetime . datetime ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:0> , <NUM_LIT:0> ) <EOL> mock_utcnow . return_value = time_ <EOL> c = self . _create_test_cdr ( ) <EOL> mock_utcnow . return_value = time_ + datetime . timedelta ( seconds = <NUM_LIT:30> ) <EOL> self . assertEqual ( [ ] , self . dbapi . get_offline_conductors ( ) ) <EOL> mock_utcnow . return_value = time_ + datetime . timedelta ( seconds = <NUM_LIT> ) <EOL> self . assertEqual ( [ c . hostname ] , self . dbapi . get_offline_conductors ( ) ) </s>
