Adding JUnit support

pyresttest/macros.py

@@ -36,6 +36,10 @@ def simple_print(self, x):
         if x: 
             print(x)
 
+    # Called at the begining and end of the test suite
+    def start_testset(self, input): lambda x: simple_print(x)
+    def end_testset(self, input): lambda x: simple_print(x)
+
     # Logging outputs, these are part of the lifecycle    
     def start_macro(self, input): lambda x: simple_print(x)
     def pre_request(self, input): lambda x: simple_print(x)  # Called just before submitting requests
@@ -59,6 +63,10 @@ class TestSetConfig(object):
     verbose = False
     ssl_insecure = False
     skip_term_colors = False  # Turn off output term colors
+    junit = False # Write junit output
+    junit_path = None # Path to write junit file
+    working_directory = None # Working directory
+    name = '' # TestSetName
 
     # Binding and creation of generators
     variable_binds = None
@@ -198,5 +206,7 @@ def parse_configuration(node, base_config=None):
                 gen = parse_generator(generator_config)
                 gen_map[str(generator_name)] = gen
             testset_config.generators = gen_map
+        elif key == u'testset':
+            testset_config.name = str(value)            
 
-    return testset_config
+    return testset_config

pyresttest/parsing.py

@@ -129,6 +129,10 @@ def parse_command_line_args(args_in):
                       action="store_true", dest="absolute_urls")
     parser.add_option(u'--skip_term_colors', help='Turn off the output term colors',
                       action='store_true', default=False, dest="skip_term_colors")
+    parser.add_option(u'--junit', help='Enable junit output',
+                      action="store_true", default=False, dest="junit")
+    parser.add_option(u'--junit-path', help='Set path to junit ouput (when junit is enabled)',
+                      action='store', type="string", dest="junit_path")
 
     (args, unparsed_args) = parser.parse_args(args_in)
     args = vars(args)
@@ -149,4 +153,4 @@ def parse_command_line_args(args_in):
 
     # So modules can be loaded from current folder
     args['cwd'] = os.path.realpath(os.path.abspath(os.getcwd()))
-    return args
+    return args

pyresttest/resttest.py

@@ -5,6 +5,8 @@
 import yaml
 import pycurl
 import logging
+import re
+from xml.etree import cElementTree as ET # For JUnit output
 
 # Python 3 compatibility
 if sys.version_info[0] > 2:
@@ -169,24 +171,177 @@ class LoggerCallbacks(MacroCallbacks):
     """ Uses a standard python logger """
     def log_status(self, input):
         logger.info(str(input))
-    def log_intermediate(self, input): 
+    def log_intermediate(self, input):
         logger.debug(str(input))
     def log_failure(self, input):
         logger.error(str(input))
     def log_success(self, input):
         logger.info(str(input))
 
