gecko-dev/testing/web-platform/tests/tools/pywebsocket/test/mock.py

# 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