91 lines
		
	
	
		
			2.4 KiB
		
	
	
	
		
			Python
		
	
	
	
			
		
		
	
	
			91 lines
		
	
	
		
			2.4 KiB
		
	
	
	
		
			Python
		
	
	
	
| 
 | |
| """ webtest
 | |
| """
 | |
| 
 | |
| import py
 | |
| 
 | |
| try:
 | |
|     from pypy.translator.js.main import rpython2javascript
 | |
|     from pypy.translator.js import commproxy
 | |
|     
 | |
|     commproxy.USE_MOCHIKIT = False
 | |
| except ImportError:
 | |
|     py.test.skip("PyPy not found")
 | |
| 
 | |
| def setup_module(mod):
 | |
|     config = py.test.config._reparse([])
 | |
|     config._overwrite('_dist_import_pypy', True)
 | |
|     from py.__.test.rsession.web import TestHandler as _TestHandler
 | |
|     from py.__.test.rsession.web import MultiQueue
 | |
|     mod._TestHandler = _TestHandler
 | |
|     mod.MultiQueue = MultiQueue
 | |
| 
 | |
| def test_js_generate():
 | |
|     from py.__.test.rsession import webjs
 | |
|     from py.__.test.rsession.web import FUNCTION_LIST, IMPORTED_PYPY
 | |
|     
 | |
|     source = rpython2javascript(webjs, FUNCTION_LIST)
 | |
|     assert source
 | |
| 
 | |
| def test_parse_args():
 | |
|     class TestTestHandler(_TestHandler):
 | |
|         def __init__(self):
 | |
|             pass
 | |
|     h = TestTestHandler()
 | |
|     assert h.parse_args('foo=bar') == {'foo': 'bar'}
 | |
|     assert h.parse_args('foo=bar%20baz') == {'foo': 'bar baz'}
 | |
|     assert h.parse_args('foo%20bar=baz') == {'foo bar': 'baz'}
 | |
|     assert h.parse_args('foo=bar%baz') == {'foo': 'bar\xbaz'}
 | |
|     py.test.raises(ValueError, 'h.parse_args("foo")')
 | |
| 
 | |
| class TestMultiQueue(object):
 | |
|     def test_get_one_sessid(self):
 | |
|         mq = MultiQueue()
 | |
|         mq.put(1)
 | |
|         result = mq.get(1234)
 | |
|         assert result == 1
 | |
| 
 | |
|     def test_get_two_sessid(self):
 | |
|         mq = MultiQueue()
 | |
|         mq.put(1)
 | |
|         result = mq.get(1234)
 | |
|         assert result == 1
 | |
|         mq.put(2)
 | |
|         result = mq.get(1234)
 | |
|         assert result == 2
 | |
|         result = mq.get(5678)
 | |
|         assert result == 1
 | |
|         result = mq.get(5678)
 | |
|         assert result == 2
 | |
| 
 | |
|     def test_get_blocking(self):
 | |
|         import thread
 | |
|         result = []
 | |
|         def getitem(mq, sessid):
 | |
|             result.append(mq.get(sessid))
 | |
|         mq = MultiQueue()
 | |
|         thread.start_new_thread(getitem, (mq, 1234))
 | |
|         assert not result
 | |
|         mq.put(1)
 | |
|         py.std.time.sleep(0.1)
 | |
|         assert result == [1]
 | |
| 
 | |
|     def test_empty(self):
 | |
|         mq = MultiQueue()
 | |
|         assert mq.empty()
 | |
|         mq.put(1)
 | |
|         assert not mq.empty()
 | |
|         result = mq.get(1234)
 | |
|         result == 1
 | |
|         assert mq.empty()
 | |
|         mq.put(2)
 | |
|         result = mq.get(4567)
 | |
|         assert result == 1
 | |
|         result = mq.get(1234)
 | |
|         assert result == 2
 | |
|         assert not mq.empty()
 | |
|         result = mq.get(4567)
 | |
|         assert result == 2
 | |
|         assert mq.empty()
 | |
| 
 |