test_protocol.py 文件源码

def test_dataReceived_adds_message_to_queue(self):
        protocol, factory = self.make_protocol()
        self.patch_autospec(protocol, "processMessages")
        message = {"type": MSG_TYPE.REQUEST}
        self.expectThat(
            protocol.dataReceived(
                json.dumps(message).encode("ascii")), Is(NOT_DONE_YET))
        self.expectThat(protocol.messages, Equals(deque([message])))