Upstream version 10.39.225.0

[platform/framework/web/crosswalk.git] / src / tools / telemetry / telemetry / core / backends / chrome / websocket_unittest.py

diff --git a/src/tools/telemetry/telemetry/core/backends/chrome/websocket_unittest.py b/src/tools/telemetry/telemetry/core/backends/chrome/websocket_unittest.py
index 66f79b3..e72068d 100644 (file)
--- a/src/tools/telemetry/telemetry/core/backends/chrome/websocket_unittest.py
+++ b/src/tools/telemetry/telemetry/core/backends/chrome/websocket_unittest.py
@@ -1,11 +1,35 @@
 # Copyright 2013 The Chromium Authors. All rights reserved.
 # Use of this source code is governed by a BSD-style license that can be
 # found in the LICENSE file.
+
+import base64
+import BaseHTTPServer
+import hashlib
 import socket
+import threading
 import unittest
 
 from telemetry.core.backends.chrome import websocket
 
+
+# Minimal handler for a local websocket server.
+class _FakeWebSocketHandler(BaseHTTPServer.BaseHTTPRequestHandler):
+  def do_GET(self):
+    key = self.headers.getheader('Sec-WebSocket-Key')
+
+    value = key + '258EAFA5-E914-47DA-95CA-C5AB0DC85B11'
+    hashed = base64.encodestring(hashlib.sha1(value).digest()).strip().lower()
+
+    self.send_response(101)
+
+    self.send_header('Sec-Websocket-Accept', hashed)
+    self.send_header('upgrade', 'websocket')
+    self.send_header('connection', 'upgrade')
+    self.end_headers()
+
+    self.wfile.flush()
+
+
 class TestWebSocket(unittest.TestCase):
   def testExports(self):
     self.assertNotEqual(websocket.create_connection, None)
@@ -13,13 +37,25 @@ class TestWebSocket(unittest.TestCase):
     self.assertNotEqual(websocket.WebSocketTimeoutException, None)
 
   def testSockOpts(self):
-    ws = websocket.create_connection('ws://echo.websocket.org')
-    self.assertNotEquals(
-        ws.sock.getsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR), 0)
+    httpd = BaseHTTPServer.HTTPServer(('127.0.0.1', 0), _FakeWebSocketHandler)
+    ws_url = 'ws://127.0.0.1:%d' % httpd.server_port
+
+    threading.Thread(target=httpd.handle_request).start()
+    ws = websocket.create_connection(ws_url)
+    try:
+      self.assertNotEquals(
+          ws.sock.getsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR), 0)
+    finally:
+      ws.close()
+
+    threading.Thread(target=httpd.handle_request).start()
     ws = websocket.create_connection(
-        'ws://echo.websocket.org',
+        ws_url,
         sockopt=[(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)])
-    self.assertNotEquals(
-        ws.sock.getsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR), 0)
-    self.assertNotEquals(
-        ws.sock.getsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY), 0)
+    try:
+      self.assertNotEquals(
+          ws.sock.getsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR), 0)
+      self.assertNotEquals(
+          ws.sock.getsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY), 0)
+    finally:
+      ws.close()