if test -d ./test; then
APACHE_FAST_OUTPUT(test/Makefile)
-fi
-if test -d ./test/modules/http2; then
- APACHE_FAST_OUTPUT(test/modules/http2/Makefile)
- AC_CONFIG_FILES([test/modules/http2/config.ini])
+ AC_CONFIG_FILES([test/pyhttpd/config.ini])
fi
dnl ## Finalize the variables
--- /dev/null
+config.ini
+gen
# dbu_OBJECTS = dbu.lo
# dbu: $(dbu_OBJECTS)
# $(LINK) $(dbu_OBJECTS) $(PROGRAM_LDADD)
+
+clean:
+ rm -rf gen
+
+distclean:
+ rm -f pytest/config.ini
\ No newline at end of file
--- /dev/null
+import sys
+import os
+
+sys.path.append(os.path.join(os.path.dirname(__file__), '.'))
+
+from pyhttpd.env import HttpdTestEnv
+
+def pytest_report_header(config, startdir):
+ env = HttpdTestEnv()
+ return f"[apache httpd: {env.get_httpd_version()}, mpm: {env.mpm_type}, {env.prefix}]"
+
+
--- /dev/null
+import logging
+import os
+
+import pytest
+import sys
+
+sys.path.append(os.path.join(os.path.dirname(__file__), '../..'))
+
+from .env import CoreTestEnv
+
+
+def pytest_report_header(config, startdir):
+ env = CoreTestEnv(setup_dirs=False)
+ return f"core [apache: {env.get_httpd_version()}, mpm: {env.mpm_type}, {env.prefix}]"
+
+
+@pytest.fixture(scope="module")
+def env(pytestconfig) -> CoreTestEnv:
+ level = logging.INFO
+ console = logging.StreamHandler()
+ console.setLevel(level)
+ console.setFormatter(logging.Formatter('%(levelname)s: %(message)s'))
+ logging.getLogger('').addHandler(console)
+ logging.getLogger('').setLevel(level=level)
+ env = CoreTestEnv(pytestconfig=pytestconfig)
+ env.apache_access_log_clear()
+ env.apache_error_log_clear()
+ return env
+
+
+@pytest.fixture(autouse=True, scope="module")
+def _session_scope(env):
+ yield
+ assert env.apache_stop() == 0
+ errors, warnings = env.apache_errors_and_warnings()
+ assert (len(errors), len(warnings)) == (0, 0),\
+ f"apache logged {len(errors)} errors and {len(warnings)} warnings: \n"\
+ "{0}\n{1}\n".format("\n".join(errors), "\n".join(warnings))
+
--- /dev/null
+import inspect
+import logging
+import os
+
+from pyhttpd.env import HttpdTestEnv, HttpdTestSetup
+
+log = logging.getLogger(__name__)
+
+
+class CoreTestEnv(HttpdTestEnv):
+
+ def __init__(self, pytestconfig=None, setup_dirs=True):
+ super().__init__(pytestconfig=pytestconfig,
+ local_dir=os.path.dirname(inspect.getfile(CoreTestEnv)))
+ if setup_dirs:
+ self._setup = HttpdTestSetup(env=self)
+ self._setup.make()
+ self.issue_certs()
-import time
-
import pytest
-from h2_conf import HttpdConf
+from pyhttpd.conf import HttpdConf
class TestEncoding:
@pytest.fixture(autouse=True, scope='class')
def _class_scope(self, env):
- extras = {
- 'base': f"""
+ conf = HttpdConf(env)
+ conf.add(f"""
<Directory "{env.gen_dir}">
AllowOverride None
Options +ExecCGI -MultiViews +SymLinksIfOwnerMatch
Require all granted
</Directory>
- """,
- }
- conf = HttpdConf(env)
- conf.add_vhost_test1(extras=extras)
+ """)
+ conf.add_vhost_test1()
conf.add_vhost_test2(extras={
f"test2.{env.http_tld}": "AllowEncodedSlashes on",
})
assert env.apache_restart() == 0
yield
errors, warnings = env.apache_errors_and_warnings()
+ nl = "\n"
assert (len(errors), len(warnings)) == (TestEncoding.EXP_AH10244_ERRS, 0),\
- f"apache logged {len(errors)} errors and {len(warnings)} warnings: \n"\
- "{0}\n{1}\n".format("\n".join(errors), "\n".join(warnings))
+ f"apache logged {len(errors)} errors and {len(warnings)} warnings: \n"\
+ f"{nl.join(errors)}\n{nl.join(warnings)}\n"
env.apache_error_log_clear()
# check handling of url encodings that are accepted
"/nothing/%2e/%2e%2e/006/006.css",
"/nothing/%2e/%2e%2e/006/006%2ecss",
])
- def test_203_01(self, env, path):
+ def test_core_001_01(self, env, path):
url = env.mkurl("https", "test1", path)
r = env.curl_get(url)
assert r.response["status"] == 200
"/006/../006/006.css",
"/006/%2e%2e/006/006.css",
])
- def test_203_03(self, env, path):
+ def test_core_001_03(self, env, path):
url = env.mkurl("https", "test1", path)
r = env.curl_get(url)
assert r.response["status"] == 200
["/nothing/%25%32%65%25%32%65/%25%32%65%25%32%65/h2_env.py", 404],
["/cgi-bin/%25%32%65%25%32%65/%25%32%65%25%32%65/h2_env.py", 404],
])
- def test_203_04(self, env, path, status):
+ def test_core_001_04(self, env, path, status):
url = env.mkurl("https", "cgi", path)
r = env.curl_get(url)
assert r.response["status"] == status
["test2", "/x%252f.test", 200],
["test2", "/10%25abnormal.txt", 200],
])
- def test_203_20(self, env, host, path, status):
+ def test_core_001_20(self, env, host, path, status):
url = env.mkurl("https", host, path)
r = env.curl_get(url)
assert r.response["status"] == status
-
+++ /dev/null
-
-# no targets: we don't want to build anything by default. if you want the
-# test programs, then "make test"
-TARGETS =
-
-bin_PROGRAMS =
-
-PROGRAM_LDADD = $(EXTRA_LDFLAGS) $(PROGRAM_DEPENDENCIES) $(EXTRA_LIBS)
-PROGRAM_DEPENDENCIES = \
- $(top_srcdir)/srclib/apr-util/libaprutil.la \
- $(top_srcdir)/srclib/apr/libapr.la
-
-include $(top_builddir)/build/rules.mk
-
-test: $(bin_PROGRAMS)
-
-# example for building a test proggie
-# dbu_OBJECTS = dbu.lo
-# dbu: $(dbu_OBJECTS)
-# $(LINK) $(dbu_OBJECTS) $(PROGRAM_LDADD)
import os
import pytest
+import sys
-from h2_certs import CertificateSpec, H2TestCA
-from h2_env import H2TestEnv
+sys.path.append(os.path.join(os.path.dirname(__file__), '../..'))
+
+from .env import H2TestEnv
def pytest_report_header(config, startdir):
metafunc.parametrize('repeat', range(count))
-@pytest.fixture(scope="session")
+@pytest.fixture(scope="module")
def env(pytestconfig) -> H2TestEnv:
level = logging.INFO
console = logging.StreamHandler()
logging.getLogger('').addHandler(console)
logging.getLogger('').setLevel(level=level)
env = H2TestEnv(pytestconfig=pytestconfig)
- cert_specs = [
- CertificateSpec(domains=env.domains, key_type='rsa4096'),
- CertificateSpec(domains=env.domains_noh2, key_type='rsa2048'),
- ]
- ca = H2TestCA.create_root(name=env.http_tld,
- store_dir=os.path.join(env.server_dir, 'ca'), key_type="rsa4096")
- ca.issue_certs(cert_specs)
- env.set_ca(ca)
env.apache_access_log_clear()
env.apache_error_log_clear()
return env
-@pytest.fixture(autouse=True, scope="session")
+@pytest.fixture(autouse=True, scope="module")
def _session_scope(env):
yield
assert env.apache_stop() == 0
--- /dev/null
+import inspect
+import logging
+import os
+import subprocess
+
+from pyhttpd.certs import CertificateSpec
+from pyhttpd.conf import HttpdConf
+from pyhttpd.env import HttpdTestEnv, HttpdTestSetup
+
+log = logging.getLogger(__name__)
+
+
+class H2TestSetup(HttpdTestSetup):
+
+ def __init__(self, env: 'HttpdTestEnv'):
+ super().__init__(env=env)
+
+ def make(self):
+ super().make(add_modules=["http2", "proxy_http2"])
+ self._add_h2test()
+
+ def _add_h2test(self):
+ p = subprocess.run([self.env.apxs, '-c', 'mod_h2test.c'],
+ capture_output=True,
+ cwd=os.path.join(self.env.local_dir, 'mod_h2test'))
+ rv = p.returncode
+ if rv != 0:
+ log.error(f"compiling md_h2test failed: {p.stderr}")
+ raise Exception(f"compiling md_h2test failed: {p.stderr}")
+
+ modules_conf = os.path.join(self.env.server_dir, 'conf/modules.conf')
+ with open(modules_conf, 'a') as fd:
+ # load our test module which is not installed
+ fd.write(f"LoadModule h2test_module \"{self.env.local_dir}/mod_h2test/.libs/mod_h2test.so\"\n")
+
+
+class H2TestEnv(HttpdTestEnv):
+
+ def __init__(self, pytestconfig=None, setup_dirs=True):
+ super().__init__(pytestconfig=pytestconfig,
+ local_dir=os.path.dirname(inspect.getfile(H2TestEnv)),
+ add_base_conf="""
+ H2MinWorkers 1
+ H2MaxWorkers 64
+ """,
+ interesting_modules=["http2", "proxy_http2", "h2test"])
+ self.add_cert_specs([
+ CertificateSpec(domains=[
+ f"push.{self._http_tld}",
+ f"hints.{self._http_tld}",
+ f"ssl.{self._http_tld}",
+ f"pad0.{self._http_tld}",
+ f"pad1.{self._http_tld}",
+ f"pad2.{self._http_tld}",
+ f"pad3.{self._http_tld}",
+ f"pad8.{self._http_tld}",
+ ]),
+ CertificateSpec(domains=[f"noh2.{self.http_tld}"], key_type='rsa2048'),
+ ])
+ if setup_dirs:
+ self._setup = H2TestSetup(env=self)
+ self._setup.make()
+ self.issue_certs()
+ self.setup_data_1k_1m()
+
+
+ def setup_data_1k_1m(self):
+ s100 = "012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678\n"
+ with open(os.path.join(self.gen_dir, "data-1k"), 'w') as f:
+ for i in range(10):
+ f.write(s100)
+ with open(os.path.join(self.gen_dir, "data-10k"), 'w') as f:
+ for i in range(100):
+ f.write(s100)
+ with open(os.path.join(self.gen_dir, "data-100k"), 'w') as f:
+ for i in range(1000):
+ f.write(s100)
+ with open(os.path.join(self.gen_dir, "data-1m"), 'w') as f:
+ for i in range(10000):
+ f.write(s100)
+
+
+class H2Conf(HttpdConf):
+
+ def __init__(self, env: HttpdTestEnv, path=None):
+ super().__init__(env=env, path=path)
+
+
+ def add_vhost_noh2(self):
+ self.start_vhost(self.env.https_port, "noh2", aliases=["noh2-alias"], doc_root="htdocs/noh2", with_ssl=True)
+ self.add(f"""
+ Protocols http/1.1
+ SSLOptions +StdEnvVars""")
+ self.end_vhost()
+ self.start_vhost(self.env.http_port, "noh2", aliases=["noh2-alias"], doc_root="htdocs/noh2", with_ssl=False)
+ self.add(" Protocols http/1.1")
+ self.add(" SSLOptions +StdEnvVars")
+ self.end_vhost()
+ return self
+++ /dev/null
-#
-# mod-h2 test suite
-# check that our test infrastructure is sane
-#
-import pytest
-
-
-class TestStore:
-
- @pytest.fixture(autouse=True, scope='class')
- def _class_scope(self, env):
- env.setup_data_1k_1m()
- yield
-
- def test_000_00(self):
- assert 1 == 1
-
import pytest
-from h2_conf import HttpdConf
+from .env import H2Conf
class TestStore:
@pytest.fixture(autouse=True, scope='class')
def _class_scope(self, env):
- HttpdConf(env).add_vhost_test1().install()
+ H2Conf(env).add_vhost_test1().install()
assert env.apache_restart() == 0
# we expect to see the document from the generic server
- def test_001_01(self, env):
- r = env.curl_get(f"https://{env.domain_test1}:{env.https_port}/alive.json", 5)
+ def test_h2_001_01(self, env):
+ url = env.mkurl("https", "test1", "/alive.json")
+ r = env.curl_get(url, 5)
assert r.exit_code == 0, r.stderr + r.stdout
assert r.response["json"]
- assert True == r.response["json"]["alive"]
- assert "test1" == r.response["json"]["host"]
-
- # we expect to see the document from the generic server
- def test_001_02(self, env):
- r = env.curl_get(f"https://{env.domain_test1}:{env.https_port}/alive.json", 5)
- assert r.exit_code == 0, r.stderr
- assert r.response["json"]
- assert True == r.response["json"]["alive"]
- assert "test1" == r.response["json"]["host"]
+ assert r.response["json"]["alive"] is True
+ assert r.response["json"]["host"] == "test1"
import pytest
-from h2_conf import HttpdConf
+from .env import H2Conf
class TestStore:
@pytest.fixture(autouse=True, scope='class')
def _class_scope(self, env):
- HttpdConf(env).add_vhost_test1().add_vhost_test2().install()
+ H2Conf(env).add_vhost_test1().add_vhost_test2().install()
assert env.apache_restart() == 0
# check that we see the correct documents when using the test1 server name over http:
- def test_002_01(self, env):
+ def test_h2_002_01(self, env):
url = env.mkurl("http", "test1", "/alive.json")
r = env.curl_get(url, 5)
assert 200 == r.response["status"]
assert "test1" == r.response["json"]["host"]
# check that we see the correct documents when using the test1 server name over https:
- def test_002_02(self, env):
+ def test_h2_002_02(self, env):
url = env.mkurl("https", "test1", "/alive.json")
r = env.curl_get(url, 5)
assert 200 == r.response["status"]
assert "application/json" == r.response["header"]["content-type"]
# enforce HTTP/1.1
- def test_002_03(self, env):
+ def test_h2_002_03(self, env):
url = env.mkurl("https", "test1", "/alive.json")
r = env.curl_get(url, 5, [ "--http1.1" ])
assert 200 == r.response["status"]
assert "HTTP/1.1" == r.response["protocol"]
# enforce HTTP/2
- def test_002_04(self, env):
+ def test_h2_002_04(self, env):
url = env.mkurl("https", "test1", "/alive.json")
r = env.curl_get(url, 5, [ "--http2" ])
assert 200 == r.response["status"]
assert "HTTP/2" == r.response["protocol"]
# default is HTTP/2 on this host
- def test_002_04b(self, env):
+ def test_h2_002_04b(self, env):
url = env.mkurl("https", "test1", "/alive.json")
r = env.curl_get(url, 5)
assert 200 == r.response["status"]
assert "test1" == r.response["json"]["host"]
# although, without ALPN, we cannot select it
- def test_002_05(self, env):
+ def test_h2_002_05(self, env):
url = env.mkurl("https", "test1", "/alive.json")
r = env.curl_get(url, 5, [ "--no-alpn" ])
assert 200 == r.response["status"]
assert "test1" == r.response["json"]["host"]
# default is HTTP/1.1 on the other
- def test_002_06(self, env):
+ def test_h2_002_06(self, env):
url = env.mkurl("https", "test2", "/alive.json")
r = env.curl_get(url, 5)
assert 200 == r.response["status"]
import re
import pytest
-from h2_conf import HttpdConf
+from .env import H2Conf
class TestStore:
@pytest.fixture(autouse=True, scope='class')
def _class_scope(self, env):
- HttpdConf(env).add_vhost_cgi(
+ H2Conf(env).add_vhost_cgi(
proxy_self=True, h2proxy_self=True
).add_vhost_test1(
proxy_self=True, h2proxy_self=True
assert env.apache_restart() == 0
# check SSL environment variables from CGI script
- def test_003_01(self, env):
+ def test_h2_003_01(self, env):
url = env.mkurl("https", "cgi", "/hello.py")
r = env.curl_get(url, 5, ["--tlsv1.2"])
assert 200 == r.response["status"]
assert "" == r.response["json"]["h2push"]
# retrieve a html file from the server and compare it to its source
- def test_003_02(self, env):
- with open(env.test_src("htdocs/test1/index.html"), mode='rb') as file:
+ def test_h2_003_02(self, env):
+ with open(env.htdocs_src("test1/index.html"), mode='rb') as file:
src = file.read()
url = env.mkurl("https", "test1", "/index.html")
exp += text + "\n"
assert exp == r.response["body"].decode('utf-8')
- def test_003_10(self, env):
+ def test_h2_003_10(self, env):
self.check_necho(env, 10, "0123456789")
- def test_003_11(self, env):
+ def test_h2_003_11(self, env):
self.check_necho(env, 100, "0123456789")
- def test_003_12(self, env):
+ def test_h2_003_12(self, env):
self.check_necho(env, 1000, "0123456789")
- def test_003_13(self, env):
+ def test_h2_003_13(self, env):
self.check_necho(env, 10000, "0123456789")
- def test_003_14(self, env):
+ def test_h2_003_14(self, env):
self.check_necho(env, 100000, "0123456789")
# github issue #126
- def test_003_20(self, env):
+ def test_h2_003_20(self, env):
url = env.mkurl("https", "test1", "/006/")
r = env.curl_get(url, 5)
assert 200 == r.response["status"]
s = re.sub(r'^vary:.*\n', '', s, flags=re.MULTILINE)
return re.sub(r'^accept-ranges:.*\n', '', s, flags=re.MULTILINE)
- def test_003_21(self, env):
+ def test_h2_003_21(self, env):
url = env.mkurl("https", "test1", "/index.html")
r = env.curl_get(url, 5, ["-I"])
assert 200 == r.response["status"]
@pytest.mark.parametrize("path", [
"/004.html", "/proxy/004.html", "/h2proxy/004.html"
])
- def test_003_30(self, env, path):
+ def test_h2_003_30(self, env, path):
url = env.mkurl("https", "test1", path)
r = env.curl_get(url, 5)
assert 200 == r.response["status"]
@pytest.mark.parametrize("path", [
"/004.html", "/proxy/004.html", "/h2proxy/004.html"
])
- def test_003_31(self, env, path):
+ def test_h2_003_31(self, env, path):
url = env.mkurl("https", "test1", path)
r = env.curl_get(url, 5)
assert 200 == r.response["status"]
assert 304 == r.response["status"]
# test various response body lengths to work correctly
- def test_003_40(self, env):
+ def test_h2_003_40(self, env):
n = 1001
while n <= 1025024:
url = env.mkurl("https", "cgi", f"/mnot164.py?count={n}&text=X")
@pytest.mark.parametrize("n", [
0, 1, 1291, 1292, 80000, 80123, 81087, 98452
])
- def test_003_41(self, env, n):
+ def test_h2_003_41(self, env, n):
url = env.mkurl("https", "cgi", f"/mnot164.py?count={n}&text=X")
r = env.curl_get(url, 5)
assert 200 == r.response["status"]
@pytest.mark.parametrize("path", [
"/004.html", "/proxy/004.html", "/h2proxy/004.html"
])
- def test_003_50(self, env, path):
+ def test_h2_003_50(self, env, path):
# check that the resource supports ranges and we see its raw content-length
url = env.mkurl("https", "test1", path)
r = env.curl_get(url, 5)
import re
import pytest
-from h2_conf import HttpdConf
+from .env import H2Conf
class TestStore:
@pytest.fixture(autouse=True, scope='class')
def _class_scope(self, env):
env.setup_data_1k_1m()
- HttpdConf(env).add_vhost_cgi().install()
+ H2Conf(env).add_vhost_cgi().install()
assert env.apache_restart() == 0
# upload and GET again using curl, compare to original content
r2 = env.curl_get(r.response["header"]["location"])
assert r2.exit_code == 0
assert r2.response["status"] == 200
- with open(env.test_src(fpath), mode='rb') as file:
+ with open(env.local_src(fpath), mode='rb') as file:
src = file.read()
assert src == r2.response["body"]
- def test_004_01(self, env):
+ def test_h2_004_01(self, env):
self.curl_upload_and_verify(env, "data-1k", ["--http1.1"])
self.curl_upload_and_verify(env, "data-1k", ["--http2"])
- def test_004_02(self, env):
+ def test_h2_004_02(self, env):
self.curl_upload_and_verify(env, "data-10k", ["--http1.1"])
self.curl_upload_and_verify(env, "data-10k", ["--http2"])
- def test_004_03(self, env):
+ def test_h2_004_03(self, env):
self.curl_upload_and_verify(env, "data-100k", ["--http1.1"])
self.curl_upload_and_verify(env, "data-100k", ["--http2"])
- def test_004_04(self, env):
+ def test_h2_004_04(self, env):
self.curl_upload_and_verify(env, "data-1m", ["--http1.1"])
self.curl_upload_and_verify(env, "data-1m", ["--http2"])
- def test_004_05(self, env):
+ def test_h2_004_05(self, env):
self.curl_upload_and_verify(env, "data-1k", ["-v", "--http1.1", "-H", "Expect: 100-continue"])
self.curl_upload_and_verify(env, "data-1k", ["-v", "--http2", "-H", "Expect: 100-continue"])
@pytest.mark.skipif(True, reason="python3 regresses in chunked inputs to cgi")
- def test_004_06(self, env):
+ def test_h2_004_06(self, env):
self.curl_upload_and_verify(env, "data-1k", ["--http1.1", "-H", "Content-Length: "])
self.curl_upload_and_verify(env, "data-1k", ["--http2", "-H", "Content-Length: "])
("H2_STREAM_ID", "1"),
("H2_STREAM_TAG", r'\d+-1'),
])
- def test_004_07(self, env, name, value):
+ def test_h2_004_07(self, env, name, value):
url = env.mkurl("https", "cgi", "/env.py")
r = env.curl_post_value(url, "name", name)
assert r.exit_code == 0
# verify that we parse nghttp output correctly
def check_nghttp_body(self, env, ref_input, nghttp_output):
- with open(env.test_src(os.path.join(env.gen_dir, ref_input)), mode='rb') as f:
+ with open(env.local_src(os.path.join(env.gen_dir, ref_input)), mode='rb') as f:
refbody = f.read()
- with open(env.test_src(nghttp_output), mode='rb') as f:
+ with open(env.local_src(nghttp_output), mode='rb') as f:
text = f.read()
o = env.nghttp().parse_output(text)
assert "response" in o
assert "body" in o["response"]
if refbody != o["response"]["body"]:
- with open(env.test_src(os.path.join(env.gen_dir, '%s.parsed' % ref_input)), mode='bw') as f:
+ with open(env.local_src(os.path.join(env.gen_dir, '%s.parsed' % ref_input)), mode='bw') as f:
f.write(o["response"]["body"])
assert len(refbody) == len(o["response"]["body"])
assert refbody == o["response"]["body"]
- def test_004_20(self, env):
+ def test_h2_004_20(self, env):
self.check_nghttp_body(env, 'data-1k', 'data/nghttp-output-1k-1.txt')
self.check_nghttp_body(env, 'data-10k', 'data/nghttp-output-10k-1.txt')
self.check_nghttp_body(env, 'data-100k', 'data/nghttp-output-100k-1.txt')
assert r.exit_code == 0
assert r.response["status"] >= 200 and r.response["status"] < 300
- with open(env.test_src(fpath), mode='rb') as file:
+ with open(env.local_src(fpath), mode='rb') as file:
src = file.read()
assert src == r.response["body"]
@pytest.mark.parametrize("name", [
"data-1k", "data-10k", "data-100k", "data-1m"
])
- def test_004_21(self, env, name):
+ def test_h2_004_21(self, env, name):
self.nghttp_post_and_verify(env, name, [])
@pytest.mark.parametrize("name", [
"data-1k", "data-10k", "data-100k", "data-1m"
])
- def test_004_22(self, env, name, repeat):
+ def test_h2_004_22(self, env, name, repeat):
self.nghttp_post_and_verify(env, name, ["--no-content-length"])
# upload and GET again using nghttp, compare to original content
r2 = env.nghttp().get(r.response["header"]["location"])
assert r2.exit_code == 0
assert r2.response["status"] == 200
- with open(env.test_src(fpath), mode='rb') as file:
+ with open(env.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_004_23(self, env, name, repeat):
+ def test_h2_004_23(self, env, name, repeat):
self.nghttp_upload_and_verify(env, name, [])
@pytest.mark.parametrize("name", [
"data-1k", "data-10k", "data-100k", "data-1m"
])
- def test_004_24(self, env, name, repeat):
+ def test_h2_004_24(self, env, name, repeat):
self.nghttp_upload_and_verify(env, name, ["--expect-continue"])
@pytest.mark.parametrize("name", [
"data-1k", "data-10k", "data-100k", "data-1m"
])
- def test_004_25(self, env, name, repeat):
+ def test_h2_004_25(self, env, name, repeat):
self.nghttp_upload_and_verify(env, name, ["--no-content-length"])
- def test_004_30(self, env):
+ def test_h2_004_30(self, env):
# issue: #203
resource = "data-1k"
full_length = 1000
logfile = os.path.join(env.server_logs_dir, "test_004_30")
if os.path.isfile(logfile):
os.remove(logfile)
- HttpdConf(env).add("""
+ H2Conf(env).add("""
LogFormat "{ \\"request\\": \\"%r\\", \\"status\\": %>s, \\"bytes_resp_B\\": %B, \\"bytes_tx_O\\": %O, \\"bytes_rx_I\\": %I, \\"bytes_rx_tx_S\\": %S }" issue_203
CustomLog logs/test_004_30 issue_203
""").add_vhost_cgi().install()
assert log_h2['bytes_resp_B'] == chunk
assert log_h2['bytes_tx_O'] > chunk
- def test_004_40(self, env):
+ def test_h2_004_40(self, env):
# echo content using h2test_module "echo" handler
def post_and_verify(fname, options=None):
url = env.mkurl("https", "cgi", "/h2test/echo")
if fname == part.get_filename():
filepart = part
assert filepart
- with open(env.test_src(fpath), mode='rb') as file:
+ with open(env.local_src(fpath), mode='rb') as file:
src = file.read()
assert src == filepart.get_payload(decode=True)
import os
import pytest
-from h2_conf import HttpdConf
+from .env import H2Conf
def mk_text_file(fpath: str, lines: int):
mk_text_file(os.path.join(docs_a, fname), 8 * fsize)
self.URI_PATHS.append(f"/files/{fname}")
- HttpdConf(env).add_vhost_cgi(
+ H2Conf(env).add_vhost_cgi(
proxy_self=True, h2proxy_self=True
).add_vhost_test1(
proxy_self=True, h2proxy_self=True
).install()
assert env.apache_restart() == 0
- def test_005_01(self, env):
+ def test_h2_005_01(self, env):
url = env.mkurl("https", "cgi", self.URI_PATHS[2])
r = env.curl_get(url)
assert r.response, r.stderr + r.stdout
import pytest
-from h2_conf import HttpdConf
+from .env import H2Conf
class TestStore:
@pytest.fixture(autouse=True, scope='class')
def _class_scope(self, env):
- HttpdConf(env).add_vhost_test1().install()
+ H2Conf(env).add_vhost_test1().install()
assert env.apache_restart() == 0
# single page without any assets
- def test_006_01(self, env):
+ def test_h2_006_01(self, env):
url = env.mkurl("https", "test1", "/001.html")
r = env.nghttp().assets(url, options=["-Haccept-encoding: none"])
assert 0 == r.exit_code
]
# single image without any assets
- def test_006_02(self, env):
+ def test_h2_006_02(self, env):
url = env.mkurl("https", "test1", "/002.jpg")
r = env.nghttp().assets(url, options=["-Haccept-encoding: none"])
assert 0 == r.exit_code
]
# gophertiles, yea!
- def test_006_03(self, env):
+ def test_h2_006_03(self, env):
# create the tiles files we originally had checked in
exp_assets = [
{"status": 200, "size": "10K", "path": "/004.html"},
assert r.assets == exp_assets
# page with js and css
- def test_006_04(self, env):
+ def test_h2_006_04(self, env):
url = env.mkurl("https", "test1", "/006.html")
r = env.nghttp().assets(url, options=["-Haccept-encoding: none"])
assert 0 == r.exit_code
]
# page with image, try different window size
- def test_006_05(self, env):
+ def test_h2_006_05(self, env):
url = env.mkurl("https", "test1", "/003.html")
r = env.nghttp().assets(url, options=["--window-bits=24", "-Haccept-encoding: none"])
assert 0 == r.exit_code
import pytest
-from h2_conf import HttpdConf
+from .env import H2Conf
class TestStore:
@pytest.fixture(autouse=True, scope='class')
def _class_scope(self, env):
- HttpdConf(env).add_vhost_noh2().add_vhost_test1().add_vhost_cgi().install()
+ H2Conf(env).add_vhost_noh2().add_vhost_test1().add_vhost_cgi().install()
assert env.apache_restart() == 0
# make sure the protocol selection on the different hosts work as expected
- def test_100_01(self, env):
+ def test_h2_100_01(self, env):
# this host defaults to h2, but we can request h1
url = env.mkurl("https", "cgi", "/hello.py")
assert "2" == env.curl_protocol_version( url )
assert "1.1" == env.curl_protocol_version( url, options=[ "--http2" ] )
# access a ServerAlias, after using ServerName in SNI
- def test_100_02(self, env):
+ def test_h2_100_02(self, env):
url = env.mkurl("https", "cgi", "/hello.py")
hostname = ("cgi-alias.%s" % env.http_tld)
r = env.curl_get(url, 5, [ "-H", "Host:%s" % hostname ])
assert hostname == r.response["json"]["host"]
# access another vhost, after using ServerName in SNI, that uses same SSL setup
- def test_100_03(self, env):
+ def test_h2_100_03(self, env):
url = env.mkurl("https", "cgi", "/")
hostname = ("test1.%s" % env.http_tld)
r = env.curl_get(url, 5, [ "-H", "Host:%s" % hostname ])
# access another vhost, after using ServerName in SNI,
# that has different SSL certificate. This triggers a 421 (misdirected request) response.
- def test_100_04(self, env):
+ def test_h2_100_04(self, env):
url = env.mkurl("https", "cgi", "/hello.py")
hostname = ("noh2.%s" % env.http_tld)
r = env.curl_get(url, 5, [ "-H", "Host:%s" % hostname ])
assert 421 == r.response["status"]
# access an unknown vhost, after using ServerName in SNI
- def test_100_05(self, env):
+ def test_h2_100_05(self, env):
url = env.mkurl("https", "cgi", "/hello.py")
hostname = ("unknown.%s" % env.http_tld)
r = env.curl_get(url, 5, [ "-H", "Host:%s" % hostname ])
import re
import pytest
-from h2_conf import HttpdConf
+from .env import H2Conf
class TestStore:
@pytest.fixture(autouse=True, scope='class')
def _class_scope(self, env):
- HttpdConf(env).add(
+ H2Conf(env).add(
f"""
SSLCipherSuite ECDHE-RSA-AES256-GCM-SHA384
<Directory \"{env.server_dir}/htdocs/ssl-client-verify\">
assert env.apache_restart() == 0
# access a resource with SSL renegotiation, using HTTP/1.1
- def test_101_01(self, env):
+ def test_h2_101_01(self, env):
url = env.mkurl("https", "ssl", "/renegotiate/cipher/")
r = env.curl_get(url, options=["-v", "--http1.1", "--tlsv1.2", "--tls-max", "1.2"])
assert 0 == r.exit_code
assert 403 == r.response["status"]
# try to renegotiate the cipher, should fail with correct code
- def test_101_02(self, env):
+ def test_h2_101_02(self, env):
url = env.mkurl("https", "ssl", "/renegotiate/cipher/")
r = env.curl_get(url, options=[
"-vvv", "--tlsv1.2", "--tls-max", "1.2", "--ciphers", "ECDHE-RSA-AES256-GCM-SHA384"
# try to renegotiate a client certificate from Location
# needs to fail with correct code
- def test_101_03(self, env):
+ def test_h2_101_03(self, env):
url = env.mkurl("https", "ssl", "/renegotiate/verify/")
r = env.curl_get(url, options=["-vvv", "--tlsv1.2", "--tls-max", "1.2"])
assert 0 != r.exit_code
# try to renegotiate a client certificate from Directory
# needs to fail with correct code
- def test_101_04(self, env):
+ def test_h2_101_04(self, env):
url = env.mkurl("https", "ssl", "/ssl-client-verify/index.html")
r = env.curl_get(url, options=["-vvv", "--tlsv1.2", "--tls-max", "1.2"])
assert 0 != r.exit_code
# make 10 requests on the same connection, none should produce a status code
# reported by erki@example.ee
- def test_101_05(self, env):
+ def test_h2_101_05(self, env):
r = env.run([env.h2load, "-n", "10", "-c", "1", "-m", "1", "-vvvv",
f"{env.https_base_url}/ssl-client-verify/index.html"])
assert 0 == r.exit_code
# Check that "SSLRequireSSL" works on h2 connections
# See <https://bz.apache.org/bugzilla/show_bug.cgi?id=62654>
- def test_101_10a(self, env):
+ def test_h2_101_10a(self, env):
url = env.mkurl("https", "ssl", "/sslrequire/index.html")
r = env.curl_get(url)
assert 0 == r.exit_code
# Check that "require ssl" works on h2 connections
# See <https://bz.apache.org/bugzilla/show_bug.cgi?id=62654>
- def test_101_10b(self, env):
+ def test_h2_101_10b(self, env):
url = env.mkurl("https", "ssl", "/requiressl/index.html")
r = env.curl_get(url)
assert 0 == r.exit_code
assert 404 == r.response["status"]
# Check that status works with ErrorDoc, see pull #174, fixes #172
- def test_101_11(self, env):
+ def test_h2_101_11(self, env):
url = env.mkurl("https", "ssl", "/renegotiate/err-doc-cipher")
r = env.curl_get(url, options=[
"-vvv", "--tlsv1.2", "--tls-max", "1.2", "--ciphers", "ECDHE-RSA-AES256-GCM-SHA384"
import pytest
-from h2_conf import HttpdConf
+from .env import H2Conf
class TestStore:
@pytest.fixture(autouse=True, scope='class')
def _class_scope(self, env):
- conf = HttpdConf(env).start_vhost(env.https_port, "ssl", with_ssl=True)
+ conf = H2Conf(env).start_vhost(env.https_port, "ssl", with_ssl=True)
conf.add("""
Protocols h2 http/1.1
SSLOptions +StdEnvVars
env.mkpath("%s/htdocs/ssl-client-verify" % env.server_dir)
assert env.apache_restart() == 0
- def test_102_01(self, env):
+ def test_h2_102_01(self, env):
url = env.mkurl("https", "ssl", "/h2only.html")
r = env.curl_get(url)
assert 0 == r.exit_code
assert r.response
assert 404 == r.response["status"]
- def test_102_02(self, env):
+ def test_h2_102_02(self, env):
url = env.mkurl("https", "ssl", "/noh2.html")
r = env.curl_get(url)
assert 0 == r.exit_code
import pytest
-from h2_conf import HttpdConf
+from .env import H2Conf
class TestStore:
@pytest.fixture(autouse=True, scope='class')
def _class_scope(self, env):
- HttpdConf(env).add_vhost_test1().add_vhost_test2().add_vhost_noh2(
+ H2Conf(env).add_vhost_test1().add_vhost_test2().add_vhost_noh2(
).start_vhost(
env.https_port, "test3", doc_root="htdocs/test1", with_ssl=True
).add(
assert env.apache_restart() == 0
# accessing http://test1, will not try h2 and advertise h2 in the response
- def test_103_01(self, env):
+ def test_h2_103_01(self, env):
url = env.mkurl("http", "test1", "/index.html")
r = env.curl_get(url)
assert 0 == r.exit_code
assert "h2c" == r.response["header"]["upgrade"]
# accessing http://noh2, will not advertise, because noh2 host does not have it enabled
- def test_103_02(self, env):
+ def test_h2_103_02(self, env):
url = env.mkurl("http", "noh2", "/index.html")
r = env.curl_get(url)
assert 0 == r.exit_code
assert "upgrade" not in r.response["header"]
# accessing http://test2, will not advertise, because h2 has less preference than http/1.1
- def test_103_03(self, env):
+ def test_h2_103_03(self, env):
url = env.mkurl("http", "test2", "/index.html")
r = env.curl_get(url)
assert 0 == r.exit_code
assert "upgrade" not in r.response["header"]
# accessing https://noh2, will not advertise, because noh2 host does not have it enabled
- def test_103_04(self, env):
+ def test_h2_103_04(self, env):
url = env.mkurl("https", "noh2", "/index.html")
r = env.curl_get(url)
assert 0 == r.exit_code
assert "upgrade" not in r.response["header"]
# accessing https://test2, will not advertise, because h2 has less preference than http/1.1
- def test_103_05(self, env):
+ def test_h2_103_05(self, env):
url = env.mkurl("https", "test2", "/index.html")
r = env.curl_get(url)
assert 0 == r.exit_code
assert "upgrade" not in r.response["header"]
# accessing https://test1, will advertise h2 in the response
- def test_103_06(self, env):
+ def test_h2_103_06(self, env):
url = env.mkurl("https", "test1", "/index.html")
r = env.curl_get(url, options=["--http1.1"])
assert 0 == r.exit_code
assert "h2" == r.response["header"]["upgrade"]
# accessing https://test3, will not send Upgrade since it is suppressed
- def test_103_07(self, env):
+ def test_h2_103_07(self, env):
url = env.mkurl("https", "test3", "/index.html")
r = env.curl_get(url, options=["--http1.1"])
assert 0 == r.exit_code
assert "upgrade" not in r.response["header"]
# upgrade to h2c for a request, where h2c is preferred
- def test_103_20(self, env):
+ def test_h2_103_20(self, env):
url = env.mkurl("http", "test1", "/index.html")
r = env.nghttp().get(url, options=["-u"])
assert 200 == r.response["status"]
# upgrade to h2c for a request where http/1.1 is preferred, but the clients upgrade
# wish is honored nevertheless
- def test_103_21(self, env):
+ def test_h2_103_21(self, env):
url = env.mkurl("http", "test2", "/index.html")
r = env.nghttp().get(url, options=["-u"])
assert 404 == r.response["status"]
# ugrade to h2c on a host where h2c is not enabled will fail
- def test_103_22(self, env):
+ def test_h2_103_22(self, env):
url = env.mkurl("http", "noh2", "/index.html")
r = env.nghttp().get(url, options=["-u"])
assert not r.response
# ugrade to h2c on a host where h2c is preferred, but Upgrade is disabled
- def test_103_23(self, env):
+ def test_h2_103_23(self, env):
url = env.mkurl("http", "test1b", "/index.html")
r = env.nghttp().get(url, options=["-u"])
assert not r.response
# ugrade to h2c on a host where h2c is preferred, but Upgrade is disabled on the server,
# but allowed for a specific location
- def test_103_24(self, env):
+ def test_h2_103_24(self, env):
url = env.mkurl("http", "test1b", "/006.html")
r = env.nghttp().get(url, options=["-u"])
assert 200 == r.response["status"]
import pytest
-from h2_conf import HttpdConf
+from .env import H2Conf
def frame_padding(payload, padbits):
@pytest.fixture(autouse=True, scope='class')
def _class_scope(self, env):
- conf = HttpdConf(env)
- conf.add_vhost_cgi()
+ conf = H2Conf(env)
+ conf.start_vhost(env.https_port, "ssl", doc_root="htdocs/cgi", with_ssl=True)
+ conf.add("Protocols h2 http/1.1")
+ conf.add("AddHandler cgi-script .py")
+ conf.end_vhost()
conf.start_vhost(env.https_port, "pad0", doc_root="htdocs/cgi", with_ssl=True)
conf.add("Protocols h2 http/1.1")
conf.add("H2Padding 0")
assert env.apache_restart() == 0
# default paddings settings: 0 bits
- def test_104_01(self, env):
- url = env.mkurl("https", "cgi", "/echo.py")
+ def test_h2_104_01(self, env):
+ url = env.mkurl("https", "ssl", "/echo.py")
# we get 2 frames back: one with data and an empty one with EOF
# check the number of padding bytes is as expected
for data in ["x", "xx", "xxx", "xxxx", "xxxxx", "xxxxxx", "xxxxxxx", "xxxxxxxx"]:
]
# 0 bits of padding
- def test_104_02(self, env):
+ def test_h2_104_02(self, env):
url = env.mkurl("https", "pad0", "/echo.py")
for data in ["x", "xx", "xxx", "xxxx", "xxxxx", "xxxxxx", "xxxxxxx", "xxxxxxxx"]:
r = env.nghttp().post_data(url, data, 5)
assert r.results["paddings"] == [0, 0]
# 1 bit of padding
- def test_104_03(self, env):
+ def test_h2_104_03(self, env):
url = env.mkurl("https", "pad1", "/echo.py")
for data in ["x", "xx", "xxx", "xxxx", "xxxxx", "xxxxxx", "xxxxxxx", "xxxxxxxx"]:
r = env.nghttp().post_data(url, data, 5)
assert i in range(0, 2)
# 2 bits of padding
- def test_104_04(self, env):
+ def test_h2_104_04(self, env):
url = env.mkurl("https", "pad2", "/echo.py")
for data in ["x", "xx", "xxx", "xxxx", "xxxxx", "xxxxxx", "xxxxxxx", "xxxxxxxx"]:
r = env.nghttp().post_data(url, data, 5)
assert i in range(0, 4)
# 3 bits of padding
- def test_104_05(self, env):
+ def test_h2_104_05(self, env):
url = env.mkurl("https", "pad3", "/echo.py")
for data in ["x", "xx", "xxx", "xxxx", "xxxxx", "xxxxxx", "xxxxxxx", "xxxxxxxx"]:
r = env.nghttp().post_data(url, data, 5)
assert i in range(0, 8)
# 8 bits of padding
- def test_104_06(self, env):
+ def test_h2_104_06(self, env):
url = env.mkurl("https", "pad8", "/echo.py")
for data in ["x", "xx", "xxx", "xxxx", "xxxxx", "xxxxxx", "xxxxxxx", "xxxxxxxx"]:
r = env.nghttp().post_data(url, data, 5)
import pytest
-from h2_conf import HttpdConf
-from h2_curl import CurlPiper
+from .env import H2Conf
+from pyhttpd.curl import CurlPiper
class TestStore:
# Check that base servers 'Timeout' setting is observed on SSL handshake
- def test_105_01(self, env):
- conf = HttpdConf(env)
+ def test_h2_105_01(self, env):
+ conf = H2Conf(env)
conf.add("""
AcceptFilter http none
Timeout 1.5
sock.close()
# Check that mod_reqtimeout handshake setting takes effect
- def test_105_02(self, env):
- conf = HttpdConf(env)
+ def test_h2_105_02(self, env):
+ conf = H2Conf(env)
conf.add("""
AcceptFilter http none
Timeout 10
# Check that mod_reqtimeout handshake setting do no longer apply to handshaked
# connections. See <https://github.com/icing/mod_h2/issues/196>.
- def test_105_03(self, env):
- conf = HttpdConf(env)
+ def test_h2_105_03(self, env):
+ conf = H2Conf(env)
conf.add("""
Timeout 10
RequestReadTimeout handshake=1 header=5 body=10
])
assert 200 == r.response["status"]
- def test_105_10(self, env):
+ def test_h2_105_10(self, env):
# just a check without delays if all is fine
- conf = HttpdConf(env)
+ conf = H2Conf(env)
conf.add_vhost_cgi()
conf.install()
assert env.apache_restart() == 0
assert len("".join(stdout)) == 3 * 8192
@pytest.mark.skipif(True, reason="new feature in upcoming http2")
- def test_105_11(self, env):
+ def test_h2_105_11(self, env):
# short connection timeout, longer stream delay
# receiving the first response chunk, then timeout
- conf = HttpdConf(env)
+ conf = H2Conf(env)
conf.add_vhost_cgi()
conf.add("Timeout 1")
conf.install()
assert len("".join(stdout)) == 8192
@pytest.mark.skipif(True, reason="new feature in upcoming http2")
- def test_105_12(self, env):
+ def test_h2_105_12(self, env):
# long connection timeout, short stream timeout
# sending a slow POST
- conf = HttpdConf(env)
+ conf = H2Conf(env)
conf.add_vhost_cgi()
conf.add("Timeout 10")
conf.add("H2StreamTimeout 1")
import pytest
-from h2_conf import HttpdConf
-from h2_result import ExecResult
+from .env import H2Conf
+from pyhttpd.result import ExecResult
class TestShutdown:
@pytest.fixture(autouse=True, scope='class')
def _class_scope(self, env):
- conf = HttpdConf(env)
+ conf = H2Conf(env)
conf.add_vhost_cgi()
conf.install()
assert env.apache_restart() == 0
- def test_106_01(self, env):
+ def test_h2_106_01(self, env):
url = env.mkurl("https", "cgi", "/necho.py")
lines = 100000
text = "123456789"
import pytest
-from h2_conf import HttpdConf
+from .env import H2Conf
class TestStore:
@pytest.fixture(autouse=True, scope='class')
def _class_scope(self, env):
- HttpdConf(env).add_vhost_cgi().install()
+ H2Conf(env).add_vhost_cgi().install()
assert env.apache_restart() == 0
# let the hecho.py CGI echo chars < 0x20 in field name
# for almost all such characters, the stream gets aborted with a h2 error and
# there will be no http status, cr and lf are handled special
- def test_200_01(self, env):
+ def test_h2_200_01(self, env):
url = env.mkurl("https", "cgi", "/hecho.py")
for x in range(1, 32):
r = env.curl_post_data(url, "name=x%%%02xx&value=yz" % x)
# let the hecho.py CGI echo chars < 0x20 in field value
# for almost all such characters, the stream gets aborted with a h2 error and
# there will be no http status, cr and lf are handled special
- def test_200_02(self, env):
+ def test_h2_200_02(self, env):
url = env.mkurl("https", "cgi", "/hecho.py")
for x in range(1, 32):
if 9 != x:
assert 0 != r.exit_code, "unexpected exit code for char 0x%02x" % x
# let the hecho.py CGI echo 0x10 and 0x7f in field name and value
- def test_200_03(self, env):
+ def test_h2_200_03(self, env):
url = env.mkurl("https", "cgi", "/hecho.py")
for h in ["10", "7f"]:
r = env.curl_post_data(url, "name=x%%%s&value=yz" % h)
assert 0 != r.exit_code
# test header field lengths check, LimitRequestLine (default 8190)
- def test_200_10(self, env):
+ def test_h2_200_10(self, env):
url = env.mkurl("https", "cgi", "/")
val = "1234567890" # 10 chars
for i in range(3): # make a 10000 char string
assert 431 == r.response["status"]
# test header field lengths check, LimitRequestFieldSize (default 8190)
- def test_200_11(self, env):
+ def test_h2_200_11(self, env):
url = env.mkurl("https", "cgi", "/")
val = "1234567890" # 10 chars
for i in range(3): # make a 10000 char string
# test header field count, LimitRequestFields (default 100)
# see #201: several headers with same name are mered and count only once
- def test_200_12(self, env):
+ def test_h2_200_12(self, env):
url = env.mkurl("https", "cgi", "/")
opt = []
for i in range(98): # curl sends 2 headers itself (user-agent and accept)
# test header field count, LimitRequestFields (default 100)
# different header names count each
- def test_200_13(self, env):
+ def test_h2_200_13(self, env):
url = env.mkurl("https", "cgi", "/")
opt = []
for i in range(98): # curl sends 2 headers itself (user-agent and accept)
assert 431 == r.response["status"]
# test "LimitRequestFields 0" setting, see #200
- def test_200_14(self, env):
- conf = HttpdConf(env)
+ def test_h2_200_14(self, env):
+ conf = H2Conf(env)
conf.add("""
LimitRequestFields 20
""")
opt += ["-H", "x{0}: 1".format(i)]
r = env.curl_get(url, options=opt)
assert 431 == r.response["status"]
- conf = HttpdConf(env)
+ conf = H2Conf(env)
conf.add("""
LimitRequestFields 0
""")
assert 200 == r.response["status"]
# the uri limits
- def test_200_15(self, env):
- conf = HttpdConf(env)
+ def test_h2_200_15(self, env):
+ conf = H2Conf(env)
conf.add("""
LimitRequestLine 48
""")
assert not r.response
# invalid chars in method
- def test_200_16(self, env):
- conf = HttpdConf(env)
+ def test_h2_200_16(self, env):
+ conf = H2Conf(env)
conf.add_vhost_cgi()
conf.install()
assert env.apache_restart() == 0
import pytest
-from h2_conf import HttpdConf
+from .env import H2Conf
class TestStore:
@pytest.fixture(autouse=True, scope='class')
def _class_scope(self, env):
- HttpdConf(env).add(
+ H2Conf(env).add(
"""
KeepAlive on
MaxKeepAliveRequests 30
assert env.apache_restart() == 0
# check handling of 'if-modified-since' header
- def test_201_01(self, env):
+ def test_h2_201_01(self, env):
url = env.mkurl("https", "test1", "/006/006.css")
r = env.curl_get(url)
assert 200 == r.response["status"]
assert 200 == r.response["status"]
# check handling of 'if-none-match' header
- def test_201_02(self, env):
+ def test_h2_201_02(self, env):
url = env.mkurl("https", "test1", "/006/006.css")
r = env.curl_get(url)
assert 200 == r.response["status"]
assert 200 == r.response["status"]
@pytest.mark.skipif(True, reason="304 misses the Vary header in trunk and 2.4.x")
- def test_201_03(self, env):
+ def test_h2_201_03(self, env):
url = env.mkurl("https", "test1", "/006.html")
r = env.curl_get(url, options=["-H", "Accept-Encoding: gzip"])
assert 200 == r.response["status"]
assert "vary" in r.response["header"]
# Check if "Keep-Alive" response header is removed in HTTP/2.
- def test_201_04(self, env):
+ def test_h2_201_04(self, env):
url = env.mkurl("https", "test1", "/006.html")
r = env.curl_get(url, options=["--http1.1", "-H", "Connection: keep-alive"])
assert 200 == r.response["status"]
import os
import pytest
-from h2_conf import HttpdConf
+from .env import H2Conf
def setup_data(env):
@pytest.fixture(autouse=True, scope='class')
def _class_scope(self, env):
setup_data(env)
- HttpdConf(env).add_vhost_cgi(h2proxy_self=True).install()
+ H2Conf(env).add_vhost_cgi(h2proxy_self=True).install()
assert env.apache_restart() == 0
# check if the server survives a trailer or two
- def test_202_01(self, env):
+ def test_h2_202_01(self, env):
url = env.mkurl("https", "cgi", "/echo.py")
fpath = os.path.join(env.gen_dir, "data-1k")
r = env.nghttp().upload(url, fpath, options=["--trailer", "test: 1"])
assert 1000 == len(r.response["body"])
# check if the server survives a trailer without content-length
- def test_202_02(self, env):
+ def test_h2_202_02(self, env):
url = env.mkurl("https", "cgi", "/echo.py")
fpath = os.path.join(env.gen_dir, "data-1k")
r = env.nghttp().upload(url, fpath, options=["--trailer", "test: 2", "--no-content-length"])
assert 1000 == len(r.response["body"])
# check if echoing request headers in response from GET works
- def test_202_03(self, env):
+ def test_h2_202_03(self, env):
url = env.mkurl("https", "cgi", "/echohd.py?name=X")
r = env.nghttp().get(url, options=["--header", "X: 3"])
assert 300 > r.response["status"]
assert b"X: 3\n" == r.response["body"]
# check if echoing request headers in response from POST works
- def test_202_03b(self, env):
+ def test_h2_202_03b(self, env):
url = env.mkurl("https", "cgi", "/echohd.py?name=X")
r = env.nghttp().post_name(url, "Y", options=["--header", "X: 3b"])
assert 300 > r.response["status"]
# check if echoing request headers in response from POST works, but trailers are not seen
# This is the way CGI invocation works.
- def test_202_04(self, env):
+ def test_h2_202_04(self, env):
url = env.mkurl("https", "cgi", "/echohd.py?name=X")
r = env.nghttp().post_name(url, "Y", options=["--header", "X: 4a", "--trailer", "X: 4b"])
assert 300 > r.response["status"]
import pytest
-from h2_conf import HttpdConf
+from .env import H2Conf
class TestStore:
@pytest.fixture(autouse=True, scope='class')
def _class_scope(self, env):
- HttpdConf(env).add_vhost_test1().add_vhost_cgi().install()
+ H2Conf(env).add_vhost_test1().add_vhost_cgi().install()
assert env.apache_restart() == 0
def setup_method(self, method):
print("teardown_method: %s" % method.__name__)
# check that we normally do not see an interim response
- def test_300_01(self, env):
+ def test_h2_300_01(self, env):
url = env.mkurl("https", "test1", "/index.html")
r = env.curl_post_data(url, 'XYZ')
assert 200 == r.response["status"]
assert "previous" not in r.response
# check that we see an interim response when we ask for it
- def test_300_02(self, env):
+ def test_h2_300_02(self, env):
url = env.mkurl("https", "cgi", "/echo.py")
r = env.curl_post_data(url, 'XYZ', options=["-H", "expect: 100-continue"])
assert 200 == r.response["status"]
assert 100 == r.response["previous"]["status"]
# check proper answer on unexpected
- def test_300_03(self, env):
+ def test_h2_300_03(self, env):
url = env.mkurl("https", "cgi", "/echo.py")
r = env.curl_post_data(url, 'XYZ', options=["-H", "expect: the-unexpected"])
assert 417 == r.response["status"]
import os
import pytest
-from h2_conf import HttpdConf
+from .env import H2Conf
# The push tests depend on "nghttp"
@pytest.fixture(autouse=True, scope='class')
def _class_scope(self, env):
- HttpdConf(env).start_vhost(
+ H2Conf(env).start_vhost(
env.https_port, "push", doc_root="htdocs/test1", with_ssl=True
).add(r""" Protocols h2 http/1.1"
RewriteEngine on
# Link: header handling, various combinations
# plain resource without configured pushes
- def test_400_00(self, env):
+ def test_h2_400_00(self, env):
url = env.mkurl("https", "push", "/006.html")
r = env.nghttp().get(url)
assert 200 == r.response["status"]
assert 0 == len(promises)
# 2 link headers configured, only 1 triggers push
- def test_400_01(self, env):
+ def test_h2_400_01(self, env):
url = env.mkurl("https", "push", "/006-push.html")
r = env.nghttp().get(url, options=["-Haccept-encoding: none"])
assert 200 == r.response["status"]
assert 216 == len(promises[0]["response"]["body"])
# Same as 400_01, but with single header line configured
- def test_400_02(self, env):
+ def test_h2_400_02(self, env):
url = env.mkurl("https", "push", "/006-push2.html")
r = env.nghttp().get(url)
assert 200 == r.response["status"]
assert '/006/006.js' == promises[0]["request"]["header"][":path"]
# 2 Links, only one with correct rel attribue
- def test_400_03(self, env):
+ def test_h2_400_03(self, env):
url = env.mkurl("https", "push", "/006-push3.html")
r = env.nghttp().get(url)
assert 200 == r.response["status"]
assert '/006/006.js' == promises[0]["request"]["header"][":path"]
# Missing > in Link header, PUSH not triggered
- def test_400_04(self, env):
+ def test_h2_400_04(self, env):
url = env.mkurl("https", "push", "/006-push4.html")
r = env.nghttp().get(url)
assert 200 == r.response["status"]
assert 0 == len(promises)
# More than one value in "rel" parameter
- def test_400_05(self, env):
+ def test_h2_400_05(self, env):
url = env.mkurl("https", "push", "/006-push5.html")
r = env.nghttp().get(url)
assert 200 == r.response["status"]
assert '/006/006.css' == promises[0]["request"]["header"][":path"]
# Another "rel" parameter variation
- def test_400_06(self, env):
+ def test_h2_400_06(self, env):
url = env.mkurl("https", "push", "/006-push6.html")
r = env.nghttp().get(url)
assert 200 == r.response["status"]
assert '/006/006.css' == promises[0]["request"]["header"][":path"]
# Another "rel" parameter variation
- def test_400_07(self, env):
+ def test_h2_400_07(self, env):
url = env.mkurl("https", "push", "/006-push7.html")
r = env.nghttp().get(url)
assert 200 == r.response["status"]
assert '/006/006.css' == promises[0]["request"]["header"][":path"]
# Pushable link header with "nopush" attribute
- def test_400_08(self, env):
+ def test_h2_400_08(self, env):
url = env.mkurl("https", "push", "/006-push8.html")
r = env.nghttp().get(url)
assert 200 == r.response["status"]
assert 0 == len(promises)
# 2 H2PushResource config trigger on GET, but not on POST
- def test_400_20(self, env):
+ def test_h2_400_20(self, env):
url = env.mkurl("https", "push", "/006-push20.html")
r = env.nghttp().get(url)
assert 200 == r.response["status"]
assert 0 == len(promises)
# H2Push configured Off in location
- def test_400_30(self, env):
+ def test_h2_400_30(self, env):
url = env.mkurl("https", "push", "/006-push30.html")
r = env.nghttp().get(url)
assert 200 == r.response["status"]
assert 0 == len(promises)
# - suppress PUSH
- def test_400_50(self, env):
+ def test_h2_400_50(self, env):
url = env.mkurl("https", "push", "/006-push.html")
r = env.nghttp().get(url, options=['-H', 'accept-push-policy: none'])
assert 200 == r.response["status"]
assert 0 == len(promises)
# - default pushes desired
- def test_400_51(self, env):
+ def test_h2_400_51(self, env):
url = env.mkurl("https", "push", "/006-push.html")
r = env.nghttp().get(url, options=['-H', 'accept-push-policy: default'])
assert 200 == r.response["status"]
assert 1 == len(promises)
# - HEAD pushes desired
- def test_400_52(self, env):
+ def test_h2_400_52(self, env):
url = env.mkurl("https", "push", "/006-push.html")
r = env.nghttp().get(url, options=['-H', 'accept-push-policy: head'])
assert 200 == r.response["status"]
assert 0 == len(promises[0]["response"]["body"])
# - fast-load pushes desired
- def test_400_53(self, env):
+ def test_h2_400_53(self, env):
url = env.mkurl("https", "push", "/006-push.html")
r = env.nghttp().get(url, options=['-H', 'accept-push-policy: fast-load'])
assert 200 == r.response["status"]
import pytest
-from h2_conf import HttpdConf
+from .env import H2Conf
# The push tests depend on "nghttp"
@pytest.fixture(autouse=True, scope='class')
def _class_scope(self, env):
- HttpdConf(env).start_vhost(
+ H2Conf(env).start_vhost(
env.https_port, "hints", doc_root="htdocs/test1", with_ssl=True
).add(""" Protocols h2 http/1.1"
H2EarlyHints on
assert env.apache_restart() == 0
# H2EarlyHints enabled in general, check that it works for H2PushResource
- def test_401_31(self, env):
+ def test_h2_401_31(self, env):
url = env.mkurl("https", "hints", "/006-hints.html")
r = env.nghttp().get(url)
assert 200 == r.response["status"]
assert early["header"]["link"]
# H2EarlyHints enabled in general, but does not trigger on added response headers
- def test_401_32(self, env):
+ def test_h2_401_32(self, env):
url = env.mkurl("https", "hints", "/006-nohints.html")
r = env.nghttp().get(url)
assert 200 == r.response["status"]
import re
import pytest
-from h2_conf import HttpdConf
+from .env import H2Conf
class TestStore:
@pytest.fixture(autouse=True, scope='class')
def _class_scope(self, env):
env.setup_data_1k_1m()
- HttpdConf(env).add_vhost_cgi(proxy_self=True).install()
+ H2Conf(env).add_vhost_cgi(proxy_self=True).install()
assert env.apache_restart() == 0
def setup_method(self, method):
def teardown_method(self, method):
print("teardown_method: %s" % method.__name__)
- def test_500_01(self, env):
+ def test_h2_500_01(self, env):
url = env.mkurl("https", "cgi", "/proxy/hello.py")
r = env.curl_get(url, 5)
assert 200 == r.response["status"]
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(env.test_src(fpath), mode='rb') as file:
+ with open(env.local_src(fpath), mode='rb') as file:
src = file.read()
assert src == r2.response["body"]
- def test_500_10(self, env):
+ def test_h2_500_10(self, env):
self.curl_upload_and_verify(env, "data-1k", ["--http2"])
self.curl_upload_and_verify(env, "data-10k", ["--http2"])
self.curl_upload_and_verify(env, "data-100k", ["--http2"])
r = env.nghttp().upload(url, fpath, options=options)
assert r.exit_code == 0
assert 200 <= r.response["status"] < 300
- with open(env.test_src(fpath), mode='rb') as file:
+ with open(env.local_src(fpath), mode='rb') as file:
src = file.read()
assert src == r.response["body"]
- def test_500_20(self, env):
+ def test_h2_500_20(self, env):
self.nghttp_post_and_verify(env, "data-1k", [])
self.nghttp_post_and_verify(env, "data-10k", [])
self.nghttp_post_and_verify(env, "data-100k", [])
self.nghttp_post_and_verify(env, "data-1m", [])
- def test_500_21(self, env):
+ def test_h2_500_21(self, env):
self.nghttp_post_and_verify(env, "data-1k", ["--no-content-length"])
self.nghttp_post_and_verify(env, "data-10k", ["--no-content-length"])
self.nghttp_post_and_verify(env, "data-100k", ["--no-content-length"])
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(env.test_src(fpath), mode='rb') as file:
+ with open(env.local_src(fpath), mode='rb') as file:
src = file.read()
assert src == r2.response["body"]
- def test_500_22(self, env):
+ def test_h2_500_22(self, env):
self.nghttp_upload_and_verify(env, "data-1k", [])
self.nghttp_upload_and_verify(env, "data-10k", [])
self.nghttp_upload_and_verify(env, "data-100k", [])
self.nghttp_upload_and_verify(env, "data-1m", [])
- def test_500_23(self, env):
+ def test_h2_500_23(self, env):
self.nghttp_upload_and_verify(env, "data-1k", ["--no-content-length"])
self.nghttp_upload_and_verify(env, "data-10k", ["--no-content-length"])
self.nghttp_upload_and_verify(env, "data-100k", ["--no-content-length"])
assert 200 <= r.response["status"] < 300
assert r.response["header"]["location"]
- def test_500_24(self, env):
+ def test_h2_500_24(self, env):
for i in range(100):
self.nghttp_upload_stat(env, "data-1k", ["--no-content-length"])
import pytest
-from h2_conf import HttpdConf
+from .env import H2Conf
class TestStore:
@pytest.fixture(autouse=True, scope='class')
def _class_scope(self, env):
env.setup_data_1k_1m()
- conf = HttpdConf(env)
+ conf = H2Conf(env)
conf.add_vhost_cgi(h2proxy_self=True)
if env.verbosity > 1:
conf.add("LogLevel proxy:trace2 proxy_http2:trace2")
conf.install()
assert env.apache_restart() == 0
- def test_600_01(self, env):
+ def test_h2_600_01(self, env):
url = env.mkurl("https", "cgi", "/h2proxy/hello.py")
r = env.curl_get(url, 5)
assert r.response["status"] == 200
import pytest
-from h2_conf import HttpdConf
+from .env import H2Conf
class TestStore:
@pytest.fixture(autouse=True, scope='class')
def _class_scope(self, env):
- HttpdConf(env).add_vhost_cgi().add_vhost_test1().install()
+ H2Conf(env).add_vhost_cgi().add_vhost_test1().install()
assert env.apache_restart() == 0
def check_h2load_ok(self, env, r, n):
@pytest.mark.parametrize("start", [
1000, 80000
])
- def test_700_10(self, env, start):
+ def test_h2_700_10(self, env, start):
text = "X"
chunk = 32
for n in range(0, 5):
@pytest.mark.parametrize("conns", [
1, 2, 16, 32
])
- def test_700_11(self, env, conns):
+ def test_h2_700_11(self, env, conns):
text = "X"
start = 1200
chunk = 64
import pytest
import os
-from h2_conf import HttpdConf
+from .env import H2Conf
class TestStore:
@pytest.fixture(autouse=True, scope='class')
def _class_scope(self, env):
env.setup_data_1k_1m()
- HttpdConf(env).add_vhost_test1().install()
+ H2Conf(env).add_vhost_test1().install()
assert env.apache_restart() == 0
def check_h2load_ok(self, env, r, n):
assert 0 == r.results["h2load"]["status"]["5xx"]
# test POST on static file, slurped in by server
- def test_710_00(self, env):
+ def test_h2_710_00(self, env):
url = env.mkurl("https", "test1", "/index.html")
n = 10
m = 1
r = env.run(args)
self.check_h2load_ok(env, r, n)
- def test_710_01(self, env):
+ def test_h2_710_01(self, env):
url = env.mkurl("https", "test1", "/index.html")
n = 1000
m = 100
r = env.run(args)
self.check_h2load_ok(env, r, n)
- def test_710_02(self, env):
+ def test_h2_710_02(self, env):
url = env.mkurl("https", "test1", "/index.html")
n = 100
m = 50
import pytest
import os
-from h2_conf import HttpdConf
+from .env import H2Conf
class TestStore:
@pytest.fixture(autouse=True, scope='class')
def _class_scope(self, env):
env.setup_data_1k_1m()
- HttpdConf(env).add_vhost_cgi(proxy_self=True, h2proxy_self=True).install()
+ H2Conf(env).add_vhost_cgi(proxy_self=True, h2proxy_self=True).install()
assert env.apache_restart() == 0
def check_h2load_ok(self, env, r, n):
assert 0 == r.results["h2load"]["status"]["5xx"]
# test POST on cgi, where input is read
- def test_711_10(self, env):
+ def test_h2_711_10(self, env):
url = env.mkurl("https", "test1", "/echo.py")
n = 100
m = 5
self.check_h2load_ok(env, r, n)
# test POST on cgi via http/1.1 proxy, where input is read
- def test_711_11(self, env):
+ def test_h2_711_11(self, env):
url = env.mkurl("https", "test1", "/proxy/echo.py")
n = 100
m = 5
self.check_h2load_ok(env, r, n)
# test POST on cgi via h2proxy, where input is read
- def test_711_12(self, env):
+ def test_h2_711_12(self, env):
url = env.mkurl("https", "test1", "/h2proxy/echo.py")
n = 100
m = 5
import pytest
-from h2_conf import HttpdConf
-from h2_curl import CurlPiper
+from .env import H2Conf
+from pyhttpd.curl import CurlPiper
class TestBuffering:
@pytest.fixture(autouse=True, scope='class')
def _class_scope(self, env):
env.setup_data_1k_1m()
- conf = HttpdConf(env)
+ conf = H2Conf(env)
conf.add_vhost_cgi(h2proxy_self=True).install()
assert env.apache_restart() == 0
@pytest.mark.skip(reason="this test shows unreliable jitter")
- def test_712_01(self, env):
+ def test_h2_712_01(self, env):
# test gRPC like requests that do not end, but give answers, see #207
#
# this test works like this:
piper.stutter_check(chunks, stutter)
@pytest.mark.skipif(True, reason="new feature in upcoming http2")
- def test_712_02(self, env):
+ def test_h2_712_02(self, env):
# same as 712_01 but via mod_proxy_http2
#
url = env.mkurl("https", "cgi", "/h2proxy/h2test/echo")
piper.stutter_check(chunks, stutter)
@pytest.mark.skipif(True, reason="new feature in upcoming http2")
- def test_712_03(self, env):
+ def test_h2_712_03(self, env):
# same as 712_02 but with smaller chunks
#
url = env.mkurl("https", "cgi", "/h2proxy/h2test/echo")
creds = self._store.load_credentials(name=spec.name, key_type=key_type, single_file=spec.single_file) \
if self._store else None
if creds is None:
- creds = H2TestCA.create_credentials(spec=spec, issuer=self, key_type=key_type,
- valid_from=spec.valid_from, valid_to=spec.valid_to)
+ creds = HttpdTestCA.create_credentials(spec=spec, issuer=self, key_type=key_type,
+ valid_from=spec.valid_from, valid_to=spec.valid_to)
if self._store:
self._store.save(creds, single_file=spec.single_file)
return None
-class H2TestCA:
+class HttpdTestCA:
@classmethod
def create_root(cls, name: str, store_dir: str, key_type: str = "rsa2048") -> Credentials:
store = CertStore(fpath=store_dir)
creds = store.load_credentials(name="ca", key_type=key_type)
if creds is None:
- creds = H2TestCA._make_ca_credentials(name=name, key_type=key_type)
+ creds = HttpdTestCA._make_ca_credentials(name=name, key_type=key_type)
store.save(creds, name="ca")
creds.set_store(store)
return creds
:returns: the certificate and private key PEM file paths
"""
if spec.domains and len(spec.domains):
- creds = H2TestCA._make_server_credentials(name=spec.name, domains=spec.domains,
- issuer=issuer, valid_from=valid_from,
- valid_to=valid_to, key_type=key_type)
+ creds = HttpdTestCA._make_server_credentials(name=spec.name, domains=spec.domains,
+ issuer=issuer, valid_from=valid_from,
+ valid_to=valid_to, key_type=key_type)
elif spec.client:
- creds = H2TestCA._make_client_credentials(name=spec.name, issuer=issuer,
- email=spec.email, valid_from=valid_from,
- valid_to=valid_to, key_type=key_type)
+ creds = HttpdTestCA._make_client_credentials(name=spec.name, issuer=issuer,
+ email=spec.email, valid_from=valid_from,
+ valid_to=valid_to, key_type=key_type)
elif spec.name:
- creds = H2TestCA._make_ca_credentials(name=spec.name, issuer=issuer,
- valid_from=valid_from, valid_to=valid_to,
- key_type=key_type)
+ creds = HttpdTestCA._make_ca_credentials(name=spec.name, issuer=issuer,
+ valid_from=valid_from, valid_to=valid_to,
+ key_type=key_type)
else:
raise Exception(f"unrecognized certificate specification: {spec}")
return creds
else:
issuer_subject = None
issuer_key = pkey
- subject = H2TestCA._make_x509_name(org_name=name, parent=issuer.subject if issuer else None)
- csr = H2TestCA._make_csr(subject=subject,
- issuer_subject=issuer_subject, pkey=pkey,
- valid_from_delta=valid_from, valid_until_delta=valid_to)
- csr = H2TestCA._add_ca_usages(csr)
+ subject = HttpdTestCA._make_x509_name(org_name=name, parent=issuer.subject if issuer else None)
+ csr = HttpdTestCA._make_csr(subject=subject,
+ issuer_subject=issuer_subject, pkey=pkey,
+ valid_from_delta=valid_from, valid_until_delta=valid_to)
+ csr = HttpdTestCA._add_ca_usages(csr)
cert = csr.sign(private_key=issuer_key,
algorithm=hashes.SHA256(),
backend=default_backend())
) -> Credentials:
name = name
pkey = _private_key(key_type=key_type)
- subject = H2TestCA._make_x509_name(common_name=name, parent=issuer.subject)
- csr = H2TestCA._make_csr(subject=subject,
- issuer_subject=issuer.certificate.subject, pkey=pkey,
- valid_from_delta=valid_from, valid_until_delta=valid_to)
- csr = H2TestCA._add_leaf_usages(csr, domains=domains, issuer=issuer)
+ subject = HttpdTestCA._make_x509_name(common_name=name, parent=issuer.subject)
+ csr = HttpdTestCA._make_csr(subject=subject,
+ issuer_subject=issuer.certificate.subject, pkey=pkey,
+ valid_from_delta=valid_from, valid_until_delta=valid_to)
+ csr = HttpdTestCA._add_leaf_usages(csr, domains=domains, issuer=issuer)
cert = csr.sign(private_key=issuer.private_key,
algorithm=hashes.SHA256(),
backend=default_backend())
valid_to: timedelta = timedelta(days=89),
) -> Credentials:
pkey = _private_key(key_type=key_type)
- subject = H2TestCA._make_x509_name(common_name=name, parent=issuer.subject)
- csr = H2TestCA._make_csr(subject=subject,
- issuer_subject=issuer.certificate.subject, pkey=pkey,
- valid_from_delta=valid_from, valid_until_delta=valid_to)
- csr = H2TestCA._add_client_usages(csr, issuer=issuer, rfc82name=email)
+ subject = HttpdTestCA._make_x509_name(common_name=name, parent=issuer.subject)
+ csr = HttpdTestCA._make_csr(subject=subject,
+ issuer_subject=issuer.certificate.subject, pkey=pkey,
+ valid_from_delta=valid_from, valid_until_delta=valid_to)
+ csr = HttpdTestCA._add_client_usages(csr, issuer=issuer, rfc82name=email)
cert = csr.sign(private_key=issuer.private_key,
algorithm=hashes.SHA256(),
backend=default_backend())
-import os
+from pyhttpd.env import HttpdTestEnv
class HttpdConf(object):
- def __init__(self, env, path=None):
+ def __init__(self, env: HttpdTestEnv, path=None):
self.env = env
self._lines = []
self._has_ssl_vhost = False
""")
return self
+ def add_proxy_setup(self):
+ self.add("ProxyStatus on")
+ self.add("ProxyTimeout 5")
+ self.add("SSLProxyEngine on")
+ self.add("SSLProxyVerify none")
+ return self
+
def add_vhost_test1(self, proxy_self=False, h2proxy_self=False, extras=None):
domain = f"test1.{self.env.http_tld}"
if extras and 'base' in extras:
self.add_proxies("test1", proxy_self, h2proxy_self)
self.end_vhost()
return self
-
+
def add_vhost_test2(self, extras=None):
domain = f"test2.{self.env.http_tld}"
if extras and 'base' in extras:
self.add(" LogLevel proxy:info")
self.add(" LogLevel proxy_http:info")
return self
-
- def add_vhost_noh2(self):
- self.start_vhost(self.env.https_port, "noh2", aliases=["noh2-alias"], doc_root="htdocs/noh2", with_ssl=True)
- self.add(f"""
- Protocols http/1.1
- SSLOptions +StdEnvVars""")
- self.end_vhost()
- self.start_vhost(self.env.http_port, "noh2", aliases=["noh2-alias"], doc_root="htdocs/noh2", with_ssl=False)
- self.add(" Protocols http/1.1")
- self.add(" SSLOptions +StdEnvVars")
- self.end_vhost()
- return self
-
- def add_proxy_setup(self):
- self.add("ProxyStatus on")
- self.add("ProxyTimeout 5")
- self.add("SSLProxyEngine on")
- self.add("SSLProxyVerify none")
- return self
http_port = 40001
https_port = 40002
http_tld = tests.httpd.apache.org
-test_dir = @abs_srcdir@
-test_src_dir = @abs_srcdir@
-server_dir = @abs_srcdir@/gen/apache
-gen_dir = @abs_srcdir@/gen
+test_dir = @abs_srcdir@/..
+server_dir = @abs_srcdir@/../gen/apache
+gen_dir = @abs_srcdir@/../gen
import time
from threading import Thread
-from h2_env import H2TestEnv
+from .env import HttpdTestEnv
class CurlPiper:
- def __init__(self, env: H2TestEnv, url: str):
+ def __init__(self, env: HttpdTestEnv, url: str):
self.env = env
self.url = url
self.proc = None
from configparser import ConfigParser, ExtendedInterpolation
from urllib.parse import urlparse
-from h2_certs import Credentials
-from h2_nghttp import Nghttp
-from h2_result import ExecResult
+from .certs import Credentials, HttpdTestCA, CertificateSpec
+from .nghttp import Nghttp
+from .result import ExecResult
log = logging.getLogger(__name__)
pass
-class H2TestSetup:
+class HttpdTestSetup:
# the modules we want to load
MODULES = [
"proxy_hcheck",
]
- def __init__(self, env: 'H2TestEnv'):
+ def __init__(self, env: 'HttpdTestEnv'):
self.env = env
- def make(self):
+ def make(self, modules: List[str] = None, add_modules: List[str] = None):
self._make_dirs()
self._make_conf()
+ mod_names = modules.copy() if modules else self.MODULES.copy()
+ if add_modules:
+ mod_names.extend(add_modules)
+ self._make_modules_conf(modules=mod_names)
self._make_htdocs()
- self._make_h2test()
- self._make_modules_conf()
def _make_dirs(self):
if os.path.exists(self.env.gen_dir):
os.makedirs(self.env.server_logs_dir)
def _make_conf(self):
- conf_src_dir = os.path.join(self.env.test_dir, 'conf')
+ our_dir = os.path.dirname(inspect.getfile(Dummy))
+ conf_src_dir = os.path.join(our_dir, 'conf')
conf_dest_dir = os.path.join(self.env.server_dir, 'conf')
if not os.path.exists(conf_dest_dir):
os.makedirs(conf_dest_dir)
def _make_template(self, src, dest):
var_map = dict()
- for name, value in self.env.__class__.__dict__.items():
+ for name, value in HttpdTestEnv.__dict__.items():
if isinstance(value, property):
var_map[name] = value.fget(self.env)
t = Template(''.join(open(src).readlines()))
with open(dest, 'w') as fd:
fd.write(t.substitute(var_map))
+ def _make_modules_conf(self, modules: List[str]):
+ modules_conf = os.path.join(self.env.server_dir, 'conf/modules.conf')
+ with open(modules_conf, 'w') as fd:
+ # issue load directives for all modules we want that are shared
+ for m in modules:
+ mod_path = os.path.join(self.env.libexec_dir, f"mod_{m}.so")
+ if os.path.isfile(mod_path):
+ fd.write(f"LoadModule {m}_module \"{mod_path}\"\n")
+
def _make_htdocs(self):
+ our_dir = os.path.dirname(inspect.getfile(Dummy))
if not os.path.exists(self.env.server_docs_dir):
os.makedirs(self.env.server_docs_dir)
- shutil.copytree(os.path.join(self.env.test_dir, 'htdocs'),
+ shutil.copytree(os.path.join(our_dir, 'htdocs'),
os.path.join(self.env.server_dir, 'htdocs'),
dirs_exist_ok=True)
cgi_dir = os.path.join(self.env.server_dir, 'htdocs/cgi')
st = os.stat(cgi_file)
os.chmod(cgi_file, st.st_mode | stat.S_IEXEC)
- def _make_h2test(self):
- p = subprocess.run([self.env.apxs, '-c', 'mod_h2test.c'],
- capture_output=True,
- cwd=os.path.join(self.env.test_dir, 'mod_h2test'))
- rv = p.returncode
- if rv != 0:
- log.error(f"compiling md_h2test failed: {p.stderr}")
- raise Exception(f"compiling md_h2test failed: {p.stderr}")
- def _make_modules_conf(self):
- modules_conf = os.path.join(self.env.server_dir, 'conf/modules.conf')
- with open(modules_conf, 'w') as fd:
- # issue load directives for all modules we want that are shared
- for m in self.MODULES:
- mod_path = os.path.join(self.env.libexec_dir, f"mod_{m}.so")
- if os.path.isfile(mod_path):
- fd.write(f"LoadModule {m}_module \"{mod_path}\"\n")
- for m in ["http2", "proxy_http2"]:
- fd.write(f"LoadModule {m}_module \"{self.env.libexec_dir}/mod_{m}.so\"\n")
- # load our test module which is not installed
- fd.write(f"LoadModule h2test_module \"{self.env.test_dir}/mod_h2test/.libs/mod_h2test.so\"\n")
-
-
-class H2TestEnv:
+class HttpdTestEnv:
- def __init__(self, pytestconfig=None, setup_dirs=True):
- our_dir = os.path.dirname(inspect.getfile(Dummy))
+ def __init__(self, pytestconfig=None,
+ local_dir=None, add_base_conf: str = None,
+ interesting_modules: List[str] = None):
+ self._our_dir = os.path.dirname(inspect.getfile(Dummy))
+ self._local_dir = local_dir if local_dir else self._our_dir
self.config = ConfigParser(interpolation=ExtendedInterpolation())
- self.config.read(os.path.join(our_dir, 'config.ini'))
+ self.config.read(os.path.join(self._our_dir, 'config.ini'))
self._apxs = self.config.get('global', 'apxs')
self._prefix = self.config.get('global', 'prefix')
self._curl = self.config.get('global', 'curl_bin')
self._nghttp = self.config.get('global', 'nghttp')
self._h2load = self.config.get('global', 'h2load')
- self._ca = None
self._http_port = int(self.config.get('test', 'http_port'))
self._https_port = int(self.config.get('test', 'https_port'))
self._http_tld = self.config.get('test', 'http_tld')
self._test_dir = self.config.get('test', 'test_dir')
- self._test_src_dir = self.config.get('test', 'test_src_dir')
self._gen_dir = self.config.get('test', 'gen_dir')
self._server_dir = os.path.join(self._gen_dir, 'apache')
self._server_conf_dir = os.path.join(self._server_dir, "conf")
self._server_error_log = os.path.join(self._server_logs_dir, "error_log")
self._dso_modules = self.config.get('global', 'dso_modules').split(' ')
- self._domains = [
- f"test1.{self._http_tld}",
- f"test2.{self._http_tld}",
- f"test3.{self._http_tld}",
- f"cgi.{self._http_tld}",
- f"push.{self._http_tld}",
- f"hints.{self._http_tld}",
- f"ssl.{self._http_tld}",
- f"pad0.{self._http_tld}",
- f"pad1.{self._http_tld}",
- f"pad2.{self._http_tld}",
- f"pad3.{self._http_tld}",
- f"pad8.{self._http_tld}",
- ]
- self._domains_noh2 = [
- f"noh2.{self._http_tld}",
- ]
self._mpm_type = os.environ['MPM'] if 'MPM' in os.environ else 'event'
self._httpd_addr = "127.0.0.1"
self._test_conf = os.path.join(self._server_conf_dir, "test.conf")
self._httpd_base_conf = f"""
LoadModule mpm_{self.mpm_type}_module \"{self.libexec_dir}/mod_mpm_{self.mpm_type}.so\"
- H2MinWorkers 1
- H2MaxWorkers 64
- SSLSessionCache "shmcb:ssl_gcache_data(32000)"
+ <IfModule mod_ssl.c>
+ SSLSessionCache "shmcb:ssl_gcache_data(32000)"
+ </IfModule>
"""
+ if add_base_conf:
+ self._httpd_base_conf += f"\n{add_base_conf}"
+
self._verbosity = pytestconfig.option.verbose if pytestconfig is not None else 0
if self._verbosity >= 2:
+ log_level = "trace2"
self._httpd_base_conf += f"""
- LogLevel http2:trace2 proxy_http2:info h2test:trace2
LogLevel core:trace5 mpm_{self.mpm_type}:trace5
"""
elif self._verbosity >= 1:
- self._httpd_base_conf += "LogLevel http2:debug proxy_http2:debug h2test:debug"
+ log_level = "debug"
else:
- self._httpd_base_conf += "LogLevel http2:info proxy_http2:info"
+ log_level = "info"
+ if interesting_modules:
+ self._httpd_base_conf += "\nLogLevel"
+ for name in interesting_modules:
+ self._httpd_base_conf += f" {name}:{log_level}"
+ self._httpd_base_conf += "\n"
+
+ self._ca = None
+ self._cert_specs = [CertificateSpec(domains=[
+ f"test1.{self._http_tld}",
+ f"test2.{self._http_tld}",
+ f"test3.{self._http_tld}",
+ f"cgi.{self._http_tld}",
+ ], key_type='rsa4096')]
self._verify_certs = False
- if setup_dirs:
- self._setup = H2TestSetup(env=self)
- self._setup.make()
@property
def apxs(self) -> str:
def http_tld(self) -> str:
return self._http_tld
- @property
- def domain_test1(self) -> str:
- return self._domains[0]
-
- @property
- def domains(self) -> List[str]:
- return self._domains
-
- @property
- def domains_noh2(self) -> List[str]:
- return self._domains_noh2
-
@property
def http_base_url(self) -> str:
return self._http_base
return self._gen_dir
@property
- def test_dir(self) -> str:
- return self._test_dir
+ def local_dir(self) -> str:
+ return self._local_dir
@property
- def test_src_dir(self) -> str:
- return self._test_src_dir
+ def test_dir(self) -> str:
+ return self._test_dir
@property
def server_dir(self) -> str:
def httpd_base_conf(self) -> str:
return self._httpd_base_conf
+ def local_src(self, path):
+ return os.path.join(self.local_dir, path)
+
+ def htdocs_src(self, path):
+ return os.path.join(self._our_dir, 'htdocs', path)
+
@property
def h2load(self) -> str:
return self._h2load
def ca(self) -> Credentials:
return self._ca
- def set_ca(self, ca: Credentials):
- self._ca = ca
+ def add_cert_specs(self, specs: List[CertificateSpec]):
+ self._cert_specs.extend(specs)
+
+ def issue_certs(self):
+ if self._ca is None:
+ self._ca = HttpdTestCA.create_root(name=self.http_tld,
+ store_dir=os.path.join(self.server_dir, 'ca'), key_type="rsa4096")
+ self._ca.issue_certs(self._cert_specs)
def get_credentials_for_name(self, dns_name) -> List['Credentials']:
- for domains in [self._domains, self._domains_noh2]:
- if dns_name in domains:
- return self.ca.get_credentials_for_name(domains[0])
+ for spec in self._cert_specs:
+ if dns_name in spec.domains:
+ return self.ca.get_credentials_for_name(spec.domains[0])
return []
def has_h2load(self):
if not os.path.exists(path):
return os.makedirs(path)
- def test_src(self, path):
- return os.path.join(self._test_src_dir, path)
-
def run(self, args) -> ExecResult:
log.debug("execute: %s", " ".join(args))
start = datetime.now()
}
run.add_results({"h2load": stats})
return run
-
- def setup_data_1k_1m(self):
- s100 = "012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678\n"
- with open(os.path.join(self.gen_dir, "data-1k"), 'w') as f:
- for i in range(10):
- f.write(s100)
- with open(os.path.join(self.gen_dir, "data-10k"), 'w') as f:
- for i in range(100):
- f.write(s100)
- with open(os.path.join(self.gen_dir, "data-100k"), 'w') as f:
- for i in range(1000):
- f.write(s100)
- with open(os.path.join(self.gen_dir, "data-1m"), 'w') as f:
- for i in range(10000):
- f.write(s100)
from urllib.parse import urlparse
-from h2_result import ExecResult
+from .result import ExecResult
def _get_path(x):
projects / thirdparty / apache / httpd.git / commitdiff
