10577080486

Committed 27 Aug 2024 11:19AM UTC coverage: 97.623% (+0.04%) from 97.587%

Build # 10577080486

Build Type

Pull #608

github

Committed by

web-flow

Commit Message

Merge bdccc8480 into 76541f3af

Pull Request Pull Request #608: Add SDPX format support for SBOM

Run Details

68 of 69 new or added lines in 2 files covered. (98.55%)

4 existing lines in 1 file now uncovered.

3861 of 3955 relevant lines covered (97.62%)

3.9 hits per line

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

96.93

/cachi2/interface/cli.py

import enum
import functools
import importlib.metadata
import json
import logging
import shutil
import sys
from itertools import chain
from pathlib import Path
from typing import Any, Callable, Optional, Union

import pydantic
import typer

import cachi2.core.config as config
from cachi2.core.errors import Cachi2Error, InvalidInput, UnexpectedFormat
from cachi2.core.extras.envfile import EnvFormat, generate_envfile
from cachi2.core.models.input import Flag, PackageInput, Request, parse_user_input
from cachi2.core.models.output import BuildConfig
from cachi2.core.models.property_semantics import merge_component_properties
from cachi2.core.models.sbom import Sbom, SPDXSbom
from cachi2.core.resolver import inject_files_post, resolve_packages, supported_package_managers
from cachi2.core.rooted_path import RootedPath
from cachi2.interface.logging import LogLevel, setup_logging

app = typer.Typer()
log = logging.getLogger(__name__)

DEFAULT_SOURCE = "."
DEFAULT_OUTPUT = "./cachi2-output"


OUTFILE_OPTION = typer.Option(
    None,
    "-o",
    "--output",
    dir_okay=False,
    help="Write to this file instead of standard output.",
)


Paths = list[Path]


def _bail_out_with_error(e: Cachi2Error) -> None:
    """Report and error and set correct exit code."""
    log.error("%s: %s", type(e).__name__, str(e).replace("\n", r"\n"))
    print(f"Error: {type(e).__name__}: {e.friendly_msg()}", file=sys.stderr)
    raise typer.Exit(2 if e.is_invalid_usage else 1)


def handle_errors(cmd: Callable[..., None]) -> Callable[..., None]:
    """Decorate a CLI command function with an error handler.

    All errors will be logged at ERROR level before exiting.
    Expected errors will be printed in a friendlier format rather than showing the whole traceback.
    Errors that we consider invalid usage will result in exit code 2.
    """

    def log_error(error: Exception) -> None:
        log.error("%s: %s", type(error).__name__, str(error).replace("\n", r"\n"))

    @functools.wraps(cmd)
    def cmd_with_error_handling(*args: tuple[Any, ...], **kwargs: dict[str, Any]) -> None:
        try:
            cmd(*args, **kwargs)
        except Cachi2Error as e:
            _bail_out_with_error(e)
        except Exception as e:
            log_error(e)
            raise

    return cmd_with_error_handling


def version_callback(value: bool) -> None:
    """If --version was used, print the cachi2 version and exit."""
    if not value:
        return

    print("cachi2", importlib.metadata.version("cachi2"))
    print("Supported package managers:", ", ".join(supported_package_managers))
    raise typer.Exit()


class SBOMFormat(str, enum.Enum):
    """The type of SBOM to generate."""

    cyclonedx = "cyclonedx"
    spdx = "spdx"


@app.callback()
@handle_errors
def cachi2(  # noqa: D103; docstring becomes part of --help message
    version: bool = typer.Option(
        False,
        "--version",
        callback=version_callback,
        is_eager=True,
        help="Show version and exit.",
    ),
    config_file: Path = typer.Option(
        None,
        "--config-file",
        help="Read configuration from this file.",
        dir_okay=False,
        exists=True,
        resolve_path=True,
        readable=True,
    ),
    log_level: LogLevel = typer.Option(
        LogLevel.INFO.value,
        "--log-level",
        case_sensitive=False,
        help="Set log level.",
    ),
) -> None:
    setup_logging(log_level)
    if config_file:
        config.set_config(config_file)


