run-tests

#! /usr/bin/env python3
#
# This file is part of the Green End SFTP Server.
# Copyright (C) 2007, 2011, 2016, 2018 Richard Kettlewell
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
# General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307
# USA

import os
import sys
import re
import string
from subprocess import Popen, PIPE, STDOUT

srcdir = os.path.abspath(os.getenv('srcdir'))
builddir = os.path.abspath('.')
client = os.path.abspath("sftpclient")
server = "gesftpserver"


def rmrf(path):
    """Remove path and everything below it

    shutil.rmtree isn't suitable because it's not aggressive enough about
    permissions."""
    if os.path.lexists(path):
        if (not os.path.islink(path)) and os.path.isdir(path):
            os.chmod(path, 0o700)
            for name in os.listdir(path):
                rmrf(os.path.join(path, name))
            os.rmdir(path)
        else:
            os.remove(path)


def fatal(msg):
    """Report an error and exist with nonzero status"""
    sys.stderr.write("%s\n" % msg)
    sys.exit(1)


if os.getuid() == 0:
    fatal("These tests are not suitable for running as root.")

os.umask(0o22)                           # for consistent permissions
failed = 0
protocols = ['2', '3', '4', '5', '6', '7']
dir = 'tests'
debug = 'DEBUG' in os.environ
failfast = 'FAILFAST' in os.environ
threads = None  # use default
reorder = True

args = sys.argv[1:]
while len(args) > 0 and args[0][0] == '-':
    if args[0] == "--protocols":
        protocols = args[1].split(',')
        args = args[2:]
    elif args[0] == "--server":
        server = args[1]
        args = args[2:]
    elif args[0] == "--client":
        client = args[1]
        args = args[2:]
    elif args[0] == "--directory":
        dir = args[1]
        args = args[2:]
    elif args[0] == "--debug":
        debug = True
        args = args[1:]
    elif args[0] == "--threads":
        threads = int(args[1])
        args = args[2:]
    elif args[0] == "--no-reorder":
        reorder = False
        args = args[1:]
    elif args[0] == "--fail-fast":
        failfast = True
        args = args[1:]
    else:
        fatal("unknown option '%s'" % args[0])

server = os.path.abspath(server)
client = os.path.abspath(client)

if len(args) > 0:
    tests = args
else:
    tests = os.listdir(os.path.join(srcdir, dir))
# Execute tests in a consistent order
tests.sort()

# Clean up
rmrf(os.path.join(builddir, ',testroot'))

# Colorize output (if possible)


def colored(text, color=None, on_color=None, attrs=None):
    """Uncolorized output"""
    return text


if sys.stdout.isatty():
    try:
        # apt-get install python3-termcolor
        import termcolor
        colored = termcolor.colored
    except:
        pass

for test in tests:
    for proto in protocols:
        # Skip dotfiles, backup files, and tests that don't apply to the
        # current protocol
        if ('.' in test
            or not proto in test
            or '#' in test
                or '~' in test):
            continue
        sys.stderr.write("Testing %s/%s protocol %s ... " % (dir, test, proto))
        # Make a working directory for the test
        root = os.path.join(builddir, ',testroot', '%s.%s' % (test, proto))
        os.makedirs(root)
        os.chdir(root)
        # Make a config file if necessary
        config = "/dev/null"
        if (threads is not None) or not reorder:
            config = os.path.join(builddir, ',testroot', 'gesftpserver.conf')
            with open(config, "w") as f:
                if threads is not None:
                    print(f"threads {threads}", file=f)
                if reorder:
                    print("reorder true", file=f)
                else:
                    print("reorder false", file=f)
        # Run the client with the server as a child process
        clientcmd = [client,
                     "--force-version", proto,
                     "-P", server,
                     "-b", os.path.join(srcdir, dir, test),
                     "--program-config", config,
                     '--echo',
                     '--fix-sigpipe',   # stupid Python
                     '--no-stop-on-error']
        if debug:
            debug_path = "%s/%s-%s.debug" % (builddir, test, proto)
            # Save server debug output
            clientcmd += ["--program-debug-path", debug_path]
        # Run the client
        output = Popen(clientcmd,
                       stdout=PIPE,
                       stderr=STDOUT).communicate()[0].split(b'\n')
        # Get the output, with newlines stripped
        output = [str(s, "ASCII") for s in output]
        if output[len(output)-1] == "":
            output = output[:-1]
        # Get the script, with newlines stripped
        with open(os.path.join(srcdir, dir, '%s' % test), "r") as f:
            script = [l[:-1] for l in f]
        # Find the maximum line length in each
        omax = max([len(l) for l in output])
        smax = max([len(l) for l in script])
        errors = 0
        report = []
        # Iterate over the lines, checking them and formatting the results
        for lineno in range(0, max(len(output), len(script))):
            if lineno >= len(output):
                # Truncated output is bad
                ok = False
                sline = script[lineno]
                oline = ""
            elif lineno >= len(script):
                # Excessive output is bad
                ok = False
                sline = ""
                oline = output[lineno]
            else:
                sline = script[lineno]
                oline = output[lineno]
                if len(sline) > 0 and sline[0] == '#':
                    # Lines starting # are regexps matching expected output
                    pattern = sline[1:]
                    ok = re.match(pattern, oline)
                else:
                    # Other lines are the commands, which we expect to be
                    # echoed back.
                    ok = (oline == sline)
            if ok:
                marker = " "
                def f(t): return t
            else:
                # Mismatching lines are marked with a * and colored red (if
                # possible)
                marker = "*"
                errors += 1
                def f(t): return colored(t, "red")
            report.append(f("%s %-*s %-*s\n" %
                          (marker, smax, sline, omax, oline)))
        if errors > 0:
            sys.stderr.write(colored("FAILED\n", "red"))
            sys.stderr.write("Command:\n  %s\n" % " ".join(clientcmd))
            sys.stderr.write("Output discrepancies:\n")
            for l in report:
                sys.stderr.write(l)
            if debug:
                sys.stderr.write("Server debug:\n")
                with open(debug_path, "r") as f:
                    for l in f.readlines():
                        sys.stderr.write(l)
            failed += 1
            if failfast:
                break
        else:
            sys.stderr.write(colored("passed\n", "green"))
    if failed > 0 and failfast:
        break

if failed:
    print(colored("%d tests failed" % failed, "red"))
    sys.exit(1)
else:
    # On success, delete the test root. It's preserved for inspection on
    # failure.
    os.chdir("/")
    rmrf(os.path.join(builddir, ',testroot'))
    print(colored("OK", "green"))

# Local Variables:
# mode:python
# indent-tabs-mode:nil
# py-indent-offset:4
# End: