14182007055

Committed 31 Mar 2025 09:39PM UTC coverage: 89.45% (+3.7%) from 85.772%

Build # 14182007055

Build Type

Pull #33

github

Committed by

mocsharp

Commit Message

Add unit tests

Signed-off-by: Victor Chang <vicchang@nvidia.com>

Pull Request Pull Request #33: Enable test-app with nv-gha-runner

Run Details

46 of 51 new or added lines in 8 files covered. (90.2%)

1 existing line in 1 file now uncovered.

1789 of 2000 relevant lines covered (89.45%)

0.89 hits per line

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

91.08

/src/holoscan_cli/runner/runner.py

# SPDX-FileCopyrightText: Copyright (c) 2023-2025 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.
import json
import logging
import os
import re
import shutil
import sys
import tempfile
from argparse import Namespace
from glob import glob
from pathlib import Path
from typing import Optional

from ..common.dockerutils import create_or_use_network, docker_run, image_exists
from ..common.exceptions import ManifestReadError, UnmatchedDeviceError
from ..common.utils import (
    compare_versions,
    get_requested_gpus,
    print_manifest_json,
    run_cmd,
    run_cmd_output,
)
from .resources import get_shared_memory_size

logger = logging.getLogger("runner")


def _fetch_map_manifest(map_name: str) -> tuple[dict, dict]:
    """
    Execute HAP/MAP and fetch the manifest files.

    Args:
        map_name: HAP/MAP image name.

    Returns:
        app_info: application manifest as a python dict
        pkg_info: package manifest as a python dict
    """

    def _ensure_valid_exit_code(returncode: int) -> None:
        if returncode != 0:
            raise ManifestReadError("Error reading manifest file from the package.")

    logger.info("Reading HAP/MAP manifest...")

    with tempfile.TemporaryDirectory() as info_dir:
        docker_id = run_cmd_output(["docker", "create", map_name]).strip()
        returncode = run_cmd(
            [
                "docker",
                "cp",
                f"{docker_id}:/etc/holoscan/app.json",
                f"{info_dir}/app.json",
            ]
        )
        _ensure_valid_exit_code(returncode)
        returncode = run_cmd(
            [
                "docker",
                "cp",
                f"{docker_id}:/etc/holoscan/pkg.json",
                f"{info_dir}/pkg.json",
            ]
        )
        _ensure_valid_exit_code(returncode)
        returncode = run_cmd(["docker", "rm", "-v", docker_id])
        _ensure_valid_exit_code(returncode)

        app_json = Path(f"{info_dir}/app.json")
        pkg_json = Path(f"{info_dir}/pkg.json")

        app_info = json.loads(app_json.read_text())
        pkg_info = json.loads(pkg_json.read_text())

        print_manifest_json(app_info, "app.json")
        print_manifest_json(pkg_info, "pkg.json")

        return app_info, pkg_info


def _run_app(args: Namespace, app_info: dict, pkg_info: dict):
    """
    Executes the Holoscan Application Package.

    Args:
        args: user arguments
        app_info: application manifest dictionary
        pkg_info: package manifest dictionary

    Returns:
        returncode: command returncode
    """

    map_name: str = args.map
    input_path: Path = args.input
    output_path: Path = args.output
    quiet: bool = args.quiet
    driver: bool = args.driver
    worker: bool = args.worker
    health_check: bool = args.health_check
    fragments: Optional[str] = args.fragments
    network: str = create_or_use_network(args.network, map_name)
    nic: Optional[str] = args.nic if args.nic else None
    use_all_nics: bool = args.use_all_nics
    gpus: Optional[str] = args.gpus if args.gpus else None
    config: Optional[Path] = args.config if args.config else None
    address: Optional[str] = args.address if args.address else None
    worker_address: Optional[str] = args.worker_address if args.worker_address else None
    render: bool = args.render
    user: str = f"{args.uid}:{args.gid}"
    remove: bool = args.rm
    hostname: Optional[str] = "driver" if driver else None
    terminal: bool = args.terminal
    platform_config: str = pkg_info.get("platformConfig")
    shared_memory_size: Optional[str] = (
        get_shared_memory_size(pkg_info, worker, driver, fragments, args.shm_size)
        if args.shm_size
        else None
    )

    commands = []
    devices = _lookup_devices(args.device) if args.device is not None else []

    if driver:
        commands.append("--driver")
        logger.info("Application running in Driver mode")
    if worker:
        commands.append("--worker")
        logger.info("Application running in Worker mode")
    if fragments:
        commands.append("--fragments")
        commands.append(fragments)
        logger.info(f"Configured fragments: {fragments}")
    if address:
        commands.append("--address")
        commands.append(address)
        logger.info(f"App Driver address and port: {address}")
    if worker_address:
        commands.append("--worker_address")
        commands.append(worker_address)
        logger.info(f"App Worker address and port: {worker_address}")

    docker_run(
        hostname,
        map_name,
        input_path,
        output_path,
        app_info,
        pkg_info,
        quiet,
        commands,
        health_check,
        network,
        nic,
        use_all_nics,
        gpus,
        config,
        render,
        user,
        terminal,
        devices,
        platform_config,
        shared_memory_size,
        args.uid == 0,
        remove,
    )


