def _report_experiment()

in ftw_compatible_tool/base.py [0:0]
57 lines of code
13 McCabe index (conditional complexity)

    def _report_experiment(self):
        def check_http_code(item, http_content):
            if not http_content:
                return False
            result = re.search(r"HTTP[\S]+ (\d{3})", http_content)
            if not result:
                return False
            http_code = result.group(1)
            if isinstance(item, list):
                item = list(map(str, item))
                return http_code in item
            else:
                return str(item) == http_code

        def regex_match(item, value):
            return bool(value) and bool(re.search(unicode(item), value))

        def regex_not_match(item, value):
            return not bool(regex_match(item, value))

        check_items = {
            "status": ("raw_response", check_http_code ),
            "log_contains": ("raw_log", regex_match),
            "no_log_contains": ("raw_log", regex_not_match),
            "response_contains": ("raw_response", regex_match),
            "html_contains": ("raw_response", regex_match),
            "expect_error":
            ("raw_response",
             lambda item, value: not (bool(item) and bool(value))),
        }

        def report_experiment(query_result):
            for row in query_result:
                row = collections.OrderedDict(zip(query_result.title(), row))
                # this test not be sent
                if row["raw_request"] is None:
                    continue
                check_result = {}
                try:
                    row["output"] = ast.literal_eval(row["output"])
                    for k, v in row["output"].items():
                        if k not in check_items:
                            continue
                        check_item = check_items[k]
                        check_result[k] = check_item[1](v, row[check_item[0]])
                    self._ctx.broker.publish(
                        broker.TOPICS.SQL_COMMAND,
                        sql.SQL_UPDATE_TESTING_RESULT,
                        str(bool(not check_result \
                        or reduce(lambda x, y: x and y, check_result.values()))),
                        row["traffic_id"]
                        )
                    self._ctx.broker.publish(broker.TOPICS.CHECK_RESULT, row,
                                             check_result)
                except (re.error, SyntaxError) as e:
                    self._ctx.broker.publish(broker.TOPICS.CHECK_RESULT, row,
                                             check_result)
                    self._ctx.broker.publish(broker.TOPICS.WARNING, str(e))

        self._ctx.broker.publish(
            broker.TOPICS.SQL_COMMAND,
            sql.SQL_QUERY_RESULT,
            callback=report_experiment)