test/lib/test_utils.py

   1 #
   2 #    NEPI, a framework to manage network experiments
   3 #    Copyright (C) 2013 INRIA
   4 #
   5 #    This program is free software: you can redistribute it and/or modify
   6 #    it under the terms of the GNU General Public License version 2 as
   7 #    published by the Free Software Foundation;
   8 #
   9 #    This program is distributed in the hope that it will be useful,
  10 #    but WITHOUT ANY WARRANTY; without even the implied warranty of
  11 #    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  12 #    GNU General Public License for more details.
  13 #
  14 #    You should have received a copy of the GNU General Public License
  15 #    along with this program.  If not, see <http://www.gnu.org/licenses/>.
  16 #
  17 # Author: Alina Quereilhac <alina.quereilhac@inria.fr>
  18
  19 from __future__ import print_function
  20
  21 from nepi.resources.linux.node import LinuxNode
  22
  23 import os
  24 import sys
  25
  26 class DummyEC(object):
  27     @property
  28     def exp_id(self):
  29         return "nepi-1"
  30
  31 def create_node(hostname, username = None, identity = None):
  32     ec = DummyEC()
  33     node = LinuxNode(ec, 1)
  34
  35     node.set("hostname", hostname)
  36
  37     if username:
  38         node.set("username", username)
  39
  40     if identity:
  41         node.set("identity", identity)
  42
  43     # If we don't return the reference to the EC
  44     # it will be released by the garbage collector since
  45     # the resources only save a weak refernce to it.
  46     return node, ec
  47
  48 def skipIfNotAlive(func):
  49     name = func.__name__
  50     def wrapped(*args, **kwargs):
  51         hostname = args[1]
  52         if hostname != "localhost":
  53             username = None
  54             identity = None
  55
  56             if len(args) >= 3:
  57                 username = args[2]
  58
  59             if len(args) >= 4:
  60                 identity = args[3]
  61
  62             node, ec = create_node(hostname, username, identity)
  63
  64             if not node.is_alive():
  65                 print("*** WARNING: Skipping test %s: Node %s is not alive\n" % (
  66                     name, node.get("hostname")))
  67                 return
  68
  69         return func(*args, **kwargs)
  70
  71     return wrapped
  72
  73 def skipIfAnyNotAlive(func):
  74     name = func.__name__
  75     def wrapped(*args, **kwargs):
  76         argss = list(args)
  77         argss.pop(0)
  78
  79         for i in xrange(len(argss)/2):
  80             username = argss[i*2]
  81             hostname = argss[i*2+1]
  82             node, ec = create_node(hostname, username)
  83
  84             if not node.is_alive():
  85                 print("*** WARNING: Skipping test %s: Node %s is not alive\n" % (
  86                     name, node.get("hostname")))
  87                 return
  88
  89         return func(*args, **kwargs)
  90
  91     return wrapped
  92
  93 def skipIfAnyNotAliveWithIdentity(func):
  94     name = func.__name__
  95     def wrapped(*args, **kwargs):
  96         argss = list(args)
  97         argss.pop(0)
  98         for i in xrange(len(argss)/3):
  99             username = argss[i*3]
 100             hostname = argss[i*3+1]
 101             identity = argss[i*3+2]
 102
 103             node, ec = create_node(hostname, username, identity)
 104
 105             if not node.is_alive():
 106                 print("*** WARNING: Skipping test %s: Node %s is not alive\n" % (
 107                     name, node.get("hostname")))
 108                 return
 109
 110         return func(*args, **kwargs)
 111
 112     return wrapped
 113
 114
 115 def skipInteractive(func):
 116     name = func.__name__
 117     def wrapped(*args, **kwargs):
 118         mode = os.environ.get("NEPI_INTERACTIVE_TEST", False)
 119         mode = mode and  mode.lower() in ['true', 'yes']
 120         if not mode:
 121             print("*** WARNING: Skipping test %s: Interactive mode off \n" % name)
 122             return
 123
 124         return func(*args, **kwargs)
 125
 126     return wrapped
 127
 128 def skipIfNotPLCredentials(func):
 129     name = func.__name__
 130     def wrapped(*args, **kwargs):
 131         pl_user = os.environ.get("PL_USER")
 132         pl_pass = os.environ.get("PL_PASS")
 133         if not (pl_user and pl_pass):
 134             print("*** WARNING: Skipping test %s: Planetlab user, password and slicename not defined\n" % name)
 135             return
 136
 137         return func(*args, **kwargs)
 138
 139     return wrapped
 140
 141 def skipIfNotPythonVersion(func):
 142     name = func.__name__
 143     def wrapped(*args, **kwargs):
 144         if sys.version_info < 2.7:
 145             print("*** WARNING: Skipping test %s: total_seconds() method doesn't exist\n" % name)
 146             return
 147
 148         return func(*args, **kwargs)
 149
 150     return wrapped
 151
 152 def skipIfNotSfaCredentials(func):
 153     name = func.__name__
 154     def wrapped(*args, **kwargs):
 155         sfa_user = os.environ.get("SFA_USER")
 156         sfa_pk = os.environ.get("SFA_PK")
 157
 158         if not (sfa_user and os.path.exists(os.path.expanduser(sfa_pk))):
 159             print("*** WARNING: Skipping test %s: SFA path to private key doesn't exist\n" % name)
 160             return
 161
 162         return func(*args, **kwargs)
 163
 164     return wrapped
 165
 166 def skipIfNotSfi(func):
 167     name = func.__name__
 168     def wrapped(*args, **kwargs):
 169         try:
 170             from sfa.client.sfi import Sfi
 171             from sfa.util.xrn import hrn_to_urn
 172         except ImportError:
 173             print("*** WARNING: Skipping test %s: sfi-client or sfi-common not installed\n" % name)
 174             return
 175
 176         return func(*args, **kwargs)
 177
 178     return wrapped
 179
 180 def skipIf(cond, text):
 181     def wrapped(func, text):
 182         name = func.__name__
 183
 184         def banner(*args, **kwargs):
 185             sys.stderr.write("*** WARNING: Skipping test %s: `%s'\n" %
 186                     (name, text))
 187             return None
 188         return banner
 189
 190     return (lambda func: wrapped(func, text)) if cond else lambda func: func
 191
 192
 193