WIP Initial implementation of find_cycle_breakers

(Needs more tests.)

Author: seddonym

pyproject.toml

@@ -16,6 +16,7 @@ authors = [
 ]
 requires-python = ">=3.9"
 dependencies = [
+    "igraph>=0.11.9",
     "typing-extensions>=3.10.0.0",
 ]
 classifiers = [

src/grimp/application/graph.py

@@ -1,8 +1,9 @@
 from __future__ import annotations
+import itertools
 from typing import List, Optional, Sequence, Set, Tuple, TypedDict
 from grimp.domain.analysis import PackageDependency, Route
 from grimp.domain.valueobjects import Layer
-
+import igraph as ig  # type: ignore
 from grimp import _rustgrimp as rust  # type: ignore[attr-defined]
 from grimp.exceptions import (
     ModuleNotPresent,
@@ -444,8 +445,52 @@ def find_cycle_breakers(self, package: str) -> list[Import]:
         """
         Identify a set of imports that, if removed, would make the package locally acyclic.
         """
-        # TODO
-        return []
+        children = self.find_children(package)
+        if len(children) < 2:
+            return []
+        igraph = ig.Graph(directed=True)
+        igraph.add_vertices([package, *children])
+        edges: list[tuple[str, str]] = []
+        weights: list[int] = []
+        for downstream, upstream in itertools.permutations(children):
+            total_imports = 0
+            for expression in (
+                f"{downstream} -> {upstream}",
+                f"{downstream}.** -> {upstream}",
+                f"{downstream} -> {upstream}.**",
+                f"{downstream}.** -> {upstream}.**",
+            ):
+                total_imports += len(self.find_matching_direct_imports(expression))
+            if total_imports:
+                edges.append((downstream, upstream))
+                weights.append(total_imports)
+
+        igraph.add_edges(edges)
+        igraph.es["weight"] = weights
+
+        arc_set = igraph.feedback_arc_set(weights="weight")
+
+        squashed_imports: list[Import] = []
+        for edge_id in arc_set:
+            edge = igraph.es[edge_id]
+            squashed_imports.append(
+                {
+                    "importer": edge.source_vertex["name"],
+                    "imported": edge.target_vertex["name"],
+                }
+            )
+
+        unsquashed_imports: list[Import] = []
+        for squashed_import in squashed_imports:
+            for pattern in (
+                f"{squashed_import['importer']} -> {squashed_import['imported']}",
+                f"{squashed_import['importer']}.** -> {squashed_import['imported']}",
+                f"{squashed_import['importer']} -> {squashed_import['imported']}.**",
+                f"{squashed_import['importer']}.** -> {squashed_import['imported']}.**",
+            ):
+                unsquashed_imports.extend(self.find_matching_direct_imports(pattern))
+
+        return unsquashed_imports
 
     # Dunder methods
     # --------------

tests/unit/application/graph/test_cycle_breakers.py

@@ -1,11 +1,83 @@
 from grimp.application.graph import ImportGraph
+import pytest
 
 
 class TestFindCycleBreakers:
     def test_empty_graph(self):
         graph = ImportGraph()
-        graph.add_module("mypackage")
+        graph.add_module("pkg")
 
-        result = graph.find_cycle_breakers("mypackage")
+        result = graph.find_cycle_breakers("pkg")
 
         assert result == []
+
+    @pytest.mark.parametrize(
+        "module",
+        (
+            "pkg",
+            "pkg.foo",
+            "pkg.foo.blue",
+        ),
+    )
+    def test_graph_with_no_imports(self, module: str):
+        graph = self._build_graph_with_no_imports()
+
+        result = graph.find_cycle_breakers(module)
+
+        assert result == []
+
+    @pytest.mark.parametrize(
+        "module",
+        (
+            "pkg",
+            "pkg.foo",
+            "pkg.foo.blue",
+        ),
+    )
+    def test_acyclic_graph(self, module: str):
+        graph = self._build_acyclic_graph()
+
+        result = graph.find_cycle_breakers(module)
+
+        assert result == []
+
+    def test_one_breaker(self):
+        graph = self._build_acyclic_graph()
+        importer, imported = "pkg.bar.red.four", "pkg.foo.blue.two"
+        graph.add_import(importer=importer, imported=imported)
+        result = graph.find_cycle_breakers("pkg")
+
+        assert result == [
+            {
+                "importer": importer,
+                "imported": imported,
+            }
+        ]
+
+    def _build_graph_with_no_imports(self) -> ImportGraph:
+        graph = ImportGraph()
+        for module in (
+            "pkg",
+            "pkg.foo",
+            "pkg.foo.blue",
+            "pkg.foo.blue.one",
+            "pkg.foo.blue.two",
+            "pkg.foo.green",
+            "pkg.bar",
+            "pkg.bar.red",
+            "pkg.bar.red.three",
+            "pkg.bar.red.four",
+            "pkg.bar.yellow",
+        ):
+            graph.add_module(module)
+        return graph
+
+    def _build_acyclic_graph(self) -> ImportGraph:
+        graph = self._build_graph_with_no_imports()
+        for importer, imported in (
+            ("pkg.foo", "pkg.bar.red"),
+            ("pkg.foo.green", "pkg.bar.yellow"),
+            ("pkg.foo.blue.two", "pkg.bar.red.three"),
+        ):
+            graph.add_import(importer=importer, imported=imported)
+        return graph