mirror of
https://github.com/SickGear/SickGear.git
synced 2024-12-11 13:43:37 +00:00
146 lines
5 KiB
Python
146 lines
5 KiB
Python
from __future__ import absolute_import, division, print_function, with_statement
|
|
|
|
import signal
|
|
import socket
|
|
from subprocess import Popen
|
|
import sys
|
|
import time
|
|
|
|
from tornado.netutil import BlockingResolver, ThreadedResolver, is_valid_ip
|
|
from tornado.stack_context import ExceptionStackContext
|
|
from tornado.testing import AsyncTestCase, gen_test
|
|
from tornado.test.util import unittest
|
|
|
|
try:
|
|
from concurrent import futures
|
|
except ImportError:
|
|
futures = None
|
|
|
|
try:
|
|
import pycares
|
|
except ImportError:
|
|
pycares = None
|
|
else:
|
|
from tornado.platform.caresresolver import CaresResolver
|
|
|
|
try:
|
|
import twisted
|
|
except ImportError:
|
|
twisted = None
|
|
else:
|
|
from tornado.platform.twisted import TwistedResolver
|
|
|
|
|
|
class _ResolverTestMixin(object):
|
|
def skipOnCares(self):
|
|
# Some DNS-hijacking ISPs (e.g. Time Warner) return non-empty results
|
|
# with an NXDOMAIN status code. Most resolvers treat this as an error;
|
|
# C-ares returns the results, making the "bad_host" tests unreliable.
|
|
# C-ares will try to resolve even malformed names, such as the
|
|
# name with spaces used in this test.
|
|
if self.resolver.__class__.__name__ == 'CaresResolver':
|
|
self.skipTest("CaresResolver doesn't recognize fake NXDOMAIN")
|
|
|
|
def test_localhost(self):
|
|
self.resolver.resolve('localhost', 80, callback=self.stop)
|
|
result = self.wait()
|
|
self.assertIn((socket.AF_INET, ('127.0.0.1', 80)), result)
|
|
|
|
@gen_test
|
|
def test_future_interface(self):
|
|
addrinfo = yield self.resolver.resolve('localhost', 80,
|
|
socket.AF_UNSPEC)
|
|
self.assertIn((socket.AF_INET, ('127.0.0.1', 80)),
|
|
addrinfo)
|
|
|
|
def test_bad_host(self):
|
|
self.skipOnCares()
|
|
def handler(exc_typ, exc_val, exc_tb):
|
|
self.stop(exc_val)
|
|
return True # Halt propagation.
|
|
|
|
with ExceptionStackContext(handler):
|
|
self.resolver.resolve('an invalid domain', 80, callback=self.stop)
|
|
|
|
result = self.wait()
|
|
self.assertIsInstance(result, Exception)
|
|
|
|
@gen_test
|
|
def test_future_interface_bad_host(self):
|
|
self.skipOnCares()
|
|
with self.assertRaises(Exception):
|
|
yield self.resolver.resolve('an invalid domain', 80,
|
|
socket.AF_UNSPEC)
|
|
|
|
|
|
class BlockingResolverTest(AsyncTestCase, _ResolverTestMixin):
|
|
def setUp(self):
|
|
super(BlockingResolverTest, self).setUp()
|
|
self.resolver = BlockingResolver(io_loop=self.io_loop)
|
|
|
|
|
|
@unittest.skipIf(futures is None, "futures module not present")
|
|
class ThreadedResolverTest(AsyncTestCase, _ResolverTestMixin):
|
|
def setUp(self):
|
|
super(ThreadedResolverTest, self).setUp()
|
|
self.resolver = ThreadedResolver(io_loop=self.io_loop)
|
|
|
|
def tearDown(self):
|
|
self.resolver.close()
|
|
super(ThreadedResolverTest, self).tearDown()
|
|
|
|
|
|
@unittest.skipIf(futures is None, "futures module not present")
|
|
class ThreadedResolverImportTest(unittest.TestCase):
|
|
def test_import(self):
|
|
TIMEOUT = 5
|
|
|
|
# Test for a deadlock when importing a module that runs the
|
|
# ThreadedResolver at import-time. See resolve_test.py for
|
|
# full explanation.
|
|
command = [
|
|
sys.executable,
|
|
'-c',
|
|
'import tornado.test.resolve_test_helper']
|
|
|
|
start = time.time()
|
|
popen = Popen(command, preexec_fn=lambda: signal.alarm(TIMEOUT))
|
|
while time.time() - start < TIMEOUT:
|
|
return_code = popen.poll()
|
|
if return_code is not None:
|
|
self.assertEqual(0, return_code)
|
|
return # Success.
|
|
time.sleep(0.05)
|
|
|
|
self.fail("import timed out")
|
|
|
|
|
|
@unittest.skipIf(pycares is None, "pycares module not present")
|
|
class CaresResolverTest(AsyncTestCase, _ResolverTestMixin):
|
|
def setUp(self):
|
|
super(CaresResolverTest, self).setUp()
|
|
self.resolver = CaresResolver(io_loop=self.io_loop)
|
|
|
|
|
|
@unittest.skipIf(twisted is None, "twisted module not present")
|
|
@unittest.skipIf(getattr(twisted, '__version__', '0.0') < "12.1", "old version of twisted")
|
|
class TwistedResolverTest(AsyncTestCase, _ResolverTestMixin):
|
|
def setUp(self):
|
|
super(TwistedResolverTest, self).setUp()
|
|
self.resolver = TwistedResolver(io_loop=self.io_loop)
|
|
|
|
|
|
class IsValidIPTest(unittest.TestCase):
|
|
def test_is_valid_ip(self):
|
|
self.assertTrue(is_valid_ip('127.0.0.1'))
|
|
self.assertTrue(is_valid_ip('4.4.4.4'))
|
|
self.assertTrue(is_valid_ip('::1'))
|
|
self.assertTrue(is_valid_ip('2620:0:1cfe:face:b00c::3'))
|
|
self.assertTrue(not is_valid_ip('www.google.com'))
|
|
self.assertTrue(not is_valid_ip('localhost'))
|
|
self.assertTrue(not is_valid_ip('4.4.4.4<'))
|
|
self.assertTrue(not is_valid_ip(' 127.0.0.1'))
|
|
self.assertTrue(not is_valid_ip(''))
|
|
self.assertTrue(not is_valid_ip(' '))
|
|
self.assertTrue(not is_valid_ip('\n'))
|
|
self.assertTrue(not is_valid_ip('\x00'))
|