# Copyright 2026 European Union
# Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
# SPDX-License-Identifier: EUPL-1.2
"""Resolve steering-scoped dotenv files without changing process state.
:Author: Bulgheroni Antonio
:Description: Build an isolated environment mapping for one steering file.
.. versionadded:: 2.4
"""
from __future__ import annotations
import os
import re
from collections.abc import Mapping
from dataclasses import dataclass
from pathlib import Path
from typing import Any
from dotenv import dotenv_values
from dotenv.parser import parse_stream
from mafw.tools.path_resolution import PathResolutionContext, PathResolutionError, PathValidationPolicy, resolve_path
_ENVIRONMENT_KEY_PATTERN = re.compile(r'[A-Za-z_][A-Za-z0-9_]*\Z')
[docs]
class SteeringEnvironmentError(ValueError):
"""Raised when a steering file cannot resolve one of its dotenv files."""
[docs]
@dataclass(frozen=True)
class EnvironmentFileStatus:
"""A configured dotenv path and its safe-to-display validation message."""
configured_path: str
resolved_path: Path | None
error: str | None = None
[docs]
@dataclass(frozen=True)
class SteeringEnvironment:
"""Environment values and path statuses belonging to one steering file."""
values: dict[str, str]
files: tuple[EnvironmentFileStatus, ...]
errors: tuple[str, ...] = ()
[docs]
def resolve_steering_environment(
config: Mapping[str, Any],
steering_file: Path | str,
*,
process_environment: Mapping[str, str] | None = None,
strict: bool = True,
) -> SteeringEnvironment:
"""Merge dotenv values and the process environment for one steering file.
:param config: Raw steering mapping before interpolation.
:param steering_file: Source TOML path used to resolve relative dotenv paths.
:param process_environment: Optional environment snapshot; defaults to ``os.environ``.
:param strict: Raise when any configured entry is invalid; false returns all safe statuses.
:return: Merged values and safe per-file status records.
:raises SteeringEnvironmentError: If the setting or any configured file is invalid.
"""
source = Path(steering_file).expanduser().resolve(strict=False)
shell_environment = dict(os.environ if process_environment is None else process_environment)
entries = config.get('env_files', [])
if not isinstance(entries, list):
raise SteeringEnvironmentError(f"'env_files' in steering file '{source}' must be a list of strings.")
merged: dict[str, str] = {}
statuses: list[EnvironmentFileStatus] = []
errors: list[str] = []
path_context = PathResolutionContext.from_file(source)
for entry in entries:
if not isinstance(entry, str):
message = f"'env_files' in steering file '{source}' contains a non-string path."
errors.append(message)
if strict:
raise SteeringEnvironmentError(message)
statuses.append(EnvironmentFileStatus('<invalid entry>', None, message))
continue
try:
resolved = resolve_path(
entry,
path_context,
validation=PathValidationPolicy.EXISTING_FILE,
env=shell_environment,
)
except PathResolutionError as exc:
message = f"Dotenv path '{entry}' in steering file '{source}' is not a valid existing file."
errors.append(message)
statuses.append(EnvironmentFileStatus(entry, None, message))
if strict:
raise SteeringEnvironmentError(message) from exc
continue
try:
with resolved.open('r', encoding='utf-8') as handle:
bindings = list(parse_stream(handle))
if any(
binding.error
or (binding.key is not None and _ENVIRONMENT_KEY_PATTERN.fullmatch(binding.key) is None)
for binding in bindings
):
raise ValueError('contains invalid dotenv syntax')
parsed = dotenv_values(resolved, encoding='utf-8')
except UnicodeError as exc:
message = f"Dotenv file '{entry}' referenced by steering file '{source}' is not valid UTF-8."
errors.append(message)
statuses.append(EnvironmentFileStatus(entry, resolved, message))
if strict:
raise SteeringEnvironmentError(message) from exc
continue
except ValueError as exc:
message = f"Dotenv file '{entry}' referenced by steering file '{source}' contains invalid syntax."
errors.append(message)
statuses.append(EnvironmentFileStatus(entry, resolved, message))
if strict:
raise SteeringEnvironmentError(message) from exc
continue
except OSError as exc:
message = f"Dotenv file '{entry}' referenced by steering file '{source}' could not be read."
errors.append(message)
statuses.append(EnvironmentFileStatus(entry, resolved, message))
if strict:
raise SteeringEnvironmentError(message) from exc
continue
# A valueless dotenv key is represented by None and means undefined.
merged.update({key: value for key, value in parsed.items() if value is not None})
statuses.append(EnvironmentFileStatus(entry, resolved))
merged.update(shell_environment)
return SteeringEnvironment(merged, tuple(statuses), tuple(errors))