Improved test support

ftw/http.py

@@ -19,10 +19,6 @@
 from . import util
 
 
-# Fallback to PROTOCOL_SSLv23 if PROTOCOL_TLS is not available.
-PROTOCOL_TLS = getattr(ssl, 'PROTOCOL_TLS', ssl.PROTOCOL_SSLv23)
-
-
 class HttpResponse(object):
     def __init__(self, http_response, user_agent):
         self.response = util.ensure_binary(http_response)
@@ -299,8 +295,9 @@ def build_socket(self):
             self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
             # Check if TLS
             if self.request_object.protocol == 'https':
-                context = ssl.SSLContext(PROTOCOL_TLS)
+                context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
                 context.set_ciphers(self.CIPHERS)
+                context.load_default_certs(ssl.Purpose.SERVER_AUTH)
                 self.sock = context.wrap_socket(
                     self.sock, server_hostname=self.request_object.dest_addr)
             self.sock.connect(

ftw/logchecker.py

@@ -15,6 +15,20 @@ def set_times(self, start, end):
         self.start = start
         self.end = end
 
+    def mark_start(self, stage_id):
+        """
+        May be implemented to set up the log checker before
+        the request is being sent
+        """
+        pass
+
+    def mark_end(self, stage_id):
+        """
+        May be implemented to tell the log checker that
+        the response has been received
+        """
+        pass
+
     @abstractmethod
     def get_logs(self):
         """

ftw/pytest_plugin.py

@@ -7,7 +7,7 @@
 from .ruleset import Test
 
 
-def get_testdata(rulesets):
+def get_testdata(rulesets, use_rulesets):
     """
     In order to do test-level parametrization (is this a word?), we have to
     bundle the test data from rulesets into tuples so py.test can understand
@@ -17,7 +17,10 @@ def get_testdata(rulesets):
     for ruleset in rulesets:
         for test in ruleset.tests:
             if test.enabled:
-                testdata.append((ruleset, test))
+                args = [test]
+                if use_rulesets:
+                    args = [rulesets] + args
+            testdata.append(args)
 
     return testdata
 
@@ -127,7 +130,13 @@ def pytest_generate_tests(metafunc):
                 metafunc.config.option.ruledir_recurse, True)
         if metafunc.config.option.rule:
             rulesets = util.get_rulesets(metafunc.config.option.rule, False)
-        if 'ruleset' in metafunc.fixturenames and \
-           'test' in metafunc.fixturenames:
-            metafunc.parametrize('ruleset, test', get_testdata(rulesets),
-                                 ids=test_id)
+        if 'test' in metafunc.fixturenames:
+            use_rulesets = False
+            arg_names = ['test']
+            if 'ruleset' in metafunc.fixturenames:
+                use_rulesets = True
+                arg_names = ['ruleset'] + arg_names
+            metafunc.parametrize(
+                arg_names,
+                get_testdata(rulesets, use_rulesets),
+                ids=test_id)

ftw/ruleset.py

@@ -137,18 +137,26 @@ class Stage(object):
     This class holds information about 1 stage in a test, which contains
     1 input and 1 output
     """
-    def __init__(self, stage_dict):
+    def __init__(self, stage_dict, stage_index, test):
         self.stage_dict = stage_dict
+        self.stage_index = stage_index
+        self.test = test
         self.input = Input(**stage_dict['input'])
         self.output = Output(stage_dict['output'])
+        self.id = self.build_id()
+
+    def build_id(self):
+        rule_name = self.test.ruleset_meta["name"].split('.')[0]
+        return f'{rule_name}-{self.test.test_index}-{self.stage_index}'
 
 
 class Test(object):
     """
     This class holds information for 1 test and potentially many stages
     """
-    def __init__(self, test_dict, ruleset_meta):
+    def __init__(self, test_dict, test_index, ruleset_meta):
         self.test_dict = test_dict
+        self.test_index = test_index
         self.ruleset_meta = ruleset_meta
         self.test_title = self.test_dict['test_title']
         self.stages = self.build_stages()
@@ -160,8 +168,8 @@ def build_stages(self):
         """
         Processes and loads an array of stages from the test dictionary
         """
-        return [Stage(stage_dict['stage'])
-                for stage_dict in self.test_dict['stages']]
+        return [Stage(stage_dict['stage'], index, self)
+                for index, stage_dict in enumerate(self.test_dict['stages'])]
 
 
 class Ruleset(object):
@@ -183,8 +191,8 @@ def extract_tests(self):
         creates test objects based on input
         """
         try:
-            return [Test(test_dict, self.meta)
-                    for test_dict in self.yaml_file['tests']]
+            return [Test(test_dict, index, self.meta)
+                    for index, test_dict in enumerate(self.yaml_file['tests'])]
         except errors.TestError as e:
             e.args[1]['meta'] = self.meta
             raise e

ftw/testrunner.py

@@ -174,7 +174,6 @@ def run_stage(self, stage, logger_obj=None, http_ua=None):
         input, waits for output then compares expected vs actual output
         http_ua can be passed in to persist cookies
         """
-
         # Send our request (exceptions caught as needed)
         if stage.output.expect_error:
             with pytest.raises(errors.TestError) as excinfo:
@@ -187,11 +186,20 @@ def run_stage(self, stage, logger_obj=None, http_ua=None):
         else:
             if not http_ua:
                 http_ua = http.HttpUA()
-            start = datetime.datetime.utcnow()
+            if ((stage.output.log_contains_str or
+                    stage.output.no_log_contains_str) and
+                    logger_obj is not None):
+                logger_obj.mark_start(stage.id)
+                start = datetime.datetime.utcnow()
             http_ua.send_request(stage.input)
-            end = datetime.datetime.utcnow()
-        if (stage.output.log_contains_str or
-           stage.output.no_log_contains_str) and logger_obj is not None:
+            if ((stage.output.log_contains_str or
+                    stage.output.no_log_contains_str) and
+                    logger_obj is not None):
+                logger_obj.mark_end(stage.id)
+                end = datetime.datetime.utcnow()
+        if ((stage.output.log_contains_str or
+                stage.output.no_log_contains_str) and
+                logger_obj is not None):
             logger_obj.set_times(start, end)
             lines = logger_obj.get_logs()
             if stage.output.log_contains_str:

test/integration/test_logcontains.py

@@ -27,13 +27,13 @@ def logchecker_obj():
     return LoggerTestObj()
 
 
-def test_logcontains_withlog(logchecker_obj, ruleset, test):
+def test_logcontains_withlog(logchecker_obj, test):
     runner = testrunner.TestRunner()
     for stage in test.stages:
         runner.run_stage(stage, logchecker_obj)
 
 
-def test_logcontains_nolog(logchecker_obj, ruleset, test):
+def test_logcontains_nolog(logchecker_obj, test):
     logchecker_obj.do_nothing = True
     runner = testrunner.TestRunner()
     with(pytest.raises(AssertionError)):

test/unit/test_ruleset.py

@@ -33,11 +33,12 @@ def test_input():
 
 def test_testobj():
     with pytest.raises(KeyError) as excinfo:
-        ruleset.Test({}, {})
+        ruleset.Test({}, {}, {})
     assert 'test_title' in str(excinfo.value)
+    ruleset_meta = {'name': 'test-name.yaml'}
     stages_dict = {'test_title': 1, 'stages': [{'stage':
                    {'output': {'log_contains': 'foo'}, 'input': {}}}]}
-    ruleset.Test(stages_dict, {})
+    ruleset.Test(stages_dict, {}, ruleset_meta)
 
 
 def test_ruleset():