zhangyiss/spack
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

reporters wip: working for installs

Browse Source 
Signed-off-by: Gregory Becker <becker33@llnl.gov>
This commit is contained in:
Gregory Becker 2025-04-09 14:40:31 -07:00
parent a2441f4656
commit 559ace64e1
No known key found for this signature in database
GPG Key ID: 2362541F6D14ED84
8 changed files with 223 additions and 296 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

41
lib/spack/spack/cmd/install.py
View File

@ -331,16 +331,8 @@ def install(parser, args):
arguments.sanitize_reporter_options(args)
def reporter_factory(specs):
if args.log_format is None:
return lang.nullcontext()
return spack.report.build_context_manager(
reporter=args.reporter(), filename=report_filename(args, specs=specs), specs=specs
)
reporter = args.reporter() if args.log_format else None
install_kwargs = install_kwargs_from_args(args)
env = ev.active_environment()
if not env and not args.spec and not args.specfiles:
@ -348,9 +340,9 @@ def reporter_factory(specs):
try:
if env:
install_with_active_env(env, args, install_kwargs, reporter_factory)
install_with_active_env(env, args, install_kwargs, reporter)
else:
install_without_active_env(args, install_kwargs, reporter_factory)
install_without_active_env(args, install_kwargs, reporter)
except InstallError as e:
if args.show_log_on_error:
_dump_log_on_error(e)
@ -384,7 +376,7 @@ def _maybe_add_and_concretize(args, env, specs):
env.write(regenerate=False)
def install_with_active_env(env: ev.Environment, args, install_kwargs, reporter_factory):
def install_with_active_env(env: ev.Environment, args, install_kwargs, reporter):
specs = spack.cmd.parse_specs(args.spec)
# The following two commands are equivalent:
@ -418,8 +410,10 @@ def install_with_active_env(env: ev.Environment, args, install_kwargs, reporter_
install_kwargs["overwrite"] = [spec.dag_hash() for spec in specs_to_install]
try:
with reporter_factory(specs_to_install):
env.install_specs(specs_to_install, **install_kwargs)
report_file = report_filename(args, specs_to_install)
install_kwargs["report_file"] = report_file
install_kwargs["reporter"] = reporter
env.install_specs(specs_to_install, **install_kwargs)
finally:
if env.views:
with env.write_transaction():
@ -463,18 +457,23 @@ def concrete_specs_from_file(args):
return result
def install_without_active_env(args, install_kwargs, reporter_factory):
def install_without_active_env(args, install_kwargs, reporter):
concrete_specs = concrete_specs_from_cli(args, install_kwargs) + concrete_specs_from_file(args)
if len(concrete_specs) == 0:
tty.die("The `spack install` command requires a spec to install.")
with reporter_factory(concrete_specs):
if args.overwrite:
require_user_confirmation_for_overwrite(concrete_specs, args)
install_kwargs["overwrite"] = [spec.dag_hash() for spec in concrete_specs]
if args.overwrite:
require_user_confirmation_for_overwrite(concrete_specs, args)
install_kwargs["overwrite"] = [spec.dag_hash() for spec in concrete_specs]
installs = [s.package for s in concrete_specs]
install_kwargs["explicit"] = [s.dag_hash() for s in concrete_specs]
installs = [s.package for s in concrete_specs]
install_kwargs["explicit"] = [s.dag_hash() for s in concrete_specs]
try:
builder = PackageInstaller(installs, **install_kwargs)
builder.install()
finally:
if reporter:
report_file = report_filename(args, concrete_specs)
reporter.build_report(report_file, list(builder.reports.values()))

2
lib/spack/spack/directory_layout.py
View File

@ -284,7 +284,7 @@ def remove_install_directory(self, spec: "spack.spec.Spec", deprecated: bool = F
Raised RemoveFailedError if something goes wrong.
"""
path = self.path_for_spec(spec)
assert path.startswith(self.root)
assert path.startswith(self.root), f"PATH: {path}, ROOT: {self.root}"
if deprecated:
if os.path.exists(path):

11
lib/spack/spack/environment/environment.py
View File

@ -1906,6 +1906,10 @@ def install_specs(self, specs: Optional[List[Spec]] = None, **install_args):
roots = self.concrete_roots()
specs = specs if specs is not None else roots
# Extract reporter arguments
reporter = install_args.pop("reporter", None)
report_file = install_args.pop("report_file", None)
# Extend the set of specs to overwrite with modified dev specs and their parents
install_args["overwrite"] = {
*install_args.get("overwrite", ()),
@ -1918,7 +1922,12 @@ def install_specs(self, specs: Optional[List[Spec]] = None, **install_args):
*(s.dag_hash() for s in roots),
}
PackageInstaller([spec.package for spec in specs], **install_args).install()
try:
builder = PackageInstaller([spec.package for spec in specs], **install_args)
builder.install()
finally:
if reporter:
reporter.build_report(report_file, list(builder.reports.values()))
def all_specs_generator(self) -> Iterable[Spec]:
"""Returns a generator for all concrete specs"""

47
lib/spack/spack/installer.py
View File

@ -60,6 +60,7 @@
import spack.package_base
import spack.package_prefs as prefs
import spack.repo
import spack.report
import spack.rewiring
import spack.spec
import spack.store
@ -923,6 +924,9 @@ def __init__(
raise TypeError(f"{request} is not a valid build request")
self.request = request
# Report for tracking install success/failure
self.record = spack.report.InstallRecord(self.pkg.spec)
# Initialize the status to an active state. The status is used to
# ensure priority queue invariants when tasks are "removed" from the
# queue.
@ -1236,6 +1240,8 @@ def start(self):
Otherwise, start a process for of the requested spec and/or
dependency represented by the BuildTask."""
self.record.start()
if self.install_action == InstallAction.OVERWRITE:
self.tmpdir = tempfile.mkdtemp(dir=os.path.dirname(self.pkg.prefix), prefix=".backup")
self.backup_dir = os.path.join(self.tmpdir, "backup")
@ -1249,6 +1255,9 @@ def start(self):
pkg, pkg_id = self.pkg, self.pkg_id
self.start_time = self.start_time or time.time()
tests = install_args.get("tests")
pkg.run_tests = tests is True or tests and pkg.name in tests
# Use the binary cache to install if requested,
# save result to be handled in BuildTask.complete()
if self.use_cache:
@ -1289,12 +1298,16 @@ def poll(self):
return self.no_op or self.success_result or self.error_result or self.process_handle.poll()
def succeed(self):
self.record.succeed()
# delete the temporary backup for an overwrite
# see llnl.util.filesystem.restore_directory_transaction
if self.install_action == InstallAction.OVERWRITE:
shutil.rmtree(self.tmpdir, ignore_errors=True)
def fail(self, inner_exception):
self.record.fail(inner_exception)
if self.install_action != InstallAction.OVERWRITE:
raise inner_exception
@ -1319,12 +1332,9 @@ def complete(self):
), "Can't call `complete()` before `start()` or identified no-operation task"
install_args = self.request.install_args
pkg = self.pkg
tests = install_args.get("tests")
self.status = BuildStatus.INSTALLING
pkg.run_tests = tests is True or tests and pkg.name in tests
# If task has been identified as a no operation,
# return ExecuteResult.NOOP
if self.no_op:
@ -1376,7 +1386,7 @@ class RewireTask(Task):
"""Class for representing a rewire task for a package."""
def start(self):
pass
self.record.start()
def poll(self):
return True
@ -1399,14 +1409,19 @@ def complete(self):
unsigned = install_args.get("unsigned")
_process_binary_cache_tarball(self.pkg, explicit=self.explicit, unsigned=unsigned)
_print_installed_pkg(self.pkg.prefix)
self.record.succeed()
return ExecuteResult.SUCCESS
except BaseException as e:
tty.error(f"Failed to rewire {self.pkg.spec} from binary. {e}")
self.status = oldstatus
return ExecuteResult.MISSING_BUILD_SPEC
spack.rewiring.rewire_node(self.pkg.spec, self.explicit)
_print_installed_pkg(self.pkg.prefix)
return ExecuteResult.SUCCESS
try:
spack.rewiring.rewire_node(self.pkg.spec, self.explicit)
_print_installed_pkg(self.pkg.prefix)
self.record.succeed()
return ExecuteResult.SUCCESS
except BaseException as e:
self.record.fail(e)
class PackageInstaller:
@ -1536,6 +1551,14 @@ def __init__(
# Maximum number of concurrent packages to build
self.max_active_tasks = concurrent_packages
# Reports on install success/failure
self.reports: Dict[str, dict] = {}
for build_request in self.build_requests:
# Skip reporting for already installed specs
request_record = spack.report.RequestRecord(build_request.pkg.spec)
request_record.skip_installed()
self.reports[build_request.pkg_id] = request_record
def __repr__(self) -> str:
"""Returns a formal representation of the package installer."""
rep = f"{self.__class__.__name__}("
@ -1923,13 +1946,18 @@ def _complete_task(self, task: Task, install_status: InstallStatus) -> None:
task: the installation task for a package
install_status: the installation status for the package
"""
rc = task.complete()
try:
rc = task.complete()
except BaseException:
self.reports[task.request.pkg_id].append_record(task.record)
raise
if rc == ExecuteResult.MISSING_BUILD_SPEC:
self._requeue_with_build_spec_tasks(task)
elif rc == ExecuteResult.NO_OP:
pass
else: # if rc == ExecuteResult.SUCCESS or rc == ExecuteResult.FAILED
self._update_installed(task)
self.reports[task.request.pkg_id].append_record(task.record)
def _next_is_pri0(self) -> bool:
"""
@ -2179,6 +2207,9 @@ def start_task(
# install_status.set_term_title(f"Processing {task.pkg.name}")
tty.debug(f"Processing {pkg_id}: task={task}")
# Debug
task.record.start()
# Skip the installation if the spec is not being installed locally
# (i.e., if external or upstream) BUT flag it as installed since
# some package likely depends on it.

345
lib/spack/spack/report.py
View File

@ -1,276 +1,131 @@
# Copyright Spack Project Developers. See COPYRIGHT file for details.
#
# SPDX-License-Identifier: (Apache-2.0 OR MIT)
"""Tools to produce reports of spec installations"""
"""Hooks to produce reports of spec installations"""
import collections
import contextlib
import functools
import gzip
import os
import time
import traceback
from typing import Any, Callable, Dict, List, Type
import llnl.util.lang
import llnl.util.filesystem as fs
import spack.build_environment
import spack.install_test
import spack.installer
import spack.package_base
import spack.reporters
import spack.spec
import spack.util.spack_json as sjson
reporter = None
report_file = None
Property = collections.namedtuple("Property", ["name", "value"])
class InfoCollector:
"""Base class for context manager objects that collect information during the execution of
certain package functions.
class Record(dict):
def __getattr__(self, name):
# only called if no attribute exists
if name in self:
return self[name]
raise AttributeError(f"RequestRecord for {self.name} has no attribute {name}")
The data collected is available through the ``specs`` attribute once exited, and it's
organized as a list where each item represents the installation of one spec.
"""
wrap_class: Type
do_fn: str
_backup_do_fn: Callable
input_specs: List[spack.spec.Spec]
specs: List[Dict[str, Any]]
def __init__(self, wrap_class: Type, do_fn: str, specs: List[spack.spec.Spec]):
#: Class for which to wrap a function
self.wrap_class = wrap_class
#: Action to be reported on
self.do_fn = do_fn
#: Backup of the wrapped class function
self._backup_do_fn = getattr(self.wrap_class, do_fn)
#: Specs that will be acted on
self.input_specs = specs
#: This is where we record the data that will be included in our report
self.specs: List[Dict[str, Any]] = []
def fetch_log(self, pkg: spack.package_base.PackageBase) -> str:
"""Return the stdout log associated with the function being monitored
Args:
pkg: package under consideration
"""
raise NotImplementedError("must be implemented by derived classes")
def extract_package_from_signature(self, instance, *args, **kwargs):
"""Return the package instance, given the signature of the wrapped function."""
raise NotImplementedError("must be implemented by derived classes")
def __enter__(self):
# Initialize the spec report with the data that is available upfront.
Property = collections.namedtuple("Property", ["name", "value"])
for input_spec in self.input_specs:
name_fmt = "{0}_{1}"
name = name_fmt.format(input_spec.name, input_spec.dag_hash(length=7))
spec_record = {
"name": name,
"nerrors": None,
"nfailures": None,
"npackages": None,
"time": None,
"timestamp": time.strftime("%a, %d %b %Y %H:%M:%S", time.gmtime()),
"properties": [],
"packages": [],
}
spec_record["properties"].append(Property("architecture", input_spec.architecture))
self.init_spec_record(input_spec, spec_record)
self.specs.append(spec_record)
def gather_info(wrapped_fn):
"""Decorates a function to gather useful information for a CI report."""
@functools.wraps(wrapped_fn)
def wrapper(instance, *args, **kwargs):
pkg = self.extract_package_from_signature(instance, *args, **kwargs)
package = {
"name": pkg.name,
"id": pkg.spec.dag_hash(),
"elapsed_time": None,
"result": None,
"message": None,
"installed_from_binary_cache": False,
}
# Append the package to the correct spec report. In some
# cases it may happen that a spec that is asked to be
# installed explicitly will also be installed as a
# dependency of another spec. In this case append to both
# spec reports.
for current_spec in llnl.util.lang.dedupe([pkg.spec.root, pkg.spec]):
name = name_fmt.format(current_spec.name, current_spec.dag_hash(length=7))
try:
item = next((x for x in self.specs if x["name"] == name))
item["packages"].append(package)
except StopIteration:
pass
start_time = time.time()
try:
value = wrapped_fn(instance, *args, **kwargs)
package["stdout"] = self.fetch_log(pkg)
package["installed_from_binary_cache"] = pkg.installed_from_binary_cache
self.on_success(pkg, kwargs, package)
return value
except spack.build_environment.InstallError as exc:
# An InstallError is considered a failure (the recipe
# didn't work correctly)
package["result"] = "failure"
package["message"] = exc.message or "Installation failure"
package["stdout"] = self.fetch_log(pkg)
package["stdout"] += package["message"]
package["exception"] = exc.traceback
raise
except (Exception, BaseException) as exc:
# Everything else is an error (the installation
# failed outside of the child process)
package["result"] = "error"
package["message"] = str(exc) or "Unknown error"
package["stdout"] = self.fetch_log(pkg)
package["stdout"] += package["message"]
package["exception"] = traceback.format_exc()
raise
finally:
package["elapsed_time"] = time.time() - start_time
return wrapper
setattr(self.wrap_class, self.do_fn, gather_info(getattr(self.wrap_class, self.do_fn)))
def on_success(self, pkg: spack.package_base.PackageBase, kwargs, package_record):
"""Add additional properties on function call success."""
raise NotImplementedError("must be implemented by derived classes")
def init_spec_record(self, input_spec: spack.spec.Spec, record):
"""Add additional entries to a spec record when entering the collection context."""
def __exit__(self, exc_type, exc_val, exc_tb):
# Restore the original method in PackageBase
setattr(self.wrap_class, self.do_fn, self._backup_do_fn)
for spec in self.specs:
spec["npackages"] = len(spec["packages"])
spec["nfailures"] = len([x for x in spec["packages"] if x["result"] == "failure"])
spec["nerrors"] = len([x for x in spec["packages"] if x["result"] == "error"])
spec["time"] = sum(float(x["elapsed_time"]) for x in spec["packages"])
def __setattr__(self, name, value):
if name.startswith("_"):
super().__setattr__(name, value)
else:
self[name] = value
class BuildInfoCollector(InfoCollector):
"""Collect information for the PackageInstaller._install_task method.
class RequestRecord(Record):
def __init__(self, spec):
super().__init__()
self._spec = spec
self.name = spec.name
self.errors = None
self.nfailures = None
self.npackages = None
self.time = None
self.timestamp = time.strftime("%a, d %b %Y %H:%M:%S", time.gmtime())
self.properties = [
Property("architecture", spec.architecture),
# Property("compiler", spec.compiler),
]
self.packages = []
Args:
specs: specs whose install information will be recorded
"""
def skip_installed(self):
for dep in filter(lambda x: x.installed, self._spec.traverse()):
record = InstallRecord(dep)
record.skip(msg="Spec already installed")
self.packages.append(record)
def __init__(self, specs: List[spack.spec.Spec]):
super().__init__(spack.installer.PackageInstaller, "_install_task", specs)
def append_record(self, record):
self.packages.append(record)
def init_spec_record(self, input_spec, record):
# Check which specs are already installed and mark them as skipped
for dep in filter(lambda x: x.installed, input_spec.traverse()):
package = {
"name": dep.name,
"id": dep.dag_hash(),
"elapsed_time": "0.0",
"result": "skipped",
"message": "Spec already installed",
}
record["packages"].append(package)
def summarize(self):
self.npackages = len(self.packages)
self.nfailures = len([r for r in self.packages if r.result == "failure"])
self.nerrors = len([r for r in self.packages if r.result == "error"])
self.time = sum(float(r.elapsed_time or 0.0) for r in self.packages)
def on_success(self, pkg, kwargs, package_record):
package_record["result"] = "success"
def fetch_log(self, pkg):
class SpecRecord(Record):
def __init__(self, spec):
super().__init__()
self._spec = spec
self._package = spec.package
self._start_time = None
self.name = spec.name
self.id = spec.dag_hash()
self.elapsed_time = None
def start(self):
self._start_time = time.time()
def skip(self, msg):
self.result = "skipped"
self.elapsed_time = 0.0
self.message = msg
class InstallRecord(SpecRecord):
def __init__(self, spec):
super().__init__(spec)
self.result = None
self.message = None
self.installed_from_binary_cache = None
def fetch_log(self):
try:
if os.path.exists(pkg.install_log_path):
stream = gzip.open(pkg.install_log_path, "rt", encoding="utf-8")
if os.path.exists(self._package.install_log_path):
stream = gzip.open(self._package.install_log_path, "rt", encoding="utf-8")
else:
stream = open(pkg.log_path, encoding="utf-8")
stream = open(self._package.log_path, encoding="utf-8")
with stream as f:
return f.read()
except OSError:
return f"Cannot open log for {pkg.spec.cshort_spec}"
return f"Cannot open log for {self._spec.cshort_spec}"
def extract_package_from_signature(self, instance, *args, **kwargs):
return args[0].pkg
class TestInfoCollector(InfoCollector):
"""Collect information for the PackageBase.do_test method.
Args:
specs: specs whose install information will be recorded
record_directory: record directory for test log paths
"""
dir: str
def __init__(self, specs: List[spack.spec.Spec], record_directory: str):
super().__init__(spack.package_base.PackageBase, "do_test", specs)
self.dir = record_directory
def on_success(self, pkg, kwargs, package_record):
externals = kwargs.get("externals", False)
skip_externals = pkg.spec.external and not externals
if skip_externals:
package_record["result"] = "skipped"
package_record["result"] = "success"
def fetch_log(self, pkg: spack.package_base.PackageBase):
log_file = os.path.join(self.dir, spack.install_test.TestSuite.test_log_name(pkg.spec))
def fetch_time(self):
try:
with open(log_file, "r", encoding="utf-8") as stream:
return "".join(stream.readlines())
with open(self._package.times_log_path, "r", encoding="utf-8") as f:
data = sjson.load(f.read())
return data["total"]
except Exception:
return f"Cannot open log for {pkg.spec.cshort_spec}"
return None
def extract_package_from_signature(self, instance, *args, **kwargs):
return instance
def succeed(self):
self.result = "success"
self.stdout = self.fetch_log()
self.installed_from_binary_cache = self._package.installed_from_binary_cache
assert self._start_time, "Start time is None"
self.elapsed_time = time.time() - self._start_time
@contextlib.contextmanager
def build_context_manager(
reporter: spack.reporters.Reporter, filename: str, specs: List[spack.spec.Spec]
):
"""Decorate a package to generate a report after the installation function is executed.
Args:
reporter: object that generates the report
filename: filename for the report
specs: specs that need reporting
"""
collector = BuildInfoCollector(specs)
try:
with collector:
yield
finally:
reporter.build_report(filename, specs=collector.specs)
@contextlib.contextmanager
def test_context_manager(
reporter: spack.reporters.Reporter,
filename: str,
specs: List[spack.spec.Spec],
raw_logs_dir: str,
):
"""Decorate a package to generate a report after the test function is executed.
Args:
reporter: object that generates the report
filename: filename for the report
specs: specs that need reporting
raw_logs_dir: record directory for test log paths
"""
collector = TestInfoCollector(specs, raw_logs_dir)
try:
with collector:
yield
finally:
reporter.test_report(filename, specs=collector.specs)
def fail(self, exc):
if isinstance(exc, spack.build_environment.InstallError):
self.result = "failure"
self.message = exc.message or "Installation failure"
self.exception = exc.traceback
else:
self.result = "error"
self.message = str(exc) or "Unknown error"
self.exception = traceback.format_exc()
self.stdout = self.fetch_log() + self.message
assert self._start_time, "Start time is None"
self.elapsed_time = time.time() - self._start_time

4
lib/spack/spack/reporters/cdash.py
View File

@ -278,6 +278,8 @@ def build_report(self, report_dir, specs):
self.multiple_packages = False
num_packages = 0
for spec in specs:
spec.summarize()
# Do not generate reports for packages that were installed
# from the binary cache.
spec["packages"] = [
@ -362,6 +364,8 @@ def test_report(self, report_dir, specs):
"""Generate reports for each package in each spec."""
tty.debug("Processing test report")
for spec in specs:
spec.summarize()
duration = 0
if "time" in spec:
duration = int(spec["time"])

4
lib/spack/spack/reporters/junit.py
View File

@ -17,12 +17,16 @@ def concretization_report(self, filename, msg):
pass
def build_report(self, filename, specs):
for spec in specs:
spec.summarize()
if not (os.path.splitext(filename))[1]:
# Ensure the report name will end with the proper extension;
# otherwise, it currently defaults to the "directory" name.
filename = filename + ".xml"
report_data = {"specs": specs}
with open(filename, "w", encoding="utf-8") as f:
env = spack.tengine.make_environment()
t = env.get_template(self._jinja_template)

65
lib/spack/spack/test/cmd/install.py
View File

@ -66,30 +66,33 @@ def test_install_package_and_dependency(
assert 'errors="0"' in content
def _check_runtests_none(pkg):
assert not pkg.run_tests
def _check_runtests_dttop(pkg):
assert pkg.run_tests == (pkg.name == "dttop")
def _check_runtests_all(pkg):
assert pkg.run_tests
@pytest.mark.disable_clean_stage_check
def test_install_runtests_notests(monkeypatch, mock_packages, install_mockery):
def check(pkg):
assert not pkg.run_tests
monkeypatch.setattr(spack.package_base.PackageBase, "unit_test_check", check)
monkeypatch.setattr(spack.package_base.PackageBase, "unit_test_check", _check_runtests_none)
install("-v", "dttop")
@pytest.mark.disable_clean_stage_check
def test_install_runtests_root(monkeypatch, mock_packages, install_mockery):
def check(pkg):
assert pkg.run_tests == (pkg.name == "dttop")
monkeypatch.setattr(spack.package_base.PackageBase, "unit_test_check", check)
monkeypatch.setattr(spack.package_base.PackageBase, "unit_test_check", _check_runtests_dttop)
install("--test=root", "dttop")
@pytest.mark.disable_clean_stage_check
def test_install_runtests_all(monkeypatch, mock_packages, install_mockery):
def check(pkg):
assert pkg.run_tests
monkeypatch.setattr(spack.package_base.PackageBase, "unit_test_check", check)
monkeypatch.setattr(spack.package_base.PackageBase, "unit_test_check", _check_runtests_all)
install("--test=all", "pkg-a")
@ -377,6 +380,7 @@ def test_install_from_file(spec, concretize, error_code, tmpdir):
def test_junit_output_with_failures(tmpdir, exc_typename, msg):
with tmpdir.as_cwd():
install(
"--verbose",
"--log-format=junit",
"--log-file=test.xml",
"raiser",
@ -409,6 +413,21 @@ def test_junit_output_with_failures(tmpdir, exc_typename, msg):
assert msg in content
def _throw(task, exc_typename, exc_type, msg):
# Self is a spack.installer.Task
exc_type = getattr(builtins, exc_typename)
exc = exc_type(msg)
task.fail(exc)
def _runtime_error(task, *args, **kwargs):
_throw(task, "RuntimeError", spack.error.InstallError, "something weird happened")
def _keyboard_error(task, *args, **kwargs):
_throw(task, "KeyboardInterrupt", KeyboardInterrupt, "Ctrl-C strikes again")
@pytest.mark.disable_clean_stage_check
@pytest.mark.parametrize(
"exc_typename,expected_exc,msg",
@ -428,14 +447,17 @@ def test_junit_output_with_errors(
tmpdir,
monkeypatch,
):
def just_throw(*args, **kwargs):
exc_type = getattr(builtins, exc_typename)
raise exc_type(msg)
monkeypatch.setattr(spack.installer.PackageInstaller, "_install_task", just_throw)
throw = _keyboard_error if expected_exc == KeyboardInterrupt else _runtime_error
monkeypatch.setattr(spack.installer.BuildTask, "complete", throw)
with tmpdir.as_cwd():
install("--log-format=junit", "--log-file=test.xml", "libdwarf", fail_on_error=False)
install(
"--verbose",
"--log-format=junit",
"--log-file=test.xml",
"trivial-install-test-dependent",
fail_on_error=False,
)
assert isinstance(install.error, expected_exc)
@ -445,7 +467,7 @@ def just_throw(*args, **kwargs):
content = filename.open().read()
# Only libelf error is reported (through libdwarf root spec). libdwarf
# Only original error is reported, dependent
# install is skipped and it is not an error.
assert 'tests="0"' not in content
assert 'failures="0"' in content
@ -1079,7 +1101,10 @@ def install_use_buildcache(opt):
@pytest.mark.disable_clean_stage_check
def test_padded_install_runtests_root(install_mockery, mock_fetch):
spack.config.set("config:install_tree:padded_length", 255)
output = install("--test=root", "--no-cache", "test-build-callbacks", fail_on_error=False)
output = install(
"--verbose", "--test=root", "--no-cache", "test-build-callbacks", fail_on_error=False
)
print(output)
assert output.count("method not implemented") == 1