Expand deployment file capabilities

Author: kddejong

src/cfnlint/config.py

@@ -12,12 +12,14 @@
 import logging
 import os
 import sys
+from copy import deepcopy
 from pathlib import Path
 from typing import Any, Dict, Sequence, TypedDict
 
 from typing_extensions import Unpack
 
 import cfnlint.decode.cfn_yaml
+from cfnlint.context.parameters import ParameterSet
 from cfnlint.helpers import REGIONS, format_json_string
 from cfnlint.jsonschema import StandardValidator
 from cfnlint.version import __version__
@@ -681,15 +683,15 @@ class ManualArgs(TypedDict, total=False):
     non_zero_exit_code: str
     output_file: str
     regions: list
-    parameters: list[dict[str, Any]]
+    parameters: list[ParameterSet]
+    templates: list[str]
 
 
 # pylint: disable=too-many-public-methods
 class ConfigMixIn(TemplateArgs, CliArgs, ConfigFileArgs):
 
     def __init__(self, cli_args: list[str] | None = None, **kwargs: Unpack[ManualArgs]):
         self._manual_args = kwargs or ManualArgs()
-        self._templates_to_process = False
         CliArgs.__init__(self, cli_args)
         # configure debug as soon as we can
         TemplateArgs.__init__(self, {})
@@ -716,12 +718,17 @@ def __repr__(self):
                 "merge_configs": self.merge_configs,
                 "non_zero_exit_code": self.non_zero_exit_code,
                 "override_spec": self.override_spec,
-                "regions": self.regions,
                 "parameters": self.parameters,
+                "regions": self.regions,
                 "templates": self.templates,
             }
         )
 
+    def __eq__(self, value):
+        if not isinstance(value, ConfigMixIn):
+            return False
+        return str(self) == str(value)
+
     def _get_argument_value(self, arg_name, is_template, is_config_file):
         cli_value = getattr(self.cli_args, arg_name)
         template_value = self.template_args.get(arg_name)
@@ -816,7 +823,9 @@ def templates(self):
         file_args = self._get_argument_value("templates", False, True)
         cli_args = self._get_argument_value("templates", False, False)
 
-        if cli_alt_args:
+        if "templates" in self._manual_args:
+            filenames = self._manual_args["templates"]
+        elif cli_alt_args:
             filenames = cli_alt_args
         elif file_args:
             filenames = file_args
@@ -826,10 +835,6 @@ def templates(self):
             # No filenames found, could be piped in or be using the api.
             return None
 
-        # If we're still haven't returned, we've got templates to lint.
-        # Build up list of templates to lint.
-        self.templates_to_process = True
-
         if isinstance(filenames, str):
             filenames = [filenames]
 
@@ -880,12 +885,21 @@ def append_rules(self):
         )
 
     @property
-    def parameters(self):
-        return self._get_argument_value("parameters", True, True)
+    def parameters(self) -> list[ParameterSet]:
+        parameter_sets = self._get_argument_value("parameters", True, True)
+        results: list[ParameterSet] = []
+        for parameter_set in parameter_sets:
+            if isinstance(parameter_set, ParameterSet):
+                results.append(parameter_set)
+            else:
+                results.append(
+                    ParameterSet(
+                        source=None,
+                        parameters=parameter_set,
+                    )
+                )
 
-    @parameters.setter
-    def parameters(self, parameters: list[dict[str, Any]]):
-        self._manual_args["parameters"] = parameters
+        return results
 
     @property
     def override_spec(self):
@@ -952,10 +966,8 @@ def non_zero_exit_code(self):
     def force(self):
         return self._get_argument_value("force", False, False)
 
-    @property
-    def templates_to_process(self):
-        return self._templates_to_process
+    def evolve(self, **kwargs: Unpack[ManualArgs]) -> "ConfigMixIn":
 
-    @templates_to_process.setter
-    def templates_to_process(self, value: bool):
-        self._templates_to_process = value
+        config = deepcopy(self)
+        config._manual_args.update(kwargs)
+        return config

src/cfnlint/context/__init__.py

@@ -1,3 +1,8 @@
-__all__ = ["Context", "create_context_for_template"]
+__all__ = ["Context", "create_context_for_template", "ParameterSet"]
 
-from cfnlint.context.context import Context, Path, create_context_for_template
+from cfnlint.context.context import (
+    Context,
+    ParameterSet,
+    Path,
+    create_context_for_template,
+)

src/cfnlint/context/context.py

@@ -9,10 +9,11 @@
 from collections import deque
 from dataclasses import InitVar, dataclass, field, fields
 from functools import lru_cache
