diff --git a/.pylintrc b/.pylintrc index 8c47c29237..55db7d9f08 100644 --- a/.pylintrc +++ b/.pylintrc @@ -63,7 +63,6 @@ confidence= # no Warning level messages displayed, use"--disable=all --enable=classes # --disable=W" disable= - R0201, # Method could be a function W0613, # Unused argument %r W0640, # Cell variable %s defined in loop A variable used in a closure is defined in a loop R0902, # Too many instance attributes (%s/%s) diff --git a/samcli/cli/context.py b/samcli/cli/context.py index 993b715aa4..a69ebb9ff2 100644 --- a/samcli/cli/context.py +++ b/samcli/cli/context.py @@ -4,6 +4,7 @@ import logging import uuid +from typing import Optional, cast import boto3 import botocore @@ -33,6 +34,8 @@ class Context: properties used by every CLI command. """ + _session_id: str + def __init__(self): """ Initialize the context with default values @@ -87,7 +90,7 @@ def profile(self, value): self._refresh_session() @property - def session_id(self): + def session_id(self) -> str: """ Returns the ID of this command session. This is a randomly generated UUIDv4 which will not change until the command terminates. @@ -131,7 +134,7 @@ def template_dict(self): return None @staticmethod - def get_current_context(): + def get_current_context() -> Optional["Context"]: """ Get the current Context object from Click's context stacks. This method is safe to run within the actual command's handler that has a ``@pass_context`` annotation. Outside of the handler, you run @@ -159,7 +162,7 @@ def my_command_handler(ctx): click_core_ctx = click.get_current_context() if click_core_ctx: - return click_core_ctx.find_object(Context) or click_core_ctx.ensure_object(Context) + return cast("Context", click_core_ctx.find_object(Context) or click_core_ctx.ensure_object(Context)) return None diff --git a/samcli/cli/global_config.py b/samcli/cli/global_config.py index dc0f208c9e..9af187b908 100644 --- a/samcli/cli/global_config.py +++ b/samcli/cli/global_config.py @@ -7,6 +7,7 @@ import uuid import os from pathlib import Path +from typing import Optional, Dict, Any import click @@ -39,7 +40,7 @@ def __init__(self, config_dir=None, installation_id=None, telemetry_enabled=None self._telemetry_enabled = telemetry_enabled @property - def config_dir(self): + def config_dir(self) -> Path: if not self._config_dir: # Internal Environment variable to customize SAM CLI App Dir. Currently used only by integ tests. app_dir = os.getenv("__SAM_CLI_APP_DIR") @@ -137,7 +138,7 @@ def telemetry_enabled(self, value): self._set_value("telemetryEnabled", value) self._telemetry_enabled = value - def _get_value(self, key): + def _get_value(self, key: str) -> Optional[Any]: cfg_path = self._get_config_file_path(CONFIG_FILENAME) if not cfg_path.exists(): return None @@ -146,7 +147,7 @@ def _get_value(self, key): json_body = json.loads(body) return json_body.get(key) - def _set_value(self, key, value): + def _set_value(self, key: str, value: Any) -> Any: cfg_path = self._get_config_file_path(CONFIG_FILENAME) if not cfg_path.exists(): return self._set_json_cfg(cfg_path, key, value) @@ -186,7 +187,8 @@ def _get_or_set_uuid(self, key): return cfg_value return self._set_value(key, str(uuid.uuid4())) - def _set_json_cfg(self, filepath, key, value, json_body=None): + @staticmethod + def _set_json_cfg(filepath: Path, key: str, value: Any, json_body: Optional[Dict] = None) -> Any: """ Special logic method to add a value to a JSON configuration file. This method will write a new version of the file in question, so it will diff --git a/samcli/commands/_utils/table_print.py b/samcli/commands/_utils/table_print.py index 4fc0d9d7ea..a3263e0532 100644 --- a/samcli/commands/_utils/table_print.py +++ b/samcli/commands/_utils/table_print.py @@ -4,6 +4,7 @@ from itertools import count, zip_longest import textwrap from functools import wraps +from typing import Sized import click @@ -116,7 +117,7 @@ def pprint_columns(columns, width, margin, format_string, format_args, columns_d click.secho(format_string.format(*format_args, **columns_dict), fg=color) -def newline_per_item(iterable, counter): +def newline_per_item(iterable: Sized, counter: int) -> None: """ Adds a new line based on the index of a given iterable Parameters diff --git a/samcli/commands/deploy/deploy_context.py b/samcli/commands/deploy/deploy_context.py index 079217337a..e1af644ddd 100644 --- a/samcli/commands/deploy/deploy_context.py +++ b/samcli/commands/deploy/deploy_context.py @@ -17,6 +17,7 @@ import logging import os +from typing import Dict, List import boto3 import click @@ -208,7 +209,8 @@ def deploy( raise click.echo(str(ex)) - def merge_parameters(self, template_dict, parameter_overrides): + @staticmethod + def merge_parameters(template_dict: Dict, parameter_overrides: Dict) -> List[Dict]: """ CloudFormation CreateChangeset requires a value for every parameter from the template, either specifying a new value or use previous value. @@ -216,10 +218,11 @@ def merge_parameters(self, template_dict, parameter_overrides): generates a dict of all parameters in a format that ChangeSet API will accept + :param template_dict: :param parameter_overrides: :return: """ - parameter_values = [] + parameter_values: List[Dict] = [] if not isinstance(template_dict.get("Parameters", None), dict): return parameter_values diff --git a/samcli/commands/deploy/guided_config.py b/samcli/commands/deploy/guided_config.py index a6613cebcb..eef259af9c 100644 --- a/samcli/commands/deploy/guided_config.py +++ b/samcli/commands/deploy/guided_config.py @@ -1,6 +1,7 @@ """ Set of Utilities to deal with reading/writing to configuration file during sam deploy """ +from typing import Any import click @@ -105,5 +106,6 @@ def _save_image_repositories(self, cmd_names, config_env, samconfig, image_repos _image_repositories = [f"{key}={value}" for key, value in image_repositories.items()] samconfig.put(cmd_names, self.section, "image_repositories", _image_repositories, env=config_env) - def quote_parameter_values(self, parameter_value): + @staticmethod + def quote_parameter_values(parameter_value: Any) -> str: return '"{}"'.format(parameter_value) diff --git a/samcli/commands/deploy/guided_context.py b/samcli/commands/deploy/guided_context.py index 574c5df5b1..72004aaae9 100644 --- a/samcli/commands/deploy/guided_context.py +++ b/samcli/commands/deploy/guided_context.py @@ -3,6 +3,7 @@ """ import logging +from typing import Dict, Any import click from click.types import FuncParamType @@ -332,7 +333,10 @@ def run(self): signing_profiles=self.signing_profiles, ) - def _get_parameter_value(self, parameter_key, parameter_properties, parameter_override_from_cmdline): + @staticmethod + def _get_parameter_value( + parameter_key: str, parameter_properties: Dict, parameter_override_from_cmdline: Dict + ) -> Any: """ This function provide the value of a parameter. If the command line/config file have "override_parameter" whose key exist in the template file parameters, it will use the corresponding value. diff --git a/samcli/commands/init/init_templates.py b/samcli/commands/init/init_templates.py index 191fb11a2f..90d6db6509 100644 --- a/samcli/commands/init/init_templates.py +++ b/samcli/commands/init/init_templates.py @@ -10,7 +10,8 @@ import shutil import subprocess -from pathlib import Path # must come after Py2.7 deprecation +from pathlib import Path +from typing import Dict import click @@ -83,8 +84,11 @@ def location_from_app_template(self, package_type, runtime, base_image, dependen msg = "Can't find application template " + app_template + " - check valid values in interactive init." raise InvalidInitTemplateError(msg) from ex - def _check_app_template(self, entry, app_template): - return entry["appTemplate"] == app_template + @staticmethod + def _check_app_template(entry: Dict, app_template: str) -> bool: + # we need to cast it to bool because entry["appTemplate"] can be Any, and Any's __eq__ can return Any + # detail: https://github.com/python/mypy/issues/5697 + return bool(entry["appTemplate"] == app_template) def init_options(self, package_type, runtime, base_image, dependency_manager): if not self.clone_attempted: @@ -113,7 +117,8 @@ def _init_options_from_manifest(self, package_type, runtime, base_image, depende return list(templates_by_dep) return list(templates) - def _init_options_from_bundle(self, package_type, runtime, dependency_manager): + @staticmethod + def _init_options_from_bundle(package_type, runtime, dependency_manager): for mapping in list(itertools.chain(*(RUNTIME_DEP_TEMPLATE_MAPPING.values()))): if runtime in mapping["runtimes"] or any([r.startswith(runtime) for r in mapping["runtimes"]]): if not dependency_manager or dependency_manager == mapping["dependency_manager"]: @@ -128,7 +133,8 @@ def _init_options_from_bundle(self, package_type, runtime, dependency_manager): ) raise InvalidInitTemplateError(msg) - def _shared_dir_check(self, shared_dir): + @staticmethod + def _shared_dir_check(shared_dir: Path) -> bool: try: shared_dir.mkdir(mode=0o700, parents=True, exist_ok=True) return True @@ -146,13 +152,13 @@ def _clone_repo(self): return expected_path = os.path.normpath(os.path.join(shared_dir, self._repo_name)) if self._template_directory_exists(expected_path): - self._overwrite_existing_templates(shared_dir, expected_path) + self._overwrite_existing_templates(expected_path) else: # simply create the app templates repo self._clone_new_app_templates(shared_dir, expected_path) self.clone_attempted = True - def _overwrite_existing_templates(self, shared_dir, expected_path): + def _overwrite_existing_templates(self, expected_path: str): self.repo_path = expected_path # workflow to clone a copy to a new directory and overwrite with osutils.mkdir_temp(ignore_errors=True) as tempdir: @@ -174,11 +180,12 @@ def _overwrite_existing_templates(self, shared_dir, expected_path): if "not found" in output.lower(): click.echo("WARN: Could not clone app template repo.") - def _replace_app_templates(self, temp_path, dest_path): + @staticmethod + def _replace_app_templates(temp_path: str, dest_path: str) -> None: try: - LOG.debug("Removing old templates from %s", str(dest_path)) + LOG.debug("Removing old templates from %s", dest_path) shutil.rmtree(dest_path, onerror=rmtree_callback) - LOG.debug("Copying templates from %s to %s", str(temp_path), str(dest_path)) + LOG.debug("Copying templates from %s to %s", temp_path, dest_path) shutil.copytree(temp_path, dest_path, ignore=shutil.ignore_patterns("*.git")) except (OSError, shutil.Error) as ex: # UNSTABLE STATE @@ -208,11 +215,13 @@ def _clone_new_app_templates(self, shared_dir, expected_path): if "not found" in output.lower(): click.echo("WARN: Could not clone app template repo.") - def _template_directory_exists(self, expected_path): + @staticmethod + def _template_directory_exists(expected_path: str) -> bool: path = Path(expected_path) return path.exists() - def _git_executable(self): + @staticmethod + def _git_executable() -> str: execname = "git" if platform.system().lower() == "windows": options = [execname, "{}.cmd".format(execname), "{}.exe".format(execname), "{}.bat".format(execname)] diff --git a/samcli/commands/local/generate_event/event_generation.py b/samcli/commands/local/generate_event/event_generation.py index 222d00a698..2e35187733 100644 --- a/samcli/commands/local/generate_event/event_generation.py +++ b/samcli/commands/local/generate_event/event_generation.py @@ -24,7 +24,7 @@ class ServiceCommand(click.MultiCommand): List all of the subcommands """ - def __init__(self, events_lib, *args, **kwargs): + def __init__(self, events_lib: events.Events, *args, **kwargs): """ Constructor for the ServiceCommand class @@ -96,7 +96,7 @@ class EventTypeSubCommand(click.MultiCommand): TAGS = "tags" - def __init__(self, events_lib, top_level_cmd_name, subcmd_definition, *args, **kwargs): + def __init__(self, events_lib: events.Events, top_level_cmd_name, subcmd_definition, *args, **kwargs): """ constructor for the EventTypeSubCommand class @@ -177,8 +177,11 @@ def list_commands(self, ctx): """ return sorted(self.subcmd_definition.keys()) + @staticmethod @track_command - def cmd_implementation(self, events_lib, top_level_cmd_name, subcmd_name, *args, **kwargs): + def cmd_implementation( + events_lib: events.Events, top_level_cmd_name: str, subcmd_name: str, *args, **kwargs + ) -> str: """ calls for value substitution in the event json and returns the customized json as a string diff --git a/samcli/commands/package/package_context.py b/samcli/commands/package/package_context.py index 36ba919b81..b69b43f21d 100644 --- a/samcli/commands/package/package_context.py +++ b/samcli/commands/package/package_context.py @@ -18,6 +18,7 @@ import json import logging import os +from typing import Optional import boto3 import click @@ -139,7 +140,8 @@ def _export(self, template_path, use_json): return exported_str - def write_output(self, output_file_name, data): + @staticmethod + def write_output(output_file_name: Optional[str], data: str) -> None: if output_file_name is None: click.echo(data) return diff --git a/samcli/lib/build/app_builder.py b/samcli/lib/build/app_builder.py index fae1ea9f1d..abe6d4663d 100644 --- a/samcli/lib/build/app_builder.py +++ b/samcli/lib/build/app_builder.py @@ -7,6 +7,7 @@ import json import logging import pathlib +from typing import Dict import docker from aws_lambda_builders import RPC_PROTOCOL_VERSION as lambda_builders_protocol_version @@ -172,7 +173,8 @@ def _get_build_graph(self): build_graph.clean_redundant_definitions_and_update(not self._is_building_specific_resource) return build_graph - def update_template(self, template_dict, original_template_path, built_artifacts): + @staticmethod + def update_template(template_dict: Dict, original_template_path: str, built_artifacts: Dict[str, str]) -> Dict: """ Given the path to built artifacts, update the template to point appropriate resource CodeUris to the artifacts folder diff --git a/samcli/lib/build/build_strategy.py b/samcli/lib/build/build_strategy.py index edcdcb365f..83110cd7ce 100644 --- a/samcli/lib/build/build_strategy.py +++ b/samcli/lib/build/build_strategy.py @@ -4,6 +4,7 @@ import logging import pathlib import shutil +from abc import abstractmethod, ABC from samcli.commands.build.exceptions import MissingBuildMethodException from samcli.lib.utils import osutils @@ -14,7 +15,7 @@ LOG = logging.getLogger(__name__) -class BuildStrategy: +class BuildStrategy(ABC): """ Base class for BuildStrategy Keeps basic implementation of build, build_functions and build_layers @@ -50,12 +51,12 @@ def _build_functions(self, build_graph): return function_build_results + @abstractmethod def build_single_function_definition(self, build_definition): """ Builds single function definition and returns dictionary which contains function name as key, build location as value """ - return {} def _build_layers(self, build_graph): """ @@ -67,12 +68,12 @@ def _build_layers(self, build_graph): return layer_build_results + @abstractmethod def build_single_layer_definition(self, layer_definition): """ Builds single layer definition and returns dictionary which contains layer name as key, build location as value """ - return {} class DefaultBuildStrategy(BuildStrategy): diff --git a/samcli/lib/config/samconfig.py b/samcli/lib/config/samconfig.py index 7d20e620b9..148d858e17 100644 --- a/samcli/lib/config/samconfig.py +++ b/samcli/lib/config/samconfig.py @@ -6,6 +6,7 @@ import logging from pathlib import Path +from typing import Any, Iterable import tomlkit @@ -189,11 +190,12 @@ def _write(self): def _version(self): return self.document.get(VERSION_KEY, None) - def _version_sanity_check(self, version): + @staticmethod + 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. ") @staticmethod - def _to_key(cmd_names): + def _to_key(cmd_names: Iterable[str]) -> str: # construct a parsed name that is of the format: a_b_c_d return "_".join([cmd.replace("-", "_").replace(" ", "_") for cmd in cmd_names]) diff --git a/samcli/lib/deploy/deployer.py b/samcli/lib/deploy/deployer.py index 821485c6df..4980036510 100644 --- a/samcli/lib/deploy/deployer.py +++ b/samcli/lib/deploy/deployer.py @@ -21,6 +21,7 @@ import logging import time from datetime import datetime +from typing import Dict, List import botocore @@ -395,7 +396,8 @@ def describe_stack_events(self, stack_name, time_stamp_marker, **kwargs): # Sleep in exponential backoff mode time.sleep(math.pow(self.backoff, retry_attempts)) - def _check_stack_complete(self, status): + @staticmethod + def _check_stack_complete(status: str) -> bool: return "COMPLETE" in status and "CLEANUP" not in status def wait_for_execute(self, stack_name, changeset_type): @@ -443,10 +445,11 @@ def create_and_wait_for_changeset( except botocore.exceptions.ClientError as ex: raise DeployFailedError(stack_name=stack_name, msg=str(ex)) from ex + @staticmethod @pprint_column_names( format_string=OUTPUTS_FORMAT_STRING, format_kwargs=OUTPUTS_DEFAULTS_ARGS, table_header=OUTPUTS_TABLE_HEADER_NAME ) - def _display_stack_outputs(self, stack_outputs, **kwargs): + def _display_stack_outputs(stack_outputs: List[Dict], **kwargs) -> None: for counter, output in enumerate(stack_outputs): for k, v in [ ("Key", output.get("OutputKey")), diff --git a/samcli/lib/generated_sample_events/events.py b/samcli/lib/generated_sample_events/events.py index 7b207a519e..2f70b6fb15 100644 --- a/samcli/lib/generated_sample_events/events.py +++ b/samcli/lib/generated_sample_events/events.py @@ -6,6 +6,7 @@ import json import base64 import warnings +from typing import Dict, cast from urllib.parse import quote as url_quote with warnings.catch_warnings(): @@ -97,7 +98,8 @@ def transform_val(self, properties, val): return transformed - def encode(self, encoding_scheme, val): + @staticmethod + def encode(encoding_scheme: str, val: str) -> str: """ encodes a given val with given encoding scheme @@ -122,7 +124,8 @@ def encode(self, encoding_scheme, val): # returns original val if encoding_scheme not recognized return val - def hash(self, hashing_scheme, val): + @staticmethod + def hash(hashing_scheme: str, val: str) -> str: """ hashes a given val using given hashing_scheme @@ -143,7 +146,7 @@ def hash(self, hashing_scheme, val): # raise exception if hashing_scheme is unsupported raise ValueError("Hashing_scheme {} is not supported.".format(hashing_scheme)) - def generate_event(self, service_name, event_type, values_to_sub): + def generate_event(self, service_name: str, event_type: str, values_to_sub: Dict) -> str: """ opens the event json, substitutes the values in, and returns the customized event json @@ -178,4 +181,5 @@ def generate_event(self, service_name, event_type, values_to_sub): data = json.dumps(data, indent=2) # return the substituted file - return renderer.render(data, values_to_sub) + # According to chevron's code, it returns a str (A string containing the rendered template.) + return cast("str", renderer.render(data, values_to_sub)) diff --git a/samcli/lib/intrinsic_resolver/intrinsic_property_resolver.py b/samcli/lib/intrinsic_resolver/intrinsic_property_resolver.py index 1122867133..6b06fef653 100644 --- a/samcli/lib/intrinsic_resolver/intrinsic_property_resolver.py +++ b/samcli/lib/intrinsic_resolver/intrinsic_property_resolver.py @@ -567,7 +567,8 @@ def handle_fn_transform(self, intrinsic_value, ignore_errors): return location_data - def handle_fn_import_value(self, intrinsic_value, ignore_errors): + @staticmethod + def handle_fn_import_value(intrinsic_value, ignore_errors): """ { "Fn::ImportValue" : sharedValueToImport } This intrinsic function requires handling multiple stacks, which is not currently supported by SAM-CLI. diff --git a/samcli/lib/providers/cfn_api_provider.py b/samcli/lib/providers/cfn_api_provider.py index c784567a7f..bcaf8f2782 100644 --- a/samcli/lib/providers/cfn_api_provider.py +++ b/samcli/lib/providers/cfn_api_provider.py @@ -40,7 +40,7 @@ def extract_resources(self, resources, collector, cwd=None): resources: dict The dictionary containing the different resources within the template - collector: samcli.commands.local.lib.route_collector.RouteCollector + collector: samcli.lib.providers.api_collector.ApiCollector Instance of the API collector that where we will save the API information cwd : str diff --git a/samcli/lib/providers/cfn_base_api_provider.py b/samcli/lib/providers/cfn_base_api_provider.py index fe755e99f5..3a14055fae 100644 --- a/samcli/lib/providers/cfn_base_api_provider.py +++ b/samcli/lib/providers/cfn_base_api_provider.py @@ -1,8 +1,10 @@ """Class that parses the CloudFormation Api Template""" import logging +from typing import Dict, Union, List, Optional from samcli.commands.local.lib.swagger.parser import SwaggerParser from samcli.commands.local.lib.swagger.reader import SwaggerReader +from samcli.lib.providers.api_collector import ApiCollector from samcli.lib.providers.provider import Cors from samcli.local.apigw.local_apigw_service import Route @@ -23,7 +25,7 @@ def extract_resources(self, resources, collector, cwd=None): resources: dict The dictionary containing the different resources within the template - collector: samcli.commands.local.lib.route_collector.RouteCollector + collector: samcli.lib.providers.api_collector.ApiCollector Instance of the API collector that where we will save the API information cwd : str @@ -35,7 +37,16 @@ def extract_resources(self, resources, collector, cwd=None): """ raise NotImplementedError("not implemented") - def extract_swagger_route(self, logical_id, body, uri, binary_media, collector, cwd=None, event_type=Route.API): + @staticmethod + def extract_swagger_route( + logical_id: str, + body: Dict, + uri: Union[str, Dict], + binary_media: List, + collector: ApiCollector, + cwd: Optional[str] = None, + event_type=Route.API, + ) -> None: """ Parse the Swagger documents and adds it to the ApiCollector. @@ -53,7 +64,7 @@ def extract_swagger_route(self, logical_id, body, uri, binary_media, collector, binary_media: list The link to the binary media - collector: samcli.commands.local.lib.route_collector.RouteCollector + collector: samcli.lib.providers.api_collector.ApiCollector Instance of the Route collector that where we will save the route information cwd : str diff --git a/samcli/lib/providers/sam_api_provider.py b/samcli/lib/providers/sam_api_provider.py index 5edd490564..dcbb32913f 100644 --- a/samcli/lib/providers/sam_api_provider.py +++ b/samcli/lib/providers/sam_api_provider.py @@ -67,7 +67,7 @@ def _extract_from_serverless_api(self, logical_id, api_resource, collector, cwd= api_resource : dict Resource definition, including its properties - collector: samcli.commands.local.lib.route_collector.RouteCollector + collector: samcli.lib.providers.api_collector.ApiCollector Instance of the API collector that where we will save the API information cwd : str @@ -106,7 +106,7 @@ def _extract_from_serverless_http(self, logical_id, api_resource, collector, cwd api_resource : dict Resource definition, including its properties - collector: samcli.commands.local.lib.route_collector.RouteCollector + collector: samcli.lib.providers.api_collector.ApiCollector Instance of the API collector that where we will save the API information cwd : str @@ -143,7 +143,7 @@ def _extract_routes_from_function(self, logical_id, function_resource, collector function_resource : dict Contents of the function resource including its properties - collector: samcli.commands.local.lib.route_collector.RouteCollector + collector: samcli.lib.providers.api_collector.ApiCollector Instance of the API collector that where we will save the API information """ @@ -164,7 +164,7 @@ def extract_routes_from_events(self, function_logical_id, serverless_function_ev serverless_function_events : dict Event Dictionary of a AWS::Serverless::Function - collector: samcli.commands.local.lib.route_collector.RouteCollector + collector: samcli.lib.providers.api_collector.ApiCollector Instance of the Route collector that where we will save the route information """ count = 0 @@ -236,7 +236,7 @@ def merge_routes(collector): Parameters ---------- - collector: samcli.commands.local.lib.route_collector.RouteCollector + collector: samcli.lib.providers.api_collector.ApiCollector Collector object that holds all the APIs specified in the template Returns diff --git a/samcli/lib/samlib/wrapper.py b/samcli/lib/samlib/wrapper.py index 1f0752908a..622e18af7a 100644 --- a/samcli/lib/samlib/wrapper.py +++ b/samcli/lib/samlib/wrapper.py @@ -12,6 +12,8 @@ import json import functools +from typing import Dict + import boto3 # SAM Translator Library Internal module imports # @@ -144,7 +146,8 @@ def parse(self, sam_template, sam_plugins): if document_errors: raise InvalidDocumentException(document_errors) - def _validate(self, sam_template): + @staticmethod + def _validate(sam_template: Dict) -> None: """Validates the template and parameter values and raises exceptions if there's an issue :param dict sam_template: SAM template diff --git a/samcli/lib/telemetry/metric.py b/samcli/lib/telemetry/metric.py index 82f325a48a..bdb74e095c 100644 --- a/samcli/lib/telemetry/metric.py +++ b/samcli/lib/telemetry/metric.py @@ -7,6 +7,8 @@ import uuid import platform import logging +from typing import Optional + import click from samcli import __version__ as samcli_version @@ -327,7 +329,8 @@ def _add_common_metric_attributes(self): self._data["pyversion"] = platform.python_version() self._data["samcliVersion"] = samcli_version - def _default_session_id(self): + @staticmethod + def _default_session_id() -> Optional[str]: """ Get the default SessionId from Click Context. Fail silently if Context does not exist. @@ -341,7 +344,8 @@ def _default_session_id(self): LOG.debug("Unable to find Click Context for getting session_id.") return None - def _get_execution_environment(self): + @staticmethod + def _get_execution_environment() -> str: """ Returns the environment in which SAM CLI is running. Possible options are: diff --git a/samcli/lib/utils/hash.py b/samcli/lib/utils/hash.py index 5aedd51904..78e061e320 100644 --- a/samcli/lib/utils/hash.py +++ b/samcli/lib/utils/hash.py @@ -69,7 +69,7 @@ def dir_checksum(directory, followlinks=True): return md5_dir.hexdigest() -def str_checksum(content): +def str_checksum(content: str) -> str: """ return a md5 checksum of a given string diff --git a/samcli/lib/warnings/sam_cli_warning.py b/samcli/lib/warnings/sam_cli_warning.py index 54c8d29f3c..506529e774 100644 --- a/samcli/lib/warnings/sam_cli_warning.py +++ b/samcli/lib/warnings/sam_cli_warning.py @@ -2,6 +2,7 @@ Provides all Warnings checkers for sam template """ import logging +from typing import Dict LOG = logging.getLogger(__name__) @@ -11,7 +12,7 @@ class TemplateWarning: Top level class which all warnings should extend from. """ - def check(self, template_dict): + def check(self, template_dict): # pylint: disable=no-self-use raise Exception("NotImplementedException") @@ -115,10 +116,12 @@ def check(self, template_dict): return (True, self.WARNING_MESSAGE) return (False, "") - def _have_condition(self, function): + @staticmethod + def _have_condition(function: Dict) -> bool: condition = function.get("Condition", None) return condition is not None - def _have_deployment_preferences(self, function): + @staticmethod + def _have_deployment_preferences(function: Dict) -> bool: deployment_preference = function.get("Properties", {}).get("DeploymentPreference", None) return deployment_preference is not None diff --git a/samcli/local/apigw/local_apigw_service.py b/samcli/local/apigw/local_apigw_service.py index d940a9fada..0d1cf62006 100644 --- a/samcli/local/apigw/local_apigw_service.py +++ b/samcli/local/apigw/local_apigw_service.py @@ -346,7 +346,8 @@ def _get_current_route(self, flask_request): return route - def get_request_methods_endpoints(self, flask_request): + @staticmethod + def get_request_methods_endpoints(flask_request): """ Separated out for testing requests in request handler :param request flask_request: Flask Request diff --git a/samcli/local/docker/manager.py b/samcli/local/docker/manager.py index 5776c31121..0919562346 100644 --- a/samcli/local/docker/manager.py +++ b/samcli/local/docker/manager.py @@ -12,6 +12,7 @@ from samcli.lib.utils.stream_writer import StreamWriter from samcli.local.docker import utils +from samcli.local.docker.container import Container LOG = logging.getLogger(__name__) @@ -113,7 +114,8 @@ def run(self, container, input_data=None): container.start(input_data=input_data) - def stop(self, container): + @staticmethod + def stop(container: Container) -> None: """ Stop and delete the container @@ -181,7 +183,8 @@ def has_image(self, image_name): except docker.errors.ImageNotFound: return False - def _is_rapid_image(self, image_name): + @staticmethod + def _is_rapid_image(image_name: str) -> bool: """ Is the image tagged as a RAPID clone? diff --git a/samcli/local/layers/layer_downloader.py b/samcli/local/layers/layer_downloader.py index 6bd424264f..676d7b2c07 100644 --- a/samcli/local/layers/layer_downloader.py +++ b/samcli/local/layers/layer_downloader.py @@ -156,7 +156,8 @@ def _fetch_layer_uri(self, layer): return layer_version_response.get("Content").get("Location") - def _is_layer_cached(self, layer_path): + @staticmethod + def _is_layer_cached(layer_path: Path) -> bool: """ Checks if the layer is already cached on the system diff --git a/tests/unit/lib/build_module/test_build_strategy.py b/tests/unit/lib/build_module/test_build_strategy.py index fc4c101f8c..668ed97bd9 100644 --- a/tests/unit/lib/build_module/test_build_strategy.py +++ b/tests/unit/lib/build_module/test_build_strategy.py @@ -39,14 +39,22 @@ def setUp(self): self.build_graph.put_layer_build_definition(self.layer_build_definition2, Mock()) +class _TestBuildStrategy(BuildStrategy): + def build_single_function_definition(self, build_definition): + return {} + + def build_single_layer_definition(self, layer_definition): + return {} + + class BuildStrategyTest(BuildStrategyBaseTest): def test_build_functions_layers(self): - build_strategy = BuildStrategy(self.build_graph) + build_strategy = _TestBuildStrategy(self.build_graph) self.assertEqual(build_strategy.build(), {}) def test_build_functions_layers_mock(self): - mock_build_strategy = BuildStrategy(self.build_graph) + mock_build_strategy = _TestBuildStrategy(self.build_graph) given_build_functions_result = {"function1": "build_dir_1"} given_build_layers_result = {"layer1": "layer_dir_1"} @@ -63,7 +71,7 @@ def test_build_functions_layers_mock(self): mock_build_strategy._build_layers.assert_called_once_with(self.build_graph) def test_build_single_function_layer(self): - mock_build_strategy = BuildStrategy(self.build_graph) + mock_build_strategy = _TestBuildStrategy(self.build_graph) given_build_functions_result = [{"function1": "build_dir_1"}, {"function2": "build_dir_2"}] given_build_layers_result = [{"layer1": "layer_dir_1"}, {"layer2": "layer_dir_2"}] @@ -304,7 +312,7 @@ def test_redundant_cached_should_be_clean(self): class ParallelBuildStrategyTest(BuildStrategyBaseTest): def test_given_async_context_should_call_expected_methods(self): mock_async_context = Mock() - delegate_build_strategy = MagicMock(wraps=BuildStrategy(self.build_graph)) + delegate_build_strategy = MagicMock(wraps=_TestBuildStrategy(self.build_graph)) parallel_build_strategy = ParallelBuildStrategy(self.build_graph, delegate_build_strategy, mock_async_context) given_build_results = [ @@ -337,7 +345,7 @@ def test_given_async_context_should_call_expected_methods(self): def test_given_delegate_strategy_it_should_call_delegated_build_methods(self): # create a mock delegate build strategy - delegate_build_strategy = MagicMock(wraps=BuildStrategy(self.build_graph)) + delegate_build_strategy = MagicMock(wraps=_TestBuildStrategy(self.build_graph)) delegate_build_strategy.build_single_function_definition.return_value = { "function1": "build_location1", "function2": "build_location2",