fix the issue of starting opensearch (opensearch-project#978)

* fix the issue of starting opensearch Signed-off-by: Tianle Huang <[email protected]> * add unit tests Signed-off-by: Tianle Huang <[email protected]>
naveenpajjuri · Nov 17, 2021 · 62207e8 · 62207e8
1 parent b832bf1
commit 62207e8
Show file tree

Hide file tree

Showing 4 changed files with 27 additions and 8 deletions.
diff --git a/src/system/process.py b/src/system/process.py
@@ -23,8 +23,8 @@ def start(self, command, cwd):
         if self.started:
             raise ProcessStartedError(self.pid)
 
-        self.stdout = tempfile.NamedTemporaryFile()
-        self.stderr = tempfile.NamedTemporaryFile()
+        self.stdout = tempfile.NamedTemporaryFile(mode="r+")
+        self.stderr = tempfile.NamedTemporaryFile(mode="r+")
 
         self.process = subprocess.Popen(
             command,

diff --git a/src/test_workflow/integ_test/local_test_cluster.py b/src/test_workflow/integ_test/local_test_cluster.py
@@ -113,14 +113,12 @@ def wait_for_service(self):
                     return
             except requests.exceptions.ConnectionError:
                 logging.info("Service not available yet")
-                if self.stdout:
+                if self.process_handler.output:
                     logging.info("- stdout:")
-                    with open(os.path.join(self.work_dir, self.stdout.name), "r") as stdout:
-                        logging.info(stdout.read())
-                if self.stderr:
+                    logging.info(self.process_handler.output.read())
+                if self.process_handler.error:
                     logging.info("- stderr:")
-                    with open(os.path.join(self.work_dir, self.stderr.name), "r") as stderr:
-                        logging.info(stderr.read())
+                    logging.info(self.process_handler.error.read())
 
             time.sleep(10)
         raise ClusterCreationException("Cluster is not available after 10 attempts")

diff --git a/tests/tests_system/test_process.py b/tests/tests_system/test_process.py
@@ -19,7 +19,9 @@ def test(self):
         self.assertTrue(process_handler.started)
         self.assertIsNotNone(process_handler.pid)
         self.assertIsNotNone(process_handler.output)
+        self.assertEqual(process_handler.output.mode, "r+")
         self.assertIsNotNone(process_handler.error)
+        self.assertEqual(process_handler.error.mode, "r+")
 
         return_code, stdout_data, stderr_data = process_handler.terminate()
 

diff --git a/tests/tests_test_workflow/test_integ_workflow/integ_test/test_local_test_cluster.py b/tests/tests_test_workflow/test_integ_workflow/integ_test/test_local_test_cluster.py
@@ -68,6 +68,9 @@ def __mock_response(*args, **kwargs):
         else:
             return LocalTestClusterTests.MockResponse({"status": "red"}, 404)
 
+    def __mock_connection_error_response(*args, **kwargs):
+        raise requests.exceptions.ConnectionError()
+
     def __mock_process(*args, **kwargs):
         return LocalTestClusterTests.MockProcess(12)
 
@@ -111,6 +114,22 @@ def test_wait_for_service_cluster_unavailable(self, mock_test_recorder, *mocks):
             requests.get.assert_called_once_with("http://localhost:9200/_cluster/health", verify=False, auth=("admin", "admin"))
         self.assertEqual(str(err.exception), "Cluster is not available after 10 attempts")
 
+    @patch("time.sleep")
+    @patch.object(Process, 'output')
+    @patch.object(Process, 'error')
+    @patch("requests.get", side_effect=__mock_connection_error_response)
+    @patch("test_workflow.test_recorder.test_recorder.TestRecorder")
+    def test_wait_for_service_cluster_unavailable_connection_error(self, mock_test_recorder, *mocks):
+        local_test_cluster = LocalTestCluster(
+            self.dependency_installer, self.work_dir.name, "index-management", "", self.bundle_manifest, False, "without-security", mock_test_recorder
+        )
+        with self.assertRaises(ClusterCreationException) as err:
+            local_test_cluster.wait_for_service()
+            requests.get.assert_called_once_with("http://localhost:9200/_cluster/health", verify=False, auth=("admin", "admin"))
+        self.assertEqual(str(err.exception), "Cluster is not available after 10 attempts")
+        self.assertEqual(Process.output.read.call_count, 10)
+        self.assertEqual(Process.error.read.call_count, 10)
+
     @patch.object(Process, 'terminate', return_value=("1", "2", "3"))
     def test_terminate_process(self, *mocks):
         self.local_test_cluster.terminate_process()