Class JsonFacade

tests_python/test_debugger_json.py:83–584

class JsonFacade(object):

    def __init__(self, writer):
        self.writer = writer
        if hasattr(writer, "reader_thread"):
            writer.reader_thread.accept_xml_messages = False
        self._all_json_messages_found = []
        self._sent_launch_or_attach = False

    def mark_messages(self, expected_class, accept_message=lambda obj: True):
        ret = []
        for message_with_mark in self._all_json_messages_found:
            if not message_with_mark.marked:
                if isinstance(message_with_mark.msg, expected_class):
                    if accept_message(message_with_mark.msg):
                        message_with_mark.marked = True
                        ret.append(message_with_mark.msg)
        return ret

    def wait_for_json_message(self, expected_class, accept_message=lambda obj: True):

        def accept_json_message(msg):
            if msg.startswith("{"):
                decoded_msg = from_json(msg)

                self._all_json_messages_found.append(_MessageWithMark(decoded_msg))

                if isinstance(decoded_msg, expected_class):
                    if accept_message(decoded_msg):
                        return True
            return False

        msg = self.writer.wait_for_message(accept_json_message, unquote_msg=False, expect_xml=False)
        return from_json(msg)

    def build_accept_response(self, request, response_class=None):
        if response_class is None:
            response_class = pydevd_base_schema.get_response_class(request)

        def accept_message(response):
            if isinstance(request, dict):
                if response.request_seq == request["seq"]:
                    return True
            else:
                if response.request_seq == request.seq:
                    return True
            return False

        return (response_class, Response), accept_message

    def wait_for_response(self, request, response_class=None):
        expected_classes, accept_message = self.build_accept_response(request, response_class)
        return self.wait_for_json_message(expected_classes, accept_message)

    def write_request(self, request):
        seq = self.writer.next_seq()
        if isinstance(request, dict):
            request["seq"] = seq