Overhaul test scripts in a major way

Tests/cli.py

@@ -0,0 +1,180 @@
+#!/usr/bin/env python3
+# SPDX-License-Identifier: Apache-2.0
+
+import datetime
+import logging
+import os
+import os.path
+import re
+import sys
+import tempfile
+import uuid
+
+import click
+import yaml
+
+from scs_cert_lib import load_spec, annotate_validity, add_period, normalize_scope
+
+
+HERE = os.path.dirname(__file__)
+DEFAULT_SPECPATH = {
+    '50393e6f-2ae1-4c5c-a62c-3b75f2abef3f': os.path.join(HERE, 'scs-compatible-iaas.yaml'),
+    '1fffebe6-fd4b-44d3-a36c-fc58b4bb0180': os.path.join(HERE, 'scs-compatible-kaas.yaml'),
+}
+
+logger = logging.getLogger(__name__)
+
+
+def _load_spec(specpath):
+    specpath = DEFAULT_SPECPATH.get(normalize_scope(specpath), specpath)
+    with open(specpath, "r", encoding="UTF-8") as fileobj:
+        return load_spec(yaml.load(fileobj, Loader=yaml.SafeLoader))
+
+
+def select_valid(versions: list) -> list:
+    return [version for version in versions if version['_explicit_validity']]
+
+
+@click.group()
+def cli():
+    return
+
+
+@cli.command()
+@click.option('--version', '-V', 'version', type=str, default=None)
+@click.option('--tests', '-t', 'tests', type=str, default=None)
+@click.option('--section', '-S', 'sections', type=str, multiple=True)
+@click.argument('specpath', type=str)
+def select(specpath, version, sections, tests):
+    spec = _load_spec(specpath)
+    checkdate = datetime.date.today()
+    annotate_validity(spec['timeline'], spec['versions'], checkdate)
+    if version is None:
+        versions = select_valid(spec['versions'].values())
+    else:
+        versions = [spec['versions'].get(version)]
+        if versions[0] is None:
+            raise RuntimeError(f"Requested version '{version}' not found")
+    if not versions:
+        raise RuntimeError(f"No valid version found for {checkdate}")
+    title = spec['name']
+    if sections:
+        title += f" [sections: {', '.join(sections)}]"
+    if tests:
+        title += f" [tests: '{tests}']"
+        tests_re = re.compile(tests)
+    # collect all testcases we need
+    testcase_lookup = spec['testcases']
+    all_testcase_ids = set()
+    for version in versions:
+        for tc_id in version['testcase_ids']:
+            if sections and testcase_lookup.get(tc_id, {}).get('section') not in sections:
+                continue
+            if tests and not tests_re.match(tc_id):
+                continue
+            all_testcase_ids.add(tc_id)
+    logger.info(f"{title}: {len(all_testcase_ids)} testcases")
+    if all_testcase_ids:
+        print('\n'.join(sorted(all_testcase_ids)))
+
+
+def _update_scorecard(scorecard, report, testcase_lookup):
+    scores = scorecard.setdefault('tests', {})
+    checked_at = report['checked_at']
+    checked_at_str = str(checked_at)[:19]
+    for tc_id, result in report.get('tests', {}).items():
+        score = scores.get(tc_id)
+        if score is None or score['checked_at'] < checked_at_str:
+            testcase = testcase_lookup.get(tc_id)
+            lifetime = testcase.get('lifetime')  # leave None if not present; to be handled by add_period
+            expires_at = add_period(checked_at, lifetime)
+            scores[tc_id] = {'checked_at': checked_at_str, 'expires_at': str(expires_at), **result}
+
+
+def _prune_scorecard(scorecard):
+    scores = scorecard.setdefault('tests', {})
+    now_str = str(datetime.datetime.now())[:19]
+    for tc_id in list(scores):
+        score = scores[tc_id]
+        if score['expires_at'] >= now_str:
+            del scores[tc_id]
+
+
+def _atomic_write(path, text):
+    with tempfile.NamedTemporaryFile(
+        mode='w', encoding='UTF-8',
+        dir=os.path.dirname(path) or '.',
+        delete=False, delete_on_close=False,
+    ) as fileobj:
+        fileobj.write(text)
+    os.rename(fileobj.name, path)
+
+
+def _dump(o):
+    return yaml.safe_dump(o, default_flow_style=False, sort_keys=False, explicit_start=True)
+
+
+@cli.command()
+@click.option('--subject', '-s', 'subject', type=str)
+@click.option('--score', '-S', 'score_yaml', type=click.Path(exists=False), default=None)
+@click.option('-o', '--output', 'report_yaml', type=click.Path(exists=False), default=None)
+@click.option('--spec', 'specpath', type=click.Path(exists=True))
+def score(specpath, subject, score_yaml, report_yaml):
+    scorecard = {}
+    if score_yaml and os.path.exists(score_yaml):
+        with open(score_yaml, "r", encoding="UTF-8") as fileobj:
+            scorecard = yaml.load(fileobj, Loader=yaml.SafeLoader)
+        if not subject:
+            subject = scorecard['subject']
+        if not specpath:
+            specpath = scorecard['scope']
+    elif not subject:
+        raise click.UsageError('need to supply at least one of -s or -S')
+    elif not specpath:
+        raise click.UsageError('need to supply at least one of --spec or -S')
+    spec = _load_spec(specpath)
+    scopeuuid = spec['uuid']
+    snippet = yaml.load(sys.stdin.read(), Loader=yaml.SafeLoader)
+    if not snippet:
+        logger.warning('Empty report snippet. Bailing')
+        return
+    report = {
+        'uuid': str(uuid.uuid4()),
+        'subject': subject,
+        'scope': scopeuuid,
+        **snippet
+    }
+    if report_yaml is None:
+        ts = str(report['checked_at'])[:19]
+        ts = ts.replace(':', '').replace('-', '').replace(' ', 'T')
+        report_yaml = f'report-{ts}-{subject}.yaml'
+    _atomic_write(report_yaml, _dump(report))
+    if score_yaml:
+        scorecard.setdefault('subject', subject)
+        scorecard.setdefault('scope', scopeuuid)
+        if subject != scorecard['subject']:
+            raise RuntimeError('subjects do not match')
+        if scopeuuid != scorecard['scope']:
+            raise RuntimeError('scopes do not match')
+        _prune_scorecard(scorecard)
+        _update_scorecard(scorecard, report, spec['testcases'])
+        _atomic_write(score_yaml, _dump(scorecard))
+
+
+@cli.command()
+@click.option('--subject', '-s', 'subject', type=str)
+@click.option('--spec', 'specpath', type=click.Path(exists=True))
+@click.argument('score_yaml', type=click.Path(exists=False))
+def init(specpath, subject, score_yaml):
+    spec = _load_spec(specpath)
+    scorecard = {
+        'subject': subject,
+        'scope': spec['uuid'],
+        'tests': {},
+    }
+    _atomic_write(score_yaml, _dump(scorecard))
+
+
+if __name__ == "__main__":
+    logging.basicConfig(format='%(levelname)s: %(message)s', level=logging.INFO)
+    cli()

