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):
  31     ec = DummyEC()
  32     node = LinuxNode(ec, 1)
  33     node.set("hostname", hostname)
  34     node.set("username", username)
  35
  36     # If we don't return the reference to the EC
  37     # it will be released by the garbage collector since
  38     # the resources only save a weak refernce to it.
  39     return node, ec
  40
  41 def skipIfNotAlive(func):
  42     name = func.__name__
  43     def wrapped(*args, **kwargs):
  44         node, ec = create_node(args[1], args[2])
  45
  46         if not node.is_alive():
  47             print "*** WARNING: Skipping test %s: Node %s is not alive\n" % (
  48                 name, node.get("hostname"))
  49             return
  50
  51         return func(*args, **kwargs)
  52
  53     return wrapped
  54
  55 def skipIfAnyNotAlive(func):
  56     name = func.__name__
  57     def wrapped(*args, **kwargs):
  58         argss = list(args)
  59         argss.pop(0)
  60         for i in xrange(len(argss)/2):
  61             username = argss[i*2]
  62             hostname = argss[i*2+1]
  63             node, ec = create_node(hostname, username)
  64
  65             if not node.is_alive():
  66                 print "*** WARNING: Skipping test %s: Node %s is not alive\n" % (
  67                     name, node.get("hostname"))
  68                 return
  69
  70         return func(*args, **kwargs)
  71
  72     return wrapped
  73
  74 def skipInteractive(func):
  75     name = func.__name__
  76     def wrapped(*args, **kwargs):
  77         mode = os.environ.get("NEPI_INTERACTIVE_TEST", False)
  78         mode = mode and  mode.lower() in ['true', 'yes']
  79         if not mode:
  80             print "*** WARNING: Skipping test %s: Interactive mode off \n" % name
  81             return
  82
  83         return func(*args, **kwargs)
  84
  85     return wrapped
  86
  87 def skipIfNotPLCredentials(func):
  88     name = func.__name__
  89     def wrapped(*args, **kwargs):
  90         pl_user = os.environ.get("PL_USER")
  91         pl_pass = os.environ.get("PL_PASS")
  92         if not (pl_user and pl_pass):
  93             print "*** WARNING: Skipping test %s: Planetlab user, password and slicename not defined\n" % name
  94             return
  95
  96         return func(*args, **kwargs)
  97
  98     return wrapped
  99
 100 def skipIfNotPythonVersion(func):
 101     name = func.__name__
 102     def wrapped(*args, **kwargs):
 103         if sys.version_info < 2.7:
 104             print "*** WARNING: Skipping test %s: total_seconds() method doesn't exist\n" % name
 105             return
 106
 107         return func(*args, **kwargs)
 108
 109     return wrapped
 110
 111 def skipIfNotSfaCredentials(func):
 112     name = func.__name__
 113     def wrapped(*args, **kwargs):
 114         sfa_user = os.environ.get("SFA_USER")
 115         sfa_pk = os.environ.get("SFA_PK")
 116
 117         if not (sfa_user and os.path.exists(os.path.expanduser(sfa_pk))):
 118             print "*** WARNING: Skipping test %s: SFA path to private key doesn't exist\n" % name
 119             return
 120
 121         return func(*args, **kwargs)
 122
 123     return wrapped
 124
 125 def skipIfNotSfi(func):
 126     name = func.__name__
 127     def wrapped(*args, **kwargs):
 128         try:
 129             from sfa.client.sfi import Sfi
 130             from sfa.util.xrn import hrn_to_urn
 131         except ImportError:
 132             print "*** WARNING: Skipping test %s: sfi-client or sfi-common not installed\n" % name
 133             return
 134
 135         return func(*args, **kwargs)
 136
 137     return wrapped