+
+class JUnitCallback(MacroCallbacks):
+    """ Uses junit standard xml output """
+
+    def __init__(self):
+        self.el_test_suites = None
+        self.test_suite_current_id = 0
+        self.group_test_suite_map = None
+        self.working_directory = os.path.abspath(os.getcwd())
+        self.path = 'test-results.xml'
+
+    def start_testset(self, input):
+        self.el_test_suites = ET.Element('testsuites')
+        self.el_test_suites.set('name', self.camelizeStr(str(input)))
+        self.test_suite_current_id = 0
+        self.group_test_suite_map = dict()
+
+    def end_testset(self, input):
+        self.write_file(self.el_test_suites)
+
+    def log_status(self, input):
+        logger.info(str(input))
+
+    def log_intermediate(self, input):
+        logger.debug("LOGGER INTERMEDIATE: " + str(input))
+
+    def log_failure(self, input):
+        if isinstance(input, TestResponse):
+            el_test_suite = self.get_test_suite(input.test.group)
+            try:
+                num_tests = int(el_test_suite.get('tests', '0'))
+            except ValueError:
+                num_tests = 0
+            el_test_suite.set('tests', str(num_tests + 1))
+            try:
+                num_failures = int(el_test_suite.get('failures', '0'))
+            except ValueError:
+                num_failures = 0
+            el_test_suite.set('failures', str(num_failures + 1))
+            el_test_case = self.start_test_case(el_test_suite, input.test, "Ko")
+            failure_messages = []
+            for idx, failure in enumerate(input.failures):
+                el_failure = ET.SubElement(el_test_case, 'failure')
+                if failure.message:
+                    el_failure.set('message', failure.message)
+                if failure.failure_type:
+                    el_failure.set('type', str(failure.failure_type))
+                if failure.details:
+                    failure_messages.append("\n\n====================================== FAILURE ")
+                    failure_messages.append(str(idx))
+                    failure_messages.append(" DETAILS ======================================\n")
+                    failure_messages.append(failure.details)
+            el_system_err = ET.SubElement(el_test_case, 'system-err')
+            el_system_err.text = ''.join(failure_messages)
+        else:
+            logger.error(str(input))
+
+    def log_success(self, input):
+        if isinstance(input, TestResponse):
+            el_test_suite = self.get_test_suite(input.test.group)
+            try:
+                num_tests = int(el_test_suite.get('tests', '0'))
+            except ValueError:
+                num_tests = 0
+            el_test_suite.set('tests', str(num_tests + 1))
+            el_test_case = self.start_test_case(el_test_suite, input.test, "Ok")
+        else:
+            logger.info(str(input))
+
+    def get_test_suite(self, group_name):
+        """ Return the test suite for group_name. If it does'nt exist, it will be created. """
+        if group_name in self.group_test_suite_map.keys():
+            el_test_suite = self.group_test_suite_map[group_name]
+        else:
+            el_test_suite = ET.SubElement(self.el_test_suites, 'testsuite')
+            el_test_suite.set('id', str(self.test_suite_current_id))
+            self.test_suite_current_id += 1
+            suite_name = self.aggregate_name(self.el_test_suites.get('name',''), self.camelizeStr(group_name))
+            el_test_suite.set('name', suite_name)
+            el_test_suite.set('failures', '0')
+            self.group_test_suite_map[group_name] = el_test_suite
+        return el_test_suite
+
+    def start_test_case(self, el_suite, test, status):
+        """ Start a test case and return the Element """
+        el_test_case = ET.SubElement(el_suite,'testcase')
+        el_test_case.set('name', test.name)
+        num_assertion = 1 # At least one assertion least on status
+        if test.validators:
+            num_assertion += len(test.validators)
+        el_test_case.set('assertions', str(num_assertion))
+        testcase_classname = self.aggregate_name(el_suite.get('name',''), self.camelizeStr(test.name))
+        el_test_case.set('classname', testcase_classname)
+        el_test_case.set('status', status)
+        return el_test_case
+
+    def set_o_path(self, path, default_name='test-results.xml'):
+        """ Set output path, where the JUnit results willbe written
+        If path is incorrect (directory does not exists), an error is logged and the path stay unchanged.
+
+        path -- the path where to write JUnit output
+        default_name -- the default name of the file, if not present in the path (default 'test-results.xml')
+        """
+        with cd(self.working_directory):
+            if os.path.isdir(path):
+                self.path = os.path.join(path, default_name) # Default file name
+            else:
+                dir_path, filename = os.path.split(path)
+                if not os.path.isdir(dir_path): # The directory does not exit, log error
+                    logger.error('JUnit Error: ouput dir {0} does not exist. File will be writed to default path ({1}).'.format(dir_path, self.path))
+                else:
+                    self.path = path
+
+    def camelizeStr(self, mystr):
+        """ Return a string formatted to camelCase """
+        camelized = ''
+        if mystr:
+            pattern = re.compile('[\W_]+')
+            camelized = pattern.sub('', mystr.title())
+            camelized = camelized[0].lower() + camelized[1:]
+        return camelized
+
+    def aggregate_name(self, *args):
+        """ Aggrgate name to format a java-like package name """
+        return ".".join(args)
+
+    def set_working_directory(self, dir_path):
+        if os.path.isdir(dir_path):
+            self.working_directory = dir_path
+        else:
+            logger.error('Junit Error: setting working dir to {0} : directory does not exist.'.format(dir_path))
+
+    def write_file(self, root):
+        """ Write root elemet to file """
+        tree = ET.ElementTree(root)
+        with cd(self.working_directory):
+            logger.debug("Writing junit output to: {0}".format(self.path))
+            tree.write(self.path, encoding="UTF-8", xml_declaration=True)
+
+
 def run_testsets(testsets):
     """ Execute a set of tests, using given TestSet list input """
     group_results = dict()  # results, by group
     group_failure_counts = dict()
     total_failures = 0
     myinteractive = False
     curl_handle = pycurl.Curl()
