strict type checking (#539)

.vscode/settings.json

@@ -117,7 +117,7 @@
         "xtrace"
     ],
     "files.insertFinalNewline": true,
-    "python.analysis.typeCheckingMode": "basic",
+    "python.analysis.typeCheckingMode": "strict",
     "python.formatting.provider": "black",
     "python.linting.flake8Args": [
         "--docstring-convention=all"

Makefile

@@ -76,7 +76,7 @@ lint-isort: ## run isort
 
 lint-pylint: ## run pylint
 	@echo "Running pylint..."
-	@find runway -name '*.py' -not -path 'runway/embedded*' -not -path 'runway/templates/stacker/*' -not -path 'runway/templates/cdk-py/*' -not -path 'runway/blueprints/*' | xargs pipenv run pylint --rcfile=pyproject.toml
+	@find runway -name '*.py' -not -path 'runway/templates/cdk-py/*' | xargs pipenv run pylint --rcfile=pyproject.toml
 	@echo ""
 
 lint-pyright:

Pipfile

@@ -21,16 +21,16 @@ flake8-docstrings = "~=1.5.0"
 pep8-naming = "~=0.11"
 pydocstyle = "~=5.0.2"
 # Typing
-boto3-stubs = {extras = ["acm", "awslambda", "cloudformation", "cognito-idp", "dynamodb", "ec2", "ecr", "ecs", "iam", "kms", "route53", "s3", "ssm", "sts"],version = "*"}
+boto3-stubs = {extras = ["acm", "awslambda", "cloudformation", "cloudfront", "cognito-idp", "dynamodb", "ec2", "ecr", "ecs", "iam", "kms", "route53", "s3", "ssm", "sts"],version = "*"}
 mypy-boto3 = "*"
 # Test
 coverage = {extras = ["toml"],version = "~=5.2.1"}
 mock = "~=4.0.2"
 moto = "~=1.3"
-pytest = "~=6.0.1"
-pytest-cov = "~=2.10.1"
+pytest = "~=6.2"
+pytest-cov = "~=2.10"
 pytest-mock = "~=3.5"
-pytest-subprocess = "~=1.0.0"
+pytest-subprocess = "~=1.0"
 pytest-sugar = "*"
 testfixtures = "~=4.10.0"
 # Build

package.json

@@ -59,6 +59,6 @@
     "access": "public"
   },
   "devDependencies": {
-    "pyright": "^1.1.99"
+    "pyright": "1.1.112"
   }
 }

pyrightconfig.json

@@ -10,28 +10,26 @@
         "**/docs",
         "**/node_modules",
         "**/typings",
-        "**/infrastructure",
         "**/integration_test_infrastructure",
         "**/integration_tests",
-        "**/npm",
         "**/runway/aws_sso_botocore",
-        "**/runway/cfngin/commands",
         "**/runway/hooks/staticsite/auth_at_edge/templates",
-        "**/runway/templates",
+        "**/runway/templates/cdk-py",
         "**/scripts"
     ],
     "pythonVersion": "3.8",
     "reportDuplicateImport": "none",
     "reportImportCycles": "none",
+    "reportIncompatibleMethodOverride": "warning",
     "reportMissingTypeStubs": "none",
     "reportPrivateUsage": "none",
     "reportUnknownMemberType": "none",
-    "reportUntypedClassDecorator": "none",
-    "reportUntypedFunctionDecorator": "none",
+    "reportUnnecessaryIsInstance": "warning",
     "reportUnusedImport": "none",
     "reportUnusedVariable": "none",
     "reportWildcardImportFromLibrary": "none",
-    "typeCheckingMode": "basic",
+    "strictParameterNoneValue": false,
+    "typeCheckingMode": "strict",
     "useLibraryCodeForTypes": true,
     "venv": ".venv"
 }

runway/_cli/commands/_run_aws.py

@@ -1,7 +1,7 @@
 """``runway run-aws`` command."""
 # docs: file://./../../../docs/source/commands.rst
 import logging
-from typing import Any, Tuple
+from typing import Any, Tuple, cast
 
 import click
 from awscli.clidriver import create_clidriver
@@ -36,4 +36,4 @@ def run_aws(ctx: click.Context, args: Tuple[str, ...], **_: Any) -> None:
             if name.startswith("awscli.") and isinstance(logger, logging.Logger):
                 logger.setLevel(logging.ERROR)
     with SafeHaven(environ={"LC_CTYPE": "en_US.UTF"}):
-        ctx.exit(create_clidriver().main(list(args)))
+        ctx.exit(cast(int, create_clidriver().main(list(args))))

runway/_cli/main.py

@@ -14,7 +14,10 @@
 
 LOGGER = logging.getLogger("runway.cli")
 
-CLICK_CONTEXT_SETTINGS = dict(help_option_names=["-h", "--help"], max_content_width=999)
+CLICK_CONTEXT_SETTINGS: Dict[str, Any] = {
+    "help_option_names": ["-h", "--help"],
+    "max_content_width": 999,
+}
 
 
 class _CliGroup(click.Group):

