diff --git a/samcli/lib/config/parameter_loaders.py b/samcli/lib/config/parameter_loaders.py new file mode 100644 index 00000000000..5b7abc3c937 --- /dev/null +++ b/samcli/lib/config/parameter_loaders.py @@ -0,0 +1,359 @@ +""" +Parameter file loading utilities for SAM CLI configuration +""" + +import json +import logging +import os +from pathlib import Path +from typing import Dict, List, Optional, Tuple + +from ruamel.yaml import YAML, YAMLError + +from samcli.lib.config.exceptions import FileParseException + +LOG = logging.getLogger(__name__) + + +class ParameterFileLoader: + """ + Loads parameters from external parameter files. + Supports JSON, YAML, and ENV file formats. + Follows existing SAM CLI FileManager patterns. + """ + + @staticmethod + def is_file_url(parameter_value: Optional[str]) -> bool: + """ + Check if a parameter value is a file:// URL. + + Parameters + ---------- + parameter_value : str, optional + Parameter value to check + + Returns + ------- + bool + True if parameter_value is a file:// URL + """ + if not isinstance(parameter_value, str): + return False + return parameter_value.startswith("file://") + + @staticmethod + def parse_file_url(file_url: str) -> str: + """ + Parse file:// URL and return the file path with environment variable expansion. + + Parameters + ---------- + file_url : str + File URL in format file://path/to/file.ext + + Returns + ------- + str + Expanded file path + + Raises + ------ + ValueError + If URL format is invalid + """ + if not ParameterFileLoader.is_file_url(file_url): + raise ValueError(f"Invalid file URL format: {file_url}") + + # Extract path by removing file:// prefix + # Don't use urlparse because it doesn't handle env vars and relative paths correctly + file_path = file_url[7:] # Remove 'file://' prefix + + # Handle Windows paths (file:///C:/path/to/file) + if os.name == "nt" and file_path.startswith("/") and ":" in file_path[1:3]: + file_path = file_path[1:] + + # Expand environment variables before returning + expanded_path = os.path.expandvars(file_path) + + LOG.debug(f"Parsed file URL '{file_url}' to path '{expanded_path}'") + return expanded_path + + @staticmethod + def load_from_file(file_path: str) -> Dict: + """ + Load parameters from a file based on its extension. + + Supported formats: + - .json: JSON format + - .yaml/.yml: YAML format + - .env: Environment variable format + + Parameters + ---------- + file_path : str + Path to the parameter file + + Returns + ------- + Dict + Dictionary of loaded parameters + + Raises + ------ + FileNotFoundError + If file doesn't exist + FileParseException + If file format is invalid or unsupported + """ + path_obj = Path(file_path) + + if not path_obj.exists(): + raise FileNotFoundError(f"Parameter file not found: {file_path}") + + if not path_obj.is_file(): + raise FileParseException(f"Parameter path is not a file: {file_path}") + + extension = path_obj.suffix.lower() + + try: + if extension == ".json": + return ParameterFileLoader._load_json_file(path_obj) + elif extension in [".yaml", ".yml"]: + return ParameterFileLoader._load_yaml_file(path_obj) + elif extension == ".env": + return ParameterFileLoader._load_env_file(path_obj) + else: + raise FileParseException( + f"Unsupported parameter file format: {extension}. " f"Supported formats: .json, .yaml, .yml, .env" + ) + except Exception as e: + if isinstance(e, (FileParseException, FileNotFoundError)): + raise + raise FileParseException(f"Failed to load parameter file '{file_path}': {str(e)}") from e + + @staticmethod + def _load_json_file(file_path: Path) -> Dict: + """ + Load parameters from JSON file. + + Parameters + ---------- + file_path : Path + Path to JSON file + + Returns + ------- + Dict + Dictionary of parameters + """ + try: + content = file_path.read_text(encoding="utf-8") + params = json.loads(content) + + if not isinstance(params, dict): + raise FileParseException(f"JSON parameter file must contain an object, got {type(params)}") + + LOG.debug(f"Loaded {len(params)} parameters from JSON file: {file_path}") + return params + + except json.JSONDecodeError as e: + raise FileParseException(f"Invalid JSON in parameter file '{file_path}': {str(e)}") from e + + @staticmethod + def _load_yaml_file(file_path: Path) -> Dict: + """ + Load parameters from YAML file. + + Parameters + ---------- + file_path : Path + Path to YAML file + + Returns + ------- + Dict + Dictionary of parameters + """ + yaml = YAML(typ="safe", pure=True) + + try: + content = file_path.read_text(encoding="utf-8") + params = yaml.load(content) + + if params is None: + return {} + + if not isinstance(params, dict): + raise FileParseException(f"YAML parameter file must contain a mapping, got {type(params)}") + + LOG.debug(f"Loaded {len(params)} parameters from YAML file: {file_path}") + return params + + except YAMLError as e: + raise FileParseException(f"Invalid YAML in parameter file '{file_path}': {str(e)}") from e + + @staticmethod + def _load_env_file(file_path: Path) -> Dict: + """ + Load parameters from environment variable file. + + Format: + KEY1=value1 + KEY2=value2 + # Comments are supported + MULTILINE_KEY="line1 + line2 + line3" + + Parameters + ---------- + file_path : Path + Path to ENV file + + Returns + ------- + Dict + Dictionary of parameters + """ + params: Dict[str, str] = {} + + try: + content = file_path.read_text(encoding="utf-8") + lines = content.splitlines() + + current_key: Optional[str] = None + current_value: List[str] = [] + in_multiline = False + + for line_num, raw_line in enumerate(lines, 1): + line = raw_line.rstrip() + + # Skip empty lines and comments + if not line or line.startswith("#"): + continue + + # Handle multiline values + if in_multiline: + if line.endswith('"') and not line.endswith('\\"'): + # End of multiline value + current_value.append(line[:-1]) # Remove closing quote + if current_key is not None: + params[current_key] = "\n".join(current_value) + current_key = None + current_value = [] + in_multiline = False + else: + current_value.append(line) + continue + + # Parse key=value pairs + if "=" not in line: + LOG.warning(f"Skipping invalid line {line_num} in {file_path}: {line}") + continue + + key, value = line.split("=", 1) + key = key.strip() + value = value.strip() + + # Handle quoted values + if value.startswith('"'): + if value.endswith('"') and len(value) > 1 and not value.endswith('\\"'): + # Single line quoted value + params[key] = value[1:-1] # Remove quotes + else: + # Start of multiline value + current_key = key + current_value = [value[1:]] # Remove opening quote + in_multiline = True + else: + params[key] = value + + if in_multiline: + raise FileParseException(f"Unterminated quoted value for key '{current_key}' in {file_path}") + + LOG.debug(f"Loaded {len(params)} parameters from ENV file: {file_path}") + return params + + except Exception as e: + if isinstance(e, FileParseException): + raise + raise FileParseException(f"Failed to parse ENV file '{file_path}': {str(e)}") from e + + @staticmethod + def resolve_parameter_files(parameter_overrides: Optional[str]) -> Tuple[Dict, Dict]: + """ + Resolve parameter overrides that may contain file:// URLs. + + Parameters + ---------- + parameter_overrides : str, optional + Parameter overrides string that may contain file:// URLs + + Returns + ------- + Tuple[Dict, Dict] + Tuple of (direct_parameters, file_parameters) + - direct_parameters: Parameters specified directly as Key=Value + - file_parameters: Parameters loaded from files + """ + if not parameter_overrides: + return {}, {} + + direct_params = {} + file_params = {} + + # Split parameter overrides by spaces, handling quoted values + import shlex + + try: + parts = shlex.split(parameter_overrides) + except ValueError: + # Fall back to simple split if shlex fails + parts = parameter_overrides.split() + + for part in parts: + if ParameterFileLoader.is_file_url(part): + # Load parameters from file + try: + file_path = ParameterFileLoader.parse_file_url(part) + loaded_params = ParameterFileLoader.load_from_file(file_path) + file_params.update(loaded_params) + LOG.info(f"Loaded {len(loaded_params)} parameters from file: {file_path}") + except Exception as e: + LOG.error(f"Failed to load parameters from {part}: {str(e)}") + raise + elif "=" in part: + # Direct parameter specification + key, value = part.split("=", 1) + direct_params[key.strip()] = value.strip() + else: + LOG.warning(f"Skipping invalid parameter format: {part}") + + return direct_params, file_params + + @staticmethod + def expand_environment_variables(params: Dict) -> Dict: + """ + Expand environment variables in parameter values. + + Supports ${VAR_NAME} and $VAR_NAME syntax. + + Parameters + ---------- + params : Dict + Dictionary of parameters + + Returns + ------- + Dict + Dictionary with expanded environment variables + """ + expanded = {} + + for key, value in params.items(): + if isinstance(value, str): + expanded[key] = os.path.expandvars(value) + else: + expanded[key] = value + + return expanded diff --git a/samcli/lib/config/parameter_merger.py b/samcli/lib/config/parameter_merger.py new file mode 100644 index 00000000000..bed69945ae5 --- /dev/null +++ b/samcli/lib/config/parameter_merger.py @@ -0,0 +1,218 @@ +""" +Parameter merging utilities for SAM CLI configuration +""" + +import logging +from typing import Dict, Optional + +LOG = logging.getLogger(__name__) + + +class ParameterMerger: + """ + Handles parameter merging with precedence rules: CLI > file > config > default + Follows existing SAM CLI patterns for parameter processing + """ + + @staticmethod + def merge_parameters( + config_params: Optional[Dict] = None, + cli_params: Optional[Dict] = None, + file_params: Optional[Dict] = None, + ) -> Dict: + """ + Merge parameters with precedence rules. + + Precedence (highest to lowest): + 1. CLI parameters (--parameter-overrides) + 2. File parameters (--parameter-overrides file://params.json) + 3. Config parameters ([env.command.parameters.template_parameters]) + + Parameters + ---------- + config_params : Dict, optional + Parameters from samconfig.toml template_parameters section + cli_params : Dict, optional + Parameters from CLI --parameter-overrides + file_params : Dict, optional + Parameters loaded from external files + + Returns + ------- + Dict + Merged parameter dictionary with CLI taking precedence + """ + merged = {} + + # Start with config parameters (lowest precedence) + if config_params: + merged.update(config_params) + LOG.debug(f"Added config parameters: {list(config_params.keys())}") + + # Add file parameters (medium precedence) + if file_params: + merged.update(file_params) + LOG.debug(f"Added file parameters: {list(file_params.keys())}") + + # Add CLI parameters (highest precedence) + if cli_params: + merged.update(cli_params) + LOG.debug(f"Added CLI parameters: {list(cli_params.keys())}") + + LOG.debug(f"Final merged parameters: {list(merged.keys())}") + return merged + + @staticmethod + def merge_tags( + config_tags: Optional[Dict] = None, + cli_tags: Optional[Dict] = None, + file_tags: Optional[Dict] = None, + ) -> Dict: + """ + Merge tags with same precedence rules as parameters. + + Precedence (highest to lowest): + 1. CLI tags (--tags) + 2. File tags (--tags file://tags.json) + 3. Config tags ([env.command.parameters.template_tags]) + + Parameters + ---------- + config_tags : Dict, optional + Tags from samconfig.toml template_tags section + cli_tags : Dict, optional + Tags from CLI --tags + file_tags : Dict, optional + Tags loaded from external files + + Returns + ------- + Dict + Merged tag dictionary with CLI taking precedence + """ + merged = {} + + # Start with config tags (lowest precedence) + if config_tags: + merged.update(config_tags) + LOG.debug(f"Added config tags: {list(config_tags.keys())}") + + # Add file tags (medium precedence) + if file_tags: + merged.update(file_tags) + LOG.debug(f"Added file tags: {list(file_tags.keys())}") + + # Add CLI tags (highest precedence) + if cli_tags: + merged.update(cli_tags) + LOG.debug(f"Added CLI tags: {list(cli_tags.keys())}") + + LOG.debug(f"Final merged tags: {list(merged.keys())}") + return merged + + @staticmethod + def format_for_cloudformation(parameters: Dict) -> Dict: + """ + Format merged parameters for CloudFormation deployment. + Converts our merged dict to the format expected by CloudFormation. + + Parameters + ---------- + parameters : Dict + Merged parameters dictionary + + Returns + ------- + Dict + Parameters formatted for CloudFormation (Key=Value format) + """ + if not parameters: + return {} + + # Convert to CloudFormation parameter format + formatted = {} + for key, value in parameters.items(): + # Handle different value types + if isinstance(value, (dict, list)): + # Convert complex types to JSON strings + import json + + formatted[key] = json.dumps(value) + elif value is None: + formatted[key] = "" + else: + formatted[key] = str(value) + + return formatted + + @staticmethod + def parse_legacy_parameter_string(parameter_string: str) -> Dict: + """ + Parse legacy parameter_overrides string format. + Handles space-separated key=value pairs with proper quoting support. + + Parameters + ---------- + parameter_string : str + String in format "Key1=Value1 Key2=Value2" + + Returns + ------- + Dict + Dictionary of parsed parameters + """ + if not parameter_string or not isinstance(parameter_string, str): + return {} + + params = {} + import shlex + + try: + # Use shlex to properly handle quoted values + pairs = shlex.split(parameter_string) + for pair in pairs: + if "=" in pair: + key, value = pair.split("=", 1) + params[key.strip()] = value.strip() + else: + LOG.warning(f"Skipping invalid parameter format: {pair}") + except ValueError as e: + LOG.warning(f"Failed to parse parameter string with shlex: {e}") + # Fall back to simple split for malformed strings + pairs = parameter_string.split() + for pair in pairs: + if "=" in pair: + key, value = pair.split("=", 1) + params[key.strip()] = value.strip() + + return params + + @staticmethod + def validate_parameters(parameters: Dict, template_parameters: Optional[Dict] = None) -> Dict: + """ + Validate merged parameters against template parameter definitions. + + Parameters + ---------- + parameters : Dict + Merged parameters to validate + template_parameters : Dict, optional + Template parameter definitions from CloudFormation template + + Returns + ------- + Dict + Validated parameters (may remove invalid ones with warnings) + """ + if not template_parameters: + return parameters + + validated = {} + + for key, value in parameters.items(): + if key in template_parameters: + validated[key] = value + else: + LOG.warning(f"Parameter '{key}' not found in template parameters. Skipping.") + + return validated diff --git a/samcli/lib/config/samconfig.py b/samcli/lib/config/samconfig.py index 2cd0d41199a..ee36f7d9744 100644 --- a/samcli/lib/config/samconfig.py +++ b/samcli/lib/config/samconfig.py @@ -20,6 +20,10 @@ DEFAULT_ENV = "default" DEFAULT_GLOBAL_CMDNAME = "global" +# New section names for enhanced parameter support +TEMPLATE_PARAMETERS_SECTION = "parameters.template_parameters" +TEMPLATE_TAGS_SECTION = "parameters.template_tags" + class SamConfig: """ @@ -283,6 +287,155 @@ def _version_sanity_check(version: Any) -> None: if not isinstance(version, float): raise SamConfigVersionException(f"'{VERSION_KEY}' key is not present or is in unrecognized format. ") + def get_template_parameters(self, cmd_names, env=DEFAULT_ENV): + """ + Gets template parameters from the new [env.command.parameters.template_parameters] section format. + Falls back to parsing legacy parameter_overrides string if new format not found. + + Parameters + ---------- + cmd_names : list(str) + List representing the entire command. Ex: ["deploy"] + env : str + Optional, Name of the environment + + Returns + ------- + dict + Dictionary of template parameters. Empty dict if none found. + """ + env = env or DEFAULT_ENV + + try: + # Try new format first + template_params = self.get_all(cmd_names, TEMPLATE_PARAMETERS_SECTION, env) + if template_params: + return template_params + except KeyError: + pass + + # Fall back to parsing legacy parameter_overrides + try: + legacy_params = self.get_all(cmd_names, "parameters", env) + parameter_overrides = legacy_params.get("parameter_overrides", "") + if parameter_overrides: + return self._parse_parameter_overrides(parameter_overrides) + except KeyError: + pass + + return {} + + def get_template_tags(self, cmd_names, env=DEFAULT_ENV): + """ + Gets template tags from the new [env.command.parameters.template_tags] section format. + Falls back to parsing legacy tags string if new format not found. + + Parameters + ---------- + cmd_names : list(str) + List representing the entire command. Ex: ["deploy"] + env : str + Optional, Name of the environment + + Returns + ------- + dict + Dictionary of template tags. Empty dict if none found. + """ + env = env or DEFAULT_ENV + + try: + # Try new format first + template_tags = self.get_all(cmd_names, TEMPLATE_TAGS_SECTION, env) + if template_tags: + return template_tags + except KeyError: + pass + + # Fall back to parsing legacy tags + try: + legacy_params = self.get_all(cmd_names, "parameters", env) + tags = legacy_params.get("tags", "") + if tags: + return self._parse_parameter_overrides(tags) + except KeyError: + pass + + return {} + + def put_template_parameter(self, cmd_names, key, value, env=DEFAULT_ENV): + """ + Stores a template parameter in the new [env.command.parameters.template_parameters] section format. + + Parameters + ---------- + cmd_names : list(str) + List representing the entire command. Ex: ["deploy"] + key : str + Parameter name + value : Any + Parameter value + env : str + Optional, Name of the environment + """ + self.put(cmd_names, TEMPLATE_PARAMETERS_SECTION, key, value, env) + + def put_template_tag(self, cmd_names, key, value, env=DEFAULT_ENV): + """ + Stores a template tag in the new [env.command.parameters.template_tags] section format. + + Parameters + ---------- + cmd_names : list(str) + List representing the entire command. Ex: ["deploy"] + key : str + Tag name + value : Any + Tag value + env : str + Optional, Name of the environment + """ + self.put(cmd_names, TEMPLATE_TAGS_SECTION, key, value, env) + + @staticmethod + def _parse_parameter_overrides(parameter_overrides_str): + """ + Parse legacy parameter_overrides string format into dictionary. + Handles space-separated key=value pairs. + + Parameters + ---------- + parameter_overrides_str : str + String in format "Key1=Value1 Key2=Value2" + + Returns + ------- + dict + Dictionary of parsed parameters + """ + if not parameter_overrides_str or not isinstance(parameter_overrides_str, str): + return {} + + params = {} + # Split by spaces, but handle quoted values + import shlex + + try: + pairs = shlex.split(parameter_overrides_str) + for pair in pairs: + if "=" in pair: + key, value = pair.split("=", 1) + params[key.strip()] = value.strip() + except ValueError: + # If shlex parsing fails, fall back to simple split + pairs = parameter_overrides_str.split() + for pair in pairs: + if "=" in pair: + key, value = pair.split("=", 1) + params[key.strip()] = value.strip() + + return params + @staticmethod def to_key(cmd_names: Iterable[str]) -> str: # construct a parsed name that is of the format: a_b_c_d diff --git a/tests/unit/lib/samconfig/test_new_schema_integration.py b/tests/unit/lib/samconfig/test_new_schema_integration.py new file mode 100644 index 00000000000..4ceb610e1a1 --- /dev/null +++ b/tests/unit/lib/samconfig/test_new_schema_integration.py @@ -0,0 +1,274 @@ +""" +Unit tests for new schema integration in SamConfig +""" + +import os +import tempfile +import unittest +from pathlib import Path + +from samcli.lib.config.samconfig import SamConfig, DEFAULT_ENV, TEMPLATE_PARAMETERS_SECTION, TEMPLATE_TAGS_SECTION + + +class TestSamConfigNewSchema(unittest.TestCase): + """Test cases for new schema support in SamConfig""" + + def setUp(self): + """Set up test fixtures""" + self.temp_dir = tempfile.mkdtemp() + self.samconfig = SamConfig(self.temp_dir) + + def tearDown(self): + """Clean up test fixtures""" + if self.samconfig.exists(): + os.remove(self.samconfig.path()) + import shutil + + shutil.rmtree(self.temp_dir, ignore_errors=True) + + def test_put_get_template_parameter(self): + """Test storing and retrieving template parameters""" + cmd_names = ["deploy"] + key = "TemplateParam1" + value = "TemplateValue1" + + # Store parameter + self.samconfig.put_template_parameter(cmd_names, key, value) + self.samconfig.flush() + + # Retrieve parameter + params = self.samconfig.get_template_parameters(cmd_names) + self.assertEqual(params[key], value) + + def test_put_get_template_tag(self): + """Test storing and retrieving template tags""" + cmd_names = ["deploy"] + key = "Environment" + value = "production" + + # Store tag + self.samconfig.put_template_tag(cmd_names, key, value) + self.samconfig.flush() + + # Retrieve tag + tags = self.samconfig.get_template_tags(cmd_names) + self.assertEqual(tags[key], value) + + def test_get_template_parameters_multiple(self): + """Test retrieving multiple template parameters""" + cmd_names = ["deploy"] + params = {"Param1": "Value1", "Param2": "Value2", "Param3": 123} + + # Store multiple parameters + for key, value in params.items(): + self.samconfig.put_template_parameter(cmd_names, key, value) + self.samconfig.flush() + + # Retrieve all parameters + result = self.samconfig.get_template_parameters(cmd_names) + self.assertEqual(result, params) + + def test_get_template_tags_multiple(self): + """Test retrieving multiple template tags""" + cmd_names = ["deploy"] + tags = {"Environment": "prod", "Project": "MyApp", "CostCenter": "Engineering"} + + # Store multiple tags + for key, value in tags.items(): + self.samconfig.put_template_tag(cmd_names, key, value) + self.samconfig.flush() + + # Retrieve all tags + result = self.samconfig.get_template_tags(cmd_names) + self.assertEqual(result, tags) + + def test_get_template_parameters_different_environments(self): + """Test template parameters in different environments""" + cmd_names = ["deploy"] + + # Store parameters in different environments + self.samconfig.put_template_parameter(cmd_names, "Param1", "DefaultValue", DEFAULT_ENV) + self.samconfig.put_template_parameter(cmd_names, "Param1", "StagingValue", "staging") + self.samconfig.flush() + + # Retrieve from different environments + default_params = self.samconfig.get_template_parameters(cmd_names, DEFAULT_ENV) + staging_params = self.samconfig.get_template_parameters(cmd_names, "staging") + + self.assertEqual(default_params["Param1"], "DefaultValue") + self.assertEqual(staging_params["Param1"], "StagingValue") + + def test_get_template_parameters_fallback_to_legacy(self): + """Test fallback to legacy parameter_overrides format""" + cmd_names = ["deploy"] + legacy_string = "Param1=Value1 Param2=Value2" + + # Store in legacy format + self.samconfig.put(cmd_names, "parameters", "parameter_overrides", legacy_string) + self.samconfig.flush() + + # Should parse legacy format + params = self.samconfig.get_template_parameters(cmd_names) + expected = {"Param1": "Value1", "Param2": "Value2"} + self.assertEqual(params, expected) + + def test_get_template_tags_fallback_to_legacy(self): + """Test fallback to legacy tags format""" + cmd_names = ["deploy"] + legacy_string = "Environment=prod Project=MyApp" + + # Store in legacy format + self.samconfig.put(cmd_names, "parameters", "tags", legacy_string) + self.samconfig.flush() + + # Should parse legacy format + tags = self.samconfig.get_template_tags(cmd_names) + expected = {"Environment": "prod", "Project": "MyApp"} + self.assertEqual(tags, expected) + + def test_get_template_parameters_new_format_preferred(self): + """Test that new format takes precedence over legacy""" + cmd_names = ["deploy"] + + # Store in both formats + self.samconfig.put(cmd_names, "parameters", "parameter_overrides", "LegacyParam=LegacyValue") + self.samconfig.put_template_parameter(cmd_names, "NewParam", "NewValue") + self.samconfig.flush() + + # New format should be preferred + params = self.samconfig.get_template_parameters(cmd_names) + expected = {"NewParam": "NewValue"} + self.assertEqual(params, expected) + + def test_get_template_parameters_empty_when_none_found(self): + """Test that empty dict is returned when no parameters found""" + cmd_names = ["deploy"] + + params = self.samconfig.get_template_parameters(cmd_names) + self.assertEqual(params, {}) + + def test_get_template_tags_empty_when_none_found(self): + """Test that empty dict is returned when no tags found""" + cmd_names = ["deploy"] + + tags = self.samconfig.get_template_tags(cmd_names) + self.assertEqual(tags, {}) + + def test_parse_parameter_overrides_simple(self): + """Test parsing simple parameter overrides string""" + param_string = "Key1=Value1 Key2=Value2" + + result = SamConfig._parse_parameter_overrides(param_string) + expected = {"Key1": "Value1", "Key2": "Value2"} + self.assertEqual(result, expected) + + def test_parse_parameter_overrides_quoted(self): + """Test parsing quoted parameter overrides string""" + param_string = 'Key1="Value with spaces" Key2=SimpleValue' + + result = SamConfig._parse_parameter_overrides(param_string) + expected = {"Key1": "Value with spaces", "Key2": "SimpleValue"} + self.assertEqual(result, expected) + + def test_parse_parameter_overrides_empty(self): + """Test parsing empty parameter overrides string""" + self.assertEqual(SamConfig._parse_parameter_overrides(""), {}) + self.assertEqual(SamConfig._parse_parameter_overrides(None), {}) + self.assertEqual(SamConfig._parse_parameter_overrides(" "), {}) + + def test_parse_parameter_overrides_malformed(self): + """Test parsing malformed parameter overrides string""" + param_string = "ValidKey=ValidValue InvalidFormat" + + result = SamConfig._parse_parameter_overrides(param_string) + # Should only parse valid key=value pairs + expected = {"ValidKey": "ValidValue"} + self.assertEqual(result, expected) + + def test_template_parameters_with_complex_values(self): + """Test template parameters with complex values""" + cmd_names = ["deploy"] + + # Test different value types + self.samconfig.put_template_parameter(cmd_names, "StringParam", "string_value") + self.samconfig.put_template_parameter(cmd_names, "NumberParam", 42) + self.samconfig.put_template_parameter(cmd_names, "BooleanParam", True) + self.samconfig.put_template_parameter(cmd_names, "ListParam", [1, 2, 3]) + self.samconfig.put_template_parameter(cmd_names, "DictParam", {"key": "value"}) + self.samconfig.flush() + + params = self.samconfig.get_template_parameters(cmd_names) + + self.assertEqual(params["StringParam"], "string_value") + self.assertEqual(params["NumberParam"], 42) + self.assertEqual(params["BooleanParam"], True) + self.assertEqual(params["ListParam"], [1, 2, 3]) + self.assertEqual(params["DictParam"], {"key": "value"}) + + def test_template_parameters_multiline_values(self): + """Test template parameters with multiline values""" + cmd_names = ["deploy"] + + multiline_value = """line1 +line2 +line3""" + + self.samconfig.put_template_parameter(cmd_names, "MultilineParam", multiline_value) + self.samconfig.flush() + + params = self.samconfig.get_template_parameters(cmd_names) + self.assertEqual(params["MultilineParam"], multiline_value) + + def test_backward_compatibility_mixed_usage(self): + """Test backward compatibility with mixed old and new formats""" + cmd_names = ["deploy"] + + # Store some parameters in legacy format + self.samconfig.put(cmd_names, "parameters", "parameter_overrides", "LegacyParam=LegacyValue") + self.samconfig.put(cmd_names, "parameters", "tags", "LegacyTag=LegacyTagValue") + + # Store some parameters in new format + self.samconfig.put_template_parameter(cmd_names, "NewParam", "NewValue") + self.samconfig.put_template_tag(cmd_names, "NewTag", "NewTagValue") + + # Store other legacy parameters + self.samconfig.put(cmd_names, "parameters", "stack_name", "my-stack") + self.samconfig.put(cmd_names, "parameters", "s3_bucket", "my-bucket") + + self.samconfig.flush() + + # New format should be preferred for template_parameters and template_tags + params = self.samconfig.get_template_parameters(cmd_names) + tags = self.samconfig.get_template_tags(cmd_names) + + self.assertEqual(params, {"NewParam": "NewValue"}) + self.assertEqual(tags, {"NewTag": "NewTagValue"}) + + # Legacy parameters should still be accessible via get_all + legacy_params = self.samconfig.get_all(cmd_names, "parameters") + self.assertEqual(legacy_params["stack_name"], "my-stack") + self.assertEqual(legacy_params["s3_bucket"], "my-bucket") + + def test_global_parameters_inheritance(self): + """Test that global parameters are inherited in new format""" + from samcli.lib.config.samconfig import DEFAULT_GLOBAL_CMDNAME + + cmd_names = ["deploy"] + + # Store global template parameter + self.samconfig.put_template_parameter([DEFAULT_GLOBAL_CMDNAME], "GlobalParam", "GlobalValue") + # Store command-specific template parameter + self.samconfig.put_template_parameter(cmd_names, "CommandParam", "CommandValue") + self.samconfig.flush() + + # Should get both global and command-specific parameters + params = self.samconfig.get_template_parameters(cmd_names) + + # Note: The current get_template_parameters doesn't implement global inheritance + # This test documents the expected behavior for future enhancement + self.assertEqual(params["CommandParam"], "CommandValue") + # Global inheritance would need to be implemented in get_template_parameters + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/unit/lib/samconfig/test_parameter_loaders.py b/tests/unit/lib/samconfig/test_parameter_loaders.py new file mode 100644 index 00000000000..4e5bf97145c --- /dev/null +++ b/tests/unit/lib/samconfig/test_parameter_loaders.py @@ -0,0 +1,295 @@ +""" +Unit tests for parameter file loading functionality +""" + +import json +import os +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +from samcli.lib.config.exceptions import FileParseException +from samcli.lib.config.parameter_loaders import ParameterFileLoader + + +class TestParameterFileLoader(unittest.TestCase): + """Test cases for ParameterFileLoader class""" + + def setUp(self): + """Set up test fixtures""" + self.temp_dir = tempfile.mkdtemp() + self.temp_path = Path(self.temp_dir) + + def tearDown(self): + """Clean up test fixtures""" + import shutil + + shutil.rmtree(self.temp_dir, ignore_errors=True) + + def test_is_file_url_valid(self): + """Test identification of valid file URLs""" + self.assertTrue(ParameterFileLoader.is_file_url("file://path/to/file.json")) + self.assertTrue(ParameterFileLoader.is_file_url("file:///absolute/path/file.yaml")) + self.assertTrue(ParameterFileLoader.is_file_url("file://./relative/path/file.env")) + + def test_is_file_url_invalid(self): + """Test identification of invalid file URLs""" + self.assertFalse(ParameterFileLoader.is_file_url("http://example.com/file.json")) + self.assertFalse(ParameterFileLoader.is_file_url("https://example.com/file.json")) + self.assertFalse(ParameterFileLoader.is_file_url("path/to/file.json")) + self.assertFalse(ParameterFileLoader.is_file_url("")) + self.assertFalse(ParameterFileLoader.is_file_url(None)) + + def test_parse_file_url_unix_path(self): + """Test parsing Unix-style file URLs""" + url = "file:///path/to/file.json" + result = ParameterFileLoader.parse_file_url(url) + self.assertEqual(result, "/path/to/file.json") + + def test_parse_file_url_relative_path(self): + """Test parsing relative file URLs""" + url = "file://./relative/path/file.json" + result = ParameterFileLoader.parse_file_url(url) + self.assertEqual(result, "./relative/path/file.json") + + def test_parse_file_url_with_env_vars(self): + """Test parsing file URLs with environment variables""" + with patch.dict(os.environ, {"TEST_VAR": "test_value"}, clear=False): + url = "file://$TEST_VAR/file.json" + result = ParameterFileLoader.parse_file_url(url) + self.assertEqual(result, "test_value/file.json") + + def test_parse_file_url_invalid(self): + """Test parsing invalid file URLs""" + with self.assertRaises(ValueError): + ParameterFileLoader.parse_file_url("http://example.com/file.json") + + def test_load_json_file_valid(self): + """Test loading valid JSON parameter file""" + json_data = {"param1": "value1", "param2": "value2", "param3": 123} + json_file = self.temp_path / "params.json" + json_file.write_text(json.dumps(json_data)) + + result = ParameterFileLoader.load_from_file(str(json_file)) + self.assertEqual(result, json_data) + + def test_load_json_file_invalid(self): + """Test loading invalid JSON parameter file""" + json_file = self.temp_path / "invalid.json" + json_file.write_text('{"invalid": json}') + + with self.assertRaises(FileParseException): + ParameterFileLoader.load_from_file(str(json_file)) + + def test_load_json_file_not_object(self): + """Test loading JSON file that's not an object""" + json_file = self.temp_path / "array.json" + json_file.write_text('["not", "an", "object"]') + + with self.assertRaises(FileParseException): + ParameterFileLoader.load_from_file(str(json_file)) + + def test_load_yaml_file_valid(self): + """Test loading valid YAML parameter file""" + yaml_content = """ +param1: value1 +param2: value2 +param3: 123 +nested: + key: nested_value +""" + yaml_file = self.temp_path / "params.yaml" + yaml_file.write_text(yaml_content) + + result = ParameterFileLoader.load_from_file(str(yaml_file)) + expected = {"param1": "value1", "param2": "value2", "param3": 123, "nested": {"key": "nested_value"}} + self.assertEqual(result, expected) + + def test_load_yaml_file_empty(self): + """Test loading empty YAML file""" + yaml_file = self.temp_path / "empty.yaml" + yaml_file.write_text("") + + result = ParameterFileLoader.load_from_file(str(yaml_file)) + self.assertEqual(result, {}) + + def test_load_yaml_file_invalid(self): + """Test loading invalid YAML parameter file""" + yaml_file = self.temp_path / "invalid.yaml" + yaml_file.write_text("param1: value1\ninvalid yaml: [ unclosed") + + with self.assertRaises(FileParseException): + ParameterFileLoader.load_from_file(str(yaml_file)) + + def test_load_env_file_simple(self): + """Test loading simple ENV parameter file""" + env_content = """ +# This is a comment +PARAM1=value1 +PARAM2=value2 +PARAM3=123 +""" + env_file = self.temp_path / "params.env" + env_file.write_text(env_content) + + result = ParameterFileLoader.load_from_file(str(env_file)) + expected = {"PARAM1": "value1", "PARAM2": "value2", "PARAM3": "123"} + self.assertEqual(result, expected) + + def test_load_env_file_quoted_values(self): + """Test loading ENV file with quoted values""" + env_content = """ +SIMPLE=value +QUOTED="quoted value with spaces" +MULTILINE="line1 +line2 +line3" +""" + env_file = self.temp_path / "params.env" + env_file.write_text(env_content) + + result = ParameterFileLoader.load_from_file(str(env_file)) + expected = {"SIMPLE": "value", "QUOTED": "quoted value with spaces", "MULTILINE": "line1\nline2\nline3"} + self.assertEqual(result, expected) + + def test_load_env_file_malformed_multiline(self): + """Test loading ENV file with malformed multiline value""" + env_content = """ +PARAM1=value1 +UNTERMINATED="unclosed quote +PARAM2=value2 +""" + env_file = self.temp_path / "params.env" + env_file.write_text(env_content) + + with self.assertRaises(FileParseException): + ParameterFileLoader.load_from_file(str(env_file)) + + def test_load_from_file_unsupported_extension(self): + """Test loading file with unsupported extension""" + txt_file = self.temp_path / "params.txt" + txt_file.write_text("some content") + + with self.assertRaises(FileParseException): + ParameterFileLoader.load_from_file(str(txt_file)) + + def test_load_from_file_not_found(self): + """Test loading non-existent file""" + non_existent = self.temp_path / "does_not_exist.json" + + with self.assertRaises(FileNotFoundError): + ParameterFileLoader.load_from_file(str(non_existent)) + + def test_load_from_file_directory(self): + """Test loading directory instead of file""" + directory = self.temp_path / "not_a_file" + directory.mkdir() + + with self.assertRaises(FileParseException): + ParameterFileLoader.load_from_file(str(directory)) + + def test_resolve_parameter_files_direct_only(self): + """Test resolving parameter overrides with direct parameters only""" + param_string = "Key1=Value1 Key2=Value2" + + direct, file_params = ParameterFileLoader.resolve_parameter_files(param_string) + + expected_direct = {"Key1": "Value1", "Key2": "Value2"} + self.assertEqual(direct, expected_direct) + self.assertEqual(file_params, {}) + + def test_resolve_parameter_files_file_only(self): + """Test resolving parameter overrides with file parameters only""" + json_data = {"FileParam1": "FileValue1", "FileParam2": "FileValue2"} + json_file = self.temp_path / "params.json" + json_file.write_text(json.dumps(json_data)) + + param_string = f"file://{json_file}" + + direct, file_params = ParameterFileLoader.resolve_parameter_files(param_string) + + self.assertEqual(direct, {}) + self.assertEqual(file_params, json_data) + + def test_resolve_parameter_files_mixed(self): + """Test resolving parameter overrides with mixed parameters""" + json_data = {"FileParam": "FileValue"} + json_file = self.temp_path / "params.json" + json_file.write_text(json.dumps(json_data)) + + param_string = f"DirectParam=DirectValue file://{json_file} AnotherParam=AnotherValue" + + direct, file_params = ParameterFileLoader.resolve_parameter_files(param_string) + + expected_direct = {"DirectParam": "DirectValue", "AnotherParam": "AnotherValue"} + self.assertEqual(direct, expected_direct) + self.assertEqual(file_params, json_data) + + def test_resolve_parameter_files_empty(self): + """Test resolving empty parameter overrides""" + direct, file_params = ParameterFileLoader.resolve_parameter_files("") + self.assertEqual(direct, {}) + self.assertEqual(file_params, {}) + + direct, file_params = ParameterFileLoader.resolve_parameter_files(None) + self.assertEqual(direct, {}) + self.assertEqual(file_params, {}) + + def test_resolve_parameter_files_invalid_file(self): + """Test resolving parameter overrides with invalid file""" + param_string = "file:///nonexistent/file.json" + + with self.assertRaises(FileNotFoundError): + ParameterFileLoader.resolve_parameter_files(param_string) + + def test_expand_environment_variables(self): + """Test expanding environment variables in parameter values""" + with patch.dict(os.environ, {"TEST_VAR": "expanded"}, clear=False): + params = { + "NoVar": "simple_value", + "WithVar": "$TEST_VAR", + "WithBraces": "${TEST_VAR}_suffix", + "NonExistent": "$NON_EXISTENT_VAR", + "NumericValue": 123, + } + + result = ParameterFileLoader.expand_environment_variables(params) + + expected = { + "NoVar": "simple_value", + "WithVar": "expanded", + "WithBraces": "expanded_suffix", + "NonExistent": "$NON_EXISTENT_VAR", # Unexpanded if var doesn't exist + "NumericValue": 123, # Non-string values unchanged + } + self.assertEqual(result, expected) + + @patch("samcli.lib.config.parameter_loaders.LOG") + def test_resolve_parameter_files_logs_info(self, mock_log): + """Test that file parameter loading logs info messages""" + json_data = {"param": "value"} + json_file = self.temp_path / "params.json" + json_file.write_text(json.dumps(json_data)) + + param_string = f"file://{json_file}" + ParameterFileLoader.resolve_parameter_files(param_string) + + mock_log.info.assert_called_once() + self.assertIn("Loaded 1 parameters from file", mock_log.info.call_args[0][0]) + + @patch("samcli.lib.config.parameter_loaders.LOG") + def test_resolve_parameter_files_logs_warning_invalid_format(self, mock_log): + """Test that invalid parameter formats log warnings""" + param_string = "ValidParam=Value InvalidFormat" + + direct, file_params = ParameterFileLoader.resolve_parameter_files(param_string) + + expected_direct = {"ValidParam": "Value"} + self.assertEqual(direct, expected_direct) + self.assertEqual(file_params, {}) + mock_log.warning.assert_called_once_with("Skipping invalid parameter format: InvalidFormat") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/unit/lib/samconfig/test_parameter_merger.py b/tests/unit/lib/samconfig/test_parameter_merger.py new file mode 100644 index 00000000000..1b0e38790c5 --- /dev/null +++ b/tests/unit/lib/samconfig/test_parameter_merger.py @@ -0,0 +1,201 @@ +""" +Unit tests for parameter merger functionality +""" + +import unittest +from unittest.mock import patch + +from samcli.lib.config.parameter_merger import ParameterMerger + + +class TestParameterMerger(unittest.TestCase): + """Test cases for ParameterMerger class""" + + def test_merge_parameters_config_only(self): + """Test merging with only config parameters""" + config_params = {"Param1": "ConfigValue1", "Param2": "ConfigValue2"} + + result = ParameterMerger.merge_parameters(config_params=config_params) + + expected = {"Param1": "ConfigValue1", "Param2": "ConfigValue2"} + self.assertEqual(result, expected) + + def test_merge_parameters_cli_only(self): + """Test merging with only CLI parameters""" + cli_params = {"Param1": "CLIValue1", "Param2": "CLIValue2"} + + result = ParameterMerger.merge_parameters(cli_params=cli_params) + + expected = {"Param1": "CLIValue1", "Param2": "CLIValue2"} + self.assertEqual(result, expected) + + def test_merge_parameters_file_only(self): + """Test merging with only file parameters""" + file_params = {"Param1": "FileValue1", "Param2": "FileValue2"} + + result = ParameterMerger.merge_parameters(file_params=file_params) + + expected = {"Param1": "FileValue1", "Param2": "FileValue2"} + self.assertEqual(result, expected) + + def test_merge_parameters_precedence(self): + """Test parameter precedence: CLI > file > config""" + config_params = {"Param1": "ConfigValue", "Param2": "ConfigValue", "Param3": "ConfigValue"} + file_params = {"Param2": "FileValue", "Param3": "FileValue"} + cli_params = {"Param3": "CLIValue"} + + result = ParameterMerger.merge_parameters( + config_params=config_params, cli_params=cli_params, file_params=file_params + ) + + expected = { + "Param1": "ConfigValue", # Only in config + "Param2": "FileValue", # File overrides config + "Param3": "CLIValue", # CLI overrides both + } + self.assertEqual(result, expected) + + def test_merge_parameters_empty_inputs(self): + """Test merging with empty/None inputs""" + result = ParameterMerger.merge_parameters() + self.assertEqual(result, {}) + + result = ParameterMerger.merge_parameters(config_params={}, cli_params=None, file_params={}) + self.assertEqual(result, {}) + + def test_merge_tags_config_only(self): + """Test merging with only config tags""" + config_tags = {"Environment": "prod", "Project": "MyApp"} + + result = ParameterMerger.merge_tags(config_tags=config_tags) + + expected = {"Environment": "prod", "Project": "MyApp"} + self.assertEqual(result, expected) + + def test_merge_tags_precedence(self): + """Test tag precedence: CLI > file > config""" + config_tags = {"Environment": "config", "Project": "config", "CostCenter": "config"} + file_tags = {"Project": "file", "CostCenter": "file"} + cli_tags = {"CostCenter": "cli"} + + result = ParameterMerger.merge_tags(config_tags=config_tags, cli_tags=cli_tags, file_tags=file_tags) + + expected = { + "Environment": "config", # Only in config + "Project": "file", # File overrides config + "CostCenter": "cli", # CLI overrides both + } + self.assertEqual(result, expected) + + def test_format_for_cloudformation(self): + """Test formatting parameters for CloudFormation""" + parameters = { + "StringParam": "value", + "NumberParam": 42, + "BooleanParam": True, + "NullParam": None, + "DictParam": {"key": "value"}, + "ListParam": [1, 2, 3], + } + + result = ParameterMerger.format_for_cloudformation(parameters) + + expected = { + "StringParam": "value", + "NumberParam": "42", + "BooleanParam": "True", + "NullParam": "", + "DictParam": '{"key": "value"}', + "ListParam": "[1, 2, 3]", + } + self.assertEqual(result, expected) + + def test_format_for_cloudformation_empty(self): + """Test formatting empty parameters""" + result = ParameterMerger.format_for_cloudformation({}) + self.assertEqual(result, {}) + + result = ParameterMerger.format_for_cloudformation(None) + self.assertEqual(result, {}) + + def test_parse_legacy_parameter_string_simple(self): + """Test parsing simple parameter strings""" + param_string = "Key1=Value1 Key2=Value2" + + result = ParameterMerger.parse_legacy_parameter_string(param_string) + + expected = {"Key1": "Value1", "Key2": "Value2"} + self.assertEqual(result, expected) + + def test_parse_legacy_parameter_string_quoted(self): + """Test parsing parameter strings with quoted values""" + param_string = 'Key1="Value with spaces" Key2=SimpleValue' + + result = ParameterMerger.parse_legacy_parameter_string(param_string) + + expected = {"Key1": "Value with spaces", "Key2": "SimpleValue"} + self.assertEqual(result, expected) + + def test_parse_legacy_parameter_string_complex(self): + """Test parsing complex parameter strings""" + param_string = 'DBHost=localhost DBPort=5432 DBName="my database" ConnectionString="host=localhost;port=5432"' + + result = ParameterMerger.parse_legacy_parameter_string(param_string) + + expected = { + "DBHost": "localhost", + "DBPort": "5432", + "DBName": "my database", + "ConnectionString": "host=localhost;port=5432", + } + self.assertEqual(result, expected) + + def test_parse_legacy_parameter_string_empty(self): + """Test parsing empty/invalid parameter strings""" + self.assertEqual(ParameterMerger.parse_legacy_parameter_string(""), {}) + self.assertEqual(ParameterMerger.parse_legacy_parameter_string(None), {}) + self.assertEqual(ParameterMerger.parse_legacy_parameter_string(" "), {}) + + @patch("samcli.lib.config.parameter_merger.LOG") + def test_parse_legacy_parameter_string_invalid_format(self, mock_log): + """Test parsing parameter strings with invalid format""" + param_string = "ValidKey=ValidValue InvalidEntry AnotherKey=AnotherValue" + + result = ParameterMerger.parse_legacy_parameter_string(param_string) + + expected = {"ValidKey": "ValidValue", "AnotherKey": "AnotherValue"} + self.assertEqual(result, expected) + mock_log.warning.assert_called_once() + + def test_validate_parameters_with_template(self): + """Test parameter validation against template parameters""" + parameters = {"ValidParam1": "value1", "ValidParam2": "value2", "InvalidParam": "invalid"} + template_parameters = {"ValidParam1": {"Type": "String"}, "ValidParam2": {"Type": "String"}} + + result = ParameterMerger.validate_parameters(parameters, template_parameters) + + expected = {"ValidParam1": "value1", "ValidParam2": "value2"} + self.assertEqual(result, expected) + + def test_validate_parameters_without_template(self): + """Test parameter validation without template parameters""" + parameters = {"Param1": "value1", "Param2": "value2"} + + result = ParameterMerger.validate_parameters(parameters, None) + + self.assertEqual(result, parameters) + + @patch("samcli.lib.config.parameter_merger.LOG") + def test_validate_parameters_logs_warnings(self, mock_log): + """Test parameter validation logs warnings for invalid parameters""" + parameters = {"InvalidParam": "value"} + template_parameters = {"ValidParam": {"Type": "String"}} + + result = ParameterMerger.validate_parameters(parameters, template_parameters) + + self.assertEqual(result, {}) + mock_log.warning.assert_called_once_with("Parameter 'InvalidParam' not found in template parameters. Skipping.") + + +if __name__ == "__main__": + unittest.main()