X-Git-Url: http://git.onelab.eu/?a=blobdiff_plain;f=test%2Flib%2Ftest_utils.py;h=cd67dcddf1b76f165b35bf28b6e2b6e14a390746;hb=58a2b493f8df1072a1faa653c8abb6a3f9ba21fa;hp=e783a626ca0c7bda1678cd92fa3bc837f21dcd12;hpb=d662dcbab5a504791e306489dde627dde9221fd1;p=nepi.git diff --git a/test/lib/test_utils.py b/test/lib/test_utils.py index e783a626..cd67dcdd 100644 --- a/test/lib/test_utils.py +++ b/test/lib/test_utils.py @@ -3,9 +3,8 @@ # Copyright (C) 2013 INRIA # # This program is free software: you can redistribute it and/or modify -# it under the terms of the GNU General Public License as published by -# the Free Software Foundation, either version 3 of the License, or -# (at your option) any later version. +# it under the terms of the GNU General Public License version 2 as +# published by the Free Software Foundation; # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of @@ -17,20 +16,29 @@ # # Author: Alina Quereilhac +from __future__ import print_function + from nepi.resources.linux.node import LinuxNode import os +import sys class DummyEC(object): @property def exp_id(self): return "nepi-1" -def create_node(hostname, username): +def create_node(hostname, username = None, identity = None): ec = DummyEC() node = LinuxNode(ec, 1) + node.set("hostname", hostname) - node.set("username", username) + + if username: + node.set("username", username) + + if identity: + node.set("identity", identity) # If we don't return the reference to the EC # it will be released by the garbage collector since @@ -40,12 +48,23 @@ def create_node(hostname, username): def skipIfNotAlive(func): name = func.__name__ def wrapped(*args, **kwargs): - node, ec = create_node(args[1], args[2]) + hostname = args[1] + if hostname != "localhost": + username = None + identity = None - if not node.is_alive(): - print "*** WARNING: Skipping test %s: Node %s is not alive\n" % ( - name, node.get("hostname")) - return + if len(args) >= 3: + username = args[2] + + if len(args) >= 4: + identity = args[3] + + node, ec = create_node(hostname, username, identity) + + if not node.is_alive(): + print("*** WARNING: Skipping test %s: Node %s is not alive\n" % ( + name, node.get("hostname"))) + return return func(*args, **kwargs) @@ -56,27 +75,50 @@ def skipIfAnyNotAlive(func): def wrapped(*args, **kwargs): argss = list(args) argss.pop(0) - for i in xrange(len(argss)/2): + + for i in range(len(argss)/2): username = argss[i*2] hostname = argss[i*2+1] node, ec = create_node(hostname, username) if not node.is_alive(): - print "*** WARNING: Skipping test %s: Node %s is not alive\n" % ( - name, node.get("hostname")) + print("*** WARNING: Skipping test %s: Node %s is not alive\n" % ( + name, node.get("hostname"))) return return func(*args, **kwargs) return wrapped +def skipIfAnyNotAliveWithIdentity(func): + name = func.__name__ + def wrapped(*args, **kwargs): + argss = list(args) + argss.pop(0) + for i in range(len(argss)/3): + username = argss[i*3] + hostname = argss[i*3+1] + identity = argss[i*3+2] + + node, ec = create_node(hostname, username, identity) + + if not node.is_alive(): + print("*** WARNING: Skipping test %s: Node %s is not alive\n" % ( + name, node.get("hostname"))) + return + + return func(*args, **kwargs) + + return wrapped + + def skipInteractive(func): name = func.__name__ def wrapped(*args, **kwargs): mode = os.environ.get("NEPI_INTERACTIVE_TEST", False) mode = mode and mode.lower() in ['true', 'yes'] if not mode: - print "*** WARNING: Skipping test %s: Interactive mode off \n" % name + print("*** WARNING: Skipping test %s: Interactive mode off \n" % name) return return func(*args, **kwargs) @@ -88,12 +130,64 @@ def skipIfNotPLCredentials(func): def wrapped(*args, **kwargs): pl_user = os.environ.get("PL_USER") pl_pass = os.environ.get("PL_PASS") - pl_slice = os.environ.get("PL_SLICE") - if not (pl_user and pl_pass and pl_slice): - print "*** WARNING: Skipping test %s: Planetlab user, password and slicename not defined\n" % name + if not (pl_user and pl_pass): + print("*** WARNING: Skipping test %s: Planetlab user, password and slicename not defined\n" % name) + return + + return func(*args, **kwargs) + + return wrapped + +def skipIfNotPythonVersion(func): + name = func.__name__ + def wrapped(*args, **kwargs): + if sys.version_info < 2.7: + print("*** WARNING: Skipping test %s: total_seconds() method doesn't exist\n" % name) + return + + return func(*args, **kwargs) + + return wrapped + +def skipIfNotSfaCredentials(func): + name = func.__name__ + def wrapped(*args, **kwargs): + sfa_user = os.environ.get("SFA_USER") + sfa_pk = os.environ.get("SFA_PK") + + if not (sfa_user and os.path.exists(os.path.expanduser(sfa_pk))): + print("*** WARNING: Skipping test %s: SFA path to private key doesn't exist\n" % name) return return func(*args, **kwargs) return wrapped +def skipIfNotSfi(func): + name = func.__name__ + def wrapped(*args, **kwargs): + try: + from sfa.client.sfi import Sfi + from sfa.util.xrn import hrn_to_urn + except ImportError: + print("*** WARNING: Skipping test %s: sfi-client or sfi-common not installed\n" % name) + return + + return func(*args, **kwargs) + + return wrapped + +def skipIf(cond, text): + def wrapped(func, text): + name = func.__name__ + + def banner(*args, **kwargs): + sys.stderr.write("*** WARNING: Skipping test %s: `%s'\n" % + (name, text)) + return None + return banner + + return (lambda func: wrapped(func, text)) if cond else lambda func: func + + +