# Copyright Spack Project Developers. See COPYRIGHT file for details.

#

# SPDX-License-Identifier: (Apache-2.0 OR MIT)

import collections

import collections.abc

import contextlib

import errno

import glob

import os

import pathlib

import re

import shutil

import stat

import warnings

from collections.abc import KeysView

from typing import (

Any,

Callable,

Dict,

Iterable,

List,

Mapping,

Optional,

Sequence,

Set,

Tuple,

Union,

)

import spack

import spack.config

import spack.deptypes as dt

import spack.error

import spack.filesystem_view as fsv

import spack.hash_types as ht

import spack.installer_dispatch

import spack.llnl.util.filesystem as fs

import spack.llnl.util.tty as tty

import spack.llnl.util.tty.color as clr

import spack.package_base

import spack.paths

import spack.repo

import spack.schema.env

import spack.spec

import spack.store

import spack.user_environment as uenv

import spack.util.environment

import spack.util.hash

import spack.util.lock as lk

import spack.util.path

import spack.util.spack_json as sjson

import spack.util.spack_yaml as syaml

import spack.variant as vt

from spack import traverse

from spack.enums import ConfigScopePriority

from spack.llnl.util.filesystem import copy_tree, islink, readlink, symlink

from spack.llnl.util.lang import stable_partition

from spack.llnl.util.link_tree import ConflictingSpecsError

from spack.schema.env import TOP_LEVEL_KEY

from spack.spec import Spec

from spack.spec_filter import SpecFilter

from spack.util.path import substitute_path_variables

from .list import SpecList, SpecListError, SpecListParser

SpecPair = Tuple[Spec, Spec]

DEFAULT_USER_SPEC_GROUP = "default"

#: environment variable used to indicate the active environment

spack_env_var = "SPACK_ENV"

#: environment variable used to indicate the active environment view

spack_env_view_var = "SPACK_ENV_VIEW"

#: currently activated environment

_active_environment: Optional["Environment"] = None

# This is used in spack.main to bypass env failures if the command is `spack config edit`

# It is used in spack.cmd.config to get the path to a failed env for `spack config edit`

#: Validation error for a currently activate environment that failed to parse

_active_environment_error: Optional[spack.config.ConfigFormatError] = None

#: default path where environments are stored in the spack tree

default_env_path = os.path.join(spack.paths.var_path, "environments")

#: Name of the input yaml file for an environment

manifest_name = "spack.yaml"

#: Name of the input yaml file for an environment

lockfile_name = "spack.lock"

#: Name of the directory where environments store repos, logs, views, configs

env_subdir_name = ".spack-env"

def env_root_path() -> str:

"""Override default root path if the user specified it"""

return spack.util.path.canonicalize_path(

spack.config.get("config:environments_root", default=default_env_path)

)

def environment_name(path: Union[str, pathlib.Path]) -> str:

"""Human-readable representation of the environment.

This is the path for independent environments, and just the name

for managed environments.

"""

env_root = pathlib.Path(env_root_path()).resolve()

path_path = pathlib.Path(path)

# For a managed environment created in Spack, env.path is ENV_ROOT/NAME

# For a tracked environment from `spack env track`, the path is symlinked to ENV_ROOT/NAME

# So if ENV_ROOT/NAME resolves to env.path we know the environment is tracked/managed.

# Otherwise, it is an independent environment and we return the path.

#

# We resolve both paths fully because the env_root itself could also be a symlink,

# and any directory in env.path could be a symlink.

if (env_root / path_path.name).resolve() == path_path.resolve():

return path_path.name

else:

return str(path)

def ensure_no_disallowed_env_config_mods(scope: spack.config.ConfigScope) -> None:

config = scope.get_section("config")

if config and "environments_root" in config["config"]:

raise SpackEnvironmentError(

"Spack environments are prohibited from modifying 'config:environments_root' "

"because it can make the definition of the environment ill-posed. Please "

"remove from your environment and place it in a permanent scope such as "

"defaults, system, site, etc."

)

def default_manifest_yaml():

"""default spack.yaml file to put in new environments"""

