winbindd: Retry on expired session in cm_connect_lsa

[obnox/samba/samba-obnox.git] / selftest / testlist.py

diff --git a/selftest/testlist.py b/selftest/testlist.py
index e5351945102dc51446745e68e8c9bedd7b556006..34c7a5d31c953b6ec2424c8e3c4aeeafd7b61b6e 100644 (file)
--- a/selftest/testlist.py
+++ b/selftest/testlist.py
@@ -19,9 +19,11 @@
 
 """Selftest test list management."""
 
-__all__ = ['find_in_list', 'read_test_regexes']
+__all__ = ['find_in_list', 'read_test_regexes', 'read_testlist']
 
+import os
 import re
+import sys
 
 def find_in_list(list, fullname):
     """Find test in list.
@@ -53,3 +55,117 @@ def read_test_regexes(f):
             yield l, None
         else:
             yield test.strip(), reason.strip()
+
+
+def should_run_test(tests, name):
+    if tests == []:
+        return True
+    for test in tests:
+        if re.match(test, name):
+            return True
+    return False
+
+
+def read_testlist(inf, outf):
+    """Read a list of tests from a file.
+
+    :param inf: File-like object to read from.
+    :param outf: File-like object to write to.
+    :return: Iterator over tuples describing tests
+    """
+    while True:
+        l = inf.readline()
+        if l == '':
+            return
+        if l.startswith("-- TEST") and l.endswith(" --\n"):
+            supports_loadlist = l.startswith("-- TEST-LOADLIST")
+            name = inf.readline().rstrip("\n")
+            env = inf.readline().rstrip("\n")
+            if supports_loadlist:
+                loadlist = inf.readline().rstrip("\n")
+            else:
+                loadlist = None
+            cmdline = inf.readline().rstrip("\n")
+            yield (name, env, cmdline, loadlist)
+        else:
+            outf.write(l)
+
+
+def read_restricted_test_list(f):
+    for l in f.readlines():
+        yield l.strip()
+
+
+class RestrictedTestManager(object):
+    """Test manager which can filter individual tests that should be run."""
+
+    def __init__(self, test_list):
+        self.test_list = test_list
+        self.unused = set(self.test_list)
+
+    @classmethod
+    def from_path(cls, path):
+        f = open(path, 'r')
+        try:
+            return cls(read_restricted_test_list(f))
+        finally:
+            f.close()
+
+    def should_run_testsuite(self, name):
+        """Determine whether a testsuite should be run.
+
+        :param name: Name of the testsuite
+        :return: None if full testsuite should be run,
+            a list of subtests to run or [] if it should
+            not be run.
+        """
+        match = []
+        for r in self.test_list:
+            if r == name:
+                match = None
+                if r in self.unused:
+                    self.unused.remove(r)
+            elif r.startswith(name + "."):
+                if match is not None:
+                    match.append(r[len(name+"."):])
+                if r in self.unused:
+                    self.unused.remove(r)
+        return match
+
+    def iter_unused(self):
+        """Iterate over entry entries that were unused.
+
+        :return: Iterator over test list entries that were not used.
+        """
+        return iter(self.unused)
+
+
+def open_file_or_pipe(path, mode):
+    """Open a file or pipe.
+
+    :param path: Path to open; if it ends with | it is assumed to be a
+        command to run
+    :param mode: Mode with which to open it
+    :return: File-like object
+    """
+    if path.endswith("|"):
+        return os.popen(path[:-1], mode)
+    return open(path, mode)
+
+
+def read_testlist_file(fn, outf=None):
+    """Read testlist file.
+
+    :param fn: Path to read (assumed to be a command to run if it ends with |)
+    :param outf: File-like object to pass non-test data through to
+        (defaults to stdout)
+    :return: Iterator over test suites (see read_testlist)
+    """
+    if outf is None:
+        outf = sys.stdout
+    inf = open_file_or_pipe(fn, 'r')
+    try:
+        for testsuite in read_testlist(inf, outf):
+            yield testsuite
+    finally:
+        inf.close()