Add run_tool_with_logging as a standalone scripts
As part of the build team's effort to cleanup envseup.sh.
Remove the run_tool_with_logging rundtion from envsetup.sh and add it
as a standalone script under build/soong/bin
Test: atest run_tool_with_logging
Test: manually do source envsetup.sh and run adb devices -l and check
the event log is sent to clearcut, tested with both bash and zsh
Bug: 342066042
Change-Id: I2fbc69a064832af61d75d14b0e33a62cb56186df
diff --git a/tests/run_tool_with_logging_test.py b/tests/run_tool_with_logging_test.py
new file mode 100644
index 0000000..57a6d62
--- /dev/null
+++ b/tests/run_tool_with_logging_test.py
@@ -0,0 +1,337 @@
+# Copyright 2024 The Android Open Source Project
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import dataclasses
+import glob
+from importlib import resources
+import logging
+import os
+from pathlib import Path
+import re
+import shutil
+import signal
+import stat
+import subprocess
+import sys
+import tempfile
+import textwrap
+import time
+import unittest
+
+EXII_RETURN_CODE = 0
+INTERRUPTED_RETURN_CODE = 130
+
+
+class RunToolWithLoggingTest(unittest.TestCase):
+
+ @classmethod
+ def setUpClass(cls):
+ super().setUpClass()
+ # Configure to print logging to stdout.
+ logging.basicConfig(filename=None, level=logging.DEBUG)
+ console = logging.StreamHandler(sys.stdout)
+ logging.getLogger("").addHandler(console)
+
+ def setUp(self):
+ super().setUp()
+ self.working_dir = tempfile.TemporaryDirectory()
+ # Run all the tests from working_dir which is our temp Android build top.
+ os.chdir(self.working_dir.name)
+
+ self.logging_script_path = self._import_executable("run_tool_with_logging")
+
+ def tearDown(self):
+ self.working_dir.cleanup()
+ super().tearDown()
+
+ def test_does_not_log_when_logger_var_empty(self):
+ test_tool = TestScript.create(self.working_dir)
+
+ self._run_script_and_wait(f"""
+ export ANDROID_TOOL_LOGGER=""
+ {self.logging_script_path} "FAKE_TOOL" {test_tool.executable} arg1 arg2
+ """)
+
+ test_tool.assert_called_once_with_args("arg1 arg2")
+
+ def test_does_not_log_with_logger_unset(self):
+ test_tool = TestScript.create(self.working_dir)
+
+ self._run_script_and_wait(f"""
+ unset ANDROID_TOOL_LOGGER
+ {self.logging_script_path} "FAKE_TOOL" {test_tool.executable} arg1 arg2
+ """)
+
+ test_tool.assert_called_once_with_args("arg1 arg2")
+
+ def test_log_success_with_logger_enabled(self):
+ test_tool = TestScript.create(self.working_dir)
+ test_logger = TestScript.create(self.working_dir)
+
+ self._run_script_and_wait(f"""
+ export ANDROID_TOOL_LOGGER="{test_logger.executable}"
+ {self.logging_script_path} "FAKE_TOOL" {test_tool.executable} arg1 arg2
+ """)
+
+ test_tool.assert_called_once_with_args("arg1 arg2")
+ expected_logger_args = (
+ "--tool_tag=FAKE_TOOL --start_timestamp=\d+\.\d+ --end_timestamp="
+ "\d+\.\d+ --tool_args=arg1 arg2 --exit_code=0"
+ )
+ test_logger.assert_called_once_with_args(expected_logger_args)
+
+ def test_run_tool_output_is_same_with_and_without_logging(self):
+ test_tool = TestScript.create(self.working_dir, "echo 'tool called'")
+ test_logger = TestScript.create(self.working_dir)
+
+ run_tool_with_logging_stdout, run_tool_with_logging_stderr = (
+ self._run_script_and_wait(f"""
+ export ANDROID_TOOL_LOGGER="{test_logger.executable}"
+ {self.logging_script_path} "FAKE_TOOL" {test_tool.executable} arg1 arg2
+ """)
+ )
+
+ run_tool_without_logging_stdout, run_tool_without_logging_stderr = (
+ self._run_script_and_wait(f"""
+ export ANDROID_TOOL_LOGGER="{test_logger.executable}"
+ {test_tool.executable} arg1 arg2
+ """)
+ )
+
+ self.assertEqual(
+ run_tool_with_logging_stdout, run_tool_without_logging_stdout
+ )
+ self.assertEqual(
+ run_tool_with_logging_stderr, run_tool_without_logging_stderr
+ )
+
+ def test_logger_output_is_suppressed(self):
+ test_tool = TestScript.create(self.working_dir)
+ test_logger = TestScript.create(self.working_dir, "echo 'logger called'")
+
+ run_tool_with_logging_output, _ = self._run_script_and_wait(f"""
+ export ANDROID_TOOL_LOGGER="{test_logger.executable}"
+ {self.logging_script_path} "FAKE_TOOL" {test_tool.executable} arg1 arg2
+ """)
+
+ self.assertNotIn("logger called", run_tool_with_logging_output)
+
+ def test_logger_error_is_suppressed(self):
+ test_tool = TestScript.create(self.working_dir)
+ test_logger = TestScript.create(
+ self.working_dir, "echo 'logger failed' > /dev/stderr; exit 1"
+ )
+
+ _, err = self._run_script_and_wait(f"""
+ export ANDROID_TOOL_LOGGER="{test_logger.executable}"
+ {self.logging_script_path} "FAKE_TOOL" {test_tool.executable} arg1 arg2
+ """)
+
+ self.assertNotIn("logger failed", err)
+
+ def test_log_success_when_tool_interrupted(self):
+ test_tool = TestScript.create(self.working_dir, script_body="sleep 100")
+ test_logger = TestScript.create(self.working_dir)
+
+ process = self._run_script(f"""
+ export ANDROID_TOOL_LOGGER="{test_logger.executable}"
+ {self.logging_script_path} "FAKE_TOOL" {test_tool.executable} arg1 arg2
+ """)
+
+ pgid = os.getpgid(process.pid)
+ # Give sometime for the subprocess to start.
+ time.sleep(1)
+ # Kill the subprocess and any processes created in the same group.
+ os.killpg(pgid, signal.SIGINT)
+
+ returncode, _, _ = self._wait_for_process(process)
+ self.assertEqual(returncode, INTERRUPTED_RETURN_CODE)
+
+ expected_logger_args = (
+ "--tool_tag=FAKE_TOOL --start_timestamp=\d+\.\d+ --end_timestamp="
+ "\d+\.\d+ --tool_args=arg1 arg2 --exit_code=130"
+ )
+ test_logger.assert_called_once_with_args(expected_logger_args)
+
+ def test_logger_can_be_toggled_on(self):
+ test_tool = TestScript.create(self.working_dir)
+ test_logger = TestScript.create(self.working_dir)
+
+ self._run_script_and_wait(f"""
+ export ANDROID_TOOL_LOGGER=""
+ export ANDROID_TOOL_LOGGER="{test_logger.executable}"
+ {self.logging_script_path} "FAKE_TOOL" {test_tool.executable} arg1 arg2
+ """)
+
+ test_logger.assert_called_with_times(1)
+
+ def test_logger_can_be_toggled_off(self):
+ test_tool = TestScript.create(self.working_dir)
+ test_logger = TestScript.create(self.working_dir)
+
+ self._run_script_and_wait(f"""
+ export ANDROID_TOOL_LOGGER="{test_logger.executable}"
+ export ANDROID_TOOL_LOGGER=""
+ {self.logging_script_path} "FAKE_TOOL" {test_tool.executable} arg1 arg2
+ """)
+
+ test_logger.assert_not_called()
+
+ def test_integration_tool_event_logger_dry_run(self):
+ test_tool = TestScript.create(self.working_dir)
+ logger_path = self._import_executable("tool_event_logger")
+
+ self._run_script_and_wait(f"""
+ TMPDIR="{self.working_dir.name}"
+ export ANDROID_TOOL_LOGGER="{logger_path}"
+ export ANDROID_TOOL_LOGGER_EXTRA_ARGS="--dry_run"
+ {self.logging_script_path} "FAKE_TOOL" {test_tool.executable} arg1 arg2
+ """)
+
+ self._assert_logger_dry_run()
+
+ def test_tool_args_do_not_fail_logger(self):
+ test_tool = TestScript.create(self.working_dir)
+ logger_path = self._import_executable("tool_event_logger")
+
+ self._run_script_and_wait(f"""
+ TMPDIR="{self.working_dir.name}"
+ export ANDROID_TOOL_LOGGER="{logger_path}"
+ export ANDROID_TOOL_LOGGER_EXTRA_ARGS="--dry_run"
+ {self.logging_script_path} "FAKE_TOOL" {test_tool.executable} --tool-arg1
+ """)
+
+ self._assert_logger_dry_run()
+
+ def _import_executable(self, executable_name: str) -> Path:
+ # logger = "tool_event_logger"
+ executable_path = Path(self.working_dir.name).joinpath(executable_name)
+ with resources.as_file(
+ resources.files("testdata").joinpath(executable_name)
+ ) as p:
+ shutil.copy(p, executable_path)
+ Path.chmod(executable_path, 0o755)
+ return executable_path
+
+ def _assert_logger_dry_run(self):
+ log_files = glob.glob(self.working_dir.name + "/tool_event_logger_*/*.log")
+ self.assertEqual(len(log_files), 1)
+
+ with open(log_files[0], "r") as f:
+ lines = f.readlines()
+ self.assertEqual(len(lines), 1)
+ self.assertIn("dry run", lines[0])
+
+ def _run_script_and_wait(self, test_script: str) -> tuple[str, str]:
+ process = self._run_script(test_script)
+ returncode, out, err = self._wait_for_process(process)
+ logging.debug("script stdout: %s", out)
+ logging.debug("script stderr: %s", err)
+ self.assertEqual(returncode, EXII_RETURN_CODE)
+ return out, err
+
+ def _run_script(self, test_script: str) -> subprocess.Popen:
+ return subprocess.Popen(
+ test_script,
+ shell=True,
+ stdout=subprocess.PIPE,
+ stderr=subprocess.PIPE,
+ text=True,
+ start_new_session=True,
+ executable="/bin/bash",
+ )
+
+ def _wait_for_process(
+ self, process: subprocess.Popen
+ ) -> tuple[int, str, str]:
+ pgid = os.getpgid(process.pid)
+ out, err = process.communicate()
+ # Wait for all process in the same group to complete since the logger runs
+ # as a separate detached process.
+ self._wait_for_process_group(pgid)
+ return (process.returncode, out, err)
+
+ def _wait_for_process_group(self, pgid: int, timeout: int = 5):
+ """Waits for all subprocesses within the process group to complete."""
+ start_time = time.time()
+ while True:
+ if time.time() - start_time > timeout:
+ raise TimeoutError(
+ f"Process group did not complete after {timeout} seconds"
+ )
+ for pid in os.listdir("/proc"):
+ if pid.isdigit():
+ try:
+ if os.getpgid(int(pid)) == pgid:
+ time.sleep(0.1)
+ break
+ except (FileNotFoundError, PermissionError, ProcessLookupError):
+ pass
+ else:
+ # All processes have completed.
+ break
+
+
+@dataclasses.dataclass
+class TestScript:
+ executable: Path
+ output_file: Path
+
+ def create(temp_dir: Path, script_body: str = ""):
+ with tempfile.NamedTemporaryFile(dir=temp_dir.name, delete=False) as f:
+ output_file = f.name
+
+ with tempfile.NamedTemporaryFile(dir=temp_dir.name, delete=False) as f:
+ executable = f.name
+ executable_contents = textwrap.dedent(f"""
+ #!/bin/bash
+
+ echo "${{@}}" >> {output_file}
+ {script_body}
+ """)
+ f.write(executable_contents.encode("utf-8"))
+
+ Path.chmod(f.name, os.stat(f.name).st_mode | stat.S_IEXEC)
+
+ return TestScript(executable, output_file)
+
+ def assert_called_with_times(self, expected_call_times: int):
+ lines = self._read_contents_from_output_file()
+ assert len(lines) == expected_call_times, (
+ f"Expect to call {expected_call_times} times, but actually called"
+ f" {len(lines)} times."
+ )
+
+ def assert_called_with_args(self, expected_args: str):
+ lines = self._read_contents_from_output_file()
+ assert len(lines) > 0
+ assert re.search(expected_args, lines[0]), (
+ f"Expect to call with args {expected_args}, but actually called with"
+ f" args {lines[0]}."
+ )
+
+ def assert_not_called(self):
+ self.assert_called_with_times(0)
+
+ def assert_called_once_with_args(self, expected_args: str):
+ self.assert_called_with_times(1)
+ self.assert_called_with_args(expected_args)
+
+ def _read_contents_from_output_file(self) -> list[str]:
+ with open(self.output_file, "r") as f:
+ return f.readlines()
+
+
+if __name__ == "__main__":
+ unittest.main()