def _if_json_then_validate(value: str) -> str:
    if _looks_like_json(value):
        try:
            json.loads(value)
        except json.JSONDecodeError:
            raise typer.BadParameter(f"Looks like JSON but is not valid JSON: {value!r}")
    return value


def _looks_like_json(value: str) -> bool:
    return value.lstrip().startswith(("{", "["))


class _Input(pydantic.BaseModel, extra="forbid"):
    packages: list[PackageInput]
    flags: list[Flag] = list()


@app.command()
@handle_errors
def fetch_deps(
    raw_input: str = typer.Argument(
        ...,
        help="Specify package (within the source repo) to process. See usage examples.",
        metavar="PKG",
        callback=_if_json_then_validate,
    ),
    source: Path = typer.Option(
        DEFAULT_SOURCE,
        exists=True,
        file_okay=False,
        resolve_path=True,
        help="Process the git repository at this path.",
    ),
    output: Path = typer.Option(
        DEFAULT_OUTPUT,
        file_okay=False,
        resolve_path=True,
        help="Write output files to this directory.",
    ),
    dev_package_managers: bool = typer.Option(False, "--dev-package-managers", hidden=True),
    cgo_disable: bool = typer.Option(
        False, "--cgo-disable", help="Set CGO_ENABLED=0 while processing gomod packages."
    ),
    force_gomod_tidy: bool = typer.Option(
        False,
        "--force-gomod-tidy",
        help="Run 'go mod tidy' after downloading go dependencies.",
    ),
    gomod_vendor: bool = typer.Option(
        False,
        "--gomod-vendor",
        help=(
            "Fetch go deps via 'go mod vendor' rather than 'go mod download'. If you "
            "have a vendor/ dir, one of --gomod-vendor/--gomod-vendor-check is required."
        ),
    ),
    gomod_vendor_check: bool = typer.Option(
        False,
        "--gomod-vendor-check",
        help=(
            "Same as gomod-vendor, but will not make unexpected changes if you "
            "already have a vendor/ directory (will fail if changes would be made)."
        ),
    ),
    sbom_output_type: SBOMFormat = typer.Option(
        SBOMFormat.cyclonedx,
        "--sbom-type",
        help=("Format of generated SBOM. Default is CycloneDX"),
    ),
) -> None:
    """Fetch dependencies for supported package managers.

    \b
    # gomod package in the current directory
    cachi2 fetch-deps gomod

    \b
    # pip package in the root of the source directory
    cachi2 fetch-deps --source ./my-repo pip

    \b
    # gomod package in a subpath of the source directory (./my-repo/subpath)
    cachi2 fetch-deps --source ./my-repo '{
        "type": "gomod",
        "path": "subpath"
    }'

    \b
    # multiple packages as a JSON list
    cachi2 fetch-deps '[
        {"type": "gomod"},
        {"type": "gomod", "path": "subpath"},
        {"type": "pip", "path": "other-path"}
    ]'

    \b
    # multiple packages and flags as a JSON list
    cachi2 fetch-deps '{
        "packages": [
            {"type": "gomod"},
            {"type": "gomod", "path": "subpath"},
            {"type": "pip", "path": "other-path"}
        ],
        "flags": [
            "gomod-vendor"
        ]
    }'
    """  # noqa: D301, D202; backslashes intentional, blank line required by black

    def normalize_input() -> dict[str, list[Any]]:
        """Format raw_input so it can be parsed by the _Input class."""
        if _looks_like_json(raw_input):
            parsed_input = json.loads(raw_input)

            if isinstance(parsed_input, dict):
                if "packages" in parsed_input.keys():
                    # is a dict with list of packages and possibly flags
                    return parsed_input
                else:
                    # is a dict representing a package
                    return {"packages": [parsed_input]}
            else:
                # is a list
                return {"packages": parsed_input}
        else:
            # is a str
            return {"packages": [{"type": raw_input}]}

    def combine_option_and_json_flags(json_flags: list[Flag]) -> list[str]:
        flag_names = [
            "cgo-disable",
            "dev-package-managers",
            "force-gomod-tidy",
            "gomod-vendor",
            "gomod-vendor-check",
        ]
        flag_values = [
            cgo_disable,
            dev_package_managers,
            force_gomod_tidy,
            gomod_vendor,
            gomod_vendor_check,
        ]
        flags = [name for name, value in zip(flag_names, flag_values) if value]

        if json_flags:
            flags.extend(flag.strip() for flag in json_flags)

        return flags

    input = parse_user_input(_Input.model_validate, normalize_input())

    request = parse_user_input(
        Request.model_validate,
        {
            "source_dir": source,
            "output_dir": output,
            "packages": input.packages,
            "flags": combine_option_and_json_flags(input.flags),
        },
    )

    deps_dir = output / "deps"
    if deps_dir.exists():
        log.debug(f"Removing existing deps directory '{deps_dir}'")
        shutil.rmtree(deps_dir, ignore_errors=True)

    request_output = resolve_packages(request)

    request.output_dir.path.mkdir(parents=True, exist_ok=True)
    request.output_dir.join_within_root(".build-config.json").path.write_text(
        request_output.build_config.model_dump_json(indent=2, exclude_none=True)
    )

    if sbom_output_type == SBOMFormat.cyclonedx:
        sbom: Union[Sbom, SPDXSbom] = request_output.generate_sbom()
    else:
        sbom = request_output.generate_sbom().to_spdx()
    request.output_dir.join_within_root("bom.json").path.write_text(
        # the Sbom model has camelCase aliases in some fields
        sbom.model_dump_json(indent=2, by_alias=True, exclude_none=True)
    )

    log.info(r"All dependencies fetched successfully \o/")