Tests/iaas/openstack_test.py

@@ -7,12 +7,14 @@
 SPDX-License-Identifier: CC-BY-SA 4.0
 """
 
+from datetime import datetime
 import getopt
 import logging
 import os
 import sys
 
 import openstack
+import yaml
 
 from scs_0100_flavor_naming.flavor_names import compute_flavor_spec
 from scs_0100_flavor_naming.flavor_names_check import \
@@ -55,6 +57,8 @@ def usage(rcode=1, file=sys.stderr):
     print("Options: [-c/--os-cloud OS_CLOUD] sets cloud environment (default from OS_CLOUD env)", file=file)
     print("Runs specified testcases against the OpenStack cloud OS_CLOUD", file=file)
     print("and reports inconsistencies, errors etc. It returns 0 on success.", file=file)
+    print("Instead of listing testcase-ids, you can supply a single dash (-)", file=file)
+    print("to have them read from stdin, one testcase-id per line.", file=file)
     sys.exit(rcode)
 
 
@@ -179,7 +183,8 @@ def __init__(self):
     def __getattr__(self, key):
         val = self._values.get(key)
         if val is None:
-            logger.debug(f'... {key}')
+            # uncomment for super serious debugging
+            # logger.log(f'... {key}')
             try:
                 ret = self._functions[key](self)
             except BaseException as e:
@@ -203,28 +208,24 @@ def add_value(self, name, value):
         self._values[name] = value
 
 
-def harness(name, *check_fns):
+def harness(name, results, *check_fns):
     """Harness for evaluating testcase `name`.
 