runway/_cli/utils.py

@@ -4,7 +4,6 @@
 import logging
 import os
 import sys
-from collections.abc import MutableMapping
 from pathlib import Path
 from typing import Any, Iterator, List, Optional, Tuple
 
@@ -24,7 +23,7 @@
 LOGGER = logging.getLogger(__name__)
 
 
-class CliContext(MutableMapping):
+class CliContext:
     """CLI context object."""
 
     def __init__(

runway/_logging.py

@@ -1,7 +1,7 @@
 """Runway logging."""
 import logging
 from enum import IntEnum
-from typing import Any, Dict, Text, Tuple, Union
+from typing import Any, MutableMapping, Text, Tuple, Union
 
 
 class LogLevels(IntEnum):
@@ -62,8 +62,8 @@ def notice(self, msg: Union[Exception, str], *args: Any, **kwargs: Any) -> None:
         self.log(LogLevels.NOTICE, msg, *args, **kwargs)
 
     def process(
-        self, msg: Union[Exception, str], kwargs: Dict[str, Any]
-    ) -> Tuple[str, Dict[str, Any]]:
+        self, msg: Union[Exception, str], kwargs: MutableMapping[str, Any]
+    ) -> Tuple[str, MutableMapping[str, Any]]:
         """Process the message to append the prefix.
 
         Args:

runway/blueprints/k8s/k8s_master.py

@@ -1,7 +1,7 @@
 #!/usr/bin/env python
 """Module with k8s cluster resources."""
 import awacs.iam
-from awacs.aws import StringLike  # type: ignore  # pylint: disable=no-name-in-module
+from awacs.aws import StringLike  # pylint: disable=no-name-in-module
 from awacs.aws import Allow, Condition, PolicyDocument, Statement
 from awacs.helpers.trust import make_simple_assume_policy
 from troposphere import Export, Join, Output, Sub, ec2, eks, iam

runway/blueprints/staticsite/auth_at_edge.py

@@ -17,7 +17,7 @@
 from .staticsite import StaticSite
 
 if TYPE_CHECKING:
-    from ...cfngin.blueprints.base import BlueprintVariable
+    from ...cfngin.blueprints.type_defs import BlueprintVariableTypeDef
     from ...context.cfngin import CfnginContext
 
 LOGGER = logging.getLogger("runway")
@@ -26,7 +26,7 @@
 class AuthAtEdge(StaticSite):
     """Auth@Edge Blueprint."""
 
-    AUTH_VARIABLES: Dict[str, BlueprintVariable] = {
+    AUTH_VARIABLES: Dict[str, BlueprintVariableTypeDef] = {
         "OAuthScopes": {"type": list, "default": [], "description": "OAuth2 Scopes"},
         "PriceClass": {
             "type": str,
@@ -58,7 +58,7 @@ class AuthAtEdge(StaticSite):
         },
     }
     IAM_ARN_PREFIX = "arn:aws:iam::aws:policy/service-role/"
-    VARIABLES: Dict[str, BlueprintVariable] = {}
+    VARIABLES: Dict[str, BlueprintVariableTypeDef] = {}
 
     def __init__(
         self,
@@ -378,6 +378,7 @@ def _get_error_responses(self) -> List[cloudfront.CustomErrorResponse]:
             )
         ]
 
+    # pyright: reportIncompatibleMethodOverride=none
     def _get_cloudfront_bucket_policy_statements(
         self, bucket: s3.Bucket, oai: cloudfront.CloudFrontOriginAccessIdentity
     ) -> List[Statement]:

runway/blueprints/staticsite/staticsite.py

@@ -36,7 +36,7 @@
 if TYPE_CHECKING:
     from troposphere import Ref  # pylint: disable=ungrouped-imports
 
-    from ...cfngin.blueprints.base import BlueprintVariable
+    from ...cfngin.blueprints.type_defs import BlueprintVariableTypeDef
 
 LOGGER = logging.getLogger("runway")
 
@@ -53,7 +53,7 @@ class _IndexRewriteFunctionInfoTypeDef(TypedDict, total=False):
 class StaticSite(Blueprint):
     """CFNgin blueprint for creating S3 bucket and CloudFront distribution."""
 
-    VARIABLES: Dict[str, BlueprintVariable] = {
+    VARIABLES: Dict[str, BlueprintVariableTypeDef] = {
         "AcmCertificateArn": {
             "type": str,
             "default": "",

runway/cfngin/actions/base.py

@@ -5,7 +5,7 @@
 import os
 import sys
 import threading
-from typing import TYPE_CHECKING, Any, Callable, Optional
+from typing import TYPE_CHECKING, Any, Callable, Optional, Union
 
 import botocore.exceptions
 
@@ -143,18 +143,19 @@ def provider(self) -> Provider:
         Used for running things like hooks.
 
         """
