Factor out test suite running to a standalone script

2 files changed, 419 insertions, 199 deletions

diff --git a/test/run_test_suite.py b/test/run_test_suite.py
new file mode 100755
index 00000000..e4f5a79a
--- /dev/null
+++ b/test/run_test_suite.py
@@ -0,0 +1,380 @@
+#!/usr/bin/env python3
+
+"""Run an RDF test suite with serdi."""
+
+import argparse
+import datetime
+import difflib
+import itertools
+import os
+import re
+import shlex
+import subprocess
+import sys
+import tempfile
+import urllib.parse
+
+
+def earl_assertion(test, passed, asserter):
+    """Return a Turtle description of an assertion for the test report."""
+
+    asserter_str = ""
+    if asserter is not None:
+        asserter_str = "\n\tearl:assertedBy <%s> ;" % asserter
+
+    return """
+[]
+\ta earl:Assertion ;%s
+\tearl:subject <http://drobilla.net/sw/serd> ;
+\tearl:test <%s> ;
+\tearl:result [
+\t\ta earl:TestResult ;
+\t\tearl:outcome %s ;
+\t\tdc:date "%s"^^xsd:dateTime
+\t] .
+""" % (
+        asserter_str,
+        test,
+        "earl:passed" if passed else "earl:failed",
+        datetime.datetime.now().replace(microsecond=0).isoformat(),
+    )
+
+
+def log_error(message):
+    """Log an error message to stderr"""
+
+    sys.stderr.write("error: ")
+    sys.stderr.write(message)
+
+
+def test_thru(
+    base_uri,
+    path,
+    check_path,
+    out_test_dir,
+    flags,
+    isyntax,
+    osyntax,
+    command_prefix,
+):
+    """Test lossless round-tripping through two different syntaxes."""
+
+    assert isyntax is not None
+    assert osyntax is not None
+
+    test_name = os.path.basename(path)
+    out_path = os.path.join(out_test_dir, test_name + ".pass")
+    thru_path = os.path.join(out_test_dir, test_name + ".thru")
+
+    out_cmd = (
+        command_prefix
+        + [f for sublist in flags for f in sublist]
+        + [
+            "-i",
+            isyntax,
+            "-o",
+            isyntax,
+            "-p",
+            "foo",
+            path,
+            base_uri,
+        ]
+    )
+
+    thru_cmd = command_prefix + [
+        "-i",
+        isyntax,
+        "-o",
+        osyntax,
+        "-c",
+        "foo",
+        out_path,
+        base_uri,
+    ]
+
+    with open(out_path, "wb") as out:
+        subprocess.run(out_cmd, check=True, stdout=out)
+
+    with open(thru_path, "wb") as out:
+        subprocess.run(thru_cmd, check=True, stdout=out)
+
+    if not _file_equals(check_path, thru_path):
+        log_error(
+            "Round-tripped output {} does not match {}\n".format(
+                check_path, thru_path
+            )
+        )
+        return 1
+
+    return 0
+
+
+def _uri_path(uri):
+    path = urllib.parse.urlparse(uri).path
+    drive = os.path.splitdrive(path[1:])[0]
+    return path if not drive else path[1:]
+
+
+def _test_output_syntax(test_class):
+    """Return the output syntax use for a given test class."""
+
+    if "NTriples" in test_class or "Turtle" in test_class:
+        return "NTriples"
+
+    if "NQuads" in test_class or "Trig" in test_class:
+        return "NQuads"
+
+    raise Exception("Unknown test class <{}>".format(test_class))
+
+
+def _load_rdf(filename, base_uri, command_prefix):
+    """Load an RDF file as dictionaries via serdi (only supports URIs)."""
+
+    rdf_type = "http://www.w3.org/1999/02/22-rdf-syntax-ns#type"
+    model = {}
+    instances = {}
+
+    cmd = command_prefix + [filename, base_uri]
+    proc = subprocess.run(cmd, capture_output=True, check=True)
+    for line in proc.stdout.splitlines():
+        matches = re.match(
+            r"<([^ ]*)> <([^ ]*)> <([^ ]*)> \.", line.decode("utf-8")
+        )
+        if matches:
+            s, p, o = (matches.group(1), matches.group(2), matches.group(3))
+            if s not in model:
+                model[s] = {p: [o]}
+            elif p not in model[s]:
+                model[s][p] = [o]
+            else:
+                model[s][p].append(o)
+
+            if p == rdf_type:
+                if o not in instances:
+                    instances[o] = set([s])
+                else:
+                    instances[o].update([s])
+
+    return model, instances
+
+
+def _option_combinations(options):
+    """Return an iterator that cycles through all combinations of options."""
+
+    combinations = []
+    for count in range(len(options) + 1):
+        combinations += list(itertools.combinations(options, count))
+
+    return itertools.cycle(combinations)
+
+
+def _show_diff(from_lines, to_lines, from_filename, to_filename):
+    same = True
+    for line in difflib.unified_diff(
+        from_lines,
+        to_lines,
+        fromfile=os.path.abspath(from_filename),
+        tofile=os.path.abspath(to_filename),
+    ):
+        sys.stderr.write(line)
+        same = False
+
+    return same
+
+
+def _file_equals(patha, pathb):
+
+    for path in (patha, pathb):
+        if not os.access(path, os.F_OK):
+            log_error("missing file {}\n".format(path))
+            return False
+
+    with open(patha, "r", encoding="utf-8") as fa:
+        with open(pathb, "r", encoding="utf-8") as fb:
+            return _show_diff(fa.readlines(), fb.readlines(), patha, pathb)
+
+
+def test_suite(
+    manifest_path,
+    base_uri,
+    report_filename,
+    isyntax,
+    command_prefix,
+):
+    """Run all tests in a test suite manifest."""
+
+    mf = "http://www.w3.org/2001/sw/DataAccess/tests/test-manifest#"
+    test_dir = os.path.dirname(manifest_path)
+    model, instances = _load_rdf(manifest_path, base_uri, command_prefix)
+
+    top_dir = os.path.commonpath([os.getcwd(), os.path.abspath(test_dir)])
+    out_test_dir = os.path.relpath(test_dir, top_dir)
+
+    os.makedirs(out_test_dir, exist_ok=True)
+
+    asserter = ""
+    if os.getenv("USER") == "drobilla":
+        asserter = "http://drobilla.net/drobilla#me"
+
+    class Results:
+        def __init__(self):
+            self.n_tests = 0
+            self.n_failures = 0
+
+    def run_tests(test_class, tests, expected_return, results):
+        thru_flags = [["-e"], ["-f"], ["-b"], ["-r", "http://example.org/"]]
+        osyntax = _test_output_syntax(test_class)
+        thru_options_iter = _option_combinations(thru_flags)
+
+        for test in sorted(tests):
+            test_uri = model[test][mf + "action"][0]
+            test_uri_path = _uri_path(test_uri)
+            test_name = os.path.basename(test_uri_path)
+            test_path = os.path.join(test_dir, test_name)
+
+            command = command_prefix + ["-f", test_path, test_uri]
+            command_string = " ".join(shlex.quote(c) for c in command)
+            out_filename = os.path.join(out_test_dir, test_name + ".out")
+
+            results.n_tests += 1
+
+            if expected_return == 0:  # Positive test
+
+                # Run strict test
+                with open(out_filename, "w") as stdout:
+                    proc = subprocess.run(command, check=False, stdout=stdout)
+                    if proc.returncode == 0:
+                        passed = True
+                    else:
+                        results.n_failures += 1
+                        log_error(
+                            "Unexpected failure of command: {}\n".format(
+                                command_string
+                            )
+                        )
+
+                if proc.returncode == 0 and mf + "result" in model[test]:
+
+                    # Check output against expected output from test suite
+                    check_uri = model[test][mf + "result"][0]
+                    check_filename = os.path.basename(_uri_path(check_uri))
+                    check_path = os.path.join(test_dir, check_filename)
+
+                    if not _file_equals(check_path, out_filename):
+                        results.n_failures += 1
+                        log_error(
+                            "Output {} does not match {}".format(
+                                out_filename, check_path
+                            )
+                        )
+
+                    # Run round-trip tests
+                    results.n_failures += test_thru(
+                        test_uri,
+                        test_path,
+                        check_path,
+                        out_test_dir,
+                        list(next(thru_options_iter)),
+                        isyntax,
+                        osyntax,
+                        command_prefix,
+                    )
+
+            else:  # Negative test
+                with open(out_filename, "w") as stdout:
+                    with tempfile.TemporaryFile() as stderr:
+                        proc = subprocess.run(
+                            command, check=False, stdout=stdout, stderr=stderr
+                        )
+
+                        if proc.returncode != 0:
+                            passed = True
+                        else:
+                            results.n_failures += 1
+                            log_error(
+                                "Unexpected success of command: {}\n".format(
+                                    command_string
+                                )
+                            )
+
+                        # Check that an error message was printed
+                        stderr.seek(0, 2)  # Seek to end
+                        if stderr.tell() == 0:  # Empty
+                            results.n_failures += 1
+                            log_error(
+                                "No error message printed by command: {}\n".format(
+                                    command_string
+                                )
+                            )
+                            result = 1
+
+            # Write test report entry
+            if report_filename:
+                with open(report_filename, "a") as report:
+                    report.write(earl_assertion(test, passed, asserter))
+
+    # Run all test types in the test suite
+    results = Results()
+    ns_rdftest = "http://www.w3.org/ns/rdftest#"
+    for test_class, instances in instances.items():
+        if test_class.startswith(ns_rdftest):
+            expected = (
+                1
+                if "-l" not in command_prefix and "Negative" in test_class
+                else 0
+            )
+            run_tests(test_class, instances, expected, results)
+
+    # Print result summary
+    if results.n_failures > 0:
+        log_error(
+            "{}/{} tests failed\n".format(results.n_failures, results.n_tests)
+        )
+    else:
+        sys.stdout.write("All {} tests passed\n".format(results.n_tests))
+
+    return results.n_failures
+
+
+def main():
+    """Run the command line tool."""
+
+    parser = argparse.ArgumentParser(
+        usage="%(prog)s [OPTION]... MANIFEST BASE_URI -- [SERDI_OPTION]...",
+        description=__doc__,
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+    )
+
+    parser.add_argument("--report", help="path to write result report to")
+    parser.add_argument("--serdi", default="serdi", help="path to serdi")
+    parser.add_argument("--syntax", default="turtle", help="input syntax")
+    parser.add_argument("--wrapper", default="", help="executable wrapper")
+    parser.add_argument("manifest", help="test suite manifest.ttl file")
+    parser.add_argument("base_uri", help="base URI for tests")
+    parser.add_argument(
+        "serdi_option", nargs=argparse.REMAINDER, help="option for serdi"
+    )
+
+    args = parser.parse_args(sys.argv[1:])
+    command_prefix = (
+        shlex.split(args.wrapper) + [args.serdi] + args.serdi_option
+    )
+
+    return test_suite(
+        args.manifest,
+        args.base_uri,
+        args.report,
+        args.syntax,
+        command_prefix,
+    )
+
+
+if __name__ == "__main__":
+    try:
+        sys.exit(main())
+    except subprocess.CalledProcessError as e:
+        if e.stderr is not None:
+            sys.stderr.write(e.stderr.decode("utf-8"))
+
+        sys.stderr.write("error: %s\n" % e)
+        sys.exit(e.returncode)
diff --git a/wscript b/wscript
index 2a4085b4..0264ee13 100644
--- a/wscript
+++ b/wscript
@@ -361,185 +361,6 @@ def amalgamate(ctx):
         Logs.info('Wrote build/serd.%s' % i)
 
 
