From 998dfb9d1bdf3233ad0268ff4d0b0da0cee59ff4 Mon Sep 17 00:00:00 2001 From: Kamforka Date: Sat, 30 Sep 2023 14:59:32 +0200 Subject: [PATCH] #296 - refactor ci and cd scripts --- scripts/cd.py | 137 +++++++++++++++++++++++++------------------------- scripts/ci.py | 35 +++++-------- 2 files changed, 82 insertions(+), 90 deletions(-) diff --git a/scripts/cd.py b/scripts/cd.py index 1f91f9a3..039313a7 100755 --- a/scripts/cd.py +++ b/scripts/cd.py @@ -1,103 +1,104 @@ #!/usr/bin/env python import argparse import subprocess +from typing import List def _run_subprocess( - args: str, - init_message: str, - success_message: str, - error_message: str, - verbose=False, + command: str, + quiet=False, ): - print(init_message) - proc = subprocess.run(args, shell=True, capture_output=True) - - process_output = proc.stdout.decode() or proc.stderr.decode() - indented_process_output = "\n".join( - [f"\t{output_line}" for output_line in process_output.splitlines()] - ) - - if proc.returncode != 0: - exit_message = "\n".join([error_message, indented_process_output]) - exit(exit_message) - - if verbose: - print(indented_process_output) - - print(success_message) + if not quiet: + stdout = stderr = None + else: + stdout = stderr = subprocess.DEVNULL + + try: + subprocess.run(str.split(command), stdout=stdout, stderr=stderr, check=True) + except subprocess.CalledProcessError as err: + error_output = ( + f"ERROR: Execution of command '{command}' returned: {err.returncode}\n" + ) + print(error_output) + exit(err.returncode) -def run_all(verbose=False): +def run_all(quiet=False): print("Run all deployment tasks...") - run_build(verbose=verbose) - run_publish(verbose=verbose) + run_build(quiet=quiet) + run_upload(quiet=quiet) print("All tasks succeeded!") -def run_build(verbose: bool): +def run_build(quiet: bool): + print("Building thehive4py with the build module...") + _run_subprocess( + command="rm -rf build/ dist/", + quiet=quiet, + ) _run_subprocess( - args=("rm -rf build/ dist/ && python -m build --sdist --wheel"), - init_message="Building the package with the build module...", - success_message="Package build succeeded!", - error_message="Package build failed due to:", - verbose=verbose, + command="python -m build --sdist --wheel", + quiet=quiet, ) + print("Successfully built thehive4py!") -def run_publish(verbose: bool): +def run_upload(quiet: bool): + print("Publishing thehive4py with twine...") _run_subprocess( - args=("echo 'Publish command is not implemented yet...' && exit 1 "), - init_message="Publishing the package with twine...", - success_message="Publish succeeded!", - error_message="Publish failed due to:", - verbose=verbose, + command="twine upload dist/*", + quiet=quiet, ) + print("Successfully published thehive4py!") -def parse_arguments(): - main_parser = argparse.ArgumentParser( +def build_run_options() -> List[dict]: + return [ + {"name": "build", "help": "run build step", "func": run_build}, + {"name": "upload", "help": "run upload step", "func": run_upload}, + ] + + +def parse_arguments(run_options: List[dict]): + parser = argparse.ArgumentParser( prog="thehive4py-cd", - description="run all cd tasks or use sub commands to run cd tasks individually", + description="run all cd steps or use options to run cd steps selectively", ) - main_parser.add_argument( - "-v", - "--verbose", + parser.add_argument( + "-q", + "--quiet", action="store_true", default=False, - help="generate verbose output", + help="silence verbose output", ) - main_parser.set_defaults(func=run_all) - - subparsers = main_parser.add_subparsers(help="commands") - subparser_options = [ - { - "name": "build", - "help": "task to build the package", - "default_func": run_build, - }, - { - "name": "publish", - "help": "task to publish the package", - "default_func": run_publish, - }, - ] - for subparser_option in subparser_options: - _subparser = subparsers.add_parser( - name=subparser_option["name"], - help=subparser_option["help"], - parents=[main_parser], - add_help=False, + for run_option in run_options: + parser.add_argument( + f"--{run_option['name']}", + help=run_option["help"], + action="store_true", ) - _subparser.set_defaults(func=subparser_option["default_func"]) - return main_parser.parse_args() + return parser.parse_args() def main(): - args = parse_arguments() - args.func(verbose=args.verbose) + run_options = build_run_options() + args = parse_arguments(run_options=run_options) + + quiet = args.quiet + + selective_runs = [ + run_option["func"] + for run_option in run_options + if getattr(args, run_option["name"]) + ] + + if selective_runs: + for run in selective_runs: + run(quiet=quiet) + else: + run_all(quiet=quiet) if __name__ == "__main__": diff --git a/scripts/ci.py b/scripts/ci.py index deaffd1c..3db98fa6 100755 --- a/scripts/ci.py +++ b/scripts/ci.py @@ -6,29 +6,21 @@ def _run_subprocess( command: str, - init_message: str, - success_message: str, quiet=False, ): - print(init_message) - if not quiet: stdout = stderr = None else: stdout = stderr = subprocess.DEVNULL - import shlex - try: - subprocess.run(shlex.split(command), stdout=stdout, stderr=stderr, check=True) + subprocess.run(str.split(command), stdout=stdout, stderr=stderr, check=True) except subprocess.CalledProcessError as err: error_output = ( f"ERROR: Execution of command '{command}' returned: {err.returncode}\n" ) print(error_output) exit(err.returncode) - else: - print(success_message, end="\n\n") def check_all(quiet=False): @@ -42,57 +34,57 @@ def check_all(quiet=False): def check_lint(quiet=False): + print("Run lint checks with flake8...") _run_subprocess( command="flake8 thehive4py/ tests/", - init_message="Run lint checks with flake8...", - success_message="Lint checks succeeded!", quiet=quiet, ) + print("Lint checks succeeded!") def check_format(quiet=False): + print("Run format checks with black...") _run_subprocess( command="black --check thehive4py/ tests/", - init_message="Run format checks with black...", - success_message="Format checks succeeded!", quiet=quiet, ) + print("Format checks succeeded!") def check_type(quiet=False): + print("Run type checks with mypy...") _run_subprocess( command="mypy --install-types --non-interactive thehive4py/", - init_message="Run type checks with mypy...", - success_message="Type checks succeeded!", quiet=quiet, ) + print("Type checks succeeded!") def check_cve(quiet=False): + print("Run CVE checks with pip-audit...") _run_subprocess( command="pip-audit .", - init_message="Run CVE checks with pip-audit...", - success_message="CVE checks succeeded!", quiet=quiet, ) + print("CVE checks succeeded!") def check_security(quiet=False): + print("Run security checks with bandit...") _run_subprocess( command="bandit -r thehive4py/", - init_message="Run security checks with bandit...", - success_message="Security checks succeeded!", quiet=quiet, ) + print("Security checks succeeded!") def check_test(quiet=False): + print("Run integration tests with pytest...") _run_subprocess( command="pytest -v --cov", - init_message="Run integration tests with pytest...", - success_message="Integration tests succeeded!", quiet=quiet, ) + print("Integration tests succeeded!") def build_check_options() -> List[dict]: @@ -148,7 +140,6 @@ def main(): check(quiet=quiet) else: check_all(quiet=quiet) - print() if __name__ == "__main__":