Skip to content

Commit

Permalink
Convert doctests in test_serverbase.py to real tests in preparation f…
Browse files Browse the repository at this point in the history 
…or adding more. Move the dualchannel test it contained to the proper file.
  • Loading branch information
@jamadden
jamadden committed Oct 27, 2017
1 parent 78e0a0d commit 55ce62f
Show file tree
Hide file tree
Showing 3 changed files with 101 additions and 102 deletions.
18 changes: 8 additions & 10 deletions src/zope/server/serverbase.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -61,7 +61,7 @@ def log(self, message):
'info': logging.INFO,
'error': logging.ERROR,
'warning': logging.WARN,
}
}

def log_info(self, message, type='info'):
"""See zope.server.interfaces.IDispatcherLogging"""
Expand Down Expand Up @@ -95,11 +95,11 @@ def accept_connections(self):
if self.verbose:
self.log_info('%s started.\n'
'\tHostname: %s\n\tPort: %d%s' % (
self.SERVER_IDENT,
self.server_name,
self.port,
self.getExtraLogMessage()
))
self.SERVER_IDENT,
self.server_name,
self.port,
self.getExtraLogMessage()
))

def getExtraLogMessage(self):
r"""Additional information to be logged on startup.
Expand Down Expand Up @@ -128,11 +128,9 @@ def writable(self):

def handle_read(self):
"""See zope.server.interfaces.IDispatcherEventHandler"""
pass

def handle_connect(self):
"""See zope.server.interfaces.IDispatcherEventHandler"""
pass

def handle_accept(self):
"""See zope.server.interfaces.IDispatcherEventHandler"""
Expand All @@ -147,8 +145,8 @@ def handle_accept(self):
# address family is unknown. We don't want the whole server
# to shut down because of this.
if self.adj.log_socket_errors:
self.log_info ('warning: server accept() threw an exception',
'warning')
self.log_info('warning: server accept() threw an exception',
'warning')
return
for (level, optname, value) in self.adj.socket_options:
conn.setsockopt(level, optname, value)
Expand Down
41 changes: 41 additions & 0 deletions src/zope/server/tests/test_dualmodechannel.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -80,3 +80,44 @@ class A(object):
channel = C(None, None, A())
channel.write(b'some bytes')
self.assertTrue(channel.flush_called)


def test_channels_accept_iterables(self):
# Channels accept iterables (they special-case strings).

from zope.server.tests.test_serverbase import FakeSocket
socket = FakeSocket()
channel = DualModeChannel(socket, ('localhost', 42))

written = channel.write(b"First")
self.assertEqual(5, written)

channel.flush()
self.assertEqual(socket.data.decode('ascii'),
'First')

written = channel.write([b"\n", b"Second", b"\n", b"Third"])
self.assertEqual(13, written)

channel.flush()
self.assertEqual(socket.data.decode('ascii'),
"First\n"
"Second\n"
"Third")

def count():
yield b'\n1\n2\n3\n'
yield b'I love to count. Ha ha ha.'

written = channel.write(count())
self.assertEqual(written, 33)

channel.flush()
self.assertEqual(socket.data.decode('ascii'),
"First\n"
"Second\n"
"Third\n"
"1\n"
"2\n"
"3\n"
"I love to count. Ha ha ha.")
144 changes: 52 additions & 92 deletions src/zope/server/tests/test_serverbase.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -13,68 +13,22 @@
##############################################################################
"""Tests for zope.server.serverbase
"""
import doctest
import unittest

from zope.server.serverbase import ServerBase

def doctest_ServerBase():
r"""Regression test for ServerBase

Bug: if the `ip` argument of ServerBase is a string containing a numberic
IP address, and the verbose argument is enabled, ServerBase.__init__
would try to use self.logger before it was initialized.
class FakeSocket(object):
data = b''

We will use a subclass of ServerBase so that unit tests do not actually try
to bind to ports.
def setblocking(self, _val):
return None

>>> from zope.server.serverbase import ServerBase
>>> class ServerBaseForTest(ServerBase):
... def bind(self, addr):
... ip, port = addr
... print("Listening on %s:%d" % (ip or '*', port))
>>> sb = ServerBaseForTest('127.0.0.1', 80, start=False, verbose=True)
Listening on 127.0.0.1:80
def fileno(self):
return 42

