from __future__ import absolute_import
import collections
import json
import os
import pytest
from six.moves.urllib.error import HTTPError
from six.moves.urllib.request import (
HTTPHandler,
ProxyHandler,
Request,
build_opener,
install_opener,
urlopen,
)
import mozunit
import mozhttpd
def httpd_url(httpd, path, querystr=None):
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 collections.defaultdict(int)
@pytest.fixture(name="try_get")
def fixture_try_get(num_requests):
def try_get(httpd, querystr):
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):
def try_post(httpd, querystr):
num_requests["post_handler"] = 0
postdata = {"hamburgers": "1234"}
f = urlopen(
httpd_url(httpd, "/api/resource/", querystr),
data=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):
def try_del(httpd, querystr):
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():
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):
@mozhttpd.handlers.json_response
def get_handler(request, objid):
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):
@mozhttpd.handlers.json_response
def get_handler(request, objid):
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):
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):
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):
try_get(httpd, "")
try_get(httpd, "?foo=bar")
try_post(httpd, "")
try_post(httpd, "?foo=bar")
try_del(httpd, "")
try_del(httpd, "?foo=bar")
with pytest.raises(HTTPError) as exc_info:
urlopen(httpd_url(httpd, "/"))
assert exc_info.value.code == 404
def test_nonexistent_resources(httpd_no_urlhandlers):
with pytest.raises(HTTPError) as excinfo:
urlopen(httpd_url(httpd_no_urlhandlers, "/api/resource/"))
assert excinfo.value.code == 404
with pytest.raises(HTTPError) as excinfo:
urlopen(httpd_url(httpd_no_urlhandlers, "/api/resource/"), data=json.dumps({}))
assert excinfo.value.code == 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 f.read()
try_get(httpd_with_docroot, "")
try_get(httpd_with_docroot, "?foo=bar")
def index_contents(host):
return "{host} index".format(host=host)
@pytest.fixture(name="hosts")
def fixture_hosts():
return ("mozilla.com", "mozilla.org")
@pytest.fixture(name="docroot")
def fixture_docroot(tmpdir):
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):
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()
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() == 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()
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() == 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()