[ISV-5618] Fix check_upgrade_graph_loop for community operators

operator-pipeline-images/operatorcert/static_tests/community/bundle.py

@@ -12,7 +12,6 @@
 import re
 import subprocess
 from collections.abc import Iterator
-from typing import Any, List
 
 from operator_repo import Bundle
 from operator_repo.checks import CheckResult, Fail, Warn
@@ -106,12 +105,6 @@ def ocp_to_k8s_ver(ocp_ver: str) -> str:
         return f"{k8s.major}.{k8s.minor}"
 
 
-class GraphLoopException(Exception):
-    """
-    Exception raised when a loop is detected in the update graph
-    """
-
-
 def run_operator_sdk_bundle_validate(
     bundle: Bundle, test_suite_selector: str
 ) -> Iterator[CheckResult]:
@@ -327,68 +320,6 @@ def check_api_version_constraints(bundle: Bundle) -> Iterator[CheckResult]:
         )
 
 
-@skip_fbc
-def check_upgrade_graph_loop(bundle: Bundle) -> Iterator[CheckResult]:
-    """
-    Detect loops in the upgrade graph
-
-    Example:
-
-    Channel beta: A -> B -> C -> B
-
-    Args:
-        bundle (Bundle): Operator bundle
-
-    Yields:
-        Iterator[CheckResult]: Failure if a loop is detected
-    """
-    all_channels: set[str] = set(bundle.channels)
-    if bundle.default_channel is not None:
-        all_channels.add(bundle.default_channel)
-    operator = bundle.operator
-    for channel in sorted(all_channels):
-        visited: List[Bundle] = []
-        try:
-            channel_bundles = operator.channel_bundles(channel)
-            try:
-                graph = operator.update_graph(channel)
-            except (NotImplementedError, ValueError) as exc:
-                yield Fail(str(exc))
-                return
-            follow_graph(graph, channel_bundles[0], visited)
-        except GraphLoopException as exc:
-            yield Fail(str(exc))
-
-
-def follow_graph(graph: Any, bundle: Bundle, visited: List[Bundle]) -> List[Bundle]:
-    """
-    Follow operator upgrade graph and raise exception if loop is detected
-
-    Args:
-        graph (Any): Operator update graph
-        bundle (Bundle): Current bundle that started the graph traversal
-        visited (List[Bundle]): List of bundles visited so far
-
-    Raises:
-        GraphLoopException: Graph loop detected
-
-    Returns:
-        List[Bundle]: List of bundles visited so far
-    """
-    if bundle in visited:
-        visited.append(bundle)
-        raise GraphLoopException(f"Upgrade graph loop detected for bundle: {visited}")
-    if bundle not in graph:
-        return visited
-
-    visited.append(bundle)
-    next_bundles = graph[bundle]
-    for next_bundle in next_bundles:
-        visited_copy = visited.copy()
-        follow_graph(graph, next_bundle, visited_copy)
-    return visited
-
-
 @skip_fbc
 def check_replaces_availability(bundle: Bundle) -> Iterator[CheckResult]:
     """

operator-pipeline-images/operatorcert/static_tests/community/operator.py

@@ -7,8 +7,9 @@
 """
 
 from collections.abc import Iterator
+from typing import Dict, Optional
 
-from operator_repo import Operator
+from operator_repo import Operator, Bundle
 from operator_repo.checks import CheckResult, Fail, Warn
 from operatorcert.static_tests.helpers import skip_fbc
 
@@ -36,3 +37,76 @@ def check_ci_upgrade_graph(operator: Operator) -> Iterator[CheckResult]:
             yield Fail(
                 f"The 'updateGraph' option in ci.yaml must be one of {allowed_graphs}"
             )
+
+
+@skip_fbc
+def check_upgrade_graph_loop(operator: Operator) -> Iterator[CheckResult]:
+    """
+    Detect loops in the upgrade graph
+
+    Example:
+
+    Channel beta: A -> B -> C -> B
+
+    Args:
+        operator (Operator): Operator
+
+    Yields:
+        Iterator[CheckResult]: Failure if a loop is detected
+    """
+    all_channels: set[str] = set()
+    for bundle in operator.all_bundles():
+        all_channels.update(bundle.channels)
+        if bundle.default_channel is not None:
+            all_channels.add(bundle.default_channel)
+    for channel in sorted(all_channels):
+        try:
+            graph = operator.update_graph(channel)
+        except (NotImplementedError, ValueError) as exc:
+            yield Fail(str(exc))
+            return
+        back_edge = has_cycle(graph)
+        if back_edge is not None:
+            node_from, node_to = back_edge
+            yield Fail(f"Upgrade graph loop detected: {node_from} -> {node_to}")
+
+
+def has_cycle(graph: Dict[Bundle, set[Bundle]]) -> Optional[tuple[Bundle, Bundle]]:
+    """
+    Detects cycles in an update graph using regular DFS
+
+    Args:
+        graph: Upgrade graph
+
+    Returns:
+        - None if no cycle is detected
+        - A tuple representing the edge (from_bundle, to_bundle) causing the cycle
+    """
+
+    def dfs(
+        node: Bundle, visited: set[Bundle], rec_stack: set[Bundle]
+    ) -> Optional[tuple[Bundle, Bundle]]:
+        visited.add(node)
+        rec_stack.add(node)
+
+        for neighbor in graph.get(node, set()):
+            if neighbor not in visited:
+                result = dfs(neighbor, visited, rec_stack)
+                if result:
+                    return result
+            elif neighbor in rec_stack:
+                return (node, neighbor)
+
+        rec_stack.remove(node)
+        return None
+
+    visited: set[Bundle] = set()
+    rec_stack: set[Bundle] = set()
+
+    for node in graph:
+        if node not in visited:
+            result = dfs(node, visited, rec_stack)
+            if result:
+                return result
+
+    return None