-    Logs beginning of computation.
+    Logs beginning and end of computation.
     Calls each fn in `check_fns`.
-    Prints (to stdout) 'name: RESULT', where RESULT is one of
-
-    - 'ABORT' if an exception occurs during the function calls
-    - 'FAIL' if one of the functions has a falsy result
-    - 'PASS' otherwise
+    Records result to `results`.
     """
-    logger.debug(f'** {name}')
+    logger.info(f'*** {name}')
     try:
         result = all(check_fn() for check_fn in check_fns)
     except BaseException:
         logger.debug('exception during check', exc_info=True)
-        result = 'ABORT'
+        value = 0
     else:
-        result = ['FAIL', 'PASS'][min(1, result)]
-    # this is quite redundant
-    # logger.debug(f'** computation end for {name}')
-    print(f"{name}: {result}")
+        value = 1 if result else -1
+    result = ['FAIL', 'ABORT', 'PASS'][value + 1]
+    logger.info(f'+++ {name}: {result}')
+    results[name] = value
 
 
 def run_preflight_checks(container):
@@ -241,6 +242,15 @@ def run_preflight_checks(container):
         raise RuntimeError("OpenStack user is missing member role.")
 
 
+class _LogHandler(logging.Handler):
+    def __init__(self, level=logging.NOTSET, log=None):
+        super().__init__(level=level)
+        self.log = [] if log is None else log
+
+    def handle(self, record):
+        self.log.append(f'{record.levelname}: {record.msg}')
+
+
 def main(argv):
     # configure logging, disable verbose library logging
     logging.basicConfig(format='%(levelname)s: %(message)s', level=logging.DEBUG)
@@ -264,6 +274,9 @@ def main(argv):
         else:
             usage(2)
 
+    if len(args) == 1 and args[0] == '-':
+        args = sys.stdin.read().splitlines()
+
     testcases = [t for t in args if t.startswith('scs-')]
     if len(testcases) != len(args):
         unknown = [a for a in args if a not in testcases]
@@ -281,8 +294,24 @@ def main(argv):
         for testcase in testcases:
             print(f"{testcase}: ABORT")
         raise
+
+    results = {}
+    log = []
+    logging.root.addHandler(_LogHandler(level=logging.DEBUG, log=log))
     for testcase in testcases:
-        harness(testcase, lambda: getattr(c, testcase.replace('-', '_')))
+        harness(testcase, results, lambda: getattr(c, testcase.replace('-', '_')))
+    report = {
+        'creator': 'openstack_test.py v0.1.0',
+        'checked_at': datetime.now(),
+        'tests': {
+            key: {'result': value}
+            for key, value in results.items()
+        },
+        'log': log,
+    }
+    # don't do explicit_start here because that can easily be done by the caller using "echo ---",
+    # and then the caller can even add fields such as uuid, subject, and scope
+    yaml.safe_dump(report, sys.stdout, default_flow_style=False, sort_keys=False, explicit_start=False)
     return 0