Adding upstream version 110.0.1.upstream/110.0.1upstream

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

1 files changed, 381 insertions, 0 deletions

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()