build: add directory-oriented make workflows
Host SITL smoke / Deterministic Gazebo command path (push) Has been cancelled

This commit is contained in:
Onur Özkan committed 2026-08-30 07:05:01 +03:00
1 parent 4bc98adbc6
commit 7dc3ccb619
40 files changed
+1367 -278

No files matched your search

+275
View File
@@ -0,0 +1,275 @@
#!/usr/bin/env python3
"""Preserve application argv across GNU Make's command-line parser.
GNU Make removes variable assignments, consumes its first ``--``, and flattens
quoted goal names in ``MAKECMDGOALS``. Linux exposes the original NUL-delimited
argument vector in procfs, which lets the Makefiles keep a conventional
``make component -- application arguments`` interface without rebuilding a
command string in the shell.
"""
from __future__ import annotations
import base64
import errno
import json
import os
from pathlib import Path
import re
import sys
from typing import NoReturn
PAYLOAD_ENV = "KAPTAN_MAKE_ARGV"
ARG_COUNT_ENV = "KAPTAN_MAKE_ARG_COUNT"
_MAKE_FLAG_ENV = ("MAKEFLAGS", "MFLAGS")
class ArgumentRecoveryError(RuntimeError):
"""Raised when the original Make invocation cannot be recovered."""
def encode_arguments(arguments: list[str]) -> str:
"""Encode an argument vector into an ASCII-only environment value."""
document = json.dumps(arguments, ensure_ascii=True, separators=(",", ":"))
return base64.urlsafe_b64encode(document.encode("ascii")).decode("ascii")
def decode_arguments(encoded: str) -> list[str]:
"""Decode an argument vector produced by :func:`encode_arguments`."""
try:
document = base64.b64decode(encoded, altchars=b"-_", validate=True)
arguments = json.loads(document.decode("ascii"))
except (ValueError, UnicodeError, json.JSONDecodeError) as error:
raise ArgumentRecoveryError(
f"invalid application argument payload in {PAYLOAD_ENV}"
) from error
if not isinstance(arguments, list) or not all(
isinstance(argument, str) for argument in arguments
):
raise ArgumentRecoveryError(
f"invalid application argument payload in {PAYLOAD_ENV}"
)
return arguments
def read_process_arguments(pid: int, proc_root: Path = Path("/proc")) -> list[str]:
"""Read one process's exact NUL-delimited argv from procfs."""
path = proc_root / str(pid) / "cmdline"
try:
raw = path.read_bytes()
except OSError as error:
raise ArgumentRecoveryError(
f"cannot read {path}; exact Make argument forwarding requires Linux /proc"
) from error
if not raw:
raise ArgumentRecoveryError(f"cannot recover arguments from empty {path}")
if not raw.endswith(b"\0"):
raise ArgumentRecoveryError(
f"cannot recover unterminated arguments from {path}"
)
return [os.fsdecode(argument) for argument in raw[:-1].split(b"\0")]
def read_parent_pid(pid: int, proc_root: Path = Path("/proc")) -> int:
"""Read a process's parent PID from procfs."""
path = proc_root / str(pid) / "status"
try:
status = path.read_text(encoding="utf-8")
except OSError as error:
raise ArgumentRecoveryError(
f"cannot read {path}; exact Make argument forwarding requires Linux /proc"
) from error
for line in status.splitlines():
if line.startswith("PPid:"):
return int(line.split(":", 1)[1].strip())
raise ArgumentRecoveryError(f"cannot find PPid in {path}")
def find_make_arguments(marker: str, start_pid: int | None = None) -> list[str]:
"""Find the nearest parent Make invocation containing *marker*."""
pid = os.getppid() if start_pid is None else start_pid
visited: set[int] = set()
while pid > 1 and pid not in visited:
visited.add(pid)
arguments = read_process_arguments(pid)
executable = Path(arguments[0]).name if arguments else ""
if marker in arguments and executable.endswith("make"):
return arguments
pid = read_parent_pid(pid)
raise ArgumentRecoveryError(
f"cannot find a parent Make invocation containing {marker!r}; "
"exact argument forwarding requires Linux /proc"
)
def _tail_after_marker(make_arguments: list[str], marker: str) -> list[str]:
try:
marker_index = make_arguments.index(marker)
except ValueError as error:
raise ArgumentRecoveryError(
f"cannot find {marker!r} in the parent Make invocation"
) from error
return make_arguments[marker_index + 1 :]
def arguments_after_marker(make_arguments: list[str], marker: str) -> list[str]:
"""Return arguments following *marker*, consuming one following ``--``."""
arguments = _tail_after_marker(make_arguments, marker)
if arguments[:1] == ["--"]:
return arguments[1:]
return arguments
def select_route(
make_arguments: list[str],
component: str,
default_target: str,
local_targets: set[str],
) -> tuple[str, list[str]]:
"""Select a local Make target and the exact application payload.
``default_target`` applies only when the component has no arguments. An
explicit delimiter or backward-compatible unknown words always select the
component's ``run`` target.
"""
arguments = _tail_after_marker(make_arguments, component)
if "--" in arguments:
separator = arguments.index("--")
before_separator = arguments[:separator]
payload = arguments[separator + 1 :]
if len(before_separator) == 1 and before_separator[0] in local_targets:
return before_separator[0], payload
if before_separator:
raise ArgumentRecoveryError(
f"unexpected words before -- for {component}: "
+ " ".join(repr(word) for word in before_separator)
)
return "run", payload
if not arguments:
return default_target, []
if len(arguments) == 1 and arguments[0] in local_targets:
return arguments[0], []
return "run", arguments
def assignment_names(arguments: list[str]) -> set[str]:
"""Return environment names which GNU Make may treat as assignments."""
names: set[str] = set()
for argument in arguments[1:]:
for operator in (":::=", "::=", ":=", "+=", "?=", "!=", "="):
if operator in argument:
name, _ = argument.split(operator, 1)
if name:
names.add(name)
break
return names
def scrub_make_environment(
environment: dict[str, str], make_arguments: list[str]
) -> dict[str, str]:
"""Remove recursive-Make flags and command-line variable side effects."""
clean = environment.copy()
clean.pop("MAKEOVERRIDES", None)
for name in _MAKE_FLAG_ENV:
value = clean.get(name, "")
separator = re.search(r"(?:^|\s)--(?:\s|$)", value)
flags = value[: separator.start()].strip() if separator else value.strip()
if flags:
clean[name] = flags
else:
clean.pop(name, None)
for name in assignment_names(make_arguments):
clean.pop(name, None)
return clean
def route(
component: str,
make_program: str,
default_target: str,
local_targets: set[str],
) -> NoReturn:
"""Recover argv and exec the component's selected recursive Make target."""
make_arguments = find_make_arguments(component)
target, payload = select_route(
make_arguments, component, default_target, local_targets
)
environment = scrub_make_environment(dict(os.environ), make_arguments)
environment[PAYLOAD_ENV] = encode_arguments(payload)
environment[ARG_COUNT_ENV] = str(len(payload))
os.execvpe(
make_program,
[make_program, "--no-print-directory", "-C", component, target],
environment,
)
def execute(marker: str, command: list[str]) -> NoReturn:
"""Append recovered arguments to *command* and exec it without a shell."""
if not command:
raise ArgumentRecoveryError("exec requires a command prefix")
environment = dict(os.environ)
encoded = environment.pop(PAYLOAD_ENV, None)
environment.pop(ARG_COUNT_ENV, None)
if encoded is None:
make_arguments = find_make_arguments(marker)
arguments = arguments_after_marker(make_arguments, marker)
environment = scrub_make_environment(environment, make_arguments)
else:
arguments = decode_arguments(encoded)
environment = scrub_make_environment(environment, ["make"])
os.execvpe(command[0], [*command, *arguments], environment)
def usage() -> str:
return (
"usage:\n"
" make-argv.py route COMPONENT MAKE_PROGRAM DEFAULT_TARGET [LOCAL_TARGET ...]\n"
" make-argv.py exec MARKER COMMAND [COMMAND_ARG ...]"
)
def main(arguments: list[str] | None = None) -> int:
arguments = sys.argv[1:] if arguments is None else arguments
try:
if arguments[:1] == ["route"] and len(arguments) >= 4:
route(arguments[1], arguments[2], arguments[3], set(arguments[4:]))
elif arguments[:1] == ["exec"] and len(arguments) >= 3:
execute(arguments[1], arguments[2:])
else:
raise ArgumentRecoveryError(usage())
except ArgumentRecoveryError as error:
print(f"make-argv: {error}", file=sys.stderr)
return 2
except OSError as error:
program = error.filename or "command"
detail = error.strerror or str(error)
print(f"make-argv: cannot execute {program}: {detail}", file=sys.stderr)
return 127 if error.errno == errno.ENOENT else 126
return 0
if __name__ == "__main__":
raise SystemExit(main())
+312
View File
@@ -0,0 +1,312 @@
#!/usr/bin/env python3
"""Tests for exact argument forwarding through GNU Make."""
import importlib.util
import io
import json
from pathlib import Path
import shutil
import subprocess
import sys
import tempfile
import unittest
from unittest import mock
SPEC = importlib.util.spec_from_file_location(
"make_argv", Path(__file__).with_name("make-argv.py")
)
assert SPEC is not None and SPEC.loader is not None
MAKE_ARGV = importlib.util.module_from_spec(SPEC)
SPEC.loader.exec_module(MAKE_ARGV)
class ArgumentEncodingTests(unittest.TestCase):
def test_round_trip_preserves_exact_arguments(self):
arguments = [
"--key=one",
"two words",
"",
"--key=one",
"$HOME",
"--",
"--key=two",
]
self.assertEqual(
MAKE_ARGV.decode_arguments(MAKE_ARGV.encode_arguments(arguments)),
arguments,
)
def test_rejects_invalid_payload(self):
with self.assertRaisesRegex(
MAKE_ARGV.ArgumentRecoveryError, "invalid application argument payload"
):
MAKE_ARGV.decode_arguments("not-base64!")
@mock.patch.object(
MAKE_ARGV, "execute", side_effect=FileNotFoundError(2, "missing", "tool")
)
def test_missing_application_has_a_concise_error(self, _execute):
errors = io.StringIO()
with mock.patch("sys.stderr", errors):
status = MAKE_ARGV.main(["exec", "run", "tool"])
self.assertEqual(status, 127)
self.assertEqual(errors.getvalue(), "make-argv: cannot execute tool: missing\n")
class RouteSelectionTests(unittest.TestCase):
def test_delimiter_makes_test_an_application_argument(self):
target, arguments = MAKE_ARGV.select_route(
["make", "firmware", "--", "test"],
"firmware",
"run",
{"test", "clippy"},
)
self.assertEqual(target, "run")
self.assertEqual(arguments, ["test"])
def test_no_delimiter_selects_a_known_local_target(self):
target, arguments = MAKE_ARGV.select_route(
["make", "firmware", "test"],
"firmware",
"run",
{"test", "clippy"},
)
self.assertEqual(target, "test")
self.assertEqual(arguments, [])
def test_configurable_no_argument_default(self):
target, arguments = MAKE_ARGV.select_route(
["make", "docs"], "docs", "check", {"help"}
)
self.assertEqual(target, "check")
self.assertEqual(arguments, [])
def test_explicit_docs_arguments_select_run_not_no_argument_default(self):
target, arguments = MAKE_ARGV.select_route(
["make", "docs", "--", "--version"], "docs", "check", {"help"}
)
self.assertEqual(target, "run")
self.assertEqual(arguments, ["--version"])
def test_unknown_words_without_delimiter_remain_application_arguments(self):
target, arguments = MAKE_ARGV.select_route(
["make", "firmware/tools/sitl", "move", "5", "two words"],
"firmware/tools/sitl",
"run",
{"help"},
)
self.assertEqual(target, "run")
self.assertEqual(arguments, ["move", "5", "two words"])
def test_local_flash_keeps_every_argument_after_delimiter(self):
target, arguments = MAKE_ARGV.select_route(
[
"make",
"firmware/boards/example",
"flash",
"--",
"--device=0483:df11",
"",
"--",
"$HOME",
],
"firmware/boards/example",
"run",
{"flash", "help"},
)
self.assertEqual(target, "flash")
self.assertEqual(arguments, ["--device=0483:df11", "", "--", "$HOME"])
class EnvironmentTests(unittest.TestCase):
def test_scrubs_make_flags_and_command_line_assignments(self):
environment = {
"PATH": "/bin",
"MAKEFLAGS": "-- CARGO=wrong",
"MAKEOVERRIDES": "CARGO=wrong",
"MFLAGS": "--",
"CARGO": "wrong",
"OTHER": "keep",
}
clean = MAKE_ARGV.scrub_make_environment(
environment,
["make", "firmware", "--", "CARGO=wrong", "--key=value"],
)
self.assertEqual(clean, {"PATH": "/bin", "OTHER": "keep"})
def test_preserves_make_execution_flags_without_overrides(self):
clean = MAKE_ARGV.scrub_make_environment(
{
"MAKEFLAGS": "n --no-print-directory -- --key=value",
"MFLAGS": "-n -- --key=value",
"MAKEOVERRIDES": "--key=value",
},
["make", "firmware", "--", "--key=value"],
)
self.assertEqual(
clean,
{"MAKEFLAGS": "n --no-print-directory", "MFLAGS": "-n"},
)
@mock.patch.object(MAKE_ARGV.os, "execvpe")
@mock.patch.object(MAKE_ARGV, "find_make_arguments")
def test_route_execs_recursive_make_with_encoded_payload(
self, find_make_arguments, execvpe
):
payload = ["--key=one", "two words", "", "--", "--key=one"]
find_make_arguments.return_value = [
"make",
"firmware/tools/sitl",
"--",
*payload,
]
with mock.patch.dict(
MAKE_ARGV.os.environ,
{"PATH": "/bin", "MAKEFLAGS": "n -- --key=one"},
clear=True,
):
MAKE_ARGV.route("firmware/tools/sitl", "make", "run", {"help"})
program, arguments, environment = execvpe.call_args.args
self.assertEqual(program, "make")
self.assertEqual(
arguments,
[
"make",
"--no-print-directory",
"-C",
"firmware/tools/sitl",
"run",
],
)
self.assertEqual(
MAKE_ARGV.decode_arguments(environment[MAKE_ARGV.PAYLOAD_ENV]), payload
)
self.assertEqual(environment[MAKE_ARGV.ARG_COUNT_ENV], str(len(payload)))
self.assertEqual(environment["MAKEFLAGS"], "n")
@mock.patch.object(MAKE_ARGV.os, "execvpe")
def test_exec_uses_routed_payload_without_shell_joining(self, execvpe):
payload = ["--key=one", "two words", "", "$HOME", "--", "--key=one"]
environment = {
MAKE_ARGV.PAYLOAD_ENV: MAKE_ARGV.encode_arguments(payload),
MAKE_ARGV.ARG_COUNT_ENV: str(len(payload)),
"PATH": "/bin",
"MAKEFLAGS": "-- --key=one",
}
with mock.patch.dict(MAKE_ARGV.os.environ, environment, clear=True):
MAKE_ARGV.execute("run", ["example-app", "fixed argument"])
program, arguments, child_environment = execvpe.call_args.args
self.assertEqual(program, "example-app")
self.assertEqual(arguments, ["example-app", "fixed argument", *payload])
self.assertEqual(child_environment, {"PATH": "/bin"})
@mock.patch.object(MAKE_ARGV.os, "execvpe")
@mock.patch.object(MAKE_ARGV, "find_make_arguments")
def test_direct_local_make_exec_recovers_arguments(
self, find_make_arguments, execvpe
):
payload = ["--key=value", "two words", "", "--", "$HOME"]
find_make_arguments.return_value = ["make", "run", "--", *payload]
with mock.patch.dict(MAKE_ARGV.os.environ, {"PATH": "/bin"}, clear=True):
MAKE_ARGV.execute("run", ["example-app"])
self.assertEqual(execvpe.call_args.args[1], ["example-app", *payload])
class ProcfsTests(unittest.TestCase):
def test_procfs_preserves_a_final_empty_argument(self):
with tempfile.TemporaryDirectory() as directory:
process_directory = Path(directory, "123")
process_directory.mkdir()
Path(process_directory, "cmdline").write_bytes(b"make\0run\0\0")
self.assertEqual(
MAKE_ARGV.read_process_arguments(123, Path(directory)),
["make", "run", ""],
)
def test_procfs_failure_is_actionable(self):
with tempfile.TemporaryDirectory() as directory:
with self.assertRaisesRegex(
MAKE_ARGV.ArgumentRecoveryError, "requires Linux /proc"
):
MAKE_ARGV.read_process_arguments(123, Path(directory))
@unittest.skipUnless(Path("/proc/self/cmdline").is_file(), "requires Linux procfs")
class MakeIntegrationTests(unittest.TestCase):
def test_recursive_make_preserves_exact_application_argv(self):
make = shutil.which("make")
if make is None:
self.skipTest("GNU Make is unavailable")
helper = Path(__file__).with_name("make-argv.py").resolve()
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
component = root / "component"
component.mkdir()
output = root / "arguments.json"
application = root / "record-arguments.py"
application.write_text(
"import json, pathlib, sys\n"
"pathlib.Path(sys.argv[1]).write_text(json.dumps(sys.argv[2:]))\n",
encoding="utf-8",
)
(root / "Makefile").write_text(
".PHONY: component\n"
"component:\n"
f"\t+@exec {sys.executable} {helper} route component make run help\n"
".DEFAULT:\n"
"\t@:\n",
encoding="utf-8",
)
(component / "Makefile").write_text(
".PHONY: run\n"
"run:\n"
f"\t@exec {sys.executable} {helper} exec run "
f"{sys.executable} {application} {output}\n"
".DEFAULT:\n"
"\t@:\n",
encoding="utf-8",
)
payload = [
"--key=one",
"two words",
"",
"--key=one",
"$HOME",
"--",
"--key=two",
]
subprocess.run(
[make, "component", "--", *payload],
cwd=root,
check=True,
capture_output=True,
text=True,
)
self.assertEqual(json.loads(output.read_text(encoding="utf-8")), payload)
if __name__ == "__main__":
unittest.main()