-def earl_assertion(test, passed, asserter):
-    import datetime
-
-    asserter_str = ''
-    if asserter is not None:
-        asserter_str = '\n\tearl:assertedBy <%s> ;' % asserter
-
-    return '''
-[]
-	a earl:Assertion ;%s
-	earl:subject <http://drobilla.net/sw/serd> ;
-	earl:test <%s> ;
-	earl:result [
-		a earl:TestResult ;
-		earl:outcome %s ;
-		dc:date "%s"^^xsd:dateTime
-	] .
-''' % (asserter_str,
-       test,
-       'earl:passed' if passed else 'earl:failed',
-       datetime.datetime.now().replace(microsecond=0).isoformat())
-
-
-serdi = './serdi_static'
-
-
-def test_thru(check, base, path, check_path, flags, isyntax, osyntax, opts=[]):
-    out_path = path + '.pass'
-    out_cmd = [serdi] + opts + [f for sublist in flags for f in sublist] + [
-        '-i', isyntax,
-        '-o', isyntax,
-        '-p', 'foo',
-        check.tst.src_path(path), base]
-
-    thru_path = path + '.thru'
-    thru_cmd = [serdi] + opts + [
-        '-i', isyntax,
-        '-o', osyntax,
-        '-c', 'foo',
-        out_path,
-        base]
-
-    return (check(out_cmd, stdout=out_path, verbosity=0, name=out_path) and
-            check(thru_cmd, stdout=thru_path, verbosity=0, name=thru_path) and
-            check.file_equals(check_path, thru_path, verbosity=0))
-
-
-def file_uri_to_path(uri):
-    try:
-        from urlparse import urlparse  # Python 2
-    except ImportError:
-        from urllib.parse import urlparse  # Python 3
-
-    path  = urlparse(uri).path
-    drive = os.path.splitdrive(path[1:])[0]
-    return path if not drive else path[1:]
-
-
-def _test_output_syntax(test_class):
-    if 'NTriples' in test_class or 'Turtle' in test_class:
-        return 'NTriples'
-    elif 'NQuads' in test_class or 'Trig' in test_class:
-        return 'NQuads'
-    raise Exception('Unknown test class <%s>' % test_class)
-
-
-def _wrapped_command(cmd):
-    if Options.options.wrapper:
-        import shlex
-        return shlex.split(Options.options.wrapper) + cmd
-
-    return cmd
-
-
-def _load_rdf(filename):
-    "Load an RDF file into python dictionaries via serdi.  Only supports URIs."
-    import subprocess
-    import re
-
-    rdf_type = 'http://www.w3.org/1999/02/22-rdf-syntax-ns#type'
-    model = {}
-    instances = {}
-
-    cmd = _wrapped_command(['./serdi_static', filename])
-    proc = subprocess.Popen(cmd, stdout=subprocess.PIPE)
-    for line in proc.communicate()[0].splitlines():
-        matches = re.match(r'<([^ ]*)> <([^ ]*)> <([^ ]*)> \.',
-                           line.decode('utf-8'))
-        if matches:
-            s, p, o = (matches.group(1), matches.group(2), matches.group(3))
-            if s not in model:
-                model[s] = {p: [o]}
-            elif p not in model[s]:
-                model[s][p] = [o]
-            else:
-                model[s][p].append(o)
-
-            if p == rdf_type:
-                if o not in instances:
-                    instances[o] = set([s])
-                else:
-                    instances[o].update([s])
-
-    return model, instances
-
-
-def _option_combinations(options):
-    "Return an iterator that cycles through all combinations of options"
-    import itertools
-
-    combinations = []
-    for n in range(len(options) + 1):
-        combinations += list(itertools.combinations(options, n))
-
-    return itertools.cycle(combinations)
-
-
-def test_suite(ctx, base_uri, testdir, report, isyntax, options=[]):
-    srcdir = ctx.path.abspath()
-
-    mf = 'http://www.w3.org/2001/sw/DataAccess/tests/test-manifest#'
-    manifest_path = os.path.join(srcdir, 'test', testdir, 'manifest.ttl')
-    model, instances = _load_rdf(manifest_path)
-
-    asserter = ''
-    if os.getenv('USER') == 'drobilla':
-        asserter = 'http://drobilla.net/drobilla#me'
-
-    def run_tests(test_class, tests, expected_return):
-        thru_flags = [['-e'], ['-f'], ['-b'], ['-r', 'http://example.org/']]
-        osyntax = _test_output_syntax(test_class)
-        thru_options_iter = _option_combinations(thru_flags)
-        tests_name = '%s.%s' % (testdir, test_class[test_class.find('#') + 1:])
-        with ctx.group(tests_name) as check:
-            for test in sorted(tests):
-                action_node = model[test][mf + 'action'][0]
-                basename    = os.path.basename(action_node)
-                action      = os.path.join('test', testdir, basename)
-                rel_action  = os.path.join(os.path.relpath(srcdir), action)
-                uri         = base_uri + os.path.basename(action)
-                command     = [serdi] + options + ['-f', rel_action, uri]
-
-                # Run strict test
-                if expected_return == 0:
-                    result = check(command,
-                                   stdout=action + '.out',
-                                   name=action)
-                else:
-                    result = check(command,
-                                   stdout=action + '.out',
-                                   stderr=autowaf.NONEMPTY,
-                                   expected=expected_return,
-                                   name=action)
-
-                if (result and expected_return == 0 and
-                    ((mf + 'result') in model[test])):
-                    # Check output against test suite
-                    check_uri  = model[test][mf + 'result'][0]
-                    check_path = ctx.src_path(file_uri_to_path(check_uri))
-                    result     = check.file_equals(action + '.out', check_path)
-
-                    # Run round-trip tests
-                    if result:
-                        test_thru(check, uri, action, check_path,
-                                  list(next(thru_options_iter)),
-                                  isyntax, osyntax, options)
-
-                # Write test report entry
-                if report is not None:
-                    report.write(earl_assertion(test, result, asserter))
-
-    ns_rdftest = 'http://www.w3.org/ns/rdftest#'
-    for test_class, instances in instances.items():
-        if test_class.startswith(ns_rdftest):
-            expected = (1 if '-l' not in options and 'Negative' in test_class
-                        else 0)
-            run_tests(test_class, instances, expected)
-
-
 def test(tst):
     import tempfile
 
