| #!/usr/bin/env fuchsia-vendored-python |
| # Copyright 2025 The Fuchsia Authors. All rights reserved. |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| |
| import json |
| import os |
| import sys |
| import tempfile |
| import unittest |
| from pathlib import Path |
| |
| _SCRIPT_DIR = os.path.dirname(__file__) |
| sys.path.insert(0, _SCRIPT_DIR) |
| import build_utils |
| from build_utils import BazelLauncher, BazelQueryCache, CommandResult |
| |
| |
| class FindFuchsiaDirTest(unittest.TestCase): |
| def test_find_fuchsia_dir(self) -> None: |
| with tempfile.TemporaryDirectory() as tmp_dir: |
| not_fuchsia_dir = Path(tmp_dir) / "this_is_not_fuchsia" |
| not_fuchsia_dir.mkdir() |
| fuchsia_dir = Path(tmp_dir) / "this_is_fuchsia" |
| fuchsia_dir.mkdir() |
| (fuchsia_dir / ".jiri_manifest").write_text("") |
| (fuchsia_dir / "src" / "foo").mkdir(parents=True) |
| |
| # Check function when searching from the current path. |
| saved_cwd = os.getcwd() |
| try: |
| os.chdir(not_fuchsia_dir) |
| with self.assertRaises(ValueError): |
| build_utils.find_fuchsia_dir() |
| |
| for path in ( |
| fuchsia_dir, |
| fuchsia_dir / "src", |
| fuchsia_dir / "src" / "foo", |
| ): |
| os.chdir(path) |
| self.assertEqual( |
| build_utils.find_fuchsia_dir(), |
| fuchsia_dir, |
| f"From {path}", |
| ) |
| |
| # Ensure the result is absolute even if the starting path is relative. |
| os.chdir(fuchsia_dir) |
| for path in (Path("."), Path("src"), Path("src/foo")): |
| self.assertEqual( |
| build_utils.find_fuchsia_dir(path), |
| fuchsia_dir, |
| f"From {path}", |
| ) |
| |
| finally: |
| os.chdir(saved_cwd) |
| |
| # Check function when searching from a given path. |
| with self.assertRaises(ValueError): |
| build_utils.find_fuchsia_dir(not_fuchsia_dir) |
| |
| for path in ( |
| fuchsia_dir, |
| fuchsia_dir / "src", |
| fuchsia_dir / "src" / "foo", |
| ): |
| self.assertEqual( |
| build_utils.find_fuchsia_dir(path), |
| fuchsia_dir, |
| f"From {path}", |
| ) |
| |
| |
| class FindFxBuildDirTest(unittest.TestCase): |
| def test_find_fx_build_dir(self) -> None: |
| with tempfile.TemporaryDirectory() as tmp_dir: |
| fuchsia_dir = Path(tmp_dir) |
| |
| # No .fx-build-dir present -> Path() |
| self.assertEqual(build_utils.find_fx_build_dir(fuchsia_dir), None) |
| |
| # Empty .fx-build-dir content -> Path() |
| fx_build_dir_path = fuchsia_dir / ".fx-build-dir" |
| fx_build_dir_path.write_text("") |
| self.assertEqual(build_utils.find_fx_build_dir(fuchsia_dir), None) |
| |
| # Invalid .fx-build-dir content -> Path() |
| fx_build_dir_path.write_text("does/not/exist\n") |
| self.assertEqual(build_utils.find_fx_build_dir(fuchsia_dir), None) |
| |
| # Valid build directory. |
| build_dir = fuchsia_dir / "some" / "build_dir" |
| build_dir.mkdir(parents=True) |
| |
| fx_build_dir_path.write_text("some/build_dir\n") |
| self.assertEqual( |
| build_utils.find_fx_build_dir(fuchsia_dir), build_dir |
| ) |
| |
| |
| class FindBazelPathsTest(unittest.TestCase): |
| def setUp(self) -> None: |
| self._td = tempfile.TemporaryDirectory() |
| self.fuchsia_dir = Path(self._td.name) |
| config_dir = self.fuchsia_dir / "build" / "bazel" / "config" |
| config_dir.mkdir(parents=True) |
| (config_dir / "bazel_top_dir").write_text("some/top/dir\n") |
| |
| self.build_dir = self.fuchsia_dir / "out" / "build_dir" |
| self.launcher_path = self.build_dir / "some/top/dir/bazel" |
| self.workspace_path = self.build_dir / "some/top/dir/workspace" |
| |
| def tearDown(self) -> None: |
| self._td.cleanup() |
| |
| def test_find_bazel_launcher_path(self) -> None: |
| self.assertEqual( |
| build_utils.find_bazel_launcher_path( |
| self.fuchsia_dir, self.build_dir |
| ), |
| None, |
| ) |
| |
| self.launcher_path.parent.mkdir(parents=True) |
| self.launcher_path.write_text("!") |
| self.assertEqual( |
| build_utils.find_bazel_launcher_path( |
| self.fuchsia_dir, self.build_dir |
| ), |
| self.launcher_path, |
| ) |
| |
| def test_find_bazel_workspace_path(self) -> None: |
| self.assertEqual( |
| build_utils.find_bazel_workspace_path( |
| self.fuchsia_dir, self.build_dir |
| ), |
| None, |
| ) |
| |
| self.workspace_path.mkdir(parents=True) |
| self.assertEqual( |
| build_utils.find_bazel_workspace_path( |
| self.fuchsia_dir, self.build_dir |
| ), |
| self.workspace_path, |
| ) |
| |
| |
| class FindBazelWorkspacePathTest(unittest.TestCase): |
| def test_find_bazel_workspace_path(self) -> None: |
| with tempfile.TemporaryDirectory() as tmp_dir: |
| fuchsia_dir = Path(tmp_dir) |
| config_dir = fuchsia_dir / "build" / "bazel" / "config" |
| config_dir.mkdir(parents=True) |
| (config_dir / "bazel_top_dir").write_text("some/top/dir\n") |
| |
| build_dir = fuchsia_dir / "out" / "build_dir" |
| launcher_path = build_dir / "some/top/dir/bazel" |
| |
| self.assertEqual( |
| build_utils.find_bazel_launcher_path(fuchsia_dir, build_dir), |
| None, |
| ) |
| |
| launcher_path.parent.mkdir(parents=True) |
| launcher_path.write_text("!") |
| self.assertEqual( |
| build_utils.find_bazel_launcher_path(fuchsia_dir, build_dir), |
| launcher_path, |
| ) |
| |
| |
| class GetBazelRelativeTopDirTest(unittest.TestCase): |
| def test_get_bazel_relative_topdir(self) -> None: |
| with tempfile.TemporaryDirectory() as tmp_dir: |
| fuchsia_dir = Path(tmp_dir) |
| config_dir = fuchsia_dir / "build" / "bazel" / "config" |
| config_dir.mkdir(parents=True) |
| |
| main_config = config_dir / "bazel_top_dir" |
| main_config.write_text("gen/test/bazel_workspace\n") |
| |
| topdir, input_files = build_utils.get_bazel_relative_topdir( |
| fuchsia_dir |
| ) |
| self.assertEqual(topdir, "gen/test/bazel_workspace") |
| self.assertListEqual(list(input_files), [main_config]) |
| |
| topdir, input_files = build_utils.get_bazel_relative_topdir( |
| str(fuchsia_dir) |
| ) |
| self.assertEqual(topdir, "gen/test/bazel_workspace") |
| self.assertListEqual(list(input_files), [main_config]) |
| |
| |
| class ForceSymlinkTest(unittest.TestCase): |
| def test_force_symlink(self) -> None: |
| with tempfile.TemporaryDirectory() as tmp_dir: |
| tmp_path = Path(tmp_dir).resolve() |
| |
| # Create a new symlink, then ensure its embedded target is relative. |
| # The target doesn't need to exist. |
| target_path = tmp_path / "target" / "file" |
| link_path = tmp_path / "links" / "dir" / "symlink" |
| |
| build_utils.force_symlink(link_path, target_path) |
| |
| self.assertTrue(link_path.is_symlink()) |
| self.assertEqual(str(link_path.readlink()), "../../target/file") |
| |
| # Update the target to a new path, verify the symlink was updated. |
| target_path = tmp_path / "target" / "new_file" |
| |
| build_utils.force_symlink(link_path, target_path) |
| self.assertTrue(link_path.is_symlink()) |
| self.assertEqual(str(link_path.readlink()), "../../target/new_file") |
| |
| |
| class IsHexadecimalStringTest(unittest.TestCase): |
| def test_is_hexadecimal_string(self) -> None: |
| TEST_CASES = [ |
| ("", False), |
| ("0", True), |
| ("0.", False), |
| ("F", True), |
| ("G", False), |
| ("0123456789abcdefABCDEF", True), |
| ("0123456789abcdefghijklmnopqrstuvwxyz", False), |
| ] |
| for input, expected in TEST_CASES: |
| self.assertEqual( |
| build_utils.is_hexadecimal_string(input), |
| expected, |
| msg=f"For input [{input}]", |
| ) |
| |
| |
| class IsLikelyBuildIdPathTest(unittest.TestCase): |
| def test_one(self) -> None: |
| TEST_CASES = [ |
| ("", False), |
| ("/src/.build-id", False), |
| ("/src/.build-id/0", False), |
| ("/src/.build-id/00", False), |
| ("/src/.build-id/000", False), |
| ("/src/.build-id/0/foo", False), |
| ("/src/.build-id/00/foo", True), |
| ("/src/.build-id/000/foo", False), |
| ("/src/.build-id/af/foo", True), |
| ("/src/.build-id/ag/foo", False), |
| ("/src/..build-id/00/foo", False), |
| ("/src.build-id/00/foo", False), |
| ("/src/.build-id/log.txt", False), |
| (".build-id/00/foo", True), |
| ] |
| for input, expected in TEST_CASES: |
| self.assertEqual( |
| build_utils.is_likely_build_id_path(input), |
| expected, |
| msg=f"For input [{input}]", |
| ) |
| |
| |
| class IsLikelyContentHashPathTest(unittest.TestCase): |
| def test_one(self) -> None: |
| TEST_CASES = [ |
| ("", False), |
| ("/src/.build-id/0/foo", False), |
| ("/src/.build-id/00/foo", True), |
| ("/src/blobs/0123456789abcdef0000", True), |
| ("/src/blobs/01234", False), # Too short |
| ("0123456789abcdef0000", True), |
| ] |
| for input, expected in TEST_CASES: |
| self.assertEqual( |
| build_utils.is_likely_content_hash_path(input), |
| expected, |
| msg=f"For input [{input}]", |
| ) |
| |
| |
| class BazelPathsTest(unittest.TestCase): |
| def setUp(self) -> None: |
| self._td = tempfile.TemporaryDirectory() |
| self.fuchsia_dir = Path(self._td.name) / "fuchsia" |
| self.fuchsia_dir.mkdir() |
| (self.fuchsia_dir / ".jiri_manifest").write_text("") |
| |
| self.build_dir = self.fuchsia_dir / "out" / "build_dir" |
| self.build_dir.mkdir(parents=True) |
| (self.fuchsia_dir / ".fx-build-dir").write_text("out/build_dir\n") |
| |
| build_utils.BazelPaths.write_topdir_config_for_test( |
| self.fuchsia_dir, "some/top/dir" |
| ) |
| self.launcher_path = self.build_dir / "some/top/dir/bazel" |
| self.workspace_path = self.build_dir / "some/top/dir/workspace" |
| self.output_base_path = self.build_dir / "some/top/dir/output_base" |
| |
| def tearDown(self) -> None: |
| self._td.cleanup() |
| |
| def test_new_with_no_valid_fuchsia_or_build_dirs(self) -> None: |
| saved_current_dir = Path.cwd() |
| try: |
| os.chdir(self._td.name) |
| |
| # This fails because BazelPaths.new() cannot find a Fuchsia source directory |
| # from the current directory. |
| with self.assertRaises(ValueError) as cm: |
| build_utils.BazelPaths.new() |
| self.assertEqual( |
| str(cm.exception), |
| f"Could not find Fuchsia checkout directory from: {self._td.name}", |
| ) |
| finally: |
| os.chdir(saved_current_dir) |
| |
| def test_new_with_no_valid_build_dir(self) -> None: |
| saved_current_dir = Path.cwd() |
| (self.fuchsia_dir / ".fx-build-dir").write_text("out/does_not_exist") |
| try: |
| os.chdir(self._td.name) |
| |
| # This fails because BazelPaths.new() cannot find a .fx-build-dir file from |
| # the Fuchsia source directory. |
| with self.assertRaises(ValueError) as cm: |
| paths = build_utils.BazelPaths.new(fuchsia_dir=self.fuchsia_dir) |
| self.assertEqual( |
| str(cm.exception), |
| f"Could not detect current build-directory from Fuchsia directory: {self.fuchsia_dir}", |
| ) |
| finally: |
| os.chdir(saved_current_dir) |
| |
| def test_new_with_fuchsia_dir_only(self) -> None: |
| paths = build_utils.BazelPaths.new(fuchsia_dir=self.fuchsia_dir) |
| self.assertTrue(paths) |
| self.assertEqual(paths.fuchsia_dir, self.fuchsia_dir) |
| self.assertEqual(paths.ninja_build_dir, self.build_dir) |
| self.assertEqual(paths.workspace, self.workspace_path) |
| self.assertEqual(paths.output_base, self.output_base_path) |
| self.assertEqual(paths.launcher, self.launcher_path) |
| |
| def test_new_with_build_dir_only(self) -> None: |
| paths = build_utils.BazelPaths.new(build_dir=self.build_dir) |
| self.assertTrue(paths) |
| self.assertEqual(paths.fuchsia_dir, self.fuchsia_dir) |
| self.assertEqual(paths.ninja_build_dir, self.build_dir) |
| self.assertEqual(paths.workspace, self.workspace_path) |
| self.assertEqual(paths.output_base, self.output_base_path) |
| self.assertEqual(paths.launcher, self.launcher_path) |
| |
| def test_new_with_fuchsia_and_build_dirs(self) -> None: |
| paths = build_utils.BazelPaths.new( |
| fuchsia_dir=self.fuchsia_dir, build_dir=self.build_dir |
| ) |
| self.assertTrue(paths) |
| self.assertEqual(paths.fuchsia_dir, self.fuchsia_dir) |
| self.assertEqual(paths.ninja_build_dir, self.build_dir) |
| self.assertEqual(paths.workspace, self.workspace_path) |
| self.assertEqual(paths.output_base, self.output_base_path) |
| self.assertEqual(paths.launcher, self.launcher_path) |
| |
| def test_write_topdir_config_for_test(self) -> None: |
| build_utils.BazelPaths.write_topdir_config_for_test( |
| self.fuchsia_dir, "some/other/topdir" |
| ) |
| paths = build_utils.BazelPaths(self.fuchsia_dir, self.build_dir) |
| self.assertEqual(paths.top_dir, self.build_dir / "some/other/topdir") |
| |
| def test_constructor(self) -> None: |
| paths = build_utils.BazelPaths(self.fuchsia_dir, self.build_dir) |
| self.assertEqual(paths.fuchsia_dir, self.fuchsia_dir) |
| self.assertEqual(paths.ninja_build_dir, self.build_dir) |
| self.assertEqual(paths.workspace, self.workspace_path) |
| self.assertEqual(paths.output_base, self.output_base_path) |
| self.assertEqual(paths.launcher, self.launcher_path) |
| |
| |
| class MockCurrentTime(object): |
| """A mock time.time() implementation used for TimeProfileTest. |
| |
| Initial time is always 100 seconds, first call is 10 seconds |
| """ |
| |
| def __init__(self) -> None: |
| self._current_time = 100.0 |
| |
| def reset(self) -> None: |
| self._current_time = 100.0 |
| |
| def increment(self, increment: float) -> None: |
| self._current_time += increment |
| |
| def __call__(self) -> float: |
| return self._current_time |
| |
| |
| class TimeProfileTest(unittest.TestCase): |
| def setUp(self) -> None: |
| self._now = MockCurrentTime() |
| |
| def test_empty(self) -> None: |
| p = build_utils.TimeProfile(now=self._now) |
| self.assertDictEqual(p.to_json_timings(), {}) |
| |
| def test_single_step(self) -> None: |
| now = self._now |
| |
| # Verifies that if time does not increment, duration is 0. |
| p = build_utils.TimeProfile(now=now) |
| p.start("first_step", "A first step") |
| self.assertDictEqual(p.to_json_timings(), {"first_step": 0}) |
| |
| # Increment time by 10 seconds after start(), then calls to_json_timings() |
| # directly without a stop() call. |
| now.reset() |
| p = build_utils.TimeProfile(now=self._now) |
| p.start("first_step_again", "Another first step") |
| now.increment(10) |
| self.assertDictEqual(p.to_json_timings(), {"first_step_again": 10.0}) |
| |
| # Same as above, but calls stop() before increment time again. |
| now.reset() |
| p = build_utils.TimeProfile(now=self._now) |
| p.start("first_step_again", "Another first step") |
| now.increment(10) |
| p.stop() |
| now.increment(20) |
| self.assertDictEqual(p.to_json_timings(), {"first_step_again": 10.0}) |
| |
| def test_multiple_steps(self) -> None: |
| now = self._now |
| p = build_utils.TimeProfile(now=now) |
| p.start("first", "First step") |
| now.increment(20) |
| p.start("second", "Second step") |
| now.increment(10) |
| p.stop() |
| now.increment(10) |
| p.start("third", "Third step") |
| now.increment(40) |
| p.stop() |
| now.increment(1000) |
| |
| self.assertDictEqual( |
| p.to_json_timings(), |
| { |
| "first": 20, |
| "second": 10, |
| "third": 40, |
| }, |
| ) |
| |
| def test_log(self) -> None: |
| now = self._now |
| |
| log_messages = [] |
| |
| def log(msg: str) -> None: |
| log_messages.append(msg) |
| |
| p = build_utils.TimeProfile(now=now, log=log) |
| p.start("first", "first message") |
| now.increment(10) |
| p.start("second", "second message") |
| now.increment(10) |
| p.start("third", "third message") |
| |
| self.assertListEqual( |
| log_messages, ["first message", "second message", "third message"] |
| ) |
| |
| |
| class BazelBuildInvocationTest(unittest.TestCase): |
| def test_new_instance(self) -> None: |
| with self.assertRaises(ValueError) as cm: |
| build_utils.BazelBuildInvocation(bazel_targets=[], build_args=[]) |
| |
| i = build_utils.BazelBuildInvocation( |
| bazel_targets=["//src:foo"], |
| build_args=[], |
| ) |
| self.assertListEqual(i.bazel_targets, ["//src:foo"]) |
| self.assertListEqual(i.build_args, []) |
| self.assertIsNone(i.gn_label) |
| self.assertIsNone(i.gn_targets_dir) |
| self.assertIsNone(i.bazel_action_timings) |
| |
| i = build_utils.BazelBuildInvocation( |
| bazel_targets=["//first:target", "//second:target"], |
| build_args=["--config=host", "--config=remote_cache_only"], |
| gn_label="//some:bazel_action", |
| gn_targets_dir="obj/some/bazel_action/gn_targets_dir", |
| bazel_action_timings={"foo": 1.0}, |
| ) |
| self.assertListEqual( |
| i.bazel_targets, ["//first:target", "//second:target"] |
| ) |
| self.assertListEqual( |
| i.build_args, ["--config=host", "--config=remote_cache_only"] |
| ) |
| self.assertEqual(i.gn_label, "//some:bazel_action") |
| self.assertEqual( |
| i.gn_targets_dir, "obj/some/bazel_action/gn_targets_dir" |
| ) |
| self.assertDictEqual(i.bazel_action_timings, {"foo": 1}) |
| |
| def test_to_json(self) -> None: |
| i = build_utils.BazelBuildInvocation( |
| bazel_targets=["//src:foo"], |
| build_args=[], |
| ) |
| self.assertDictEqual( |
| i.to_json(), |
| { |
| "bazel_targets": ["//src:foo"], |
| "build_args": [], |
| }, |
| ) |
| |
| i = build_utils.BazelBuildInvocation( |
| bazel_targets=["//first:target", "//second:target"], |
| build_args=["--config=host", "--config=remote_cache_only"], |
| gn_label="//some:bazel_action", |
| gn_targets_dir="obj/some/bazel_action/gn_targets_dir", |
| bazel_action_timings={"bar": 42.0}, |
| ) |
| self.assertDictEqual( |
| i.to_json(), |
| { |
| "bazel_targets": ["//first:target", "//second:target"], |
| "build_args": ["--config=host", "--config=remote_cache_only"], |
| "gn_label": "//some:bazel_action", |
| "gn_targets_dir": "obj/some/bazel_action/gn_targets_dir", |
| "bazel_action_timings": {"bar": 42}, |
| }, |
| ) |
| |
| def test_from_json(self) -> None: |
| with self.assertRaises(ValueError) as cm: |
| i = build_utils.BazelBuildInvocation.from_json([]) |
| self.assertEqual(str(cm.exception), "Input JSON is not an object: []") |
| |
| with self.assertRaises(ValueError) as cm: |
| i = build_utils.BazelBuildInvocation.from_json({}) |
| self.assertEqual( |
| str(cm.exception), "Missing JSON object key 'bazel_targets'" |
| ) |
| |
| with self.assertRaises(ValueError) as cm: |
| i = build_utils.BazelBuildInvocation.from_json( |
| {"bazel_targets": []} |
| ) |
| self.assertEqual( |
| str(cm.exception), "Missing JSON object key 'build_args'" |
| ) |
| |
| with self.assertRaises(ValueError) as cm: |
| i = build_utils.BazelBuildInvocation.from_json( |
| {"bazel_targets": [], "build_args": []} |
| ) |
| self.assertEqual(str(cm.exception), "Empty bazel_targets list") |
| |
| i = build_utils.BazelBuildInvocation.from_json( |
| { |
| "bazel_targets": ["//src:foo"], |
| "build_args": [], |
| } |
| ) |
| self.assertListEqual(i.bazel_targets, ["//src:foo"]) |
| self.assertListEqual(i.build_args, []) |
| self.assertIsNone(i.gn_label) |
| self.assertIsNone(i.gn_targets_dir) |
| |
| i = build_utils.BazelBuildInvocation.from_json( |
| { |
| "bazel_targets": ["//src:foo"], |
| "build_args": ["--config=host", "--config=remote_cache_only"], |
| "gn_label": "//some:bazel_action", |
| "gn_targets_dir": "obj/some/bazel_action/gn_targets_dir", |
| } |
| ) |
| self.assertListEqual(i.bazel_targets, ["//src:foo"]) |
| self.assertListEqual( |
| i.build_args, ["--config=host", "--config=remote_cache_only"] |
| ) |
| self.assertEqual(i.gn_label, "//some:bazel_action") |
| self.assertEqual( |
| i.gn_targets_dir, "obj/some/bazel_action/gn_targets_dir" |
| ) |
| |
| |
| class LastBazelBuildInvocationsTest(unittest.TestCase): |
| def test_from_json(self) -> None: |
| with self.assertRaises(ValueError) as cm: |
| last = build_utils.LastBazelBuildInvocations.new_from_json({}) |
| self.assertEqual( |
| str(cm.exception), |
| "Input is not a JSON array, got <class 'dict'> instead!", |
| ) |
| |
| last = build_utils.LastBazelBuildInvocations.new_from_json([]) |
| self.assertListEqual(last.invocations, []) |
| |
| last = build_utils.LastBazelBuildInvocations.new_from_json( |
| [ |
| { |
| "bazel_targets": ["//first:target"], |
| "build_args": ["--config=host"], |
| "gn_label": "//first:bazel_action", |
| }, |
| { |
| "bazel_targets": ["//second:target"], |
| "build_args": ["--config=fuchsia"], |
| "gn_label": "//second:bazel_action", |
| "gn_targets_dir": "obj/second/gn_targets_dir", |
| }, |
| ] |
| ) |
| |
| self.assertEqual(len(last.invocations), 2) |
| i = last.invocations[0] |
| self.assertListEqual(i.bazel_targets, ["//first:target"]) |
| self.assertListEqual(i.build_args, ["--config=host"]) |
| self.assertEqual(i.gn_label, "//first:bazel_action") |
| self.assertIsNone(i.gn_targets_dir) |
| |
| i = last.invocations[1] |
| self.assertListEqual(i.bazel_targets, ["//second:target"]) |
| self.assertListEqual(i.build_args, ["--config=fuchsia"]) |
| self.assertEqual(i.gn_label, "//second:bazel_action") |
| self.assertEqual(i.gn_targets_dir, "obj/second/gn_targets_dir") |
| |
| def test_to_json(self) -> None: |
| last = build_utils.LastBazelBuildInvocations.new_from_json([]) |
| self.assertListEqual(last.invocations, []) |
| |
| last.append( |
| build_utils.BazelBuildInvocation( |
| bazel_targets=["//first:target"], |
| build_args=["--config=host"], |
| gn_label="//first:bazel_action", |
| ) |
| ) |
| |
| last.append( |
| build_utils.BazelBuildInvocation( |
| bazel_targets=["//second:target"], |
| build_args=["--config=fuchsia"], |
| gn_label="//second:bazel_action", |
| gn_targets_dir="obj/second/gn_targets_dir", |
| ) |
| ) |
| |
| last.to_json() |
| |
| self.assertListEqual( |
| last.to_json(), |
| [ |
| { |
| "bazel_targets": ["//first:target"], |
| "build_args": ["--config=host"], |
| "gn_label": "//first:bazel_action", |
| }, |
| { |
| "bazel_targets": ["//second:target"], |
| "build_args": ["--config=fuchsia"], |
| "gn_label": "//second:bazel_action", |
| "gn_targets_dir": "obj/second/gn_targets_dir", |
| }, |
| ], |
| ) |
| |
| def test_append_to_build_dir(self) -> None: |
| with tempfile.TemporaryDirectory() as build_dir: |
| # Create initial empty list file. |
| file_path = ( |
| build_utils.LastBazelBuildInvocations.get_build_file_path( |
| build_dir |
| ) |
| ) |
| file_path.write_text("[]") |
| |
| build_utils.LastBazelBuildInvocations.append_to_build_dir( |
| build_dir, |
| build_utils.BazelBuildInvocation( |
| bazel_targets=["//first:target"], |
| build_args=["--config=host"], |
| gn_label="//first:bazel_action", |
| ), |
| ) |
| |
| build_utils.LastBazelBuildInvocations.append_to_build_dir( |
| build_dir, |
| build_utils.BazelBuildInvocation( |
| bazel_targets=["//second:target"], |
| build_args=["--config=fuchsia"], |
| gn_label="//second:bazel_action", |
| gn_targets_dir="obj/second/gn_targets_dir", |
| ), |
| ) |
| |
| last_invocations = ( |
| build_utils.LastBazelBuildInvocations.new_from_build(build_dir) |
| ) |
| self.assertListEqual( |
| last_invocations.to_json(), |
| [ |
| { |
| "bazel_targets": ["//first:target"], |
| "build_args": ["--config=host"], |
| "gn_label": "//first:bazel_action", |
| }, |
| { |
| "bazel_targets": ["//second:target"], |
| "build_args": ["--config=fuchsia"], |
| "gn_label": "//second:bazel_action", |
| "gn_targets_dir": "obj/second/gn_targets_dir", |
| }, |
| ], |
| ) |
| |
| |
| class MockBazelLauncher(BazelLauncher): |
| """A BazelLauncher sub-class used to mock subprocess invocation. |
| |
| The class manages a FIFO of CommandResult values that is |
| filled by calling push_result(), and which is consumed when |
| run_command() is called. |
| """ |
| |
| def __init__(self) -> None: |
| """Create instance.""" |
| |
| def log(msg: str) -> None: |
| self.logs.append(msg) |
| |
| def log_error(msg: str) -> None: |
| self.errors.append(msg) |
| |
| super().__init__("/path/to/bazel", log=log, log_err=log_error) |
| self.commands: list[list[str]] = [] |
| self.result_queue: list[CommandResult] = [] |
| self.logs: list[str] = [] |
| self.errors: list[str] = [] |
| |
| def push_result( |
| self, returncode: int = 0, stdout: str = "", stderr: str = "" |
| ) -> None: |
| """Add one result value to the FIFO. |
| |
| Args: |
| returncode: Optional process return code. default to 0. |
| stdout: Optional process stdout, as a string, default to empty. |
| stderr: Optional process stderr, as a string, default to empty. |
| """ |
| self.result_queue.append(CommandResult(returncode, stdout, stderr)) |
| |
| def run_command_internal( |
| self, |
| cmd_args: list[str], |
| print_stdout: bool = False, |
| print_stderr: bool = False, |
| ) -> CommandResult: |
| """Simulate command invocation by popping one value from the FIFO. |
| |
| Args: |
| cmd_args: Command arguments, these are simply saved into |
| self.commands for later inspection. |
| print_stderr: Optional flag, set to True to not capture stdout. |
| print_stderr: Optional flag, set to True to not capture stderr. |
| Returns: |
| The CommandResult at the start of the FIFO. |
| Raises: |
| AssertionError if there are no results in the FIFO. |
| """ |
| self.commands.append(cmd_args[:]) |
| assert self.result_queue |
| result = self.result_queue[0] |
| self.result_queue = self.result_queue[1:] |
| return result |
| |
| |
| class BazelQueryCacheTest(unittest.TestCase): |
| def setUp(self) -> None: |
| self._td = tempfile.TemporaryDirectory() |
| self._root = Path(self._td.name) |
| |
| def tearDown(self) -> None: |
| self._td.cleanup() |
| |
| def test_cache(self) -> None: |
| launcher = MockBazelLauncher() |
| cache_dir = self._root / "cache" |
| cache = BazelQueryCache(cache_dir) |
| |
| launcher.push_result(stdout="some\nresult\nlines") |
| result = cache.get_query_output("query", ["deps(//src:foo)"], launcher) |
| |
| self.assertListEqual(result, ["some", "result", "lines"]) |
| self.assertTrue(cache_dir.is_dir()) |
| key, args = cache.compute_cache_key_and_args( |
| "query", ["deps(//src:foo)"] |
| ) |
| key_file = cache_dir / f"{key}.json" |
| self.assertTrue(key_file.exists()) |
| with key_file.open() as f: |
| value = json.load(f) |
| self.assertDictEqual( |
| value, |
| { |
| "key_args": ["query", "deps(//src:foo)"], |
| "output_lines": ["some", "result", "lines"], |
| }, |
| ) |
| |
| # A second call with the same query and arguments should retrieve |
| # the result from the cache, and not run any command. To detect this |
| # push a new launcher result, which will not be popped as no |
| # command will be run by the cache. |
| launcher.push_result(stdout="some\nother\nresult\nlines") |
| result2 = cache.get_query_output("query", ["deps(//src:foo)"], launcher) |
| self.assertListEqual(result2, result) |
| |
| # By changing the query type, this will pop the last pushed |
| # result value, while creating a new cache entry. |
| result3 = cache.get_query_output( |
| "cquery", ["deps(//src:foo)"], launcher |
| ) |
| self.assertListEqual(result3, ["some", "other", "result", "lines"]) |
| |
| key3, args3 = cache.compute_cache_key_and_args( |
| "cquery", ["deps(//src:foo)"] |
| ) |
| self.assertNotEqual(key3, key) |
| key3_file = cache_dir / f"{key3}.json" |
| self.assertTrue(key3_file.exists()) |
| with key3_file.open() as f: |
| value3 = json.load(f) |
| self.assertDictEqual( |
| value3, |
| { |
| "key_args": ["cquery", "deps(//src:foo)"], |
| "output_lines": ["some", "other", "result", "lines"], |
| }, |
| ) |
| |
| def test_cache_with_starlark_file(self) -> None: |
| launcher = MockBazelLauncher() |
| cache_dir = self._root / "cache" |
| cache = BazelQueryCache(cache_dir) |
| |
| # Create starlark file. Exact content does not matter as real Bazel |
| # queries are not run by this test. |
| starlark_file = self._root / "query.starlark" |
| starlark_file.write_text("1") |
| |
| query_cmd = "query" |
| query_args = [ |
| "deps(//src:foo)", |
| "--output=starlark", |
| "--starlark:file", |
| str(starlark_file), |
| ] |
| |
| # First invocation creates cache entry. |
| launcher.push_result(stdout="some\nresult\nlines") |
| result = cache.get_query_output(query_cmd, query_args, launcher) |
| self.assertListEqual(result, ["some", "result", "lines"]) |
| self.assertTrue(cache_dir.is_dir()) |
| |
| key, args = cache.compute_cache_key_and_args(query_cmd, query_args) |
| key_file = cache_dir / f"{key}.json" |
| self.assertTrue(key_file.exists()) |
| with key_file.open() as f: |
| value = json.load(f) |
| self.assertDictEqual( |
| value, |
| { |
| "key_args": [query_cmd] + query_args, |
| "output_lines": ["some", "result", "lines"], |
| }, |
| ) |
| |
| # Second invocation returns the cache value directly without |
| # invoking anything. |
| launcher.push_result(stdout="other\nresult\nlines") |
| result2 = cache.get_query_output(query_cmd, query_args, launcher) |
| self.assertEqual(result2, result) |
| |
| key, args = cache.compute_cache_key_and_args(query_cmd, query_args) |
| |
| # Now modifying the contenf ot the input file should change |
| # the key value and force a launcher invocation. The arguments |
| # will be the same though. |
| starlark_file.write_text("2") |
| result3 = cache.get_query_output(query_cmd, query_args, launcher) |
| self.assertEqual(result3, ["other", "result", "lines"]) |
| |
| key2, args2 = cache.compute_cache_key_and_args(query_cmd, query_args) |
| self.assertNotEqual(key2, key) |
| self.assertEqual(args, args2) |
| |
| # Now change the query args to use --starlark:file=FILE |
| # This will end up creating a new cache key. |
| query_args = [ |
| "deps(//src:foo)", |
| "--output=starlark", |
| f"--starlark:file={starlark_file}", |
| ] |
| key3, args3 = cache.compute_cache_key_and_args(query_cmd, query_args) |
| self.assertNotEqual(key3, key) |
| self.assertNotEqual(key3, key2) |
| |
| # Modifying the file should change the key, but not the args. |
| starlark_file.write_text("3") |
| key4, args4 = cache.compute_cache_key_and_args(query_cmd, query_args) |
| self.assertNotEqual(key4, key3) |
| self.assertNotEqual(key4, key2) |
| self.assertNotEqual(key4, key) |
| self.assertEqual(args3, args4) |
| |
| |
| if __name__ == "__main__": |
| unittest.main() |
