Separate intrinsic environment variables

tmt-owned variables are no longer included in the
`HasEnvironment.environment` return values. Separating the user-owned
and tmt-owned variables will give us precise control over their
ordering.

Patch is also dropping support of `Environment.update(foo: HasEnvironment)`
shortcut - an actual `Environment` instances are required now. Once a
class offers both user-provided and tmt-owned variables, `update()`
would have hard time to pick which set to include in the target
environment ("both" might also be an answer...).

This should be the penultimate patch of the environment precedence
series; the next and almost-final patch will provide unified helper to
construct environments so they follow the same process.

Related to #2609.

Author: happz

tmt/base/plan.py

@@ -57,7 +57,7 @@
     style,
     to_yaml,
 )
-from tmt.utils.environment import Environment, EnvVarValue, HasEnvironment
+from tmt.utils.environment import Environment, EnvVarValue, HasEnvironment, HasIntrinsicEnvironment
 
 if TYPE_CHECKING:
     import tmt.cli
@@ -224,6 +224,7 @@ class Plan(
     HasUserAnchorPath,
     HasPlanWorkdir,
     HasEnvironment,
+    HasIntrinsicEnvironment,
     Core,
     tmt.export.Exportable['Plan'],
     tmt.lint.Lintable['Plan'],
@@ -458,11 +459,7 @@ def draw_test_serial_number(self, test: Test) -> int:
     _environment_from_importing: Environment = field(default_factory=Environment, internal=True)
 
     @property
-    def _environment_from_intrinsics(self) -> Environment:
-        """
-        Environment variables derived from the plan properties.
-        """
-
+    def intrinsic_environment(self) -> Environment:
         environment = Environment(
             {
                 'TMT_VERSION': EnvVarValue(tmt.__version__),
@@ -549,7 +546,6 @@ def environment(self) -> Environment:
                     **self._environment_from_importing,
                     **self._environment_from_cli,
                     **self.my_run.environment,
-                    **self._environment_from_intrinsics,
                 }
             )
 
@@ -558,7 +554,6 @@ def environment(self) -> Environment:
                 **self._environment_from_fmf,
                 **self._environment_from_importing,
                 **self._environment_from_cli,
-                **self._environment_from_intrinsics,
             }
         )
 

tmt/guest/__init__.py

@@ -40,6 +40,7 @@
 import tmt.steps
 import tmt.steps.scripts
 import tmt.utils