-
+    myconfig = TestSetConfig()
+    if len(testsets) > 0:
+        myconfig = testsets[0].config
+    testset_name = myconfig.name
+
     # Invoked during macro execution to report results
     # FIXME  I need to set up for logging before/after/during requests
-    callbacks = LoggerCallbacks()
+    if myconfig.junit:
+        callbacks = JUnitCallback()
+        if myconfig.working_directory is not None:
+            callbacks.set_working_directory(myconfig.working_directory)
+        if myconfig.junit_path is not None:
+            callbacks.set_o_path(myconfig.junit_path)
+    else:
+        callbacks = LoggerCallbacks()
+
+    callbacks.start_testset(testset_name)
 
     for testset in testsets:
         mytests = testset.tests
@@ -283,16 +438,17 @@ def run_testsets(testsets):
         total_failures = total_failures + failures
 
         passfail = {True: u'SUCCEEDED: ', False: u'FAILED: '}
-        output_string = "Test Group {0} {1}: {2}/{3} Tests Passed!".format(group, passfail[failures == 0], str(test_count - failures), str(test_count)) 
-        
+        output_string = "Test Group {0} {1}: {2}/{3} Tests Passed!".format(group, passfail[failures == 0], str(test_count - failures), str(test_count))
+
         if myconfig.skip_term_colors:
-            print(output_string)    
+            print(output_string)
         else:
             if failures > 0:
                 print('\033[91m' + output_string + '\033[0m')
             else:
                 print('\033[92m' + output_string + '\033[0m')
 
+    callbacks.end_testset(testset_name)
     return total_failures
 
 
@@ -415,6 +571,13 @@ def main(args):
         if 'skip_term_colors' in args and args['skip_term_colors'] is not None:
             t.config.skip_term_colors = safe_to_bool(args['skip_term_colors'])
 
+        if 'junit' in args and args['junit'] is not None:
+            t.config.junit = safe_to_bool(args['junit'])
+            if 'junit_path' in args and args['junit_path'] is not None:
+                t.config.junit_path = args['junit_path']
+
+        t.config.working_directory = os.path.dirname(test_file)
+
     # Execute all testsets
     failures = run_testsets(tests)
 

pyresttest/tests.py

@@ -417,7 +417,10 @@ def execute_macro(self, testset_config=TestSetConfig(), context=None, cmdline_ar
             callbacks.log_status(result.response_headers)
 
         # TODO add string escape on body output
-        callbacks.log_intermediate(result)
+        if result.passed is True:
+            callbacks.log_success(result)
+        else:
+            callbacks.log_failure(result)
 
         return result
 
@@ -428,14 +431,14 @@ def configure_curl(self, timeout=DEFAULT_TIMEOUT, context=None, curl_handle=None
             curl = curl_handle
 
             try:  # Check the curl handle isn't closed, and reuse it if possible
-                curl.getinfo(curl.HTTP_CODE)                
+                curl.getinfo(curl.HTTP_CODE)
                 # Below clears the cookies & curl options for clean run
                 # But retains the DNS cache and connection pool
                 curl.reset()
                 curl.setopt(curl.COOKIELIST, "ALL")
             except pycurl.error:
                 curl = pycurl.Curl()
-            
+
         else:
             curl = pycurl.Curl()
 
@@ -454,8 +457,8 @@ def configure_curl(self, timeout=DEFAULT_TIMEOUT, context=None, curl_handle=None
             curl.setopt(curl.READFUNCTION, MyIO(bod).read)
 
         if self.auth_username and self.auth_password:
-            curl.setopt(pycurl.USERPWD, 
-                parsing.encode_unicode_bytes(self.auth_username) + b':' + 
+            curl.setopt(pycurl.USERPWD,
+                parsing.encode_unicode_bytes(self.auth_username) + b':' +
                 parsing.encode_unicode_bytes(self.auth_password))
             if self.auth_type:
                 curl.setopt(pycurl.HTTPAUTH, self.auth_type)
@@ -498,7 +501,7 @@ def configure_curl(self, timeout=DEFAULT_TIMEOUT, context=None, curl_handle=None
                 curl.setopt(pycurl.POSTFIELDSIZE, len(bod))
 
         # Template headers as needed and convert headers dictionary to list of header entries
-        
+
         head = self.get_headers(context=context)
         head = copy.copy(head)  # We're going to mutate it, need to copy