Add JobSummary report for GitHub

scripts/check-urls.py

@@ -1,4 +1,3 @@
-import collections
 import fileinput
 import os
 import re
@@ -9,6 +8,8 @@
 from queue import SimpleQueue
 from typing import Optional
 
+from github_job_summary import JobSummary
+
 """
 Read file names from stdin
 Extract all URL-like strings
@@ -77,9 +78,8 @@ class Curl:
 # print(URL_RE_PATTERN)
 URL_REGEX = re.compile(URL_RE_PATTERN, re.MULTILINE)
 
-BROKEN_URLS = collections.defaultdict(list)
-
-EXTRACTED_URLS_WITH_FILES = {k: [] for k in URLS_TO_IGNORE}
+# URL : [Files]
+EXTRACTED_URLS_WITH_FILES: dict[str, [str]] = {k: [] for k in URLS_TO_IGNORE}
 
 
 def url_extractor(text, filename):
@@ -167,6 +167,7 @@ def process_finished_task(task) -> None:
     )
     if task.ret_code == expected_ret_code:
         print("OK:", "'%s' %.2fs" % (task.url, task.age))
+        JOB_SUMMARY.add_success(task.url)
         return
 
     if task.ret_code == Curl.HTTP_RETURNED_ERROR and expected_http_code:
@@ -176,14 +177,15 @@ def process_finished_task(task) -> None:
         http_code = int(match.groupdict()["http_code"])
         if http_code == expected_http_code:
             print("OK HTTP:", "'%s' %.2fs" % (task.url, task.age))
+            JOB_SUMMARY.add_success(task.url)
             return
 
     print(
         "Expected %d got %d for '%s': %s"
         % (expected_ret_code, task.ret_code, task.url, task.stderr),
         file=sys.stderr,
     )
-    BROKEN_URLS[task.url] = EXTRACTED_URLS_WITH_FILES[task.url]
+    JOB_SUMMARY.add_error(f"Broken URL '{task.url}': {task.stderr}Files: {EXTRACTED_URLS_WITH_FILES[task.url]}")
 
 
 WORKER_QUEUE = SimpleQueue()
@@ -219,7 +221,11 @@ def url_checker(num_workers=8):
     print("Worker finished")
 
 
-def main(files):
+JOB_SUMMARY = JobSummary(os.environ.get('GITHUB_STEP_SUMMARY', "step_summary.md"))
+JOB_SUMMARY.add_header("Test all URLs")
+
+
+def main(files: [str]) -> int:
     checker = threading.Thread(target=url_checker)
     checker.start()
 
@@ -230,18 +236,14 @@ def main(files):
     WORKER_QUEUE.put_nowait(None)
     checker.join()
 
-    if BROKEN_URLS:
-        print("Errors:", file=sys.stdout, flush=True)
-    for url, files in BROKEN_URLS.items():
-        print(
-            "BROKEN URL: '%s' in files: %s"
-            % (url, ", ".join("'%s'" % f for f in files)),
-            file=sys.stderr,
-            flush=True,
-        )
-    if BROKEN_URLS:
-        exit(1)
+    JOB_SUMMARY.finalize("Checked {total} failed **{failed}**\nGood={success}")
+    if JOB_SUMMARY.has_errors:
+        print(JOB_SUMMARY, file=sys.stderr, flush=True)
+        return 1
+    else:
+        print(JOB_SUMMARY, file=sys.stdout, flush=True)
+        return 0
 
 
 if __name__ == "__main__":
-    main([filename.strip() for filename in fileinput.input()])
+    exit(main([filename.strip() for filename in fileinput.input()]))

scripts/github_job_summary.py

@@ -0,0 +1,68 @@
+import contextlib
+from typing import TextIO
+from threading import Lock
+
+
+class JobSummary:
+
+    def __init__(self, filename: str):
+        """
+
+        :param filename: Use $GITHUB_STEP_SUMMARY inside GitHub
+        """
+        self.__file: TextIO = open(filename, mode='wt')
+        self._errors = []
+        self._success = []
+        self._lock = Lock()
+
+    def close(self):
+        assert not self.__file.closed
+        self.__file.close()
+
+    def __str__(self) -> str:
+        if not self.has_errors:
+            return "OK"
+        lines = ["Errors:"] + self._errors
+        return '\n\n'.join(lines)
+
+    def _write_line(self, line):
+        with self._lock:
+            self.__file.write(line)
+
+    @property
+    def has_errors(self) -> bool:
+        return bool(self._errors)
+
+    def add_header(self, text: str, level: int = 3):
+        self._write_line(f"{'#' * level} {text}\n\n")
+
+    def add_error(self, text: str):
+        if not self._errors:
+            self._write_line("Errors:\n")
+        self._errors.append(text)
+        self._write_line(f"\n1. :x: {text}\n")
+
+    def add_success(self, text: str):
+        self._success.append(text)
+
+    def finalize(self, format_str: str):
+        total = len(self._success) + len(self._errors)
+        self._write_line('\n' + format_str.format(
+            total=total,
+            success=len(self._success),
+            failed=len(self._errors)
+        ) + '\n')
+
+
+def test():
+    summary: JobSummary
+    with contextlib.closing(JobSummary("test.md")) as summary:
+        summary.add_header("Test results")
+        summary.add_error("Text for 1 error message")
+        summary.add_error("Text for 2 error message")
+        summary.add_success("OK")
+        summary.finalize("Total={total}, success={success}, failed={failed}")
+
+
+if __name__ == '__main__':
+    test()