test_pipes.py (10428B)
1 import os 2 import unittest 3 import time 4 import json 5 import urllib 6 7 import pytest 8 9 wptserve = pytest.importorskip("wptserve") 10 from .base import TestUsingServer, doc_root 11 12 13 class TestStatus(TestUsingServer): 14 def test_status(self): 15 resp = self.request("/document.txt", query="pipe=status(202)") 16 self.assertEqual(resp.getcode(), 202) 17 18 class TestHeader(TestUsingServer): 19 def test_not_set(self): 20 resp = self.request("/document.txt", query="pipe=header(X-TEST,PASS)") 21 self.assertEqual(resp.info()["X-TEST"], "PASS") 22 23 def test_set(self): 24 resp = self.request("/document.txt", query="pipe=header(Content-Type,text/html)") 25 self.assertEqual(resp.info()["Content-Type"], "text/html") 26 27 def test_multiple(self): 28 resp = self.request("/document.txt", query="pipe=header(X-Test,PASS)|header(Content-Type,text/html)") 29 self.assertEqual(resp.info()["X-TEST"], "PASS") 30 self.assertEqual(resp.info()["Content-Type"], "text/html") 31 32 def test_multiple_same(self): 33 resp = self.request("/document.txt", query="pipe=header(Content-Type,FAIL)|header(Content-Type,text/html)") 34 self.assertEqual(resp.info()["Content-Type"], "text/html") 35 36 def test_multiple_append(self): 37 resp = self.request("/document.txt", query="pipe=header(X-Test,1)|header(X-Test,2,True)") 38 self.assert_multiple_headers(resp, "X-Test", ["1", "2"]) 39 40 def test_semicolon(self): 41 resp = self.request("/document.txt", query="pipe=header(Refresh,3;url=http://example.com)") 42 self.assertEqual(resp.info()["Refresh"], "3;url=http://example.com") 43 44 def test_escape_comma(self): 45 resp = self.request("/document.txt", query=r"pipe=header(Expires,Thu\,%2014%20Aug%201986%2018:00:00%20GMT)") 46 self.assertEqual(resp.info()["Expires"], "Thu, 14 Aug 1986 18:00:00 GMT") 47 48 def test_escape_parenthesis(self): 49 resp = self.request("/document.txt", query=r"pipe=header(User-Agent,Mozilla/5.0%20(X11;%20Linux%20x86_64;%20rv:12.0\)") 50 self.assertEqual(resp.info()["User-Agent"], "Mozilla/5.0 (X11; Linux x86_64; rv:12.0)") 51 52 class TestSlice(TestUsingServer): 53 def test_both_bounds(self): 54 resp = self.request("/document.txt", query="pipe=slice(1,10)") 55 with open(os.path.join(doc_root, "document.txt"), 'rb') as f: 56 expected = f.read() 57 self.assertEqual(resp.read(), expected[1:10]) 58 59 def test_no_upper(self): 60 resp = self.request("/document.txt", query="pipe=slice(1)") 61 with open(os.path.join(doc_root, "document.txt"), 'rb') as f: 62 expected = f.read() 63 self.assertEqual(resp.read(), expected[1:]) 64 65 def test_no_lower(self): 66 resp = self.request("/document.txt", query="pipe=slice(null,10)") 67 with open(os.path.join(doc_root, "document.txt"), 'rb') as f: 68 expected = f.read() 69 self.assertEqual(resp.read(), expected[:10]) 70 71 class TestSub(TestUsingServer): 72 def test_sub_config(self): 73 resp = self.request("/sub.txt", query="pipe=sub") 74 expected = b"localhost localhost %i" % self.server.port 75 self.assertEqual(resp.read().rstrip(), expected) 76 77 def test_sub_file_hash(self): 78 resp = self.request("/sub_file_hash.sub.txt") 79 expected = b""" 80 md5: JmI1W8fMHfSfCarYOSxJcw== 81 sha1: nqpWqEw4IW8NjD6R375gtrQvtTo= 82 sha224: RqQ6fMmta6n9TuA/vgTZK2EqmidqnrwBAmQLRQ== 83 sha256: G6Ljg1uPejQxqFmvFOcV/loqnjPTW5GSOePOfM/u0jw= 84 sha384: lkXHChh1BXHN5nT5BYhi1x67E1CyYbPKRKoF2LTm5GivuEFpVVYtvEBHtPr74N9E 85 sha512: r8eLGRTc7ZznZkFjeVLyo6/FyQdra9qmlYCwKKxm3kfQAswRS9+3HsYk3thLUhcFmmWhK4dXaICzJwGFonfXwg==""" 86 self.assertEqual(resp.read().rstrip(), expected.strip()) 87 88 def test_sub_file_hash_unrecognized(self): 89 with self.assertRaises(urllib.error.HTTPError) as cm: 90 self.request("/sub_file_hash_unrecognized.sub.txt") 91 cm.exception.close() 92 93 def test_sub_headers(self): 94 resp = self.request("/sub_headers.txt", query="pipe=sub", headers={"X-Test": "PASS"}) 95 expected = b"PASS" 96 self.assertEqual(resp.read().rstrip(), expected) 97 98 def test_sub_location(self): 99 resp = self.request("/sub_location.sub.txt?query_string") 100 expected = """ 101 host: localhost:{0} 102 hostname: localhost 103 path: /sub_location.sub.txt 104 pathname: /sub_location.sub.txt 105 port: {0} 106 query: ?query_string 107 scheme: http 108 server: http://localhost:{0}""".format(self.server.port).encode("ascii") 109 self.assertEqual(resp.read().rstrip(), expected.strip()) 110 111 def test_sub_params(self): 112 resp = self.request("/sub_params.txt", query="plus+pct-20%20pct-3D%3D=PLUS+PCT-20%20PCT-3D%3D&pipe=sub") 113 expected = b"PLUS PCT-20 PCT-3D=" 114 self.assertEqual(resp.read().rstrip(), expected) 115 116 def test_sub_url_base(self): 117 resp = self.request("/sub_url_base.sub.txt") 118 self.assertEqual(resp.read().rstrip(), b"Before / After") 119 120 def test_sub_url_base_via_filename_with_query(self): 121 resp = self.request("/sub_url_base.sub.txt?pipe=slice(5,10)") 122 self.assertEqual(resp.read().rstrip(), b"e / A") 123 124 def test_sub_uuid(self): 125 resp = self.request("/sub_uuid.sub.txt") 126 self.assertRegex(resp.read().rstrip(), b"Before [a-f0-9-]+ After") 127 128 def test_sub_var(self): 129 resp = self.request("/sub_var.sub.txt") 130 port = self.server.port 131 expected = b"localhost %d A %d B localhost C" % (port, port) 132 self.assertEqual(resp.read().rstrip(), expected) 133 134 def test_sub_fs_path(self): 135 resp = self.request("/subdir/sub_path.sub.txt") 136 root = os.path.abspath(doc_root) 137 expected = """%(root)s%(sep)ssubdir%(sep)ssub_path.sub.txt 138 %(root)s%(sep)ssub_path.sub.txt 139 %(root)s%(sep)ssub_path.sub.txt 140 """ % {"root": root, "sep": os.path.sep} 141 self.assertEqual(resp.read(), expected.encode("utf8")) 142 143 def test_sub_header_or_default(self): 144 resp = self.request("/sub_header_or_default.sub.txt", headers={"X-Present": "OK"}) 145 expected = b"OK\nabsent-default" 146 self.assertEqual(resp.read().rstrip(), expected) 147 148 class TestTrickle(TestUsingServer): 149 def test_trickle(self): 150 # Actually testing that the response trickles in is not that easy 151 clock_info = time.get_clock_info("monotonic") 152 t0 = time.monotonic() 153 with self.request("/document.txt", query="pipe=trickle(1:d2:5:d1:r2)") as resp: 154 actual = resp.read() 155 t1 = time.monotonic() 156 with open(os.path.join(doc_root, "document.txt"), "rb") as f: 157 expected = f.read() 158 self.assertEqual(actual, expected) 159 self.assertGreater(t1 - t0, 6 - clock_info.resolution) 160 161 def test_headers(self): 162 resp = self.request("/document.txt", query="pipe=trickle(d0.01)") 163 self.assertEqual(resp.info()["Cache-Control"], "no-cache, no-store, must-revalidate") 164 self.assertEqual(resp.info()["Pragma"], "no-cache") 165 self.assertEqual(resp.info()["Expires"], "0") 166 167 class TestPipesWithVariousHandlers(TestUsingServer): 168 def test_with_python_file_handler(self): 169 resp = self.request("/test_string.py", query="pipe=slice(null,2)") 170 self.assertEqual(resp.read(), b"PA") 171 172 def test_with_python_func_handler(self): 173 @wptserve.handlers.handler 174 def handler(request, response): 175 return "PASS" 176 route = ("GET", "/test/test_pipes_1/", handler) 177 self.server.router.register(*route) 178 resp = self.request(route[1], query="pipe=slice(null,2)") 179 self.assertEqual(resp.read(), b"PA") 180 181 def test_with_python_func_handler_using_response_writer(self): 182 @wptserve.handlers.handler 183 def handler(request, response): 184 response.writer.write_content("PASS") 185 route = ("GET", "/test/test_pipes_1/", handler) 186 self.server.router.register(*route) 187 resp = self.request(route[1], query="pipe=slice(null,2)") 188 # slice has not been applied to the response, because response.writer was used. 189 self.assertEqual(resp.read(), b"PASS") 190 191 def test_header_pipe_with_python_func_using_response_writer(self): 192 @wptserve.handlers.handler 193 def handler(request, response): 194 response.writer.write_content("CONTENT") 195 route = ("GET", "/test/test_pipes_1/", handler) 196 self.server.router.register(*route) 197 resp = self.request(route[1], query="pipe=header(X-TEST,FAIL)") 198 # header pipe was ignored, because response.writer was used. 199 self.assertFalse(resp.info().get("X-TEST")) 200 self.assertEqual(resp.read(), b"CONTENT") 201 202 def test_with_json_handler(self): 203 @wptserve.handlers.json_handler 204 def handler(request, response): 205 return json.dumps({'data': 'PASS'}) 206 route = ("GET", "/test/test_pipes_2/", handler) 207 self.server.router.register(*route) 208 resp = self.request(route[1], query="pipe=slice(null,2)") 209 self.assertEqual(resp.read(), b'"{') 210 211 def test_slice_with_as_is_handler(self): 212 resp = self.request("/test.asis", query="pipe=slice(null,2)") 213 self.assertEqual(202, resp.getcode()) 214 self.assertEqual("Giraffe", resp.msg) 215 self.assertEqual("PASS", resp.info()["X-Test"]) 216 # slice has not been applied to the response, because response.writer was used. 217 self.assertEqual(b"Content", resp.read()) 218 219 def test_headers_with_as_is_handler(self): 220 resp = self.request("/test.asis", query="pipe=header(X-TEST,FAIL)") 221 self.assertEqual(202, resp.getcode()) 222 self.assertEqual("Giraffe", resp.msg) 223 # header pipe was ignored. 224 self.assertEqual("PASS", resp.info()["X-TEST"]) 225 self.assertEqual(b"Content", resp.read()) 226 227 def test_trickle_with_as_is_handler(self): 228 t0 = time.time() 229 resp = self.request("/test.asis", query="pipe=trickle(1:d2:5:d1:r2)") 230 t1 = time.time() 231 self.assertTrue(b'Content' in resp.read()) 232 self.assertGreater(6, t1-t0) 233 234 def test_gzip_handler(self): 235 resp = self.request("/document.txt", query="pipe=gzip") 236 self.assertEqual(resp.getcode(), 200) 237 238 def test_sub_default_py(self): 239 route = ("GET", "/defaultsubpy", wptserve.handlers.python_script_handler) 240 self.server.router.register(*route) 241 resp = self.request("/defaultsubpy") 242 self.assertEqual(b"localhost", resp.read()) 243 244 245 if __name__ == '__main__': 246 unittest.main()