refactor: Use python to rewrite process testing shell script. (#24)

Signed-off-by: Hongli Chen <[email protected]>

test/openjd/adaptor_runtime/integ/process/scripts/echo_sleep_n_times.py

@@ -0,0 +1,17 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+
+import sys
+import time
+
+
+def repeat_message(message, count):
+    for _ in range(count):
+        print(message)
+        print(message, file=sys.stderr)
+        time.sleep(0.01)
+
+
+if __name__ == "__main__":
+    message = sys.argv[1]
+    num_times = int(sys.argv[2])
+    repeat_message(message, num_times)

test/openjd/adaptor_runtime/integ/process/scripts/signals_test.py

@@ -0,0 +1,39 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+
+import signal
+import sys
+import time
+from datetime import datetime
+import logging
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+
+def flush_logging():
+    for handler in logger.handlers:
+        handler.flush()
+
+
+def func_trap(signum, frame):
+    logger.info(f"Trapped: {signal.Signals(signum).name}")
+    flush_logging()
+    if exit_after_signal:
+        exit(0)
+
+
+def set_signal_handlers():
+    signals = [signal.SIGTERM, signal.SIGINT]
+    for sig in signals:
+        signal.signal(sig, func_trap)
+
+
+if __name__ == "__main__":
+    exit_after_signal = sys.argv[1] == "True"
+    logger.info("Starting signals_test.py Script")
+    flush_logging()
+    set_signal_handlers()
+
+    while True:
+        logger.info(datetime.now().strftime("%Y-%m-%d_%H:%M:%S"))
+        time.sleep(1)

test/openjd/adaptor_runtime/integ/process/test_integration_logging_subprocess.py

@@ -5,6 +5,7 @@
 import logging
 import os
 import re
+import sys
 import time
 from logging import DEBUG
 from unittest import mock
@@ -38,18 +39,19 @@ def test_stop_process(self, grace_period, expected_output, caplog: pytest.LogCap
         Testing that we stop the process immediately and after SIGTERM fails.
         """
         test_file = os.path.join(
-            os.path.abspath(os.path.dirname(__file__)), "scripts", "no_sigterm.sh"
+            os.path.abspath(os.path.dirname(__file__)), "scripts", "signals_test.py"
         )
         caplog.set_level(DEBUG)
-        p = LoggingSubprocess(args=[test_file])
+        p = LoggingSubprocess(args=[sys.executable, test_file, "False"])
 
         # This is because we are giving the subprocess time to load and
         # register the sigterm signal handler with the OS.
-        while "Starting no_sigterm.sh Script" not in caplog.text:
-            True
+        while "Starting signals_test.py Script" not in caplog.text:
+            time.sleep(0.2)
 
         p.terminate(grace_period)
 
+        assert "Starting signals_test.py Script" in caplog.text
         for output in expected_output:
             assert output in caplog.text
 
@@ -60,22 +62,22 @@ def test_terminate_process(self, caplog):
         when SIGTERM was sent and SIGKILL was not needed.
         """
         test_file = os.path.join(
-            os.path.abspath(os.path.dirname(__file__)), "scripts", "print_signals.sh"
+            os.path.abspath(os.path.dirname(__file__)), "scripts", "signals_test.py"
         )
         caplog.set_level(DEBUG)
-        p = LoggingSubprocess(args=[test_file])
+        p = LoggingSubprocess(args=[sys.executable, test_file, "True"])
 
         # This is because we are giving the subprocess time to load and ignore the sigterm signal.
-        while "Starting print_signals.sh Script" not in caplog.text:
-            True
+        while "Starting signals_test.py Script" not in caplog.text:
+            time.sleep(0.2)
 
         p.terminate(5)  # Sometimes, when this is 1 second the process doesn't terminate in time.
 
         assert (
             "Sending the SIGTERM signal to pid=" in caplog.text
         )  # Asserting the SIGTERM signal was sent to the subprocess
         assert (
-            "Trapped: TERM" in caplog.text
+            "Trapped: SIGTERM" in caplog.text
         )  # Asserting the SIGTERM was received by the subprocess.
         assert (
             "now sending the SIGKILL signal." not in caplog.text
@@ -120,11 +122,11 @@ def test_log_levels(self, log_level: int, caplog):
         message = "Hello World"
 
         test_file = os.path.join(
-            os.path.abspath(os.path.dirname(__file__)), "scripts", "echo_sleep_n_times.sh"
+            os.path.abspath(os.path.dirname(__file__)), "scripts", "echo_sleep_n_times.py"
         )
         # WHEN
         p = LoggingSubprocess(
-            args=[test_file, message, "1"],
+            args=[sys.executable, test_file, message, "1"],
         )
         p.wait()
 
@@ -157,17 +159,19 @@ def test_stdouthandler_invoked(self, regex, output, echo_count, stdout, stderr):
         regex_callbacks = [RegexCallback([regex], callback)]
         regex_handler = RegexHandler(regex_callbacks)
         test_file = os.path.join(
-            os.path.abspath(os.path.dirname(__file__)), "scripts", "echo_sleep_n_times.sh"
+            os.path.abspath(os.path.dirname(__file__)), "scripts", "echo_sleep_n_times.py"
         )
         # WHEN
         p = LoggingSubprocess(
-            args=[test_file, output, str(echo_count)],
+            args=[sys.executable, test_file, output, str(echo_count)],
             stdout_handler=regex_handler if stdout else None,
             stderr_handler=regex_handler if stderr else None,
         )
         p.wait()
         time.sleep(0.01)  # magic sleep - logging handler has a delay and test can exit too fast
-
+        print(
+            [sys.executable, test_file, output, str(echo_count)],
+        )
         # THEN
         assert callback.call_count == echo_count * (stdout + stderr)
         assert all(c[0][0].re == regex for c in callback.call_args_list)
@@ -198,15 +202,15 @@ def test_multiple_processes_invoked_independently(self, regex, output, echo_coun
         stderr_handlers = regex_handlers[num_procs:]
 
         test_file = os.path.join(
-            os.path.abspath(os.path.dirname(__file__)), "scripts", "echo_sleep_n_times.sh"
+            os.path.abspath(os.path.dirname(__file__)), "scripts", "echo_sleep_n_times.py"
         )
 
         # WHEN
         procs = []
         for i in range(num_procs):
             procs.append(
                 LoggingSubprocess(
-                    args=[test_file, output, str(echo_count)],
+                    args=[sys.executable, test_file, output, str(echo_count)],
                     stdout_handler=stdout_handlers[i],
                     stderr_handler=stderr_handlers[i],
                 )

test/openjd/adaptor_runtime/integ/process/test_integration_managed_process.py

@@ -4,6 +4,7 @@
 
 import os
 import re
+import sys
 import time
 from logging import INFO
 from typing import List
@@ -59,12 +60,13 @@ def test_regexhandler_invoked(self, regex, output, echo_count, stdout, stderr):
         # GIVEN
         class FakeManagedProcess(ManagedProcess):
             def get_executable(self) -> str:
-                return os.path.join(
-                    os.path.abspath(os.path.dirname(__file__)), "scripts", "echo_sleep_n_times.sh"
-                )
+                return sys.executable
 
             def get_arguments(self) -> List[str]:
-                return [output, str(echo_count)]
+                test_file = os.path.join(
+                    os.path.abspath(os.path.dirname(__file__)), "scripts", "echo_sleep_n_times.py"
+                )
+                return [test_file, output, str(echo_count)]
 
             def get_startup_directory(self) -> str | None:
                 return None