Basic unit tests for operations (#81)

Author: debonte

sarif/operations/blame_op.py

@@ -6,14 +6,38 @@
 import os
 import subprocess
 import sys
+from typing import Callable, Iterable, List, Union
 import urllib.parse
 import urllib.request
 
 from sarif.sarif_file import SarifFileSet
 
 
+def _run_git_blame(repo_path: str, file_path: str) -> List[bytes]:
+    cmd = ["git", "blame", "--porcelain", _make_path_git_compatible(file_path)]
+    with subprocess.Popen(cmd, stdout=subprocess.PIPE, cwd=repo_path) as proc:
+        result = []
+        if proc.stdout:
+            result = [x for x in proc.stdout.readlines()]
+
+        # Ensure process terminates
+        proc.communicate()
+        if proc.returncode:
+            cmd_str = " ".join(cmd)
+            sys.stderr.write(
+                f"WARNING: Command `{cmd_str} "
+                f"failed with exit code {proc.returncode} in {repo_path}\n"
+            )
+
+        return result
+
+
 def enhance_with_blame(
-    input_files: SarifFileSet, repo_path: str, output: str, output_multiple_files: bool
+    input_files: SarifFileSet,
+    repo_path: str,
+    output: str,
+    output_multiple_files: bool,
+    run_git_blame: Callable[[str, str], List[bytes]] = _run_git_blame,
 ):
     """
     Enhance SARIF files with information from `git blame`.  The `git` command is run in the current
@@ -26,7 +50,7 @@ def enhance_with_blame(
     if not os.path.isdir(repo_path):
         raise ValueError(f"No git repository directory found at {repo_path}")
 
-    _enhance_with_blame(input_files, repo_path)
+    _enhance_with_blame(input_files, repo_path, run_git_blame)
 
     for input_file in input_files:
         input_file_name = input_file.get_file_name()
@@ -57,7 +81,11 @@ def enhance_with_blame(
             )
 
 
-def _enhance_with_blame(input_files, repo_path):
+def _enhance_with_blame(
+    input_files: SarifFileSet,
+    repo_path: str,
+    run_git_blame: Callable[[str, str], List[bytes]],
+):
     """
     Run `git blame --porcelain` for each file path listed in input_files.
     Then enhance the results in error_list by adding a "blame" property including "hash", "author"
@@ -73,7 +101,7 @@ def _enhance_with_blame(input_files, repo_path):
         "in",
         repo_path,
     )
-    file_blame_info = _run_git_blame_on_files(files_to_blame, repo_path)
+    file_blame_info = _run_git_blame_on_files(files_to_blame, repo_path, run_git_blame)
 
     # Now join up blame output with result list
     blame_info_count = 0
@@ -106,44 +134,40 @@ def _make_path_git_compatible(file_path):
         return file_path
 
 
-def _run_git_blame_on_files(files_to_blame, repo_path):
+def _run_git_blame_on_files(
+    files_to_blame: Iterable[str],
+    repo_path: str,
+    run_git_blame: Callable[[str, str], List[bytes]],
+):
     file_blame_info = {}
     for file_path in files_to_blame:
-        cmd = ["git", "blame", "--porcelain", _make_path_git_compatible(file_path)]
-        with subprocess.Popen(cmd, stdout=subprocess.PIPE, cwd=repo_path) as proc:
-            blame_info = {"commits": {}, "line_to_commit": {}}
-            file_blame_info[file_path] = blame_info
-            commit_hash: str | None = None
-            for line_bytes in proc.stdout.readlines():
-                # Convert byte sequence to string and remove trailing LF
-                line_string = line_bytes.decode("utf-8")[:-1]
-                # Now parse output from git blame --porcelain
-                if commit_hash:
-                    if line_string.startswith("\t"):
-                        commit_hash = None
-                        # Ignore line contents = source code
-                    elif " " in line_string:
-                        space_pos = line_string.index(" ")
-                        key = line_string[0:space_pos]
-                        value = line_string[space_pos + 1 :].strip()
-                        blame_info["commits"][commit_hash][key] = value
-                    else:
-                        # e.g. "boundary"
-                        key = line_string
-                        blame_info["commits"][commit_hash][key] = True
+        git_blame_output = run_git_blame(repo_path, file_path)
+        blame_info = {"commits": {}, "line_to_commit": {}}
+        file_blame_info[file_path] = blame_info
+        commit_hash: Union[str, None] = None
+
+        for line_bytes in git_blame_output:
+            # Convert byte sequence to string and remove trailing LF
+            line_string = line_bytes.decode("utf-8")[:-1]
+            # Now parse output from git blame --porcelain
+            if commit_hash:
+                if line_string.startswith("\t"):
+                    commit_hash = None
+                    # Ignore line contents = source code
+                elif " " in line_string:
+                    space_pos = line_string.index(" ")
+                    key = line_string[0:space_pos]
+                    value = line_string[space_pos + 1 :].strip()
+                    blame_info["commits"][commit_hash][key] = value
                 else:
-                    commit_line_info = line_string.split(" ")
-                    commit_hash = commit_line_info[0]
-                    commit_line = commit_line_info[2]
-                    blame_info["commits"].setdefault(commit_hash, {})
-                    blame_info["line_to_commit"][commit_line] = commit_hash
-
-            # Ensure process terminates
-            proc.communicate()
-            if proc.returncode:
-                cmd_str = " ".join(cmd)
-                sys.stderr.write(
-                    f"WARNING: Command `{cmd_str} "
-                    f"failed with exit code {proc.returncode} in {repo_path}\n"
-                )
+                    # e.g. "boundary"
+                    key = line_string
+                    blame_info["commits"][commit_hash][key] = True
+            else:
+                commit_line_info = line_string.split(" ")
+                commit_hash = commit_line_info[0]
+                commit_line = commit_line_info[2]
+                blame_info["commits"].setdefault(commit_hash, {})
+                blame_info["line_to_commit"][commit_line] = commit_hash
+
     return file_blame_info