return """\

# This is a Spack Environment file.

#

# It describes a set of packages to be installed, along with

# configuration settings.

spack:

# add package specs to the `specs` list

specs: []

view: true

concretizer:

unify: {}

""".format("true" if spack.config.get("concretizer:unify") else "false")

sep_re = re.escape(os.sep)

#: regex for validating environment names

valid_environment_name_re = rf"^\w[{sep_re}\w-]*$"

#: version of the lockfile format. Must increase monotonically.

CURRENT_LOCKFILE_VERSION = 7

READER_CLS = {

1: spack.spec.SpecfileV1,

2: spack.spec.SpecfileV1,

3: spack.spec.SpecfileV2,

4: spack.spec.SpecfileV3,

5: spack.spec.SpecfileV4,

6: spack.spec.SpecfileV5,

7: spack.spec.SpecfileV5,

}

# Magic names

# The name of the standalone spec list in the manifest yaml

USER_SPECS_KEY = "specs"

# The name of the default view (the view loaded on env.activate)

default_view_name = "default"

# Default behavior to link all packages into views (vs. only root packages)

default_view_link = "all"

# (DEPRECATED) Use as the heading/name in the manifest is deprecated.

# The key for any concrete specs included in a lockfile.

lockfile_include_key = "include_concrete"

# The name/heading for include paths in the manifest file.

manifest_include_name = "include"

def installed_specs():

"""

Returns the specs of packages installed in the active environment or None

if no packages are installed.

"""

env = active_environment()

hashes = env.all_hashes() if env else None

return spack.store.STORE.db.query(hashes=hashes)

def valid_env_name(name):

return re.match(valid_environment_name_re, name)

def validate_env_name(name):

if not valid_env_name(name):

raise ValueError(

f"{name}: names may only contain letters, numbers, _, and -, and may not start with -."

)

return name

def activate(env, use_env_repo=False):

"""Activate an environment.

To activate an environment, we add its manifest's configuration scope to the

existing Spack configuration, and we set active to the current environment.

Arguments:

env (Environment): the environment to activate

use_env_repo (bool): use the packages exactly as they appear in the

environment's repository

"""

global _active_environment

try:

_active_environment = env

# Fail early to avoid ending in an invalid state

if not isinstance(env, Environment):

raise TypeError("`env` should be of type {0}".format(Environment.__name__))

# Check if we need to reinitialize spack.store.STORE and spack.repo.REPO due to

# config changes.

install_tree_before = spack.config.get("config:install_tree")

upstreams_before = spack.config.get("upstreams")

repos_before = spack.config.get("repos")

env.manifest.prepare_config_scope()

install_tree_after = spack.config.get("config:install_tree")

upstreams_after = spack.config.get("upstreams")

repos_after = spack.config.get("repos")

if install_tree_before != install_tree_after or upstreams_before != upstreams_after:

setattr(env, "store_token", spack.store.reinitialize())

if repos_before != repos_after:

setattr(env, "repo_token", spack.repo.PATH)

spack.repo.PATH.disable()

new_repo = spack.repo.RepoPath.from_config(spack.config.CONFIG)

if use_env_repo:

new_repo.put_first(env.repo)

spack.repo.enable_repo(new_repo)

tty.debug(f"Using environment '{env.name}'")

except Exception:

_active_environment = None

raise

def deactivate():

"""Undo any configuration or repo settings modified by ``activate()``."""

global _active_environment

if not _active_environment:

return

# If any config changes affected spack.store.STORE or spack.repo.PATH, undo them.

store = getattr(_active_environment, "store_token", None)

if store is not None:

spack.store.restore(store)

delattr(_active_environment, "store_token")

repo = getattr(_active_environment, "repo_token", None)

if repo is not None:

spack.repo.PATH.disable()

spack.repo.enable_repo(repo)

_active_environment.manifest.deactivate_config_scope()

tty.debug(f"Deactivated environment '{_active_environment.name}'")

_active_environment = None

def active_environment() -> Optional["Environment"]:

"""Returns the active environment when there is any"""

return _active_environment

def _root(name):

"""Non-validating version of root(), to be used internally."""

return os.path.join(env_root_path(), name)

def root(name):

"""Get the root directory for an environment by name."""

validate_env_name(name)

return _root(name)

def exists(name):

"""Whether an environment with this name exists or not."""

