fix the duplicated running issue with --argumentfile options

src/pabot/arguments.py

@@ -1,5 +1,8 @@
+import atexit
 import multiprocessing
+import os.path
 import re
+import tempfile
 from typing import Dict, List, Optional, Tuple
 
 from robot import __version__ as ROBOT_VERSION
@@ -65,8 +68,47 @@ def parse_args(
         options_for_subprocesses["name"] = "Suites"
     opts = _delete_none_keys(options)
     opts_sub = _delete_none_keys(options_for_subprocesses)
+    _replace_arg_files(pabot_args, opts_sub)
     return opts, datasources, pabot_args, opts_sub
 
+# remove options from argumentfile according to different scenarios.
+# -t/--test/--task shall be removed if --testlevelsplit options exists
+# -s/--suite shall be removed if --testlevelsplit options does not exist
+def _replace_arg_files(pabot_args, opts_sub):
+    if not opts_sub.get('argumentfile') or not opts_sub['argumentfile']:
+        return
+    arg_file_list = opts_sub['argumentfile']
+    temp_file_list = []
+    test_level = pabot_args.get('testlevelsplit')
+
+    for arg_file_path in arg_file_list:
+        with open(arg_file_path, 'r') as arg_file:
+            arg_file_lines = arg_file.readlines()
+        if not arg_file_lines:
+            continue
+        with tempfile.NamedTemporaryFile(delete=False) as temp_file:
+            for line in arg_file_lines:
+                if test_level and _is_test_option(line):
+                    continue
+                elif not test_level and _is_suite_option(line):
+                    continue
+                temp_file.write(line.encode('utf-8'))
+        temp_file_list.append(temp_file.name)
+
+    opts_sub['argumentfile'] = temp_file_list
+    atexit.register(cleanup_temp_file, temp_file_list)
+
+def _is_suite_option(line):
+    return line.startswith('-s ') or line.startswith('--suite ')
+
+def _is_test_option(line):
+    return line.startswith('-t ') or line.startswith('--test ') or line.startswith('--task ')
+
+# clean the temp argument files before exiting the pabot process
+def cleanup_temp_file(temp_file_list):
+    for temp_file in temp_file_list:
+        if os.path.exists(temp_file):
+            os.remove(temp_file)
 
 def _parse_shard(arg):
     # type: (str) -> Tuple[int, int]

tests/argument_file/arg_case_options.txt

@@ -0,0 +1,2 @@
+--test  Test 1-1
+-t Test 1-3

tests/argument_file/arg_mixed_options.txt

@@ -0,0 +1,4 @@
+--test  *3
+--suite  Suite-2
+-s  Suite-1
+--loglevel debug

tests/argument_file/arg_suite_options.txt

@@ -0,0 +1,2 @@
+--suite  Suite-1
+-s Suite-3

tests/argument_file/suite-1.robot

@@ -0,0 +1,10 @@
+*** Test Cases ***
+Test 1-1
+    Log  ${TEST_NAME}
+
+Test 1-2
+    Log  ${TEST_NAME}
+
+Test 1-3
+    Log  ${TEST_NAME}
+    Log   debug message   level=DEBUG

tests/argument_file/suite-2.robot

@@ -0,0 +1,10 @@
+*** Test Cases ***
+Test 2-1
+    Log  ${TEST_NAME}
+
+Test 2-2
+    Log  ${TEST_NAME}
+
+Test 2-3
+    Log  ${TEST_NAME}
+    Log   debug message   level=DEBUG

tests/argument_file/suite-3.robot

@@ -0,0 +1,9 @@
+*** Test Cases ***
+Test 3-1
+    Log  ${TEST_NAME}
+
+Test 3-2
+    Log  ${TEST_NAME}
+
+Test 3-3
+    Log  ${TEST_NAME}

tests/test_arguments_output.py

@@ -1,3 +1,4 @@
+import os
 import sys
 import tempfile
 import textwrap
@@ -65,3 +66,195 @@ def test_argumentfile_outputs(self):
         else:
             self.assertIn(b"PASSED", stdout, stderr)
             self.assertIn(b"failed", stdout, stderr)
+
+    # copy all prepared files to temporary directory
+    def _copy_files_to_tmp(self):
+        for item in os.listdir("tests/argument_file"):
+            src_item = os.path.join("tests/argument_file", item)
+            dst_item = os.path.join(self.tmpdir, item)
+            shutil.copy2(src_item, dst_item)
+
+    # read file content
+    def _read_output_file(self, file_path):
+        with open(file_path, 'r', encoding='utf-8') as output_file:
+            content = output_file.read()
+        return content.replace("=", "").encode("utf-8")
+
+    def test_case_level_with_case_args_file(self):
+        self._copy_files_to_tmp()
+
+        process = subprocess.Popen(
+          [
+            sys.executable,
+            "-m", "pabot.pabot",
+            "--processes",
+            "2",
+            "--testlevelsplit",
+            "--argumentfile",
+            os.sep.join([self.tmpdir, "arg_case_options.txt"]),
+            "--outputdir",
+            self.tmpdir,
+            self.tmpdir
+          ],
+          cwd=self.tmpdir,
+          stdout=subprocess.PIPE,
+          stderr=subprocess.PIPE
+        )
+        stdout, stderr = process.communicate()
+
+        # validate the final output
+        self.assertIn(b'2 tests, 2 passed, 0 failed, 0 skipped.', stdout)
+
+        # validate the output for each process, there should be only one case for each process
+        pabot_result_0 = "{}/pabot_results/0/robot_stdout.out".format(self.tmpdir)
+        pabot_result_1 = "{}/pabot_results/1/robot_stdout.out".format(self.tmpdir)
+
+        stdout_0 = self._read_output_file(pabot_result_0)
+        stdout_1 = self._read_output_file(pabot_result_1)
+
+        self.assertIn(b'1 test, 1 passed, 0 failed', stdout_0)
+        self.assertIn(b'1 test, 1 passed, 0 failed', stdout_1)
+
+    def test_suite_level_with_case_args_file(self):
+        self._copy_files_to_tmp()
+
+        process = subprocess.Popen(
+          [
+            sys.executable,
+            "-m", "pabot.pabot",
+            "--processes",
+            "2",
+            "--argumentfile",
+            os.sep.join([self.tmpdir, "arg_case_options.txt"]),
+            "--outputdir",
+            self.tmpdir,
+            self.tmpdir
+          ],
+          cwd=self.tmpdir,
+          stdout=subprocess.PIPE,
+          stderr=subprocess.PIPE
+        )
+        stdout, stderr = process.communicate()
+
+        # validate the final output
+        self.assertIn(b'2 tests, 2 passed, 0 failed, 0 skipped.', stdout)
+
+        # validate the output for each process, there should be only one case for each process
+        pabot_result_0 = "{}/pabot_results/0/robot_stdout.out".format(self.tmpdir)
+
+        stdout_0 = self._read_output_file(pabot_result_0)
+
+        self.assertIn(b'2 tests, 2 passed, 0 failed', stdout_0)
+
+    def test_suite_level_with_suite_args_file(self):
+        self._copy_files_to_tmp()
+
+        process = subprocess.Popen(
+          [
+            sys.executable,
+            "-m", "pabot.pabot",
+            "--processes",
+            "2",
+            "-A",
+            os.sep.join([self.tmpdir, "arg_suite_options.txt"]),
+            "--outputdir",
+            self.tmpdir,
+            self.tmpdir
+          ],
+          cwd=self.tmpdir,
+          stdout=subprocess.PIPE,
+          stderr=subprocess.PIPE
+        )
+        stdout, stderr = process.communicate()
+
+        # validate the final output
+        self.assertIn(b'6 tests, 6 passed, 0 failed, 0 skipped.', stdout)
+
+        # validate the output for each process, there should be only one case for each process
+        pabot_result_0 = "{}/pabot_results/0/robot_stdout.out".format(self.tmpdir)
+        pabot_result_1 = "{}/pabot_results/1/robot_stdout.out".format(self.tmpdir)
+
+        stdout_0 = self._read_output_file(pabot_result_0)
+        stdout_1 = self._read_output_file(pabot_result_1)
+
+        self.assertIn(b'3 tests, 3 passed, 0 failed', stdout_0)
+        self.assertIn(b'3 tests, 3 passed, 0 failed', stdout_1)
+
+    def test_case_level_with_suite_args_file(self):
+        self._copy_files_to_tmp()
+
+        process = subprocess.Popen(
+          [
+            sys.executable,
+            "-m", "pabot.pabot",
+            "--processes",
+            "2",
+            "--testlevelsplit",
+            "-A",
+            os.sep.join([self.tmpdir, "arg_suite_options.txt"]),
+            "--outputdir",
+            self.tmpdir,
+            self.tmpdir
+          ],
+          cwd=self.tmpdir,
+          stdout=subprocess.PIPE,
+          stderr=subprocess.PIPE
+        )
+        stdout, stderr = process.communicate()
+
+        # validate the final output
+        self.assertIn(b'6 tests, 6 passed, 0 failed, 0 skipped.', stdout)
+
+        # validate the output for each process, there should be only one case for each process
+        pabot_result_0 = "{}/pabot_results/0/robot_stdout.out".format(self.tmpdir)
+        pabot_result_5 = "{}/pabot_results/5/robot_stdout.out".format(self.tmpdir)
+
+        stdout_0 = self._read_output_file(pabot_result_0)
+        stdout_5 = self._read_output_file(pabot_result_5)
+
+        self.assertIn(b'1 test, 1 passed, 0 failed', stdout_0)
+        self.assertIn(b'1 test, 1 passed, 0 failed', stdout_5)
+
+    def test_suite_level_with_mixed_args_file(self):
+        self._copy_files_to_tmp()
+
+        process = subprocess.Popen(
+          [
+            sys.executable,
+            "-m", "pabot.pabot",
+            "--testlevelsplit",
+            "--processes",
+            "2",
+            "--argumentfile",
+            os.sep.join([self.tmpdir, "arg_mixed_options.txt"]),
+            "--outputdir",
+            self.tmpdir,
+            self.tmpdir
+          ],
+          cwd=self.tmpdir,
+          stdout=subprocess.PIPE,
+          stderr=subprocess.PIPE
+        )
+        stdout, stderr = process.communicate()
+
+        # validate the final output
+        self.assertIn(b'2 tests, 2 passed, 0 failed, 0 skipped.', stdout)
+
+        # validate the output for each process, there should be only one case for each process
+        pabot_result_0 = "{}/pabot_results/0/robot_stdout.out".format(self.tmpdir)
+        pabot_result_1 = "{}/pabot_results/1/robot_stdout.out".format(self.tmpdir)
+
+        stdout_0 = self._read_output_file(pabot_result_0)
+        stdout_1 = self._read_output_file(pabot_result_1)
+
+        self.assertIn(b'1 test, 1 passed, 0 failed', stdout_0)
+        self.assertIn(b'1 test, 1 passed, 0 failed', stdout_1)
+
+        # validate if the --loglevel option works
+        pabot_output_0 = "{}/pabot_results/0/output.xml".format(self.tmpdir)
+        pabot_output_1 = "{}/pabot_results/1/output.xml".format(self.tmpdir)
+        output_0 = self._read_output_file(pabot_output_0)
+        output_1 = self._read_output_file(pabot_output_1)
+
+        self.assertIn(b'debug message', output_0)
+        self.assertIn(b'debug message', output_1)