tor-browser

mock.py (7040B)

---

# 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.
"""

from __future__ import absolute_import

import six
import six.moves.queue
from six.moves import range

from pywebsocket3 import common, util
from pywebsocket3.stream import Stream, StreamOptions


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 = b'fake_address'

   def write(self, data):
       """Override mod_python.apache.mp_conn.write.

       data should be bytes when touching this method manually.
       """

       self._write_data.append(data)

   def written_data(self):
       """Get bytes written to this mock."""

       return b''.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 b''
       end_index = self._read_data.find(b'\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 b''
       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 = six.moves.queue.Queue()

   def readline(self):
       """Override mod_python.apache.mp_conn.readline."""
       line = bytearray()
       while True:
           c = self._queue.get()
           line.append(c)
           if c == ord(b'\n'):
               return bytes(line)

   def read(self, length):
       """Override mod_python.apache.mp_conn.read."""

       data = bytearray()
       for unused in range(length):
           data.append(self._queue.get())

       return bytes(data)

   def put_bytes(self, bytes):
       """Put bytes to be read from this mock.

       Args:
           bytes: bytes to be read.
       """

       for byte in six.iterbytes(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 = list(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 = Stream(self, StreamOptions())
       self.ws_close_code = None
       self.ws_close_reason = None
       self.ws_version = common.VERSION_HYBI_LATEST
       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