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