Trying to make the tests work in Windows and Solaris, everywhere
else just works
This commit is contained in:
parent
62c744e7d1
commit
b4dfafaba6
|
@ -549,17 +549,20 @@ def server(evt):
|
||||||
serv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
serv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||||||
serv.settimeout(3)
|
serv.settimeout(3)
|
||||||
serv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
serv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
||||||
serv.bind(("", 9091))
|
serv.bind(("", 9093))
|
||||||
serv.listen(5)
|
serv.listen(5)
|
||||||
try:
|
try:
|
||||||
conn, addr = serv.accept()
|
conn, addr = serv.accept()
|
||||||
except socket.timeout:
|
|
||||||
pass
|
|
||||||
else:
|
|
||||||
conn.send("1 Hola mundo\n")
|
conn.send("1 Hola mundo\n")
|
||||||
conn.recv(13)
|
cantdata = 0
|
||||||
|
while cantdata < 13:
|
||||||
|
print "len:", cantdata
|
||||||
|
data = conn.recv(13-cantdata)
|
||||||
|
cantdata += len(data)
|
||||||
conn.send("2 No more lines\n")
|
conn.send("2 No more lines\n")
|
||||||
conn.close()
|
conn.close()
|
||||||
|
except socket.timeout:
|
||||||
|
pass
|
||||||
finally:
|
finally:
|
||||||
serv.close()
|
serv.close()
|
||||||
evt.set()
|
evt.set()
|
||||||
|
@ -567,7 +570,7 @@ def server(evt):
|
||||||
class FTPWrapperTests(unittest.TestCase):
|
class FTPWrapperTests(unittest.TestCase):
|
||||||
|
|
||||||
def setUp(self):
|
def setUp(self):
|
||||||
ftplib.FTP.port = 9091
|
ftplib.FTP.port = 9093
|
||||||
self.evt = threading.Event()
|
self.evt = threading.Event()
|
||||||
threading.Thread(target=server, args=(self.evt,)).start()
|
threading.Thread(target=server, args=(self.evt,)).start()
|
||||||
time.sleep(.1)
|
time.sleep(.1)
|
||||||
|
@ -577,28 +580,27 @@ class FTPWrapperTests(unittest.TestCase):
|
||||||
|
|
||||||
def testBasic(self):
|
def testBasic(self):
|
||||||
# connects
|
# connects
|
||||||
ftp = urllib.ftpwrapper("myuser", "mypass", "localhost", 9091, [])
|
ftp = urllib.ftpwrapper("myuser", "mypass", "localhost", 9093, [])
|
||||||
ftp.ftp.sock.close()
|
ftp.ftp.sock.close()
|
||||||
|
|
||||||
def testTimeoutDefault(self):
|
def testTimeoutDefault(self):
|
||||||
# default
|
# default
|
||||||
ftp = urllib.ftpwrapper("myuser", "mypass", "localhost", 9091, [])
|
ftp = urllib.ftpwrapper("myuser", "mypass", "localhost", 9093, [])
|
||||||
self.assertTrue(ftp.ftp.sock.gettimeout() is None)
|
self.assertTrue(ftp.ftp.sock.gettimeout() is None)
|
||||||
ftp.ftp.sock.close()
|
ftp.ftp.sock.close()
|
||||||
|
|
||||||
def testTimeoutValue(self):
|
def testTimeoutValue(self):
|
||||||
# a value
|
# a value
|
||||||
ftp = urllib.ftpwrapper("myuser", "mypass", "localhost", 9091, [], timeout=30)
|
ftp = urllib.ftpwrapper("myuser", "mypass", "localhost", 9093, [], timeout=30)
|
||||||
self.assertEqual(ftp.ftp.sock.gettimeout(), 30)
|
self.assertEqual(ftp.ftp.sock.gettimeout(), 30)
|
||||||
ftp.ftp.sock.close()
|
ftp.ftp.sock.close()
|
||||||
|
|
||||||
|
|
||||||
def testTimeoutNone(self):
|
def testTimeoutNone(self):
|
||||||
# None, having other default
|
# None, having other default
|
||||||
previous = socket.getdefaulttimeout()
|
previous = socket.getdefaulttimeout()
|
||||||
socket.setdefaulttimeout(30)
|
socket.setdefaulttimeout(30)
|
||||||
try:
|
try:
|
||||||
ftp = urllib.ftpwrapper("myuser", "mypass", "localhost", 9091, [], timeout=30)
|
ftp = urllib.ftpwrapper("myuser", "mypass", "localhost", 9093, [])
|
||||||
finally:
|
finally:
|
||||||
socket.setdefaulttimeout(previous)
|
socket.setdefaulttimeout(previous)
|
||||||
self.assertEqual(ftp.ftp.sock.gettimeout(), 30)
|
self.assertEqual(ftp.ftp.sock.gettimeout(), 30)
|
||||||
|
|
Loading…
Reference in New Issue