def _lookup_devices(devices: list[str]) -> list[str]:
    """
    Looks up matching devices in /dev and returns a list
    of fully qualified device paths.

    Raises exception if any devices cannot be found.
    """
    matched_devices = []
    unmatched_devices = []

    for dev in devices:
        pattern = dev
        if not pattern.startswith("/"):
            pattern = "/dev/" + pattern
        found_devices = glob(pattern)
        if len(found_devices) == 0:
            unmatched_devices.append(dev)
        else:
            matched_devices.extend(found_devices)

    if len(unmatched_devices) > 0:
        raise UnmatchedDeviceError(unmatched_devices)

    return matched_devices


def _dependency_verification(map_name: str) -> bool:
    """Check if all the dependencies are installed or not.

    Args:
        map_name: HAP/MAP name

    Returns:
        True if all dependencies are satisfied, otherwise False.
    """
    logger.info("Checking dependencies...")

    # check for docker
    prog = "docker"
    logger.info(f'--> Verifying if "{prog}" is installed...\n')
    if not shutil.which(prog):
        logger.error(
            '"%s" not installed, please install it from https://docs.docker.com/engine/install/.',
            prog,
        )
        return False

    buildx_paths = [
        os.path.expandvars("$HOME/.docker/cli-plugins"),
        "/usr/local/lib/docker/cli-plugins",
        "/usr/local/libexec/docker/cli-plugins",
        "/usr/lib/docker/cli-plugins",
        "/usr/libexec/docker/cli-plugins",
    ]
    prog = "docker-buildx"
    logger.info('--> Verifying if "%s" is installed...\n', prog)
    buildx_found = False
    for path in buildx_paths:
        if shutil.which(os.path.join(path, prog)):
            buildx_found = True

    if not buildx_found:
        logger.error(
            '"%s" not installed, please install it from https://docs.docker.com/engine/install/.',
            prog,
        )
        return False

    # check for map image
    logger.info('--> Verifying if "%s" is available...\n', map_name)
    if not image_exists(map_name):
        logger.error("Unable to fetch required image.")
        return False

    return True


def _pkg_specific_dependency_verification(pkg_info: dict) -> bool:
    """Checks for any package specific dependencies.

    Currently it verifies the following dependencies:
    * If gpu has been requested by the application, verify that nvidia-ctk is installed.
    Note: when running inside a Docker container, always assume nvidia-ctk is installed.
    Args:
        pkg_info: package manifest as a python dict

    Returns:
        True if all dependencies are satisfied, otherwise False.
    """
    if os.path.exists("/.dockerenv"):
        logger.info("--> Skipping nvidia-ctk check inside Docker...\n")
        return True

    requested_gpus = get_requested_gpus(pkg_info)
    if requested_gpus > 0:
        # check for NVIDIA Container TOolkit
        prog = "nvidia-ctk"
        logger.info('--> Verifying if "%s" is installed...\n', prog)
        if not shutil.which(prog):
            logger.error(
                '"%s" not installed, please install NVIDIA Container Toolkit.', prog
            )
            return False

        logger.info('--> Verifying "%s" version...\n', prog)
        output = run_cmd_output(["nvidia-ctk", "--version"], "version")
        match = re.search(r"([0-9]+\.[0-9]+\.[0-9]+)", output)
        min_ctk_version = "1.12.0"
        recommended_ctk_version = "1.14.1"

        if match is None:
            logger.error(
                f"Error detecting NVIDIA Container Toolkit version. "
                f"Version {min_ctk_version}+ is required ({recommended_ctk_version}+ recommended)."
            )
            return False
        elif compare_versions(min_ctk_version, match.group()) > 0:
            logger.error(
                f"Found '{prog}' Version {match.group()}. "
                f"Version {min_ctk_version}+ is required ({recommended_ctk_version}+ recommended)."
            )
            return False

    return True


