Feature/sg 1057 add breaking change tests

.circleci/config.yml

@@ -220,6 +220,36 @@ jobs:
       - store_artifacts:
           path: ~/sg_logs
 
+  breaking_change_check:
+    parameters:
+      py_version:
+        type: string
+        default: latest
+    docker:
+      - image: cimg/python:<< parameters.py_version >>
+    resource_class: small
+    steps:
+      - checkout
+      - run:
+          name: Detect breaking changes
+          no_output_timeout: 30m
+          command: |
+            python3 -m venv venv
+            . venv/bin/activate
+            python3 -m pip install pip==23.1.2
+            python3 -m pip install -e .
+            python3 -m pip install gitpython==3.1.0
+            coverage run --source=super_gradients -m unittest tests/breaking_change_tests/unit_test.py
+            coverage run --source=super_gradients -m unittest tests/breaking_change_tests/test_detect_breaking_change.py
+      - run:
+          name: Remove new environment when failed
+          command: "rm -r venv"
+          when: on_fail
+      - slack/notify:
+          channel: "sg-integration-tests"
+          event: fail
+          template: basic_fail_1 # see https://github.com/CircleCI-Public/slack-orb/wiki#templates.
+
   change_rc_to_b:
      description: "change rc in the tag to b"
      docker:
@@ -906,6 +936,12 @@ workflows:
           requires:
             - deci-common/persist_version_info
 
+      - breaking_change_check:
+          name: "breaking-change-check"
+          py_version: << pipeline.parameters.build_py_version >>
+          requires:
+            - deci-common/persist_version_info
+
       - deci-common/codeartifact_login:
           repo_name: "deci-packages"
           <<: *release_candidate_filter

requirements.dev.txt

@@ -1,3 +1,4 @@
 flake8==5.0.4
 black==22.10.0
 pre-commit==2.20.0
+gitpython>=3.1.0

tests/breaking_change_tests/breaking_changes_detection.py