FROM_OUTPUT_DIR_ARG = typer.Argument(
    ...,
    exists=True,
    file_okay=False,
    resolve_path=True,
    help="The output directory populated by a previous fetch-deps command.",
)
FOR_OUTPUT_DIR_OPTION = typer.Option(
    None,
    resolve_path=True,
    help="Generate output as if the output directory was at this path instead.",
)


@app.command()
@handle_errors
def generate_env(
    from_output_dir: Path = FROM_OUTPUT_DIR_ARG,
    for_output_dir: Optional[Path] = FOR_OUTPUT_DIR_OPTION,
    output: Optional[Path] = OUTFILE_OPTION,
    fmt: Optional[EnvFormat] = typer.Option(
        None,
        "-f",
        "--format",
        help="Specify format to use. Default json or based on output file name.",
    ),
) -> None:
    """Generate the environment variables needed to use the fetched dependencies."""
    fmt = fmt or (EnvFormat.based_on_suffix(output) if output else EnvFormat.json)
    for_output_dir = for_output_dir or from_output_dir
    fetch_deps_output = _get_build_config(from_output_dir)

    env_file_content = generate_envfile(fetch_deps_output, fmt, for_output_dir)

    if output:
        with output.open("w") as f:
            print(env_file_content, file=f)
    else:
        print(env_file_content)


@app.command()
@handle_errors
def inject_files(
    from_output_dir: Path = FROM_OUTPUT_DIR_ARG,
    for_output_dir: Optional[Path] = FOR_OUTPUT_DIR_OPTION,
) -> None:
    """Inject the project files needed to use the fetched dependencies."""
    for_output_dir = for_output_dir or from_output_dir
    fetch_deps_output = _get_build_config(from_output_dir)

    for project_file in fetch_deps_output.project_files:
        if project_file.abspath.exists():
            log.info("Overwriting %s", project_file.abspath)
        else:
            log.info("Creating %s", project_file.abspath)
            project_file.abspath.parent.mkdir(exist_ok=True, parents=True)

        content = project_file.resolve_content(output_dir=for_output_dir)
        project_file.abspath.write_text(content)

    inject_files_post(
        from_output_dir=from_output_dir,
        for_output_dir=for_output_dir,
        options=fetch_deps_output.options,
    )


