aboutsummaryrefslogtreecommitdiff
path: root/dummyserver/testcase.py
blob: 35769efb2be378d8af2e40a69e5f88c6e768574b (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
import unittest
import socket
import threading
from nose.plugins.skip import SkipTest
from tornado import ioloop, web, wsgi

from dummyserver.server import (
    SocketServerThread,
    run_tornado_app,
    run_loop_in_thread,
    DEFAULT_CERTS,
)
from dummyserver.handlers import TestingApp
from dummyserver.proxy import ProxyHandler



class SocketDummyServerTestCase(unittest.TestCase):
    """
    A simple socket-based server is created for this class that is good for
    exactly one request.
    """
    scheme = 'http'
    host = 'localhost'

    @classmethod
    def _start_server(cls, socket_handler):
        ready_event = threading.Event()
        cls.server_thread = SocketServerThread(socket_handler=socket_handler,
                                               ready_event=ready_event,
                                               host=cls.host)
        cls.server_thread.start()
        ready_event.wait()
        cls.port = cls.server_thread.port

    @classmethod
    def tearDownClass(cls):
        if hasattr(cls, 'server_thread'):
            cls.server_thread.join(0.1)


class HTTPDummyServerTestCase(unittest.TestCase):
    scheme = 'http'
    host = 'localhost'
    host_alt = '127.0.0.1'  # Some tests need two hosts
    certs = DEFAULT_CERTS

    @classmethod
    def _start_server(cls):
        cls.io_loop = ioloop.IOLoop()
        app = wsgi.WSGIContainer(TestingApp())
        cls.server, cls.port = run_tornado_app(app, cls.io_loop, cls.certs,
                                               cls.scheme, cls.host)
        cls.server_thread = run_loop_in_thread(cls.io_loop)

    @classmethod
    def _stop_server(cls):
        cls.io_loop.add_callback(cls.server.stop)
        cls.io_loop.add_callback(cls.io_loop.stop)
        cls.server_thread.join()

    @classmethod
    def setUpClass(cls):
        cls._start_server()

    @classmethod
    def tearDownClass(cls):
        cls._stop_server()


class HTTPSDummyServerTestCase(HTTPDummyServerTestCase):
    scheme = 'https'
    host = 'localhost'
    certs = DEFAULT_CERTS


class HTTPDummyProxyTestCase(unittest.TestCase):

    http_host = 'localhost'
    http_host_alt = '127.0.0.1'

    https_host = 'localhost'
    https_host_alt = '127.0.0.1'
    https_certs = DEFAULT_CERTS

    proxy_host = 'localhost'
    proxy_host_alt = '127.0.0.1'

    @classmethod
    def setUpClass(cls):
        cls.io_loop = ioloop.IOLoop()

        app = wsgi.WSGIContainer(TestingApp())
        cls.http_server, cls.http_port = run_tornado_app(
            app, cls.io_loop, None, 'http', cls.http_host)

        app = wsgi.WSGIContainer(TestingApp())
        cls.https_server, cls.https_port = run_tornado_app(
            app, cls.io_loop, cls.https_certs, 'https', cls.http_host)

        app = web.Application([(r'.*', ProxyHandler)])
        cls.proxy_server, cls.proxy_port = run_tornado_app(
            app, cls.io_loop, None, 'http', cls.proxy_host)

        cls.server_thread = run_loop_in_thread(cls.io_loop)

    @classmethod
    def tearDownClass(cls):
        cls.io_loop.add_callback(cls.http_server.stop)
        cls.io_loop.add_callback(cls.https_server.stop)
        cls.io_loop.add_callback(cls.proxy_server.stop)
        cls.io_loop.add_callback(cls.io_loop.stop)
        cls.server_thread.join()


class IPv6HTTPDummyServerTestCase(HTTPDummyServerTestCase):
    host = '::1'

    @classmethod
    def setUpClass(cls):
        if not socket.has_ipv6:
            raise SkipTest('IPv6 not available')
        else:
            super(IPv6HTTPDummyServerTestCase, cls).setUpClass()