1 files changed, 21 insertions, 89 deletions

diff --git a/meta/lib/oeqa/utils/decorators.py b/meta/lib/oeqa/utils/decorators.py
index 615fd956b5..ea90164e5e 100644
--- a/meta/lib/oeqa/utils/decorators.py
+++ b/meta/lib/oeqa/utils/decorators.py
@@ -1,6 +1,8 @@
+#
 # Copyright (C) 2013 Intel Corporation
 #
-# Released under the MIT license (see COPYING.MIT)
+# SPDX-License-Identifier: MIT
+#
 
 # Some custom decorators that can be used by unittests
 # Most useful is skipUnlessPassed which can be used for
@@ -14,91 +16,6 @@ import threading
 import signal
 from functools import wraps
 
-#get the "result" object from one of the upper frames provided that one of these upper frames is a unittest.case frame
-class getResults(object):
-    def __init__(self):
-        #dynamically determine the unittest.case frame and use it to get the name of the test method
-        ident = threading.current_thread().ident
-        upperf = sys._current_frames()[ident]
-        while (upperf.f_globals['__name__'] != 'unittest.case'):
-            upperf = upperf.f_back
-
-        def handleList(items):
-            ret = []
-            # items is a list of tuples, (test, failure) or (_ErrorHandler(), Exception())
-            for i in items:
-                s = i[0].id()
-                #Handle the _ErrorHolder objects from skipModule failures
-                if "setUpModule (" in s:
-                    ret.append(s.replace("setUpModule (", "").replace(")",""))
-                else:
-                    ret.append(s)
-                # Append also the test without the full path
-                testname = s.split('.')[-1]
-                if testname:
-                    ret.append(testname)
-            return ret
-        self.faillist = handleList(upperf.f_locals['result'].failures)
-        self.errorlist = handleList(upperf.f_locals['result'].errors)
-        self.skiplist = handleList(upperf.f_locals['result'].skipped)
-
-    def getFailList(self):
-        return self.faillist
-
-    def getErrorList(self):
-        return self.errorlist
-
-    def getSkipList(self):
-        return self.skiplist
-
-class skipIfFailure(object):
-
-    def __init__(self,testcase):
-        self.testcase = testcase
-
-    def __call__(self,f):
-        @wraps(f)
-        def wrapped_f(*args, **kwargs):
-            res = getResults()
-            if self.testcase in (res.getFailList() or res.getErrorList()):
-                raise unittest.SkipTest("Testcase dependency not met: %s" % self.testcase)
-            return f(*args, **kwargs)
-        wrapped_f.__name__ = f.__name__
-        return wrapped_f
-
-class skipIfSkipped(object):
-
-    def __init__(self,testcase):
-        self.testcase = testcase
-
-    def __call__(self,f):
-        @wraps(f)
-        def wrapped_f(*args, **kwargs):
-            res = getResults()
-            if self.testcase in res.getSkipList():
-                raise unittest.SkipTest("Testcase dependency not met: %s" % self.testcase)
-            return f(*args, **kwargs)
-        wrapped_f.__name__ = f.__name__
-        return wrapped_f
-
-class skipUnlessPassed(object):
-
-    def __init__(self,testcase):
-        self.testcase = testcase
-
-    def __call__(self,f):
-        @wraps(f)
-        def wrapped_f(*args, **kwargs):
-            res = getResults()
-            if self.testcase in res.getSkipList() or \
-                    self.testcase in res.getFailList() or \
-                    self.testcase in res.getErrorList():
-                raise unittest.SkipTest("Testcase dependency not met: %s" % self.testcase)
-            return f(*args, **kwargs)
-        wrapped_f.__name__ = f.__name__
-        wrapped_f._depends_on = self.testcase
-        return wrapped_f
-
 class testcase(object):
     def __init__(self, test_case):
         self.test_case = test_case
@@ -172,27 +89,42 @@ def LogResults(original_class):
 
         #check status of tests and record it
 
+        tcid = self.id()
         for (name, msg) in result.errors:
-            if (self._testMethodName == str(name).split(' ')[0]) and (class_name in str(name).split(' ')[1]):
+            if tcid == name.id():
                 local_log.results("Testcase "+str(test_case)+": ERROR")
                 local_log.results("Testcase "+str(test_case)+":\n"+msg)
                 passed = False
         for (name, msg) in result.failures:
-            if (self._testMethodName == str(name).split(' ')[0]) and (class_name in str(name).split(' ')[1]):
+            if tcid == name.id():
                 local_log.results("Testcase "+str(test_case)+": FAILED")
                 local_log.results("Testcase "+str(test_case)+":\n"+msg)
                 passed = False
         for (name, msg) in result.skipped:
-            if (self._testMethodName == str(name).split(' ')[0]) and (class_name in str(name).split(' ')[1]):
+            if tcid == name.id():
                 local_log.results("Testcase "+str(test_case)+": SKIPPED")
                 passed = False
         if passed:
             local_log.results("Testcase "+str(test_case)+": PASSED")
 
+        # XXX: In order to avoid race condition when test if exists the linkfile
+        # use bb.utils.lock, the best solution is to create a unique name for the
+        # link file.
+        try:
+            import bb
+            has_bb = True
+            lockfilename = linkfile + '.lock'
+        except ImportError:
+            has_bb = False
+
+        if has_bb:
+            lf = bb.utils.lockfile(lockfilename, block=True)
         # Create symlink to the current log
         if os.path.lexists(linkfile):
             os.remove(linkfile)
         os.symlink(logfile, linkfile)
+        if has_bb:
+            bb.utils.unlockfile(lf)
 
     original_class.run = run