test_protocol.py 文件源码

def test_jsonDecoder(self):
        '''SockJSWireProtocolWrapper can use a json.JSONDecoder subclass for
        receives.

        '''
        class SetDecoder(json.JSONDecoder):
            def __init__(self, *args, **kwargs):
                kwargs['object_hook'] = self.set_object_hook
                super(SetDecoder, self).__init__(*args, **kwargs)

            def set_object_hook(self, obj):
                if isinstance(obj, dict) and obj.get('!set'):
                    return set(obj['!set'])
                return obj

        factory = P.SockJSWireProtocolWrappingFactory(
            self.wrappedFactory,
            jsonDecoder=SetDecoder)

        encodingProtocol = factory.buildProtocol(self.address)
        encodingProtocol.makeConnection(self.transport)

        encodingProtocol.dataReceived(b'{"!set": [1, 2, 3]}')
        self.assertEqual(self.receivedData, [{1, 2, 3}])