From 6bf0a5cb5034a7e684dcc3500e841785237ce2dd Mon Sep 17 00:00:00 2001 From: Daniel Baumann Date: Sun, 7 Apr 2024 19:32:43 +0200 Subject: Adding upstream version 1:115.7.0. Signed-off-by: Daniel Baumann --- testing/mozbase/mozhttpd/tests/api.py | 381 ++++++++++++++++++++++++++++++++++ 1 file changed, 381 insertions(+) create mode 100644 testing/mozbase/mozhttpd/tests/api.py (limited to 'testing/mozbase/mozhttpd/tests/api.py') diff --git a/testing/mozbase/mozhttpd/tests/api.py b/testing/mozbase/mozhttpd/tests/api.py new file mode 100644 index 0000000000..c2fce58be9 --- /dev/null +++ b/testing/mozbase/mozhttpd/tests/api.py @@ -0,0 +1,381 @@ +#!/usr/bin/env python + +# This Source Code Form is subject to the terms of the Mozilla Public +# License, v. 2.0. If a copy of the MPL was not distributed with this file, +# You can obtain one at http://mozilla.org/MPL/2.0/. + +import collections +import json +import os + +import mozhttpd +import mozunit +import pytest +from six import ensure_binary, ensure_str +from six.moves.urllib.error import HTTPError +from six.moves.urllib.request import ( + HTTPHandler, + ProxyHandler, + Request, + build_opener, + install_opener, + urlopen, +) + + +def httpd_url(httpd, path, querystr=None): + """Return the URL to a started MozHttpd server for the given info.""" + + url = "http://127.0.0.1:{port}{path}".format( + port=httpd.httpd.server_port, + path=path, + ) + + if querystr is not None: + url = "{url}?{querystr}".format( + url=url, + querystr=querystr, + ) + + return url + + +@pytest.fixture(name="num_requests") +def fixture_num_requests(): + """Return a defaultdict to count requests to HTTP handlers.""" + return collections.defaultdict(int) + + +@pytest.fixture(name="try_get") +def fixture_try_get(num_requests): + """Return a function to try GET requests to the server.""" + + def try_get(httpd, querystr): + """Try GET requests to the server.""" + + num_requests["get_handler"] = 0 + + f = urlopen(httpd_url(httpd, "/api/resource/1", querystr)) + + assert f.getcode() == 200 + assert json.loads(f.read()) == {"called": 1, "id": "1", "query": querystr} + assert num_requests["get_handler"] == 1 + + return try_get + + +@pytest.fixture(name="try_post") +def fixture_try_post(num_requests): + """Return a function to try POST calls to the server.""" + + def try_post(httpd, querystr): + """Try POST calls to the server.""" + + num_requests["post_handler"] = 0 + + postdata = {"hamburgers": "1234"} + + f = urlopen( + httpd_url(httpd, "/api/resource/", querystr), + data=ensure_binary(json.dumps(postdata)), + ) + + assert f.getcode() == 201 + assert json.loads(f.read()) == { + "called": 1, + "data": postdata, + "query": querystr, + } + assert num_requests["post_handler"] == 1 + + return try_post + + +@pytest.fixture(name="try_del") +def fixture_try_del(num_requests): + """Return a function to try DEL calls to the server.""" + + def try_del(httpd, querystr): + """Try DEL calls to the server.""" + + num_requests["del_handler"] = 0 + + opener = build_opener(HTTPHandler) + request = Request(httpd_url(httpd, "/api/resource/1", querystr)) + request.get_method = lambda: "DEL" + f = opener.open(request) + + assert f.getcode() == 200 + assert json.loads(f.read()) == {"called": 1, "id": "1", "query": querystr} + assert num_requests["del_handler"] == 1 + + return try_del + + +@pytest.fixture(name="httpd_no_urlhandlers") +def fixture_httpd_no_urlhandlers(): + """Yields a started MozHttpd server with no URL handlers.""" + httpd = mozhttpd.MozHttpd(port=0) + httpd.start(block=False) + yield httpd + httpd.stop() + + +@pytest.fixture(name="httpd_with_docroot") +def fixture_httpd_with_docroot(num_requests): + """Yields a started MozHttpd server with docroot set.""" + + @mozhttpd.handlers.json_response + def get_handler(request, objid): + """Handler for HTTP GET requests.""" + + num_requests["get_handler"] += 1 + + return ( + 200, + { + "called": num_requests["get_handler"], + "id": objid, + "query": request.query, + }, + ) + + httpd = mozhttpd.MozHttpd( + port=0, + docroot=os.path.dirname(os.path.abspath(__file__)), + urlhandlers=[ + { + "method": "GET", + "path": "/api/resource/([^/]+)/?", + "function": get_handler, + } + ], + ) + + httpd.start(block=False) + yield httpd + httpd.stop() + + +@pytest.fixture(name="httpd") +def fixture_httpd(num_requests): + """Yield a started MozHttpd server.""" + + @mozhttpd.handlers.json_response + def get_handler(request, objid): + """Handler for HTTP GET requests.""" + + num_requests["get_handler"] += 1 + + return ( + 200, + { + "called": num_requests["get_handler"], + "id": objid, + "query": request.query, + }, + ) + + @mozhttpd.handlers.json_response + def post_handler(request): + """Handler for HTTP POST requests.""" + + num_requests["post_handler"] += 1 + + return ( + 201, + { + "called": num_requests["post_handler"], + "data": json.loads(request.body), + "query": request.query, + }, + ) + + @mozhttpd.handlers.json_response + def del_handler(request, objid): + """Handler for HTTP DEL requests.""" + + num_requests["del_handler"] += 1 + + return ( + 200, + { + "called": num_requests["del_handler"], + "id": objid, + "query": request.query, + }, + ) + + httpd = mozhttpd.MozHttpd( + port=0, + urlhandlers=[ + { + "method": "GET", + "path": "/api/resource/([^/]+)/?", + "function": get_handler, + }, + { + "method": "POST", + "path": "/api/resource/?", + "function": post_handler, + }, + { + "method": "DEL", + "path": "/api/resource/([^/]+)/?", + "function": del_handler, + }, + ], + ) + + httpd.start(block=False) + yield httpd + httpd.stop() + + +def test_api(httpd, try_get, try_post, try_del): + # GET requests + try_get(httpd, "") + try_get(httpd, "?foo=bar") + + # POST requests + try_post(httpd, "") + try_post(httpd, "?foo=bar") + + # DEL requests + try_del(httpd, "") + try_del(httpd, "?foo=bar") + + # GET: By default we don't serve any files if we just define an API + with pytest.raises(HTTPError) as exc_info: + urlopen(httpd_url(httpd, "/")) + + assert exc_info.value.code == 404 + + +def test_nonexistent_resources(httpd_no_urlhandlers): + # GET: Return 404 for non-existent endpoint + with pytest.raises(HTTPError) as excinfo: + urlopen(httpd_url(httpd_no_urlhandlers, "/api/resource/")) + assert excinfo.value.code == 404 + + # POST: POST should also return 404 + with pytest.raises(HTTPError) as excinfo: + urlopen( + httpd_url(httpd_no_urlhandlers, "/api/resource/"), + data=ensure_binary(json.dumps({})), + ) + assert excinfo.value.code == 404 + + # DEL: DEL should also return 404 + opener = build_opener(HTTPHandler) + request = Request(httpd_url(httpd_no_urlhandlers, "/api/resource/")) + request.get_method = lambda: "DEL" + + with pytest.raises(HTTPError) as excinfo: + opener.open(request) + assert excinfo.value.code == 404 + + +def test_api_with_docroot(httpd_with_docroot, try_get): + f = urlopen(httpd_url(httpd_with_docroot, "/")) + assert f.getcode() == 200 + assert "Directory listing for" in ensure_str(f.read()) + + # Make sure API methods still work + try_get(httpd_with_docroot, "") + try_get(httpd_with_docroot, "?foo=bar") + + +def index_contents(host): + """Return the expected index contents for the given host.""" + return "{host} index".format(host=host) + + +@pytest.fixture(name="hosts") +def fixture_hosts(): + """Returns a tuple of hosts.""" + return ("mozilla.com", "mozilla.org") + + +@pytest.fixture(name="docroot") +def fixture_docroot(tmpdir): + """Returns a path object to a temporary docroot directory.""" + docroot = tmpdir.mkdir("docroot") + index_file = docroot.join("index.html") + index_file.write(index_contents("*")) + + yield docroot + + docroot.remove() + + +@pytest.fixture(name="httpd_with_proxy_handler") +def fixture_httpd_with_proxy_handler(docroot): + """Yields a started MozHttpd server for the proxy test.""" + + httpd = mozhttpd.MozHttpd(port=0, docroot=str(docroot)) + httpd.start(block=False) + + port = httpd.httpd.server_port + proxy_support = ProxyHandler( + { + "http": "http://127.0.0.1:{port:d}".format(port=port), + } + ) + install_opener(build_opener(proxy_support)) + + yield httpd + + httpd.stop() + + # Reset proxy opener in case it changed + install_opener(None) + + +def test_proxy(httpd_with_proxy_handler, hosts): + for host in hosts: + f = urlopen("http://{host}/".format(host=host)) + assert f.getcode() == 200 + assert f.read() == ensure_binary(index_contents("*")) + + +@pytest.fixture(name="httpd_with_proxy_host_dirs") +def fixture_httpd_with_proxy_host_dirs(docroot, hosts): + for host in hosts: + index_file = docroot.mkdir(host).join("index.html") + index_file.write(index_contents(host)) + + httpd = mozhttpd.MozHttpd(port=0, docroot=str(docroot), proxy_host_dirs=True) + + httpd.start(block=False) + + port = httpd.httpd.server_port + proxy_support = ProxyHandler( + {"http": "http://127.0.0.1:{port:d}".format(port=port)} + ) + install_opener(build_opener(proxy_support)) + + yield httpd + + httpd.stop() + + # Reset proxy opener in case it changed + install_opener(None) + + +def test_proxy_separate_directories(httpd_with_proxy_host_dirs, hosts): + for host in hosts: + f = urlopen("http://{host}/".format(host=host)) + assert f.getcode() == 200 + assert f.read() == ensure_binary(index_contents(host)) + + unproxied_host = "notmozilla.org" + + with pytest.raises(HTTPError) as excinfo: + urlopen("http://{host}/".format(host=unproxied_host)) + + assert excinfo.value.code == 404 + + +if __name__ == "__main__": + mozunit.main() -- cgit v1.2.3