fix: resolve pabot issue with duplicated test runs and incomplete exe…

…cutions

Addressed a problem in Pabot where test runs could be duplicated or test executions might not complete properly in specific scenarios.

qase-robotframework/changelog.md

@@ -1,3 +1,11 @@
+# qase-robotframework 3.2.3
+
+## What's new
+
+Fixed an issue with Pabot
+Resolved an issue where test runs were sometimes duplicated or test executions failed to complete under certain
+conditions.
+
 # qase-robotframework 3.2.2
 
 ## What's new

qase-robotframework/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "qase-robotframework"
-version = "3.2.2"
+version = "3.2.3"
 description = "Qase Robot Framework Plugin"
 readme = "README.md"
 authors = [{name = "Qase Team", email = "[email protected]"}]

qase-robotframework/src/qase/robotframework/listener.py

@@ -23,6 +23,10 @@ def get_pool_id():
     return BuiltIn().get_variable_value('${PABOTQUEUEINDEX}', None)
 
 
+def get_last_level_flag():
+    return BuiltIn().get_variable_value('${PABOTISLASTEXECUTIONINPOOL}', None)
+
+
 class Listener:
     ROBOT_LISTENER_API_VERSION = 3
 
@@ -34,6 +38,7 @@ def __init__(self):
         self.runtime = QaseRuntimeSingleton.get_instance()
         self.tests = {}
         self.pabot_index = None
+        self.last_level_flag = None
 
         if config.config.debug:
             logger.setLevel(logging.DEBUG)
@@ -46,6 +51,7 @@ def __init__(self):
 
     def start_suite(self, suite, result):
         self.pabot_index = get_pool_id()
+        self.last_level_flag = get_last_level_flag()
         if self.pabot_index is not None:
             try:
                 if int(self.pabot_index) == 0:
@@ -134,8 +140,8 @@ def end_test(self, test, result):
         )
 
     def close(self):
-        if self.pabot_index is not None:
-            if int(self.pabot_index) == 0:
+        if self.last_level_flag is not None:
+            if int(self.last_level_flag) == 1:
                 Listener.drop_run_id()
             else:
                 self.reporter.complete_worker()