return valid_env_name(name) and os.path.lexists(os.path.join(_root(name), manifest_name))

def active(name):

"""True if the named environment is active."""

return _active_environment and name == _active_environment.name

def is_env_dir(path):

"""Whether a directory contains a spack environment."""

return os.path.isdir(path) and os.path.exists(os.path.join(path, manifest_name))

def as_env_dir(name_or_dir):

"""Translate an environment name or directory to the environment directory"""

if is_env_dir(name_or_dir):

return name_or_dir

else:

validate_env_name(name_or_dir)

if not exists(name_or_dir):

raise SpackEnvironmentError("no such environment '%s'" % name_or_dir)

return _root(name_or_dir)

def environment_from_name_or_dir(name_or_dir):

"""Get an environment with the supplied name."""

return Environment(as_env_dir(name_or_dir))

def read(name):

"""Get an environment with the supplied name."""

validate_env_name(name)

if not exists(name):

raise SpackEnvironmentError("no such environment '%s'" % name)

return Environment(root(name))

def create(

name: str,

init_file: Optional[Union[str, pathlib.Path]] = None,

with_view: Optional[Union[str, pathlib.Path, bool]] = None,

keep_relative: bool = False,

include_concrete: Optional[List[str]] = None,

) -> "Environment":

"""Create a managed environment in Spack and returns it.

A managed environment is created in a root directory managed by this Spack instance, so that

Spack can keep track of them.

Files with suffix ``.json`` or ``.lock`` are considered lockfiles. Files with any other name

are considered manifest files.

Args:

name: name of the managed environment

init_file: either a lockfile, a manifest file, or None

with_view: whether a view should be maintained for the environment. If the value is a

string, it specifies the path to the view

keep_relative: if True, develop paths are copied verbatim into the new environment file,

otherwise they are made absolute

include_concrete: concrete environment names/paths to be included

"""

environment_dir = environment_dir_from_name(name, exists_ok=False)

return create_in_dir(

environment_dir,

init_file=init_file,

with_view=with_view,

keep_relative=keep_relative,

include_concrete=include_concrete,

)

def create_in_dir(

root: Union[str, pathlib.Path],

init_file: Optional[Union[str, pathlib.Path]] = None,

with_view: Optional[Union[str, pathlib.Path, bool]] = None,

keep_relative: bool = False,

include_concrete: Optional[List[str]] = None,

) -> "Environment":

"""Create an environment in the directory passed as input and returns it.

Files with suffix ``.json`` or ``.lock`` are considered lockfiles. Files with any other name

are considered manifest files.

Args:

root: directory where to create the environment.

init_file: either a lockfile, a manifest file, an env directory, or None

with_view: whether a view should be maintained for the environment. If the value is a

string, it specifies the path to the view

keep_relative: if True, develop paths are copied verbatim into the new environment file,

otherwise they are made absolute

include_concrete: concrete environment names/paths to be included

"""

# If the initfile is a named environment, get its path

if init_file and exists(str(init_file)):

init_file = read(str(init_file)).path

initialize_environment_dir(root, envfile=init_file)

if with_view is None and keep_relative:

return Environment(root)

try:

manifest = EnvironmentManifestFile(root)

if with_view is not None:

manifest.set_default_view(with_view)

if include_concrete is not None:

set_included_envs_to_env_paths(include_concrete)

validate_included_envs_exists(include_concrete)

validate_included_envs_concrete(include_concrete)

manifest.set_include_concrete(include_concrete)

manifest.flush()

except (spack.config.ConfigFormatError, SpackEnvironmentConfigError) as e:

shutil.rmtree(root)

raise e

env = Environment(root)

if init_file:

if os.path.isdir(init_file):

init_file_dir = init_file

copied = True

else:

init_file_dir = os.path.abspath(os.path.dirname(init_file))

copied = False

if not keep_relative:

if env.path != init_file_dir:

# If we are here, we are creating an environment based on an

# spack.yaml file in another directory, and moreover we want

# dev paths in this environment to refer to their original

# locations.

# If the full env was copied including internal files, only rewrite

# relative paths outside of env

_rewrite_relative_dev_paths_on_relocation(env, init_file_dir, copied_env=copied)