@@ -554,6 +375,7 @@ def test(tst):
         except Exception:
             pass
 
+    serdi = './serdi_static'
     srcdir = tst.path.abspath()
 
     with tst.group('Unit') as check:
@@ -634,27 +456,45 @@ def test(tst):
         except ImportError:
             Logs.warn('Failed to import rdflib, not running NEWS tests')
 
-    # Serd-specific test suites
-    serd_base = 'http://drobilla.net/sw/serd/test/'
-    test_suite(tst, serd_base + 'good/', 'good', None, 'Turtle')
-    test_suite(tst, serd_base + 'bad/', 'bad', None, 'Turtle')
-    test_suite(tst, serd_base + 'lax/', 'lax', None, 'Turtle', ['-l'])
-    test_suite(tst, serd_base + 'lax/', 'lax', None, 'Turtle')
+    run_test_suite = ['../test/run_test_suite.py', '--serdi', './serdi_static']
+
+    with tst.group('TestSuites') as check:
+        # Run serd-specific test suites
+        serd_base = 'http://drobilla.net/sw/serd/test/'
+        check(run_test_suite + ['../test/good/manifest.ttl', serd_base + 'good/'])
+        check(run_test_suite + ['../test/bad/manifest.ttl', serd_base + 'bad/'])
+        check(run_test_suite + ['../test/lax/manifest.ttl', serd_base + 'lax/', '--', '-l'])
+        check(run_test_suite + ['../test/lax/manifest.ttl', serd_base + 'lax/'])
 
