include_concrete: read from older env formats properly (#45766)

* include_concrete: read from older env formats properly
* spack env rm: fix logic for checking env includes
* regression test

Signed-off-by: Todd Gamblin <tgamblin@llnl.gov>

3 files changed, 43 insertions, 39 deletions

diff --git a/lib/spack/spack/cmd/env.py b/lib/spack/spack/cmd/env.py
index 2ccb88fd1a..b943f3d3bd 100644
--- a/lib/spack/spack/cmd/env.py
+++ b/lib/spack/spack/cmd/env.py
@@ -468,32 +468,30 @@ def env_remove(args):
     This removes an environment managed by Spack. Directory environments
     and manifests embedded in repositories should be removed manually.
     """
-    read_envs = []
+    remove_envs = []
     valid_envs = []
     bad_envs = []
-    invalid_envs = []
 
     for env_name in ev.all_environment_names():
         try:
             env = ev.read(env_name)
-            valid_envs.append(env_name)
+            valid_envs.append(env)
 
             if env_name in args.rm_env:
-                read_envs.append(env)
+                remove_envs.append(env)
         except (spack.config.ConfigFormatError, ev.SpackEnvironmentConfigError):
-            invalid_envs.append(env_name)
-
             if env_name in args.rm_env:
                 bad_envs.append(env_name)
 
-        # Check if env is linked to another before trying to remove
-        for name in valid_envs:
+    # Check if remove_env is included from another env before trying to remove
+    for env in valid_envs:
+        for remove_env in remove_envs:
             # don't check if environment is included to itself
-            if name == env_name:
+            if env.name == remove_env.name:
                 continue
-            environ = ev.Environment(ev.root(name))
-            if ev.root(env_name) in environ.included_concrete_envs:
-                msg = f'Environment "{env_name}" is being used by environment "{name}"'
+
+            if remove_env.path in env.included_concrete_envs:
+                msg = f'Environment "{remove_env.name}" is being used by environment "{env.name}"'
                 if args.force:
                     tty.warn(msg)
                 else:
@@ -506,7 +504,7 @@ def env_remove(args):
         if not answer:
             tty.die("Will not remove any environments")
 
-    for env in read_envs:
+    for env in remove_envs:
         name = env.name
         if env.active:
             tty.die(f"Environment {name} can't be removed while activated.")
diff --git a/lib/spack/spack/environment/environment.py b/lib/spack/spack/environment/environment.py
index ef51cb38d6..4114a1cdb9 100644
--- a/lib/spack/spack/environment/environment.py
+++ b/lib/spack/spack/environment/environment.py
@@ -1214,7 +1214,6 @@ class Environment:
     def include_concrete_envs(self):
         """Copy and save the included envs' specs internally"""
 
-        lockfile_meta = None
         root_hash_seen = set()
         concrete_hash_seen = set()
         self.included_concrete_spec_data = {}
@@ -1225,37 +1224,26 @@ class Environment:
                 raise SpackEnvironmentError(f"Unable to find env at {env_path}")
 
             env = Environment(env_path)
-
-            with open(env.lock_path) as f:
-                lockfile_as_dict = env._read_lockfile(f)
-
-            # Lockfile_meta must match each env and use at least format version 5
-            if lockfile_meta is None:
-                lockfile_meta = lockfile_as_dict["_meta"]
-            elif lockfile_meta != lockfile_as_dict["_meta"]:
-                raise SpackEnvironmentError("All lockfile _meta values must match")
-            elif lockfile_meta["lockfile-version"] < 5:
-                raise SpackEnvironmentError("The lockfile format must be at version 5 or higher")
+            self.included_concrete_spec_data[env_path] = {"roots": [], "concrete_specs": {}}
 
             # Copy unique root specs from env
-            self.included_concrete_spec_data[env_path] = {"roots": []}
-            for root_dict in lockfile_as_dict["roots"]:
+            for root_dict in env._concrete_roots_dict():
                 if root_dict["hash"] not in root_hash_seen:
                     self.included_concrete_spec_data[env_path]["roots"].append(root_dict)
                     root_hash_seen.add(root_dict["hash"])
 
             # Copy unique concrete specs from env
-            for concrete_spec in lockfile_as_dict["concrete_specs"]:
-                if concrete_spec not in concrete_hash_seen:
-                    self.included_concrete_spec_data[env_path].update(
-                        {"concrete_specs": lockfile_as_dict["concrete_specs"]}
+            for dag_hash, spec_details in env._concrete_specs_dict().items():
+                if dag_hash not in concrete_hash_seen:
+                    self.included_concrete_spec_data[env_path]["concrete_specs"].update(
+                        {dag_hash: spec_details}
                     )
-                    concrete_hash_seen.add(concrete_spec)
+                    concrete_hash_seen.add(dag_hash)
 
-            if "include_concrete" in lockfile_as_dict.keys():
-                self.included_concrete_spec_data[env_path]["include_concrete"] = lockfile_as_dict[
-                    "include_concrete"
-                ]
+            # Copy transitive include data
+            transitive = env.included_concrete_spec_data
+            if transitive:
+                self.included_concrete_spec_data[env_path]["include_concrete"] = transitive
 
         self._read_lockfile_dict(self._to_lockfile_dict())
         self.write()
@@ -2173,16 +2161,23 @@ class Environment:
 
         return specs
 
-    def _to_lockfile_dict(self):
-        """Create a dictionary to store a lockfile for this environment."""
+    def _concrete_specs_dict(self):
         concrete_specs = {}
         for s in traverse.traverse_nodes(self.specs_by_hash.values(), key=traverse.by_dag_hash):
             spec_dict = s.node_dict_with_hashes(hash=ht.dag_hash)
             # Assumes no legacy formats, since this was just created.
             spec_dict[ht.dag_hash.name] = s.dag_hash()
             concrete_specs[s.dag_hash()] = spec_dict
+        return concrete_specs
 
+    def _concrete_roots_dict(self):
         hash_spec_list = zip(self.concretized_order, self.concretized_user_specs)
+        return [{"hash": h, "spec": str(s)} for h, s in hash_spec_list]
+
+    def _to_lockfile_dict(self):
+        """Create a dictionary to store a lockfile for this environment."""
+        concrete_specs = self._concrete_specs_dict()
+        root_specs = self._concrete_roots_dict()
 
         spack_dict = {"version": spack.spack_version}
         spack_commit = spack.main.get_spack_commit()
@@ -2203,7 +2198,7 @@ class Environment:
             # spack version information
             "spack": spack_dict,
             # users specs + hashes are the 'roots' of the environment
-            "roots": [{"hash": h, "spec": str(s)} for h, s in hash_spec_list],
+            "roots": root_specs,
             # Concrete specs by hash, including dependencies
             "concrete_specs": concrete_specs,
         }
diff --git a/lib/spack/spack/test/cmd/env.py b/lib/spack/spack/test/cmd/env.py
index 77fa41d54f..9c32ab7610 100644
--- a/lib/spack/spack/test/cmd/env.py
+++ b/lib/spack/spack/test/cmd/env.py
@@ -1734,6 +1734,17 @@ def test_env_include_concrete_env_yaml(env_name):
     assert test.path in combined_yaml["include_concrete"]
 
 
+@pytest.mark.regression("45766")
+@pytest.mark.parametrize("format", ["v1", "v2", "v3"])
+def test_env_include_concrete_old_env(format, tmpdir):
+    lockfile = os.path.join(spack.paths.test_path, "data", "legacy_env", f"{format}.lock")
+    # create an env from old .lock file -- this does not update the format
+    env("create", "old-env", lockfile)
+    env("create", "--include-concrete", "old-env", "test")
+
+    assert ev.read("old-env").all_specs() == ev.read("test").all_specs()
+
+
 def test_env_bad_include_concrete_env():
     with pytest.raises(ev.SpackEnvironmentError):
         env("create", "--include-concrete", "nonexistant_env", "combined_env")