refactor: reduce cognitive complexity for SonarCloud quality gate

Author: nikolay-e

src/treemapper/diffctx/edges/config/terraform.py

@@ -74,19 +74,25 @@ def _collect_tf_dirs_and_sources(tf_files: list[Path]) -> tuple[set[Path], set[s
     return tf_dirs, module_sources
 
 
+def _resolve_module_paths(src: str, tf_dirs: set[Path], repo_root: Path | None) -> list[Path]:
+    paths: list[Path] = []
+    for tf_dir in tf_dirs:
+        try:
+            paths.append((tf_dir / src).resolve())
+        except (ValueError, OSError):
+            pass
+    if repo_root:
+        try:
+            paths.append((repo_root / src.lstrip("./")).resolve())
+        except (ValueError, OSError):
+            pass
+    return paths
+
+
 def _is_in_module(candidate: Path, module_sources: set[str], tf_dirs: set[Path], repo_root: Path | None = None) -> bool:
     for src in module_sources:
-        for tf_dir in tf_dirs:
-            try:
-                module_path = (tf_dir / src).resolve()
-                if candidate.is_relative_to(module_path):
-                    return True
-            except (ValueError, OSError):
-                pass
-        if repo_root:
+        for module_path in _resolve_module_paths(src, tf_dirs, repo_root):
             try:
-                clean_src = src.lstrip("./")
-                module_path = (repo_root / clean_src).resolve()
                 if candidate.is_relative_to(module_path):
                     return True
             except (ValueError, OSError):
@@ -134,7 +140,10 @@ def _extract_qualified_defs(content: str) -> set[str]:
 )
 
 
-def _candidate_references_changed_defs_strict(content: str, changed_defs: set[str]) -> bool:
+_TF_RESOURCE_SKIP_TYPES = frozenset({"var", "local", "data", "module", "path", "terraform", "each", "self", "count"})
+
+
+def _has_non_generic_var_local_ref(content: str, changed_defs: set[str]) -> bool:
     for match in _TF_VAR_REF_RE.finditer(content):
         name = match.group(1)
         if name in changed_defs and name not in _TF_GENERIC_NAMES:
@@ -143,23 +152,29 @@ def _candidate_references_changed_defs_strict(content: str, changed_defs: set[st
         name = match.group(1)
         if name in changed_defs and name not in _TF_GENERIC_NAMES:
             return True
+    return False
+
+
+def _has_data_module_resource_ref(content: str, changed_defs: set[str]) -> bool:
     for match in _TF_DATA_REF_RE.finditer(content):
         data_type, data_name = match.group(1), match.group(2)
         if f"{data_type}.{data_name}" in changed_defs or data_name in changed_defs:
             return True
     for match in _TF_MODULE_REF_RE.finditer(content):
         if match.group(1) in changed_defs:
             return True
-    skip_types = {"var", "local", "data", "module", "path", "terraform", "each", "self", "count"}
     for match in _TF_RESOURCE_REF_RE.finditer(content):
         res_type, res_name, _ = match.groups()
-        if res_type in skip_types:
-            continue
-        if f"{res_type}.{res_name}" in changed_defs or res_name in changed_defs:
-            return True
+        if res_type not in _TF_RESOURCE_SKIP_TYPES:
+            if f"{res_type}.{res_name}" in changed_defs or res_name in changed_defs:
+                return True
     return False
 
 
+def _candidate_references_changed_defs_strict(content: str, changed_defs: set[str]) -> bool:
+    return _has_non_generic_var_local_ref(content, changed_defs) or _has_data_module_resource_ref(content, changed_defs)
+
+
 class _TFIndex:
     var_defs: dict[str, list[FragmentId]]
     resource_defs: dict[str, list[FragmentId]]
@@ -203,28 +218,37 @@ def discover_related_files(
                 pass
 
         changed_set = set(changed_files)
-        discovered: list[Path] = []
-
-        for candidate in all_candidate_files:
-            if candidate in changed_set or not _is_terraform_file(candidate):
-                continue
-            if _is_in_module(candidate, module_sources, tf_dirs, repo_root):
-                discovered.append(candidate)
-                continue
-            if candidate.parent not in tf_dirs:
-                continue
-            try:
-                content = candidate.read_text(encoding="utf-8")
-            except (OSError, UnicodeDecodeError):
-                continue
-            if _candidate_references_changed_defs_strict(content, changed_defs):
-                discovered.append(candidate)
-                continue
-            candidate_defs = _extract_qualified_defs(content)
-            if candidate_defs and any(_candidate_references_changed_defs_strict(c, candidate_defs) for c in changed_contents):
-                discovered.append(candidate)
-
-        return discovered
+        return [
+            c
+            for c in all_candidate_files
+            if c not in changed_set
+            and _is_terraform_file(c)
+            and self._is_related(c, module_sources, tf_dirs, repo_root, changed_defs, changed_contents)
+        ]
+
+    def _is_related(
+        self,
+        candidate: Path,
+        module_sources: set[str],
+        tf_dirs: set[Path],
+        repo_root: Path | None,
+        changed_defs: set[str],
+        changed_contents: list[str],
+    ) -> bool:
+        if _is_in_module(candidate, module_sources, tf_dirs, repo_root):
+            return True
+        if candidate.parent not in tf_dirs:
+            return False
+        try:
+            content = candidate.read_text(encoding="utf-8")
+        except (OSError, UnicodeDecodeError):
+            return False
+        if _candidate_references_changed_defs_strict(content, changed_defs):
+            return True
+        candidate_defs = _extract_qualified_defs(content)
+        return bool(candidate_defs) and any(
+            _candidate_references_changed_defs_strict(c, candidate_defs) for c in changed_contents
+        )
 
     def build(self, fragments: list[Fragment], repo_root: Path | None = None) -> EdgeDict:
         tf_frags = [f for f in fragments if _is_terraform_file(f.path)]

src/treemapper/diffctx/edges/semantic/jvm.py

@@ -333,18 +333,24 @@ def _link_refs(
         class_to_frags: dict[str, list[FragmentId]],
         edges: EdgeDict,
     ) -> None:
+        self._link_inheritance_refs(jf, class_to_frags, edges)
+        self._link_type_refs(jf, class_to_frags, edges)
+        self._link_annotation_refs(jf, class_to_frags, edges)
+
+    def _link_inheritance_refs(self, jf: Fragment, class_to_frags: dict[str, list[FragmentId]], edges: EdgeDict) -> None:
         for inh_ref in _extract_inheritance(jf.content, jf.path):
             for fid in class_to_frags.get(inh_ref.lower(), []):
                 if fid != jf.id:
                     self.add_edge(edges, jf.id, fid, self.inheritance_weight)
 
+    def _link_type_refs(self, jf: Fragment, class_to_frags: dict[str, list[FragmentId]], edges: EdgeDict) -> None:
         for type_ref in _extract_type_refs(jf.content):
-            if type_ref in _JVM_STDLIB_TYPES:
-                continue
-            for fid in class_to_frags.get(type_ref.lower(), []):
-                if fid != jf.id:
-                    self.add_edge(edges, jf.id, fid, self.type_weight)
+            if type_ref not in _JVM_STDLIB_TYPES:
+                for fid in class_to_frags.get(type_ref.lower(), []):
+                    if fid != jf.id:
+                        self.add_edge(edges, jf.id, fid, self.type_weight)
 
+    def _link_annotation_refs(self, jf: Fragment, class_to_frags: dict[str, list[FragmentId]], edges: EdgeDict) -> None:
         for ann_ref in _extract_annotations(jf.content):
             for fid in class_to_frags.get(ann_ref.lower(), []):
                 if fid != jf.id:

src/treemapper/diffctx/graph.py

@@ -87,42 +87,48 @@ def build_graph(fragments: list[Fragment], repo_root: Path | None = None) -> Gra
 _HUB_OUT_DEGREE_THRESHOLD = 3
 
 
-def _apply_hub_suppression(
+def _suppress_high_in_degree(
     edges: dict[tuple[FragmentId, FragmentId], float],
     edge_categories: dict[tuple[FragmentId, FragmentId], str],
 ) -> dict[tuple[FragmentId, FragmentId], float]:
-    if not edges:
-        return edges
-
     in_degree: dict[FragmentId, int] = {}
-    for _src, dst in edges.keys():
+    for _src, dst in edges:
         in_degree[dst] = in_degree.get(dst, 0) + 1
 
-    if not in_degree:
-        return edges
-
     degrees = sorted(in_degree.values())
     mid = len(degrees) // 2
     d_median = (degrees[mid] + degrees[~mid]) / 2.0
 
     suppressed: dict[tuple[FragmentId, FragmentId], float] = {}
     for (src, dst), weight in edges.items():
         deg = in_degree.get(dst, 0)
-        cat = edge_categories.get((src, dst), "generic")
-        if deg > d_median and cat not in _SUPPRESSION_EXEMPT:
+        if deg > d_median and edge_categories.get((src, dst), "generic") not in _SUPPRESSION_EXEMPT:
             weight = weight / max(1.0, math.log(1 + deg))
         suppressed[(src, dst)] = weight
+    return suppressed
+
 
+def _suppress_semantic_hubs(
+    edges: dict[tuple[FragmentId, FragmentId], float],
+    edge_categories: dict[tuple[FragmentId, FragmentId], str],
+) -> dict[tuple[FragmentId, FragmentId], float]:
     sem_out_files: dict[FragmentId, set[Path]] = {}
     for (src, dst), cat in edge_categories.items():
         if cat == "semantic":
-            if src not in sem_out_files:
-                sem_out_files[src] = set()
-            sem_out_files[src].add(dst.path)
+            sem_out_files.setdefault(src, set()).add(dst.path)
 
-    for src, dst in list(suppressed.keys()):
+    for src, dst in list(edges):
         out_file_deg = len(sem_out_files.get(src, set()))
         if out_file_deg >= _HUB_OUT_DEGREE_THRESHOLD and edge_categories.get((src, dst)) == "semantic":
-            suppressed[(src, dst)] = suppressed[(src, dst)] / math.sqrt(out_file_deg)
+            edges[(src, dst)] = edges[(src, dst)] / math.sqrt(out_file_deg)
+    return edges
 
-    return suppressed
+
+def _apply_hub_suppression(
+    edges: dict[tuple[FragmentId, FragmentId], float],
+    edge_categories: dict[tuple[FragmentId, FragmentId], str],
+) -> dict[tuple[FragmentId, FragmentId], float]:
+    if not edges:
+        return edges
+    suppressed = _suppress_high_in_degree(edges, edge_categories)
+    return _suppress_semantic_hubs(suppressed, edge_categories)