#296 - refactor ci and cd scripts

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__":

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__":