29762829348

Committed 20 Jul 2026 05:13PM UTC coverage: 80.089% (+1.5%) from 78.598%

Build # 29762829348

Build Type

Pull #201

github

Committed by

wyli

Commit Message

feat: improve shell command plan readability

Render replay steps as labeled, multiline subshell blocks while preserving exact argv, cwd, and environment semantics.

Co-authored-by: Codex <noreply@openai.com>

Pull Request Pull Request #201: feat: add JSON and shell dry-run plans for build-container

Coverage Stats

299 of 326 new or added lines in 7 files covered. (91.72%)

4 existing lines in 4 files now uncovered.

3411 of 4259 relevant lines covered (80.09%)

0.8 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

93.9

/src/holoscan_cli/command_plan.py

# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0
#
# 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.

"""Invocation-scoped structured command plans.

The first public slice records process steps for ``build-container``.  The
recorder is deliberately private and active only for ``--dryrun --json`` or
``--dryrun --shell`` so the normal execution path and human dry-run output stay
unchanged.
"""

from __future__ import annotations

import argparse
import json
import os
import re
import shlex
from contextlib import contextmanager
from contextvars import ContextVar
from dataclasses import dataclass, field
from pathlib import Path
from typing import Iterator, Mapping, Optional, Sequence


class CommandPlanError(RuntimeError):
    """Raised when an invocation cannot produce a complete v1 plan."""


_ACTIVE_RECORDER: ContextVar[Optional["PlanRecorder"]] = ContextVar(
    "holoscan_cli_command_plan", default=None
)

_ENV_NAME = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")
_SENSITIVE_EXACT_NAMES = {
    "API_KEY",
    "AWS_ACCESS_KEY_ID",
    "AWS_SECRET_ACCESS_KEY",
    "NGC_API_KEY",
    "NGC_CLI_API_KEY",
    "PASSWORD",
    "SECRET",
    "TOKEN",
}
_SENSITIVE_SUFFIXES = ("_API_KEY", "_PASSWORD", "_SECRET", "_TOKEN")
_DOCKER_VALUE_OPTIONS = {"-e", "--env", "--build-arg"}


def add_plan_output_arguments(parser: argparse.ArgumentParser) -> None:
    """Add the command-plan output formats to one audited action parser."""

    output = parser.add_mutually_exclusive_group()
    output.add_argument(
        "--json",
        dest="plan_format",
        action="store_const",
        const="json",
        help="With --dryrun, print a machine-readable command plan",
    )
    output.add_argument(
        "--shell",
        dest="plan_format",
        action="store_const",
        const="shell",
        help="With --dryrun, print a copyable Bash command plan",
    )
    parser.set_defaults(plan_format=None)


def get_active_recorder() -> Optional["PlanRecorder"]:
    """Return the recorder for the current invocation, if any."""

    return _ACTIVE_RECORDER.get()


def command_plan_active() -> bool:
    """Return whether structured planning is active in this context."""

    return get_active_recorder() is not None


def record_probe_fallback(message: str) -> None:
    """Attach a warning when the real resolver uses a documented fallback."""

    recorder = get_active_recorder()
    if recorder is not None:
        recorder.add_warning("probe_fallback_used", message)


def _normalize_env_name(name: str) -> str:
    return re.sub(r"[^A-Za-z0-9]+", "_", name).strip("_").upper()


def _is_sensitive_env_name(name: str) -> bool:
    normalized = _normalize_env_name(name)
    return normalized in _SENSITIVE_EXACT_NAMES or normalized.endswith(_SENSITIVE_SUFFIXES)


def _redact_assignment(assignment: str) -> tuple[str, bool, Optional[str]]:
    """Return public assignment, redaction state, and named-env requirement."""

    if "=" not in assignment:
        name = assignment
        required = name if _ENV_NAME.fullmatch(name) else None
        return assignment, False, required

    name, _value = assignment.split("=", 1)
    if not _is_sensitive_env_name(name):
        return assignment, False, None
    return f"{name}=<redacted>", True, None


