docker/docker-ubuntu-vnc-desktop
1
0
Fork 0
mirror of https://github.com/fcwu/docker-ubuntu-vnc-desktop synced 2026-08-05 16:12:41 +02:00
Code Issues Releases Activity

Push submodule websockify of noVNC

Browse Source
This commit is contained in:
Doro Wu
2015-05-13 10:05:09 +08:00
parent 1bc6fcb092
commit 89a5132f55
79 changed files with 11786 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
noVNC/utils/websockify/tests/b64_vs_utf8.py
Executable
+29
View File
@@ -0,0 +1,29 @@
#!/usr/bin/env python
from base64 import b64encode, b64decode
from codecs import (utf_8_encode, utf_8_decode,
latin_1_encode, latin_1_decode)
import random, time
buf_len = 10000
iterations = 10000
print "Generating random input buffer"
r = random.Random()
buf = "".join([chr(r.randint(0, 255)) for i in range(buf_len)])
tests = {'UTF8 encode': lambda: utf_8_encode(unicode(buf, 'latin-1'))[0],
'B64 encode': lambda: b64encode(buf)}
utf8_buf = tests['UTF8 encode']()
b64_buf = tests['B64 encode']()
tests.update({'UTF8 decode': lambda: latin_1_encode(utf_8_decode(utf8_buf)[0])[0],
'B64 decode': lambda: b64decode(b64_buf)})
print "Running tests"
for test in 'UTF8 encode', 'B64 encode', 'UTF8 decode', 'B64 decode':
start = time.time()
for i in range(iterations):
res_buf = tests[test]()
print "%s took %s seconds (result size %s)" % (
test, (time.time() - start), len(res_buf))
noVNC/utils/websockify/tests/base64.html
+91
View File
@@ -0,0 +1,91 @@
<!DOCTYPE html>
<html xmlns="http://www.w3.org/1999/xhtml">
<head>
<title>Native Base64 Tests</title>
<script src="include/util.js"></script>
<script src="include/webutil.js"></script>
<script src="include/base64.js"></script>
</head>
<body>
<h1>Native Base64 Tests</h1>
<br>
Messages:<br>
<textarea id="debug" style="font-size: 9px;" cols=80 rows=25></textarea>
<br>
</body>
<script>
function debug(str) {
console.log(str);
cell = $D('debug');
cell.innerHTML += str + "\n";
cell.scrollTop = cell.scrollHeight;
}
function assertRun(code, result) {
try {
var actual = eval(code);
} catch (exc) {
debug("FAIL: '" + code + "' threw an exception");
fail += 1;
return false;
}
if (actual !== result) {
debug("FAIL: '" + code + "' returned '" + actual + "', expected '" + result + "'");
fail += 1;
return false;
}
debug("PASS: '" + code + "' returned expected '" + result +"'");
pass += 1;
return true;
}
function Base64_decode(data) {
var arr = Base64.decode (data);
return arr.map(function (num) {
return String.fromCharCode(num); } ).join('');
}
window.onload = function() {
var str;
debug('onload');
fail = 0;
pass = 0;
assertRun('window.btoa("hello world")', 'aGVsbG8gd29ybGQ=');
assertRun('window.btoa("a")', 'YQ==');
assertRun('window.btoa("ab")', 'YWI=');
assertRun('window.btoa("abc")', 'YWJj');
assertRun('window.btoa("abcd")', 'YWJjZA==');
assertRun('window.btoa("abcde")', 'YWJjZGU=');
assertRun('window.btoa("abcdef")', 'YWJjZGVm');
assertRun('window.btoa("abcdefg")', 'YWJjZGVmZw==');
assertRun('window.btoa("abcdefgh")', 'YWJjZGVmZ2g=');
assertRun('window.atob("aGVsbG8gd29ybGQ=")', 'hello world');
assertRun('Base64_decode("aGVsbG8gd29ybGQ=")', 'hello world');
assertRun('window.atob("YQ==")', 'a');
assertRun('Base64_decode("YQ==")', 'a');
assertRun('window.atob("YWI=")', 'ab');
assertRun('Base64_decode("YWI=")', 'ab');
assertRun('window.atob("YWJj")', 'abc');
assertRun('Base64_decode("YWJj")', 'abc');
assertRun('window.atob("YWJjZA==")', 'abcd');
assertRun('Base64_decode("YWJjZA==")', 'abcd');
assertRun('window.atob("YWJjZGU=")', 'abcde');
assertRun('Base64_decode("YWJjZGU=")', 'abcde');
assertRun('window.atob("YWJjZGVm")', 'abcdef');
assertRun('Base64_decode("YWJjZGVm")', 'abcdef');
assertRun('typeof window.btoa', 'function');
assertRun('window.btoa("")', '');
assertRun('window.btoa(null)', '');
assertRun('window.atob(window.btoa(window))', window.toString()); // "[object DOMWindow]"
assertRun('window.btoa("\\u0080\\u0081")', 'gIE=');
debug("Tests failed: " + fail);
debug("Tests passed: " + pass);
}
</script>
noVNC/utils/websockify/tests/base64.js
+12
View File
@@ -0,0 +1,12 @@
// The following results in 'hello [MANGLED]'
//
// Filed as https://github.com/ry/node/issues/issue/402
var sys = require("sys"),
buf = new Buffer(1024), len,
str1 = "aGVsbG8g", // 'hello '
str2 = "d29ybGQ=", // 'world'
len = buf.write(str1, 0, 'base64');
len += buf.write(str2, len, 'base64');
sys.log("decoded result: " + buf.toString('binary', 0, len));
noVNC/utils/websockify/tests/echo.html
+148
View File
@@ -0,0 +1,148 @@
<html>
<head>
<title>WebSockets Echo Test</title>
<script src="include/util.js"></script>
<script src="include/webutil.js"></script>
<script src="include/base64.js"></script>
<script src="include/websock.js"></script>
<!-- Uncomment to activate firebug lite -->
<!--
<script type='text/javascript'
src='http://getfirebug.com/releases/lite/1.2/firebug-lite-compressed.js'></script>
-->
</head>
<body>
Host: <input id='host' style='width:100'>&nbsp;
Port: <input id='port' style='width:50'>&nbsp;
Encrypt: <input id='encrypt' type='checkbox'>&nbsp;
<input id='connectButton' type='button' value='Start' style='width:100px'
onclick="connect();">&nbsp;
<br>
Log:<br>
<textarea id="messages" style="font-size: 9;" cols=80 rows=25></textarea>
</body>
<script>
var ws, host = null, port = null,
msg_cnt = 0, send_cnt = 1, echoDelay = 500,
echo_ref;
function message(str) {
console.log(str);
cell = $D('messages');
cell.innerHTML += msg_cnt + ": " + str + "\n";
cell.scrollTop = cell.scrollHeight;
msg_cnt++;
}
Array.prototype.pushStr = function (str) {
var n = str.length;
for (var i=0; i < n; i++) {
this.push(str.charCodeAt(i));
}
}
function send_msg() {
var str = "Message #" + send_cnt;
ws.send_string(str);
message("Sent message: '" + str + "'");
send_cnt++;
}
function update_stats() {
$D('sent').innerHTML = sent;
$D('received').innerHTML = received;
$D('errors').innerHTML = errors;
}
function connect() {
var host = $D('host').value,
port = $D('port').value,
scheme = "ws://", uri;
console.log(">> connect");
if ((!host) || (!port)) {
console.log("must set host and port");
return;
}
if (ws) {
ws.close();
}
if ($D('encrypt').checked) {
scheme = "wss://";
}
uri = scheme + host + ":" + port;
message("connecting to " + uri);
ws = new Websock();
ws.open(uri);
ws.on('message', function(e) {
//console.log(">> WebSockets.onmessage");
var str = ws.rQshiftStr();
message("Received message '" + str + "'");
//console.log("<< WebSockets.onmessage");
});
ws.on('open', function(e) {
console.log(">> WebSockets.onopen");
echo_ref = setInterval(send_msg, echoDelay);
console.log("<< WebSockets.onopen");
});
ws.on('close', function(e) {
console.log(">> WebSockets.onclose");
if (echo_ref) {
clearInterval(echo_ref);
echo_ref = null;
}
console.log("<< WebSockets.onclose");
});
ws.on('error', function(e) {
console.log(">> WebSockets.onerror");
if (echo_ref) {
clearInterval(echo_ref);
echo_ref = null;
}
console.log("<< WebSockets.onerror");
});
$D('connectButton').value = "Stop";
$D('connectButton').onclick = disconnect;
console.log("<< connect");
}
function disconnect() {
console.log(">> disconnect");
if (ws) {
ws.close();
}
if (echo_ref) {
clearInterval(echo_ref);
}
$D('connectButton').value = "Start";
$D('connectButton').onclick = connect;
console.log("<< disconnect");
}
window.onload = function() {
console.log("onload");
var url = document.location.href;
$D('host').value = (url.match(/host=([^&#]*)/) || ['',window.location.hostname])[1];
$D('port').value = (url.match(/port=([^&#]*)/) || ['',window.location.port])[1];
}
</script>
</html>
noVNC/utils/websockify/tests/echo.py
Executable
+76
View File
@@ -0,0 +1,76 @@
#!/usr/bin/env python
'''
A WebSocket server that echos back whatever it receives from the client.
Copyright 2010 Joel Martin
Licensed under LGPL version 3 (see docs/LICENSE.LGPL-3)
You can make a cert/key with openssl using:
openssl req -new -x509 -days 365 -nodes -out self.pem -keyout self.pem
as taken from http://docs.python.org/dev/library/ssl.html#certificates
'''
import os, sys, select, optparse, logging
sys.path.insert(0,os.path.join(os.path.dirname(__file__), ".."))
from websockify.websocket import WebSocketServer, WebSocketRequestHandler
class WebSocketEcho(WebSocketRequestHandler):
"""
WebSockets server that echos back whatever is received from the
client. """
buffer_size = 8096
def new_websocket_client(self):
"""
Echo back whatever is received.
"""
cqueue = []
c_pend = 0
cpartial = ""
rlist = [self.request]
while True:
wlist = []
if cqueue or c_pend: wlist.append(self.request)
ins, outs, excepts = select.select(rlist, wlist, [], 1)
if excepts: raise Exception("Socket exception")
if self.request in outs:
# Send queued target data to the client
c_pend = self.send_frames(cqueue)
cqueue = []
if self.request in ins:
# Receive client data, decode it, and send it back
frames, closed = self.recv_frames()
cqueue.extend(frames)
if closed:
self.send_close()
if __name__ == '__main__':
parser = optparse.OptionParser(usage="%prog [options] listen_port")
parser.add_option("--verbose", "-v", action="store_true",
help="verbose messages and per frame traffic")
parser.add_option("--cert", default="self.pem",
help="SSL certificate file")
parser.add_option("--key", default=None,
help="SSL key file (if separate from cert)")
parser.add_option("--ssl-only", action="store_true",
help="disallow non-encrypted connections")
(opts, args) = parser.parse_args()
try:
if len(args) != 1: raise
opts.listen_port = int(args[0])
except:
parser.error("Invalid arguments")
logging.basicConfig(level=logging.INFO)
opts.web = "."
server = WebSocketServer(WebSocketEcho, **opts.__dict__)
server.start_server()
noVNC/utils/websockify/tests/echo.rb
Executable
+69
View File
@@ -0,0 +1,69 @@
#!/usr/bin/env ruby
# A WebSocket server that echos back whatever it receives from the client.
# Copyright 2011 Joel Martin
# Licensed under LGPL version 3 (see docs/LICENSE.LGPL-3)
require 'socket'
$: << "other"
$: << "../other"
require 'websocket'
class WebSocketEcho < WebSocketServer
# Echo back whatever is received
def new_websocket_client(client)
cqueue = []
c_pend = 0
rlist = [client]
loop do
wlist = []
if cqueue.length > 0 or c_pend
wlist << client
end
ins, outs, excepts = IO.select(rlist, wlist, nil, 1)
if excepts.length > 0
raise Exception, "Socket exception"
end
if outs.include?(client)
# Send queued data to the client
c_pend = send_frames(cqueue)
cqueue = []
end
if ins.include?(client)
# Receive client data, decode it, and send it back
frames, closed = recv_frames
cqueue += frames
if closed
raise EClose, closed
end
end
end # loop
end
end
port = ARGV[0].to_i || 8080
puts "Starting server on port #{port}"
server_cert = nil
server_key = nil
if ARGV.length > 2
server_cert = ARGV[1]
server_key = ARGV[2]
end
server = WebSocketEcho.new('listen_port' => port, 'verbose' => true,
'server_cert' => server_cert, 'server_key' => server_key)
server.start
server.join
puts "Server has been terminated"
# vim: sw=2
noVNC/utils/websockify/tests/include
Symlink
+1
View File
@@ -0,0 +1 @@
../include
noVNC/utils/websockify/tests/latency.html
+290
View File
@@ -0,0 +1,290 @@
<html>
<head>
<title>WebSockets Latency Test</title>
<script src="include/base64.js"></script>
<script src="include/util.js"></script>
<script src="include/webutil.js"></script>
<script src="include/websock.js"></script>
<!-- Uncomment to activate firebug lite -->
<!--
<script type='text/javascript'
src='http://getfirebug.com/releases/lite/1.2/firebug-lite-compressed.js'></script>
-->
</head>
<body>
Host: <input id='host' style='width:100'>&nbsp;
Port: <input id='port' style='width:50'>&nbsp;
Encrypt: <input id='encrypt' type='checkbox'>
<br>
Payload Size: <input id='payload_size' style='width:50'>&nbsp;
Send Delay (ms): <input id='sendDelay' style='width:50' value="10">&nbsp;
<input id='connectButton' type='button' value='Start' style='width:100px'
onclick="connect();">&nbsp;
<br><br>
<table border=1>
<tr>
<th align="right">Packets sent:</th>
<td align="right"><div id='sent'></div></td>
</tr><tr>
<th align="right">Packets Received:</th>
<td align="right"><div id='received'></div></td>
</tr><tr>
<th align="right">Average Latency:</th>
<td align="right"><div id='laverage'></div></td>
</tr><tr>
<th align="right">40 Frame Running Average Latency:</th>
<td align="right"><div id='lrunning'></div></td>
</tr><tr>
<th align="right">Minimum Latency:</th>
<td align="right"><div id='lmin'></div></td>
</tr><tr>
<th align="right">Maximum Latency:</th>
<td align="right"><div id='lmax'></div></td>
</tr>
</table>
<br>
Messages:<br>
<textarea id="messages" style="font-size: 9;" cols=80 rows=10></textarea>
</body>
<script>
var host = null, port = null, sendDelay = 0, actualSendDelay,
ws = null, send_ref = null,
sent, received, latencies, ltotal, laverage, lrunning, lmin, lmax,
run_length = 40,
payload_size = 2000, payload,
msg_cnt = 0, recv_seq = 0, send_seq = 0;
Array.prototype.pushStr = function (str) {
var n = str.length;
for (var i=0; i < n; i++) {
this.push(str.charCodeAt(i));
}
}
function message(str) {
console.log(str);
cell = $D('messages');
msg_cnt++;
cell.innerHTML += msg_cnt + ": " + str + "\n";
cell.scrollTop = cell.scrollHeight;
}
function add (x,y) {
return parseInt(x,10)+parseInt(y,10);
}
function recvMsg(data) {
//console.log(">> check_respond");
var i, now, arr, first, last, arr, latency;
now = (new Date()).getTime(); // Early as possible
arr = ws.rQshiftBytes(ws.rQlen());
first = String.fromCharCode(arr.shift());
last = String.fromCharCode(arr.pop());
if (first != "^") {
message("Error: packet missing start char '^'");
disconnect();
return;
}
if (last != "$") {
message("Error: packet missing end char '$'");
disconnect();
return;
}
arr = arr.map(function(num) {
return String.fromCharCode(num);
} ).join('').split(':');
seq = arr[0];
timestamp = parseInt(arr[1],10);
rpayload = arr[2];
if (seq != recv_seq) {
message("Error: expected seq " + recv_seq + " but got " + seq);
disconnect();
return;
}
recv_seq++;
if (payload !== rpayload) {
message("Payload corrupt");
disconnect();
return;
}
received++;
latency = now - timestamp;
latencies.push(latency);
if (latencies.length > run_length) {
latencies.shift();
}
ltotal += latency;
laverage = ltotal / received;
lrunning = 0;
for (var i=0; i < latencies.length; i++) {
lrunning += latencies[i];
}
lrunning = lrunning / latencies.length;
if (latency < lmin) {
lmin = latency;
}
if (latency > lmax) {
lmax = latency;
}
showStats();
//console.log("<< check_respond");
}
function sendMsg() {
var arr = [];
if (! ws.flush() ) {
message("WebSocket not ready, backing off");
actualSendDelay = actualSendDelay * 2;
send_ref = setTimeout(sendMsg, actualSendDelay);
return false;
} else {
// Scale the delay down to the requested minimum
if (actualSendDelay > sendDelay) {
message("WebSocket ready, increasing presure");
actualSendDelay = Math.max(actualSendDelay / 2, sendDelay);
}
}
timestamp = (new Date()).getTime();
arr.pushStr("^" + send_seq + ":" + timestamp + ":" + payload + "$");
send_seq ++;
ws.send(arr);
sent++;
showStats();
send_ref = setTimeout(sendMsg, actualSendDelay);
}
function showStats() {
$D('sent').innerHTML = sent;
$D('received').innerHTML = received;
$D('laverage').innerHTML = laverage.toFixed(2);
$D('lrunning').innerHTML = lrunning.toFixed(2);
$D('lmin').innerHTML = lmin.toFixed(2);
$D('lmax').innerHTML = lmax.toFixed(2);
}
function init_ws() {
console.log(">> init_ws");
var scheme = "ws://";
if ($D('encrypt').checked) {
scheme = "wss://";
}
var uri = scheme + host + ":" + port;
console.log("connecting to " + uri);
ws = new Websock();
ws.maxBufferedAmount = 5000;
ws.open(uri);
ws.on('message', function() {
recvMsg();
});
ws.on('open', function() {
send_ref = setTimeout(sendMsg, sendDelay);
});
ws.on('close', function(e) {
disconnect();
});
ws.on('error', function(e) {
message("Websock error: " + e);
disconnect();
});
console.log("<< init_ws");
}
function connect() {
console.log(">> connect");
host = $D('host').value;
port = $D('port').value;
payload_size = parseInt($D('payload_size').value, 10);
sendDelay = parseInt($D('sendDelay').value, 10);
if ((!host) || (!port)) {
console.log("must set host and port");
return;
}
if (ws) {
ws.close();
}
init_ws();
// Populate payload data
var numlist = []
for (var i=0; i < payload_size; i++) {
numlist.push( Math.floor(Math.random()*10) );
}
payload = numlist.join('');
// Initialize stats
sent = 0;
received = 0;
latencies = [];
ltotal = 0;
laverage = 0;
lrunning = 0;
lmin = 999999999;
lmax = 0;
actualSendDelay = sendDelay;
$D('connectButton').value = "Stop";
$D('connectButton').onclick = disconnect;
console.log("<< connect");
}
function disconnect() {
console.log(">> disconnect");
if (ws) {
ws.close();
}
if (send_ref) {
clearInterval(send_ref);
send_ref = null;
}
showStats(); // Final numbers
recv_seq = 0;
send_seq = 0;
$D('connectButton').value = "Start";
$D('connectButton').onclick = connect;
console.log("<< disconnect");
}
window.onload = function() {
console.log("onload");
if (Websock_native) {
message("Using native WebSockets");
} else {
message("initializing web-socket-js flash bridge");
}
var url = document.location.href;
$D('host').value = (url.match(/host=([^&#]*)/) || ['',window.location.hostname])[1];
$D('port').value = (url.match(/port=([^&#]*)/) || ['',window.location.port])[1];
$D('payload_size').value = payload_size;
}
</script>
</html>
noVNC/utils/websockify/tests/latency.py
Symlink
+1
View File
@@ -0,0 +1 @@
echo.py
noVNC/utils/websockify/tests/load.html
+250
View File
@@ -0,0 +1,250 @@
<html>
<head>
<title>WebSockets Load Test</title>
<script src="include/util.js"></script>
<script src="include/webutil.js"></script>
<script src="include/base64.js"></script>
<script src="include/websock.js"></script>
<!-- Uncomment to activate firebug lite -->
<!--
<script type='text/javascript'
src='http://getfirebug.com/releases/lite/1.2/firebug-lite-compressed.js'></script>
-->
</head>
<body>
Host: <input id='host' style='width:100'>&nbsp;
Port: <input id='port' style='width:50'>&nbsp;
Encrypt: <input id='encrypt' type='checkbox'>&nbsp;
Send Delay (ms): <input id='sendDelay' style='width:50' value="100">&nbsp;
<input id='connectButton' type='button' value='Start' style='width:100px'
onclick="connect();">&nbsp;
<br><br>
<table border=1>
<tr>
<th align="right">Packets sent:</th>
<td align="right"><div id='sent'>0</div></td>
</tr><tr>
<th align="right">Good Packets Received:</th>
<td align="right"><div id='received'>0</div></td>
</tr><tr>
<th align="right">Errors (Bad Packets Received:)</th>
<td align="right"><div id='errors'>0</div></td>
</tr>
</table>
<br>
Errors:<br>
<textarea id="error" style="font-size: 9;" cols=80 rows=25></textarea>
</body>
<script>
function error(str) {
console.error(str);
cell = $D('error');
cell.innerHTML += errors + ": " + str + "\n";
cell.scrollTop = cell.scrollHeight;
}
var host = null, port = null, sendDelay = 0;
var ws = null, update_ref = null, send_ref = null;
var sent = 0, received = 0, errors = 0;
var max_send = 2000;
var recv_seq = 0, send_seq = 0;
Array.prototype.pushStr = function (str) {
var n = str.length;
for (var i=0; i < n; i++) {
this.push(str.charCodeAt(i));
}
}
function add (x,y) {
return parseInt(x,10)+parseInt(y,10);
}
function check_respond(data) {
//console.log(">> check_respond");
var first, last, str, length, chksum, nums, arr;
first = String.fromCharCode(data.shift());
last = String.fromCharCode(data.pop());
if (first != "^") {
errors++;
error("Packet missing start char '^'");
return;
}
if (last != "$") {
errors++;
error("Packet missing end char '$'");
return;
}
arr = data.map(function(num) {
return String.fromCharCode(num);
} ).join('').split(':');
seq = arr[0];
length = arr[1];
chksum = arr[2];
nums = arr[3];
//console.log(" length:" + length + " chksum:" + chksum + " nums:" + nums);
if (seq != recv_seq) {
errors++;
error("Expected seq " + recv_seq + " but got " + seq);
recv_seq = parseInt(seq,10) + 1; // Back on track
return;
}
recv_seq++;
if (nums.length != length) {
errors++;
error("Expected length " + length + " but got " + nums.length);
return;
}
//real_chksum = nums.reduce(add);
real_chksum = 0;
for (var i=0; i < nums.length; i++) {
real_chksum += parseInt(nums.charAt(i), 10);
}
if (real_chksum != chksum) {
errors++
error("Expected chksum " + chksum + " but real chksum is " + real_chksum);
return;
}
received++;
//console.log(" Packet checks out: length:" + length + " chksum:" + chksum);
//console.log("<< check_respond");
}
function send() {
var length = Math.floor(Math.random()*(max_send-9)) + 10; // 10 - max_send
var numlist = [], arr = [];
for (var i=0; i < length; i++) {
numlist.push( Math.floor(Math.random()*10) );
}
//chksum = numlist.reduce(add);
chksum = 0;
for (var i=0; i < numlist.length; i++) {
chksum += parseInt(numlist[i], 10);
}
var nums = numlist.join('');
arr.pushStr("^" + send_seq + ":" + length + ":" + chksum + ":" + nums + "$")
send_seq ++;
ws.send(arr);
sent++;
}
function update_stats() {
$D('sent').innerHTML = sent;
$D('received').innerHTML = received;
$D('errors').innerHTML = errors;
}
function init_ws() {
console.log(">> init_ws");
var scheme = "ws://";
if ($D('encrypt').checked) {
scheme = "wss://";
}
var uri = scheme + host + ":" + port;
console.log("connecting to " + uri);
ws = new Websock();
ws.open(uri);
ws.on('message', function() {
//console.log(">> WebSockets.onmessage");
arr = ws.rQshiftBytes(ws.rQlen());
check_respond(arr);
//console.log("<< WebSockets.onmessage");
});
ws.on('open', function() {
console.log(">> WebSockets.onopen");
send_ref = setInterval(send, sendDelay);
console.log("<< WebSockets.onopen");
});
ws.on('close', function(e) {
console.log(">> WebSockets.onclose");
clearInterval(send_ref);
console.log("<< WebSockets.onclose");
});
ws.on('error', function(e) {
console.log(">> WebSockets.onerror");
console.log(" " + e);
console.log("<< WebSockets.onerror");
});
console.log("<< init_ws");
}
function connect() {
console.log(">> connect");
host = $D('host').value;
port = $D('port').value;
sendDelay = parseInt($D('sendDelay').value, 10);
if ((!host) || (!port)) {
console.log("must set host and port");
return;
}
if (ws) {
ws.close();
}
init_ws();
update_ref = setInterval(update_stats, 1);
$D('connectButton').value = "Stop";
$D('connectButton').onclick = disconnect;
console.log("<< connect");
}
function disconnect() {
console.log(">> disconnect");
if (ws) {
ws.close();
}
clearInterval(update_ref);
update_stats(); // Final numbers
recv_seq = 0;
send_seq = 0;
$D('connectButton').value = "Start";
$D('connectButton').onclick = connect;
console.log("<< disconnect");
}
/* If no builtin websockets then load web_socket.js */
if (window.WebSocket) {
VNC_native_ws = true;
} else {
VNC_native_ws = false;
console.log("Loading web-socket-js flash bridge");
var extra = "<script src='include/web-socket-js/swfobject.js'><\/script>";
extra += "<script src='include/web-socket-js/FABridge.js'><\/script>";
extra += "<script src='include/web-socket-js/web_socket.js'><\/script>";
document.write(extra);
}
window.onload = function() {
console.log("onload");
if (!VNC_native_ws) {
console.log("initializing web-socket-js flash bridge");
WebSocket.__swfLocation = "include/web-socket-js/WebSocketMain.swf";
WebSocket.__initialize();
}
var url = document.location.href;
$D('host').value = (url.match(/host=([^&#]*)/) || ['',''])[1];
$D('port').value = (url.match(/port=([^&#]*)/) || ['',''])[1];
}
</script>
</html>
noVNC/utils/websockify/tests/load.py
Executable
+170
View File
@@ -0,0 +1,170 @@
#!/usr/bin/env python
'''
WebSocket server-side load test program. Sends and receives traffic
that has a random payload (length and content) that is checksummed and
given a sequence number. Any errors are reported and counted.
'''
import sys, os, select, random, time, optparse, logging
sys.path.insert(0,os.path.join(os.path.dirname(__file__), ".."))
from websockify.websocket import WebSocketServer, WebSocketRequestHandler
class WebSocketLoadServer(WebSocketServer):
recv_cnt = 0
send_cnt = 0
def __init__(self, *args, **kwargs):
self.delay = kwargs.pop('delay')
WebSocketServer.__init__(self, *args, **kwargs)
class WebSocketLoad(WebSocketRequestHandler):
max_packet_size = 10000
def new_websocket_client(self):
print "Prepopulating random array"
self.rand_array = []
for i in range(0, self.max_packet_size):
self.rand_array.append(random.randint(0, 9))
self.errors = 0
self.send_cnt = 0
self.recv_cnt = 0
try:
self.responder(self.request)
except:
print "accumulated errors:", self.errors
self.errors = 0
raise
def responder(self, client):
c_pend = 0
cqueue = []
cpartial = ""
socks = [client]
last_send = time.time() * 1000
while True:
ins, outs, excepts = select.select(socks, socks, socks, 1)
if excepts: raise Exception("Socket exception")
if client in ins:
frames, closed = self.recv_frames()
err = self.check(frames)
if err:
self.errors = self.errors + 1
print err
if closed:
self.send_close()
now = time.time() * 1000
if client in outs:
if c_pend:
last_send = now
c_pend = self.send_frames()
elif now > (last_send + self.server.delay):
last_send = now
c_pend = self.send_frames([self.generate()])
def generate(self):
length = random.randint(10, self.max_packet_size)
numlist = self.rand_array[self.max_packet_size-length:]
# Error in length
#numlist.append(5)
chksum = sum(numlist)
# Error in checksum
#numlist[0] = 5
nums = "".join( [str(n) for n in numlist] )
data = "^%d:%d:%d:%s$" % (self.send_cnt, length, chksum, nums)
self.send_cnt += 1
return data
def check(self, frames):
err = ""
for data in frames:
if data.count('$') > 1:
raise Exception("Multiple parts within single packet")
if len(data) == 0:
self.traffic("_")
continue
if data[0] != "^":
err += "buf did not start with '^'\n"
continue
try:
cnt, length, chksum, nums = data[1:-1].split(':')
cnt = int(cnt)
length = int(length)
chksum = int(chksum)
except:
print "\n<BOF>" + repr(data) + "<EOF>"
err += "Invalid data format\n"
continue
if self.recv_cnt != cnt:
err += "Expected count %d but got %d\n" % (self.recv_cnt, cnt)
self.recv_cnt = cnt + 1
continue
self.recv_cnt += 1
if len(nums) != length:
err += "Expected length %d but got %d\n" % (length, len(nums))
continue
inv = nums.translate(None, "0123456789")
if inv:
err += "Invalid characters found: %s\n" % inv
continue
real_chksum = 0
for num in nums:
real_chksum += int(num)
if real_chksum != chksum:
err += "Expected checksum %d but real chksum is %d\n" % (chksum, real_chksum)
return err
if __name__ == '__main__':
parser = optparse.OptionParser(usage="%prog [options] listen_port")
parser.add_option("--verbose", "-v", action="store_true",
help="verbose messages and per frame traffic")
parser.add_option("--cert", default="self.pem",
help="SSL certificate file")
parser.add_option("--key", default=None,
help="SSL key file (if separate from cert)")
parser.add_option("--ssl-only", action="store_true",
help="disallow non-encrypted connections")
(opts, args) = parser.parse_args()
try:
if len(args) != 1: raise
opts.listen_port = int(args[0])
if len(args) not in [1,2]: raise
opts.listen_port = int(args[0])
if len(args) == 2:
opts.delay = int(args[1])
else:
opts.delay = 10
except:
parser.error("Invalid arguments")
logging.basicConfig(level=logging.INFO)
opts.web = "."
server = WebSocketLoadServer(WebSocketLoad, **opts.__dict__)
server.start_server()
noVNC/utils/websockify/tests/plain_echo.html
+168
View File
@@ -0,0 +1,168 @@
<html>
<head>
<title>WebSockets Echo Test</title>
<script src="include/util.js"></script>
<script src="include/webutil.js"></script>
<!-- Uncomment to activate firebug lite -->
<!--
<script type='text/javascript'
src='http://getfirebug.com/releases/lite/1.2/firebug-lite-compressed.js'></script>
-->
</head>
<body>
Host: <input id='host' style='width:100'>&nbsp;
Port: <input id='port' style='width:50'>&nbsp;
Encrypt: <input id='encrypt' type='checkbox'>&nbsp;
<input id='connectButton' type='button' value='Start' style='width:100px'
onclick="connect();">&nbsp;
<br>
Log:<br>
<textarea id="messages" style="font-size: 9;" cols=80 rows=25></textarea>
</body>
<script>
var ws, host = null, port = null,
msg_cnt = 0, send_cnt = 1, echoDelay = 500,
echo_ref;
function message(str) {
console.log(str);
cell = $D('messages');
cell.innerHTML += msg_cnt + ": " + str + "\n";
cell.scrollTop = cell.scrollHeight;
msg_cnt++;
}
Array.prototype.pushStr = function (str) {
var n = str.length;
for (var i=0; i < n; i++) {
this.push(str.charCodeAt(i));
}
}
function send_msg() {
if (ws.bufferedAmount > 0) {
console.log("Delaying send");
return;
}
var str = "Message #" + send_cnt, arr = [];
ws.send(str);
message("Sent message: '" + str + "'");
send_cnt++;
}
function update_stats() {
$D('sent').innerHTML = sent;
$D('received').innerHTML = received;
$D('errors').innerHTML = errors;
}
function init_ws() {
console.log(">> init_ws");
console.log("<< init_ws");
}
function connect() {
var host = $D('host').value,
port = $D('port').value,
scheme = "ws://", uri;
console.log(">> connect");
if ((!host) || (!port)) {
console.log("must set host and port");
return;
}
if (ws) {
ws.close();
}
if ($D('encrypt').checked) {
scheme = "wss://";
}
uri = scheme + host + ":" + port;
message("connecting to " + uri);
ws = new WebSocket(uri, "base64");
ws.onmessage = function(e) {
//console.log(">> WebSockets.onmessage");
message("Received message '" + e.data + "'");
//console.log("<< WebSockets.onmessage");
};
ws.onopen = function(e) {
console.log(">> WebSockets.onopen");
echo_ref = setInterval(send_msg, echoDelay);
console.log("<< WebSockets.onopen");
};
ws.onclose = function(e) {
console.log(">> WebSockets.onclose");
if (echo_ref) {
clearInterval(echo_ref);
echo_ref = null;
}
console.log("<< WebSockets.onclose");
};
ws.onerror = function(e) {
console.log(">> WebSockets.onerror");
if (echo_ref) {
clearInterval(echo_ref);
echo_ref = null;
}
console.log("<< WebSockets.onerror");
};
$D('connectButton').value = "Stop";
$D('connectButton').onclick = disconnect;
console.log("<< connect");
}
function disconnect() {
console.log(">> disconnect");
if (ws) {
ws.close();
}
if (echo_ref) {
clearInterval(echo_ref);
}
$D('connectButton').value = "Start";
$D('connectButton').onclick = connect;
console.log("<< disconnect");
}
/* If no builtin websockets then load web_socket.js */
if (window.WebSocket) {
VNC_native_ws = true;
} else {
VNC_native_ws = false;
console.log("Loading web-socket-js flash bridge");
var extra = "<script src='include/web-socket-js/swfobject.js'><\/script>";
extra += "<script src='include/web-socket-js/FABridge.js'><\/script>";
extra += "<script src='include/web-socket-js/web_socket.js'><\/script>";
document.write(extra);
}
window.onload = function() {
console.log("onload");
if (!VNC_native_ws) {
console.log("initializing web-socket-js flash bridge");
WebSocket.__swfLocation = "include/web-socket-js/WebSocketMain.swf";
WebSocket.__initialize();
}
var url = document.location.href;
$D('host').value = (url.match(/host=([^&#]*)/) || ['',''])[1];
$D('port').value = (url.match(/port=([^&#]*)/) || ['',''])[1];
}
</script>
</html>
noVNC/utils/websockify/tests/simple.html
+68
View File
@@ -0,0 +1,68 @@
<html>
<head>
<title>Websock Simple Client</title>
<script src="include/util.js"></script>
<script src="include/base64.js"></script>
<script src="include/websock.js"></script>
</head>
<body>
WebSocket/websockify URI: <input id='target'>&nbsp;
<input id='connectButton' type='button' value='Connect'
onclick="connect();">
<br> <br>
<input id='sendText'>&nbsp;
<input id='sendButton' type='button' value='Send' disabled
onclick="send();">&nbsp;
<br> <br>
Log:<br><textarea id="messages" cols=80 rows=25></textarea>
</body>
<script>
var $D = function(id) { return document.getElementById(id); },
ws = null, msgs = $D('messages');
function msg(str) {
msgs.innerHTML += str + "\n";
msgs.scrollTop = msgs.scrollHeight;
}
function connect() {
var uri = $D('target').value;
ws = new Websock()
msg("connecting to: " + uri);
ws.open(uri);
ws.on('open', function () {
msg("Connected");
});
ws.on('message', function () {
msg("Received: " + ws.rQshiftStr());
});
ws.on('close', function () {
disconnect();
msg("Disconnected");
});
$D('connectButton').value = "Disconnect";
$D('connectButton').onclick = disconnect;
$D('sendButton').disabled = false;
}
function disconnect() {
if (ws) { ws.close(); }
ws = null;
$D('connectButton').value = "Connect";
$D('connectButton').onclick = connect;
$D('sendButton').disabled = true;
}
function send() {
msg("Sending: " + $D('sendText').value);
ws.send_string($D('sendText').value);
};
</script>
</html>
noVNC/utils/websockify/tests/test_websocket.py
+243
View File
@@ -0,0 +1,243 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# Copyright(c)2013 NTT corp. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
""" Unit tests for websocket """
import errno
import os
import logging
import select
import shutil
import socket
import ssl
import stubout
import sys
import tempfile
import unittest
from ssl import SSLError
from websockify import websocket as websocket
from SimpleHTTPServer import SimpleHTTPRequestHandler
class MockConnection(object):
def __init__(self, path):
self.path = path
def makefile(self, mode='r', bufsize=-1):
return open(self.path, mode, bufsize)
class WebSocketTestCase(unittest.TestCase):
def _init_logger(self, tmpdir):
name = 'websocket-unittest'
logger = logging.getLogger(name)
logger.setLevel(logging.DEBUG)
logger.propagate = True
filename = "%s.log" % (name)
handler = logging.FileHandler(filename)
handler.setFormatter(logging.Formatter("%(message)s"))
logger.addHandler(handler)
def setUp(self):
"""Called automatically before each test."""
super(WebSocketTestCase, self).setUp()
self.stubs = stubout.StubOutForTesting()
# Temporary dir for test data
self.tmpdir = tempfile.mkdtemp()
# Put log somewhere persistent
self._init_logger('./')
# Mock this out cause it screws tests up
self.stubs.Set(os, 'chdir', lambda *args, **kwargs: None)
self.server = self._get_websockserver(daemon=True,
ssl_only=False)
self.soc = self.server.socket('localhost')
def tearDown(self):
"""Called automatically after each test."""
self.stubs.UnsetAll()
shutil.rmtree(self.tmpdir)
super(WebSocketTestCase, self).tearDown()
def _get_websockserver(self, **kwargs):
return websocket.WebSocketServer(listen_host='localhost',
listen_port=80,
key=self.tmpdir,
web=self.tmpdir,
record=self.tmpdir,
**kwargs)
def _mock_os_open_oserror(self, file, flags):
raise OSError('')
def _mock_os_close_oserror(self, fd):
raise OSError('')
def _mock_os_close_oserror_EBADF(self, fd):
raise OSError(errno.EBADF, '')
def _mock_socket(self, *args, **kwargs):
return self.soc
def _mock_select(self, rlist, wlist, xlist, timeout=None):
return '_mock_select'
def _mock_select_exception(self, rlist, wlist, xlist, timeout=None):
raise Exception
def _mock_select_keyboardinterrupt(self, rlist, wlist,
xlist, timeout=None):
raise KeyboardInterrupt
def _mock_select_systemexit(self, rlist, wlist, xlist, timeout=None):
sys.exit()
def test_daemonize_error(self):
soc = self._get_websockserver(daemon=True, ssl_only=1, idle_timeout=1)
self.stubs.Set(os, 'fork', lambda *args: None)
self.stubs.Set(os, 'setsid', lambda *args: None)
self.stubs.Set(os, 'close', self._mock_os_close_oserror)
self.assertRaises(OSError, soc.daemonize, keepfd=None, chdir='./')
def test_daemonize_EBADF_error(self):
soc = self._get_websockserver(daemon=True, ssl_only=1, idle_timeout=1)
self.stubs.Set(os, 'fork', lambda *args: None)
self.stubs.Set(os, 'setsid', lambda *args: None)
self.stubs.Set(os, 'close', self._mock_os_close_oserror_EBADF)
self.stubs.Set(os, 'open', self._mock_os_open_oserror)
self.assertRaises(OSError, soc.daemonize, keepfd=None, chdir='./')
def test_decode_hybi(self):
soc = self._get_websockserver(daemon=False, ssl_only=1, idle_timeout=1)
self.assertRaises(Exception, soc.decode_hybi, 'a' * 128,
base64=True)
def test_do_websocket_handshake(self):
soc = self._get_websockserver(daemon=True, ssl_only=0, idle_timeout=1)
soc.scheme = 'scheme'
headers = {'Sec-WebSocket-Protocol': 'binary',
'Sec-WebSocket-Version': '7',
'Sec-WebSocket-Key': 'foo'}
soc.do_websocket_handshake(headers, '127.0.0.1')
def test_do_handshake(self):
soc = self._get_websockserver(daemon=True, ssl_only=0, idle_timeout=1)
self.stubs.Set(select, 'select', self._mock_select)
self.stubs.Set(socket._socketobject, 'recv', lambda *args: 'mock_recv')
self.assertRaises(Exception, soc.do_handshake, self.soc, '127.0.0.1')
def test_do_handshake_ssl_error(self):
soc = self._get_websockserver(daemon=True, ssl_only=0, idle_timeout=1)
def _mock_wrap_socket(*args, **kwargs):
from ssl import SSLError
raise SSLError('unit test exception')
self.stubs.Set(select, 'select', self._mock_select)
self.stubs.Set(socket._socketobject, 'recv', lambda *args: '\x16')
self.stubs.Set(ssl, 'wrap_socket', _mock_wrap_socket)
self.assertRaises(SSLError, soc.do_handshake, self.soc, '127.0.0.1')
def test_fallback_SIGCHILD(self):
soc = self._get_websockserver(daemon=True, ssl_only=0, idle_timeout=1)
soc.fallback_SIGCHLD(None, None)
def test_start_server_Exception(self):
soc = self._get_websockserver(daemon=False, ssl_only=1, idle_timeout=1)
self.stubs.Set(websocket.WebSocketServer, 'socket', self._mock_socket)
self.stubs.Set(websocket.WebSocketServer, 'daemonize',
lambda *args, **kwargs: None)
self.stubs.Set(select, 'select', self._mock_select_exception)
self.assertEqual(None, soc.start_server())
def test_start_server_KeyboardInterrupt(self):
soc = self._get_websockserver(daemon=False, ssl_only=1, idle_timeout=1)
self.stubs.Set(websocket.WebSocketServer, 'socket', self._mock_socket)
self.stubs.Set(websocket.WebSocketServer, 'daemonize',
lambda *args, **kwargs: None)
self.stubs.Set(select, 'select', self._mock_select_keyboardinterrupt)
self.assertEqual(None, soc.start_server())
def test_start_server_systemexit(self):
websocket.ssl = None
self.stubs.Set(websocket.WebSocketServer, 'socket', self._mock_socket)
self.stubs.Set(websocket.WebSocketServer, 'daemonize',
lambda *args, **kwargs: None)
self.stubs.Set(select, 'select', self._mock_select_systemexit)
soc = self._get_websockserver(daemon=True, ssl_only=0, idle_timeout=1,
verbose=True)
self.assertEqual(None, soc.start_server())
def test_WSRequestHandle_do_GET_nofile(self):
request = 'GET /tmp.txt HTTP/0.9'
with tempfile.NamedTemporaryFile() as test_file:
test_file.write(request)
test_file.flush()
test_file.seek(0)
con = MockConnection(test_file.name)
soc = websocket.WSRequestHandler(con, "127.0.0.1", file_only=True)
soc.path = ''
soc.headers = {'upgrade': ''}
self.stubs.Set(SimpleHTTPRequestHandler, 'send_response',
lambda *args: None)
soc.do_GET()
self.assertEqual(404, soc.last_code)
def test_WSRequestHandle_do_GET_hidden_resource(self):
request = 'GET /tmp.txt HTTP/0.9'
with tempfile.NamedTemporaryFile() as test_file:
test_file.write(request)
test_file.flush()
test_file.seek(0)
con = MockConnection(test_file.name)
soc = websocket.WSRequestHandler(con, '127.0.0.1', no_parent=True)
soc.path = test_file.name + '?'
soc.headers = {'upgrade': ''}
soc.webroot = 'no match startswith'
self.stubs.Set(SimpleHTTPRequestHandler,
'send_response',
lambda *args: None)
soc.do_GET()
self.assertEqual(403, soc.last_code)
def testsocket_set_keepalive_options(self):
keepcnt = 12
keepidle = 34
keepintvl = 56
sock = self.server.socket('localhost',
tcp_keepcnt=keepcnt,
tcp_keepidle=keepidle,
tcp_keepintvl=keepintvl)
self.assertEqual(sock.getsockopt(socket.SOL_TCP,
socket.TCP_KEEPCNT), keepcnt)
self.assertEqual(sock.getsockopt(socket.SOL_TCP,
socket.TCP_KEEPIDLE), keepidle)
self.assertEqual(sock.getsockopt(socket.SOL_TCP,
socket.TCP_KEEPINTVL), keepintvl)
sock = self.server.socket('localhost',
tcp_keepalive=False,
tcp_keepcnt=keepcnt,
tcp_keepidle=keepidle,
tcp_keepintvl=keepintvl)
self.assertNotEqual(sock.getsockopt(socket.SOL_TCP,
socket.TCP_KEEPCNT), keepcnt)
self.assertNotEqual(sock.getsockopt(socket.SOL_TCP,
socket.TCP_KEEPIDLE), keepidle)
self.assertNotEqual(sock.getsockopt(socket.SOL_TCP,
socket.TCP_KEEPINTVL), keepintvl)
noVNC/utils/websockify/tests/test_websocketproxy.py
+127
View File
@@ -0,0 +1,127 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# Copyright(c)2013 NTT corp. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
""" Unit tests for websocketproxy """
import os
import logging
import select
import shutil
import stubout
import subprocess
import tempfile
import time
import unittest
from websockify import websocketproxy
class MockSocket(object):
def __init__(*args, **kwargs):
pass
def shutdown(*args):
pass
def close(*args):
pass
class WebSocketProxyTest(unittest.TestCase):
def _init_logger(self, tmpdir):
name = 'websocket-unittest'
logger = logging.getLogger(name)
logger.setLevel(logging.DEBUG)
logger.propagate = True
filename = "%s.log" % (name)
handler = logging.FileHandler(filename)
handler.setFormatter(logging.Formatter("%(message)s"))
logger.addHandler(handler)
def setUp(self):
"""Called automatically before each test."""
super(WebSocketProxyTest, self).setUp()
self.soc = ''
self.stubs = stubout.StubOutForTesting()
# Temporary dir for test data
self.tmpdir = tempfile.mkdtemp()
# Put log somewhere persistent
self._init_logger('./')
# Mock this out cause it screws tests up
self.stubs.Set(os, 'chdir', lambda *args, **kwargs: None)
def tearDown(self):
"""Called automatically after each test."""
self.stubs.UnsetAll()
shutil.rmtree(self.tmpdir)
super(WebSocketProxyTest, self).tearDown()
def _get_websockproxy(self, **kwargs):
return websocketproxy.WebSocketProxy(key=self.tmpdir,
web=self.tmpdir,
record=self.tmpdir,
**kwargs)
def test_run_wrap_cmd(self):
web_socket_proxy = self._get_websockproxy()
web_socket_proxy.__dict__["wrap_cmd"] = "wrap_cmd"
def mock_Popen(*args, **kwargs):
return '_mock_cmd'
self.stubs.Set(subprocess, 'Popen', mock_Popen)
web_socket_proxy.run_wrap_cmd()
self.assertEquals(web_socket_proxy.spawn_message, True)
def test_started(self):
web_socket_proxy = self._get_websockproxy()
web_socket_proxy.__dict__["spawn_message"] = False
web_socket_proxy.__dict__["wrap_cmd"] = "wrap_cmd"
def mock_run_wrap_cmd(*args, **kwargs):
web_socket_proxy.__dict__["spawn_message"] = True
self.stubs.Set(web_socket_proxy, 'run_wrap_cmd', mock_run_wrap_cmd)
web_socket_proxy.started()
self.assertEquals(web_socket_proxy.__dict__["spawn_message"], True)
def test_poll(self):
web_socket_proxy = self._get_websockproxy()
web_socket_proxy.__dict__["wrap_cmd"] = "wrap_cmd"
web_socket_proxy.__dict__["wrap_mode"] = "respawn"
web_socket_proxy.__dict__["wrap_times"] = [99999999]
web_socket_proxy.__dict__["spawn_message"] = True
web_socket_proxy.__dict__["cmd"] = None
self.stubs.Set(time, 'time', lambda: 100000000.000)
web_socket_proxy.poll()
self.assertEquals(web_socket_proxy.spawn_message, False)
def test_new_client(self):
web_socket_proxy = self._get_websockproxy()
web_socket_proxy.__dict__["verbose"] = "verbose"
web_socket_proxy.__dict__["daemon"] = None
web_socket_proxy.__dict__["client"] = "client"
self.stubs.Set(web_socket_proxy, 'socket', MockSocket)
def mock_select(*args, **kwargs):
ins = None
outs = None
excepts = "excepts"
return ins, outs, excepts
self.stubs.Set(select, 'select', mock_select)
self.assertRaises(Exception, web_socket_proxy.new_websocket_client)
noVNC/utils/websockify/tests/tox.ini
+20
View File
@@ -0,0 +1,20 @@
# Tox (http://tox.testrun.org/) is a tool for running tests
# in multiple virtualenvs. This configuration file will run the
# test suite on all supported python versions. To use it, "pip install tox"
# and then run "tox" from this directory.
[tox]
envlist = py24,py25,py26,py27,py30
setupdir = ../
[testenv]
commands = nosetests {posargs}
deps =
mox
nose
# At some point we should enable this since tox epdctes it to exist but
# the code will need pep8ising first.
#[testenv:pep8]
#commands = flake8
#dep = flake8
noVNC/utils/websockify/tests/utf8-list.py
Executable
+19
View File
@@ -0,0 +1,19 @@
#!/usr/bin/python
'''
Display UTF-8 encoding for 0-255.'''
import sys, os, socket, ssl, time, traceback
from select import select
sys.path.insert(0,os.path.join(os.path.dirname(__file__), ".."))
from websockify.websocket import WebSocketServer
if __name__ == '__main__':
print "val: hybi_base64 | hybi_binary"
for c in range(0, 256):
hybi_base64 = WebSocketServer.encode_hybi(chr(c), opcode=1,
base64=True)
hybi_binary = WebSocketServer.encode_hybi(chr(c), opcode=2,
base64=False)
print "%d: %s | %s" % (c, repr(hybi_base64), repr(hybi_binary))