Separate intrinsic environment variables

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'],
@@ -335,7 +336,11 @@ def __init__(
                 self._initialize_worktree()
 
         # Expand all environment and context variables in the node
-        with self.environment.as_environ():
+        environment = Environment()
+
+        environment.update(self.environment, self.intrinsic_environment)
+
+        with environment.as_environ():
             expand_node_data(node.data, self.fmf_context)  # pyright: ignore[reportUnknownVariableType, reportUnknownArgumentType]
 
         # Initialize test steps
@@ -458,11 +463,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 +550,6 @@ def environment(self) -> Environment:
                     **self._environment_from_importing,
                     **self._environment_from_cli,
                     **self.my_run.environment,
-                    **self._environment_from_intrinsics,
                 }
             )
 
@@ -558,7 +558,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,15 @@ def plan_environment(self) -> Environment:
 
         return Environment()
 
+    @property
+    def intrinsic_environment(self) -> Environment:
+        environment = Environment()
+
+        if self.plan_environment_path is not None:
+            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 +2268,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))