X-Git-Url: http://git.onelab.eu/?a=blobdiff_plain;f=system%2Futils.py;h=51205bd8cc0128983d698d90982c06273112a4a6;hb=457d31694f7b0f60e2a6fea230e9a3572b9d2b78;hp=b246b1851c1c06fc42f42b9dac655a61127c55b3;hpb=69f11b3ec62477fde6291df69ab3ef7fb10ae5b9;p=tests.git diff --git a/system/utils.py b/system/utils.py index b246b18..51205bd 100644 --- a/system/utils.py +++ b/system/utils.py @@ -1,135 +1,92 @@ -# $Id$ +# -*- python3 -*- +# Thierry Parmentelat +# Copyright (C) 2015 INRIA +# +import sys import time import os -import commands +import re +import glob +import subprocess from pprint import PrettyPrinter +options={} + +def init_options(options_arg): + global options + options = options_arg + # how could this accept a list again ? def header(message): - now=time.strftime("%H:%M:%S", time.localtime()) - print "*",now,'--',message - -def pprint(message,spec,depth=2): - now=time.strftime("%H:%M:%S", time.localtime()) - print ">",now,"--",message - PrettyPrinter(indent=8,depth=depth).pprint(spec) + now = time.strftime("%H:%M:%S", time.localtime()) + print("*", now, '--', message) -def show_site_spec (site): - print '======== site',site['site_fields']['name'] - for (k,v) in site.iteritems(): - if k=='nodes': - if v: - print '\t\t','nodes : ', - for node in v: - print node['node_fields']['hostname'],'', - print '' - elif k=='users': - if v: - print '\t\tusers : ', - for user in v: - print user['name'],'', - print '' - elif k == 'site_fields': - print '\t\tlogin_base',':',v['login_base'] - elif k == 'address_fields': - pass - else: - print '\t\t',k, - PrettyPrinter(indent=8,depth=2).pprint(v) - -def show_initscript_spec (initscript): - print '======== initscript',initscript['initscript_fields']['name'] +def pprint(message, spec, depth=2): + now = time.strftime("%H:%M:%S", time.localtime()) + print(">", now, "--", message) + PrettyPrinter(indent=8, depth=depth).pprint(spec) -def show_key_spec (key): - print '======== key',key['name'] -def show_slice_spec (slice): - print '======== slice',slice['slice_fields']['name'] - for (k,v) in slice.iteritems(): - if k=='nodenames': - if v: - print '\t\tnodes : ', - for nodename in v: - print nodename,'', - print '' - elif k=='usernames': - if v: - print '\t\tusers : ', - for username in v: - print username,'', - print '' - elif k=='slice_fields': - print '\t\tfields',':', - print 'max_nodes=',v['max_nodes'], - print '' - else: - print '\t\t',k,v -def show_test_spec (message,all_plc_specs): - now=time.strftime("%H:%M:%S", time.localtime()) - print ">",now,"--",message - for plc_spec in all_plc_specs: - show_test_spec_pass (plc_spec,1) - show_test_spec_pass (plc_spec,2) +def system(command, background=False, silent=False, dry_run=None): + dry_run = dry_run if dry_run is not None else getattr(options, 'dry_run', False) + if dry_run: + print('dry_run:', command) + return 0 + + if silent : + if command.find(';') >= 0: + command = "({}) 2> /dev/null".format(command) + else: command += " 2> /dev/null" + if background: + command += " &" + if silent: + print('.', end=' ') + sys.stdout.flush() + else: + now = time.strftime("%H:%M:%S", time.localtime()) + # don't show in summary + print("->", now, '--', end=' ') + sys.stdout.flush() + if not silent: + command = "set -x; " + command + return os.system(command) -def show_test_spec_pass (plc_spec,passno): - for (key,val) in plc_spec.iteritems(): - if passno == 2: - if key == 'sites': - for site in val: - show_site_spec(site) - elif key=='initscripts': - for initscript in val: - show_initscript_spec (initscript) - elif key=='slices': - for slice in val: - show_slice_spec (slice) - elif key=='keys': - for key in val: - show_key_spec (key) - elif passno == 1: - if key not in ['sites','initscripts','slices','keys']: - print '\t',key,':',val +### WARNING : this ALWAYS does its job, even in dry_run mode +def output_of (command): + import subprocess + (code, string) = subprocess.getstatusoutput(command) + return (code, string) -def system(command): - now=time.strftime("%H:%M:%S", time.localtime()) - print "+",now,':',command - return os.system("set -x; " + command) -# checks whether a given hostname/ip responds to ping -ping_timeout_option = None -def check_ping (hostname): - # check OS (support for macos) - global ping_timeout_option - if not ping_timeout_option: - (status,osname) = commands.getstatusoutput("uname -s") - if status != 0: - raise Exception, "Cannot figure your OS name" - if osname == "Linux": - ping_timeout_option="-w" - elif osname == "Darwin": - ping_timeout_option="-t" +# convenience: translating shell-like pattern into regexp +def match (string, pattern): + # tmp - there's probably much simpler + # rewrite * into .*, ? into . + pattern = pattern.replace("*",".*") + pattern = pattern.replace("?",".") + return re.compile(pattern).match(string) + +def locate_hooks_scripts (message, path, extensions): + print(message, 'searching', path, 'for extensions', extensions) + scripts = [] + for ext in extensions: + # skip helper programs + scripts += glob.glob (path+'/[a-zA-Z]*.' + ext) + return scripts + +# quick & dirty - should probably use the parseroption object instead +# and move to TestMain as well +exclude_options_keys = [ 'ensure_value' , 'read_file', 'read_module' ] +def show_options (message, options): + now = time.strftime("%H:%M:%S", time.localtime()) + print(">", now, "--", message) + for k in dir(options): + if k.find("_") == 0: + continue + if k in exclude_options_keys: + continue + print(" ", k, ":", getattr(options, k)) - command="ping -c 1 %s 1 %s"%(ping_timeout_option,hostname) - (status,output) = commands.getstatusoutput(command) - return status == 0 -# inserts a backslash before each occurence of the following chars -# \ " ' < > & | ; ( ) $ * ~ -def backslash_shell_specials (command): - result='' - for char in command: - if char in "\\\"'<>&|;()$*~": - result +='\\'+char - else: - result +=char - return result -# check main IP address against the provided hostname -def is_local (hostname): - if hostname == "localhost": - return True - import socket - local_ip = socket.gethostbyname(socket.gethostname()) - remote_ip = socket.gethostbyname(hostname) - return local_ip==remote_ip