def _public_argv(argv: Sequence[str]) -> tuple[list[str], bool, list[str]]:
    """Redact recognized literal credentials in Docker option assignments."""

    public = [str(token) for token in argv]
    redacted = False
    required: set[str] = set()
    index = 0

    while index < len(public):
        token = public[index]
        if token in _DOCKER_VALUE_OPTIONS and index + 1 < len(public):
            replacement, changed, required_name = _redact_assignment(public[index + 1])
            public[index + 1] = replacement
            redacted = redacted or changed
            if required_name:
                required.add(required_name)
            index += 2
            continue

        for prefix in ("--env=", "-e=", "--build-arg="):
            if token.startswith(prefix):
                replacement, changed, required_name = _redact_assignment(token[len(prefix) :])
                public[index] = f"{prefix}{replacement}"
                redacted = redacted or changed
                if required_name:
                    required.add(required_name)
                break
        index += 1

    return public, redacted, sorted(required)


def _environment_delta(
    baseline: Mapping[str, str],
    effective: Mapping[str, str],
    explicit_set: Optional[Mapping[str, str]] = None,
) -> tuple[dict[str, str], list[str], bool]:
    changed: dict[str, str] = {}
    redacted = False
    explicit_names = set(explicit_set or {})
    for name in sorted(effective):
        value = str(effective[name])
        if baseline.get(name) == value and name not in explicit_names:
            continue
        if _is_sensitive_env_name(name):
            changed[name] = "<redacted>"
            redacted = True
        else:
            changed[name] = value
    unset = sorted(name for name in baseline if name not in effective)
    return changed, unset, redacted


def _shell_argv_groups(argv: Sequence[str]) -> list[list[str]]:
    """Group argv tokens into readable lines without changing shell semantics."""

    tokens = [str(token) for token in argv]
    if not tokens:
        return []

    head_length = 2 if len(tokens) > 1 and not tokens[1].startswith("-") else 1
    return [tokens[:head_length], *[[token] for token in tokens[head_length:]]]


def _shell_group_lines(
    groups: Sequence[Sequence[str]], *, first_indent: str, continuation_indent: str
) -> list[str]:
    """Quote grouped argv and add Bash line continuations."""

    lines = []
    for index, group in enumerate(groups):
        indent = first_indent if index == 0 else continuation_indent
        suffix = " \\" if index + 1 < len(groups) else ""
        lines.append(f"{indent}{shlex.join([str(token) for token in group])}{suffix}")
    return lines


def _process_shell(
    argv: Sequence[str],
    cwd: str,
    environment_set: Mapping[str, str],
    environment_unset: Sequence[str],
) -> str:
    command_groups = _shell_argv_groups(argv)
    lines = ["(", f"  {shlex.join(['cd', '--', cwd])} && \\"]
    if environment_set or environment_unset:
        lines.append("    env \\")
        env_groups = [["-u", name] for name in environment_unset]
        env_groups.extend([[f"{name}={value}"] for name, value in environment_set.items()])
        for group in env_groups:
            lines.append(f"      {shlex.join(group)} \\")
        lines.extend(
            _shell_group_lines(
                command_groups,
                first_indent="      ",
                continuation_indent="        ",
            )
        )
    else:
        lines.extend(
            _shell_group_lines(
                command_groups,
                first_indent="    ",
                continuation_indent="      ",
            )
        )
    lines.append(")")
    return "\n".join(lines)


@dataclass
class ProcessStep:
    """A process invocation plus its private parity data."""

    id: str
    role: str
    argv: list[str]
    private_argv: list[str] = field(repr=False)
    shell: str
    cwd: str
    environment: dict
    check: bool
    privilege: str
    redacted: bool
    destructive: bool = False

    def public_dict(self) -> dict:
        return {
            "id": self.id,
            "kind": "process",
            "role": self.role,
            "argv": self.argv,
            "shell": self.shell,
            "cwd": self.cwd,
            "environment": self.environment,
            "check": self.check,
            "privilege": self.privilege,
            "redacted": self.redacted,
            "destructive": self.destructive,
        }


