+ map_on_slices.__doc__ = TestSlice.__dict__[method.__name__].__doc__
+ return map_on_slices
+
+def bonding_redirector(method):
+ bonding_name = method.__name__.replace('bonding_', '')
+ def redirect(self):
+ bonding_method = TestBonding.__dict__[bonding_name]
+ return bonding_method(self.test_bonding)
+ # maintain __name__ for ignore_result
+ redirect.__name__ = method.__name__
+ # restore the doc text
+ redirect.__doc__ = TestBonding.__dict__[bonding_name].__doc__
+ return redirect
+
+# run a step but return True so that we can go on
+def ignore_result(method):
+ def ignoring(self):
+ # ssh_slice_ignore->ssh_slice
+ ref_name = method.__name__.replace('_ignore', '').replace('force_', '')
+ ref_method = TestPlc.__dict__[ref_name]
+ result = ref_method(self)
+ print("Actual (but ignored) result for {ref_name} is {result}".format(**locals()))
+ return Ignored(result)
+ name = method.__name__.replace('_ignore', '').replace('force_', '')
+ ignoring.__name__ = name
+ ignoring.__doc__ = "ignored version of " + name
+ return ignoring
+
+# a variant that expects the TestSlice method to return a list of CompleterTasks that
+# are then merged into a single Completer run to avoid wating for all the slices
+# esp. useful when a test fails of course
+# because we need to pass arguments we use a class instead..
+class slice_mapper__tasks(object):
+ # could not get this to work with named arguments
+ def __init__(self, timeout_minutes, silent_minutes, period_seconds):
+ self.timeout = timedelta(minutes = timeout_minutes)
+ self.silent = timedelta(minutes = silent_minutes)
+ self.period = timedelta(seconds = period_seconds)
+ def __call__(self, method):
+ decorator_self=self
+ # compute augmented method name
+ method_name = method.__name__ + "__tasks"
+ # locate in TestSlice
+ slice_method = TestSlice.__dict__[ method_name ]
+ def wrappee(self):
+ tasks=[]
+ for slice_spec in self.plc_spec['slices']:
+ site_spec = self.locate_site (slice_spec['sitename'])
+ test_site = TestSite(self, site_spec)
+ test_slice = TestSlice(self, test_site, slice_spec)
+ tasks += slice_method (test_slice, self.options)
+ return Completer (tasks, message=method.__name__).\
+ run(decorator_self.timeout, decorator_self.silent, decorator_self.period)
+ # restore the doc text from the TestSlice method even if a bit odd
+ wrappee.__name__ = method.__name__
+ wrappee.__doc__ = slice_method.__doc__
+ return wrappee
+
+def auth_sfa_mapper(method):
git://git.onelab.eu / tests.git / blobdiff