_rewrite_relative_repos_paths_on_relocation(env, init_file_dir, copied_env=copied)

return env

def _rewrite_relative_dev_paths_on_relocation(env, init_file_dir, copied_env=False):

"""When initializing the environment from a manifest file and we plan

to store the environment in a different directory, we have to rewrite

relative paths to absolute ones."""

with env:

dev_specs = spack.config.get("develop", default={}, scope=env.scope_name)

if not dev_specs:

return

for name, entry in dev_specs.items():

dev_path = substitute_path_variables(entry["path"])

expanded_path = spack.util.path.canonicalize_path(dev_path, default_wd=init_file_dir)

# Skip if the substituted and expanded path is the same (e.g. when absolute)

if entry["path"] == expanded_path:

continue

# If copied and it's inside the env, we copied it and don't need to relativize

if copied_env and expanded_path.startswith(init_file_dir):

continue

tty.debug("Expanding develop path for {0} to {1}".format(name, expanded_path))

dev_specs[name]["path"] = expanded_path

spack.config.set("develop", dev_specs, scope=env.scope_name)

env._dev_specs = None

# If we changed the environment's spack.yaml scope, that will not be reflected

# in the manifest that we read

env._re_read()

def _rewrite_relative_repos_paths_on_relocation(env, init_file_dir, copied_env=False):

"""When initializing the environment from a manifest file and we plan

to store the environment in a different directory, we have to rewrite

relative repo paths to absolute ones and expand environment variables."""

with env:

repos_specs = spack.config.get("repos", default={}, scope=env.scope_name)

if not repos_specs:

return

for name, entry in list(repos_specs.items()):

# only rewrite when we have a path-based repository

if not isinstance(entry, str):

continue

repo_path = substitute_path_variables(entry)

expanded_path = spack.util.path.canonicalize_path(repo_path, default_wd=init_file_dir)

# Skip if the substituted and expanded path is the same (e.g. when absolute)

if entry == expanded_path:

continue

# If copied and it's inside the env, we copied it and don't need to relativize

if copied_env and expanded_path.startswith(init_file_dir):

continue

tty.debug("Expanding repo path for {0} to {1}".format(entry, expanded_path))

repos_specs[name] = expanded_path

spack.config.set("repos", repos_specs, scope=env.scope_name)

env.repos_specs = None

# If we changed the environment's spack.yaml scope, that will not be reflected

# in the manifest that we read

env._re_read()

def environment_dir_from_name(name: str, exists_ok: bool = True) -> str:

"""Returns the directory associated with a named environment.

Args:

name: name of the environment

exists_ok: if False, raise an error if the environment exists already

Raises:

SpackEnvironmentError: if exists_ok is False and the environment exists already

"""

if not exists_ok and exists(name):

raise SpackEnvironmentError(f"'{name}': environment already exists at {root(name)}")

ensure_env_root_path_exists()

validate_env_name(name)

return root(name)

def ensure_env_root_path_exists():

if not os.path.isdir(env_root_path()):

fs.mkdirp(env_root_path())

def set_included_envs_to_env_paths(include_concrete: List[str]) -> None:

"""If the included environment(s) is the environment name

it is replaced by the path to the environment

Args:

include_concrete: list of env name or path to env"""

for i, env_name in enumerate(include_concrete):

if is_env_dir(env_name):

include_concrete[i] = env_name

elif exists(env_name):

include_concrete[i] = root(env_name)

def validate_included_envs_exists(include_concrete: List[str]) -> None:

"""Checks that all of the included environments exist

Args:

include_concrete: list of already existing concrete environments to include

Raises:

SpackEnvironmentError: if any of the included environments do not exist

"""

missing_envs = set()

for i, env_name in enumerate(include_concrete):

if not is_env_dir(env_name):

missing_envs.add(env_name)

if missing_envs:

msg = "The following environment(s) are missing: {0}".format(", ".join(missing_envs))

raise SpackEnvironmentError(msg)

def validate_included_envs_concrete(include_concrete: List[str]) -> None:

"""Checks that all of the included environments are concrete

Args:

include_concrete: list of already existing concrete environments to include

Raises:

SpackEnvironmentError: if any of the included environments are not concrete

"""

non_concrete_envs = set()

