ashihaby/play-sockjs — pull request #15 by dosht (2013-08-18 16:42:24+00:00)
Title: websocket

u will need to add this test to the python tests

in sockjs-protocol-0.3.3.py:
from utils_03 import WebSocektRFC6455
class WebsocketRFC6455(Test):
def test_transport(self):
ws_url = 'ws:' + base_url.split(':',1)[1] + \
'/000/' + str(uuid.uuid4()) + '/websocket'
ws = WebSocektRFC6455(ws_url).connect()
self.assertEqual(ws.recv(), u'o')
ws.send(u'["a"]')
self.assertEqual(ws.recv(), u'a["a"]')
# ws.close()

def test_close(self):
pass

def test_broken_json(self):
pass

in utils_03.py:
class TimeoutException(Exception): pass

class WebSocektRFC6455(object):

def __init__(self, url):
self.ws = WebSocketClient(url)
self.queue = Queue.Queue()
def onmessage(m):
print "received: %s" % m
self.queue.put(unicode(str(m), 'utf-8'))

self.ws.received_message = onmessage

def connect(self):
self.ws.connect()
return self

def send(self, msg):
self.ws.send(msg)
return self

def recv(self, timeout=1.0):
try:
return self.queue.get(timeout=timeout)
except Queue.Empty:
raise TimeoutException("websocket didn't receive any thing")

def close(self):
self.ws.close()