refactor: move parse and related function to common

src/BUILD

@@ -12,6 +12,14 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 load("@rules_python//python:py_binary.bzl", "py_binary")
+load("@rules_python//python:py_library.bzl", "py_library")
+
+py_library(
+    name = "common",
+    srcs = ["common.py"],
+    imports = ["."],
+    visibility = ["//visibility:public"],
+)
 
 # Tool filter compile_commands.json file
 py_binary(
@@ -24,6 +32,7 @@ py_binary(
     name = "codechecker_script",
     srcs = ["codechecker_script.py"],
     visibility = ["//visibility:public"],
+    deps = [":common"],
 )
 
 py_binary(

src/codechecker_script.py

@@ -21,10 +21,7 @@
 import os
 import plistlib
 import re
-import shlex
-import subprocess
-import sys
-
+from common import fail, parse, check_results, stage, execute, build_env
 
 START_PATH = r"\/(?:(?!\.\s+)\S)+"
 BAZEL_PATHS = {
@@ -66,47 +63,6 @@ def parse_args(argv=None):
     return args
 
 
-def fail(codechecker_log, message, exit_code=1):
-    """Print error message and return exit code"""
-    logging.error(message)
-    print()
-    print("*" * 50)
-    print("codechecker script execution FAILED!")
-    if codechecker_log:
-        print(f"See: {codechecker_log}")
-        print("*" * 50)
-        try:
-            with open(codechecker_log, encoding="utf-8") as log_file:
-                print(log_file.read())
-        except IOError:
-            print("File not accessible")
-    else:
-        print(message)
-    print("*" * 50)
-    print()
-    sys.exit(exit_code)
-
-
-def read_file(codechecker_log, filename):
-    """Read text file and return its contents"""
-    if not os.path.isfile(filename):
-        fail(codechecker_log, f"File not found: {filename}")
-    with open(filename, encoding="utf-8") as handle:
-        return handle.read()
-
-
-def separator(method="info"):
-    """Print log separator line to logging.info() or other logging methods"""
-    getattr(logging, method)("#" * 23)
-
-
-def stage(title, method="info"):
-    """Print stage title into log"""
-    separator(method)
-    getattr(logging, method)("### " + title)
-    separator(method)
-
-
 def setup(verbosity, codechecker_log):
     """Setup logging parameters for execution session"""
     if verbosity == "INFO":
@@ -119,9 +75,8 @@ def setup(verbosity, codechecker_log):
 
     if codechecker_log:
         logging.basicConfig(
-            filename=codechecker_log,
-            level=log_level,
-            format=log_format)
+            filename=codechecker_log, level=log_level, format=log_format
+        )
     else:
         logging.basicConfig(level=log_level, format=log_format)
 
@@ -144,52 +99,6 @@ def input_data(args):
     logging.debug("")
 
 
-def execute(codechecker_log, cmd, env=None, codes=None):
-    """Execute CodeChecker commands"""
-    if codes is None:
-        codes = [0]
-    with subprocess.Popen(
-        cmd,
-        env=env,
-        shell=True,
-        stdin=subprocess.PIPE,
-        stdout=subprocess.PIPE,
-        stderr=subprocess.PIPE,
-    ) as process:
-        stdout, stderr = process.communicate()
-        stdout = stdout.decode("utf-8")
-        stderr = stderr.decode("utf-8")
-        if process.returncode not in codes:
-            fail(codechecker_log,
-                 f"\ncommand: {cmd}\nstdout: {stdout}\nstderr: {stderr}\n")
-        logging.debug("Executing: %s", cmd)
-        # logging.debug("Output:\n\n%s\n", stdout)
-    return stdout
-
-
-def build_env(args):
-    """Return environment"""
-    env = os.environ.copy()
-    for entry in args.env:
-        if "=" not in entry:
-            fail(args.log, f"Environment entry is not KEY=VALUE: {entry}")
-        key, value = entry.split("=", 1)
-        env[key] = value
-    # Note: This is a workaround, CodeChecker requires the PATH to be set
-    if "PATH" not in env:
-        env["PATH"] = "/bin"
-    if env.get("CC_ANALYZERS_FROM_PATH"):
-        logging.debug("CC_ANALYZERS_FROM_PATH is set: use analyzers from PATH")
-    elif env.get("CC_ANALYZER_BIN"):
-        logging.debug("CC_ANALYZER_BIN is set by the configuration")
-    else:
-        env["CC_ANALYZER_BIN"] = (
-            f"clangsa:{args.clang};clang-tidy:{args.clang_tidy}"
-        )
-    logging.debug("env: %s", str(env))
-    return env
-
-
 def prepare(codechecker_files):
     """Prepare CodeChecker execution environment"""
     stage("CodeChecker files:")
@@ -201,7 +110,7 @@ def prepare(codechecker_files):
 def analyze(args):
     """Run CodeChecker analyze command"""
     stage("CodeChecker analyze:")
-    env = build_env(args)
+    env = build_env(args.env, args.log, args.clang, args.clang_tidy)
     output = execute(
         args.log,
         f"{args.codechecker} analyzers --details",
@@ -341,95 +250,25 @@ def update_file_paths(codechecker_files):
     resolve_symlinks(codechecker_files)
 
 
-def parse(args):
-    """Run CodeChecker parse commands"""
-    stage("CodeChecker parse:")
-    env = build_env(args)
-    logging.info("CodeChecker parse -e json")
-    codechecker_parse = (
-        f"{args.codechecker} parse --config "
-        f"{args.config} {args.output}/data"
-    )
-    # Save results to JSON file
-    command = (
-        f"{codechecker_parse} --export=json > "
-        f"{args.output}/result.json"
-    )
-    execute(args.log, command, env=env, codes=[0, 2])
-    # Save results as HTML report
-    logging.info("CodeChecker parse -e html")
-    command = (
-        codechecker_parse
-        + " --export=html --output="
-        + args.output
-        + "/report"
-    )
-    execute(args.log, command, env=env, codes=[0, 2])
-    # Save results to text file
-    logging.info("CodeChecker parse to text result")
-    result_file = args.output + "/result.txt"
-    command = codechecker_parse + " > " + result_file
-    execute(args.log, command, env=env, codes=[0, 2])
-    logging.info("Result:\n\n%s\n", read_file(args.log, result_file))
-
-
 def run(args):
     """Perform all steps for "bazel build" phase"""
     prepare(args.output)
     analyze(args)
-    parse(args)
+    parse(
+        args.output,
+        args.codechecker,
+        args.config,
+        args.env,
+        args.log,
+        args.clang,
+        args.clang_tidy,
+    )
     update_file_paths(args.output)
 
 
-def check_results(args):
-    """Check/verify CodeChecker results"""
-    stage("Checking result:")
-    # Get results file and read it
-    result_file = args.output + "/result.txt"
-    logging.info("Find CodeChecker results in bazel-bin")
-    logging.info("      all artifacts: %s/", args.output)
-    logging.info("      HTML report:   %s/report/index.html", args.output)
-    logging.info("      result file:   %s", result_file)
-    results = read_file(args.log, result_file)
-    logging.info("Results: \n\n%s\n", results)
-    # Collect defect severities to detect
-    if args.severities is None:
-        fail(args.log,
-             "CodeChecker defect severities are invalid: "
-             f"{str(args.severities)}")
-    severities = shlex.split(args.severities)
-    # Add HIGH severity by default
-    if not severities:
-        severities.append("HIGH")
-    # We should always detect CRITICAL defects
-    if "CRITICAL" not in severities:
-        severities.append("CRITICAL")
-    logging.debug("Severities: %s", str(severities))
-    issues = dict.fromkeys(severities, 0)
-    logging.debug("Issues: %s", str(issues))
-    # Grep results for defects according to severities
-    for issue in issues:
-        found = re.findall(rf"^{issue} .* (\d+)", results, re.M)
-        defects = sum(int(number) for number in found)
-        logging.debug("   %s : %s = %d", issue, str(found), defects)
-        issues[issue] = defects
-    logging.info("Defects: %s", str(issues))
-    # Check collected defects
-    passed = True
-    conclusion = ""
-    for issue, num in issues.items():
-        if num > 0:
-            passed = False
-            conclusion += f"{issue:>15} : {num}\n"
-    if passed:
-        logging.info("No defects found by CodeChecker")
-    else:
-        fail(args.log, f"CodeChecker found defects:\n{conclusion}")
-
-
 def test(args):
     """Perform all steps for "bazel test" phase"""
-    check_results(args)
+    check_results(args.output, args.log, args.severities)
 
 
 def main():