class PlanRecorder:
    """Record a complete, ordered command plan for one CLI invocation."""

    def __init__(self) -> None:
        self._environment = dict(os.environ)
        self.steps: list[ProcessStep] = []
        self.limitations: list[dict] = []
        self.warnings: list[dict] = []

    @contextmanager
    def activate(self) -> Iterator["PlanRecorder"]:
        if get_active_recorder() is not None:
            raise CommandPlanError("nested command-plan recorders are not supported")
        token = _ACTIVE_RECORDER.set(self)
        try:
            yield self
        finally:
            _ACTIVE_RECORDER.reset(token)

    def add_warning(self, code: str, message: str, step_id: Optional[str] = None) -> None:
        warning = {"code": code, "message": message}
        if step_id is not None:
            warning["step_id"] = step_id
        if warning not in self.warnings:
            self.warnings.append(warning)

    def record_process(
        self,
        argv: Sequence[str],
        *,
        role: str,
        cwd: Optional[os.PathLike[str] | str] = None,
        env: Optional[Mapping[str, str]] = None,
        explicit_env: Optional[Mapping[str, str]] = None,
        check: bool,
        privilege: str = "user",
        destructive: bool = False,
    ) -> ProcessStep:
        if role not in {"probe", "action", "cleanup"}:
            raise CommandPlanError(f"unsupported process role: {role}")

        private_argv = [str(token) for token in argv]
        if not private_argv:
            raise CommandPlanError("process argv must contain at least one token")
        public_argv, argv_redacted, required = _public_argv(private_argv)
        if env is not None:
            if not explicit_env:
                raise CommandPlanError(
                    "replacement subprocess environments are not supported in v1 plans"
                )
            expected_env = dict(self._environment)
            expected_env.update({str(name): str(value) for name, value in explicit_env.items()})
            if dict(env) != expected_env:
                raise CommandPlanError(
                    "replacement subprocess environments are not supported in v1 plans"
                )
        effective_env = os.environ if env is None else env
        missing_required = [name for name in required if name not in effective_env]
        if missing_required:
            names = ", ".join(missing_required)
            raise CommandPlanError(
                f"Docker environment reference is unset during planning: {names}"
            )
        environment_set, environment_unset, env_redacted = _environment_delta(
            self._environment, effective_env, explicit_env
        )
        external_required = [name for name in required if name not in environment_set]
        resolved_cwd = str(Path.cwd() if cwd is None else Path(cwd).resolve())
        step_id = f"step-{len(self.steps) + 1:03d}"
        redacted = argv_redacted or env_redacted
        environment = {
            "inherit": True,
            "set": environment_set,
            "unset": environment_unset,
            "required": external_required,
        }
        step = ProcessStep(
            id=step_id,
            role=role,
            argv=public_argv,
            private_argv=private_argv,
            shell=_process_shell(public_argv, resolved_cwd, environment_set, environment_unset),
            cwd=resolved_cwd,
            environment=environment,
            check=check,
            privilege=privilege,
            redacted=redacted,
            destructive=destructive,
        )
        self.steps.append(step)
        if redacted:
            self.add_warning(
                "redacted_value",
                "A credential-like literal was redacted from the public command plan.",
                step_id,
            )
        return step

    def _replay(self) -> dict:
        self._ensure_complete()
        replay_steps = [step for step in self.steps if step.role != "probe"]
        required = sorted(
            {name for step in replay_steps for name in step.environment.get("required", [])}
        )

        unavailable_reason = None
        if any(step.redacted for step in replay_steps):
            unavailable_reason = "redacted_literal"
        elif any(not step.check for step in replay_steps):
            unavailable_reason = "nonfatal_exit_handling"

        if unavailable_reason is not None:
            return {
                "format": "bash",
                "script": None,
                "required_environment": required,
                "unavailable_reason": unavailable_reason,
            }

        lines = ["#!/usr/bin/env bash", "set -e", ""]
        for name in required:
            lines.append(f': "${{{name}?Set {name} before running this script}}"')
        if required:
            lines.append("")
        for index, step in enumerate(replay_steps):
            summary = shlex.join(step.argv[:2]).replace("\r", "\\r").replace("\n", "\\n")
            lines.append(f"# {step.id} ({step.role}): {summary}")
            lines.append(step.shell)
            if index + 1 < len(replay_steps):
                lines.append("")
        return {
            "format": "bash",
            "script": "\n".join(lines) + "\n",
            "required_environment": required,
            "unavailable_reason": None,
        }

    def _ensure_complete(self) -> None:
        if not any(step.role in {"action", "cleanup"} for step in self.steps):
            raise CommandPlanError("the resolved branch contains no action steps")

    def payload(self) -> dict:
        replay = self._replay()
        warnings = list(self.warnings)
        if replay["script"] is None:
            warnings.append(
                {
                    "code": "shell_replay_unavailable",
                    "message": f"Bash replay unavailable: {replay['unavailable_reason']}",
                }
            )
        return {
            "schema_version": 1,
            "status": "dryrun",
            "scope": "current_cli_process",
            "host_resolved": True,
            "complete": True,
            "steps": [step.public_dict() for step in self.steps],
            "replay": replay,
            "limitations": list(self.limitations),
            "warnings": warnings,
        }

    def json_text(self) -> str:
        """Render the complete JSON artifact without writing stdout."""

        return json.dumps(self.payload(), indent=2)

    def shell_text(self) -> str:
        """Render Bash or fail before anything is written to stdout."""

        replay = self._replay()
        if replay["script"] is None:
            raise CommandPlanError(f"Bash replay unavailable: {replay['unavailable_reason']}")
        return str(replay["script"])