+        if not self.provider_builder:
+            raise ValueError("ProviderBuilder required to build a provider")
         return self.provider_builder.build()
 
     def build_provider(self, stack: Stack) -> Provider:
-        """Build a :class:`runway.cfngin.providers.base.BaseProvider`.
+        """Build a CFNgin provider.
 
         Args:
             stack: Stack the action will be executed on.
 
-        Returns:
-            Suitable for operating on the given :class:`runway.cfngin.stack.Stack`.
-
         """
+        if not self.provider_builder:
+            raise ValueError("ProviderBuilder required to build a provider")
         return self.provider_builder.build(region=stack.region, profile=stack.profile)
 
     def ensure_cfn_bucket(self) -> None:
@@ -172,13 +173,26 @@ def execute(self, **kwargs: Any) -> None:
             LOGGER.error(str(err))
             sys.exit(1)
 
-    def pre_run(self, **_: Any) -> None:
+    def pre_run(
+        self, *, dump: Union[bool, str] = False, outline: bool = False, **__kwargs: Any
+    ) -> None:
         """Perform steps before running the action."""
 
-    def post_run(self, **_: Any) -> None:
+    def post_run(
+        self, *, dump: Union[bool, str] = False, outline: bool = False, **__kwargs: Any
+    ) -> None:
         """Perform steps after running the action."""
 
-    def run(self, *_args: Any, **_kwargs: Any) -> None:
+    def run(
+        self,
+        *,
+        concurrency: int = 0,
+        dump: Union[bool, str] = False,
+        force: bool = False,
+        outline: bool = False,
+        tail: bool = False,
+        **_kwargs: Any
+    ) -> None:
         """Abstract method for running the action."""
         raise NotImplementedError('Subclass must implement "run" method')
 

runway/cfngin/actions/build.py

@@ -371,7 +371,7 @@ def _launch_stack(  # pylint: disable=R
             # a successful update.
             elif provider.is_stack_failed(provider_stack):
                 reason = status.reason
-                if "rolling" in reason:
+                if reason and "rolling" in reason:
                     reason = reason.replace("rolling", "rolled")
                 status_reason = provider.get_rollback_status_reason(stack.fqn)
                 LOGGER.info("%s:roll back reason: %s", stack.fqn, status_reason)
@@ -534,14 +534,15 @@ def pre_run(  # pylint: disable=arguments-differ
             outline=outline,
         )
 
-    def run(  # pylint: disable=arguments-differ
+    def run(
         self,
         *,
         concurrency: int = 0,
         dump: Union[bool, str] = False,
+        force: bool = False,  # pylint: disable=unused-argument
         outline: bool = False,
         tail: bool = False,
-        **_: Any
+        **_kwargs: Any
     ) -> None:
         """Kicks off the build/update of the stacks in the stack_definitions.
 

runway/cfngin/actions/destroy.py

@@ -2,7 +2,7 @@
 from __future__ import annotations
 
 import logging
-from typing import TYPE_CHECKING, Any, Callable, Optional
+from typing import TYPE_CHECKING, Any, Callable, Optional, Union
 
 from ..exceptions import StackDoesNotExist
 from ..hooks.utils import handle_hooks
@@ -75,19 +75,32 @@ def _destroy_stack(
         provider.destroy_stack(provider_stack)
         return DESTROYING_STATUS
 
-    def pre_run(self, **kwargs):
+    def pre_run(
+        self,
+        *,
+        dump: Union[bool, str] = False,  # pylint: disable=unused-argument
+        outline: bool = False,
+        **__kwargs: Any
+    ) -> None:
         """Any steps that need to be taken prior to running the action."""
         pre_destroy = self.context.config.pre_destroy
-        if not kwargs.get("outline") and pre_destroy:
+        if not outline and pre_destroy:
             handle_hooks(
                 stage="pre_destroy",
                 hooks=pre_destroy,
                 provider=self.provider,
                 context=self.context,
             )
 
-    def run(  # pylint: disable=arguments-differ
-        self, *, concurrency: int = 0, force: bool = False, tail: bool = False, **_: Any
+    def run(
+        self,
+        *,
+        concurrency: int = 0,
+        dump: Union[bool, str] = False,  # pylint: disable=unused-argument
+        force: bool = False,
+        outline: bool = False,  # pylint: disable=unused-argument
+        tail: bool = False,
+        **_kwargs: Any
     ) -> None:
         """Kicks off the destruction of the stacks in the stack_definitions."""
         plan = self._generate_plan(
@@ -108,8 +121,12 @@ def run(  # pylint: disable=arguments-differ
         else:
             plan.outline(message='To execute this plan, run with --force" flag.')
 
-    def post_run(  # pylint: disable=arguments-differ
-        self, *, outline: bool = False, **_: Any
+    def post_run(
+        self,
+        *,
+        dump: Union[bool, str] = False,  # pylint: disable=unused-argument
+        outline: bool = False,
+        **__kwargs: Any
     ) -> None:
         """Any steps that need to be taken after running the action."""
         if not outline and self.context.config.post_destroy: