ci: Add public interface command check

bin/public_interface.py

@@ -13,7 +13,11 @@
 import json
 import os.path
 import pkgutil
-from typing import Any, Dict, Set, Union
+import sys
+from pathlib import Path
+from typing import Any, Dict, List, NamedTuple, Set, Union
+
+_ARGUMENT_SELF = {"kind": "POSITIONAL_OR_KEYWORD", "name": "self"}
 
 
 class InterfaceScanner:
@@ -91,21 +95,141 @@ def _print(signature: Dict[str, inspect.Signature], variables: Set[str]) -> None
     print(json.dumps(result, indent=2, sort_keys=True))
 
 
+class _BreakingChanges(NamedTuple):
+    deleted_variables: List[str]
+    deleted_routines: List[str]
+    incompatible_routines: List[str]
+
+    def is_empty(self) -> bool:
+        return not any([self.deleted_variables, self.deleted_routines, self.incompatible_routines])
+
+    @staticmethod
+    def _argument_to_str(argument: Dict[str, Any]) -> str:
+        if "default" in argument:
+            return f'{argument["name"]}={argument["default"]}'
+        return str(argument["name"])
+
+    def print_markdown(
+        self,
+        original_routines: Dict[str, List[Dict[str, Any]]],
+        routines: Dict[str, List[Dict[str, Any]]],
+    ) -> None:
+        """Print all breaking changes in markdown."""
+        print("\n# Compatibility breaking changes:")
+        print("**These changes are considered breaking changes and may break packages consuming")
+        print("the PyPI package [aws-sam-translator](https://pypi.org/project/aws-sam-translator/).")
+        print("Please consider revisiting these changes to make sure they are intentional:**")
+        if self.deleted_variables:
+            print("\n## Deleted module level variables")
+            for name in self.deleted_variables:
+                print(f"- {name}")
+        if self.deleted_variables:
+            print("\n## Deleted routines")
+            for name in self.deleted_routines:
+                print(f"- {name}")
+        if self.incompatible_routines:
+            print("\n## Deleted routines")
+            for name in self.incompatible_routines:
+                before = f"({', '.join(self._argument_to_str(arg) for arg in original_routines[name])})"
+                after = f"({', '.join(self._argument_to_str(arg) for arg in routines[name])})"
+                print(f"- {name}: `{before}` -> `{after}`")
+
+
+def _only_new_optional_arguments_or_existing_arguments_optionalized_or_var_arguments(
+    original_arguments: List[Dict[str, Any]], arguments: List[Dict[str, Any]]
+) -> bool:
+    if len(original_arguments) > len(arguments):
+        return False
+    for i, original_argument in enumerate(original_arguments):
+        if original_argument == arguments[i]:
+            continue
+        if (
+            original_argument["name"] == arguments[i]["name"]
+            and original_argument["kind"] == arguments[i]["kind"]
+            and "default" not in original_argument
+            and "default" in arguments[i]
+        ):
+            continue
+        return False
+    # it is an optional argument when it has a default value:
+    return all(
+        [
+            "default" in argument or argument["kind"] in ("VAR_KEYWORD", "VAR_POSITIONAL")
+            for argument in arguments[len(original_arguments) :]
+        ]
+    )
+
+
+def _is_compatible(original_arguments: List[Dict[str, Any]], arguments: List[Dict[str, Any]]) -> bool:
+    """
+    If there is an argument change, it is compatible only when
+    - new optional arguments are added or existing arguments become optional.
+    - var arguments (*args, **kwargs) are added
+    - self is removed (method -> staticmethod).
+    - combination of above
+    """
+    if (
+        original_arguments == arguments
+        or _only_new_optional_arguments_or_existing_arguments_optionalized_or_var_arguments(
+            original_arguments, arguments
+        )
+    ):
+        return True
+    if original_arguments and original_arguments[0] == _ARGUMENT_SELF:
+        original_arguments_without_self = original_arguments[1:]
+        if _is_compatible(original_arguments_without_self, arguments):
+            return True
+    return False
+
+
+def _detect_breaking_changes(
+    original_routines: Dict[str, List[Dict[str, Any]]],
+    original_variables: Set[str],
+    routines: Dict[str, List[Dict[str, Any]]],
+    variables: Set[str],
+) -> _BreakingChanges:
+    deleted_routines: List[str] = []
+    incompatible_routines: List[str] = []
+    for routine_path, arguments in original_routines.items():
+        if routine_path not in routines:
+            deleted_routines.append(routine_path)
+        elif not _is_compatible(arguments, routines[routine_path]):
+            incompatible_routines.append(routine_path)
+    return _BreakingChanges(
+        sorted(set(original_variables) - set(variables)), sorted(deleted_routines), sorted(incompatible_routines)
+    )
+
+
 def main() -> None:
     parser = argparse.ArgumentParser(description=__doc__)
 
     subparsers = parser.add_subparsers(dest="command")
     extract = subparsers.add_parser("extract", help="Extract public interfaces")
     extract.add_argument("--module", help="The module to extract public interfaces", type=str, default="samtranslator")
+    check = subparsers.add_parser("check", help="Check public interface changes")
+    check.add_argument("original_json", help="The original public interface JSON file", type=Path)
+    check.add_argument("new_json", help="The new public interface JSON file", type=Path)
     args = parser.parse_args()
 
     if args.command == "extract":
         scanner = InterfaceScanner()
         scanner.scan_interfaces_recursively(args.module)
         _print(scanner.signatures, scanner.variables)
-    # TODO: handle compare
+    elif args.command == "check":
+        original_json = json.loads(args.original_json.read_text())
+        new_json = json.loads(args.new_json.read_text())
+        breaking_changes = _detect_breaking_changes(
+            original_json["routines"], original_json["variables"], new_json["routines"], new_json["variables"]
+        )
+        if breaking_changes.is_empty():
+            sys.stderr.write("No compatibility breaking changes detected.\n")
+        else:
+            sys.stderr.write("Compatibility breaking changes detected!!!\n")
+            breaking_changes.print_markdown(original_json["routines"], new_json["routines"])
+            sys.exit(1)
     else:
         parser.print_help()
+        sys.exit(1)
 
 
 if __name__ == "__main__":

tests/bin/test_public_interface.py

@@ -0,0 +1,175 @@
+from unittest import TestCase
+
+from bin.public_interface import _BreakingChanges, _detect_breaking_changes
+
+
+class TestDetectBreakingChanges(TestCase):
+    def test_missing_variables(self):
+        self.assertEqual(
+            _detect_breaking_changes(
+                {},
+                ["samtranslator.model.CONST_X", "samtranslator.model.CONST_Y"],
+                {},
+                ["samtranslator.model.CONST_X", "samtranslator.model.CONST_Z"],
+            ),
+            _BreakingChanges(
+                deleted_variables=["samtranslator.model.CONST_Y"], deleted_routines=[], incompatible_routines=[]
+            ),
+        )
+
+    def test_missing_routines(self):
+        self.assertEqual(
+            _detect_breaking_changes(
+                {"samtranslator.model.do_something": []},
+                [],
+                {"samtranslator.model.do_something_2": []},
+                [],
+            ),
+            _BreakingChanges(
+                deleted_variables=[], deleted_routines=["samtranslator.model.do_something"], incompatible_routines=[]
+            ),
+        )
+
+    def test_routines_still_compatible_when_adding_optional_arguments(self):
+        self.assertEqual(
+            _detect_breaking_changes(
+                {"samtranslator.model.do_something": []},
+                [],
+                {
+                    "samtranslator.model.do_something": [
+                        {"name": "new_arg", "kind": "POSITIONAL_OR_KEYWORD", "default": False},
+                        {"name": "new_arg_2", "kind": "POSITIONAL_OR_KEYWORD", "default": None},
+                    ]
+                },
+                [],
+            ),
+            _BreakingChanges(deleted_variables=[], deleted_routines=[], incompatible_routines=[]),
+        )
+
+    def test_routines_still_compatible_when_optionalize_existing_arguments(self):
+        self.assertEqual(
+            _detect_breaking_changes(
+                {
+                    "samtranslator.model.do_something": [
+                        {
+                            "name": "arg",
+                            "kind": "POSITIONAL_OR_KEYWORD",
+                        },
+                        {
+                            "name": "arg_2",
+                            "kind": "POSITIONAL_OR_KEYWORD",
+                        },
+                    ]
+                },
+                [],
+                {
+                    "samtranslator.model.do_something": [
+                        {"name": "arg", "kind": "POSITIONAL_OR_KEYWORD", "default": False},
+                        {"name": "arg_2", "kind": "POSITIONAL_OR_KEYWORD", "default": None},
+                    ]
+                },
+                [],
+            ),
+            _BreakingChanges(deleted_variables=[], deleted_routines=[], incompatible_routines=[]),
+        )
+
+    def test_routines_still_compatible_when_adding_var_arguments(self):
+        self.assertEqual(
+            _detect_breaking_changes(
+                {"samtranslator.model.do_something": []},
+                [],
+                {
+                    "samtranslator.model.do_something": [
+                        {"name": "args", "kind": "VAR_POSITIONAL"},
+                        {"name": "kwargs", "kind": "VAR_KEYWORD"},
+                    ]
+                },
+                [],
+            ),
+            _BreakingChanges(deleted_variables=[], deleted_routines=[], incompatible_routines=[]),
+        )
+
+    def test_routines_still_compatible_when_converting_from_method_to_staticmethod(self):
+        self.assertEqual(
+            _detect_breaking_changes(
+                {
+                    "samtranslator.model.do_something": [
+                        {"kind": "POSITIONAL_OR_KEYWORD", "name": "self"},
+                        {"name": "some_arg", "kind": "POSITIONAL_OR_KEYWORD"},
+                    ]
+                },
+                [],
+                {"samtranslator.model.do_something": [{"name": "some_arg", "kind": "POSITIONAL_OR_KEYWORD"}]},
+                [],
+            ),
+            _BreakingChanges(deleted_variables=[], deleted_routines=[], incompatible_routines=[]),
+        )
+
+    def test_routines_still_compatible_when_converting_from_method_to_staticmethod_and_adding_optional_arguments(self):
+        self.assertEqual(
+            _detect_breaking_changes(
+                {
+                    "samtranslator.model.do_something": [
+                        {"kind": "POSITIONAL_OR_KEYWORD", "name": "self"},
+                        {"name": "some_arg", "kind": "POSITIONAL_OR_KEYWORD"},
+                    ]
+                },
+                [],
+                {
+                    "samtranslator.model.do_something": [
+                        {"name": "some_arg", "kind": "POSITIONAL_OR_KEYWORD"},
+                        {"name": "new_arg", "kind": "POSITIONAL_OR_KEYWORD", "default": False},
+                        {"name": "new_arg_2", "kind": "POSITIONAL_OR_KEYWORD", "default": None},
+                    ]
+                },
+                [],
+            ),
+            _BreakingChanges(deleted_variables=[], deleted_routines=[], incompatible_routines=[]),
+        )
+
+    def test_routines_incompatible_when_changing_default_value(self):
+        self.assertEqual(
+            _detect_breaking_changes(
+                {
+                    "samtranslator.model.do_something": [
+                        {"kind": "POSITIONAL_OR_KEYWORD", "name": "self"},
+                        {"name": "some_arg", "kind": "POSITIONAL_OR_KEYWORD", "default": 0},
+                    ]
+                },
+                [],
+                {
+                    "samtranslator.model.do_something": [
+                        {"kind": "POSITIONAL_OR_KEYWORD", "name": "self"},
+                        {"name": "some_arg", "kind": "POSITIONAL_OR_KEYWORD", "default": 1},
+                    ]
+                },
+                [],
+            ),
+            _BreakingChanges(
+                deleted_variables=[], deleted_routines=[], incompatible_routines=["samtranslator.model.do_something"]
+            ),
+        )
+
+    def test_routines_incompatible_when_add_new_arguments(self):
+        self.assertEqual(
+            _detect_breaking_changes(
+                {
+                    "samtranslator.model.do_something": [
+                        {"kind": "POSITIONAL_OR_KEYWORD", "name": "self"},
+                        {"name": "some_arg", "kind": "POSITIONAL_OR_KEYWORD"},
+                    ]
+                },
+                [],
+                {
+                    "samtranslator.model.do_something": [
+                        {"kind": "POSITIONAL_OR_KEYWORD", "name": "self"},
+                        {"name": "some_arg", "kind": "POSITIONAL_OR_KEYWORD"},
+                        {"name": "new_arg", "kind": "POSITIONAL_OR_KEYWORD"},
+                    ]
+                },
+                [],
+            ),
+            _BreakingChanges(
+                deleted_variables=[], deleted_routines=[], incompatible_routines=["samtranslator.model.do_something"]
+            ),
+        )