for env_path in include_concrete:

if not os.path.exists(os.path.join(env_path, lockfile_name)):

non_concrete_envs.add(environment_name(env_path))

if non_concrete_envs:

msg = "The following environment(s) are not concrete: {0}\nPlease run:".format(

", ".join(non_concrete_envs)

)

for env in non_concrete_envs:

msg += f"\n\t`spack -e {env} concretize`"

raise SpackEnvironmentError(msg)

def all_environment_names():

"""List the names of environments that currently exist."""

# just return empty if the env path does not exist. A read-only

# operation like list should not try to create a directory.

if not os.path.exists(env_root_path()):

return []

env_root = pathlib.Path(env_root_path()).resolve()

def yaml_paths():

for root, dirs, files in os.walk(env_root, topdown=True, followlinks=True):

dirs[:] = [

d

for d in dirs

if not d.startswith(".") and not env_root.samefile(os.path.join(root, d))

]

if manifest_name in files:

yield os.path.join(root, manifest_name)

names = []

for yaml_path in yaml_paths():

candidate = str(pathlib.Path(yaml_path).relative_to(env_root).parent)

if valid_env_name(candidate):

names.append(candidate)

return names

def all_environments():

"""Generator for all managed Environments."""

for name in all_environment_names():

yield read(name)

def _read_yaml(str_or_file):

"""Read YAML from a file for round-trip parsing."""

try:

data = syaml.load_config(str_or_file)

except syaml.SpackYAMLError as e:

raise SpackEnvironmentConfigError(

f"Invalid environment configuration detected: {e.message}", e.filename

)

filename = getattr(str_or_file, "name", None)

spack.config.validate(data, spack.schema.env.schema, filename)

return data

def _write_yaml(data, str_or_file):

"""Write YAML to a file preserving comments and dict order."""

filename = getattr(str_or_file, "name", None)

spack.config.validate(data, spack.schema.env.schema, filename)

syaml.dump_config(data, str_or_file, default_flow_style=False)

def _is_dev_spec_and_has_changed(spec):

"""Check if the passed spec is a dev build and whether it has changed since the

last installation"""

# First check if this is a dev build and in the process already try to get

# the dev_path

if not spec.variants.get("dev_path", None):

return False

# Now we can check whether the code changed since the last installation

if not spec.installed:

# Not installed -> nothing to compare against

return False

# hook so packages can use to write their own method for checking the dev_path

# use package so attributes about concretization such as variant state can be

# utilized

return spec.package.detect_dev_src_change()

def _error_on_nonempty_view_dir(new_root):

"""Defensively error when the target view path already exists and is not an

empty directory. This usually happens when the view symlink was removed, but

not the directory it points to. In those cases, it's better to just error when

the new view dir is non-empty, since it indicates the user removed part but not

all of the view, and it likely in an inconsistent state."""

# Check if the target path lexists

try:

st = os.lstat(new_root)

except OSError:

return

# Empty directories are fine

if stat.S_ISDIR(st.st_mode) and len(os.listdir(new_root)) == 0:

return

# Anything else is an error

raise SpackEnvironmentViewError(

"Failed to generate environment view, because the target {} already "

"exists or is not empty. To update the view, remove this path, and run "

"`spack env view regenerate`".format(new_root)

)

class ViewDescriptor:

def __init__(

self,

base_path: str,

root: str,

*,

projections: Optional[Dict[str, str]] = None,

select: Optional[List[str]] = None,

exclude: Optional[List[str]] = None,

link: str = default_view_link,

link_type: fsv.LinkType = "symlink",

groups: Optional[Union[str, List[str]]] = None,

) -> None:

self.base = base_path

self.raw_root = root

self.root = spack.util.path.canonicalize_path(root, default_wd=base_path)

self.projections = projections or {}

self.select = select or []

self.exclude = exclude or []

self.link_type: fsv.LinkType = fsv.canonicalize_link_type(link_type)

self.link = link

if isinstance(groups, str):

groups = [groups]

self.groups: Optional[List[str]] = groups

def select_fn(self, spec: Spec) -> bool:

return any(spec.satisfies(s) for s in self.select)

def exclude_fn(self, spec: Spec) -> bool:

