test_nntplib.py 1.6 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273
  1. import socket
  2. import nntplib
  3. import time
  4. import unittest
  5. try:
  6. import threading
  7. except ImportError:
  8. threading = None
  9. from unittest import TestCase
  10. from test import test_support
  11. HOST = test_support.HOST
  12. def server(evt, serv, evil=False):
  13. serv.listen(5)
  14. try:
  15. conn, addr = serv.accept()
  16. except socket.timeout:
  17. pass
  18. else:
  19. if evil:
  20. conn.send("1 I'm too long response" * 3000 + "\n")
  21. else:
  22. conn.send("1 I'm OK response\n")
  23. conn.close()
  24. finally:
  25. serv.close()
  26. evt.set()
  27. class BaseServerTest(TestCase):
  28. def setUp(self):
  29. self.evt = threading.Event()
  30. self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
  31. self.sock.settimeout(3)
  32. self.port = test_support.bind_port(self.sock)
  33. threading.Thread(
  34. target=server,
  35. args=(self.evt, self.sock, self.evil)).start()
  36. time.sleep(.1)
  37. def tearDown(self):
  38. self.evt.wait()
  39. @unittest.skipUnless(threading, 'threading required')
  40. class ServerTests(BaseServerTest):
  41. evil = False
  42. def test_basic_connect(self):
  43. nntp = nntplib.NNTP('localhost', self.port)
  44. nntp.sock.close()
  45. @unittest.skipUnless(threading, 'threading required')
  46. class EvilServerTests(BaseServerTest):
  47. evil = True
  48. def test_too_long_line(self):
  49. self.assertRaises(nntplib.NNTPDataError,
  50. nntplib.NNTP, 'localhost', self.port)
  51. def test_main(verbose=None):
  52. test_support.run_unittest(EvilServerTests)
  53. test_support.run_unittest(ServerTests)
  54. if __name__ == '__main__':
  55. test_main()