Ensure reset of feature flags and constants after config deployment (#871)

Resolves #872


This pull request introduces a robust and safer mechanism for handling
feature flag and constant overrides during configuration-based
deployments. The main improvement is the replacement of the previous
global override function with a context manager, ensuring that any
changes to feature flags and constants are always reverted after
deployment, even if errors occur. This prevents state leakage between
deployments and improves reliability. The pull request also updates
validation logic, enhances logging, and adds comprehensive tests to
verify correct override and restoration behavior.

Key changes include:

**Override Mechanism Improvements**
- Replaced the global `apply_config_overrides` function with a
`config_overrides_scope` context manager in
`src/fabric_cicd/_common/_config_utils.py`. This ensures that feature
flags and constants are overridden only for the duration of the
deployment and are always restored to their original values afterward,
even if exceptions occur.
[[1]](diffhunk://#diff-a595a996e85e1b2ab0a30945c022432c66c2c63fdd87c97030327c412c713485L150-R197)
[[2]](diffhunk://#diff-a8555a82747092e8d1f2ee36f31c947b4e5a264ed6c6d8de48d8daf850372704L15-R15)
[[3]](diffhunk://#diff-a8555a82747092e8d1f2ee36f31c947b4e5a264ed6c6d8de48d8daf850372704L408-R408)
- Updated all usages and tests to use the new context manager, removing
references to the old function.
[[1]](diffhunk://#diff-591b837760640c6d5c7b21d70575e6a262d56881f9ff3f993dbf10064c61d468L12-R14)
[[2]](diffhunk://#diff-591b837760640c6d5c7b21d70575e6a262d56881f9ff3f993dbf10064c61d468L375-R531)

**Validation and Logging Enhancements**
- Improved validation logic for the `features` and `constants` sections,
including more granular checks for environment-specific mappings and
individual constant keys.
[[1]](diffhunk://#diff-2b2dbdc5a0a864161ffbaf5211918409f6b2a5ea15a9e23d17ff803e6f9b2834L347-R347)
[[2]](diffhunk://#diff-2b2dbdc5a0a864161ffbaf5211918409f6b2a5ea15a9e23d17ff803e6f9b2834L356-R365)
[[3]](diffhunk://#diff-2b2dbdc5a0a864161ffbaf5211918409f6b2a5ea15a9e23d17ff803e6f9b2834L1012-R1047)
- Changed the logging level for skipped `features` and `constants` to
warnings instead of debug, making potential misconfigurations more
visible.
[[1]](diffhunk://#diff-1482256f9dd419ccd2b7281c95d3b1fa902a4af31630ad0e9dbca891241d5f2fL305-R306)
[[2]](diffhunk://#diff-2b2dbdc5a0a864161ffbaf5211918409f6b2a5ea15a9e23d17ff803e6f9b2834L1084-L1097)

**Testing Improvements**
- Added extensive tests to verify that feature flags and constants are
correctly overridden within the context and always restored after,
including when exceptions are raised. Tests also check that user-set
values are preserved and that environment-specific overrides work as
intended.

**Documentation**
- Updated documentation to reflect the new override behavior and logging
changes for `features` and `constants`.

**Bug Fix**
- Fixes issue #872: Ensures that override behavior for feature flags and
constants in config deployment is correct and state-safe.
(.changes/unreleased/fixed-20260311-110454.yaml)

---------

Co-authored-by: Shira Sassoon <shirasassoon@microsoft.com>

Author: shirasassoon

.changes/unreleased/fixed-20260311-110454.yaml

@@ -0,0 +1,8 @@
+kind: fixed
+body: Fix override behavior of feature flags and constants in config deployment
+time: 2026-03-11T11:04:54.7911345+02:00
+custom:
+    Author: shirasassoon
+    AuthorLink: https://github.com/shirasassoon
+    Issue: "872"
+    IssueLink: https://github.com/microsoft/fabric-cicd/issues/872

docs/how_to/config_deployment.md

@@ -302,8 +302,8 @@ Fields are categorized as **required** or **optional**, which affects how missin
 | `unpublish.exclude_regex`               | ❌       | Debug logged, setting skipped   |
 | `unpublish.items_to_include`            | ❌       | Debug logged, setting skipped   |
 | `unpublish.skip`                        | ❌       | Defaults to `False`             |
-| `features`                              | ❌       | Debug logged, setting skipped   |
-| `constants`                             | ❌       | Debug logged, setting skipped   |
+| `features`                              | ❌       | Warning logged, setting skipped |
+| `constants`                             | ❌       | Warning logged, setting skipped |
 
 ### Selective Environment Configuration
 

src/fabric_cicd/_common/_config_utils.py

@@ -3,7 +3,9 @@
 
 """Utilities for YAML-based deployment configuration."""
 
+import contextlib
 import logging
+from collections.abc import Generator
 from typing import Optional, Union
 
 from fabric_cicd import constants
@@ -147,30 +149,47 @@ def extract_unpublish_settings(config: dict, environment: str) -> dict:
     return settings
 
 
-def apply_config_overrides(config: dict, environment: str) -> None:
-    """Apply feature flags and constants overrides from config.
+@contextlib.contextmanager
+def config_overrides_scope(config: dict, environment: str) -> Generator[None, None, None]:
+    """
+    Context manager that applies feature flags and constants
+    overrides from config  and guarantees cleanup.
+
+    Feature flags and constants are restored to their pre-call values on exit,
+    ensuring no state leaks between deployments.
 
     Args:
         config: Configuration dictionary
         environment: Target environment for deployment
     """
-    if "features" in config:
-        features = config["features"]
-        features_list = features.get(environment, []) if isinstance(features, dict) else features
-
-        for feature in features_list:
-            constants.FEATURE_FLAG.add(feature)
-            logger.info(f"Enabled feature flag: {feature}")
-
-    if "constants" in config:
-        constants_section = config["constants"]
-        # Check if it's an environment mapping (all values are dicts)
-        if all(isinstance(v, dict) for v in constants_section.values()):
-            constants_dict = constants_section.get(environment, {})
-        else:
-            constants_dict = constants_section
-
-        for key, value in constants_dict.items():
-            if hasattr(constants, key):
-                setattr(constants, key, value)
-                logger.warning(f"Override constant {key} = {value}")
+    # Snapshot current state before any changes
+    original_feature_flags = constants.FEATURE_FLAG.copy()
+    overridden_keys = {}
+
+    try:
+        # Set feature flags
+        if "features" in config:
+            features = config["features"]
+            features_list = features.get(environment, []) if isinstance(features, dict) else features
+            for feature in features_list:
+                constants.FEATURE_FLAG.add(feature)
+                logger.info(f"Enabled feature flag: {feature}")
+
+        # Apply constants overrides
+        if "constants" in config:
+            constants_section = config["constants"]
+            for key in list(constants_section.keys()):
+                value = get_config_value(constants_section, key, environment)
+                if value is not None and hasattr(constants, key):
+                    overridden_keys[key] = getattr(constants, key)
+                    setattr(constants, key, value)
+                    logger.warning(f"Override constant {key} = {value}")
+        yield
+
+    finally:
+        # Restore original state — guaranteed even if deployment raises
+        constants.FEATURE_FLAG.clear()
+        constants.FEATURE_FLAG.update(original_feature_flags)
+
+        for key, original_value in overridden_keys.items():
+            setattr(constants, key, original_value)

src/fabric_cicd/_common/_config_validator.py

@@ -344,7 +344,7 @@ def _validate_environment_exists(self) -> None:
             if any(
                 field_name in section and isinstance(section[field_name], dict)
                 for section, field_name, _, _, _ in _get_config_fields(self.config)
-                if not (field_name == "constants" and _is_regular_constants_dict(section.get(field_name, {})))
+                if field_name != "constants"
             ):
                 self.errors.append(constants.CONFIG_VALIDATION_MSGS["environment"]["no_env_with_mappings"])
             return
@@ -353,8 +353,16 @@ def _validate_environment_exists(self) -> None:
         for section, field_name, display_name, is_required, log_warning in _get_config_fields(self.config):
             if field_name in section:
                 field_value = section[field_name]
-                # Handle constants special case
-                if field_name == "constants" and _is_regular_constants_dict(field_value):
+                # Handle constants special case — check each constant's value individually
+                if field_name == "constants":
+                    if isinstance(field_value, dict):
+                        for const_key, const_val in field_value.items():
+                            if isinstance(const_val, dict) and self.environment not in const_val:
+                                available_envs = list(const_val.keys())
+                                logger.warning(
+                                    f"Environment '{self.environment}' not found in 'constants.{const_key}'. "
+                                    f"Available environments: {available_envs}. This constant will be skipped."
+                                )
                     continue
 
                 # If it's a dict (environment mapping), check if target environment exists
@@ -593,8 +601,8 @@ def _resolve_path_field(
 
         for env_key, path_str in paths_to_resolve.items():
             try:
-                path = Path(path_str)
                 env_desc = f" for environment '{env_key}'" if env_key != "_default" else ""
+                path = Path(path_str)
 
                 if path.is_absolute():
                     resolved_path = path
@@ -652,13 +660,9 @@ def _resolve_path_field(
                 if isinstance(field_value, str):
                     if section_name:
                         self.config[section_name][field_name] = str(resolved_path)
-                    else:
-                        self.config[field_name] = str(resolved_path)
                 else:
                     if section_name:
                         self.config[section_name][field_name][env_key] = str(resolved_path)
-                    else:
-                        self.config[field_name][env_key] = str(resolved_path)
 
             except (OSError, ValueError) as e:
                 self.errors.append(
@@ -975,11 +979,6 @@ def _validate_features_section(self, features: any) -> None:
 
             # Validate each environment's features list
             for env, features_list in features.items():
-                if not features_list:
-                    self.errors.append(
-                        constants.CONFIG_VALIDATION_MSGS["operation"]["empty_section_env"].format("features", env)
-                    )
-                    continue
                 self._validate_features_list(features_list, f"features.{env}")
             return
 
@@ -1009,49 +1008,45 @@ def _validate_constants_section(self, constants_section: any) -> None:
             )
             return
 
-        # Check if all values are dictionaries (contains environment mapping)
-        if constants_section and all(isinstance(value, dict) for value in constants_section.values()):
-            # Validate environment mapping
-            if not self._validate_environment_mapping(constants_section, "constants", dict):
-                return
-
-            # Validate each environment's constants dictionary
-            for env, env_constants in constants_section.items():
-                if not env_constants:
-                    self.errors.append(
-                        constants.CONFIG_VALIDATION_MSGS["operation"]["empty_section_env"].format("constants", env)
-                    )
-                    continue
-                self._validate_constants_dict(env_constants, f"constants.{env}")
-        else:
-            # Simple constants dictionary
-            self._validate_constants_dict(constants_section, "constants")
-
-    def _validate_constants_dict(self, constants_dict: dict, context: str) -> None:
-        """Validate a constants dictionary with proper context for error messages."""
-        for key, value in constants_dict.items():
+        # Validate each constant key individually — values can be flat or per-key env mappings
+        for key, value in constants_section.items():
             if not isinstance(key, str) or not key.strip():
                 self.errors.append(
-                    constants.CONFIG_VALIDATION_MSGS["operation"]["invalid_constant_key"].format(context, key)
+                    constants.CONFIG_VALIDATION_MSGS["operation"]["invalid_constant_key"].format("constants", key)
                 )
                 continue
 
             # Validate that the constant exists in the constants module
             if not hasattr(constants, key):
                 self.errors.append(
-                    constants.CONFIG_VALIDATION_MSGS["operation"]["unknown_constant"].format(key, context)
+                    constants.CONFIG_VALIDATION_MSGS["operation"]["unknown_constant"].format(key, "constants")
                 )
-                continue
 
-            # Validate URL constants
-            if key in _URL_CONSTANTS:
-                if not isinstance(value, str):
-                    self.errors.append(f"'{key}' in '{context}' must be a string URL, got {type(value).__name__}")
-                    continue
-                try:
-                    validate_api_url(value, f"{context}.{key}")
-                except InputError as e:
-                    self.errors.append(str(e))
+            if isinstance(value, dict):
+                # Per-key environment mapping: { KEY: { dev: val, prod: val } }
+                for env, env_value in value.items():
+                    if not isinstance(env, str) or not env.strip():
+                        self.errors.append(
+                            constants.CONFIG_VALIDATION_MSGS["environment"]["invalid_env_key"].format(
+                                f"constants.{key}", type(env).__name__
+                            )
+                        )
+                        continue
+                    self._validate_single_constant(key, env_value, f"constants.{key}.{env}")
+            else:
+                # Flat value: { KEY: val }
+                self._validate_single_constant(key, value, f"constants.{key}")
+
+    def _validate_single_constant(self, key: str, value: any, context: str) -> None:
+        """Validate a single constant value."""
+        if key in _URL_CONSTANTS:
+            if not isinstance(value, str):
+                self.errors.append(f"'{context}' must be a string URL, got {type(value).__name__}")
+                return
+            try:
+                validate_api_url(value, context)
+            except InputError as e:
+                self.errors.append(str(e))
 
 
 def _get_config_fields(config: dict) -> list[tuple[dict, str, str, bool, bool]]:
@@ -1081,20 +1076,12 @@ def _get_config_fields(config: dict) -> list[tuple[dict, str, str, bool, bool]]:
         (config.get("unpublish", {}), "exclude_regex", "unpublish.exclude_regex", False, False),
         (config.get("unpublish", {}), "items_to_include", "unpublish.items_to_include", False, False),
         (config.get("unpublish", {}), "skip", "unpublish.skip", False, False),
-        # Top-level sections - optional (debug if missing)
-        (config, "features", "features", False, False),
-        (config, "constants", "constants", False, False),
+        # Top-level sections - optional (warn if missing)
+        (config, "features", "features", False, True),
+        (config, "constants", "constants", False, True),
     ]
 
 
-def _is_regular_constants_dict(constants_value: dict) -> bool:
-    """Check if constants section is a regular dict (not environment mapping)."""
-    if not isinstance(constants_value, dict) or not constants_value:
-        return True
-    # Environment mapping if ALL values are dicts, regular dict otherwise
-    return not all(isinstance(value, dict) for value in constants_value.values())
-
-
 def _find_git_root(path: Path) -> Optional[Path]:
     """Find the git repository root for a given path."""
     current = path.resolve()

src/fabric_cicd/publish.py

@@ -12,7 +12,7 @@
 import fabric_cicd._items as items
 from fabric_cicd import constants
 from fabric_cicd._common._config_utils import (
-    apply_config_overrides,
+    config_overrides_scope,
     extract_publish_settings,
     extract_unpublish_settings,
     extract_workspace_settings,
@@ -405,42 +405,41 @@ def deploy_with_config(
     unpublish_settings = extract_unpublish_settings(config, environment)
 
     # Apply feature flags and constants if specified
-    apply_config_overrides(config, environment)
-
-    # Create FabricWorkspace object with extracted settings
-    workspace = FabricWorkspace(
-        repository_directory=workspace_settings["repository_directory"],
-        item_type_in_scope=workspace_settings.get("item_types_in_scope"),
-        environment=environment,
-        workspace_id=workspace_settings.get("workspace_id"),
-        workspace_name=workspace_settings.get("workspace_name"),
-        token_credential=token_credential,
-        parameter_file_path=workspace_settings.get("parameter_file_path"),
-    )
-    # Execute deployment operations based on skip settings
-    if not publish_settings.get("skip", False):
-        publish_all_items(
-            workspace,
-            item_name_exclude_regex=publish_settings.get("exclude_regex"),
-            folder_path_exclude_regex=publish_settings.get("folder_exclude_regex"),
-            folder_path_to_include=publish_settings.get("folder_path_to_include"),
-            items_to_include=publish_settings.get("items_to_include"),
-            shortcut_exclude_regex=publish_settings.get("shortcut_exclude_regex"),
+    with config_overrides_scope(config, environment):
+        # Create FabricWorkspace object with extracted settings
+        workspace = FabricWorkspace(
+            repository_directory=workspace_settings["repository_directory"],
+            item_type_in_scope=workspace_settings.get("item_types_in_scope"),
+            environment=environment,
+            workspace_id=workspace_settings.get("workspace_id"),
+            workspace_name=workspace_settings.get("workspace_name"),
+            token_credential=token_credential,
+            parameter_file_path=workspace_settings.get("parameter_file_path"),
         )
-    else:
-        logger.info(f"Skipping publish operation for environment '{environment}'")
-
-    if not unpublish_settings.get("skip", False):
-        unpublish_all_orphan_items(
-            workspace,
-            item_name_exclude_regex=unpublish_settings.get("exclude_regex", "^$"),
-            items_to_include=unpublish_settings.get("items_to_include"),
+        # Execute deployment operations based on skip settings
+        if not publish_settings.get("skip", False):
+            publish_all_items(
+                workspace,
+                item_name_exclude_regex=publish_settings.get("exclude_regex"),
+                folder_path_exclude_regex=publish_settings.get("folder_exclude_regex"),
+                folder_path_to_include=publish_settings.get("folder_path_to_include"),
+                items_to_include=publish_settings.get("items_to_include"),
+                shortcut_exclude_regex=publish_settings.get("shortcut_exclude_regex"),
+            )
+        else:
+            logger.info(f"Skipping publish operation for environment '{environment}'")
+
+        if not unpublish_settings.get("skip", False):
+            unpublish_all_orphan_items(
+                workspace,
+                item_name_exclude_regex=unpublish_settings.get("exclude_regex", "^$"),
+                items_to_include=unpublish_settings.get("items_to_include"),
+            )
+        else:
+            logger.info(f"Skipping unpublish operation for environment '{environment}'")
+
+        logger.info("Config-based deployment completed successfully")
+        return DeploymentResult(
+            status=DeploymentStatus.COMPLETED,
+            message="Deployment completed successfully",
         )
-    else:
-        logger.info(f"Skipping unpublish operation for environment '{environment}'")
-
-    logger.info("Config-based deployment completed successfully")
-    return DeploymentResult(
-        status=DeploymentStatus.COMPLETED,
-        message="Deployment completed successfully",
-    )