return not any(spec.satisfies(e) for e in self.exclude)

def update_root(self, new_path: str) -> None:

self.raw_root = new_path

self.root = spack.util.path.canonicalize_path(new_path, default_wd=self.base)

def __eq__(self, other: object) -> bool:

return isinstance(other, ViewDescriptor) and all(

[

self.root == other.root,

self.projections == other.projections,

self.select == other.select,

self.exclude == other.exclude,

self.link == other.link,

self.link_type == other.link_type,

]

)

def to_dict(self):

ret = syaml.syaml_dict([("root", self.raw_root)])

if self.projections:

ret["projections"] = self.projections

if self.select:

ret["select"] = self.select

if self.exclude:

ret["exclude"] = self.exclude

if self.link_type:

ret["link_type"] = self.link_type

if self.link != default_view_link:

ret["link"] = self.link

return ret

@staticmethod

def from_dict(base_path: str, d) -> "ViewDescriptor":

return ViewDescriptor(

base_path,

d["root"],

projections=d.get("projections", {}),

select=d.get("select", []),

exclude=d.get("exclude", []),

link=d.get("link", default_view_link),

link_type=d.get("link_type", "symlink"),

groups=d.get("group", None),

)

@property

def _current_root(self) -> Optional[str]:

if not islink(self.root):

return None

root = readlink(self.root)

if os.path.isabs(root):

return root

root_dir = os.path.dirname(self.root)

return os.path.join(root_dir, root)

def _next_root(self, specs):

content_hash = self.content_hash(specs)

root_dir = os.path.dirname(self.root)

root_name = os.path.basename(self.root)

return os.path.join(root_dir, "._%s" % root_name, content_hash)

def content_hash(self, specs):

d = syaml.syaml_dict(

[

("descriptor", self.to_dict()),

("specs", [(spec.dag_hash(), spec.prefix) for spec in sorted(specs)]),

]

)

contents = sjson.dump(d)

return spack.util.hash.b32_hash(contents)

def get_projection_for_spec(self, spec):

"""Get projection for spec. This function does not require the view

to exist on the filesystem."""

return self._view(self.root).get_projection_for_spec(spec)

def view(self, new: Optional[str] = None) -> fsv.SimpleFilesystemView:

"""

Returns a view object for the *underlying* view directory. This means that the

self.root symlink is followed, and that the view has to exist on the filesystem

(unless ``new``). This function is useful when writing to the view.

Raise if new is None and there is no current view

Arguments:

new: If a string, create a FilesystemView rooted at that path. Default None. This

should only be used to regenerate the view, and cannot be used to access specs.

"""

path = new if new else self._current_root

if not path:

# This can only be hit if we write a future bug

raise SpackEnvironmentViewError(

f"Attempting to get nonexistent view from environment. View root is at {self.root}"

)

return self._view(path)

def _view(self, root: str) -> fsv.SimpleFilesystemView:

"""Returns a view object for a given root dir."""

return fsv.SimpleFilesystemView(

root,

spack.store.STORE.layout,

ignore_conflicts=True,

projections=self.projections,

link_type=self.link_type,

)

def __contains__(self, spec):

"""Is the spec described by the view descriptor

Note: This does not claim the spec is already linked in the view.

It merely checks that the spec is selected if a select operation is

specified and is not excluded if an exclude operator is specified.

"""

if self.select:

if not self.select_fn(spec):

return False

if self.exclude:

if not self.exclude_fn(spec):

return False

return True

def specs_for_view(self, concrete_roots: List[Spec]) -> List[Spec]:

"""Flatten the DAGs of the concrete roots, keep only unique, selected, and installed specs

in topological order from root to leaf."""

if self.link == "all":

deptype = dt.LINK | dt.RUN

elif self.link == "run":

deptype = dt.RUN

else:

deptype = dt.NONE

specs = traverse.traverse_nodes(

concrete_roots, order="topo", deptype=deptype, key=traverse.by_dag_hash

)

# Filter selected, installed specs

with spack.store.STORE.db.read_transaction():

result = [s for s in specs if s in self and s.installed]

return self._exclude_duplicate_runtimes(result)

def regenerate(self, env: "Environment") -> None:

if self.groups is None:

