| import inspect |
| import os |
| import re |
| import pytest |
| |
| from .env import H2Conf, H2TestEnv |
| |
| |
| @pytest.mark.skipif(condition=H2TestEnv.is_unsupported, reason="mod_http2 not supported here") |
| class TestProxy: |
| |
| @pytest.fixture(autouse=True, scope='class') |
| def _class_scope(self, env): |
| H2Conf(env).add_vhost_cgi(proxy_self=True).install() |
| assert env.apache_restart() == 0 |
| |
| def local_src(self, fname): |
| return os.path.join(os.path.dirname(inspect.getfile(TestProxy)), fname) |
| |
| def setup_method(self, method): |
| print("setup_method: %s" % method.__name__) |
| |
| def teardown_method(self, method): |
| print("teardown_method: %s" % method.__name__) |
| |
| def test_h2_500_01(self, env): |
| url = env.mkurl("https", "cgi", "/proxy/hello.py") |
| r = env.curl_get(url, 5) |
| assert r.response["status"] == 200 |
| assert "HTTP/1.1" == r.response["json"]["protocol"] |
| assert r.response["json"]["https"] == "" |
| assert r.response["json"]["ssl_protocol"] == "" |
| assert r.response["json"]["h2"] == "" |
| assert r.response["json"]["h2push"] == "" |
| |
| # upload and GET again using curl, compare to original content |
| def curl_upload_and_verify(self, env, fname, options=None): |
| url = env.mkurl("https", "cgi", "/proxy/upload.py") |
| fpath = os.path.join(env.gen_dir, fname) |
| r = env.curl_upload(url, fpath, options=options) |
| assert r.exit_code == 0 |
| assert 200 <= r.response["status"] < 300 |
| |
| # why is the scheme wrong? |
| r2 = env.curl_get(re.sub(r'http:', 'https:', r.response["header"]["location"])) |
| assert r2.exit_code == 0 |
| assert r2.response["status"] == 200 |
| with open(self.local_src(fpath), mode='rb') as file: |
| src = file.read() |
| assert r2.response["body"] == src |
| |
| @pytest.mark.parametrize("name", [ |
| "data-1k", "data-10k", "data-100k", "data-1m", |
| ]) |
| def test_h2_500_10(self, env, name, repeat): |
| self.curl_upload_and_verify(env, name, ["--http2"]) |
| |
| def test_h2_500_11(self, env): |
| self.curl_upload_and_verify(env, "data-1k", [ |
| "--http1.1", "-H", "Content-Length:", "-H", "Transfer-Encoding: chunked" |
| ]) |
| self.curl_upload_and_verify(env, "data-1k", ["--http2", "-H", "Content-Length:"]) |
| |
| # POST some data using nghttp and see it echo'ed properly back |
| def nghttp_post_and_verify(self, env, fname, options=None): |
| url = env.mkurl("https", "cgi", "/proxy/echo.py") |
| fpath = os.path.join(env.gen_dir, fname) |
| r = env.nghttp().upload(url, fpath, options=options) |
| assert r.exit_code == 0 |
| assert 200 <= r.response["status"] < 300 |
| with open(self.local_src(fpath), mode='rb') as file: |
| src = file.read() |
| if r.response["body"] != src: |
| with open(os.path.join(env.gen_dir, "nghttp.out"), 'w') as fd: |
| fd.write(r.outraw.decode()) |
| fd.write("\nstderr:\n") |
| fd.write(r.stderr) |
| assert r.response["body"] == src |
| |
| @pytest.mark.parametrize("name", [ |
| "data-1k", "data-10k", "data-100k", "data-1m", |
| ]) |
| def test_h2_500_20(self, env, name, repeat): |
| self.nghttp_post_and_verify(env, name, []) |
| |
| @pytest.mark.parametrize("name", [ |
| "data-1k", "data-10k", "data-100k", "data-1m", |
| ]) |
| def test_h2_500_21(self, env, name, repeat): |
| self.nghttp_post_and_verify(env, name, ["--no-content-length"]) |
| |
| # upload and GET again using nghttp, compare to original content |
| def nghttp_upload_and_verify(self, env, fname, options=None): |
| url = env.mkurl("https", "cgi", "/proxy/upload.py") |
| fpath = os.path.join(env.gen_dir, fname) |
| |
| r = env.nghttp().upload_file(url, fpath, options=options) |
| assert r.exit_code == 0 |
| assert 200 <= r.response["status"] < 300 |
| assert r.response["header"]["location"] |
| |
| # why is the scheme wrong? |
| r2 = env.nghttp().get(re.sub(r'http:', 'https:', r.response["header"]["location"])) |
| assert r2.exit_code == 0 |
| assert r2.response["status"] == 200 |
| with open(self.local_src(fpath), mode='rb') as file: |
| src = file.read() |
| assert src == r2.response["body"] |
| |
| @pytest.mark.parametrize("name", [ |
| "data-1k", "data-10k", "data-100k", "data-1m", |
| ]) |
| def test_h2_500_22(self, env, name): |
| self.nghttp_upload_and_verify(env, name, []) |
| |
| @pytest.mark.parametrize("name", [ |
| "data-1k", "data-10k", "data-100k", "data-1m", |
| ]) |
| def test_h2_500_23(self, env, name): |
| self.nghttp_upload_and_verify(env, name, ["--no-content-length"]) |
| |
| # upload using nghttp and check returned status |
| def nghttp_upload_stat(self, env, fname, options=None): |
| url = env.mkurl("https", "cgi", "/proxy/upload.py") |
| fpath = os.path.join(env.gen_dir, fname) |
| |
| r = env.nghttp().upload_file(url, fpath, options=options) |
| assert r.exit_code == 0 |
| assert 200 <= r.response["status"] < 300 |
| assert r.response["header"]["location"] |
| |
| def test_h2_500_24(self, env): |
| for i in range(50): |
| self.nghttp_upload_stat(env, "data-1k", ["--no-content-length"]) |
| |
| # lets do some error tests |
| def test_h2_500_30(self, env): |
| url = env.mkurl("https", "cgi", "/proxy/h2test/error?status=500") |
| r = env.curl_get(url) |
| assert r.exit_code == 0, r |
| assert r.response['status'] == 500 |
| url = env.mkurl("https", "cgi", "/proxy/h2test/error?error=timeout") |
| r = env.curl_get(url) |
| assert r.exit_code == 0, r |
| assert r.response['status'] == 408 |
| |
| # produce an error during response body |
| def test_h2_500_31(self, env, repeat): |
| pytest.skip("needs fix in core protocol handling") |
| url = env.mkurl("https", "cgi", "/proxy/h2test/error?body_error=timeout") |
| r = env.curl_get(url) |
| assert r.exit_code != 0, r |
| |
| # produce an error, fail to generate an error bucket |
| def test_h2_500_32(self, env, repeat): |
| pytest.skip("needs fix in core protocol handling") |
| url = env.mkurl("https", "cgi", "/proxy/h2test/error?body_error=timeout&error_bucket=0") |
| r = env.curl_get(url) |
| assert r.exit_code != 0, r |
