Fix when 2 function have the same name, run in parallel (#135)

* Fix when 2 function have the same name, run in parallel

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* Fix typo, do not use shell=True to run grep command

* use built-in types

* use built-in types

* check that we run from git repositoyrt

* Remove commented line

* remove deplicate debug log

---------

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>

Author: myakove

apps/jira_utils/jira_information.py

@@ -1,20 +1,19 @@
+from __future__ import annotations
+
+import concurrent.futures
 import logging
+import os
 import re
 import sys
-import os
-import concurrent.futures
 from functools import lru_cache
+from typing import Any
 
 import click
-from jira import JIRA, JIRAError, Issue
-
+from jira import JIRA, Issue, JIRAError
 from simple_logger.logger import get_logger
-
-from apps.utils import ListParamType
-from typing import Dict, List, Set, Tuple, Any
-
 from tenacity import retry, retry_if_exception_type, stop_after_attempt, wait_fixed
-from apps.utils import all_python_files, get_util_config
+
+from apps.utils import ListParamType, all_python_files, get_util_config
 
 LOGGER = get_logger(name=__name__)
 
@@ -29,7 +28,7 @@ def get_issue(
     return jira.issue(id=jira_id, fields="status, issuetype, fixVersions")
 
 
-def get_jira_ids_from_file_content(file_content: str, issue_pattern: str, jira_url: str) -> Set[str]:
+def get_jira_ids_from_file_content(file_content: str, issue_pattern: str, jira_url: str) -> set[str]:
     """
     Try to find all Jira tickets in a given file content.
 
@@ -55,7 +54,7 @@ def get_jira_ids_from_file_content(file_content: str, issue_pattern: str, jira_u
     return set(_pytest_jira_marker_bugs + _jira_id_arguments + _jira_url_jiras)
 
 
-def get_jiras_from_python_files(issue_pattern: str, jira_url: str) -> Dict[str, Set[str]]:
+def get_jiras_from_python_files(issue_pattern: str, jira_url: str) -> dict[str, set[str]]:
     """
     Get all python files from the current directory and get list of jira ids from each of them
 
@@ -68,7 +67,7 @@ def get_jiras_from_python_files(issue_pattern: str, jira_url: str) -> Dict[str,
 
     Note: any line containing <skip-jira_utils-check> would be not be checked for presence of a jira id
     """
-    jira_found: Dict[str, Set[str]] = {}
+    jira_found: dict[str, set[str]] = {}
     for filename in all_python_files():
         file_content = []
         with open(filename) as fd:
@@ -91,12 +90,12 @@ def get_jiras_from_python_files(issue_pattern: str, jira_url: str) -> Dict[str,
 def get_jira_information(
     jira_object: JIRA,
     jira_id: str,
-    skip_project_ids: List[str],
-    resolved_status: List[str],
-    jira_target_versions: List[str],
+    skip_project_ids: list[str],
+    resolved_status: list[str],
+    jira_target_versions: list[str],
     target_version_str: str,
     file_name: str,
-) -> Tuple[str, str]:
+) -> tuple[str, str]:
     jira_error_string = ""
     try:
         # check resolved status:
@@ -133,11 +132,11 @@ def process_jira_command_line_config_file(
     url: str,
     token: str,
     issue_pattern: str,
-    resolved_statuses: List[str],
+    resolved_statuses: list[str],
     version_string_not_targeted_jiras: str,
-    target_versions: List[str],
-    skip_projects: List[str],
-) -> Dict[str, Any]:
+    target_versions: list[str],
+    skip_projects: list[str],
+) -> dict[str, Any]:
     # Process all the arguments passed from command line or config file or environment variable
     config_dict = get_util_config(util_name="pyutils-jira", config_file_path=config_file_path)
     url = url or config_dict.get("url", "")
@@ -201,11 +200,11 @@ def process_jira_command_line_config_file(
 @click.option("--verbose", default=False, is_flag=True)
 def get_jira_mismatch(
     config_file_path: str,
-    target_versions: List[str],
+    target_versions: list[str],
     url: str,
     token: str,
-    skip_projects: List[str],
-    resolved_statuses: List[str],
+    skip_projects: list[str],
+    resolved_statuses: list[str],
     issue_pattern: str,
     version_string_not_targeted_jiras: str,
     verbose: bool,
@@ -228,7 +227,7 @@ def get_jira_mismatch(
     )
 
     jira_obj = JIRA(token_auth=jira_config_dict["token"], options={"server": jira_config_dict["url"]})
-    jira_error: Dict[str, str] = {}
+    jira_error: dict[str, str] = {}
 
     if jira_id_dict := get_jiras_from_python_files(
         issue_pattern=jira_config_dict["issue_pattern"], jira_url=jira_config_dict["url"]

apps/polarion/polarion_set_automated.py

@@ -1,27 +1,27 @@
 from __future__ import annotations
+
 import logging
 import os
-from simple_logger.logger import get_logger
 import sys
-import click
 
-from apps.polarion.polarion_utils import get_polarion_project_id, find_polarion_ids, update_polarion_ids
-from typing import List, Dict, Optional
+import click
+from simple_logger.logger import get_logger
 
+from apps.polarion.polarion_utils import find_polarion_ids, get_polarion_project_id, update_polarion_ids
 
 LOGGER = get_logger(name=__name__)
 
 
-def approve_tests(polarion_project_id: str, added_ids: List[str]) -> Dict[str, List[str]]:
+def approve_tests(polarion_project_id: str, added_ids: list[str]) -> dict[str, list[str]]:
     LOGGER.debug(f"Following polarion ids were added: {added_ids}")
     return update_polarion_ids(
         polarion_ids=list(added_ids), project_id=polarion_project_id, is_automated=True, is_approved=True
     )
 
 
 def remove_approved_tests(
-    polarion_project_id: str, branch: str, added_ids: Optional[List[str]] = None
-) -> Dict[str, List[str]]:
+    polarion_project_id: str, branch: str, added_ids: list[str] | None = None
+) -> dict[str, list[str]]:
     removed_polarions = {}
     added_ids = added_ids or []
     if removed_ids := set(

apps/polarion/polarion_utils.py

@@ -1,12 +1,13 @@
 from __future__ import annotations
+
 import re
-import sys
-from simple_logger.logger import get_logger
 import shlex
 import subprocess
+import sys
+
+from simple_logger.logger import get_logger
 
 from apps.utils import get_util_config
-from typing import Dict, List
 
 LOGGER = get_logger(name=__name__)
 AUTOMATED = "automated"
@@ -19,8 +20,8 @@ def git_diff(branch: str) -> str:
     return data.decode()
 
 
-def git_diff_lines(branch: str) -> Dict[str, List[str]]:
-    diff: Dict[str, List[str]] = {}
+def git_diff_lines(branch: str) -> dict[str, list[str]]:
+    diff: dict[str, list[str]] = {}
     for line in git_diff(branch=branch).splitlines():
         LOGGER.debug(line)
         if line.startswith("+"):
@@ -31,13 +32,13 @@ def git_diff_lines(branch: str) -> Dict[str, List[str]]:
 
 
 def validate_polarion_requirements(
-    polarion_test_ids: List[str],
+    polarion_test_ids: list[str],
     polarion_project_id: str,
-) -> List[str]:
-    tests_with_missing_requirements: List[str] = []
+) -> list[str]:
+    tests_with_missing_requirements: list[str] = []
     if polarion_test_ids:
-        from pylero.work_item import TestCase, Requirement
         from pylero.exceptions import PyleroLibException
+        from pylero.work_item import Requirement, TestCase
 
         for _id in polarion_test_ids:
             has_req = False
@@ -57,7 +58,7 @@ def validate_polarion_requirements(
     return tests_with_missing_requirements
 
 
-def find_polarion_ids(polarion_project_id: str, string_to_match: str, branch: str) -> List[str]:
+def find_polarion_ids(polarion_project_id: str, string_to_match: str, branch: str) -> list[str]:
     return re.findall(
         rf"pytest.mark.polarion.*({polarion_project_id}-[0-9]+)",
         "\n".join(git_diff_lines(branch=branch).get(string_to_match, [])),
@@ -74,14 +75,14 @@ def get_polarion_project_id(util_name: str, config_file_path: str) -> str:
 
 
 def update_polarion_ids(
-    project_id: str, is_automated: bool, polarion_ids: List[str], is_approved: bool = False
-) -> Dict[str, List[str]]:
-    updated_ids: Dict[str, List[str]] = {}
+    project_id: str, is_automated: bool, polarion_ids: list[str], is_approved: bool = False
+) -> dict[str, list[str]]:
+    updated_ids: dict[str, list[str]] = {}
     if polarion_ids:
         automation_status = AUTOMATED if is_automated else NOT_AUTOMATED
 
-        from pylero.work_item import TestCase
         from pylero.exceptions import PyleroLibException
+        from pylero.work_item import TestCase
 
         for id in polarion_ids:
             try:

apps/polarion/polarion_verify_tc_requirements.py

@@ -1,10 +1,11 @@
 import logging
-from simple_logger.logger import get_logger
 import os
-import click
 import sys
 
-from apps.polarion.polarion_utils import validate_polarion_requirements, find_polarion_ids, get_polarion_project_id
+import click
+from simple_logger.logger import get_logger
+
+from apps.polarion.polarion_utils import find_polarion_ids, get_polarion_project_id, validate_polarion_requirements
 
 LOGGER = get_logger(name="polarion-verify-tc-requirements")
 

apps/unused_code/unused_code.py

@@ -1,20 +1,23 @@
+from __future__ import annotations
+
 import ast
 import logging
 import os
 import subprocess
 import sys
+from concurrent.futures import Future, ThreadPoolExecutor, as_completed
+from typing import Any, Iterable
 
 import click
 from simple_logger.logger import get_logger
 
-from apps.utils import all_python_files, ListParamType, get_util_config
-from typing import Any, Iterable, List
+from apps.utils import ListParamType, all_python_files, get_util_config
 
 LOGGER = get_logger(name=__name__)
 
 
 def is_fixture_autouse(func: ast.FunctionDef) -> bool:
-    deco_list: List[Any] = func.decorator_list
+    deco_list: list[Any] = func.decorator_list
     for deco in deco_list or []:
         if not hasattr(deco, "func"):
             continue
@@ -39,17 +42,55 @@ def _iter_functions(tree: ast.Module) -> Iterable[ast.FunctionDef]:
             yield elm
 
 
-def is_ignore_function_list(ignore_prefix_list: List[str], function: ast.FunctionDef) -> bool:
+def is_ignore_function_list(ignore_prefix_list: list[str], function: ast.FunctionDef) -> bool:
     ignore_function_lists = [
         function.name for ignore_prefix in ignore_prefix_list if function.name.startswith(ignore_prefix)
     ]
     if ignore_function_lists:
-        LOGGER.debug(f"Following functions are getting skipped: {ignore_function_lists}")
         return True
 
     return False
 
 
+def process_file(py_file: str, func_ignore_prefix: list[str], file_ignore_list: list[str]) -> str:
+    if os.path.basename(py_file) in file_ignore_list:
+        LOGGER.debug(f"Skipping file: {py_file}")
+        return ""
+
+    with open(py_file) as fd:
+        tree = ast.parse(source=fd.read())
+
+    for func in _iter_functions(tree=tree):
+        if func_ignore_prefix and is_ignore_function_list(ignore_prefix_list=func_ignore_prefix, function=func):
+            LOGGER.debug(f"Skipping function: {func.name}")
+            continue
+
+        if is_fixture_autouse(func=func):
+            LOGGER.debug(f"Skipping `autouse` fixture function: {func.name}")
+            continue
+
+        used = False
+        _func_grep_found = subprocess.check_output(["git", "grep", "-w", func.name], shell=False)
+
+        for entry in _func_grep_found.decode().splitlines():
+            _, _line = entry.split(":", 1)
+
+            if f"def {func.name}" in _line:
+                continue
+
+            if _line.strip().startswith("#"):
+                continue
+
+            if func.name in _line:
+                used = True
+                break
+
+        if not used:
+            return f"{os.path.relpath(py_file)}:{func.name}:{func.lineno}:{func.col_offset} Is not used anywhere in the code."
+
+    return ""
+
+
 @click.command()
 @click.option(
     "--config-file-path",
@@ -67,42 +108,39 @@ def is_ignore_function_list(ignore_prefix_list: List[str], function: ast.Functio
     help="Provide a comma-separated string or list of function prefixes to exclude",
     type=ListParamType(),
 )
-@click.option("--verbose", default=False, is_flag=True)
+@click.option("--verbose", "-v", default=False, is_flag=True)
 def get_unused_functions(
-    config_file_path: Any, exclude_files: Any, exclude_function_prefixes: Any, verbose: bool
-) -> Any:
+    config_file_path: str, exclude_files: list[str], exclude_function_prefixes: list[str], verbose: bool
+) -> None:
     LOGGER.setLevel(logging.DEBUG if verbose else logging.INFO)
 
-    _unused_functions = []
+    unused_functions: list[str] = []
     unused_code_config = get_util_config(util_name="pyutils-unusedcode", config_file_path=config_file_path)
     func_ignore_prefix = exclude_function_prefixes or unused_code_config.get("exclude_function_prefix", [])
     file_ignore_list = exclude_files or unused_code_config.get("exclude_files", [])
 
-    for py_file in all_python_files():
-        if os.path.basename(py_file) in file_ignore_list:
-            continue
-        with open(py_file) as fd:
-            tree = ast.parse(source=fd.read())
+    jobs: list[Future] = []
+    if not os.path.exists(".git"):
+        LOGGER.error("Must be run from a git repository")
+        sys.exit(1)
 
-        for func in _iter_functions(tree=tree):
-            if func_ignore_prefix and is_ignore_function_list(ignore_prefix_list=func_ignore_prefix, function=func):
-                continue
+    with ThreadPoolExecutor() as executor:
+        for py_file in all_python_files():
+            jobs.append(
+                executor.submit(
+                    process_file,
+                    py_file=py_file,
+                    func_ignore_prefix=func_ignore_prefix,
+                    file_ignore_list=file_ignore_list,
+                )
+            )
 
-            if is_fixture_autouse(func=func):
-                continue
+        for result in as_completed(jobs):
+            if unused_func := result.result():
+                unused_functions.append(unused_func)
 
-            _used = subprocess.check_output(
-                f"git grep -w '{func.name}' | wc -l",
-                shell=True,
-            )
-            used = int(_used.strip())
-            if used < 2:
-                _unused_functions.append(
-                    f"{os.path.relpath(py_file)}:{func.name}:{func.lineno}:{func.col_offset} Is"
-                    " not used anywhere in the code.",
-                )
-    if _unused_functions:
-        click.echo("\n".join(_unused_functions))
+    if unused_functions:
+        click.echo("\n".join(unused_functions))
         sys.exit(1)