diff --git a/infra/base-images/base-builder/indexer/manifest_constants.py b/infra/base-images/base-builder/indexer/manifest_constants.py index 9c99e5297df4..c8390c8769da 100644 --- a/infra/base-images/base-builder/indexer/manifest_constants.py +++ b/infra/base-images/base-builder/indexer/manifest_constants.py @@ -32,6 +32,10 @@ OBJ_DIR = Path("obj") # Directory for indexer data. INDEX_DIR = Path("idx") +# Relative source file root in the index. +INDEX_RELATIVE_SOURCES = INDEX_DIR / "relative" +# Absolute source file root in the index. +INDEX_ABSOLUTE_SOURCES = INDEX_DIR / "absolute" # The index database filename. INDEX_DB = Path("db.sqlite") # Library directory, where shared libraries are copied - inside obj. diff --git a/infra/base-images/base-builder/indexer/utils.py b/infra/base-images/base-builder/indexer/utils.py index 131ddb0af61b..7374af1528be 100644 --- a/infra/base-images/base-builder/indexer/utils.py +++ b/infra/base-images/base-builder/indexer/utils.py @@ -15,12 +15,13 @@ """Utils for snapshotting shared libraries.""" +from collections.abc import Mapping, Sequence import dataclasses import os import pathlib import re import subprocess -from typing import Final, Sequence +from typing import Final, Protocol from absl import logging @@ -69,22 +70,40 @@ def _parse_ld_trace_output(output: str) -> Sequence[SharedLibrary]: return shared_libraries -def get_shared_libraries( - binary_path: os.PathLike[str], -) -> Sequence[SharedLibrary]: - """Copies the shared libraries to the shared directory.""" - env = os.environ.copy() - env["LD_TRACE_LOADED_OBJECTS"] = "1" - env["LD_BIND_NOW"] = "1" +class CommandRunner(Protocol): + """Runs `command` with environment `env` and returns its stdout.""" + + def __call__( + self, + command: Sequence[str | os.PathLike[str]], + env: Mapping[str, str] | None = None, + ) -> bytes: + pass + - result = subprocess.run( - [_LD_BINARY_PATH, binary_path], +def run_subprocess( + command: Sequence[str | os.PathLike[str]], + env: Mapping[str, str] | None = None, +) -> bytes: + return subprocess.run( + command, capture_output=True, env=env, check=True, - ) + ).stdout + - return _parse_ld_trace_output(result.stdout.decode()) +def get_shared_libraries( + binary_path: os.PathLike[str], + command_runner: CommandRunner = run_subprocess, +) -> Sequence[SharedLibrary]: + """Copies the shared libraries to the shared directory.""" + env = os.environ | { + "LD_TRACE_LOADED_OBJECTS": "1", + "LD_BIND_NOW": "1", + } + stdout_bytes = command_runner([_LD_BINARY_PATH, binary_path], env=env) + return _parse_ld_trace_output(stdout_bytes.decode()) def copy_shared_libraries(