|
15 | 15 |
|
16 | 16 | """Utils for snapshotting shared libraries.""" |
17 | 17 |
|
| 18 | +from collections.abc import Mapping, Sequence |
18 | 19 | import dataclasses |
19 | 20 | import os |
20 | 21 | import pathlib |
21 | 22 | import re |
22 | 23 | import subprocess |
23 | | -from typing import Final, Sequence |
| 24 | +from typing import Final, Protocol |
24 | 25 |
|
25 | 26 | from absl import logging |
26 | 27 |
|
@@ -69,22 +70,40 @@ def _parse_ld_trace_output(output: str) -> Sequence[SharedLibrary]: |
69 | 70 | return shared_libraries |
70 | 71 |
|
71 | 72 |
|
72 | | -def get_shared_libraries( |
73 | | - binary_path: os.PathLike[str], |
74 | | -) -> Sequence[SharedLibrary]: |
75 | | - """Copies the shared libraries to the shared directory.""" |
76 | | - env = os.environ.copy() |
77 | | - env["LD_TRACE_LOADED_OBJECTS"] = "1" |
78 | | - env["LD_BIND_NOW"] = "1" |
| 73 | +class CommandRunner(Protocol): |
| 74 | + """Runs `command` with environment `env` and returns its stdout.""" |
| 75 | + |
| 76 | + def __call__( |
| 77 | + self, |
| 78 | + command: Sequence[str | os.PathLike[str]], |
| 79 | + env: Mapping[str, str] | None = None, |
| 80 | + ) -> bytes: |
| 81 | + pass |
| 82 | + |
79 | 83 |
|
80 | | - result = subprocess.run( |
81 | | - [_LD_BINARY_PATH, binary_path], |
| 84 | +def run_subprocess( |
| 85 | + command: Sequence[str | os.PathLike[str]], |
| 86 | + env: Mapping[str, str] | None = None, |
| 87 | +) -> bytes: |
| 88 | + return subprocess.run( |
| 89 | + command, |
82 | 90 | capture_output=True, |
83 | 91 | env=env, |
84 | 92 | check=True, |
85 | | - ) |
| 93 | + ).stdout |
| 94 | + |
86 | 95 |
|
87 | | - return _parse_ld_trace_output(result.stdout.decode()) |
| 96 | +def get_shared_libraries( |
| 97 | + binary_path: os.PathLike[str], |
| 98 | + command_runner: CommandRunner = run_subprocess, |
| 99 | +) -> Sequence[SharedLibrary]: |
| 100 | + """Copies the shared libraries to the shared directory.""" |
| 101 | + env = os.environ | { |
| 102 | + "LD_TRACE_LOADED_OBJECTS": "1", |
| 103 | + "LD_BIND_NOW": "1", |
| 104 | + } |
| 105 | + stdout_bytes = command_runner([_LD_BINARY_PATH, binary_path], env=env) |
| 106 | + return _parse_ld_trace_output(stdout_bytes.decode()) |
88 | 107 |
|
89 | 108 |
|
90 | 109 | def copy_shared_libraries( |
|
0 commit comments