Fix integ tests for windows platform (#3811)

Signed-off-by: Zelin Hao <[email protected]>

src/system/execute.py

@@ -20,7 +20,7 @@ def execute(command: str, dir: str, capture: bool = True, raise_on_failure: bool
     :returns a tuple containing the exit code, stdout, and stderr.
     """
     logging.info(f'Executing "{command}" in {dir}')
-    result = subprocess.run(command, cwd=dir, shell=True, capture_output=capture, text=True)
+    result = subprocess.run(command, cwd=dir, shell=True, capture_output=capture, text=True, encoding='utf-8')
     if raise_on_failure:
         result.check_returncode()
     return result.returncode, result.stdout, result.stderr

src/system/process.py

@@ -26,8 +26,8 @@ def start(self, command: str, cwd: str, require_sudo: bool = False) -> None:
         if self.started:
             raise ProcessStartedError(self.pid)
 
-        self.stdout = tempfile.NamedTemporaryFile(mode="r+", delete=False)
-        self.stderr = tempfile.NamedTemporaryFile(mode="r+", delete=False)
+        self.stdout = tempfile.NamedTemporaryFile(mode="r+", delete=False, encoding='utf-8')
+        self.stderr = tempfile.NamedTemporaryFile(mode="r+", delete=False, encoding='utf-8')
 
         self.require_sudo = require_sudo
 
@@ -58,15 +58,33 @@ def terminate(self) -> int:
         logging.info(f"Process killed with exit code {self.process.returncode}")
 
         if self.stdout:
-            self.__stdout_data__ = open(self.stdout.name, 'r').read()
+            self.stdout.seek(0)
+            self.__stdout_data__ = self.stdout.read()
+            self.stdout.flush()
             self.stdout.close()
-            os.remove(self.stdout.name)
+            for proc in psutil.process_iter():
+                try:
+                    for item in proc.open_files():
+                        if self.stdout.name == item.path:
+                            raise Exception(f"stdout {item} is being used by process {proc}")
+                except Exception as err:
+                    logging.error(f"{err.args}")
+            os.unlink(self.stdout.name)
             self.stdout = None
 
         if self.stderr:
-            self.__stderr_data__ = open(self.stderr.name, 'r').read()
+            self.stderr.seek(0)
+            self.__stderr_data__ = self.stderr.read()
+            self.stderr.flush()
             self.stderr.close()
-            os.remove(self.stderr.name)
+            for proc in psutil.process_iter():
+                try:
+                    for item in proc.open_files():
+                        if self.stderr.name == item.path:
+                            raise Exception(f"stderr {item} is being used by process {proc}")
+                except Exception as err:
+                    logging.error(f"{err.args}")
+            os.unlink(self.stderr.name)
             self.stderr = None
 
         self.return_code = self.process.returncode

src/test_workflow/test_recorder/test_recorder.py

@@ -49,9 +49,9 @@ def _create_base_folder_structure(self, component_name: str, component_test_conf
         return os.path.realpath(dest_directory)
 
     def _generate_std_files(self, stdout: str, stderr: str, output_path: str) -> None:
-        with open(os.path.join(output_path, "stdout.txt"), "w") as stdout_file:
+        with open(os.path.join(output_path, "stdout.txt"), "w", encoding='utf-8') as stdout_file:
             stdout_file.write(stdout)
-        with open(os.path.join(output_path, "stderr.txt"), "w") as stderr_file:
+        with open(os.path.join(output_path, "stderr.txt"), "w", encoding='utf-8') as stderr_file:
             stderr_file.write(stderr)
 
     def _generate_yml(self, test_result_data: TestResultData, output_path: str) -> str:
@@ -68,7 +68,7 @@ def _generate_yml(self, test_result_data: TestResultData, output_path: str) -> s
             "test_result": "PASS" if (test_result_data.exit_code == 0) else "FAIL",
             "test_result_files": test_result_file
         }
-        with open(os.path.join(output_path, "%s.yml" % test_result_data.component_name), "w") as file:
+        with open(os.path.join(output_path, "%s.yml" % test_result_data.component_name), "w", encoding='utf-8') as file:
             yaml.dump(outcome, file)
         return os.path.realpath("%s.yml" % test_result_data.component_name)
 

tests/tests_system/test_process.py

@@ -42,7 +42,7 @@ def test(self) -> None:
     def test_file_open_mode(self, mock_tempfile: MagicMock) -> None:
         process_handler = Process()
         process_handler.start("./tests/tests_system/data/wait_for_input.sh", ".")
-        mock_tempfile.assert_has_calls([call(delete=False, mode='r+'), call(delete=False, mode='r+')])
+        mock_tempfile.assert_has_calls([call(delete=False, encoding='utf-8', mode='r+')])
 
     def test_start_twice(self) -> None:
         process_handler = Process()
@@ -60,3 +60,14 @@ def test_terminate_unstarted_process(self) -> None:
             process_handler.terminate()
 
         self.assertEqual(str(ctx.exception), "Process has not started")
+
+    @patch('psutil.Process')
+    @patch('subprocess.Popen')
+    @patch('psutil.process_iter')
+    def test_terminate_file_not_closed(self, procs: MagicMock, subprocess: MagicMock, process: MagicMock) -> None:
+        process_handler = Process()
+
+        process_handler.start("mock_command", "mock_cwd")
+        process_handler.terminate()
+
+        procs.assert_called

tests/tests_test_workflow/test_recorder/test_test_recorder.py

@@ -80,7 +80,7 @@ def test_generate_yml(self, mock_local_cluster_logs: Mock, mock_remote_cluster_l
 
         component_yml = test_recorder._generate_yml(mock_test_result_data, "working-directory")
 
-        mock_open.assert_has_calls([call(os.path.join("working-directory", "sql.yml"), 'w')])
+        mock_open.assert_has_calls([call(os.path.join("working-directory", "sql.yml"), 'w', encoding='utf-8')])
         mock_file_path.assert_called_once_with(None, "sql", "with-security")
         mock_list_files.assert_called()
         mock_update_path.assert_called()
@@ -114,6 +114,24 @@ def test_get_list_files(self, mock_local_cluster_logs: Mock, mock_remote_cluster
         list_files = test_recorder._get_list_files("mock_dir")
         self.assertEqual(list_files, ["file1", "file2"])
 
+    @patch("builtins.open", new_callable=mock_open)
+    def test_generate_std_files(self, mock_open: Mock, *mock: Any) -> None:
+        test_recorder = TestRecorder(
+            "1234",
+            "integ-test",
+            "working-directory"
+        )
+        test_recorder._generate_std_files("mock_stdout", "mock_stderr", "mock_path")
+
+        mock_file = MagicMock()
+        mock_open.return_value = mock_file
+
+        mock_open.assert_has_calls([call(os.path.join("mock_path", "stdout.txt"), 'w', encoding='utf-8')])
+        mock_open.assert_has_calls([call(os.path.join("mock_path", "stderr.txt"), 'w', encoding='utf-8')])
+
+        mock_open.assert_has_calls([call().write("mock_stdout")])
+        mock_open.assert_has_calls([call().write("mock_stderr")])
+
 
 class TestLocalClusterLogs(unittest.TestCase):