import subprocess
import base64
import glob
import os
import random
import click
from ruamel.yaml import YAML
from pcvs import PATH_INSTDIR, io
from pcvs.backend import config
from pcvs.helpers import git, system, utils
from pcvs.helpers.exceptions import (ConfigException, ProfileException,
ValidationException)
from pcvs.helpers.system import MetaDict
PROFILE_EXISTING = dict()
[docs]def init():
"""Initialization callback, loading available profiles on disk.
"""
global PROFILE_EXISTING
PROFILE_EXISTING = {}
# this first loop defines configuration order
priority_paths = utils.storage_order()
priority_paths.reverse()
for token in priority_paths: # reverse order (overriding)
PROFILE_EXISTING[token] = []
for pfile in glob.glob(os.path.join(utils.STORAGES[token], 'profile', "*.yml")):
PROFILE_EXISTING[token].append(
(os.path.basename(pfile)[:-4], pfile))
[docs]def list_profiles(scope=None):
"""Return a list of valid profiles found on disk.
:param scope: restriction on scope, defaults to None
:type scope: str, optional
:return: dict of 3 dicts ('user', 'local' & 'global') or a single dict (if
'scope' was set), containing, for each profile name, the filepath.
:rtype: dict
"""
global PROFILE_EXISTING
assert (scope in utils.STORAGES.keys() or scope is None)
if scope is None:
return PROFILE_EXISTING
else:
return PROFILE_EXISTING[scope]
[docs]def list_templates():
"""List available templates to be used for boostraping profiles.
:return: a list of valid templates.
:rtype: list"""
array = list()
for f in os.listdir(os.path.join(PATH_INSTDIR, "templates", "profile")):
array.append((os.path.splitext(f)[0], f))
return array
[docs]class Profile:
""" A profile represents the most complete object the user can provide.
It is built upon 5 components, called configuration blocks (or basic
blocks), one of each kind (compiler, runtime, machine, criterion & group)
and gathers all required information to start a validation process. A
profile object is the basic representation to be manipulated by the user.
.. note::
A profile object can be confused with
:class:`pcvs.helpers.system.MetaConfig`. While both are carrying the
whole user configuration, a Profile object is used to build/manipulate
it, while a Metaconfig is the actual internal representation of a
complete run config.
:param _name: profile name, should be unique for a given scope
:type _name: str
:param _scope: profile scope, allowed values in `storage_order()`, defaults
to None
:type _scope: str
:param _exists: return True if the profile exists on disk.
:type _exists: bool
:param _file: profile file absolute path
:type _file: str
"""
def __init__(self, name, scope=None):
"""Constructor method.
:param name: profile name
:type name: str
:param scope: desired scope, automatically set if not provided
:type scope: str, optional
"""
utils.check_valid_scope(scope)
self._name = name
self._scope = scope
self._details = MetaDict()
self._exists = False
self._file = None
self._retrieve_file()
def _retrieve_file(self):
"""From current representation, determine the profile file path.
This function relies on known profiles & path concatenation.
"""
self._file = None
# determine proper scope is not given
if self._scope is None:
allowed_scopes = utils.storage_order()
else:
allowed_scopes = [self._scope]
# available profiles lookup to find if it exist.
for sc in allowed_scopes:
for pair in PROFILE_EXISTING[sc]:
if self._name == pair[0]:
self._file = pair[1]
self._scope = sc
self._exists = True
return
# case where the scope were not provided
# AND no pre-existing profile were found. assume scope as 'local'
if self._scope is None:
self._scope = 'local'
# this code is executed ONLY when a new profile is created
# otherwise the for loop above would have trigger a profile
# in that case, _file is computed through path concatenation
# but the _exists is set to False
self._file = os.path.join(
utils.STORAGES[self._scope], 'profile', self._name + ".yml")
self._exists = False
[docs] def get_unique_id(self):
"""Compute unique hash string identifying a profile.
This is required to make distinction between multiple profiles, based on
its content (banks relies on such unicity).
:return: an hashed version of profile content
:rtype: str
"""
return git.generate_data_hash(str(self._details))
[docs] def fill(self, raw):
"""Update the given profile with content stored in parameter.
:param raw: tree of (key, values) pairs to update
:type raw: dict
"""
# some checks
assert (isinstance(raw, dict))
# fill is called either from 'build' (dict of configurationBlock)
# of from 'clone' (dict of raw file inputs)
for k, v in raw.items():
if isinstance(v, config.ConfigurationBlock):
self._details[k] = MetaDict(v.dump())
else:
self._details[k] = v
[docs] def dump(self):
"""Return the full profile content as a single regular dict.
This function loads the profile on disk first.
:return: a regular dict for this profile
:rtype: dict
"""
# self.load_from_disk()
return MetaDict(self._details).to_dict()
[docs] def is_found(self):
"""Check if the current profile exists on disk.
:return: True if the file exist on disk
:rtype: bool
"""
return self._exists
@property
def scope(self):
"""Return the profile scope.
:return: profile scope
:rtype: str
"""
return self._scope
@property
def full_name(self):
"""Return fully-qualified profile name (scope + name).
:return: the unique profile name.
:rtype: str
"""
return ".".join([self._scope, self._name])
[docs] def load_from_disk(self):
"""Load the profile from its representation on disk.
:raises NotFoundError: profile does not exist
:raises NotFoundError: profile path is not valid
"""
if not self._exists:
raise ProfileException.NotFoundError(self._name)
self._retrieve_file()
if not os.path.isfile(self._file):
raise ProfileException.NotFoundError(self._file)
io.console.debug("Load {} ({})".format(self._name, self._scope))
with open(self._file) as f:
self._details = MetaDict(YAML(typ='safe').load(f))
[docs] def load_template(self, name="default"):
"""Populate the profile from templates of 5 basic config. blocks.
Filepath still need to be determined via `retrieve_file()` call.
"""
self._exists = True
self._file = None
filepath = os.path.join(
PATH_INSTDIR, "templates", "profile", name)+".yml"
if not os.path.isfile(filepath):
raise ProfileException.NotFoundError(
"{} is not a valid base name.\nPlease use pcvs profile list --all".format(name))
with open(filepath, "r") as fh:
self.fill(YAML(typ='safe').load(fh))
[docs] def check(self, allow_legacy=True):
"""Ensure profile meets scheme requirements, as a concatenation of 5
configuration block schemes.
:raises FormatError: A 'kind' is missing from
profile OR incorrect profile.
"""
try:
err_dbg = list()
for k in self._details.keys():
if k not in config.CONFIG_BLOCKS:
err_dbg.append(k)
if err_dbg:
raise ValidationException.FormatError("Unknown kind in Profile", invalid_kinds=err_dbg)
for kind in config.CONFIG_BLOCKS:
# if kind not in self._details:
# raise ValidationException.FormatError(
# "Missing '{}' in profile".format(kind))
system.ValidationScheme(kind).validate(
self._details[kind], filepath=self._name)
except ValidationException.FormatError as e:
if not allow_legacy:
raise e
# Is the profile a legacy format ?
# Attempt to convert it on the fly
proc = subprocess.Popen(
"pcvs_convert '{}' --stdout -k profile --skip-unknown".format(
self._file),
stderr=subprocess.DEVNULL,
stdout=subprocess.PIPE,
shell=True)
fds = proc.communicate()
if proc.returncode != 0:
raise e
converted_data = YAML(typ='safe').load(fds[0].decode('utf-8'))
self.fill(converted_data)
self.check(allow_legacy=False)
io.console.warning("Legacy format for profile '{}'".format(self._name))
io.console.warning("Please consider updating it with `pcvs_convert -k profile`")
[docs] def flush_to_disk(self):
"""Write down profile to disk.
Also, ensure the filepath is valid and profile content is compliant with
schemes.
"""
self._retrieve_file()
#self.check()
# just in case the block subprefix does not exist yet
prefix_file = os.path.dirname(self._file)
if not os.path.isdir(prefix_file):
os.makedirs(prefix_file, exist_ok=True)
with open(self._file, 'w') as f:
YAML(typ='safe').dump(self._details.to_dict(), f)
[docs] def clone(self, clone):
"""Duplicate a valid profile into the current one.
:param clone: a valid profile object
:type clone: :class:`Profile`
"""
self._retrieve_file()
io.console.info("Compute target prefix: {}".format(self._file))
assert (not os.path.isfile(self._file))
self._details = clone._details
[docs] def delete(self):
"""Remove the current profile from disk.
It does not destroy the Python object, though.
"""
io.console.info("delete {}".format(self._file))
os.remove(self._file)
[docs] def display(self):
"""Display profile data into stdout/file.
"""
io.console.print_header("Profile View")
io.console.print_section("Scope: {}".format(self._scope.capitalize()))
io.console.print_section("Profile details:")
if self._details:
io.console.print_section("Details:")
for k, v in self._details.items():
io.console.print_item("{}: {}".format(k, v))
[docs] def edit(self):
"""Open the editor to manipulate profile content.
:param e: an editor program to use instead of defaults
:type e: str
:raises Exception: Something happened while editing the file
.. warning::
If the edition failed (validation failed) a rejected file is created
in the current directory containing the rejected profile. Once
manually edited, it may be submitted again through `pcvs profile
import`.
"""
assert (self._file is not None)
if not os.path.exists(self._file):
return
with open(self._file, 'r') as fh:
stream = fh.read()
edited_stream = click.edit(
stream, extension=".yml", require_save=True)
if edited_stream is not None:
edited_yaml = MetaDict(YAML(typ='safe').load(edited_stream))
self.fill(edited_yaml)
self.flush_to_disk()
try:
self.check()
except Exception as e:
raise e
[docs] def edit_plugin(self):
"""Edit the 'runtime.plugin' section of the current profile.
:param e: an editor program to use instead of defaults
:type e: str
:raises Exception: Something happened while editing the file.
.. warning::
If the edition failed (validation failed) a rejected file is created
in the current directory containing the rejected profile. Once
manually edited, it may be submitted again through `pcvs profile
import`.
"""
if not os.path.exists(self._file):
return
self.load_from_disk()
if 'plugin' in self._details['runtime'].keys():
plugin_code = base64.b64decode(
self._details['runtime']['plugin']).decode('utf-8')
else:
plugin_code = """import math
from pcvs.plugins import Plugin
class MyPlugin(Plugin):
step = Plugin.Step.TEST_EVAL
def run(self, *args, **kwargs):
# this dict maps keys (it name) with values (it value)
# returns True if the combination should be used
return True
"""
try:
edited_code = click.edit(
plugin_code, extension=".py", require_save=True)
if edited_code is not None:
self._details['runtime']['plugin'] = base64.b64encode(
edited_code.encode('utf-8'))
self.flush_to_disk()
except Exception as e:
raise e
[docs] def split_into_configs(self, prefix, blocklist, scope=None):
"""Convert the given profile into a list of basic blocks.
This is the reverse operation of creating a profile (not the 'opposite').
:param prefix: common prefix name used to name basic blocks.
:type prefix: str
:param blocklist: list of config.blocks to generate (all 5 by default
but can be retrained)
:type blocklist: list
:param scope: config block scope, defaults to None
:type scope: str, optional
:raises AlreadyExistError: the created configuration
block name already exist
:return: list of created :class:`ConfigurationBlock`
:rtype: list
"""
objs = list()
if 'all' in blocklist:
blocklist = config.CONFIG_BLOCKS
if scope is None:
scope = self._scope
for name in blocklist:
c = config.ConfigurationBlock(name, prefix, scope)
if c.is_found():
raise ConfigException.AlreadyExistError(c.full_name)
else:
c.fill(self._details[name])
objs.append(c)
return objs
@property
def compiler(self):
"""Access the 'compiler' section.
:return: the 'compiler' dict segment
:rtype: dict
"""
return self._details['compiler']
@property
def runtime(self):
"""Access the 'runtime' section.
:return: the 'runtime' dict segment
:rtype: dict
"""
return self._details['runtime']
@property
def criterion(self):
"""Access the 'criterion' section.
:return: the 'criterion' dict segment
:rtype: dict
"""
return self._details['criterion']
@property
def group(self):
"""Access the 'group' section.
:return: the 'group' dict segment
:rtype: dict
"""
return self._details['group']
@property
def machine(self):
"""Access the 'machine' section.
:return: the 'machine' dict segment
:rtype: dict
"""
return self._details['machine']