-    # Standard test suites
-    with open('earl.ttl', 'w') as report:
-        report.write('@prefix earl: <http://www.w3.org/ns/earl#> .\n'
-                     '@prefix dc: <http://purl.org/dc/elements/1.1/> .\n')
+        # Start test report for standard test suites
+        report_filename = 'earl.ttl'
+        with open(report_filename, 'w') as report:
+            report.write('@prefix earl: <http://www.w3.org/ns/earl#> .\n'
+                         '@prefix dc: <http://purl.org/dc/elements/1.1/> .\n')
 
-        with open(os.path.join(srcdir, 'serd.ttl')) as serd_ttl:
-            report.writelines(serd_ttl)
+            with open(os.path.join(srcdir, 'serd.ttl')) as serd_ttl:
+                report.writelines(serd_ttl)
 
+        # Run standard test suites
         w3c_base = 'http://www.w3.org/2013/'
-        test_suite(tst, w3c_base + 'TurtleTests/',
-                   'TurtleTests', report, 'Turtle')
-        test_suite(tst, w3c_base + 'NTriplesTests/',
-                   'NTriplesTests', report, 'NTriples')
-        test_suite(tst, w3c_base + 'NQuadsTests/',
-                   'NQuadsTests', report, 'NQuads')
-        test_suite(tst, w3c_base + 'TriGTests/',
-                   'TriGTests', report, 'Trig', ['-a'])
+
+        check(run_test_suite + [
+            '--syntax', 'Turtle',
+            '--report', report_filename,
+            '../test/TurtleTests/manifest.ttl', w3c_base + 'TurtleTests/'])
+
+        check(run_test_suite + [
+            '--syntax', 'NTriples',
+            '--report', report_filename,
+            '../test/NTriplesTests/manifest.ttl', w3c_base + 'NTriplesTests/'])
+
+        check(run_test_suite + [
+            '--syntax', 'NQuads',
+            '--report', report_filename,
+            '../test/NQuadsTests/manifest.ttl', w3c_base + 'NQuadsTests/'])
+
+        check(run_test_suite + [
+            '--syntax', 'TriG',
+            '--report', report_filename,
+            '../test/TriGTests/manifest.ttl', w3c_base + 'TriGTests/',
+            '--', '-a'])