-from typing import Any, Deque, Iterator, Sequence, Set, Tuple
+from typing import TYPE_CHECKING, Any, Deque, Iterator, Set, Tuple
 
 from cfnlint.context._mappings import Mappings
 from cfnlint.context.conditions._conditions import Conditions
+from cfnlint.context.parameters import ParameterSet
 from cfnlint.helpers import (
     BOOLEAN_STRINGS_TRUE,
     FUNCTIONS,
@@ -22,6 +23,9 @@
 )
 from cfnlint.schema import PROVIDER_SCHEMA_MANAGER, AttributeDict
 
+if TYPE_CHECKING:
+    from cfnlint.template import Template
+
 _PSEUDOPARAMS_NON_REGION = ["AWS::AccountId", "AWS::NoValue", "AWS::StackName"]
 
 
@@ -132,12 +136,12 @@ class Context:
     """
 
     # what regions we are processing
-    regions: Sequence[str] = field(
+    regions: list[str] = field(
         init=True, default_factory=lambda: list([REGION_PRIMARY])
     )
 
     # supported functions at this point in the template
-    functions: Sequence[str] = field(init=True, default_factory=list)
+    functions: list[str] = field(init=True, default_factory=list)
 
     path: Path = field(init=True, default_factory=Path)
 
@@ -153,7 +157,7 @@ class Context:
         init=True, default_factory=lambda: set(PSEUDOPARAMS)
     )
 
-    # Combiniation of storing any resolved ref
+    # Combination of storing any resolved ref
     # and adds in any Refs available from things like Fn::Sub
     ref_values: dict[str, Any] = field(init=True, default_factory=dict)
 
@@ -163,6 +167,9 @@ class Context:
     is_resolved_value: bool = field(init=True, default=False)
     resolve_pseudo_parameters: bool = field(init=True, default=True)
 
+    # Deployment parameters
+    parameter_sets: list[ParameterSet] | None = field(init=True, default_factory=list)
+
     def evolve(self, **kwargs) -> "Context":
         """
         Create a new context without merging together attributes
@@ -441,7 +448,9 @@ def _init_transforms(transforms: Any) -> Transforms:
     return Transforms([])
 
 
-def create_context_for_template(cfn):
+def create_context_for_template(
+    cfn: Template,
+) -> "Context":
     parameters = {}
     try:
         parameters = _init_parameters(cfn.template.get("Parameters", {}))
@@ -476,5 +485,6 @@ def create_context_for_template(cfn):
         regions=cfn.regions,
         path=Path(),
         functions=["Fn::Transform"],
-        ref_values=cfn.parameters or {},
+        ref_values={},
+        parameter_sets=[],
     )

src/cfnlint/context/parameters.py

@@ -0,0 +1,16 @@
+"""
+Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+SPDX-License-Identifier: MIT-0
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any
+
+
+@dataclass
+class ParameterSet:
+
+    source: str | None = field(default=None)
+    parameters: dict[str, Any] = field(default_factory=dict)

src/cfnlint/core.py

@@ -12,12 +12,9 @@
 from cfnlint.exceptions import UnexpectedRuleException
 from cfnlint.match import Match
 from cfnlint.rules import RulesCollection
-
-from cfnlint.runner.exceptions import UnexpectedRuleException
 from cfnlint.runner.template.runner import _run_template
 
 
-
 def get_rules(
     append_rules: list[str],
     ignore_rules: list[str],

src/cfnlint/decode/__init__.py

@@ -3,6 +3,15 @@
 SPDX-License-Identifier: MIT-0
 """
 
+__all__ = [
+    "create_match_file_error",
+    "create_match_json_parser_error",
+    "create_match_yaml_parser_error",
+    "decode",
+    "decode_str",
+    "convert_dict",
+]
+
 from cfnlint.decode.decode import (
     create_match_file_error,
     create_match_json_parser_error,

src/cfnlint/decode/decode.py

@@ -37,7 +37,7 @@ def _decode(
 ) -> Decode:
     """Decode payload using yaml_f and json_f, using filename for log output."""
     template = None
-    matches = []
+    matches: Matches = []
     try:
         template = yaml_f(payload)
     except IOError as e:
@@ -145,7 +145,10 @@ def _decode(
                 message="Template needs to be an object.",
             )
         ]
-    return (template, matches)
+
+    if matches:
+        return None, matches
+    return template, matches
 
 
 def create_match_yaml_parser_error(parser_error, filename):

src/cfnlint/exceptions.py

