Introduce the mock module to improve unit tests.
There are a few socket corner-cases that are incredibly hard to reproduce in a unit test suite, so we use mock to trigger the exceptions instead.
This commit is contained in:
parent
cc4867064b
commit
e5b125eec8
|
@ -87,6 +87,9 @@ class _FileLike:
|
||||||
|
|
||||||
class Writer(_FileLike):
|
class Writer(_FileLike):
|
||||||
def flush(self):
|
def flush(self):
|
||||||
|
"""
|
||||||
|
May raise NetLibDisconnect
|
||||||
|
"""
|
||||||
if hasattr(self.o, "flush"):
|
if hasattr(self.o, "flush"):
|
||||||
try:
|
try:
|
||||||
self.o.flush()
|
self.o.flush()
|
||||||
|
@ -94,6 +97,9 @@ class Writer(_FileLike):
|
||||||
raise NetLibDisconnect(str(v))
|
raise NetLibDisconnect(str(v))
|
||||||
|
|
||||||
def write(self, v):
|
def write(self, v):
|
||||||
|
"""
|
||||||
|
May raise NetLibDisconnect
|
||||||
|
"""
|
||||||
if v:
|
if v:
|
||||||
try:
|
try:
|
||||||
if hasattr(self.o, "sendall"):
|
if hasattr(self.o, "sendall"):
|
||||||
|
|
|
@ -1,5 +1,6 @@
|
||||||
import cStringIO, threading, Queue, time
|
import cStringIO, threading, Queue, time, socket
|
||||||
from netlib import tcp, certutils, test
|
from netlib import tcp, certutils, test
|
||||||
|
import mock
|
||||||
import tutils
|
import tutils
|
||||||
|
|
||||||
class SNIHandler(tcp.BaseHandler):
|
class SNIHandler(tcp.BaseHandler):
|
||||||
|
@ -275,6 +276,22 @@ class TestFileLike:
|
||||||
s.write("x")
|
s.write("x")
|
||||||
assert s.get_log() == "xx"
|
assert s.get_log() == "xx"
|
||||||
|
|
||||||
|
def test_writer_flush_error(self):
|
||||||
|
s = cStringIO.StringIO()
|
||||||
|
s = tcp.Writer(s)
|
||||||
|
o = mock.MagicMock()
|
||||||
|
o.flush = mock.MagicMock(side_effect=socket.error)
|
||||||
|
s.o = o
|
||||||
|
tutils.raises(tcp.NetLibDisconnect, s.flush)
|
||||||
|
|
||||||
|
def test_reader_read_error(self):
|
||||||
|
s = cStringIO.StringIO("foobar\nfoobar")
|
||||||
|
s = tcp.Reader(s)
|
||||||
|
o = mock.MagicMock()
|
||||||
|
o.read = mock.MagicMock(side_effect=socket.error)
|
||||||
|
s.o = o
|
||||||
|
tutils.raises(tcp.NetLibDisconnect, s.read, 10)
|
||||||
|
|
||||||
def test_reset_timestamps(self):
|
def test_reset_timestamps(self):
|
||||||
s = cStringIO.StringIO("foobar\nfoobar")
|
s = cStringIO.StringIO("foobar\nfoobar")
|
||||||
s = tcp.Reader(s)
|
s = tcp.Reader(s)
|
||||||
|
|
Loading…
Reference in New Issue