@@ -0,0 +1,243 @@
+import ast
+from abc import ABC
+from pathlib import Path
+from typing import List, Dict, Union
+
+from termcolor import colored
+from dataclasses import dataclass, field, asdict
+
+from .code_parser import parse_functions_signatures, parse_imports
+
+
+MODULE_PATH_COLOR = "yellow"
+SOURCE_CODE_COLOR = "blue"
+BREAKING_OBJECT_COLOR = "red"
+
+
+@dataclass
+class AbstractBreakingChange(ABC):
+    line_num: int
+
+    @property
+    def description(self) -> str:
+        raise NotImplementedError()
+
+    @property
+    def breaking_type_name(self) -> str:
+        raise NotImplementedError()
+
+
+@dataclass
+class ClassRemoved(AbstractBreakingChange):
+    class_name: str
+    line_num: int
+
+    @property
+    def description(self) -> str:
+        return f"{colored(self.class_name, SOURCE_CODE_COLOR)} -> {colored('X', BREAKING_OBJECT_COLOR)}"
+
+    @property
+    def breaking_type_name(self) -> str:
+        return "CLASS REMOVED"
+
+
+@dataclass
+class ImportRemoved(AbstractBreakingChange):
+    import_name: str
+    line_num: int
+
+    @property
+    def description(self) -> str:
+        return f"{colored(self.import_name, SOURCE_CODE_COLOR)} -> {colored('X', BREAKING_OBJECT_COLOR)}"
+
+    @property
+    def breaking_type_name(self) -> str:
+        return "IMPORT REMOVED"
+
+
+@dataclass
+class FunctionRemoved(AbstractBreakingChange):
+    function_name: str
+    line_num: int
+
+    @property
+    def description(self) -> str:
+        return f"{colored(self.function_name, SOURCE_CODE_COLOR)} -> {colored('X', BREAKING_OBJECT_COLOR)}"
+
+    @property
+    def breaking_type_name(self) -> str:
+        return "FUNCTION REMOVED"
+
+
+@dataclass
+class ParameterRemoved(AbstractBreakingChange):
+    parameter_name: str
+    function_name: str
+    line_num: int
+
+    @property
+    def description(self) -> str:
+        source_fn_colored = colored(self.function_name, SOURCE_CODE_COLOR)
+        current_fn_colored = colored(self.function_name, "yellow")
+        param_colored = colored(self.parameter_name, BREAKING_OBJECT_COLOR)
+        return f"{source_fn_colored}(..., {param_colored}) -> {current_fn_colored}(...)"
+
+    @property
+    def breaking_type_name(self) -> str:
+        return "FUNCTION PARAMETER REMOVED"
+
+
+@dataclass
+class RequiredParameterAdded(AbstractBreakingChange):
+    parameter_name: str
+    function_name: str
+    line_num: int
+
+    @property
+    def description(self) -> str:
+        source_fn_colored = colored(self.function_name, SOURCE_CODE_COLOR)
+        current_fn_colored = colored(self.function_name, "yellow")
+        param_colored = colored(self.parameter_name, BREAKING_OBJECT_COLOR)
+        return f"{source_fn_colored}(...) -> {current_fn_colored}(..., {param_colored})"
+
+    @property
+    def breaking_type_name(self) -> str:
+        return "FUNCTION PARAMETER ADDED"
+
+
+@dataclass
+class BreakingChanges:
+    module_path: str
+    classes_removed: List[ClassRemoved] = field(default_factory=list)
+    imports_removed: List[ImportRemoved] = field(default_factory=list)
+    functions_removed: List[FunctionRemoved] = field(default_factory=list)
+    params_removed: List[ParameterRemoved] = field(default_factory=list)
+    required_params_added: List[RequiredParameterAdded] = field(default_factory=list)
+
+    def __str__(self) -> str:
+        summary = ""
+        module_path_colored = colored(self.module_path, MODULE_PATH_COLOR)
+
+        breaking_changes: List[AbstractBreakingChange] = (
+            self.classes_removed + self.imports_removed + self.functions_removed + self.params_removed + self.required_params_added
+        )
+        for breaking_change in breaking_changes:
+
+            summary += "{:<70} {:<8} {:<30} {}\n".format(
+                module_path_colored, breaking_change.line_num, breaking_change.breaking_type_name, breaking_change.description
+            )
+
+        return summary
+
+    def json(self) -> Dict[str, List[str]]:
+        return asdict(self)
+
+    @property
+    def is_empty(self) -> bool:
+        return len(self.classes_removed + self.imports_removed + self.functions_removed + self.params_removed + self.required_params_added) == 0
+
+
+def extract_code_breaking_changes(module_path: str, source_code: str, current_code: str) -> BreakingChanges:
+    """Compares two versions of code to identify breaking changes.
+
+    :param module_path: The path to the module being compared.
+    :param source_code: The source version of the code.
+    :param current_code: The modified version of the code.
+    :return: A BreakingChanges object detailing the differences.
+    """
+    breaking_changes = BreakingChanges(module_path=module_path)
+
+    source_classes = {node.name: node for node in ast.walk(ast.parse(source_code)) if isinstance(node, ast.ClassDef)}
+    current_classes = {node.name: node for node in ast.walk(ast.parse(current_code)) if isinstance(node, ast.ClassDef)}
+
+    # ClassRemoved
+    for class_name, source_class in source_classes.items():
+        if class_name not in current_classes:
+            breaking_changes.classes_removed.append(
+                ClassRemoved(
+                    class_name=class_name,
+                    line_num=source_class.lineno,
+                )
+            )
+
+    # FUNCTION SIGNATURES
+    source_functions_signatures = parse_functions_signatures(source_code)
+    current_functions_signatures = parse_functions_signatures(current_code)
+    for function_name, source_function_signature in source_functions_signatures.items():
+
+        if function_name in current_functions_signatures:
+            current_function_signature = current_functions_signatures[function_name]
+
+            # ParameterRemoved
+            for source_param in source_function_signature.params.all:
+                if source_param not in current_function_signature.params.all:
+                    breaking_changes.params_removed.append(
+                        ParameterRemoved(
+                            function_name=function_name,
+                            parameter_name=source_param,
+                            line_num=current_function_signature.line_num,
+                        )
+                    )
+
+            # RequiredParameterAdded
+            for current_param in current_function_signature.params.required:
+                if current_param not in source_function_signature.params.required:
+                    breaking_changes.required_params_added.append(
+                        RequiredParameterAdded(
+                            function_name=function_name,
+                            parameter_name=current_param,
+                            line_num=current_function_signature.line_num,
+                        )
+                    )
+
+        else:
+            # FunctionRemoved
+            breaking_changes.functions_removed.append(
+                FunctionRemoved(
+                    function_name=function_name,
+                    line_num=source_function_signature.line_num,
+                )
+            )
+
+    # Check import ONLY if __init__ file.
+    if module_path.endswith("__init__.py"):
+        source_imports = parse_imports(code=source_code)
+        current_imports = parse_imports(code=current_code)
+        breaking_changes.imports_removed = [
+            ImportRemoved(import_name=source_import, line_num=0) for source_import in source_imports if source_import not in current_imports
+        ]
+    return breaking_changes
+
+
+def analyze_breaking_changes(verbose: bool = 1) -> List[Dict[str, Union[str, List]]]:
+    """Analyze changes between the current branch (HEAD) and the master branch.
+    :param verbose: If True, print the summary of breaking changes in a nicely formatted way
+    :return:        List of changes, where each change is a dictionary listing each type of change for each module.
+    """
+    # GitHelper requires `git` library which should NOT be required for the other functions
+    from .git_utils import GitHelper
+
+    root_dir = str(Path(__file__).resolve().parents[2])
+    git_explorer = GitHelper(git_path=root_dir)
+
+    summary = ""
+    breaking_changes_list = []
+    for module_path in git_explorer.diff_files(source_branch="master", current_branch="HEAD"):
+
+        master_code = git_explorer.load_branch_file(branch="master", file_path=module_path)
+        head_code = git_explorer.load_branch_file(branch="HEAD", file_path=module_path)
+        breaking_changes = extract_code_breaking_changes(module_path=module_path, source_code=master_code, current_code=head_code)
+
+        if not breaking_changes.is_empty:
+            breaking_changes_list.append(breaking_changes.json())
+        summary += str(breaking_changes)
+
+    if verbose:
+        if summary:
+            print("{:<60} {:<8} {:<30} {}\n".format("MODULE", "LINE NO", "BREAKING TYPE", "DESCRIPTION (Master -> HEAD)"))
+            print("-" * 175 + "\n")
+            print(summary)
+        else:
+            print(colored("NO BREAKING CHANGE DETECTED!", "green"))
+
+    return breaking_changes_list