sarif/operations/emacs_op.py

@@ -20,13 +20,14 @@
 
 
 def generate_compile(
-    input_files: sarif_file.SarifFileSet, output: str, output_multiple_files: bool
+    input_files: sarif_file.SarifFileSet,
+    output: str,
+    output_multiple_files: bool,
+    date_val: datetime = datetime.now(),
 ):
     """
     Generate txt file from the input files.
     """
-    date_val = datetime.now()
-
     output_file = output
     if output_multiple_files:
         for input_file in input_files:

sarif/operations/html_op.py

@@ -5,6 +5,7 @@
 import base64
 from datetime import datetime
 import os
+from typing import Union
 
 from jinja2 import Environment, FileSystemLoader, select_autoescape
 
@@ -23,15 +24,14 @@
 
 def generate_html(
     input_files: sarif_file.SarifFileSet,
-    image_file: str,
+    image_file: Union[str, None],
     output: str,
     output_multiple_files: bool,
+    date_val: datetime = datetime.now(),
 ):
     """
     Generate HTML file from the input files.
     """
-    date_val = datetime.now()
-
     if image_file:
         image_mime_type = "image/" + os.path.splitext(image_file)[-1]
         if image_mime_type == "image/jpg":

sarif/operations/info_op.py

@@ -5,6 +5,8 @@
 import datetime
 import os
 
+from sarif.sarif_file import SarifFileSet
+
 _BYTES_PER_MIB = 1024 * 1024
 _BYTES_PER_KIB = 1024
 
@@ -103,7 +105,7 @@ def _generate_info_to_file(sarif_files, file_out):
     return file_count
 
 
-def generate_info(sarif_files, output):
+def generate_info(sarif_files: SarifFileSet, output: str):
     """
     Print structure information about the provided `sarif_files`.
     """

sarif/operations/trend_op.py

@@ -3,15 +3,19 @@
 """
 
 import csv
-from typing import Dict, List
+from typing import Dict, List, Literal
 
 from sarif import sarif_file
 from sarif.sarif_file import SarifFileSet
 
 TIMESTAMP_COLUMNS = ["Date", "Tool", *sarif_file.SARIF_SEVERITIES_WITH_NONE]
 
 
-def generate_trend_csv(input_files: SarifFileSet, output_file: str, dateformat: str):
+def generate_trend_csv(
+    input_files: SarifFileSet,
+    output_file: str,
+    dateformat: Literal["dmy", "mdy", "ymd"],
+) -> None:
     """
     Generate a timeline csv of the issues from the SARIF files.  Each SARIF file must contain a
     timestamp of the form 20211012T110000Z in its filename.

sarif/operations/word_op.py

@@ -10,6 +10,7 @@
 
 from datetime import datetime
 import os
+from typing import Union
 
 import docx
 from docx import oxml
@@ -23,9 +24,10 @@
 
 def generate_word_docs_from_sarif_inputs(
     input_files: sarif_file.SarifFileSet,
-    image_file: str,
+    image_file: Union[str, None],
     output: str,
     output_multiple_files: bool,
+    date_val: datetime = datetime.now(),
 ):
     """
     Convert SARIF input to Word file output.
@@ -50,23 +52,26 @@ def generate_word_docs_from_sarif_inputs(
                 report,
                 os.path.join(output, output_file_name),
                 image_file,
+                date_val,
             )
         output_file_name = "static_analysis_output.docx"
         output_file = os.path.join(output, output_file_name)
 
     source_description = input_files.get_description()
     print("Writing Word summary of", source_description, "to", output_file_name)
     report = input_files.get_report()
-    _generate_word_summary(input_files, report, output_file, image_file)
+    _generate_word_summary(input_files, report, output_file, image_file, date_val)
 
 
-def _generate_word_summary(sarif_data, report, output_file, image_file):
+def _generate_word_summary(
+    sarif_data, report, output_file, image_file: Union[str, None], date_val: datetime
+):
     # Create a new document
     document = docx.Document()
 
     severities = report.get_severities()
     _add_heading_and_highlevel_info(
-        document, sarif_data, report, severities, output_file, image_file
+        document, sarif_data, report, severities, output_file, image_file, date_val
     )
     _dump_errors_summary_by_sev(document, report, severities)
     _dump_each_error_in_detail(document, report, severities)
@@ -76,7 +81,13 @@ def _generate_word_summary(sarif_data, report, output_file, image_file):
 
 
 def _add_heading_and_highlevel_info(
-    document, sarif_data, report, severities, output_file, image_path
+    document,
+    sarif_data,
+    report,
+    severities,
+    output_file,
+    image_path: Union[str, None],
+    date_val: datetime,
 ):
     tool_name = ", ".join(sarif_data.get_distinct_tool_names())
     heading = f"Sarif Summary: {tool_name}"
@@ -87,7 +98,7 @@ def _add_heading_and_highlevel_info(
         last_paragraph.alignment = text.WD_PARAGRAPH_ALIGNMENT.CENTER
 
     document.add_heading(heading, 0)
-    document.add_paragraph(f"Document generated on: {datetime.now()}")
+    document.add_paragraph(f"Document generated on: {date_val}")
 
     sevs = ", ".join(severities)
     document.add_paragraph(