[ISV-4971] Skip subset of static checks for FBC operators (#672)

Author: mantomas

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

@@ -17,6 +17,7 @@
 from operator_repo.checks import CheckResult, Fail, Warn
 from operator_repo.utils import lookup_dict
 from operatorcert import utils
+from operatorcert.static_tests.helpers import skip_fbc
 from semver import Version
 
 from .validations import (
@@ -179,6 +180,7 @@ def check_required_fields(bundle: Bundle) -> Iterator[CheckResult]:
         yield Fail(message) if fatal else Warn(message)
 
 
+@skip_fbc
 def check_dangling_bundles(bundle: Bundle) -> Iterator[CheckResult]:
     """
     Check dangling bundles in the operator update graph
@@ -216,6 +218,7 @@ def check_dangling_bundles(bundle: Bundle) -> Iterator[CheckResult]:
             yield Fail(f"Channel {channel} has dangling bundles: {dangling_bundles}")
 
 
+@skip_fbc
 def check_api_version_constraints(bundle: Bundle) -> Iterator[CheckResult]:
     """Check that the ocp and k8s api version constraints are consistent"""
     ocp_versions_str = bundle.annotations.get("com.redhat.openshift.versions")
@@ -297,6 +300,7 @@ 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
@@ -358,6 +362,7 @@ def follow_graph(graph: Any, bundle: Bundle, visited: List[Bundle]) -> List[Bund
     return visited
 
 
+@skip_fbc
 def check_replaces_availability(bundle: Bundle) -> Iterator[CheckResult]:
     """
     Check if the current bundle and the replaced bundle support the same OCP versions

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

@@ -10,8 +10,10 @@
 
 from operator_repo import Operator
 from operator_repo.checks import CheckResult, Fail, Warn
+from operatorcert.static_tests.helpers import skip_fbc
 
 
+@skip_fbc
 def check_operator_name_unique(operator: Operator) -> Iterator[CheckResult]:
     """Ensure all operator's bundles use the same operator name in their CSV"""
     names = {bundle.csv_operator_name for bundle in operator}

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

@@ -0,0 +1,40 @@
+""" Static tests helper utilities. """
+
+import logging
+
+from functools import wraps
+from typing import Any, Callable, Iterator
+from operator_repo import Operator, Bundle
+
+
+LOGGER = logging.getLogger("operator-cert")
+
+
+def skip_fbc(func: Callable[..., Any]) -> Callable[..., Any]:
+    """
+    Decorator to skip a static check for FBC enabled operators.
+
+    First argument of the decorated function should be either an Operator or a Bundle,
+    otherwise the 'check' will be executed as usual.
+    """
+
+    @wraps(func)
+    def wrapper(*args: Any, **kwargs: Any) -> Iterator[Any]:
+        first_arg = args[0]
+        if isinstance(first_arg, Bundle):
+            operator = first_arg.operator
+        elif isinstance(first_arg, Operator):
+            operator = first_arg
+
+        config = operator.config if operator else {}
+        if not config.get("fbc", {}).get("enabled", False):
+            yield from func(*args, **kwargs)
+
+        LOGGER.info(
+            "Skipping %s for FBC enabled operator %s",
+            func.__name__,
+            operator.operator_name,
+        )
+        yield from []
+
+    return wrapper

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

@@ -4,6 +4,7 @@
 
 from operator_repo import Bundle
 from operator_repo.checks import CheckResult, Fail, Warn
+from operatorcert.static_tests.helpers import skip_fbc
 
 PRUNED_GRAPH_ERROR = (
     "olm.skipRange annotation is set but replaces field is not set in the CSV. "
@@ -15,6 +16,7 @@
 )
 
 
+@skip_fbc
 def check_pruned_graph(bundle: Bundle) -> Iterator[CheckResult]:
     """
     Check if the update graph is pruned when the bundle is added to the index

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

@@ -1,7 +1,7 @@
 import re
 from pathlib import Path
 from typing import Any, Dict, List, Optional, Set
-from unittest.mock import MagicMock, patch
+from unittest.mock import MagicMock, patch, PropertyMock
 
 import pytest
 from operator_repo import Repo
@@ -291,9 +291,7 @@ def test_metadata_container_image_validation(
     assert expected_successes.intersection(collected_results.keys()) == set()
 
 
-@patch("operator_repo.core.Operator.config")
-def test_check_dangling_bundles(mock_config: MagicMock, tmp_path: Path) -> None:
-    mock_config.get.return_value = "replaces-mode"
+def test_check_dangling_bundles(tmp_path: Path) -> None:
     create_files(
         tmp_path,
         bundle_files("hello", "0.0.1"),
@@ -303,16 +301,23 @@ def test_check_dangling_bundles(mock_config: MagicMock, tmp_path: Path) -> None:
     repo = Repo(tmp_path)
     operator = repo.operator("hello")
     bundle3 = operator.bundle("0.0.3")
-    failures = list(check_dangling_bundles(bundle3))
-    assert failures == []
 
-    mock_config.get.return_value = "unknown-mode"
-    is_loop = list(check_dangling_bundles(bundle3))
-    assert is_loop == [
-        Fail("Operator(hello): unsupported updateGraph value: unknown-mode")
-    ]
+    with patch.object(
+        type(operator), "config", new_callable=PropertyMock
+    ) as mock_config:
+        mock_config.return_value = {"updateGraph": "replaces-mode"}
+        failures = list(check_dangling_bundles(bundle3))
+        assert failures == []
+
+    with patch.object(
+        type(operator), "config", new_callable=PropertyMock
+    ) as mock_config:
+        mock_config.return_value = {"updateGraph": "unknown-mode"}
+        is_loop = list(check_dangling_bundles(bundle3))
+        assert is_loop == [
+            Fail("Operator(hello): unsupported updateGraph value: unknown-mode")
+        ]
 
-    mock_config.get.return_value = "replaces-mode"
     # Bundle 0.0.2 is not referenced by any bundle and it is not a HEAD of channel
     create_files(
         tmp_path,
@@ -323,11 +328,16 @@ def test_check_dangling_bundles(mock_config: MagicMock, tmp_path: Path) -> None:
     repo = Repo(tmp_path)
     operator = repo.operator("hello")
     bundle3 = operator.bundle("0.0.3")
-    failures = list(check_dangling_bundles(bundle3))
-    assert len(failures) == 1 and isinstance(failures[0], Fail)
-    assert (
-        failures[0].reason == "Channel beta has dangling bundles: {Bundle(hello/0.0.2)}"
-    )
+    with patch.object(
+        type(operator), "config", new_callable=PropertyMock
+    ) as mock_config:
+        mock_config.return_value = {"updateGraph": "replaces-mode"}
+        failures = list(check_dangling_bundles(bundle3))
+        assert len(failures) == 1 and isinstance(failures[0], Fail)
+        assert (
+            failures[0].reason
+            == "Channel beta has dangling bundles: {Bundle(hello/0.0.2)}"
+        )
 
     # Malformed .spec.replaces
     create_files(
@@ -338,9 +348,16 @@ def test_check_dangling_bundles(mock_config: MagicMock, tmp_path: Path) -> None:
     repo = Repo(tmp_path)
     operator = repo.operator("malformed")
     bundle1 = operator.bundle("0.0.1")
-    failures = list(check_dangling_bundles(bundle1))
-    assert len(failures) == 1 and isinstance(failures[0], Fail)
-    assert "Bundle(malformed/0.0.1) has invalid 'replaces' field:" in failures[0].reason
+    with patch.object(
+        type(operator), "config", new_callable=PropertyMock
+    ) as mock_config:
+        mock_config.return_value = {"updateGraph": "replaces-mode"}
+        failures = list(check_dangling_bundles(bundle1))
+        assert len(failures) == 1 and isinstance(failures[0], Fail)
+        assert (
+            "Bundle(malformed/0.0.1) has invalid 'replaces' field:"
+            in failures[0].reason
+        )
 
 
 @pytest.mark.parametrize(
@@ -444,9 +461,7 @@ def test_check_api_version_constraints(
     assert set(check_api_version_constraints(bundle)) == expected
 
 
-@patch("operator_repo.core.Operator.config")
-def test_check_upgrade_graph_loop(mock_config: MagicMock, tmp_path: Path) -> None:
-    mock_config.get.return_value = "replaces-mode"
+def test_check_upgrade_graph_loop(tmp_path: Path) -> None:
     create_files(
         tmp_path,
         bundle_files("hello", "0.0.1"),
@@ -456,16 +471,22 @@ def test_check_upgrade_graph_loop(mock_config: MagicMock, tmp_path: Path) -> Non
     repo = Repo(tmp_path)
     operator = repo.operator("hello")
     bundle = operator.bundle("0.0.1")
-    is_loop = list(check_upgrade_graph_loop(bundle))
-    assert is_loop == []
-
-    mock_config.get.return_value = "unknown-mode"
-    is_loop = list(check_upgrade_graph_loop(bundle))
-    assert is_loop == [
-        Fail("Operator(hello): unsupported updateGraph value: unknown-mode")
-    ]
+    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")
+        ]
 
-    mock_config.get.return_value = "replaces-mode"
     # Both bundles replace each other
     create_files(
         tmp_path,
@@ -476,13 +497,17 @@ def test_check_upgrade_graph_loop(mock_config: MagicMock, tmp_path: Path) -> Non
     repo = Repo(tmp_path)
     operator = repo.operator("hello")
     bundle = operator.bundle("0.0.1")
-    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)]"
-    )
+    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(
@@ -493,9 +518,16 @@ def test_check_upgrade_graph_loop(mock_config: MagicMock, tmp_path: Path) -> Non
     repo = Repo(tmp_path)
     operator = repo.operator("malformed")
     bundle = operator.bundle("0.0.1")
-    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
+    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(

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

@@ -0,0 +1,51 @@
+from typing import Any, Iterator
+
+from operatorcert.static_tests.helpers import skip_fbc
+from operator_repo import Operator, Bundle
+from unittest.mock import call, MagicMock, patch
+
+
+@patch("operatorcert.static_tests.helpers.LOGGER")
+@patch.object(Operator, "probe", return_value=True)
+@patch.object(Bundle, "probe", return_value=True)
+def test_skip__fbc(
+    mock_bundle: MagicMock, mock_operator: MagicMock, mock_logger: MagicMock
+) -> None:
+
+    @skip_fbc
+    def check_bundle(bundle: Bundle) -> Iterator[str]:
+        yield "processed"
+
+    @skip_fbc
+    def check_operator(operator: Operator) -> Iterator[str]:
+        yield "processed"
+
+    operator = Operator("test-operator")
+    operator.config = {"fbc": {"enabled": True}}
+    bundle = Bundle("test-bundle", operator)
+
+    assert list(check_bundle(bundle)) == []
+    assert list(check_operator(operator)) == []
+
+    mock_logger.assert_has_calls(
+        [
+            call.info(
+                "Skipping %s for FBC enabled operator %s",
+                "check_bundle",
+                "test-operator",
+            ),
+            call.info(
+                "Skipping %s for FBC enabled operator %s",
+                "check_operator",
+                "test-operator",
+            ),
+        ]
+    )
+
+    operator.config = {"fbc": {"enabled": False}}
+    assert list(check_bundle(bundle)) == ["processed"]
+    assert list(check_operator(operator)) == ["processed"]
+
+    operator.config = {}
+    assert list(check_bundle(bundle)) == ["processed"]
+    assert list(check_operator(operator)) == ["processed"]