def _prevalidate_sbom_files_args(sbom_files_to_merge: Paths) -> Paths:
    def enough_files_for_merge(sbom_files_to_merge: Paths) -> Paths:
        if len(sbom_files_to_merge) < 2:
            # NOTE: an exception here happens during argument evaluation phase
            # i.e. outside of handle_errors() decorator. Simply raising
            # an exception here will not produce correct exit code, thus
            # the explicit call to exception wrapper.
            _bail_out_with_error(InvalidInput("Need at least two different SBOM files"))
        return sbom_files_to_merge

    def all_files_are_jsons(sbom_files_to_merge: Paths) -> Paths:
        for sbom_file in sbom_files_to_merge:
            try:
                json.loads(sbom_file.read_text())
            except ValueError:
                # See comment in enough_files_for_merge()
                _bail_out_with_error(
                    UnexpectedFormat(f"{sbom_file} does not look like a SBOM file")
                )
        return sbom_files_to_merge

    return all_files_are_jsons(enough_files_for_merge(list(set(sbom_files_to_merge))))


@app.command()
@handle_errors
def merge_sboms(
    sbom_files_to_merge: Paths = typer.Argument(
        ...,
        callback=_prevalidate_sbom_files_args,
        exists=True,
        file_okay=True,
        dir_okay=False,
        resolve_path=True,
        readable=True,
        help="Names of files with SBOMs to merge.",
    ),
    output_sbom_file_name: Optional[Path] = OUTFILE_OPTION,
) -> None:
    """Merge two or more SBOMs into one.

    The command works with Cachi2-generated SBOMs only. You might want to run

    cachi2 fetch-deps <args...>

    first to produce SBOMs to merge.
    """
    sboms_to_merge = []
    for sbom_file in sbom_files_to_merge:
        try:
            sboms_to_merge.append(Sbom.model_validate_json(sbom_file.read_text()))
        except pydantic.ValidationError:
            raise UnexpectedFormat(f"{sbom_file} does not appear to be a valid Cachi2 SBOM.")
    sbom = Sbom(
        components=merge_component_properties(
            chain.from_iterable(s.components for s in sboms_to_merge)
        )
    )
    sbom_json = sbom.model_dump_json(indent=2, by_alias=True, exclude_none=True)

    if output_sbom_file_name is not None:
        output_sbom_file_name.write_text(sbom_json)
    else:
        print(sbom_json)


def _get_build_config(output_dir: Path) -> BuildConfig:
    build_config_json = RootedPath(output_dir).join_within_root(".build-config.json").path
    if not build_config_json.exists():
        raise InvalidInput(
            f"No .build-config.json found in {output_dir}. "
            "Please use a directory populated by a previous fetch-deps command."
        )
    return BuildConfig.model_validate_json(build_config_json.read_text())