operator-pipeline-images/operatorcert/static_tests/helpers.py

@@ -31,8 +31,7 @@ def wrapper(*args: Any, **kwargs: Any) -> Iterator[Any]:
         config = operator.config if operator else {}
         if not config.get("fbc", {}).get("enabled", False):
             yield from func(*args, **kwargs)
-
-        if operator:
+        else:
             LOGGER.info(
                 "Skipping %s for FBC enabled operator %s",
                 func.__name__,

operator-pipeline-images/tests/static_tests/community/test_bundle.py

@@ -1,6 +1,6 @@
 import re
 from pathlib import Path
-from typing import Any, Dict, List, Optional, Set
+from typing import Any, Dict, Optional, Set
 from unittest.mock import MagicMock, patch, PropertyMock
 
 import pytest
@@ -14,7 +14,6 @@
     run_operator_sdk_bundle_validate,
     check_api_version_constraints,
     check_replaces_availability,
-    check_upgrade_graph_loop,
     check_using_fbc,
     ocp_to_k8s_ver,
 )
@@ -512,75 +511,6 @@ def test_check_api_version_constraints(
     assert set(check_api_version_constraints(bundle)) == expected
 
 
-def test_check_upgrade_graph_loop(tmp_path: Path) -> None:
-    create_files(
-        tmp_path,
-        bundle_files("hello", "0.0.1"),
-        bundle_files("hello", "0.0.2", csv={"spec": {"replaces": "hello.v0.0.1"}}),
-    )
-
-    repo = Repo(tmp_path)
-    operator = repo.operator("hello")
-    bundle = operator.bundle("0.0.1")
-    with patch.object(
-        type(operator), "config", new_callable=PropertyMock
-    ) as mock_config:
-        mock_config.return_value = {"updateGraph": "replaces-mode"}
-        is_loop = list(check_upgrade_graph_loop(bundle))
-        assert is_loop == []
-
-    with patch.object(
-        type(operator), "config", new_callable=PropertyMock
-    ) as mock_config:
-        mock_config.return_value = {"updateGraph": "unknown-mode"}
-        is_loop = list(check_upgrade_graph_loop(bundle))
-        assert is_loop == [
-            Fail("Operator(hello): unsupported updateGraph value: unknown-mode")
-        ]
-
-    # Both bundles replace each other
-    create_files(
-        tmp_path,
-        bundle_files("hello", "0.0.1", csv={"spec": {"replaces": "hello.v0.0.2"}}),
-        bundle_files("hello", "0.0.2", csv={"spec": {"replaces": "hello.v0.0.1"}}),
-    )
-
-    repo = Repo(tmp_path)
-    operator = repo.operator("hello")
-    bundle = operator.bundle("0.0.1")
-    with patch.object(
-        type(operator), "config", new_callable=PropertyMock
-    ) as mock_config:
-        mock_config.return_value = {"updateGraph": "replaces-mode"}
-        is_loop = list(check_upgrade_graph_loop(bundle))
-        assert len(is_loop) == 1 and isinstance(is_loop[0], Fail)
-        assert (
-            is_loop[0].reason
-            == "Upgrade graph loop detected for bundle: [Bundle(hello/0.0.1), "
-            "Bundle(hello/0.0.2), Bundle(hello/0.0.1)]"
-        )
-
-    # Malformed .spec.replaces
-    create_files(
-        tmp_path,
-        bundle_files("malformed", "0.0.1", csv={"spec": {"replaces": ""}}),
-    )
-
-    repo = Repo(tmp_path)
-    operator = repo.operator("malformed")
-    bundle = operator.bundle("0.0.1")
-    with patch.object(
-        type(operator), "config", new_callable=PropertyMock
-    ) as mock_config:
-        mock_config.return_value = {"updateGraph": "replaces-mode"}
-        failures = list(check_upgrade_graph_loop(bundle))
-        assert len(failures) == 1 and isinstance(failures[0], Fail)
-        assert (
-            "Bundle(malformed/0.0.1) has invalid 'replaces' field:"
-            in failures[0].reason
-        )
-
-
 def test_check_replaces_availability_no_replaces(
     tmp_path: Path,
 ) -> None:

operator-pipeline-images/tests/static_tests/community/test_operator.py

@@ -2,11 +2,13 @@
 from typing import Any
 
 import pytest
+from unittest.mock import patch, PropertyMock
 from operator_repo import Repo
 from operator_repo.checks import Fail, Warn
 from operatorcert.static_tests.community.operator import (
     check_ci_upgrade_graph,
     check_operator_name_unique,
+    check_upgrade_graph_loop,
 )
 from tests.utils import bundle_files, create_files
 
@@ -76,3 +78,72 @@ def test_check_ci_upgrade_graph(graph_mode: str, expected: Any, tmp_path: Path)
     result = check_ci_upgrade_graph(operator)
 
     assert set(result) == expected
+
+
+def test_check_upgrade_graph_loop(tmp_path: Path) -> None:
+    create_files(
+        tmp_path,
+        bundle_files("hello", "0.0.1"),
+        bundle_files("hello", "0.0.2", csv={"spec": {"replaces": "hello.v0.0.1"}}),
+    )
+
+    repo = Repo(tmp_path)
+    operator = repo.operator("hello")
+    bundle = operator.bundle("0.0.1")
+    with patch.object(
+        type(operator), "config", new_callable=PropertyMock
+    ) as mock_config:
+        mock_config.return_value = {"updateGraph": "replaces-mode"}
+        is_loop = list(check_upgrade_graph_loop(operator))
+        assert is_loop == []
+
+    with patch.object(
+        type(operator), "config", new_callable=PropertyMock
+    ) as mock_config:
+        mock_config.return_value = {"updateGraph": "unknown-mode"}
+        is_loop = list(check_upgrade_graph_loop(operator))
+        assert is_loop == [
+            Fail("Operator(hello): unsupported updateGraph value: unknown-mode")
+        ]
+
+    # Both bundles replace each other
+    create_files(
+        tmp_path,
+        bundle_files("hello", "0.0.1", csv={"spec": {"replaces": "hello.v0.0.2"}}),
+        bundle_files("hello", "0.0.2", csv={"spec": {"replaces": "hello.v0.0.1"}}),
+    )
+
+    repo = Repo(tmp_path)
+    operator = repo.operator("hello")
+    bundle = operator.bundle("0.0.1")
+    with patch.object(
+        type(operator), "config", new_callable=PropertyMock
+    ) as mock_config:
+        mock_config.return_value = {"updateGraph": "replaces-mode"}
+        is_loop = list(check_upgrade_graph_loop(operator))
+        assert len(is_loop) == 1 and isinstance(is_loop[0], Fail)
+        assert (
+            "Upgrade graph loop detected:" in is_loop[0].reason
+            and "Bundle(hello/0.0.1)" in is_loop[0].reason
+            and "Bundle(hello/0.0.2)" in is_loop[0].reason
+        )
+
+    # Malformed .spec.replaces
+    create_files(
+        tmp_path,
+        bundle_files("malformed", "0.0.1", csv={"spec": {"replaces": ""}}),
+    )
+
+    repo = Repo(tmp_path)
+    operator = repo.operator("malformed")
+    bundle = operator.bundle("0.0.1")
+    with patch.object(
+        type(operator), "config", new_callable=PropertyMock
+    ) as mock_config:
+        mock_config.return_value = {"updateGraph": "replaces-mode"}
+        failures = list(check_upgrade_graph_loop(operator))
+        assert len(failures) == 1 and isinstance(failures[0], Fail)
+        assert (
+            "Bundle(malformed/0.0.1) has invalid 'replaces' field:"
+            in failures[0].reason
+        )

operator-pipeline-images/tests/static_tests/test_helpers.py

@@ -46,13 +46,17 @@ def check_unknown(unknown: None) -> Iterator[str]:
         ]
     )
 
+    mock_logger.reset_mock()
     operator.config = {"fbc": {"enabled": False}}
     assert list(check_bundle(bundle)) == ["processed"]
     assert list(check_operator(operator)) == ["processed"]
+    mock_logger.assert_not_called()
 
+    mock_logger.reset_mock()
     operator.config = {}
     assert list(check_bundle(bundle)) == ["processed"]
     assert list(check_operator(operator)) == ["processed"]
+    mock_logger.assert_not_called()
 
     # if no operator provided wrapped func is executed
     assert list(check_unknown(None)) == ["processed"]