29760972341

Committed 20 Jul 2026 04:45PM UTC coverage: 79.991% (+1.4%) from 78.598%

Build # 29760972341

Build Type

push

github

Committed by

wyli

Commit Message

feat: add structured dry-run command plans

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

Coverage Stats

273 of 300 new or added lines in 7 files covered. (91.0%)

3 existing lines in 3 files now uncovered.

3386 of 4233 relevant lines covered (79.99%)

0.8 hits per line

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

93.05

/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 _process_shell(
    argv: Sequence[str],
    cwd: str,
    environment_set: Mapping[str, str],
    environment_unset: Sequence[str],
) -> str:
    command = [str(token) for token in argv]
    if environment_set or environment_unset:
        env_argv = ["env"]
        for name in environment_unset:
            env_argv.extend(["-u", name])
        env_argv.extend(f"{name}={value}" for name, value in environment_set.items())
        command = [*env_argv, *command]
    return f"{shlex.join(['cd', '--', cwd])} && {shlex.join(command)}"


@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]
        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}}"')
        lines.extend(step.shell for step in replay_steps)
        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 _process_shell(	1✔
179	argv: Sequence[str],
180	cwd: str,
181	environment_set: Mapping[str, str],
182	environment_unset: Sequence[str],
183	) -> str:
184	command = [str(token) for token in argv]	1✔
185	if environment_set or environment_unset:	1✔
186	env_argv = ["env"]	1✔
187	for name in environment_unset:	1✔
NEW 188	env_argv.extend(["-u", name])	×
189	env_argv.extend(f"{name}={value}" for name, value in environment_set.items())	1✔
190	command = [env_argv, command]	1✔
191	return f"{shlex.join(['cd', '--', cwd])} && {shlex.join(command)}"	1✔
192
193
194	@dataclass	1✔
195	class ProcessStep:	1✔
196	"""A process invocation plus its private parity data."""
197
198	id: str	1✔
199	role: str	1✔
200	argv: list[str]	1✔
201	private_argv: list[str] = field(repr=False)	1✔
202	shell: str	1✔
203	cwd: str	1✔
204	environment: dict	1✔
205	check: bool	1✔
206	privilege: str	1✔
207	redacted: bool	1✔
208	destructive: bool = False	1✔
209
210	def public_dict(self) -> dict:	1✔
211	return {	1✔
212	"id": self.id,
213	"kind": "process",
214	"role": self.role,
215	"argv": self.argv,
216	"shell": self.shell,
217	"cwd": self.cwd,
218	"environment": self.environment,
219	"check": self.check,
220	"privilege": self.privilege,
221	"redacted": self.redacted,
222	"destructive": self.destructive,
223	}
224
225
226	class PlanRecorder:	1✔
227	"""Record a complete, ordered command plan for one CLI invocation."""
228
229	def __init__(self) -> None:	1✔
230	self._environment = dict(os.environ)	1✔
231	self.steps: list[ProcessStep] = []	1✔
232	self.limitations: list[dict] = []	1✔
233	self.warnings: list[dict] = []	1✔
234
235	@contextmanager	1✔
236	def activate(self) -> Iterator["PlanRecorder"]:	1✔
237	if get_active_recorder() is not None:	1✔
NEW 238	raise CommandPlanError("nested command-plan recorders are not supported")	×
239	token = _ACTIVE_RECORDER.set(self)	1✔
240	try:	1✔
241	yield self	1✔
242	finally:
243	_ACTIVE_RECORDER.reset(token)	1✔
244
245	def add_warning(self, code: str, message: str, step_id: Optional[str] = None) -> None:	1✔
246	warning = {"code": code, "message": message}	1✔
247	if step_id is not None:	1✔
248	warning["step_id"] = step_id	1✔
249	if warning not in self.warnings:	1✔
250	self.warnings.append(warning)	1✔
251
252	def record_process(	1✔
253	self,
254	argv: Sequence[str],
255	*,
256	role: str,
257	cwd: Optional[os.PathLike[str] \| str] = None,
258	env: Optional[Mapping[str, str]] = None,
259	explicit_env: Optional[Mapping[str, str]] = None,
260	check: bool,
261	privilege: str = "user",
262	destructive: bool = False,
263	) -> ProcessStep:
264	if role not in {"probe", "action", "cleanup"}:	1✔
NEW 265	raise CommandPlanError(f"unsupported process role: {role}")	×
266
267	private_argv = [str(token) for token in argv]	1✔
268	public_argv, argv_redacted, required = _public_argv(private_argv)	1✔
269	if env is not None:	1✔
270	if not explicit_env:	1✔
271	raise CommandPlanError(	1✔
272	"replacement subprocess environments are not supported in v1 plans"
273	)
274	expected_env = dict(self._environment)	1✔
275	expected_env.update({str(name): str(value) for name, value in explicit_env.items()})	1✔
276	if dict(env) != expected_env:	1✔
NEW 277	raise CommandPlanError(	×
278	"replacement subprocess environments are not supported in v1 plans"
279	)
280	effective_env = os.environ if env is None else env	1✔
281	missing_required = [name for name in required if name not in effective_env]	1✔
282	if missing_required:	1✔
283	names = ", ".join(missing_required)	1✔
284	raise CommandPlanError(	1✔
285	f"Docker environment reference is unset during planning: {names}"
286	)
287	environment_set, environment_unset, env_redacted = _environment_delta(	1✔
288	self._environment, effective_env, explicit_env
289	)
290	external_required = [name for name in required if name not in environment_set]	1✔
291	resolved_cwd = str(Path.cwd() if cwd is None else Path(cwd).resolve())	1✔
292	step_id = f"step-{len(self.steps) + 1:03d}"	1✔
293	redacted = argv_redacted or env_redacted	1✔
294	environment = {	1✔
295	"inherit": True,
296	"set": environment_set,
297	"unset": environment_unset,
298	"required": external_required,
299	}
300	step = ProcessStep(	1✔
301	id=step_id,
302	role=role,
303	argv=public_argv,
304	private_argv=private_argv,
305	shell=_process_shell(public_argv, resolved_cwd, environment_set, environment_unset),
306	cwd=resolved_cwd,
307	environment=environment,
308	check=check,
309	privilege=privilege,
310	redacted=redacted,
311	destructive=destructive,
312	)
313	self.steps.append(step)	1✔
314	if redacted:	1✔
315	self.add_warning(	1✔
316	"redacted_value",
317	"A credential-like literal was redacted from the public command plan.",
318	step_id,
319	)
320	return step	1✔
321
322	def _replay(self) -> dict:	1✔
323	self._ensure_complete()	1✔
324	replay_steps = [step for step in self.steps if step.role != "probe"]	1✔
325	required = sorted(	1✔
326	{name for step in replay_steps for name in step.environment.get("required", [])}
327	)
328
329	unavailable_reason = None	1✔
330	if any(step.redacted for step in replay_steps):	1✔
331	unavailable_reason = "redacted_literal"	1✔
332	elif any(not step.check for step in replay_steps):	1✔
NEW 333	unavailable_reason = "nonfatal_exit_handling"	×
334
335	if unavailable_reason is not None:	1✔
336	return {	1✔
337	"format": "bash",
338	"script": None,
339	"required_environment": required,
340	"unavailable_reason": unavailable_reason,
341	}
342
343	lines = ["#!/usr/bin/env bash", "set -e"]	1✔
344	for name in required:	1✔
345	lines.append(f': "${{{name}?Set {name} before running this script}}"')	1✔
346	lines.extend(step.shell for step in replay_steps)	1✔
347	return {	1✔
348	"format": "bash",
349	"script": "\n".join(lines) + "\n",
350	"required_environment": required,
351	"unavailable_reason": None,
352	}
353
354	def _ensure_complete(self) -> None:	1✔
355	if not any(step.role in {"action", "cleanup"} for step in self.steps):	1✔
356	raise CommandPlanError("the resolved branch contains no action steps")	1✔
357
358	def payload(self) -> dict:	1✔
359	replay = self._replay()	1✔
360	warnings = list(self.warnings)	1✔
361	if replay["script"] is None:	1✔
362	warnings.append(	1✔
363	{
364	"code": "shell_replay_unavailable",
365	"message": f"Bash replay unavailable: {replay['unavailable_reason']}",
366	}
367	)
368	return {	1✔
369	"schema_version": 1,
370	"status": "dryrun",
371	"scope": "current_cli_process",
372	"host_resolved": True,
373	"complete": True,
374	"steps": [step.public_dict() for step in self.steps],
375	"replay": replay,
376	"limitations": list(self.limitations),
377	"warnings": warnings,
378	}
379
380	def json_text(self) -> str:	1✔
381	"""Render the complete JSON artifact without writing stdout."""
382
383	return json.dumps(self.payload(), indent=2)	1✔
384
385	def shell_text(self) -> str:	1✔
386	"""Render Bash or fail before anything is written to stdout."""
387
388	replay = self._replay()	1✔
389	if replay["script"] is None:	1✔
390	raise CommandPlanError(f"Bash replay unavailable: {replay['unavailable_reason']}")	1✔
391	return str(replay["script"])	1✔

nvidia-holoscan / holoscan-cli / 29760972341

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