1
0
Fork 0

Compare commits

...

4 Commits

6 changed files with 234 additions and 47 deletions

View File

@ -60,7 +60,8 @@ def get_checks(
def run_suite(
targets: Iterable[Operator | Bundle], suite_name: str = "operator_repo.checks"
targets: Iterable[Repo | Operator | Bundle],
suite_name: str = "operator_repo.checks",
) -> Iterable[CheckResult]:
checks = get_checks(suite_name)
for target in targets:

View File

@ -11,7 +11,9 @@ def check_operator_name(bundle: Bundle) -> Iterator[CheckResult]:
"""Check if the operator names used in CSV, metadata and filesystem are consistent"""
name = bundle.annotations.get("operators.operatorframework.io.bundle.package.v1")
if name is None:
yield Fail(f"Bundle does not define the operator name in annotations.yaml")
yield Fail(
bundle, f"Bundle does not define the operator name in annotations.yaml"
)
return
if name != bundle.csv_operator_name:
yield Fail(

View File

@ -18,7 +18,7 @@ def check_upgrade(operator: Operator) -> Iterator[CheckResult]:
if dangling_bundles:
yield Fail(
operator,
f"Channel {channel} has dangling bundles: {dangling_bundles}.",
f"Channel {channel} has dangling bundles: {dangling_bundles}",
)
except Exception as exc:
yield Fail(operator, str(exc))

View File

@ -25,48 +25,59 @@ def indent(depth: int) -> str:
return " " * depth
def _list(
target: Union[Repo, Operator, Bundle], recursive: bool = False, depth: int = 0
) -> None:
def show_repo(repo: Repo, recursive: bool = False, depth: int = 0) -> None:
print(indent(depth) + str(repo))
for operator in repo:
if recursive:
show_operator(operator, recursive, depth + 1)
else:
print(indent(depth + 1) + str(operator))
def show_operator(operator: Operator, recursive: bool = False, depth: int = 0) -> None:
print(indent(depth) + str(operator))
for bundle in operator:
if recursive:
show_bundle(bundle, recursive, depth + 1)
else:
print(indent(depth + 1) + str(bundle))
def show_bundle(bundle: Bundle, recursive: bool = False, depth: int = 0) -> None:
print(indent(depth) + str(bundle))
csv_annotations = bundle.csv.get("metadata", {}).get("annotations", {})
info = [
("Description", csv_annotations.get("description", "")),
("Name", f"{bundle.csv_operator_name}.v{bundle.csv_operator_version}"),
("Channels", ", ".join(bundle.channels)),
("Default channel", bundle.default_channel),
("Container image", csv_annotations.get("containerImage", "")),
("Replaces", bundle.csv.get("spec", {}).get("replaces", "")),
("Skips", bundle.csv.get("spec", {}).get("skips", [])),
]
max_width = max(len(key) for key, _ in info)
for key, value in info:
message = f"{key.ljust(max_width + 1)}: {value}"
print(indent(depth + 1) + message)
def show(target: Union[Repo, Operator, Bundle], recursive: bool = False) -> None:
if isinstance(target, Repo):
print(indent(depth) + str(target))
for operator in target:
if recursive:
_list(operator, True, depth + 1)
else:
print(indent(depth + 1) + str(operator))
show_repo(target, recursive, 0)
elif isinstance(target, Operator):
print(indent(depth) + str(target))
for bundle in target:
if recursive:
_list(bundle, True, depth + 1)
else:
print(indent(depth + 1) + str(bundle))
show_operator(target, recursive, 1 * recursive)
elif isinstance(target, Bundle):
print(indent(depth) + str(target))
csv_annotations = target.csv.get("metadata", {}).get("annotations", {})
info = [
("Description", csv_annotations.get("description", "")),
("Name", f"{target.csv_operator_name}.{target.csv_operator_version}"),
("Channels", ", ".join(target.channels)),
("Default channel", target.default_channel),
("Container image", csv_annotations.get("containerImage", "")),
("Replaces", target.csv.get("spec", {}).get("replaces", "")),
("Skips", target.csv.get("spec", {}).get("skips", [])),
]
max_width = max(len(key) for key, _ in info)
for key, value in info:
message = f"{key.ljust(max_width+1)}: {value}"
print(indent(depth + 1) + message)
show_bundle(target, recursive, 2 * recursive)
def action_list(repo_path, *what: str, recursive: bool = False) -> None:
repo = Repo(repo_path)
if not what:
_list(repo, recursive)
if what:
targets = (parse_target(repo, x) for x in what)
else:
for target in what:
_list(parse_target(repo, target), recursive)
targets = [repo]
for target in targets:
show(target, recursive)
def _walk(
@ -84,14 +95,15 @@ def action_check(
repo_path: Path, suite: str, *what: str, recursive: bool = False
) -> None:
repo = Repo(repo_path)
if recursive:
if what:
targets = chain(_walk(parse_target(repo, x)) for x in what)
else:
targets = chain(_walk(x) for x in repo)
if what:
targets = [parse_target(repo, x) for x in what]
else:
targets = [parse_target(repo, x) for x in what] or repo.all_operators()
for result in run_suite(targets, suite_name=suite):
targets = repo.all_operators()
if recursive:
all_targets = chain.from_iterable(_walk(x) for x in targets)
else:
all_targets = targets
for result in run_suite(all_targets, suite_name=suite):
print(result)

View File

@ -16,8 +16,28 @@ log = logging.getLogger(__name__)
def _find_yaml(path: Path) -> Path:
"""Look for yaml files with alternate extensions"""
"""
Find a YAML file by looking for files with alternate extensions.
This function searches for a YAML file by trying multiple extensions (.yaml and .yml)
and checking if the file exists. It is used to locate YAML files with different possible
extensions in order to provide flexibility when specifying file paths.
Args:
path (Path): The path to the file with or without a YAML extension.
Returns:
Path: The path to the found YAML file.
Raises:
FileNotFoundError: If a YAML file with any of the tried extensions cannot be found.
Example:
file_path = Path("my_file.json")
yaml_path = _find_yaml(file_path)
# If "my_file.yaml" or "my_file.yml" exists, yaml_path will point to the found YAML file.
# Otherwise, a FileNotFoundError will be raised.
"""
if path.is_file():
return path
tries = [path]
@ -33,8 +53,28 @@ def _find_yaml(path: Path) -> Path:
def _load_yaml_strict(path: Path) -> Any:
"""Returns the parsed contents of the YAML file at the given path"""
"""
Load and parse the contents of a YAML file at the given path.
This function reads the contents of the specified YAML file and attempts to parse it
using the `yaml.safe_load` method. If the YAML document contains multiple documents or
if it's not a valid YAML document, exceptions are raised accordingly.
Args:
path (Path): The path to the YAML file to be loaded.
Returns:
Any: The parsed contents of the YAML file.
Raises:
OperatorRepoException: If the YAML file contains multiple documents.
OperatorRepoException: If the YAML file is not a valid YAML document.
Example:
yaml_path = Path("my_file.yaml")
yaml_content = _load_yaml_strict(yaml_path)
# The parsed contents of the YAML file will be stored in the `yaml_content` variable.
"""
log.debug("Loading %s", path)
with path.open("r") as yaml_file:
try:
@ -48,13 +88,58 @@ def _load_yaml_strict(path: Path) -> Any:
def load_yaml(path: Path) -> Any:
"""Same as _load_yaml_strict but tries both .yaml and .yml extensions"""
"""
Load and parse the contents of a YAML file at the given path with alternate extensions.
Args:
path (Path): The path to the file with or without a YAML extension.
Returns:
Any: The parsed contents of the YAML file.
Raises:
OperatorRepoException: If the YAML file contains multiple documents.
OperatorRepoException: If the YAML file is not a valid YAML document.
Example:
file_path = Path("my_file.json")
yaml_content = load_yaml(file_path)
# If "my_file.yaml" or "my_file.yml" exists, the parsed contents of the YAML file
# will be stored in the `yaml_content` variable.
"""
return _load_yaml_strict(_find_yaml(path))
def lookup_dict(
data: dict[str, Any], path: str, default: Any = None, separator: str = "."
) -> Any:
"""
Retrieve a value from a nested dictionary using a specific path of keys.
This function allows you to access a value in a nested dictionary by providing a
dot-separated path of keys. If the path exists in the dictionary, the corresponding
value is returned. If the path does not exist, the specified default value is returned.
Args:
data (dict): The nested dictionary from which to retrieve the value.
path (str): A dot-separated string representing the path to the desired value.
default (Any, optional): The value to return if the path does not exist. Defaults to None.
separator (str, optional): The separator used to split the path into keys. Defaults to ".".
Returns:
Any: The value at the specified path if found, otherwise the default value.
Example:
data = {
"a": {
"b": {
"c": 42
}
}
}
value = lookup_dict(data, "a.b.c")
# value will be 42
"""
keys = path.split(separator)
subtree = data
for key in keys:

View File

@ -4,6 +4,28 @@ import yaml
def merge(a, b, path=None):
"""
Recursively merge two dictionaries, with values from the second dictionary (b)
overwriting corresponding values in the first dictionary (a). This function can
handle nested dictionaries.
Args:
a (dict): The base dictionary to merge into.
b (dict): The dictionary with values to merge into the base dictionary.
path (list of str, optional): A list representing the current path in the recursive merge.
This argument is used internally for handling nested dictionaries. Users typically
don't need to provide this argument. Defaults to None.
Returns:
dict: The merged dictionary (a) after incorporating values from the second dictionary (b).
Example:
dict_a = {"key1": 42, "key2": {"subkey1": "value1"}}
dict_b = {"key2": {"subkey2": "value2"}, "key3": "value3"}
merged_dict = merge(dict_a, dict_b)
# merged_dict will be:
# {"key1": 42, "key2": {"subkey1": "value1", "subkey2": "value2"}, "key3": "value3"}
"""
if path is None:
path = []
for key in b:
@ -18,6 +40,37 @@ def merge(a, b, path=None):
def create_files(path, *contents):
"""
Create files and directories at the specified path based on the provided content.
This function allows you to create files and directories at a specified path.
The function accepts a variable number of content dictionaries, where each
dictionary represents a file or directory to be created. The keys in the dictionary
represent the filenames or directory names, and the values represent the content
of the files or subdirectories.
Args:
path (str): The path where the files and directories should be created.
*contents (dict): Variable number of dictionaries representing files and directories.
For files, the dictionary should have a single key-value pair where the key is
the filename and the value is the content of the file. For directories, the
dictionary should have a single key with a value of None.
Returns:
None
Example:
create_files(
"/my_folder",
{"file1.txt": "Hello, World!"},
{"subfolder": None},
{"config.yaml": {"key": "value"}},
)
In this example, the function will create a file "file1.txt" with content "Hello, World!"
in the "/my_folder" directory, create an empty subdirectory "subfolder", and create a
file "config.yaml" with the specified YAML content.
"""
root = Path(path)
for element in contents:
for file_name, content in element.items():
@ -35,6 +88,40 @@ def create_files(path, *contents):
def bundle_files(
operator_name, bundle_version, annotations=None, csv=None, other_files=None
):
"""
Create a bundle of files and metadata for an Operator package.
This function generates a bundle of files and metadata for an Operator package,
including annotations, a CSV (ClusterServiceVersion) file, and other additional files.
Args:
operator_name (str): The name of the Operator.
bundle_version (str): The version of the Operator bundle.
annotations (dict, optional): Additional annotations for the bundle annotations.yaml file.
Defaults to None.
csv (dict, optional): Additional content to merge with the base CSV (ClusterServiceVersion)
file. Defaults to None.
other_files (dict, optional): Additional files to be included in the bundle.
Defaults to None.
Returns:
dict: A dictionary representing the bundle, including annotations.yaml and CSV files,
and other additional files.
Example:
bundle = bundle_files(
operator_name="my-operator",
bundle_version="1.0.0",
annotations={"custom.annotation": "value"},
csv={"spec": {"installModes": ["AllNamespaces"]}},
other_files={"README.md": "# My Operator Documentation"}
)
In this example, the function will create a dictionary representing a bundle for the Operator
"my-operator" with version "1.0.0". The annotations.yaml file will contain the provided custom
annotation, the CSV file will have additional installation modes, and the README.md file will
be included as an additional file in the bundle.
"""
bundle_path = f"operators/{operator_name}/{bundle_version}"
base_annotations = {
"operators.operatorframework.io.bundle.mediatype.v1": "registry+v1",