Update web-platform-tests to revision e92532746b7615dcccdfa060937a87664816b1db

tests/wpt/web-platform-tests/tools/pywebsocket/test/mock.py

@@ -0,0 +1,221 @@
+# Copyright 2011, Google Inc.
+# All rights reserved.
+#
+# Redistribution and use in source and binary forms, with or without
+# modification, are permitted provided that the following conditions are
+# met:
+#
+#     * Redistributions of source code must retain the above copyright
+# notice, this list of conditions and the following disclaimer.
+#     * Redistributions in binary form must reproduce the above
+# copyright notice, this list of conditions and the following disclaimer
+# in the documentation and/or other materials provided with the
+# distribution.
+#     * Neither the name of Google Inc. nor the names of its
+# contributors may be used to endorse or promote products derived from
+# this software without specific prior written permission.
+#
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+# OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+# DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+# THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+
+"""Mocks for testing.
+"""
+
+
+import Queue
+import threading
+
+from mod_pywebsocket import common
+from mod_pywebsocket.stream import StreamHixie75
+
+
+class _MockConnBase(object):
+    """Base class of mocks for mod_python.apache.mp_conn.
+
+    This enables tests to check what is written to a (mock) mp_conn.
+    """
+
+    def __init__(self):
+        self._write_data = []
+        self.remote_addr = 'fake_address'
+
+    def write(self, data):
+        """Override mod_python.apache.mp_conn.write."""
+
+        self._write_data.append(data)
+
+    def written_data(self):
+        """Get bytes written to this mock."""
+
+        return ''.join(self._write_data)
+
+
+class MockConn(_MockConnBase):
+    """Mock for mod_python.apache.mp_conn.
+
+    This enables tests to specify what should be read from a (mock) mp_conn as
+    well as to check what is written to it.
+    """
+
+    def __init__(self, read_data):
+        """Constructs an instance.
+
+        Args:
+            read_data: bytes that should be returned when read* methods are
+            called.
+        """
+
+        _MockConnBase.__init__(self)
+        self._read_data = read_data
+        self._read_pos = 0
+
+    def readline(self):
+        """Override mod_python.apache.mp_conn.readline."""
+
+        if self._read_pos >= len(self._read_data):
+            return ''
+        end_index = self._read_data.find('\n', self._read_pos) + 1
+        if not end_index:
+            end_index = len(self._read_data)
+        return self._read_up_to(end_index)
+
+    def read(self, length):
+        """Override mod_python.apache.mp_conn.read."""
+
+        if self._read_pos >= len(self._read_data):
+            return ''
+        end_index = min(len(self._read_data), self._read_pos + length)
+        return self._read_up_to(end_index)
+
+    def _read_up_to(self, end_index):
+        line = self._read_data[self._read_pos:end_index]
+        self._read_pos = end_index
+        return line
+
+
+class MockBlockingConn(_MockConnBase):
+    """Blocking mock for mod_python.apache.mp_conn.
+
+    This enables tests to specify what should be read from a (mock) mp_conn as
+    well as to check what is written to it.
+    Callers of read* methods will block if there is no bytes available.
+    """
+
+    def __init__(self):
+        _MockConnBase.__init__(self)
+        self._queue = Queue.Queue()
+
+    def readline(self):
+        """Override mod_python.apache.mp_conn.readline."""
+        line = ''
+        while True:
+            c = self._queue.get()
+            line += c
+            if c == '\n':
+                return line
+
+    def read(self, length):
+        """Override mod_python.apache.mp_conn.read."""
+
+        data = ''
+        for unused in range(length):
+            data += self._queue.get()
+        return data
+
+    def put_bytes(self, bytes):
+        """Put bytes to be read from this mock.
+
+        Args:
+            bytes: bytes to be read.
+        """
+
+        for byte in bytes:
+            self._queue.put(byte)
+
+
+class MockTable(dict):
+    """Mock table.
+
+    This mimics mod_python mp_table. Note that only the methods used by
+    tests are overridden.
+    """
+
+    def __init__(self, copy_from={}):
+        if isinstance(copy_from, dict):
+            copy_from = copy_from.items()
+        for key, value in copy_from:
+            self.__setitem__(key, value)
+
+    def __getitem__(self, key):
+        return super(MockTable, self).__getitem__(key.lower())
+
+    def __setitem__(self, key, value):
+        super(MockTable, self).__setitem__(key.lower(), value)
+
+    def get(self, key, def_value=None):
+        return super(MockTable, self).get(key.lower(), def_value)
+
+
+class MockRequest(object):
+    """Mock request.
+
+    This mimics mod_python request.
+    """
+
+    def __init__(self, uri=None, headers_in={}, connection=None, method='GET',
+                 protocol='HTTP/1.1', is_https=False):
+        """Construct an instance.
+
+        Arguments:
+            uri: URI of the request.
+            headers_in: Request headers.
+            connection: Connection used for the request.
+            method: request method.
+            is_https: Whether this request is over SSL.
+
+        See the document of mod_python Request for details.
+        """
+        self.uri = uri
+        self.unparsed_uri = uri
+        self.connection = connection
+        self.method = method
+        self.protocol = protocol
+        self.headers_in = MockTable(headers_in)
+        # self.is_https_ needs to be accessible from tests.  To avoid name
+        # conflict with self.is_https(), it is named as such.
+        self.is_https_ = is_https
+        self.ws_stream = StreamHixie75(self, True)
+        self.ws_close_code = None
+        self.ws_close_reason = None
+        self.ws_version = common.VERSION_HYBI00
+        self.ws_deflate = False
+
+    def is_https(self):
+        """Return whether this request is over SSL."""
+        return self.is_https_
+
+
+class MockDispatcher(object):
+    """Mock for dispatch.Dispatcher."""
+
+    def __init__(self):
+        self.do_extra_handshake_called = False
+
+    def do_extra_handshake(self, conn_context):
+        self.do_extra_handshake_called = True
+
+    def transfer_data(self, conn_context):
+        pass
+
+
+# vi:sts=4 sw=4 et