"""


def doctest_ServerBase_startup_logging():
r"""Test for ServerBase verbose startup logging
We will use a subclass of ServerBase so that unit tests do not actually try
to bind to ports.
>>> from zope.server.serverbase import ServerBase
>>> class ServerBaseForTest(ServerBase):
... def bind(self, addr):
... self.socket = FakeSocket()
... def log_info(self, message, level='info'):
... print(message.expandtabs())
>>> sb = ServerBaseForTest('example.com', 80, start=True, verbose=True)
zope.server.serverbase started.
Hostname: example.com
Port: 80
Subclasses can add extra information there
>>> class ServerForTest(ServerBaseForTest):
... def getExtraLogMessage(self):
... return '\n\tURL: http://example.com/'
>>> sb = ServerForTest('example.com', 80, start=True, verbose=True)
zope.server.serverbase started.
Hostname: example.com
Port: 80
URL: http://example.com/
"""

class FakeSocket:
data = b''
setblocking = lambda *_: None
fileno = lambda *_: 42
getpeername = lambda *_: ('localhost', 42)
def getpeername(self):
return ('localhost', 42)

def listen(self, *args):
pass
Expand All @@ -84,53 +38,59 @@ def send(self, data):
return len(data)


def channels_accept_iterables():
r"""
Channels accept iterables (they special-case strings).
class TestServerBase(unittest.TestCase):
def test_ServerBase_ip_string_verbose(self):
# Regression test for ServerBase

>>> from zope.server.dualmodechannel import DualModeChannel
>>> socket = FakeSocket()
>>> channel = DualModeChannel(socket, ('localhost', 42))
# Bug: if the `ip` argument of ServerBase is a string
# containing a numeric IP address, and the verbose argument
# is enabled, ServerBase.__init__ would try to use self.logger
# before it was initialized.

>>> channel.write(b"First")
5
# We will use a subclass of ServerBase so that unit tests do
# not actually try to bind to ports.

>>> channel.flush()
>>> print(socket.data.decode('ascii'))
First
bound = []

>>> channel.write([b"\n", b"Second", b"\n", b"Third"])
13
class ServerBaseForTest(ServerBase):
def bind(self, addr):
ip, port = addr
bound.append("Listening on %s:%d" % (ip or '*', port))
ServerBaseForTest('127.0.0.1', 80, start=False, verbose=True)
self.assertEqual(bound,
['Listening on 127.0.0.1:80'])

>>> channel.flush()
>>> print(socket.data.decode('ascii'))
First
Second
Third
def test_ServerBase_startup_logging(self):
# Test for ServerBase verbose startup logging

>>> def count():
... yield b'\n1\n2\n3\n'
... yield b'I love to count. Ha ha ha.'
# We will use a subclass of ServerBase so that unit tests do
# not actually try to bind to ports.

>>> channel.write(count())
33
logs = []

>>> channel.flush()
>>> print(socket.data.decode('ascii'))
First
Second
Third
1
2
3
I love to count. Ha ha ha.
class ServerBaseForTest(ServerBase):
def bind(self, addr):
self.socket = FakeSocket()

"""
def log_info(self, message, type='info'):
logs.append(message.expandtabs())

ServerBaseForTest('example.com', 80, start=True, verbose=True)
self.assertEqual(logs[0],
"zope.server.serverbase started.\n"
" Hostname: example.com\n"
" Port: 80")

def test_suite():
return doctest.DocTestSuite()
# Subclasses can add extra information there

class ServerForTest(ServerBaseForTest):
def getExtraLogMessage(self):
return '\n\tURL: http://example.com/'

if __name__ == '__main__':
unittest.main(defaultTest='test_suite')
del logs[:]
ServerForTest('example.com', 80, start=True, verbose=True)
self.assertEqual(logs[0],
"zope.server.serverbase started.\n"
" Hostname: example.com\n"
" Port: 80\n"
" URL: http://example.com/")

0 comments on commit 55ce62f

Please sign in to comment.