<s> """<STR_LIT>""" <EOL> import sys <EOL> import mock <EOL> from mox3 import mox <EOL> from oslo_concurrency import processutils <EOL> from oslo_config import cfg <EOL> from oslo_service import service as _service <EOL> import testtools <EOL> from nova import exception <EOL> from nova import manager <EOL> from nova import objects <EOL> from nova import rpc <EOL> from nova import service <EOL> from nova import test <EOL> from nova . tests . unit import utils <EOL> from nova import wsgi <EOL> test_service_opts = [ <EOL> cfg . StrOpt ( "<STR_LIT>" , <EOL> default = "<STR_LIT>" , <EOL> help = "<STR_LIT>" ) , <EOL> cfg . StrOpt ( "<STR_LIT>" , <EOL> default = '<STR_LIT:127.0.0.1>' , <EOL> help = "<STR_LIT>" ) , <EOL> cfg . IntOpt ( "<STR_LIT>" , <EOL> default = <NUM_LIT:0> , <EOL> help = "<STR_LIT>" ) , <EOL> ] <EOL> CONF = cfg . CONF <EOL> CONF . register_opts ( test_service_opts ) <EOL> class FakeManager ( manager . Manager ) : <EOL> """<STR_LIT>""" <EOL> def test_method ( self ) : <EOL> return '<STR_LIT>' <EOL> class ExtendedService ( service . Service ) : <EOL> def test_method ( self ) : <EOL> return '<STR_LIT>' <EOL> class ServiceManagerTestCase ( test . NoDBTestCase ) : <EOL> """<STR_LIT>""" <EOL> def test_message_gets_to_manager ( self ) : <EOL> serv = service . Service ( '<STR_LIT:test>' , <EOL> '<STR_LIT:test>' , <EOL> '<STR_LIT:test>' , <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( '<STR_LIT>' , serv . test_method ( ) ) <EOL> def test_override_manager_method ( self ) : <EOL> serv = ExtendedService ( '<STR_LIT:test>' , <EOL> '<STR_LIT:test>' , <EOL> '<STR_LIT:test>' , <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( '<STR_LIT>' , serv . test_method ( ) ) <EOL> def test_service_with_min_down_time ( self ) : <EOL> self . flags ( service_down_time = <NUM_LIT:10> , report_interval = <NUM_LIT:10> ) <EOL> service . Service ( '<STR_LIT:test>' , <EOL> '<STR_LIT:test>' , <EOL> '<STR_LIT:test>' , <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( <NUM_LIT> , CONF . service_down_time ) <EOL> class ServiceTestCase ( test . NoDBTestCase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> super ( ServiceTestCase , self ) . setUp ( ) <EOL> self . host = '<STR_LIT:foo>' <EOL> self . binary = '<STR_LIT>' <EOL> self . topic = '<STR_LIT>' <EOL> self . flags ( use_local = True , group = '<STR_LIT>' ) <EOL> def test_create ( self ) : <EOL> app = service . Service . create ( host = self . host , binary = self . binary , <EOL> topic = self . topic ) <EOL> self . assertTrue ( app ) <EOL> def test_repr ( self ) : <EOL> serv = service . Service ( self . host , <EOL> self . binary , <EOL> self . topic , <EOL> '<STR_LIT>' ) <EOL> exp = "<STR_LIT>" "<STR_LIT>" <EOL> self . assertEqual ( exp , repr ( serv ) ) <EOL> def _service_start_mocks ( self ) : <EOL> self . mox . StubOutWithMock ( objects . Service , '<STR_LIT>' ) <EOL> self . mox . StubOutWithMock ( objects . Service , '<STR_LIT>' ) <EOL> objects . Service . get_by_host_and_binary ( mox . IgnoreArg ( ) , self . host , <EOL> self . binary ) <EOL> objects . Service . create ( ) <EOL> def test_init_and_start_hooks ( self ) : <EOL> self . manager_mock = self . mox . CreateMock ( FakeManager ) <EOL> self . mox . StubOutWithMock ( sys . modules [ __name__ ] , <EOL> '<STR_LIT>' , use_mock_anything = True ) <EOL> self . mox . StubOutWithMock ( self . manager_mock , '<STR_LIT>' ) <EOL> self . mox . StubOutWithMock ( self . manager_mock , '<STR_LIT>' ) <EOL> self . mox . StubOutWithMock ( self . manager_mock , '<STR_LIT>' ) <EOL> FakeManager ( host = self . host ) . AndReturn ( self . manager_mock ) <EOL> self . manager_mock . service_name = self . topic <EOL> self . manager_mock . additional_endpoints = [ ] <EOL> self . manager_mock . init_host ( ) <EOL> self . _service_start_mocks ( ) <EOL> self . manager_mock . pre_start_hook ( ) <EOL> self . manager_mock . post_start_hook ( ) <EOL> self . mox . ReplayAll ( ) <EOL> serv = service . Service ( self . host , <EOL> self . binary , <EOL> self . topic , <EOL> '<STR_LIT>' ) <EOL> serv . start ( ) <EOL> def _test_service_check_create_race ( self , ex ) : <EOL> self . manager_mock = self . mox . CreateMock ( FakeManager ) <EOL> self . mox . StubOutWithMock ( sys . modules [ __name__ ] , '<STR_LIT>' , <EOL> use_mock_anything = True ) <EOL> self . mox . StubOutWithMock ( self . manager_mock , '<STR_LIT>' ) <EOL> self . mox . StubOutWithMock ( self . manager_mock , '<STR_LIT>' ) <EOL> self . mox . StubOutWithMock ( self . manager_mock , '<STR_LIT>' ) <EOL> self . mox . StubOutWithMock ( objects . Service , '<STR_LIT>' ) <EOL> self . mox . StubOutWithMock ( objects . Service , '<STR_LIT>' ) <EOL> FakeManager ( host = self . host ) . AndReturn ( self . manager_mock ) <EOL> self . manager_mock . init_host ( ) <EOL> objects . Service . get_by_host_and_binary ( mox . IgnoreArg ( ) , self . host , <EOL> self . binary ) <EOL> objects . Service . create ( ) . AndRaise ( ex ) <EOL> class TestException ( Exception ) : <EOL> pass <EOL> objects . Service . get_by_host_and_binary ( <EOL> mox . IgnoreArg ( ) , self . host , self . binary ) . AndRaise ( TestException ( ) ) <EOL> self . mox . ReplayAll ( ) <EOL> serv = service . Service ( self . host , <EOL> self . binary , <EOL> self . topic , <EOL> '<STR_LIT>' ) <EOL> self . assertRaises ( TestException , serv . start ) <EOL> def test_service_check_create_race_topic_exists ( self ) : <EOL> ex = exception . ServiceTopicExists ( host = '<STR_LIT:foo>' , topic = '<STR_LIT:bar>' ) <EOL> self . _test_service_check_create_race ( ex ) <EOL> def test_service_check_create_race_binary_exists ( self ) : <EOL> ex = exception . ServiceBinaryExists ( host = '<STR_LIT:foo>' , binary = '<STR_LIT:bar>' ) <EOL> self . _test_service_check_create_race ( ex ) <EOL> def test_parent_graceful_shutdown ( self ) : <EOL> self . manager_mock = self . mox . CreateMock ( FakeManager ) <EOL> self . mox . StubOutWithMock ( sys . modules [ __name__ ] , <EOL> '<STR_LIT>' , use_mock_anything = True ) <EOL> self . mox . StubOutWithMock ( self . manager_mock , '<STR_LIT>' ) <EOL> self . mox . StubOutWithMock ( self . manager_mock , '<STR_LIT>' ) <EOL> self . mox . StubOutWithMock ( self . manager_mock , '<STR_LIT>' ) <EOL> self . mox . StubOutWithMock ( _service . Service , '<STR_LIT>' ) <EOL> FakeManager ( host = self . host ) . AndReturn ( self . manager_mock ) <EOL> self . manager_mock . service_name = self . topic <EOL> self . manager_mock . additional_endpoints = [ ] <EOL> self . manager_mock . init_host ( ) <EOL> self . _service_start_mocks ( ) <EOL> self . manager_mock . pre_start_hook ( ) <EOL> self . manager_mock . post_start_hook ( ) <EOL> _service . Service . stop ( ) <EOL> self . mox . ReplayAll ( ) <EOL> serv = service . Service ( self . host , <EOL> self . binary , <EOL> self . topic , <EOL> '<STR_LIT>' ) <EOL> serv . start ( ) <EOL> serv . stop ( ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_parent_graceful_shutdown_with_cleanup_host ( <EOL> self , mock_svc_get_by_host_and_binary , mock_API ) : <EOL> mock_manager = mock . Mock ( ) <EOL> serv = service . Service ( self . host , <EOL> self . binary , <EOL> self . topic , <EOL> '<STR_LIT>' ) <EOL> serv . manager = mock_manager <EOL> serv . manager . additional_endpoints = [ ] <EOL> serv . start ( ) <EOL> serv . manager . init_host . assert_called_with ( ) <EOL> serv . stop ( ) <EOL> serv . manager . cleanup_host . assert_called_with ( ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> @ mock . patch . object ( rpc , '<STR_LIT>' ) <EOL> def test_service_stop_waits_for_rpcserver ( <EOL> self , mock_rpc , mock_svc_get_by_host_and_binary , mock_API ) : <EOL> serv = service . Service ( self . host , <EOL> self . binary , <EOL> self . topic , <EOL> '<STR_LIT>' ) <EOL> serv . start ( ) <EOL> serv . stop ( ) <EOL> serv . rpcserver . start . assert_called_once_with ( ) <EOL> serv . rpcserver . stop . assert_called_once_with ( ) <EOL> serv . rpcserver . wait . assert_called_once_with ( ) <EOL> def test_reset ( self ) : <EOL> serv = service . Service ( self . host , <EOL> self . binary , <EOL> self . topic , <EOL> '<STR_LIT>' ) <EOL> with mock . patch . object ( serv . manager , '<STR_LIT>' ) as mock_reset : <EOL> serv . reset ( ) <EOL> mock_reset . assert_called_once_with ( ) <EOL> class TestWSGIService ( test . NoDBTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( TestWSGIService , self ) . setUp ( ) <EOL> self . stubs . Set ( wsgi . Loader , "<STR_LIT>" , mox . MockAnything ( ) ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_service_start_creates_record ( self , mock_create , mock_get ) : <EOL> mock_get . return_value = None <EOL> test_service = service . WSGIService ( "<STR_LIT>" ) <EOL> test_service . start ( ) <EOL> self . assertTrue ( mock_create . called ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_service_start_does_not_create_record ( self , mock_create , mock_get ) : <EOL> test_service = service . WSGIService ( "<STR_LIT>" ) <EOL> test_service . start ( ) <EOL> self . assertFalse ( mock_create . called ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_service_random_port ( self , mock_get ) : <EOL> test_service = service . WSGIService ( "<STR_LIT>" ) <EOL> test_service . start ( ) <EOL> self . assertNotEqual ( <NUM_LIT:0> , test_service . port ) <EOL> test_service . stop ( ) <EOL> def test_workers_set_default ( self ) : <EOL> test_service = service . WSGIService ( "<STR_LIT>" ) <EOL> self . assertEqual ( test_service . workers , processutils . get_worker_count ( ) ) <EOL> def test_workers_set_good_user_setting ( self ) : <EOL> CONF . set_override ( '<STR_LIT>' , <NUM_LIT:8> ) <EOL> test_service = service . WSGIService ( "<STR_LIT>" ) <EOL> self . assertEqual ( test_service . workers , <NUM_LIT:8> ) <EOL> def test_workers_set_zero_user_setting ( self ) : <EOL> CONF . set_override ( '<STR_LIT>' , <NUM_LIT:0> ) <EOL> test_service = service . WSGIService ( "<STR_LIT>" ) <EOL> self . assertEqual ( test_service . workers , processutils . get_worker_count ( ) ) <EOL> def test_service_start_with_illegal_workers ( self ) : <EOL> CONF . set_override ( "<STR_LIT>" , - <NUM_LIT:1> ) <EOL> self . assertRaises ( exception . InvalidInput , <EOL> service . WSGIService , "<STR_LIT>" ) <EOL> def test_openstack_compute_api_workers_set_default ( self ) : <EOL> test_service = service . WSGIService ( "<STR_LIT>" ) <EOL> self . assertEqual ( test_service . workers , processutils . get_worker_count ( ) ) <EOL> def test_openstack_compute_api_workers_set_good_user_setting ( self ) : <EOL> CONF . set_override ( '<STR_LIT>' , <NUM_LIT:8> ) <EOL> test_service = service . WSGIService ( "<STR_LIT>" ) <EOL> self . assertEqual ( test_service . workers , <NUM_LIT:8> ) <EOL> def test_openstack_compute_api_workers_set_zero_user_setting ( self ) : <EOL> CONF . set_override ( '<STR_LIT>' , <NUM_LIT:0> ) <EOL> test_service = service . WSGIService ( "<STR_LIT>" ) <EOL> self . assertEqual ( test_service . workers , processutils . get_worker_count ( ) ) <EOL> def test_openstack_compute_api_service_start_with_illegal_workers ( self ) : <EOL> CONF . set_override ( "<STR_LIT>" , - <NUM_LIT:1> ) <EOL> self . assertRaises ( exception . InvalidInput , <EOL> service . WSGIService , "<STR_LIT>" ) <EOL> @ testtools . skipIf ( not utils . is_ipv6_supported ( ) , "<STR_LIT>" ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_service_random_port_with_ipv6 ( self , mock_get ) : <EOL> CONF . set_default ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> test_service = service . WSGIService ( "<STR_LIT>" ) <EOL> test_service . start ( ) <EOL> self . assertEqual ( "<STR_LIT>" , test_service . host ) <EOL> self . assertNotEqual ( <NUM_LIT:0> , test_service . port ) <EOL> test_service . stop ( ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_reset_pool_size_to_default ( self , mock_get ) : <EOL> test_service = service . WSGIService ( "<STR_LIT>" ) <EOL> test_service . start ( ) <EOL> test_service . stop ( ) <EOL> self . assertEqual ( test_service . server . _pool . size , <NUM_LIT:0> ) <EOL> test_service . reset ( ) <EOL> test_service . start ( ) <EOL> self . assertEqual ( test_service . server . _pool . size , <EOL> CONF . wsgi . default_pool_size ) <EOL> class TestLauncher ( test . NoDBTestCase ) : <EOL> @ mock . patch . object ( _service , '<STR_LIT>' ) <EOL> def test_launch_app ( self , mock_launch ) : <EOL> service . _launcher = None <EOL> service . serve ( mock . sentinel . service ) <EOL> mock_launch . assert_called_once_with ( mock . ANY , <EOL> mock . sentinel . service , <EOL> workers = None ) <EOL> @ mock . patch . object ( _service , '<STR_LIT>' ) <EOL> def test_launch_app_with_workers ( self , mock_launch ) : <EOL> service . _launcher = None <EOL> service . serve ( mock . sentinel . service , workers = mock . sentinel . workers ) <EOL> mock_launch . assert_called_once_with ( mock . ANY , <EOL> mock . sentinel . service , <EOL> workers = mock . sentinel . workers ) <EOL> @ mock . patch . object ( _service , '<STR_LIT>' ) <EOL> def test_launch_app_more_than_once_raises ( self , mock_launch ) : <EOL> service . _launcher = None <EOL> service . serve ( mock . sentinel . service ) <EOL> self . assertRaises ( RuntimeError , service . serve , mock . sentinel . service ) </s>
