<s> """<STR_LIT>""" <EOL> import json <EOL> import os <EOL> import time <EOL> from threading import Thread <EOL> import sh <EOL> from pygments import highlight <EOL> from pygments . lexers import JsonLexer <EOL> from pygments . formatters import Terminal256Formatter <EOL> from prompt_toolkit import prompt <EOL> from . common import utils <EOL> from . common . utils import log <EOL> from . common . runtime import rpc <EOL> from . common . runtime . backends import LocalBackend <EOL> from . common . runtime . runner import ServiceRunner <EOL> from . common . commands import HutCmd <EOL> from . common . exceptions import ConfigError <EOL> from . toolkit_utils import * <EOL> from . builder import Service , stacks , get_docker , OS_TYPE <EOL> from . commands import UserCmd <EOL> class TestRunner : <EOL> def __init__ ( self , port ) : <EOL> utils . SERVER_URL = "<STR_LIT>" . format ( port ) <EOL> def call_service ( self , msg ) : <EOL> r = stackhut_api_call ( '<STR_LIT>' , msg ) <EOL> result = highlight ( json . dumps ( r , indent = <NUM_LIT:4> ) , JsonLexer ( ) , Terminal256Formatter ( ) ) <EOL> log . info ( "<STR_LIT>" . format ( utils . SERVER_URL , result ) ) <EOL> def test_file ( self , fname ) : <EOL> with open ( fname , "<STR_LIT:r>" ) as f : <EOL> msg = json . load ( f ) <EOL> self . call_service ( msg ) <EOL> def test_interactive ( self , service_name ) : <EOL> contract = rpc . load_contract_file ( ) <EOL> interfaces = contract . interfaces <EOL> log . info ( "<STR_LIT>" . format ( len ( interfaces ) , list ( interfaces . keys ( ) ) ) ) <EOL> for i in contract . interfaces . values ( ) : <EOL> log . info ( "<STR_LIT>" . <EOL> format ( i . name , len ( i . functions ) ) ) <EOL> for f in i . functions . values ( ) : <EOL> log . info ( "<STR_LIT>" . format ( rpc . render_signature ( f ) ) ) <EOL> while True : <EOL> ( iface , fname ) = prompt ( '<STR_LIT>' ) . split ( '<STR_LIT:.>' ) <EOL> f = contract . interface ( iface ) . function ( fname ) <EOL> values = [ prompt ( '<STR_LIT>' . format ( p . type , p . name ) ) <EOL> for p in f . params ] <EOL> eval_values = [ json . loads ( x ) for x in values ] <EOL> if utils . VERBOSE : <EOL> pp_values = highlight ( json . dumps ( eval_values , indent = <NUM_LIT:4> ) , JsonLexer ( ) , Terminal256Formatter ( ) ) <EOL> log . debug ( "<STR_LIT>" . format ( f . full_name , pp_values ) ) <EOL> msg = { <EOL> "<STR_LIT>" : service_name , <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : f . full_name , <EOL> "<STR_LIT>" : eval_values <EOL> } <EOL> } <EOL> self . call_service ( msg ) <EOL> class TestRequestCmd ( HutCmd , UserCmd ) : <EOL> """<STR_LIT>""" <EOL> name = '<STR_LIT:test>' <EOL> description = "<STR_LIT>" <EOL> @ staticmethod <EOL> def register ( sp ) : <EOL> sp . add_argument ( "<STR_LIT:port>" , nargs = '<STR_LIT:?>' , default = '<STR_LIT>' , help = "<STR_LIT>" , type = int ) <EOL> sp . add_argument ( "<STR_LIT>" , '<STR_LIT>' , metavar = '<STR_LIT>' , help = "<STR_LIT>" ) <EOL> sp . add_argument ( "<STR_LIT>" , '<STR_LIT>' , action = '<STR_LIT:store_true>' , help = "<STR_LIT>" ) <EOL> def __init__ ( self , args ) : <EOL> super ( ) . __init__ ( args ) <EOL> self . port = args . port <EOL> self . fname = args . file <EOL> self . interactive = args . interactive <EOL> def run ( self ) : <EOL> tester = TestRunner ( self . port ) <EOL> if self . fname is not None : <EOL> tester . test_file ( self . fname ) <EOL> elif self . interactive : <EOL> tester . test_interactive ( self . hutcfg . service_short_name ( self . usercfg . username ) ) <EOL> else : <EOL> raise AssertionError ( "<STR_LIT>" ) <EOL> return <NUM_LIT:0> <EOL> class RunService : <EOL> def __init__ ( self , port , args , username , service_full_name ) : <EOL> import random <EOL> host_store_dir = os . path . abspath ( LocalBackend . local_store ) <EOL> os . mkdir ( host_store_dir ) if not os . path . exists ( host_store_dir ) else None <EOL> self . name = '<STR_LIT>' . format ( random . randrange ( <NUM_LIT> ) ) <EOL> res_flag = '<STR_LIT:z>' if OS_TYPE == '<STR_LIT>' else '<STR_LIT>' <EOL> verbose_mode = '<STR_LIT>' if args . verbose else None <EOL> uid_gid = '<STR_LIT>' . format ( os . getuid ( ) , os . getgid ( ) ) <EOL> args = [ <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' . format ( port ) , <EOL> '<STR_LIT>' , '<STR_LIT>' . format ( host_store_dir , LocalBackend . local_store , res_flag ) , <EOL> '<STR_LIT>' , '<STR_LIT>' . format ( self . name ) , <EOL> '<STR_LIT>' if args . privileged else None , <EOL> '<STR_LIT>' , service_full_name , '<STR_LIT>' , verbose_mode , <EOL> '<STR_LIT>' , '<STR_LIT>' , uid_gid , '<STR_LIT>' , username <EOL> ] <EOL> self . args = [ x for x in args if x is not None ] <EOL> def __enter__ ( self ) : <EOL> return self <EOL> def __exit__ ( self , exc_type , exc_val , exc_tb ) : <EOL> docker = get_docker ( ) <EOL> docker . run_docker_sh ( '<STR_LIT>' , [ '<STR_LIT>' , '<STR_LIT:5>' , self . name ] ) <EOL> log . info ( "<STR_LIT>" ) <EOL> def start ( self ) : <EOL> log . info ( "<STR_LIT>" ) <EOL> docker_cmd = sh . Command ( "<STR_LIT>" ) <EOL> self . p = docker_cmd ( self . args , _bg = True , _out = lambda x : log . debug ( "<STR_LIT>" . format ( x . rstrip ( ) ) ) , <EOL> _err = lambda x : log . error ( "<STR_LIT>" . format ( x . rstrip ( ) ) ) ) <EOL> class RunContainerCmd ( HutCmd , UserCmd ) : <EOL> """<STR_LIT>""" <EOL> name = '<STR_LIT>' <EOL> description = "<STR_LIT>" <EOL> @ staticmethod <EOL> def register ( sp ) : <EOL> sp . add_argument ( "<STR_LIT:port>" , nargs = '<STR_LIT:?>' , default = '<STR_LIT>' , help = "<STR_LIT>" , type = int ) <EOL> sp . add_argument ( "<STR_LIT>" , '<STR_LIT>' , action = '<STR_LIT:store_true>' , help = "<STR_LIT>" ) <EOL> sp . add_argument ( "<STR_LIT>" , '<STR_LIT>' , action = '<STR_LIT:store_true>' , help = "<STR_LIT>" ) <EOL> sp . add_argument ( "<STR_LIT>" , '<STR_LIT:-c>' , metavar = '<STR_LIT>' , help = "<STR_LIT>" ) <EOL> sp . add_argument ( "<STR_LIT>" , '<STR_LIT>' , action = '<STR_LIT:store_true>' , help = "<STR_LIT>" ) <EOL> sp . add_argument ( "<STR_LIT>" , metavar = '<STR_LIT>' , help = "<STR_LIT>" ) <EOL> sp . add_argument ( "<STR_LIT>" , '<STR_LIT>' , action = '<STR_LIT:store_true>' , help = "<STR_LIT>" ) <EOL> def __init__ ( self , args ) : <EOL> self . port = args . port <EOL> self . force = args . force <EOL> self . privileged = args . privileged <EOL> self . clone = args . clone <EOL> self . test = args . test <EOL> if self . clone : <EOL> from posixpath import basename <EOL> from urllib . parse import urlparse <EOL> dir = basename ( urlparse ( self . clone ) . path ) <EOL> try : <EOL> sh . git . clone ( self . clone ) <EOL> except sh . ErrorReturnCode_128 : <EOL> raise RuntimeError ( "<STR_LIT>" . format ( dir ) ) <EOL> log . debug ( "<STR_LIT>" . format ( self . clone , dir ) ) <EOL> utils . change_root_dir ( dir ) <EOL> self . dir = utils . ROOT_DIR <EOL> super ( ) . __init__ ( args ) <EOL> def sigterm_handler ( self , signo , frame ) : <EOL> log . debug ( "<STR_LIT>" . format ( signo ) ) <EOL> raise KeyboardInterrupt <EOL> def run ( self ) : <EOL> from posixpath import basename <EOL> import signal <EOL> super ( ) . run ( ) <EOL> service = Service ( self . hutcfg , self . usercfg . username ) <EOL> service . build_push ( force = self . force ) <EOL> docker = get_docker ( ) <EOL> log . info ( "<STR_LIT>" . <EOL> format ( self . hutcfg . service_short_name ( self . usercfg . username ) , docker . ip , self . port ) ) <EOL> log . info ( "<STR_LIT>" <EOL> "<STR_LIT>" . format ( docker . ip , self . port , basename ( utils . ROOT_DIR ) ) ) <EOL> signal . signal ( signal . SIGTERM , self . sigterm_handler ) <EOL> signal . signal ( signal . SIGINT , self . sigterm_handler ) <EOL> try : <EOL> with RunService ( self . port , self . args , self . usercfg . username , service . full_name ) as sh_service : <EOL> sh_service . start ( ) <EOL> if self . test : <EOL> time . sleep ( <NUM_LIT:2> ) <EOL> TestRequestCmd ( self . args ) . run ( ) <EOL> else : <EOL> log . debug ( "<STR_LIT>" ) <EOL> sh_service . p . wait ( ) <EOL> except KeyboardInterrupt : <EOL> log . debug ( "<STR_LIT>" ) <EOL> log . info ( "<STR_LIT>" ) <EOL> return <NUM_LIT:0> <EOL> class RunCmd ( RunContainerCmd ) : <EOL> """<STR_LIT>""" <EOL> name = '<STR_LIT>' <EOL> class RunHostCmd ( HutCmd , UserCmd ) : <EOL> """<STR_LIT>""" <EOL> name = '<STR_LIT>' <EOL> description = "<STR_LIT>" <EOL> @ staticmethod <EOL> def register ( sp ) : <EOL> sp . add_argument ( "<STR_LIT:port>" , nargs = '<STR_LIT:?>' , default = '<STR_LIT>' , help = "<STR_LIT>" , type = int ) <EOL> def __init__ ( self , args ) : <EOL> super ( ) . __init__ ( args ) <EOL> self . port = args . port <EOL> def run ( self ) : <EOL> toolkit_stack = stacks [ self . hutcfg . stack ] <EOL> toolkit_stack . copy_shim ( ) <EOL> rpc . generate_contract_file ( ) <EOL> try : <EOL> backend = LocalBackend ( self . hutcfg , self . usercfg . username , port = self . port ) <EOL> with ServiceRunner ( backend , self . hutcfg ) as runner : <EOL> log . info ( "<STR_LIT>" . format ( backend . service_short_name , self . port ) ) <EOL> runner . run ( ) <EOL> except ConfigError as err : <EOL> log . error ( '<STR_LIT>' , str ( err ) ) <EOL> finally : <EOL> toolkit_stack . del_shim ( ) <EOL> os . remove ( rpc . REQ_FIFO ) if os . path . exists ( rpc . REQ_FIFO ) else None <EOL> os . remove ( rpc . RESP_FIFO ) if os . path . exists ( rpc . RESP_FIFO ) else None <EOL> COMMANDS = [ <EOL> RunContainerCmd , RunCmd , RunHostCmd , TestRequestCmd , <EOL> ] </s>