+import tmt.utils.environment
 import tmt.utils.wait
 from tmt._compat.typing import Self
 from tmt.ansible import (
@@ -1718,6 +1719,7 @@ class Guest(
     # TODO: `Guest` does "have" environment, but it's a genuine attribute,
     # not a property, and this interface will not work.
     # tmt.utils.HasEnvironment,
+    tmt.utils.environment.HasIntrinsicEnvironment,
     tmt.utils.Common,
 ):
     """
@@ -1933,6 +1935,14 @@ def plan_environment(self) -> Environment:
 
         return Environment()
 
+    @property
+    def intrinsic_environment(self) -> Environment:
+        environment = Environment()
+
+        environment['TMT_PLAN_ENVIRONMENT_FILE'] = EnvVarValue(self.plan_environment_path)
+
+        return environment
+
     @classmethod
     def options(cls, how: Optional[str] = None) -> list[tmt.options.ClickOptionDecoratorType]:
         """
@@ -2257,14 +2267,12 @@ def _prepare_command_environment(
             environment.update(self.environment)
 
             if isinstance(self.parent, tmt.steps.Step):
-                environment.update(self.parent.plan)
-
-            # TODO: this was owned by plan, but at wrong position, and it will
-            # be owned by plan again once the dust of environment untangling
-            # settles. Follow https://github.com/teemtee/tmt/issues/4241 for
-            # more.
-            if self.plan_environment_path:
-                environment['TMT_PLAN_ENVIRONMENT_FILE'] = EnvVarValue(self.plan_environment_path)
+                environment.update(self.parent.plan.environment)
+
+            environment.update(self.intrinsic_environment)
+
+            if isinstance(self.parent, tmt.steps.Step):
+                environment.update(self.parent.plan.intrinsic_environment)
 
         else:
             # Create a copy of given environment - this prevents any

tmt/steps/context/abort.py

@@ -5,7 +5,7 @@
 import tmt.utils
 from tmt.container import container
 from tmt.utils import Path
-from tmt.utils.environment import Environment, HasEnvironment
+from tmt.utils.environment import Environment, HasIntrinsicEnvironment
 
 
 class AbortStep(tmt.utils.GeneralError):
@@ -15,7 +15,7 @@ class AbortStep(tmt.utils.GeneralError):
 
 
 @container
-class AbortContext(HasEnvironment):
+class AbortContext(HasIntrinsicEnvironment):
     """
     Provides API for handling a phase-requested abort of a step.
     """
@@ -43,5 +43,5 @@ def requested(self) -> bool:
         return self.request_path.exists()
 
     @property
-    def environment(self) -> Environment:
+    def intrinsic_environment(self) -> Environment:
         return Environment()

tmt/steps/context/pidfile.py

@@ -84,7 +84,7 @@
 from tmt.guest import Guest, TransferOptions
 from tmt.steps import safe_filename
 from tmt.utils import Path, ShellScript
-from tmt.utils.environment import Environment, EnvVarValue, HasEnvironment
+from tmt.utils.environment import Environment, EnvVarValue, HasIntrinsicEnvironment
 from tmt.utils.templates import render_template
 
 TEST_PIDFILE_FILENAME = 'tmt-test.pid'
@@ -193,7 +193,7 @@ def effective_pidfile_root() -> Path:
 
 
 @container
-class PidFileContext(HasEnvironment):
+class PidFileContext(HasIntrinsicEnvironment):
     #: Phase owning this context.
     phase: tmt.steps.BasePlugin[Any, Any]
 
@@ -220,7 +220,7 @@ def pidfile_lock_path(self) -> Path:
         return effective_pidfile_root() / TEST_PIDFILE_LOCK_FILENAME
 
     @property
-    def environment(self) -> Environment:
+    def intrinsic_environment(self) -> Environment:
         return Environment(
             {
                 'TMT_TEST_PIDFILE': EnvVarValue(self.pidfile_path),

tmt/steps/context/reboot.py

@@ -9,7 +9,7 @@
 from tmt.container import MetadataContainer, container
 from tmt.guest import Guest, RebootMode, SoftRebootModes
 from tmt.utils import Path, ShellScript
-from tmt.utils.environment import Environment, EnvVarValue, HasEnvironment
+from tmt.utils.environment import Environment, EnvVarValue, HasIntrinsicEnvironment
 from tmt.utils.wait import Deadline, Waiting
 
 if TYPE_CHECKING:
@@ -27,7 +27,7 @@ class RebootData(MetadataContainer):
 
 
 @container
-class RebootContext(HasEnvironment):
+class RebootContext(HasIntrinsicEnvironment):
     """
     Tracks information about guest reboots.
     """
@@ -76,7 +76,7 @@ def requested(self) -> bool:
         return self.soft_requested or self.hard_requested
 
     @property
-    def environment(self) -> Environment:
+    def intrinsic_environment(self) -> Environment:
         environment = Environment()
 
         # Set all supported reboot variables

tmt/steps/context/restart.py

@@ -4,14 +4,14 @@
 import tmt.utils
 from tmt.container import container
 from tmt.guest import Guest
-from tmt.utils.environment import Environment, EnvVarValue, HasEnvironment
+from tmt.utils.environment import Environment, EnvVarValue, HasIntrinsicEnvironment
 
 if TYPE_CHECKING:
     from tmt.steps.context.reboot import RebootContext
 
 
 @container
-class RestartContext(HasEnvironment):
+class RestartContext(HasIntrinsicEnvironment):
     """
     Tracks information about restarts of an action, e.g. a test script.
     """
@@ -50,7 +50,7 @@ def requested(self) -> bool:
         return self.is_requested_test()
 
     @property
-    def environment(self) -> Environment:
+    def intrinsic_environment(self) -> Environment:
         return Environment({'TMT_TEST_RESTART_COUNT': EnvVarValue(str(self.restart_counter))})
 
     def handle_restart(self, reboot: Optional['RebootContext'] = None) -> bool:

tmt/steps/context/restraint.py

@@ -2,11 +2,11 @@
 
 import tmt.log
 from tmt.container import container
-from tmt.utils.environment import Environment, EnvVarValue, HasEnvironment
+from tmt.utils.environment import Environment, EnvVarValue, HasIntrinsicEnvironment
 
 
 @container
-class RestraintContext(HasEnvironment):
+class RestraintContext(HasIntrinsicEnvironment):
     """
     Provides restraint-related context for execution.
     """
@@ -22,7 +22,7 @@ class RestraintContext(HasEnvironment):
     taskname: Optional[str] = None
 
     @property
-    def environment(self) -> Environment:
+    def intrinsic_environment(self) -> Environment:
         environment = Environment()
 
         environment["TMT_RESTRAINT_COMPATIBLE"] = EnvVarValue(str(int(self.enabled)))

tmt/steps/execute/__init__.py

@@ -48,7 +48,7 @@
     Stopwatch,
     configure_bool_constant,
 )
-from tmt.utils.environment import Environment, EnvVarValue, HasEnvironment
+from tmt.utils.environment import Environment, EnvVarValue, HasEnvironment, HasIntrinsicEnvironment
 
 if TYPE_CHECKING:
     import tmt.base.plan
@@ -111,7 +111,7 @@ class ExecuteStepData(tmt.steps.WhereableStepData, tmt.steps.StepData):
 
 
 @container
-class TestInvocation(HasStepWorkdir, HasEnvironment):
+class TestInvocation(HasStepWorkdir, HasEnvironment, HasIntrinsicEnvironment):
     """
     A bundle describing one test invocation.
 
@@ -352,56 +352,62 @@ def restraint(self) -> RestraintContext:
             logger=self.logger,
         )
 
+    @property
+    def intrinsic_environment(self) -> Environment:
+        # narrow type
+        assert isinstance(self.phase.step.plan.my_run, tmt.base.run.Run)
+
+        environment = Environment()
+
+        environment["TMT_TEST_NAME"] = EnvVarValue(self.test.name)
+        environment["TMT_TEST_INVOCATION_PATH"] = EnvVarValue(self.path)
+        environment["TMT_TEST_DATA"] = EnvVarValue(self.test_data_path)
+        environment["TMT_TEST_SUBMITTED_FILES"] = EnvVarValue(self.submission_log_path)
+        environment['TMT_TEST_SERIAL_NUMBER'] = EnvVarValue(str(self.test.serial_number))
+        environment["TMT_TEST_METADATA"] = EnvVarValue(self.path / TEST_METADATA_FILENAME)
+
+        environment['TMT_TEST_ITERATION_ID'] = EnvVarValue(
+            f"{self.phase.step.plan.my_run.unique_id}-{self.test.serial_number}"
+        )
+
+        environment['TMT_SOURCE_DIR'] = EnvVarValue(self.discover_phase.source_dir)
+
+        environment.update(
+            # Add variables from plan
+            self.phase.step.plan.intrinsic_environment,
+            # Add variables from guest
+            self.guest.intrinsic_environment,
+            # Add variables from invocation contexts
+            self.abort.intrinsic_environment,
+            self.reboot.intrinsic_environment,
+            self.restart.intrinsic_environment,
+            self.pidfile.intrinsic_environment,
+            self.restraint.intrinsic_environment,
+            # Add variables the framework wants to expose
+            self.test.test_framework.get_environment_variables(self, self.logger),
+        )
+
+        return environment
+
     @property
     def environment(self) -> Environment:
         if self._environment is None:
             # narrow type
-            assert isinstance(self.phase.parent, Execute)
-
-            # narrow type
-            assert isinstance(self.phase.parent.plan.my_run, tmt.base.run.Run)
+            assert isinstance(self.phase.step.plan.my_run, tmt.base.run.Run)
 
             environment = Environment()
 
             environment.update(
                 self.guest.environment,
                 self.test.environment,
                 self.guest.plan_environment,
-                self.phase.parent.plan.environment,
+                self.phase.step.plan.environment,
             )
 
-            environment["TMT_TEST_NAME"] = EnvVarValue(self.test.name)
-            environment["TMT_TEST_INVOCATION_PATH"] = EnvVarValue(self.path)
-            environment["TMT_TEST_DATA"] = EnvVarValue(self.test_data_path)
-            environment["TMT_TEST_SUBMITTED_FILES"] = EnvVarValue(self.submission_log_path)
-            environment['TMT_TEST_SERIAL_NUMBER'] = EnvVarValue(str(self.test.serial_number))
-            environment["TMT_TEST_METADATA"] = EnvVarValue(self.path / TEST_METADATA_FILENAME)
-
-            environment['TMT_TEST_ITERATION_ID'] = EnvVarValue(
-                f"{self.phase.parent.plan.my_run.unique_id}-{self.test.serial_number}"
-            )
-
-            environment['TMT_SOURCE_DIR'] = EnvVarValue(self.discover_phase.source_dir)
-
         else:
             environment = self._environment
 
-        # TODO: this was owned by plan, but at wrong position, and it will
-        # be owned by plan again once the dust of environment untangling
-        # settles. Follow https://github.com/teemtee/tmt/issues/4241 for
-        # more.
-        environment['TMT_PLAN_ENVIRONMENT_FILE'] = EnvVarValue(self.guest.plan_environment_path)
-
-        environment.update(
-            # Add variables from invocation contexts
-            self.abort,
-            self.reboot,
-            self.restart,
-            self.pidfile,
-            self.restraint,
-            # Add variables the framework wants to expose
-            self.test.test_framework.get_environment_variables(self, self.logger),
-        )
+        environment.update(self.intrinsic_environment)
 
         self._environment = environment
 

tmt/steps/prepare/shell.py

@@ -151,13 +151,6 @@ def go(
             self.step.plan.environment,
         )
 
-        # TODO: this was owned by plan, but at wrong position, and it will
-        # be owned by plan again once the dust of environment untangling
-        # settles. Follow https://github.com/teemtee/tmt/issues/4241 for
-        # more.
-        if guest.plan_environment_path:
-            environment['TMT_PLAN_ENVIRONMENT_FILE'] = EnvVarValue(guest.plan_environment_path)
-
         # Give a short summary
         overview = fmf.utils.listed(self.data.script, 'script')
         logger.info('overview', f'{overview} found', 'green')
@@ -269,8 +262,12 @@ def _invoke_script(
             script_log_filepath.touch()
 
             script_environment = environment.copy()
-            script_environment.update(reboot_context)
-            script_environment.update(pidfile_context)
+            script_environment.update(
+                self.step.plan.intrinsic_environment,
+                guest.intrinsic_environment,
+                reboot_context.intrinsic_environment,
+                pidfile_context.intrinsic_environment,
+            )
 
             pull_options = DEFAULT_PULL_OPTIONS.copy()
             pull_options.exclude.append(str(script_log_filepath))

tmt/steps/provision/local.py

@@ -20,7 +20,7 @@
     Path,
     ShellScript,
 )
-from tmt.utils.environment import Environment, EnvVarValue
+from tmt.utils.environment import Environment
 from tmt.utils.hints import get_hint
 from tmt.utils.wait import Waiting
 
@@ -73,14 +73,12 @@ def _prepare_command_environment(
             environment.update(self.environment)
 
             if isinstance(self.parent, tmt.steps.Step):
-                environment.update(self.parent.plan)
-
-            # TODO: this was owned by plan, but at wrong position, and it will
-            # be owned by plan again once the dust of environment untangling
-            # settles. Follow https://github.com/teemtee/tmt/issues/4241 for
-            # more.
-            if self.plan_environment_path:
-                environment['TMT_PLAN_ENVIRONMENT_FILE'] = EnvVarValue(self.plan_environment_path)
+                environment.update(self.parent.plan.environment)
+
+            environment.update(self.intrinsic_environment)
+
+            if isinstance(self.parent, tmt.steps.Step):
+                environment.update(self.parent.plan.intrinsic_environment)
 
         else:
             environment = super()._prepare_command_environment(environment=environment)