12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273 |
- import socket
- import nntplib
- import time
- import unittest
- try:
- import threading
- except ImportError:
- threading = None
- from unittest import TestCase
- from test import test_support
- HOST = test_support.HOST
- def server(evt, serv, evil=False):
- serv.listen(5)
- try:
- conn, addr = serv.accept()
- except socket.timeout:
- pass
- else:
- if evil:
- conn.send("1 I'm too long response" * 3000 + "\n")
- else:
- conn.send("1 I'm OK response\n")
- conn.close()
- finally:
- serv.close()
- evt.set()
- class BaseServerTest(TestCase):
- def setUp(self):
- self.evt = threading.Event()
- self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
- self.sock.settimeout(3)
- self.port = test_support.bind_port(self.sock)
- threading.Thread(
- target=server,
- args=(self.evt, self.sock, self.evil)).start()
- time.sleep(.1)
- def tearDown(self):
- self.evt.wait()
- @unittest.skipUnless(threading, 'threading required')
- class ServerTests(BaseServerTest):
- evil = False
- def test_basic_connect(self):
- nntp = nntplib.NNTP('localhost', self.port)
- nntp.sock.close()
- @unittest.skipUnless(threading, 'threading required')
- class EvilServerTests(BaseServerTest):
- evil = True
- def test_too_long_line(self):
- self.assertRaises(nntplib.NNTPDataError,
- nntplib.NNTP, 'localhost', self.port)
- def test_main(verbose=None):
- test_support.run_unittest(EvilServerTests)
- test_support.run_unittest(ServerTests)
- if __name__ == '__main__':
- test_main()
|