api.py (9594B)
1 #!/usr/bin/env python 2 3 # This Source Code Form is subject to the terms of the Mozilla Public 4 # License, v. 2.0. If a copy of the MPL was not distributed with this file, 5 # You can obtain one at http://mozilla.org/MPL/2.0/. 6 7 import collections 8 import json 9 import os 10 from urllib.error import HTTPError 11 from urllib.request import ( 12 HTTPHandler, 13 ProxyHandler, 14 Request, 15 build_opener, 16 install_opener, 17 urlopen, 18 ) 19 20 import mozhttpd 21 import mozunit 22 import pytest 23 24 25 def httpd_url(httpd, path, querystr=None): 26 """Return the URL to a started MozHttpd server for the given info.""" 27 28 url = f"http://127.0.0.1:{httpd.httpd.server_port}{path}" 29 30 if querystr is not None: 31 url = f"{url}?{querystr}" 32 33 return url 34 35 36 @pytest.fixture(name="num_requests") 37 def fixture_num_requests(): 38 """Return a defaultdict to count requests to HTTP handlers.""" 39 return collections.defaultdict(int) 40 41 42 @pytest.fixture(name="try_get") 43 def fixture_try_get(num_requests): 44 """Return a function to try GET requests to the server.""" 45 46 def try_get(httpd, querystr): 47 """Try GET requests to the server.""" 48 49 num_requests["get_handler"] = 0 50 51 f = urlopen(httpd_url(httpd, "/api/resource/1", querystr)) 52 53 assert f.getcode() == 200 54 assert json.loads(f.read()) == {"called": 1, "id": "1", "query": querystr} 55 assert num_requests["get_handler"] == 1 56 57 return try_get 58 59 60 @pytest.fixture(name="try_post") 61 def fixture_try_post(num_requests): 62 """Return a function to try POST calls to the server.""" 63 64 def try_post(httpd, querystr): 65 """Try POST calls to the server.""" 66 67 num_requests["post_handler"] = 0 68 69 postdata = {"hamburgers": "1234"} 70 71 f = urlopen( 72 httpd_url(httpd, "/api/resource/", querystr), 73 data=json.dumps(postdata).encode(), 74 ) 75 76 assert f.getcode() == 201 77 assert json.loads(f.read()) == { 78 "called": 1, 79 "data": postdata, 80 "query": querystr, 81 } 82 assert num_requests["post_handler"] == 1 83 84 return try_post 85 86 87 @pytest.fixture(name="try_del") 88 def fixture_try_del(num_requests): 89 """Return a function to try DEL calls to the server.""" 90 91 def try_del(httpd, querystr): 92 """Try DEL calls to the server.""" 93 94 num_requests["del_handler"] = 0 95 96 opener = build_opener(HTTPHandler) 97 request = Request(httpd_url(httpd, "/api/resource/1", querystr)) 98 request.get_method = lambda: "DEL" 99 f = opener.open(request) 100 101 assert f.getcode() == 200 102 assert json.loads(f.read()) == {"called": 1, "id": "1", "query": querystr} 103 assert num_requests["del_handler"] == 1 104 105 return try_del 106 107 108 @pytest.fixture(name="httpd_no_urlhandlers") 109 def fixture_httpd_no_urlhandlers(): 110 """Yields a started MozHttpd server with no URL handlers.""" 111 httpd = mozhttpd.MozHttpd(port=0) 112 httpd.start(block=False) 113 yield httpd 114 httpd.stop() 115 116 117 @pytest.fixture(name="httpd_with_docroot") 118 def fixture_httpd_with_docroot(num_requests): 119 """Yields a started MozHttpd server with docroot set.""" 120 121 @mozhttpd.handlers.json_response 122 def get_handler(request, objid): 123 """Handler for HTTP GET requests.""" 124 125 num_requests["get_handler"] += 1 126 127 return ( 128 200, 129 { 130 "called": num_requests["get_handler"], 131 "id": objid, 132 "query": request.query, 133 }, 134 ) 135 136 httpd = mozhttpd.MozHttpd( 137 port=0, 138 docroot=os.path.dirname(os.path.abspath(__file__)), 139 urlhandlers=[ 140 { 141 "method": "GET", 142 "path": "/api/resource/([^/]+)/?", 143 "function": get_handler, 144 } 145 ], 146 ) 147 148 httpd.start(block=False) 149 yield httpd 150 httpd.stop() 151 152 153 @pytest.fixture(name="httpd") 154 def fixture_httpd(num_requests): 155 """Yield a started MozHttpd server.""" 156 157 @mozhttpd.handlers.json_response 158 def get_handler(request, objid): 159 """Handler for HTTP GET requests.""" 160 161 num_requests["get_handler"] += 1 162 163 return ( 164 200, 165 { 166 "called": num_requests["get_handler"], 167 "id": objid, 168 "query": request.query, 169 }, 170 ) 171 172 @mozhttpd.handlers.json_response 173 def post_handler(request): 174 """Handler for HTTP POST requests.""" 175 176 num_requests["post_handler"] += 1 177 178 return ( 179 201, 180 { 181 "called": num_requests["post_handler"], 182 "data": json.loads(request.body), 183 "query": request.query, 184 }, 185 ) 186 187 @mozhttpd.handlers.json_response 188 def del_handler(request, objid): 189 """Handler for HTTP DEL requests.""" 190 191 num_requests["del_handler"] += 1 192 193 return ( 194 200, 195 { 196 "called": num_requests["del_handler"], 197 "id": objid, 198 "query": request.query, 199 }, 200 ) 201 202 httpd = mozhttpd.MozHttpd( 203 port=0, 204 urlhandlers=[ 205 { 206 "method": "GET", 207 "path": "/api/resource/([^/]+)/?", 208 "function": get_handler, 209 }, 210 { 211 "method": "POST", 212 "path": "/api/resource/?", 213 "function": post_handler, 214 }, 215 { 216 "method": "DEL", 217 "path": "/api/resource/([^/]+)/?", 218 "function": del_handler, 219 }, 220 ], 221 ) 222 223 httpd.start(block=False) 224 yield httpd 225 httpd.stop() 226 227 228 def test_api(httpd, try_get, try_post, try_del): 229 # GET requests 230 try_get(httpd, "") 231 try_get(httpd, "?foo=bar") 232 233 # POST requests 234 try_post(httpd, "") 235 try_post(httpd, "?foo=bar") 236 237 # DEL requests 238 try_del(httpd, "") 239 try_del(httpd, "?foo=bar") 240 241 # GET: By default we don't serve any files if we just define an API 242 with pytest.raises(HTTPError) as exc_info: 243 urlopen(httpd_url(httpd, "/")) 244 245 assert exc_info.value.code == 404 246 247 248 def test_nonexistent_resources(httpd_no_urlhandlers): 249 # GET: Return 404 for non-existent endpoint 250 with pytest.raises(HTTPError) as excinfo: 251 urlopen(httpd_url(httpd_no_urlhandlers, "/api/resource/")) 252 assert excinfo.value.code == 404 253 254 # POST: POST should also return 404 255 with pytest.raises(HTTPError) as excinfo: 256 urlopen( 257 httpd_url(httpd_no_urlhandlers, "/api/resource/"), 258 data=json.dumps({}).encode(), 259 ) 260 assert excinfo.value.code == 404 261 262 # DEL: DEL should also return 404 263 opener = build_opener(HTTPHandler) 264 request = Request(httpd_url(httpd_no_urlhandlers, "/api/resource/")) 265 request.get_method = lambda: "DEL" 266 267 with pytest.raises(HTTPError) as excinfo: 268 opener.open(request) 269 assert excinfo.value.code == 404 270 271 272 def test_api_with_docroot(httpd_with_docroot, try_get): 273 f = urlopen(httpd_url(httpd_with_docroot, "/")) 274 assert f.getcode() == 200 275 assert "Directory listing for" in f.read().decode() 276 277 # Make sure API methods still work 278 try_get(httpd_with_docroot, "") 279 try_get(httpd_with_docroot, "?foo=bar") 280 281 282 def index_contents(host): 283 """Return the expected index contents for the given host.""" 284 return f"{host} index" 285 286 287 @pytest.fixture(name="hosts") 288 def fixture_hosts(): 289 """Returns a tuple of hosts.""" 290 return ("mozilla.com", "mozilla.org") 291 292 293 @pytest.fixture(name="docroot") 294 def fixture_docroot(tmpdir): 295 """Returns a path object to a temporary docroot directory.""" 296 docroot = tmpdir.mkdir("docroot") 297 index_file = docroot.join("index.html") 298 index_file.write(index_contents("*")) 299 300 yield docroot 301 302 docroot.remove() 303 304 305 @pytest.fixture(name="httpd_with_proxy_handler") 306 def fixture_httpd_with_proxy_handler(docroot): 307 """Yields a started MozHttpd server for the proxy test.""" 308 309 httpd = mozhttpd.MozHttpd(port=0, docroot=str(docroot)) 310 httpd.start(block=False) 311 312 port = httpd.httpd.server_port 313 proxy_support = ProxyHandler({ 314 "http": f"http://127.0.0.1:{port:d}", 315 }) 316 install_opener(build_opener(proxy_support)) 317 318 yield httpd 319 320 httpd.stop() 321 322 # Reset proxy opener in case it changed 323 install_opener(None) 324 325 326 def test_proxy(httpd_with_proxy_handler, hosts): 327 for host in hosts: 328 f = urlopen(f"http://{host}/") 329 assert f.getcode() == 200 330 assert f.read() == index_contents("*").encode() 331 332 333 @pytest.fixture(name="httpd_with_proxy_host_dirs") 334 def fixture_httpd_with_proxy_host_dirs(docroot, hosts): 335 for host in hosts: 336 index_file = docroot.mkdir(host).join("index.html") 337 index_file.write(index_contents(host)) 338 339 httpd = mozhttpd.MozHttpd(port=0, docroot=str(docroot), proxy_host_dirs=True) 340 341 httpd.start(block=False) 342 343 port = httpd.httpd.server_port 344 proxy_support = ProxyHandler({"http": f"http://127.0.0.1:{port:d}"}) 345 install_opener(build_opener(proxy_support)) 346 347 yield httpd 348 349 httpd.stop() 350 351 # Reset proxy opener in case it changed 352 install_opener(None) 353 354 355 def test_proxy_separate_directories(httpd_with_proxy_host_dirs, hosts): 356 for host in hosts: 357 f = urlopen(f"http://{host}/") 358 assert f.getcode() == 200 359 assert f.read() == index_contents(host).encode() 360 361 unproxied_host = "notmozilla.org" 362 363 with pytest.raises(HTTPError) as excinfo: 364 urlopen(f"http://{unproxied_host}/") 365 366 assert excinfo.value.code == 404 367 368 369 if __name__ == "__main__": 370 mozunit.main()