1
0
Fork 0

Compare commits

...

2 Commits

Author SHA1 Message Date
Maurizio Porrato 885e63565f cli: fail cleanly if the given directory is not a valid operator repo 2023-08-17 12:09:56 +01:00
Maurizio Porrato a99949c386 Add typing to utility test functions 2023-08-17 11:57:00 +01:00
2 changed files with 24 additions and 8 deletions

View File

@ -5,13 +5,15 @@
import argparse
import logging
import sys
from collections.abc import Iterator
from itertools import chain
from pathlib import Path
from typing import Union
from typing import Union, Optional
from .checks import get_checks, run_suite
from .classes import Bundle, Operator, Repo
from .exceptions import OperatorRepoException
def parse_target(repo: Repo, target: str) -> Union[Operator, Bundle]:
@ -115,6 +117,16 @@ def action_check_list(suite: str) -> None:
print(f" - {display_name}: {check.__doc__}")
def _get_repo(path: Optional[Path]) -> Repo:
if not path:
path = Path.cwd()
try:
return Repo(path)
except OperatorRepoException:
print(f"{path} is not a valid operator repository")
sys.exit(1)
def main() -> None:
main_parser = argparse.ArgumentParser(
description="Operator repository manipulation tool",
@ -161,7 +173,6 @@ def main() -> None:
)
args = main_parser.parse_args()
# print(args)
verbosity = {0: logging.ERROR, 1: logging.WARNING, 2: logging.INFO}
log = logging.getLogger(__package__)
@ -173,13 +184,13 @@ def main() -> None:
log.addHandler(handler)
if args.action in ("list", "ls"):
action_list(args.repo or Path.cwd(), *args.target, recursive=args.recursive)
action_list(_get_repo(args.repo), *args.target, recursive=args.recursive)
elif args.action == "check":
if args.list:
action_check_list(args.suite)
else:
action_check(
args.repo or Path.cwd(),
_get_repo(args.repo),
args.suite,
*args.target,
recursive=args.recursive,

View File

@ -1,9 +1,10 @@
from pathlib import Path
from typing import Optional, Union
import yaml
def merge(a, b, path=None):
def merge(a: dict, b: dict, path: Optional[list[str]] = None) -> dict:
"""
Recursively merge two dictionaries, with values from the second dictionary (b)
overwriting corresponding values in the first dictionary (a). This function can
@ -39,7 +40,7 @@ def merge(a, b, path=None):
return a
def create_files(path, *contents):
def create_files(path: Union[str, Path], *contents: dict) -> None:
"""
Create files and directories at the specified path based on the provided content.
@ -86,8 +87,12 @@ def create_files(path, *contents):
def bundle_files(
operator_name, bundle_version, annotations=None, csv=None, other_files=None
):
operator_name: str,
bundle_version: str,
annotations: dict = None,
csv: dict = None,
other_files: dict = None,
) -> dict:
"""
Create a bundle of files and metadata for an Operator package.