Skip to content

Commit

Permalink
Move static check from common to isv
Browse files Browse the repository at this point in the history
The checks are usually skipped by support to it doesn't make sense to
have them in isv pipelines.

Signed-off-by: Ales Raszka <[email protected]>
  • Loading branch information
Allda committed May 2, 2024
1 parent dfd0823 commit f4d697a
Show file tree
Hide file tree
Showing 4 changed files with 270 additions and 274 deletions.
118 changes: 1 addition & 117 deletions operator-pipeline-images/operatorcert/static_tests/common/bundle.py
Original file line number Diff line number Diff line change
@@ -1,17 +1,9 @@
"""A common test suite for operator bundles"""

from collections.abc import Iterator
from typing import Any, List

from operator_repo import Bundle
from operator_repo.checks import CheckResult, Fail, Warn
from operatorcert import utils


class GraphLoopException(Exception):
"""
Exception raised when a loop is detected in the update graph
"""
from operator_repo.checks import CheckResult, Warn


def check_operator_name(bundle: Bundle) -> Iterator[CheckResult]:
Expand All @@ -31,111 +23,3 @@ def check_operator_name(bundle: Bundle) -> Iterator[CheckResult]:
"To fix this issue define the annotation in "
"'metadata/annotations.yaml' file that matches the CSV name."
)


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


def check_replaces_availability(bundle: Bundle) -> Iterator[CheckResult]:
"""
Check if the current bundle and the replaced bundle support the same OCP versions
Args:
bundle (Bundle): Operator bundle
Yields:
Iterator[CheckResult]: Failure if the version of the replaced bundle
does not match with the current bundle
"""

replaces = bundle.csv.get("spec", {}).get("replaces")
if not replaces:
return
delimiter = ".v" if ".v" in replaces else "."
replaces_version = replaces.split(delimiter, 1)[1]
replaces_bundle = bundle.operator.bundle(replaces_version)
ocp_versions_str = bundle.annotations.get("com.redhat.openshift.versions")
replaces_ocp_version_str = replaces_bundle.annotations.get(
"com.redhat.openshift.versions"
)
if ocp_versions_str == replaces_ocp_version_str:
# The annotations match, no need to check further
return
organization = bundle.operator.repo.config.get("organization")

indexes = set(utils.get_ocp_supported_versions(organization, ocp_versions_str))
replaces_indexes = set(
utils.get_ocp_supported_versions(organization, replaces_ocp_version_str)
)

if indexes - replaces_indexes == set():
# The replaces bundle supports all the same versions as the current bundle
return
yield Fail(
f"Replaces bundle {replaces_bundle} {sorted(replaces_indexes)} does not support "
f"the same OCP versions as bundle {bundle} {sorted(indexes)}. In order to fix this issue, "
"align the OCP version range to match the range of the replaced bundle. "
"This can be done by setting the `com.redhat.openshift.versions` annotation in the "
"`metadata/annotations.yaml` file.\n"
f"`{bundle}` - `{ocp_versions_str}`\n"
f"`{replaces_bundle}` - `{replaces_ocp_version_str}`"
)
yield from []
Original file line number Diff line number Diff line change
Expand Up @@ -11,12 +11,13 @@
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
from operator_repo.utils import lookup_dict
from semver import Version
from operatorcert import utils
from semver import Version

from .validations import (
validate_capabilities,
Expand Down Expand Up @@ -54,6 +55,12 @@
}


class GraphLoopException(Exception):
"""
Exception raised when a loop is detected in the update graph
"""


def _parse_semver(version: str) -> Version:
return Version.parse(version.strip(), optional_minor_and_patch=True).replace(
prerelease=None, build=None
Expand Down Expand Up @@ -288,3 +295,111 @@ def check_api_version_constraints(bundle: Bundle) -> Iterator[CheckResult]:
f"OCP version(s) {conflicting_str} conflict with "
f"minKubeVersion={k8s_version_min}"
)


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


def check_replaces_availability(bundle: Bundle) -> Iterator[CheckResult]:
"""
Check if the current bundle and the replaced bundle support the same OCP versions
Args:
bundle (Bundle): Operator bundle
Yields:
Iterator[CheckResult]: Failure if the version of the replaced bundle
does not match with the current bundle
"""

replaces = bundle.csv.get("spec", {}).get("replaces")
if not replaces:
return
delimiter = ".v" if ".v" in replaces else "."
replaces_version = replaces.split(delimiter, 1)[1]
replaces_bundle = bundle.operator.bundle(replaces_version)
ocp_versions_str = bundle.annotations.get("com.redhat.openshift.versions")
replaces_ocp_version_str = replaces_bundle.annotations.get(
"com.redhat.openshift.versions"
)
if ocp_versions_str == replaces_ocp_version_str:
# The annotations match, no need to check further
return
organization = bundle.operator.repo.config.get("organization")

indexes = set(utils.get_ocp_supported_versions(organization, ocp_versions_str))
replaces_indexes = set(
utils.get_ocp_supported_versions(organization, replaces_ocp_version_str)
)

if indexes - replaces_indexes == set():
# The replaces bundle supports all the same versions as the current bundle
return
yield Fail(
f"Replaces bundle {replaces_bundle} {sorted(replaces_indexes)} does not support "
f"the same OCP versions as bundle {bundle} {sorted(indexes)}. In order to fix this issue, "
"align the OCP version range to match the range of the replaced bundle. "
"This can be done by setting the `com.redhat.openshift.versions` annotation in the "
"`metadata/annotations.yaml` file.\n"
f"`{bundle}` - `{ocp_versions_str}`\n"
f"`{replaces_bundle}` - `{replaces_ocp_version_str}`"
)
yield from []
Loading

0 comments on commit f4d697a

Please sign in to comment.