concrete_roots = env.concrete_roots()

else:

concrete_roots = [c for g in self.groups for _, c in env.concretized_specs_by(group=g)]

specs = self.specs_for_view(concrete_roots)

# To ensure there are no conflicts with packages being installed

# that cannot be resolved or have repos that have been removed

# we always regenerate the view from scratch.

# We will do this by hashing the view contents and putting the view

# in a directory by hash, and then having a symlink to the real

# view in the root. The real root for a view at /dirname/basename

# will be /dirname/._basename_<hash>.

# This allows for atomic swaps when we update the view

# cache the roots because the way we determine which is which does

# not work while we are updating

new_root = self._next_root(specs)

old_root = self._current_root

if new_root == old_root:

tty.debug(f"View at {self.root} does not need regeneration.")

return

_error_on_nonempty_view_dir(new_root)

# construct view at new_root

if specs:

tty.msg(f"Updating view at {self.root}")

view = self.view(new=new_root)

root_dirname = os.path.dirname(self.root)

tmp_symlink_name = os.path.join(root_dirname, "._view_link")

# Remove self.root if is it an empty dir, since we need a symlink there. Note that rmdir

# fails if self.root is a symlink.

try:

os.rmdir(self.root)

except (FileNotFoundError, NotADirectoryError):

pass

except OSError as e:

if e.errno == errno.ENOTEMPTY:

msg = "it is a non-empty directory"

elif e.errno == errno.EACCES:

msg = "of insufficient permissions"

else:

raise

raise SpackEnvironmentViewError(

f"The environment view in {self.root} cannot not be created because {msg}."

) from e

# Create a new view

try:

fs.mkdirp(new_root)

view.add_specs(*specs)

# create symlink from tmp_symlink_name to new_root

if os.path.exists(tmp_symlink_name):

os.unlink(tmp_symlink_name)

symlink(new_root, tmp_symlink_name)

# mv symlink atomically over root symlink to old_root

fs.rename(tmp_symlink_name, self.root)

except Exception as e:

# Clean up new view and temporary symlink on any failure.

try:

shutil.rmtree(new_root, ignore_errors=True)

os.unlink(tmp_symlink_name)

except OSError:

pass

# Give an informative error message for the typical error case: two specs, same package

# project to same prefix.

if isinstance(e, ConflictingSpecsError):

spec_a = e.args[0].format(color=clr.get_color_when())

spec_b = e.args[1].format(color=clr.get_color_when())

raise SpackEnvironmentViewError(

f"The environment view in {self.root} could not be created, "

"because the following two specs project to the same prefix:\n"

f" {spec_a}, and\n"

f" {spec_b}.\n"

" To resolve this issue:\n"

" a. use `concretization:unify:true` to ensure there is only one "

"package per spec in the environment, or\n"

" b. disable views with `view:false`, or\n"

" c. create custom view projections."

) from e

raise

# Remove the old root when it's in the same folder as the new root. This guards

# against removal of an arbitrary path when the original symlink in self.root

# was not created by the environment, but by the user.

if (

old_root

and os.path.exists(old_root)

and os.path.samefile(os.path.dirname(new_root), os.path.dirname(old_root))

):

try:

shutil.rmtree(old_root)

except OSError as e:

msg = "Failed to remove old view at %s\n" % old_root

msg += str(e)

tty.warn(msg)

def _exclude_duplicate_runtimes(self, specs: List[Spec]) -> List[Spec]:

"""Stably filter out duplicates of "runtime" tagged packages, keeping only latest."""

# Maps packages tagged "runtime" to the spec with latest version.

latest: Dict[str, Spec] = {}

for s in specs:

if "runtime" not in getattr(s.package, "tags", ()):

continue

elif s.name not in latest or latest[s.name].version < s.version:

latest[s.name] = s

return [x for x in specs if x.name not in latest or latest[x.name] is x]

def env_subdir_path(manifest_dir: Union[str, pathlib.Path]) -> str:

"""Path to where the environment stores repos, logs, views, configs.

Args:

manifest_dir: directory containing the environment manifest file

Returns: directory the environment uses to manage its files

"""

return os.path.join(str(manifest_dir), env_subdir_name)

class ConcretizedRootInfo: