codekingpro/portable-devtools

1import asyncio
2import concurrent.futures
3import threading
4
5from wsgiref.validate import validator
6
7from tornado.routing import RuleRouter
8from tornado.testing import AsyncHTTPTestCase, gen_test
9from tornado.wsgi import WSGIContainer
10
11
12class WSGIAppMixin:
13    # TODO: Now that WSGIAdapter is gone, this is a pretty weak test.
14    def get_executor(self):
15        raise NotImplementedError()
16
17    def get_app(self):
18        executor = self.get_executor()
19        # The barrier test in DummyExecutorTest will always wait the full
20        # value of this timeout, so we don't want it to be too high.
21        self.barrier = threading.Barrier(2, timeout=0.3)
22
23        def make_container(app):
24            return WSGIContainer(validator(app), executor=executor)
25
26        return RuleRouter(
27            [
28                ("/simple", make_container(self.simple_wsgi_app)),
29                ("/barrier", make_container(self.barrier_wsgi_app)),
30                ("/streaming_barrier", make_container(self.streaming_barrier_wsgi_app)),
31            ]
32        )
33
34    def respond_plain(self, start_response):
35        status = "200 OK"
36        response_headers = [("Content-Type", "text/plain")]
37        start_response(status, response_headers)
38
39    def simple_wsgi_app(self, environ, start_response):
40        self.respond_plain(start_response)
41        return [b"Hello world!"]
42
43    def barrier_wsgi_app(self, environ, start_response):
44        self.respond_plain(start_response)
45        try:
46            n = self.barrier.wait()
47        except threading.BrokenBarrierError:
48            return [b"broken barrier"]
49        else:
50            return [b"ok %d" % n]
51
52    def streaming_barrier_wsgi_app(self, environ, start_response):
53        self.respond_plain(start_response)
54        yield b"ok "
55        try:
56            n = self.barrier.wait()
57        except threading.BrokenBarrierError:
58            yield b"broken barrier"
59        else:
60            yield b"%d" % n
61
62
63class WSGIContainerDummyExecutorTest(WSGIAppMixin, AsyncHTTPTestCase):
64    def get_executor(self):
65        return None
66
67    def test_simple(self):
68        response = self.fetch("/simple")
69        self.assertEqual(response.body, b"Hello world!")
70
71    @gen_test
72    async def test_concurrent_barrier(self):
73        self.barrier.reset()
74        resps = await asyncio.gather(
75            self.http_client.fetch(self.get_url("/barrier")),
76            self.http_client.fetch(self.get_url("/barrier")),
77        )
78        for resp in resps:
79            self.assertEqual(resp.body, b"broken barrier")
80
81    @gen_test
82    async def test_concurrent_streaming_barrier(self):
83        self.barrier.reset()
84        resps = await asyncio.gather(
85            self.http_client.fetch(self.get_url("/streaming_barrier")),
86            self.http_client.fetch(self.get_url("/streaming_barrier")),
87        )
88        for resp in resps:
89            self.assertEqual(resp.body, b"ok broken barrier")
90
91
92class WSGIContainerThreadPoolTest(WSGIAppMixin, AsyncHTTPTestCase):
93    def get_executor(self):
94        return concurrent.futures.ThreadPoolExecutor()
95
96    def test_simple(self):
97        response = self.fetch("/simple")
98        self.assertEqual(response.body, b"Hello world!")
99
100    @gen_test
101    async def test_concurrent_barrier(self):
102        self.barrier.reset()
103        resps = await asyncio.gather(
104            self.http_client.fetch(self.get_url("/barrier")),
105            self.http_client.fetch(self.get_url("/barrier")),
106        )
107        self.assertEqual([b"ok 0", b"ok 1"], sorted([resp.body for resp in resps]))
108
109    @gen_test
110    async def test_concurrent_streaming_barrier(self):
111        self.barrier.reset()
112        resps = await asyncio.gather(
113            self.http_client.fetch(self.get_url("/streaming_barrier")),
114            self.http_client.fetch(self.get_url("/streaming_barrier")),
115        )
116        self.assertEqual([b"ok 0", b"ok 1"], sorted([resp.body for resp in resps]))
117