Extract report construction; simplify report concatenation

Signed-off-by: Matthias Büchse <matthias.buechse@alasca.cloud>

Author: mbuechse

Tests/scs-compliance-check.py

@@ -19,7 +19,6 @@
 
 import os
 import os.path
-import uuid
 import re
 import sys
 import shlex
@@ -29,7 +28,8 @@
 import logging
 import yaml
 
-from scs_cert_lib import load_spec, annotate_validity, eval_buckets, TESTCASE_VERDICTS
+from scs_cert_lib import load_spec, annotate_validity, eval_buckets, TESTCASE_VERDICTS, \
+    make_report
 
 
 logger = logging.getLogger(__name__)
@@ -232,24 +232,6 @@ def print_report(testcase_lookup: dict, tc_ids: list, results: dict, partial=Fal
                 print(f"      > {testcase['url']}")
 
 
-def create_report(config, spec, log, results):
-    return {
-        "uuid": str(uuid.uuid4()),
-        "creator": "SCS test suite; version=0.1.0",  # TODO put actual version of test suite here
-        "scope": spec['uuid'],
-        "checked_at": datetime.datetime.now(),
-        "reference_date": config.checkdate,
-        "subject": config.subject,
-        "tests": {
-            testcase_id: {
-                "result": value,
-            }
-            for testcase_id, value in results.items()
-        },
-        "log": log,
-    }
-
-
 def main(argv):
     """Entry point for the checker"""
     logging.basicConfig(format='%(levelname)s: %(message)s', level=logging.INFO)
@@ -318,7 +300,7 @@ def main(argv):
             print(f"- {version['version']}:", end=' ')
             print_report(testcase_lookup, version['testcase_ids'], results, partial, config.verbose)
     if config.output:
-        report = create_report(config, spec, log, results)
+        report = make_report(spec['uuid'], config.subject, results, log)
         with open(config.output, 'w', encoding='UTF-8') as fileobj:
             yaml.safe_dump(report, fileobj, default_flow_style=False, sort_keys=False, explicit_start=True)
     num_error = len([tc_id for tc_id, value in results.items() if value != 1])

Tests/scs-test-runner.py

@@ -159,18 +159,18 @@ def _run_commands(commands, num_workers=5):
 def _concat_files(source_paths, target_path):
     with open(target_path, 'wb') as tfileobj:
         for path in source_paths:
-            with open(path, 'rb') as sfileobj:
-                shutil.copyfileobj(sfileobj, tfileobj)
+            try:
+                with open(path, 'rb') as sfileobj:
+                    shutil.copyfileobj(sfileobj, tfileobj)
+            except FileNotFoundError:
+                logger.warning(f"Skipping non-extant {path}")
 
 
-def _move_file(source_path, target_path):
-    # for Windows people, remove target first, but don't try too hard (Windows is notoriously bad at this)
-    # this two-stage delete-rename approach does have a tiny (irrelevant) race condition (thx Windows)
+def _remove_file(path):
     try:
-        os.remove(target_path)
+        os.remove(path)
     except FileNotFoundError:
         pass
-    os.rename(source_path, target_path)
 
 
 @cli.command()
@@ -213,18 +213,17 @@ def run(cfg, scopes, subjects, sections, preset, num_workers, monitor_url, repor
     logger.debug(f'running tests for scope(s) {", ".join(scopes)} and subject(s) {", ".join(subjects)}')
     logger.debug(f'monitor url: {monitor_url}, num_workers: {num_workers}, output: {report_yaml}')
     with tempfile.TemporaryDirectory(dir=cfg.cwd) as tdirname:
-        report_yaml_tmp = os.path.join(tdirname, 'report.yaml')
         jobs = [(scope, subject) for scope in scopes for subject in subjects]
         outputs = [os.path.join(tdirname, f'report-{idx}.yaml') for idx in range(len(jobs))]
         commands = [cfg.build_check_command(job[0], job[1], sections, output) for job, output in zip(jobs, outputs)]
         _run_commands(commands, num_workers=num_workers)
-        _concat_files(outputs, report_yaml_tmp)
         if report_yaml is None:
-            report_yaml = report_yaml_tmp
-        else:
-            _move_file(report_yaml_tmp, report_yaml)
+            report_yaml = os.path.join(tdirname, 'report.yaml')
+        _concat_files(outputs, report_yaml)
+        _remove_file(report_yaml + '.sig')
         subprocess.run(**cfg.build_sign_command(report_yaml))
         subprocess.run(**cfg.build_upload_command(report_yaml, monitor_url))
+        sys.stdout.write('\n')  # curl output does not end in newline
     return 0
 
 

Tests/scs_cert_lib.py

@@ -9,9 +9,11 @@
 from collections import defaultdict
 from datetime import datetime, date, timedelta
 import logging
+import uuid
 
 
 logger = logging.getLogger(__name__)
+__VERSION__ = "20260623"
 
 # valid keywords for various parts of the spec, to be checked using `check_keywords`
 KEYWORDS = {
@@ -219,3 +221,22 @@ def evaluate(results, testcase_ids) -> int:
         results.get(testcase_id, {}).get('result') or 0
         for testcase_id in testcase_ids
     ], default=0)
+
+
+def make_report(scopeuuid, subject, results, log, creator=None, checked_at=None):
+    if creator is None:
+        creator = f"SCS test suite; version={__VERSION__}"
+    return {
+        "uuid": str(uuid.uuid4()),
+        "creator": creator,
+        "scope": scopeuuid,
+        "checked_at": checked_at or datetime.now(),
+        "subject": subject,
+        "tests": {
+            testcase_id: {
+                "result": value,
+            }
+            for testcase_id, value in results.items()
+        },
+        "log": log,
+    }