def execute_run_command(args: Namespace):
    """
    Entrypoint for the Holoscan Run command.

    Args:
        args: Namespace object containing user arguments.

    Returns:
        None
    """
    if not _dependency_verification(args.map):
        logger.error("Execution Aborted")
        sys.exit(2)
    try:
        # Fetch application manifest from MAP
        app_info, pkg_info = _fetch_map_manifest(args.map)
    except Exception as e:
        logger.error(f"Failed to fetch MAP manifest: {e}")
        sys.exit(2)

    if not _pkg_specific_dependency_verification(pkg_info):
        logger.error("Execution Aborted")
        sys.exit(2)
    try:
        # Run Holoscan Application
        _run_app(args, app_info, pkg_info)
    except Exception as ex:
        logger.debug(ex, exc_info=True)
        logger.error(f"Error executing {args.map}: {ex}")
        sys.exit(2)

1	# SPDX-FileCopyrightText: Copyright (c) 2023-2025 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	import json	1✔
16	import logging	1✔
17	import os	1✔
18	import re	1✔
19	import shutil	1✔
20	import sys	1✔
21	import tempfile	1✔
22	from argparse import Namespace	1✔
23	from glob import glob	1✔
24	from pathlib import Path	1✔
25	from typing import Optional	1✔
26
27	from ..common.dockerutils import create_or_use_network, docker_run, image_exists	1✔
28	from ..common.exceptions import ManifestReadError, UnmatchedDeviceError	1✔
29	from ..common.utils import (	1✔
30	compare_versions,
31	get_requested_gpus,
32	print_manifest_json,
33	run_cmd,
34	run_cmd_output,
35	)
36	from .resources import get_shared_memory_size	1✔
37
38	logger = logging.getLogger("runner")	1✔
39
40
41	def _fetch_map_manifest(map_name: str) -> tuple[dict, dict]:	1✔
42	"""
43	Execute HAP/MAP and fetch the manifest files.
44
45	Args:
46	map_name: HAP/MAP image name.
47
48	Returns:
49	app_info: application manifest as a python dict
50	pkg_info: package manifest as a python dict
51	"""
52
53	def _ensure_valid_exit_code(returncode: int) -> None:	1✔
54	if returncode != 0:	1✔
55	raise ManifestReadError("Error reading manifest file from the package.")	1✔
56
57	logger.info("Reading HAP/MAP manifest...")	1✔
58
59	with tempfile.TemporaryDirectory() as info_dir:	1✔
60	docker_id = run_cmd_output(["docker", "create", map_name]).strip()	1✔
61	returncode = run_cmd(	1✔
62	[
63	"docker",
64	"cp",
65	f"{docker_id}:/etc/holoscan/app.json",
66	f"{info_dir}/app.json",
67	]
68	)
69	_ensure_valid_exit_code(returncode)	1✔
70	returncode = run_cmd(	1✔
71	[
72	"docker",
73	"cp",
74	f"{docker_id}:/etc/holoscan/pkg.json",
75	f"{info_dir}/pkg.json",
76	]
77	)
78	_ensure_valid_exit_code(returncode)	1✔
79	returncode = run_cmd(["docker", "rm", "-v", docker_id])	1✔
80	_ensure_valid_exit_code(returncode)	1✔
81
82	app_json = Path(f"{info_dir}/app.json")	1✔
83	pkg_json = Path(f"{info_dir}/pkg.json")	1✔
84
85	app_info = json.loads(app_json.read_text())	1✔
86	pkg_info = json.loads(pkg_json.read_text())	1✔
87
88	print_manifest_json(app_info, "app.json")	1✔
89	print_manifest_json(pkg_info, "pkg.json")	1✔
90
91	return app_info, pkg_info	1✔
92
93
94	def _run_app(args: Namespace, app_info: dict, pkg_info: dict):	1✔
95	"""
96	Executes the Holoscan Application Package.
97
98	Args:
99	args: user arguments
100	app_info: application manifest dictionary
101	pkg_info: package manifest dictionary
102
103	Returns:
104	returncode: command returncode
105	"""
106
107	map_name: str = args.map	1✔
108	input_path: Path = args.input	1✔
109	output_path: Path = args.output	1✔
110	quiet: bool = args.quiet	1✔
111	driver: bool = args.driver	1✔
112	worker: bool = args.worker	1✔
113	health_check: bool = args.health_check	1✔
114	fragments: Optional[str] = args.fragments	1✔
115	network: str = create_or_use_network(args.network, map_name)	1✔
116	nic: Optional[str] = args.nic if args.nic else None	1✔
117	use_all_nics: bool = args.use_all_nics	1✔
118	gpus: Optional[str] = args.gpus if args.gpus else None	1✔
119	config: Optional[Path] = args.config if args.config else None	1✔
120	address: Optional[str] = args.address if args.address else None	1✔
121	worker_address: Optional[str] = args.worker_address if args.worker_address else None	1✔
122	render: bool = args.render	1✔
123	user: str = f"{args.uid}:{args.gid}"	1✔
124	remove: bool = args.rm	1✔
125	hostname: Optional[str] = "driver" if driver else None	1✔
126	terminal: bool = args.terminal	1✔
127	platform_config: str = pkg_info.get("platformConfig")	1✔
128	shared_memory_size: Optional[str] = (	1✔
129	get_shared_memory_size(pkg_info, worker, driver, fragments, args.shm_size)
130	if args.shm_size
131	else None
132	)
133
134	commands = []	1✔
135	devices = _lookup_devices(args.device) if args.device is not None else []	1✔
136
137	if driver:	1✔
138	commands.append("--driver")	1✔
139	logger.info("Application running in Driver mode")	1✔
140	if worker:	1✔
141	commands.append("--worker")	1✔
142	logger.info("Application running in Worker mode")	1✔
143	if fragments:	1✔
144	commands.append("--fragments")	1✔
145	commands.append(fragments)	1✔
146	logger.info(f"Configured fragments: {fragments}")	1✔
147	if address:	1✔
148	commands.append("--address")	1✔
149	commands.append(address)	1✔
150	logger.info(f"App Driver address and port: {address}")	1✔
151	if worker_address:	1✔
152	commands.append("--worker_address")	1✔
153	commands.append(worker_address)	1✔
154	logger.info(f"App Worker address and port: {worker_address}")	1✔
155
156	docker_run(	1✔
157	hostname,
158	map_name,
159	input_path,
160	output_path,
161	app_info,
162	pkg_info,
163	quiet,
164	commands,
165	health_check,
166	network,
167	nic,
168	use_all_nics,
169	gpus,
170	config,
171	render,
172	user,
173	terminal,
174	devices,
175	platform_config,
176	shared_memory_size,
177	args.uid == 0,
178	remove,
179	)
180
181
182	def _lookup_devices(devices: list[str]) -> list[str]:	1✔
183	"""
184	Looks up matching devices in /dev and returns a list
185	of fully qualified device paths.
186
187	Raises exception if any devices cannot be found.
188	"""
189	matched_devices = []	1✔
190	unmatched_devices = []	1✔
191
192	for dev in devices:	1✔
193	pattern = dev	1✔
194	if not pattern.startswith("/"):	1✔
195	pattern = "/dev/" + pattern	1✔
196	found_devices = glob(pattern)	1✔
197	if len(found_devices) == 0:	1✔
198	unmatched_devices.append(dev)	1✔
199	else:
200	matched_devices.extend(found_devices)	1✔
201
202	if len(unmatched_devices) > 0:	1✔
203	raise UnmatchedDeviceError(unmatched_devices)	1✔
204
205	return matched_devices	1✔
206
207
208	def _dependency_verification(map_name: str) -> bool:	1✔
209	"""Check if all the dependencies are installed or not.
210
211	Args:
212	map_name: HAP/MAP name
213
214	Returns:
215	True if all dependencies are satisfied, otherwise False.
216	"""
217	logger.info("Checking dependencies...")	1✔
218
219	# check for docker
220	prog = "docker"	1✔
221	logger.info(f'--> Verifying if "{prog}" is installed...\n')	1✔
222	if not shutil.which(prog):	1✔
223	logger.error(	1✔
224	'"%s" not installed, please install it from https://docs.docker.com/engine/install/.',
225	prog,
226	)
227	return False	1✔
228
229	buildx_paths = [	1✔
230	os.path.expandvars("$HOME/.docker/cli-plugins"),
231	"/usr/local/lib/docker/cli-plugins",
232	"/usr/local/libexec/docker/cli-plugins",
233	"/usr/lib/docker/cli-plugins",
234	"/usr/libexec/docker/cli-plugins",
235	]
236	prog = "docker-buildx"	1✔
237	logger.info('--> Verifying if "%s" is installed...\n', prog)	1✔
238	buildx_found = False	1✔
239	for path in buildx_paths:	1✔
240	if shutil.which(os.path.join(path, prog)):	1✔
241	buildx_found = True	1✔
242
243	if not buildx_found:	1✔
244	logger.error(	1✔
245	'"%s" not installed, please install it from https://docs.docker.com/engine/install/.',
246	prog,
247	)
248	return False	1✔
249
250	# check for map image
251	logger.info('--> Verifying if "%s" is available...\n', map_name)	1✔
252	if not image_exists(map_name):	1✔
253	logger.error("Unable to fetch required image.")	×
254	return False	×
255
256	return True	1✔
257
258
259	def _pkg_specific_dependency_verification(pkg_info: dict) -> bool:	1✔
260	"""Checks for any package specific dependencies.
261
262	Currently it verifies the following dependencies:
263	* If gpu has been requested by the application, verify that nvidia-ctk is installed.
264	Note: when running inside a Docker container, always assume nvidia-ctk is installed.
265	Args:
266	pkg_info: package manifest as a python dict
267
268	Returns:
269	True if all dependencies are satisfied, otherwise False.
270	"""
271	if os.path.exists("/.dockerenv"):	1✔
UNCOV 272	logger.info("--> Skipping nvidia-ctk check inside Docker...\n")	×
273	return True	×
274
275	requested_gpus = get_requested_gpus(pkg_info)	1✔
276	if requested_gpus > 0:	1✔
277	# check for NVIDIA Container TOolkit
278	prog = "nvidia-ctk"	1✔
279	logger.info('--> Verifying if "%s" is installed...\n', prog)	1✔
280	if not shutil.which(prog):	1✔
281	logger.error(	×
282	'"%s" not installed, please install NVIDIA Container Toolkit.', prog
283	)
284	return False	×
285
286	logger.info('--> Verifying "%s" version...\n', prog)	1✔
287	output = run_cmd_output(["nvidia-ctk", "--version"], "version")	1✔
288	match = re.search(r"([0-9]+\.[0-9]+\.[0-9]+)", output)	1✔
289	min_ctk_version = "1.12.0"	1✔
290	recommended_ctk_version = "1.14.1"	1✔
291
292	if match is None:	1✔
293	logger.error(	×
294	f"Error detecting NVIDIA Container Toolkit version. "
295	f"Version {min_ctk_version}+ is required ({recommended_ctk_version}+ recommended)."
296	)
297	return False	×
298	elif compare_versions(min_ctk_version, match.group()) > 0:	1✔
299	logger.error(	1✔
300	f"Found '{prog}' Version {match.group()}. "
301	f"Version {min_ctk_version}+ is required ({recommended_ctk_version}+ recommended)."
302	)
303	return False	1✔
304
305	return True	1✔
306
307
308	def execute_run_command(args: Namespace):	1✔
309	"""
310	Entrypoint for the Holoscan Run command.
311
312	Args:
313	args: Namespace object containing user arguments.
314
315	Returns:
316	None
317	"""
318	if not _dependency_verification(args.map):	1✔
319	logger.error("Execution Aborted")	1✔
320	sys.exit(2)	1✔
321	try:	1✔
322	# Fetch application manifest from MAP
323	app_info, pkg_info = _fetch_map_manifest(args.map)	1✔
324	except Exception as e:	1✔
325	logger.error(f"Failed to fetch MAP manifest: {e}")	1✔
326	sys.exit(2)	1✔
327
328	if not _pkg_specific_dependency_verification(pkg_info):	1✔
329	logger.error("Execution Aborted")	×
330	sys.exit(2)	×
331	try:	1✔
332	# Run Holoscan Application
333	_run_app(args, app_info, pkg_info)	1✔
334	except Exception as ex:	×
335	logger.debug(ex, exc_info=True)	×
336	logger.error(f"Error executing {args.map}: {ex}")	×
337	sys.exit(2)	×

nvidia-holoscan / holoscan-cli / 14182007055

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