1	import enum	4✔
2	import functools	4✔
3	import importlib.metadata	4✔
4	import json	4✔
5	import logging	4✔
6	import shutil	4✔
7	import sys	4✔
8	from itertools import chain	4✔
9	from pathlib import Path	4✔
10	from typing import Any, Callable, Optional, Union	4✔
11
12	import pydantic	4✔
13	import typer	4✔
14
15	import cachi2.core.config as config	4✔
16	from cachi2.core.errors import Cachi2Error, InvalidInput, UnexpectedFormat	4✔
17	from cachi2.core.extras.envfile import EnvFormat, generate_envfile	4✔
18	from cachi2.core.models.input import Flag, PackageInput, Request, parse_user_input	4✔
19	from cachi2.core.models.output import BuildConfig	4✔
20	from cachi2.core.models.property_semantics import merge_component_properties	4✔
21	from cachi2.core.models.sbom import Sbom, SPDXSbom	4✔
22	from cachi2.core.resolver import inject_files_post, resolve_packages, supported_package_managers	4✔
23	from cachi2.core.rooted_path import RootedPath	4✔
24	from cachi2.interface.logging import LogLevel, setup_logging	4✔
25
26	app = typer.Typer()	4✔
27	log = logging.getLogger(__name__)	4✔
28
29	DEFAULT_SOURCE = "."	4✔
30	DEFAULT_OUTPUT = "./cachi2-output"	4✔
31
32
33	OUTFILE_OPTION = typer.Option(	4✔
34	None,
35	"-o",
36	"--output",
37	dir_okay=False,
38	help="Write to this file instead of standard output.",
39	)
40
41
42	Paths = list[Path]	4✔
43
44
45	def _bail_out_with_error(e: Cachi2Error) -> None:	4✔
46	"""Report and error and set correct exit code."""
47	log.error("%s: %s", type(e).__name__, str(e).replace("\n", r"\n"))	4✔
48	print(f"Error: {type(e).__name__}: {e.friendly_msg()}", file=sys.stderr)	4✔
49	raise typer.Exit(2 if e.is_invalid_usage else 1)	4✔
50
51
52	def handle_errors(cmd: Callable[..., None]) -> Callable[..., None]:	4✔
53	"""Decorate a CLI command function with an error handler.
54
55	All errors will be logged at ERROR level before exiting.
56	Expected errors will be printed in a friendlier format rather than showing the whole traceback.
57	Errors that we consider invalid usage will result in exit code 2.
58	"""
59
60	def log_error(error: Exception) -> None:	4✔
UNCOV 61	log.error("%s: %s", type(error).__name__, str(error).replace("\n", r"\n"))	×
62
63	@functools.wraps(cmd)	4✔
64	def cmd_with_error_handling(args: tuple[Any, ...], *kwargs: dict[str, Any]) -> None:	4✔
65	try:	4✔
66	cmd(args, *kwargs)	4✔
67	except Cachi2Error as e:	4✔
68	_bail_out_with_error(e)	4✔
UNCOV 69	except Exception as e:	×
UNCOV 70	log_error(e)	×
UNCOV 71	raise	×
72
73	return cmd_with_error_handling	4✔
74
75
76	def version_callback(value: bool) -> None:	4✔
77	"""If --version was used, print the cachi2 version and exit."""
78	if not value:	4✔
79	return	4✔
80
81	print("cachi2", importlib.metadata.version("cachi2"))	4✔
82	print("Supported package managers:", ", ".join(supported_package_managers))	4✔
83	raise typer.Exit()	4✔
84
85
86	class SBOMFormat(str, enum.Enum):	4✔
87	"""The type of SBOM to generate."""
88
89	cyclonedx = "cyclonedx"	4✔
90	spdx = "spdx"	4✔
91
92
93	@app.callback()	4✔
94	@handle_errors	4✔
95	def cachi2( # noqa: D103; docstring becomes part of --help message	4✔
96	version: bool = typer.Option(
97	False,
98	"--version",
99	callback=version_callback,
100	is_eager=True,
101	help="Show version and exit.",
102	),
103	config_file: Path = typer.Option(
104	None,
105	"--config-file",
106	help="Read configuration from this file.",
107	dir_okay=False,
108	exists=True,
109	resolve_path=True,
110	readable=True,
111	),
112	log_level: LogLevel = typer.Option(
113	LogLevel.INFO.value,
114	"--log-level",
115	case_sensitive=False,
116	help="Set log level.",
117	),
118	) -> None:
119	setup_logging(log_level)	4✔
120	if config_file:	4✔
121	config.set_config(config_file)	4✔
122
123
124	def _if_json_then_validate(value: str) -> str:	4✔
125	if _looks_like_json(value):	4✔
126	try:	4✔
127	json.loads(value)	4✔
128	except json.JSONDecodeError:	4✔
129	raise typer.BadParameter(f"Looks like JSON but is not valid JSON: {value!r}")	4✔
130	return value	4✔
131
132
133	def _looks_like_json(value: str) -> bool:	4✔
134	return value.lstrip().startswith(("{", "["))	4✔
135
136
137	class _Input(pydantic.BaseModel, extra="forbid"):	4✔
138	packages: list[PackageInput]	4✔
139	flags: list[Flag] = list()	4✔
140
141
142	@app.command()	4✔
143	@handle_errors	4✔
144	def fetch_deps(	4✔
145	raw_input: str = typer.Argument(
146	...,
147	help="Specify package (within the source repo) to process. See usage examples.",
148	metavar="PKG",
149	callback=_if_json_then_validate,
150	),
151	source: Path = typer.Option(
152	DEFAULT_SOURCE,
153	exists=True,
154	file_okay=False,
155	resolve_path=True,
156	help="Process the git repository at this path.",
157	),
158	output: Path = typer.Option(
159	DEFAULT_OUTPUT,
160	file_okay=False,
161	resolve_path=True,
162	help="Write output files to this directory.",
163	),
164	dev_package_managers: bool = typer.Option(False, "--dev-package-managers", hidden=True),
165	cgo_disable: bool = typer.Option(
166	False, "--cgo-disable", help="Set CGO_ENABLED=0 while processing gomod packages."
167	),
168	force_gomod_tidy: bool = typer.Option(
169	False,
170	"--force-gomod-tidy",
171	help="Run 'go mod tidy' after downloading go dependencies.",
172	),
173	gomod_vendor: bool = typer.Option(
174	False,
175	"--gomod-vendor",
176	help=(
177	"Fetch go deps via 'go mod vendor' rather than 'go mod download'. If you "
178	"have a vendor/ dir, one of --gomod-vendor/--gomod-vendor-check is required."
179	),
180	),
181	gomod_vendor_check: bool = typer.Option(
182	False,
183	"--gomod-vendor-check",
184	help=(
185	"Same as gomod-vendor, but will not make unexpected changes if you "
186	"already have a vendor/ directory (will fail if changes would be made)."
187	),
188	),
189	sbom_output_type: SBOMFormat = typer.Option(
190	SBOMFormat.cyclonedx,
191	"--sbom-type",
192	help=("Format of generated SBOM. Default is CycloneDX"),
193	),
194	) -> None:
195	"""Fetch dependencies for supported package managers.
196
197	\b
198	# gomod package in the current directory
199	cachi2 fetch-deps gomod
200
201	\b
202	# pip package in the root of the source directory
203	cachi2 fetch-deps --source ./my-repo pip
204
205	\b
206	# gomod package in a subpath of the source directory (./my-repo/subpath)
207	cachi2 fetch-deps --source ./my-repo '{
208	"type": "gomod",
209	"path": "subpath"
210	}'
211
212	\b
213	# multiple packages as a JSON list
214	cachi2 fetch-deps '[
215	{"type": "gomod"},
216	{"type": "gomod", "path": "subpath"},
217	{"type": "pip", "path": "other-path"}
218	]'
219
220	\b
221	# multiple packages and flags as a JSON list
222	cachi2 fetch-deps '{
223	"packages": [
224	{"type": "gomod"},
225	{"type": "gomod", "path": "subpath"},
226	{"type": "pip", "path": "other-path"}
227	],
228	"flags": [
229	"gomod-vendor"
230	]
231	}'
232	""" # noqa: D301, D202; backslashes intentional, blank line required by black
233
234	def normalize_input() -> dict[str, list[Any]]:	4✔
235	"""Format raw_input so it can be parsed by the _Input class."""
236	if _looks_like_json(raw_input):	4✔
237	parsed_input = json.loads(raw_input)	4✔
238
239	if isinstance(parsed_input, dict):	4✔
240	if "packages" in parsed_input.keys():	4✔
241	# is a dict with list of packages and possibly flags
242	return parsed_input	4✔
243	else:
244	# is a dict representing a package
245	return {"packages": [parsed_input]}	4✔
246	else:
247	# is a list
248	return {"packages": parsed_input}	4✔
249	else:
250	# is a str
251	return {"packages": [{"type": raw_input}]}	4✔
252
253	def combine_option_and_json_flags(json_flags: list[Flag]) -> list[str]:	4✔
254	flag_names = [	4✔
255	"cgo-disable",
256	"dev-package-managers",
257	"force-gomod-tidy",
258	"gomod-vendor",
259	"gomod-vendor-check",
260	]
261	flag_values = [	4✔
262	cgo_disable,
263	dev_package_managers,
264	force_gomod_tidy,
265	gomod_vendor,
266	gomod_vendor_check,
267	]
268	flags = [name for name, value in zip(flag_names, flag_values) if value]	4✔
269
270	if json_flags:	4✔
271	flags.extend(flag.strip() for flag in json_flags)	4✔
272
273	return flags	4✔
274
275	input = parse_user_input(_Input.model_validate, normalize_input())	4✔
276
277	request = parse_user_input(	4✔
278	Request.model_validate,
279	{
280	"source_dir": source,
281	"output_dir": output,
282	"packages": input.packages,
283	"flags": combine_option_and_json_flags(input.flags),
284	},
285	)
286
287	deps_dir = output / "deps"	4✔
288	if deps_dir.exists():	4✔
289	log.debug(f"Removing existing deps directory '{deps_dir}'")	4✔
290	shutil.rmtree(deps_dir, ignore_errors=True)	4✔
291
292	request_output = resolve_packages(request)	4✔
293
294	request.output_dir.path.mkdir(parents=True, exist_ok=True)	4✔
295	request.output_dir.join_within_root(".build-config.json").path.write_text(	4✔
296	request_output.build_config.model_dump_json(indent=2, exclude_none=True)
297	)
298
299	if sbom_output_type == SBOMFormat.cyclonedx:	4✔
300	sbom: Union[Sbom, SPDXSbom] = request_output.generate_sbom()	4✔
301	else:
302	sbom = request_output.generate_sbom().to_spdx()	4✔
303	request.output_dir.join_within_root("bom.json").path.write_text(	4✔
304	# the Sbom model has camelCase aliases in some fields
305	sbom.model_dump_json(indent=2, by_alias=True, exclude_none=True)
306	)
307
308	log.info(r"All dependencies fetched successfully \o/")	4✔
309
310
311	FROM_OUTPUT_DIR_ARG = typer.Argument(	4✔
312	...,
313	exists=True,
314	file_okay=False,
315	resolve_path=True,
316	help="The output directory populated by a previous fetch-deps command.",
317	)
318	FOR_OUTPUT_DIR_OPTION = typer.Option(	4✔
319	None,
320	resolve_path=True,
321	help="Generate output as if the output directory was at this path instead.",
322	)
323
324
325	@app.command()	4✔
326	@handle_errors	4✔
327	def generate_env(	4✔
328	from_output_dir: Path = FROM_OUTPUT_DIR_ARG,
329	for_output_dir: Optional[Path] = FOR_OUTPUT_DIR_OPTION,
330	output: Optional[Path] = OUTFILE_OPTION,
331	fmt: Optional[EnvFormat] = typer.Option(
332	None,
333	"-f",
334	"--format",
335	help="Specify format to use. Default json or based on output file name.",
336	),
337	) -> None:
338	"""Generate the environment variables needed to use the fetched dependencies."""
339	fmt = fmt or (EnvFormat.based_on_suffix(output) if output else EnvFormat.json)	4✔
340	for_output_dir = for_output_dir or from_output_dir	4✔
341	fetch_deps_output = _get_build_config(from_output_dir)	4✔
342
343	env_file_content = generate_envfile(fetch_deps_output, fmt, for_output_dir)	4✔
344
345	if output:	4✔
346	with output.open("w") as f:	4✔
347	print(env_file_content, file=f)	4✔
348	else:
349	print(env_file_content)	4✔
350
351
352	@app.command()	4✔
353	@handle_errors	4✔
354	def inject_files(	4✔
355	from_output_dir: Path = FROM_OUTPUT_DIR_ARG,
356	for_output_dir: Optional[Path] = FOR_OUTPUT_DIR_OPTION,
357	) -> None:
358	"""Inject the project files needed to use the fetched dependencies."""
359	for_output_dir = for_output_dir or from_output_dir	4✔
360	fetch_deps_output = _get_build_config(from_output_dir)	4✔
361
362	for project_file in fetch_deps_output.project_files:	4✔
363	if project_file.abspath.exists():	4✔
364	log.info("Overwriting %s", project_file.abspath)	4✔
365	else:
366	log.info("Creating %s", project_file.abspath)	4✔
367	project_file.abspath.parent.mkdir(exist_ok=True, parents=True)	4✔
368
369	content = project_file.resolve_content(output_dir=for_output_dir)	4✔
370	project_file.abspath.write_text(content)	4✔
371
372	inject_files_post(	4✔
373	from_output_dir=from_output_dir,
374	for_output_dir=for_output_dir,
375	options=fetch_deps_output.options,
376	)
377
378
379	def _prevalidate_sbom_files_args(sbom_files_to_merge: Paths) -> Paths:	4✔
380	def enough_files_for_merge(sbom_files_to_merge: Paths) -> Paths:	4✔
381	if len(sbom_files_to_merge) < 2:	4✔
382	# NOTE: an exception here happens during argument evaluation phase
383	# i.e. outside of handle_errors() decorator. Simply raising
384	# an exception here will not produce correct exit code, thus
385	# the explicit call to exception wrapper.
386	_bail_out_with_error(InvalidInput("Need at least two different SBOM files"))	4✔
387	return sbom_files_to_merge	4✔
388
389	def all_files_are_jsons(sbom_files_to_merge: Paths) -> Paths:	4✔
390	for sbom_file in sbom_files_to_merge:	4✔
391	try:	4✔
392	json.loads(sbom_file.read_text())	4✔
393	except ValueError:	4✔
394	# See comment in enough_files_for_merge()
395	_bail_out_with_error(	4✔
396	UnexpectedFormat(f"{sbom_file} does not look like a SBOM file")
397	)
398	return sbom_files_to_merge	4✔
399
400	return all_files_are_jsons(enough_files_for_merge(list(set(sbom_files_to_merge))))	4✔
401
402
403	@app.command()	4✔
404	@handle_errors	4✔
405	def merge_sboms(	4✔
406	sbom_files_to_merge: Paths = typer.Argument(
407	...,
408	callback=_prevalidate_sbom_files_args,
409	exists=True,
410	file_okay=True,
411	dir_okay=False,
412	resolve_path=True,
413	readable=True,
414	help="Names of files with SBOMs to merge.",
415	),
416	output_sbom_file_name: Optional[Path] = OUTFILE_OPTION,
417	) -> None:
418	"""Merge two or more SBOMs into one.
419
420	The command works with Cachi2-generated SBOMs only. You might want to run
421
422	cachi2 fetch-deps <args...>
423
424	first to produce SBOMs to merge.
425	"""
426	sboms_to_merge = []	4✔
427	for sbom_file in sbom_files_to_merge:	4✔
428	try:	4✔
429	sboms_to_merge.append(Sbom.model_validate_json(sbom_file.read_text()))	4✔
430	except pydantic.ValidationError:	4✔
431	raise UnexpectedFormat(f"{sbom_file} does not appear to be a valid Cachi2 SBOM.")	4✔
432	sbom = Sbom(	4✔
433	components=merge_component_properties(
434	chain.from_iterable(s.components for s in sboms_to_merge)
435	)
436	)
437	sbom_json = sbom.model_dump_json(indent=2, by_alias=True, exclude_none=True)	4✔
438
439	if output_sbom_file_name is not None:	4✔
440	output_sbom_file_name.write_text(sbom_json)	4✔
441	else:
442	print(sbom_json)	4✔
443
444
445	def _get_build_config(output_dir: Path) -> BuildConfig:	4✔
446	build_config_json = RootedPath(output_dir).join_within_root(".build-config.json").path	4✔
447	if not build_config_json.exists():	4✔
NEW 448	raise InvalidInput(	×
449	f"No .build-config.json found in {output_dir}. "
450	"Please use a directory populated by a previous fetch-deps command."
451	)
452	return BuildConfig.model_validate_json(build_config_json.read_text())	4✔

containerbuildsystem / cachi2 / 10577080486

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