Added a method to provide plugin information that is included in artifact zip

.pre-commit-config.yaml

@@ -49,7 +49,7 @@ repos:
   - id: bandit
     files: ^(src|python)/
 - repo: https://github.com/pre-commit/mirrors-mypy
-  rev: v0.711
+  rev: v0.790
   hooks:
   - id: mypy
     files: ^src/

python/rpdk/python/codegen.py

@@ -1,5 +1,6 @@
 import logging
 import os
+import re
 import shutil
 import zipfile
 from pathlib import PurePosixPath
@@ -15,6 +16,7 @@
 from rpdk.core.jsonutils.resolver import ContainerType, resolve_models
 from rpdk.core.plugin_base import LanguagePlugin
 
+from . import __version__
 from .resolver import contains_model, translate_type
 
 LOG = logging.getLogger(__name__)
@@ -169,6 +171,9 @@ def generate(self, project):
 
         LOG.debug("Generate complete")
 
+    def get_plugin_information(self, project):
+        return self._get_plugin_information(project)
+
     def _pre_package(self, build_path):
         f = TemporaryFile("w+b")  # pylint: disable=R1732
 
@@ -233,6 +238,37 @@ def _make_pip_command(base_path):
             str(base_path / "build"),
         ]
 
+    @staticmethod
+    def _get_plugin_information(project):
+        requirements_file = project.root / "requirements.txt"
+        plugin_version = None
+
+        with open(requirements_file) as f:
+            line = f.readline()
+            while line:
+                line = line.strip()
+                if line.startswith(SUPPORT_LIB_NAME):
+                    line_without_prefix = line[len(SUPPORT_LIB_NAME) :]
+
+                    semi_colon = ";"
+                    if semi_colon in line_without_prefix:
+                        index = line_without_prefix.index(semi_colon)
+                        line_without_prefix = line_without_prefix[0:index].strip()
+
+                    plugin_version = re.split("=\\s*", line_without_prefix.strip())[-1]
+                    break
+
+                line = f.readline()
+
+        plugin_info = {
+            "plugin-tool-version": __version__,
+            "plugin-name": "python",
+        }
+        if plugin_version and plugin_version is not None:
+            plugin_info["plugin-version"] = plugin_version
+
+        return plugin_info
+
     @classmethod
     def _docker_build(cls, external_path):
         internal_path = PurePosixPath("/project")

setup.cfg

@@ -16,8 +16,8 @@ exclude =
 max-complexity = 10
 max-line-length = 88
 select = C,E,F,W,B,B950
-# C812, C815, W503 clash with black, F723 false positive
-ignore = E501,C812,C815,C816,W503,F723
+# C812, C815, E203, W503 clash with black, F723 false positive
+ignore = E203,E501,C812,C815,C816,W503,F723
 
 [isort]
 line_length = 88

src/cloudformation_cli_python_lib/recast.py

@@ -14,7 +14,7 @@ def recast_object(
     if not isinstance(json_data, dict):
         raise InvalidRequest(f"Can only parse dict items, not {type(json_data)}")
     # if type is Any, we leave it as is
-    if cls == typing.Any:
+    if cls is typing.Any:
         return
     for k, v in json_data.items():
         if isinstance(v, dict):
@@ -35,7 +35,7 @@ def recast_object(
 
 def _recast_lists(cls: Any, k: str, v: List[Any], classes: Dict[str, Any]) -> List[Any]:
     # Leave as is if type is Any
-    if cls == typing.Any:
+    if cls is typing.Any:
         return v
     if "__dataclass_fields__" not in dir(cls):
         pass
@@ -64,7 +64,7 @@ def cast_sequence_item(cls: Any, k: str, item: Any, classes: Dict[str, Any]) ->
 
 
 def _recast_primitive(cls: Any, k: str, v: Any) -> Any:
-    if cls == typing.Any:
+    if cls is typing.Any:
         # If the type is Any, we cannot guess what the original type was, so we leave
         # it as a string
         return v
@@ -127,5 +127,5 @@ def get_forward_ref_type() -> Any:
     # introspection is valid:
     # https://docs.python.org/3/library/typing.html#typing.ForwardRef
     if "ForwardRef" in dir(typing):
-        return typing.ForwardRef  # type: ignore
+        return typing.ForwardRef
     return typing._ForwardRef  # type: ignore

tests/lib/resource_test.py

@@ -61,7 +61,7 @@ def resource():
 
 def patch_and_raise(resource, str_to_patch, exc_cls, entrypoint):
     with patch.object(resource, str_to_patch) as mock_parse:
-        mock_parse.side_effect = exc_cls("hahaha")
+        mock_parse.side_effect = exc_cls("exception")
         # "un-apply" decorator
         event = entrypoint.__wrapped__(resource, {}, None)  # pylint: disable=no-member
     return event

tests/plugin/codegen_test.py

@@ -14,6 +14,7 @@
 from requests.exceptions import ConnectionError as RequestsConnectionError
 from rpdk.core.exceptions import DownstreamError
 from rpdk.core.project import Project
+from rpdk.python.__init__ import __version__
 from rpdk.python.codegen import (
     SUPPORT_LIB_NAME,
     SUPPORT_LIB_PKG,
@@ -289,6 +290,68 @@ def test__docker_build_good_path(plugin, tmp_path):
     )
 
 
+def test_get_plugin_information_format1(project):
+    plugin_information = project._plugin.get_plugin_information(project)
+
+    assert plugin_information["plugin-tool-version"] == __version__
+    assert plugin_information["plugin-name"] == "python"
+    assert plugin_information["plugin-version"] == "2.1.3"
+
+
+def test_get_plugin_information_format2(project):
+    existing_content = ""
+    with open(project.root / "requirements.txt", "r+") as f:
+        existing_content = f.read()
+        f.seek(0)
+        f.write("cloudformation-cli-python-lib>=2.1.3; python_version < '3.8'")
+        f.truncate()
+
+    plugin_information = project._plugin.get_plugin_information(project)
+
+    with open(project.root / "requirements.txt", "w") as f:
+        f.write(existing_content)
+
+    assert plugin_information["plugin-tool-version"] == __version__
+    assert plugin_information["plugin-name"] == "python"
+    assert plugin_information["plugin-version"] == "2.1.3"
+
+
+def test_get_plugin_information_format3(project):
+    existing_content = ""
+    with open(project.root / "requirements.txt", "r+") as f:
+        existing_content = f.read()
+        f.seek(0)
+        f.write("dummy_line\ncloudformation-cli-python-lib; python_version < '3.8'")
+        f.truncate()
+
+    plugin_information = project._plugin.get_plugin_information(project)
+
+    with open(project.root / "requirements.txt", "w") as f:
+        f.write(existing_content)
+
+    assert plugin_information["plugin-tool-version"] == __version__
+    assert plugin_information["plugin-name"] == "python"
+    assert "plugin-version" not in plugin_information
+
+
+def test_get_plugin_information_no_support_lib(project):
+    existing_content = ""
+    with open(project.root / "requirements.txt", "r+") as f:
+        existing_content = f.read()
+        f.seek(0)
+        f.write("dummy_line1\ndummy_line2")
+        f.truncate()
+
+    plugin_information = project._plugin.get_plugin_information(project)
+
+    with open(project.root / "requirements.txt", "w") as f:
+        f.write(existing_content)
+
+    assert plugin_information["plugin-tool-version"] == __version__
+    assert plugin_information["plugin-name"] == "python"
+    assert "plugin-version" not in plugin_information
+
+
 @pytest.mark.parametrize(
     "exception",
     [