tests/integration: small Py testdriver (cmocka style)

tests/pydnstest/scenario.py

-import traceback
 import dns.message
 import dns.rrset
 import dns.rcode
@@ -53,7 +52,7 @@ class Entry:
             try:
                 self.match_part(code, msg)
             except Exception as e:
-                raise Exception("when matching %s: %s" % (code, str(e)))
+                raise Exception("%s: %s" % (code, str(e)))
 
     def set_match(self, fields):
         """ Set conditions for message comparison [all, flags, question, answer, authority, additional] """
@@ -100,7 +99,7 @@ class Entry:
         elif self.section == 'ADDITIONAL':
             self.message.additional.append(rr)
         else:
-            raise Exception('attempted to add record in section %s' % self.section)
+            raise Exception('bad section %s' % self.section)
 
 
     def __rr_from_str(self, owner, args):
@@ -192,8 +191,7 @@ class Step:
         elif self.type == 'TIME_PASSES':
             return self.__time_passes(ctx)
         else:
-            print('%d %s (%d entries) => NOOP' % (self.id, self.type, len(self.data)))
-            return None
+            raise Exception('step %s unsupported' % self.type)
 
     def __check_answer(self, ctx):
         """ Compare answer from previously resolved query. """
@@ -249,6 +247,6 @@ class Scenario:
                 self.current_step = step
                 step.play(ctx)
         except Exception as e:
-            raise Exception('on step #%d "%s": %s\n%s' % (step.id, step.type, str(e), traceback.format_exc()))
+            raise Exception('step #%d %s' % (step.id, str(e)))
 
 

tests/pydnstest/test.py

+#!/usr/bin/env python
+
+class Test:
+    """ Small library to imitate CMocka output. """
+
+    def __init__(self):
+        self.tests = []
+
+    def add(self, name, test, *args):
+        """ Add named test to set. """
+        self.tests.append((name, test, args))
+
+    def run(self):
+        """ Run planned tests. """
+        planned = len(self.tests)
+        passed = 0
+
+        print('[==========] Running %d test(s).' % planned)
+        for name, test_callback, args in self.tests:
+            print('[ RUN      ] %s' % name)
+            try:
+                test_callback(*args)
+                passed += 1
+                print('[       OK ] %s' % name)
+            except Exception as e:
+                print('[     FAIL ] %s (%s)' % (name, str(e)))
+
+        print('[==========] %d test(s) run.' % planned)
+        if passed == planned:
+            print('[  PASSED  ] %d test(s).' % passed)
+            return 0
+        else:
+            print('[  FAILED  ] %d test(s).' % (planned - passed))
+            return 1

tests/pydnstest/testserver.py

 import select, socket, threading, struct, sys, os
 import dns.message
+import test
 
 def recv_message(stream):
     """ Receive DNS/TCP message. """
@@ -100,9 +101,8 @@ class TestServer:
             address = (address, 0)
         return address
 
-def module_test():
+def test_sendrecv():
     """ Module self-test code. """
-    result = 0
     server = TestServer(None)
     client = server.client()
     server.start()
@@ -114,20 +114,17 @@ def module_test():
             raise Exception('no answer received')
         if not query.is_response(answer):
             raise Exception('not a mirror response')
-        print('[ OK ] testserver')
-    except Exception as e:
-        print('[FAIL] testserver %s' % str(e))
-        result = 1
     finally:
         client.close()
         server.stop()
-        return result
 
 if __name__ == '__main__':
 
     # Self-test code
     if '--test' in sys.argv:
-        sys.exit(module_test())
+        test = test.Test()
+        test.add('testserver/sendrecv', test_sendrecv)
+        sys.exit(test.run())
 
     # Mirror server
     server = TestServer(None, socket.AF_INET, '127.0.0.1')

tests/test_integration.py

 #!/usr/bin/env python
 import sys, os, fileinput
-from pydnstest import scenario, testserver
+from pydnstest import scenario, testserver, test
 import _test_integration as mock_ctx
 
 # Test debugging
@@ -99,13 +99,15 @@ def parse_file(file_in):
         raise Exception('line %d: %s' % (file_in.lineno(), str(e)))
 
 
-def parse_object(path):
+def find_objects(path):
     """ Recursively scan file/directory for scenarios. """
+    result = []
     if os.path.isdir(path):
         for e in os.listdir(path):
-            parse_object(os.path.join(path, e))
+            result += find_objects(os.path.join(path, e))
     elif os.path.isfile(path):
-        play_object(path)
+        result.append(path)
+    return result
 
 
 def play_object(path):
@@ -116,52 +118,48 @@ def play_object(path):
     scenario = None
     try:
         scenario = parse_file(file_in)
-    except Exception as e:
-        print('%s %s' % (os.path.basename(path), str(e)))
-    file_in.close()
-    if scenario is None:
-        return
+    finally:
+        file_in.close()
 
     # Play scenario
     server = testserver.TestServer(scenario)
     server.start()
     mock_ctx.init()
-    mock_ctx.set_server(server)
     try:
+        mock_ctx.set_server(server)
         if TEST_DEBUG > 0:
-            print('--- server listening at %s ---' % str(server.server.server_address))
+            print('--- server listening at %s ---' % str(server.address()))
             print('--- scenario parsed, any key to continue ---')
             sys.stdin.readline()
         scenario.play(mock_ctx)
-        print('%s OK' % os.path.basename(path))
-    except Exception as e:
-        print('%s %s' % (os.path.basename(path), str(e)))
     finally:
         server.stop()
         mock_ctx.deinit()
 
-def module_test():
+def test_ipc(*args):
+    for arg in args:
+        print arg
     """ Module self-test code. """
-    result = 0
     server = testserver.TestServer(None)
     server.start()
     mock_ctx.set_server(server)
     try:
         mock_ctx.test_connect()
-        print('[ OK ] test connection')
-    except Exception as e:
-        print('[FAIL] test connection: %s' % str(e))
-        result = 1
     finally:
         server.stop()
-        return result
 
 if __name__ == '__main__':
 
+    test = test.Test()
+
     # Self-test code
     if '--test' in sys.argv:
-        sys.exit(module_test())
+        test.add('integration/ipc', test_ipc)
+    else:
+        # Scan for scenarios
+        for arg in sys.argv[1:]:
+            objects = find_objects(arg)
+            for path in objects:
+                test.add(path, play_object, path)
 
-    # Process path
-    for arg in sys.argv[1:]:
-        parse_object(arg)
+    sys.exit(test.run())