Compare commits
4 Commits
8d5fde5fcd
...
15eb937453
Author | SHA1 | Date |
---|---|---|
Maurizio Porrato | 15eb937453 | |
Maurizio Porrato | da06509790 | |
Maurizio Porrato | 71d2129cfb | |
Maurizio Porrato | b0f6bea925 |
|
@ -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:
|
||||
|
|
|
@ -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(
|
||||
|
|
|
@ -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))
|
||||
|
|
|
@ -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)
|
||||
|
||||
|
||||
|
|
|
@ -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:
|
||||
|
|
|
@ -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",
|
||||
|
|
Loading…
Reference in New Issue