1	# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
2	# SPDX-License-Identifier: Apache-2.0
3	#
4	# Licensed under the Apache License, Version 2.0 (the "License");
5	# you may not use this file except in compliance with the License.
6	# You may obtain a copy of the License at
7	#
8	# http://www.apache.org/licenses/LICENSE-2.0
9	#
10	# Unless required by applicable law or agreed to in writing, software
11	# distributed under the License is distributed on an "AS IS" BASIS,
12	# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13	# See the License for the specific language governing permissions and
14	# limitations under the License.
15
16	"""Invocation-scoped structured command plans.
17
18	The first public slice records process steps for ``build-container``. The
19	recorder is deliberately private and active only for ``--dryrun --json`` or
20	``--dryrun --shell`` so the normal execution path and human dry-run output stay
21	unchanged.
22	"""
23
24	from __future__ import annotations	1✔
25
26	import argparse	1✔
27	import json	1✔
28	import os	1✔
29	import re	1✔
30	import shlex	1✔
31	from contextlib import contextmanager	1✔
32	from contextvars import ContextVar	1✔
33	from dataclasses import dataclass, field	1✔
34	from pathlib import Path	1✔
35	from typing import Iterator, Mapping, Optional, Sequence	1✔
36
37
38	class CommandPlanError(RuntimeError):	1✔
39	"""Raised when an invocation cannot produce a complete v1 plan."""
40
41
42	_ACTIVE_RECORDER: ContextVar[Optional["PlanRecorder"]] = ContextVar(	1✔
43	"holoscan_cli_command_plan", default=None
44	)
45
46	_ENV_NAME = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")	1✔
47	_SENSITIVE_EXACT_NAMES = {	1✔
48	"API_KEY",
49	"AWS_ACCESS_KEY_ID",
50	"AWS_SECRET_ACCESS_KEY",
51	"NGC_API_KEY",
52	"NGC_CLI_API_KEY",
53	"PASSWORD",
54	"SECRET",
55	"TOKEN",
56	}
57	_SENSITIVE_SUFFIXES = ("_API_KEY", "_PASSWORD", "_SECRET", "_TOKEN")	1✔
58	_DOCKER_VALUE_OPTIONS = {"-e", "--env", "--build-arg"}	1✔
59
60
61	def add_plan_output_arguments(parser: argparse.ArgumentParser) -> None:	1✔
62	"""Add the command-plan output formats to one audited action parser."""
63
64	output = parser.add_mutually_exclusive_group()	1✔
65	output.add_argument(	1✔
66	"--json",
67	dest="plan_format",
68	action="store_const",
69	const="json",
70	help="With --dryrun, print a machine-readable command plan",
71	)
72	output.add_argument(	1✔
73	"--shell",
74	dest="plan_format",
75	action="store_const",
76	const="shell",
77	help="With --dryrun, print a copyable Bash command plan",
78	)
79	parser.set_defaults(plan_format=None)	1✔
80
81
82	def get_active_recorder() -> Optional["PlanRecorder"]:	1✔
83	"""Return the recorder for the current invocation, if any."""
84
85	return _ACTIVE_RECORDER.get()	1✔
86
87
88	def command_plan_active() -> bool:	1✔
89	"""Return whether structured planning is active in this context."""
90
91	return get_active_recorder() is not None	1✔
92
93
94	def record_probe_fallback(message: str) -> None:	1✔
95	"""Attach a warning when the real resolver uses a documented fallback."""
96
97	recorder = get_active_recorder()	1✔
98	if recorder is not None:	1✔
99	recorder.add_warning("probe_fallback_used", message)	1✔
100
101
102	def _normalize_env_name(name: str) -> str:	1✔
103	return re.sub(r"[^A-Za-z0-9]+", "_", name).strip("_").upper()	1✔
104
105
106	def _is_sensitive_env_name(name: str) -> bool:	1✔
107	normalized = _normalize_env_name(name)	1✔
108	return normalized in _SENSITIVE_EXACT_NAMES or normalized.endswith(_SENSITIVE_SUFFIXES)	1✔
109
110
111	def _redact_assignment(assignment: str) -> tuple[str, bool, Optional[str]]:	1✔
112	"""Return public assignment, redaction state, and named-env requirement."""
113
114	if "=" not in assignment:	1✔
115	name = assignment	1✔
116	required = name if _ENV_NAME.fullmatch(name) else None	1✔
117	return assignment, False, required	1✔
118
119	name, _value = assignment.split("=", 1)	1✔
120	if not _is_sensitive_env_name(name):	1✔
121	return assignment, False, None	1✔
122	return f"{name}=<redacted>", True, None	1✔
123
124
125	def _public_argv(argv: Sequence[str]) -> tuple[list[str], bool, list[str]]:	1✔
126	"""Redact recognized literal credentials in Docker option assignments."""
127
128	public = [str(token) for token in argv]	1✔
129	redacted = False	1✔
130	required: set[str] = set()	1✔
131	index = 0	1✔
132
133	while index < len(public):	1✔
134	token = public[index]	1✔
135	if token in _DOCKER_VALUE_OPTIONS and index + 1 < len(public):	1✔
136	replacement, changed, required_name = _redact_assignment(public[index + 1])	1✔
137	public[index + 1] = replacement	1✔
138	redacted = redacted or changed	1✔
139	if required_name:	1✔
140	required.add(required_name)	1✔
141	index += 2	1✔
142	continue	1✔
143
144	for prefix in ("--env=", "-e=", "--build-arg="):	1✔
145	if token.startswith(prefix):	1✔
NEW 146	replacement, changed, required_name = _redact_assignment(token[len(prefix) :])	×
NEW 147	public[index] = f"{prefix}{replacement}"	×
NEW 148	redacted = redacted or changed	×
NEW 149	if required_name:	×
NEW 150	required.add(required_name)	×
NEW 151	break	×
152	index += 1	1✔
153
154	return public, redacted, sorted(required)	1✔
155
156
157	def _environment_delta(	1✔
158	baseline: Mapping[str, str],
159	effective: Mapping[str, str],
160	explicit_set: Optional[Mapping[str, str]] = None,
161	) -> tuple[dict[str, str], list[str], bool]:
162	changed: dict[str, str] = {}	1✔
163	redacted = False	1✔
164	explicit_names = set(explicit_set or {})	1✔
165	for name in sorted(effective):	1✔
166	value = str(effective[name])	1✔
167	if baseline.get(name) == value and name not in explicit_names:	1✔
168	continue	1✔
169	if _is_sensitive_env_name(name):	1✔
NEW 170	changed[name] = "<redacted>"	×
NEW 171	redacted = True	×
172	else:
173	changed[name] = value	1✔
174	unset = sorted(name for name in baseline if name not in effective)	1✔
175	return changed, unset, redacted	1✔
176
177
178	def _shell_argv_groups(argv: Sequence[str]) -> list[list[str]]:	1✔
179	"""Group argv tokens into readable lines without changing shell semantics."""
180
181	tokens = [str(token) for token in argv]	1✔
182	if not tokens:	1✔
NEW 183	return []	×
184
185	head_length = 2 if len(tokens) > 1 and not tokens[1].startswith("-") else 1	1✔
186	return [tokens[:head_length], *[[token] for token in tokens[head_length:]]]	1✔
187
188
189	def _shell_group_lines(	1✔
190	groups: Sequence[Sequence[str]], *, first_indent: str, continuation_indent: str
191	) -> list[str]:
192	"""Quote grouped argv and add Bash line continuations."""
193
194	lines = []	1✔
195	for index, group in enumerate(groups):	1✔
196	indent = first_indent if index == 0 else continuation_indent	1✔
197	suffix = " \\" if index + 1 < len(groups) else ""	1✔
198	lines.append(f"{indent}{shlex.join([str(token) for token in group])}{suffix}")	1✔
199	return lines	1✔
200
201
202	def _process_shell(	1✔
203	argv: Sequence[str],
204	cwd: str,
205	environment_set: Mapping[str, str],
206	environment_unset: Sequence[str],
207	) -> str:
208	command_groups = _shell_argv_groups(argv)	1✔
209	lines = ["(", f" {shlex.join(['cd', '--', cwd])} && \\"]	1✔
210	if environment_set or environment_unset:	1✔
211	lines.append(" env \\")	1✔
212	env_groups = [["-u", name] for name in environment_unset]	1✔
213	env_groups.extend([[f"{name}={value}"] for name, value in environment_set.items()])	1✔
214	for group in env_groups:	1✔
215	lines.append(f" {shlex.join(group)} \\")	1✔
216	lines.extend(	1✔
217	_shell_group_lines(
218	command_groups,
219	first_indent=" ",
220	continuation_indent=" ",
221	)
222	)
223	else:
224	lines.extend(	1✔
225	_shell_group_lines(
226	command_groups,
227	first_indent=" ",
228	continuation_indent=" ",
229	)
230	)
231	lines.append(")")	1✔
232	return "\n".join(lines)	1✔
233
234
235	@dataclass	1✔
236	class ProcessStep:	1✔
237	"""A process invocation plus its private parity data."""
238
239	id: str	1✔
240	role: str	1✔
241	argv: list[str]	1✔
242	private_argv: list[str] = field(repr=False)	1✔
243	shell: str	1✔
244	cwd: str	1✔
245	environment: dict	1✔
246	check: bool	1✔
247	privilege: str	1✔
248	redacted: bool	1✔
249	destructive: bool = False	1✔
250
251	def public_dict(self) -> dict:	1✔
252	return {	1✔
253	"id": self.id,
254	"kind": "process",
255	"role": self.role,
256	"argv": self.argv,
257	"shell": self.shell,
258	"cwd": self.cwd,
259	"environment": self.environment,
260	"check": self.check,
261	"privilege": self.privilege,
262	"redacted": self.redacted,
263	"destructive": self.destructive,
264	}
265
266
267	class PlanRecorder:	1✔
268	"""Record a complete, ordered command plan for one CLI invocation."""
269
270	def __init__(self) -> None:	1✔
271	self._environment = dict(os.environ)	1✔
272	self.steps: list[ProcessStep] = []	1✔
273	self.limitations: list[dict] = []	1✔
274	self.warnings: list[dict] = []	1✔
275
276	@contextmanager	1✔
277	def activate(self) -> Iterator["PlanRecorder"]:	1✔
278	if get_active_recorder() is not None:	1✔
NEW 279	raise CommandPlanError("nested command-plan recorders are not supported")	×
280	token = _ACTIVE_RECORDER.set(self)	1✔
281	try:	1✔
282	yield self	1✔
283	finally:
284	_ACTIVE_RECORDER.reset(token)	1✔
285
286	def add_warning(self, code: str, message: str, step_id: Optional[str] = None) -> None:	1✔
287	warning = {"code": code, "message": message}	1✔
288	if step_id is not None:	1✔
289	warning["step_id"] = step_id	1✔
290	if warning not in self.warnings:	1✔
291	self.warnings.append(warning)	1✔
292
293	def record_process(	1✔
294	self,
295	argv: Sequence[str],
296	*,
297	role: str,
298	cwd: Optional[os.PathLike[str] \| str] = None,
299	env: Optional[Mapping[str, str]] = None,
300	explicit_env: Optional[Mapping[str, str]] = None,
301	check: bool,
302	privilege: str = "user",
303	destructive: bool = False,
304	) -> ProcessStep:
305	if role not in {"probe", "action", "cleanup"}:	1✔
NEW 306	raise CommandPlanError(f"unsupported process role: {role}")	×
307
308	private_argv = [str(token) for token in argv]	1✔
309	if not private_argv:	1✔
310	raise CommandPlanError("process argv must contain at least one token")	1✔
311	public_argv, argv_redacted, required = _public_argv(private_argv)	1✔
312	if env is not None:	1✔
313	if not explicit_env:	1✔
314	raise CommandPlanError(	1✔
315	"replacement subprocess environments are not supported in v1 plans"
316	)
317	expected_env = dict(self._environment)	1✔
318	expected_env.update({str(name): str(value) for name, value in explicit_env.items()})	1✔
319	if dict(env) != expected_env:	1✔
NEW 320	raise CommandPlanError(	×
321	"replacement subprocess environments are not supported in v1 plans"
322	)
323	effective_env = os.environ if env is None else env	1✔
324	missing_required = [name for name in required if name not in effective_env]	1✔
325	if missing_required:	1✔
326	names = ", ".join(missing_required)	1✔
327	raise CommandPlanError(	1✔
328	f"Docker environment reference is unset during planning: {names}"
329	)
330	environment_set, environment_unset, env_redacted = _environment_delta(	1✔
331	self._environment, effective_env, explicit_env
332	)
333	external_required = [name for name in required if name not in environment_set]	1✔
334	resolved_cwd = str(Path.cwd() if cwd is None else Path(cwd).resolve())	1✔
335	step_id = f"step-{len(self.steps) + 1:03d}"	1✔
336	redacted = argv_redacted or env_redacted	1✔
337	environment = {	1✔
338	"inherit": True,
339	"set": environment_set,
340	"unset": environment_unset,
341	"required": external_required,
342	}
343	step = ProcessStep(	1✔
344	id=step_id,
345	role=role,
346	argv=public_argv,
347	private_argv=private_argv,
348	shell=_process_shell(public_argv, resolved_cwd, environment_set, environment_unset),
349	cwd=resolved_cwd,
350	environment=environment,
351	check=check,
352	privilege=privilege,
353	redacted=redacted,
354	destructive=destructive,
355	)
356	self.steps.append(step)	1✔
357	if redacted:	1✔
358	self.add_warning(	1✔
359	"redacted_value",
360	"A credential-like literal was redacted from the public command plan.",
361	step_id,
362	)
363	return step	1✔
364
365	def _replay(self) -> dict:	1✔
366	self._ensure_complete()	1✔
367	replay_steps = [step for step in self.steps if step.role != "probe"]	1✔
368	required = sorted(	1✔
369	{name for step in replay_steps for name in step.environment.get("required", [])}
370	)
371
372	unavailable_reason = None	1✔
373	if any(step.redacted for step in replay_steps):	1✔
374	unavailable_reason = "redacted_literal"	1✔
375	elif any(not step.check for step in replay_steps):	1✔
NEW 376	unavailable_reason = "nonfatal_exit_handling"	×
377
378	if unavailable_reason is not None:	1✔
379	return {	1✔
380	"format": "bash",
381	"script": None,
382	"required_environment": required,
383	"unavailable_reason": unavailable_reason,
384	}
385
386	lines = ["#!/usr/bin/env bash", "set -e", ""]	1✔
387	for name in required:	1✔
388	lines.append(f': "${{{name}?Set {name} before running this script}}"')	1✔
389	if required:	1✔
390	lines.append("")	1✔
391	for index, step in enumerate(replay_steps):	1✔
392	summary = shlex.join(step.argv[:2]).replace("\r", "\\r").replace("\n", "\\n")	1✔
393	lines.append(f"# {step.id} ({step.role}): {summary}")	1✔
394	lines.append(step.shell)	1✔
395	if index + 1 < len(replay_steps):	1✔
396	lines.append("")	1✔
397	return {	1✔
398	"format": "bash",
399	"script": "\n".join(lines) + "\n",
400	"required_environment": required,
401	"unavailable_reason": None,
402	}
403
404	def _ensure_complete(self) -> None:	1✔
405	if not any(step.role in {"action", "cleanup"} for step in self.steps):	1✔
406	raise CommandPlanError("the resolved branch contains no action steps")	1✔
407
408	def payload(self) -> dict:	1✔
409	replay = self._replay()	1✔
410	warnings = list(self.warnings)	1✔
411	if replay["script"] is None:	1✔
412	warnings.append(	1✔
413	{
414	"code": "shell_replay_unavailable",
415	"message": f"Bash replay unavailable: {replay['unavailable_reason']}",
416	}
417	)
418	return {	1✔
419	"schema_version": 1,
420	"status": "dryrun",
421	"scope": "current_cli_process",
422	"host_resolved": True,
423	"complete": True,
424	"steps": [step.public_dict() for step in self.steps],
425	"replay": replay,
426	"limitations": list(self.limitations),
427	"warnings": warnings,
428	}
429
430	def json_text(self) -> str:	1✔
431	"""Render the complete JSON artifact without writing stdout."""
432
433	return json.dumps(self.payload(), indent=2)	1✔
434
435	def shell_text(self) -> str:	1✔
436	"""Render Bash or fail before anything is written to stdout."""
437
438	replay = self._replay()	1✔
439	if replay["script"] is None:	1✔
440	raise CommandPlanError(f"Bash replay unavailable: {replay['unavailable_reason']}")	1✔
441	return str(replay["script"])	1✔

nvidia-holoscan / holoscan-cli / 29762829348

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous