in lib/ramble/spack/report.py [0:0]
def __enter__(self):
# Initialize the spec report with the data that is available upfront.
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 = {
'name': name,
'nerrors': None,
'nfailures': None,
'npackages': None,
'time': None,
'timestamp': time.strftime(
"%a, %d %b %Y %H:%M:%S", time.gmtime()
),
'properties': [],
'packages': []
}
self.specs.append(spec)
Property = collections.namedtuple('Property', ['name', 'value'])
spec['properties'].append(
Property('architecture', input_spec.architecture)
)
spec['properties'].append(
Property('compiler', input_spec.compiler))
# Check which specs are already installed and mark them as skipped
# only for install_task
if self.do_fn == '_install_task':
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'
}
spec['packages'].append(package)
def gather_info(do_fn):
"""Decorates do_fn to gather useful information for
a CI report.
It's defined here to capture the environment and build
this context as the installations proceed.
"""
@functools.wraps(do_fn)
def wrapper(instance, *args, **kwargs):
if isinstance(instance, spack.package.PackageBase):
pkg = instance
elif hasattr(args[0], 'pkg'):
pkg = args[0].pkg
else:
raise Exception
# We accounted before for what is already installed
installed_already = pkg.spec.installed
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 s in llnl.util.lang.dedupe([pkg.spec.root, pkg.spec]):
name = name_fmt.format(s.name, s.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()
value = None
try:
value = do_fn(instance, *args, **kwargs)
externals = kwargs.get('externals', False)
skip_externals = pkg.spec.external and not externals
if do_fn.__name__ == 'do_test' and skip_externals:
package['result'] = 'skipped'
else:
package['result'] = 'success'
package['stdout'] = fetch_log(pkg, do_fn, self.dir)
package['installed_from_binary_cache'] = \
pkg.installed_from_binary_cache
if do_fn.__name__ == '_install_task' and installed_already:
return
except spack.build_environment.InstallError as e:
# An InstallError is considered a failure (the recipe
# didn't work correctly)
package['result'] = 'failure'
package['message'] = e.message or 'Installation failure'
package['stdout'] = fetch_log(pkg, do_fn, self.dir)
package['stdout'] += package['message']
package['exception'] = e.traceback
raise
except (Exception, BaseException) as e:
# Everything else is an error (the installation
# failed outside of the child process)
package['result'] = 'error'
package['stdout'] = fetch_log(pkg, do_fn, self.dir)
package['message'] = str(e) or 'Unknown error'
package['exception'] = traceback.format_exc()
raise
finally:
package['elapsed_time'] = time.time() - start_time
return value
return wrapper
setattr(self.wrap_class, self.do_fn, gather_info(
getattr(self.wrap_class, self.do_fn)
))