<s> import unittest <EOL> import uuid <EOL> import shutil <EOL> import time <EOL> import os <EOL> from alaudacli import cmd_parser , cmd_processor , service , auth <EOL> from alaudacli . service import Service <EOL> class ServiceTest ( unittest . TestCase ) : <EOL> @ classmethod <EOL> def setUpClass ( cls ) : <EOL> try : <EOL> service . Service . remove ( '<STR_LIT>' ) <EOL> except : <EOL> pass <EOL> try : <EOL> service . Service . remove ( '<STR_LIT>' ) <EOL> except : <EOL> pass <EOL> api_endpoint , token , username = auth . load_token ( ) <EOL> if '<STR_LIT>' in api_endpoint : <EOL> cls . registry = '<STR_LIT>' <EOL> else : <EOL> cls . registry = '<STR_LIT>' <EOL> def _create_service ( self , name , image , port , start = False , volume = '<STR_LIT>' , envvar = '<STR_LIT>' , link = '<STR_LIT>' ) : <EOL> cmd = '<STR_LIT>' <EOL> if start : <EOL> cmd = '<STR_LIT>' <EOL> argv = [ '<STR_LIT>' , cmd , name , image , '<STR_LIT>' , str ( port ) ] <EOL> if volume : <EOL> argv += [ '<STR_LIT>' , volume ] <EOL> if envvar : <EOL> argv += [ '<STR_LIT>' , envvar ] <EOL> args = cmd_parser . parse_cmds ( argv ) <EOL> cmd_processor . process_cmds ( args ) <EOL> def _wait_for_service ( self , name , timeout_in_seconds = <NUM_LIT> ) : <EOL> for i in range ( <NUM_LIT:0> , timeout_in_seconds ) : <EOL> service = Service . fetch ( name ) <EOL> if service . get_state ( ) == '<STR_LIT>' : <EOL> time . sleep ( <NUM_LIT:1> ) <EOL> continue <EOL> elif service . get_state ( ) == '<STR_LIT>' : <EOL> print '<STR_LIT>' . format ( name ) <EOL> return <EOL> elif service . get_state ( ) == '<STR_LIT>' : <EOL> print '<STR_LIT>' . format ( name ) <EOL> return <EOL> else : <EOL> raise Exception ( '<STR_LIT>' . format ( name ) ) <EOL> raise Exception ( '<STR_LIT>' . format ( name ) ) <EOL> def _start_service ( self , name ) : <EOL> argv = [ '<STR_LIT>' , '<STR_LIT:start>' , name ] <EOL> args = cmd_parser . parse_cmds ( argv ) <EOL> cmd_processor . process_cmds ( args ) <EOL> self . _wait_for_service ( '<STR_LIT>' ) <EOL> def _list_services ( self ) : <EOL> argv = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> args = cmd_parser . parse_cmds ( argv ) <EOL> cmd_processor . process_cmds ( args ) <EOL> def _inspect_service ( self , name ) : <EOL> argv = [ '<STR_LIT>' , '<STR_LIT>' , name ] <EOL> args = cmd_parser . parse_cmds ( argv ) <EOL> cmd_processor . process_cmds ( args ) <EOL> def _get_service_logs ( self , name ) : <EOL> argv = [ '<STR_LIT>' , '<STR_LIT>' , name ] <EOL> args = cmd_parser . parse_cmds ( argv ) <EOL> cmd_processor . process_cmds ( args ) <EOL> def _scale_service ( self , name , target_num_instances ) : <EOL> argv = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' . format ( name , target_num_instances ) ] <EOL> args = cmd_parser . parse_cmds ( argv ) <EOL> cmd_processor . process_cmds ( args ) <EOL> def _stop_service ( self , name ) : <EOL> argv = [ '<STR_LIT>' , '<STR_LIT>' , name ] <EOL> args = cmd_parser . parse_cmds ( argv ) <EOL> cmd_processor . process_cmds ( args ) <EOL> def _remove_service ( self , name ) : <EOL> argv = [ '<STR_LIT>' , '<STR_LIT>' , name ] <EOL> args = cmd_parser . parse_cmds ( argv ) <EOL> cmd_processor . process_cmds ( args ) <EOL> def _enable_autoscaling ( self , name , config ) : <EOL> argv = [ '<STR_LIT>' , '<STR_LIT>' , name , '<STR_LIT>' , config ] <EOL> args = cmd_parser . parse_cmds ( argv ) <EOL> cmd_processor . process_cmds ( args ) <EOL> def _disable_autoscaling ( self , name ) : <EOL> argv = [ '<STR_LIT>' , '<STR_LIT>' , name ] <EOL> args = cmd_parser . parse_cmds ( argv ) <EOL> cmd_processor . process_cmds ( args ) <EOL> def test_stateless_service ( self ) : <EOL> name = '<STR_LIT>' <EOL> image = self . registry + '<STR_LIT>' <EOL> self . _create_service ( name , image , <NUM_LIT> ) <EOL> self . _wait_for_service ( name ) <EOL> self . _start_service ( name ) <EOL> self . _list_services ( ) <EOL> self . _inspect_service ( name ) <EOL> self . _get_service_logs ( name ) <EOL> self . _scale_service ( name , <NUM_LIT:2> ) <EOL> self . _wait_for_service ( name ) <EOL> self . _enable_autoscaling ( name , '<STR_LIT>' ) <EOL> self . _wait_for_service ( name ) <EOL> self . _disable_autoscaling ( name ) <EOL> self . _wait_for_service ( name ) <EOL> self . _stop_service ( name ) <EOL> self . _wait_for_service ( name ) <EOL> self . _remove_service ( name ) <EOL> def test_stateful_service ( self ) : <EOL> name = '<STR_LIT>' <EOL> image = self . registry + '<STR_LIT>' <EOL> self . _create_service ( name , image , <NUM_LIT> , True , <EOL> volume = '<STR_LIT>' , <EOL> envvar = '<STR_LIT>' ) <EOL> self . _wait_for_service ( name ) <EOL> self . _inspect_service ( name ) <EOL> self . _stop_service ( name ) <EOL> self . _wait_for_service ( name ) <EOL> self . _remove_service ( name ) <EOL> def test_service_linking ( self ) : <EOL> db = '<STR_LIT>' <EOL> db_image = self . registry + '<STR_LIT>' <EOL> web = '<STR_LIT>' <EOL> web_image = self . registry + '<STR_LIT>' <EOL> self . _create_service ( db , db_image , <NUM_LIT> , True , <EOL> volume = '<STR_LIT>' , <EOL> envvar = '<STR_LIT>' ) <EOL> self . _wait_for_service ( db ) <EOL> self . _create_service ( web , web_image , <NUM_LIT> , <EOL> link = '<STR_LIT>' ) <EOL> self . _wait_for_service ( web ) <EOL> self . _remove_service ( web ) <EOL> self . _remove_service ( db ) <EOL> class BuildTest ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . repo_name = '<STR_LIT>' <EOL> self . namespace = '<STR_LIT>' <EOL> self . source_path = '<STR_LIT>' . format ( uuid . uuid4 ( ) ) <EOL> os . makedirs ( self . source_path ) <EOL> with open ( os . path . join ( self . source_path , '<STR_LIT>' ) , '<STR_LIT:w>' ) as f : <EOL> f . writelines ( [ <EOL> '<STR_LIT>' , '<STR_LIT>' <EOL> ] ) <EOL> def tearDown ( self ) : <EOL> shutil . rmtree ( self . source_path ) <EOL> def test_create_build ( self ) : <EOL> argv = [ <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , self . source_path , '<STR_LIT>' , self . repo_name , <EOL> '<STR_LIT>' , self . namespace , '<STR_LIT>' , '<STR_LIT>' <EOL> ] <EOL> args = cmd_parser . parse_cmds ( argv ) <EOL> cmd_processor . process_cmds ( args ) </s>