@@ -1,3 +1,11 @@
+"""
+Copyright 2019 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+SPDX-License-Identifier: MIT-0
+"""
+
+from __future__ import annotations
+
+
 class CfnLintExitException(Exception):
     """
     An exception that is raised to indicate that the CloudFormation linter should exit.
@@ -10,11 +18,11 @@ class CfnLintExitException(Exception):
         exit_code (int): The exit code to be used when the linter exits.
 
     Methods:
-        __init__(self, exit_code: int) -> None:
+        __init__(self, msg: str | None=None, exit_code: int=1) -> None:
             Initialize a new CfnLintExitException instance with the specified exit code.
     """
 
-    def __init__(self, msg=None, exit_code=1):
+    def __init__(self, msg: str | None = None, exit_code: int = 1):
         """
         Initialize a new CfnLintExitException instance with the specified exit code.
 
@@ -49,9 +57,17 @@ class UnexpectedRuleException(CfnLintExitException):
 
 class DuplicateRuleError(CfnLintExitException):
     """
-    The data associated with a particular path could not be loaded.
-    :ivar data_path: The data path that the user attempted to load.
+    An exception that is raised when an unexpected error occurs while loading rules.
+
+    This exception is raised when the CloudFormation linter encounters a rule with a
+    duplicate ID.
     """
 
     def __init__(self, rule_id: str):
+        """
+        Initialize a new CfnLintExitException instance with the specified exit code.
+
+        Args:
+            rule_id (str): The rule ID that a duplicate was found for.
+        """
         super().__init__(f"Rule already included: {rule_id}")

src/cfnlint/helpers.py

@@ -7,6 +7,7 @@
 
 from __future__ import annotations
 
+import dataclasses
 import datetime
 import fnmatch
 import gzip
@@ -627,6 +628,8 @@ def converter(o):  # pylint: disable=R1710
         """Help convert date/time into strings"""
         if isinstance(o, datetime.datetime):
             return o.__str__()  # pylint: disable=unnecessary-dunder-call
+        elif dataclasses.is_dataclass(o):
+            return dataclasses.asdict(o)
 
     return json.dumps(
         json_string, indent=1, sort_keys=True, separators=(",", ": "), default=converter

src/cfnlint/jsonschema/validators.py

@@ -25,7 +25,7 @@
 from typing import Any, Callable
 
 from cfnlint.conditions import UnknownSatisfisfaction
-from cfnlint.context import Context, create_context_for_template
+from cfnlint.context import Context
 from cfnlint.helpers import is_function
 from cfnlint.jsonschema import _keywords, _keywords_cfn, _resolvers_cfn
 from cfnlint.jsonschema._filter import FunctionFilter
@@ -92,7 +92,7 @@ class Validator:
 
         def __post_init__(self):
             if self.context is None:
-                self.context = create_context_for_template(self.cfn)
+                self.context = self.cfn.context.evolve()
             if self.resolver is None:
                 self.resolver = RefResolver.from_schema(
                     schema=self.schema,

src/cfnlint/match.py

@@ -118,6 +118,8 @@ def create(
         if linenumberend is None:
             linenumberend = linenumber
 
+        filename = getattr(rulematch_obj, "filename", filename)
+
         return cls(
             linenumber=linenumber,
             columnnumber=columnnumber,

src/cfnlint/rules/deployment_files/Parameters.py

@@ -79,18 +79,23 @@ def _build_schema(self, instance: Any) -> dict[str, Any]:
         return schema
 
     def validate(self, validator: Validator, _: Any, instance: Any, schema: Any):
-        if validator.cfn.parameters is None:
+        if validator.context.parameter_sets is None:
             return
 
-        cfn_validator = self.extend_validator(
-            validator=validator,
-            schema=self._build_schema(instance),
-            context=validator.context,
-        ).evolve(
-            context=validator.context.evolve(strict_types=False),
-            function_filter=validator.function_filter.evolve(
-                add_cfn_lint_keyword=False,
-            ),
-        )
-
-        yield from super()._iter_errors(cfn_validator, validator.cfn.parameters)
+        for parameter_set in validator.context.parameter_sets:
+
+            cfn_validator = self.extend_validator(
+                validator=validator,
+                schema=self._build_schema(instance),
+                context=validator.context,
+            ).evolve(
+                context=validator.context.evolve(strict_types=False),
+                function_filter=validator.function_filter.evolve(
+                    add_cfn_lint_keyword=False,
+                ),
+            )
+
+            for err in super()._iter_errors(cfn_validator, parameter_set.parameters):
+                if parameter_set.source:
+                    err.extra_args["filename"] = parameter_set.source
+                yield err