backup/warcprox
1
0
Fork 0
mirror of https://github.com/internetarchive/warcprox.git synced 2025-01-18 13:22:09 +01:00
Code Issues Projects Releases Activity

set up fixtures once for all tests

Browse Source
This commit is contained in:
Noah Levitt 2015-08-20 22:58:30 +00:00
parent 3d90b9c2e9
commit 788bc69f47
2 changed files with 11 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
warcprox/tests/conftest.py
View File

@ -5,7 +5,7 @@ def pytest_addoption(parser):
parser.addoption('--rethinkdb-servers', dest='rethinkdb_servers', parser.addoption('--rethinkdb-servers', dest='rethinkdb_servers',
help='rethink db servers for dedup, e.g. db0.foo.org,db0.foo.org:38015,db1.foo.org') help='rethink db servers for dedup, e.g. db0.foo.org,db0.foo.org:38015,db1.foo.org')
@pytest.fixture @pytest.fixture(scope="module")
def rethinkdb_servers(request): def rethinkdb_servers(request):
return request.config.getoption("--rethinkdb-servers") return request.config.getoption("--rethinkdb-servers")

20
warcprox/tests/test_warcprox.py
View File

@ -59,7 +59,7 @@ class _TestHttpRequestHandler(http_server.BaseHTTPRequestHandler):
self.connection.sendall(headers) self.connection.sendall(headers)
self.connection.sendall(payload) self.connection.sendall(payload)
@pytest.fixture() @pytest.fixture(scope="module")
def cert(request): def cert(request):
f = tempfile.NamedTemporaryFile(prefix='warcprox-test-https-', suffix='.pem', delete=False) f = tempfile.NamedTemporaryFile(prefix='warcprox-test-https-', suffix='.pem', delete=False)
@ -92,7 +92,7 @@ def cert(request):
finally: finally:
f.close() f.close()
@pytest.fixture() @pytest.fixture(scope="module")
def http_daemon(request): def http_daemon(request):
http_daemon = http_server.HTTPServer(('localhost', 0), http_daemon = http_server.HTTPServer(('localhost', 0),
RequestHandlerClass=_TestHttpRequestHandler) RequestHandlerClass=_TestHttpRequestHandler)
@ -110,7 +110,7 @@ def http_daemon(request):
return http_daemon return http_daemon
@pytest.fixture() @pytest.fixture(scope="module")
def https_daemon(request, cert): def https_daemon(request, cert):
# http://www.piware.de/2011/01/creating-an-https-server-in-python/ # http://www.piware.de/2011/01/creating-an-https-server-in-python/
https_daemon = http_server.HTTPServer(('localhost', 0), https_daemon = http_server.HTTPServer(('localhost', 0),
@ -131,7 +131,7 @@ def https_daemon(request, cert):
return https_daemon return https_daemon
@pytest.fixture() @pytest.fixture(scope="module")
def dedup_db(request, rethinkdb_servers): def dedup_db(request, rethinkdb_servers):
if rethinkdb_servers: if rethinkdb_servers:
servers = rethinkdb_servers.split(",") servers = rethinkdb_servers.split(",")
@ -156,7 +156,7 @@ def dedup_db(request, rethinkdb_servers):
return ddb return ddb
@pytest.fixture() @pytest.fixture(scope="module")
def warcprox_(request, dedup_db): def warcprox_(request, dedup_db):
f = tempfile.NamedTemporaryFile(prefix='warcprox-test-ca-', suffix='.pem', delete=True) f = tempfile.NamedTemporaryFile(prefix='warcprox-test-ca-', suffix='.pem', delete=True)
f.close() # delete it, or CertificateAuthority will try to read it f.close() # delete it, or CertificateAuthority will try to read it
@ -212,12 +212,12 @@ def warcprox_(request, dedup_db):
return warcprox_ return warcprox_
@pytest.fixture() @pytest.fixture(scope="module")
def archiving_proxies(warcprox_): def archiving_proxies(warcprox_):
archiving_proxy = 'http://localhost:{}'.format(warcprox_.proxy.server_port) archiving_proxy = 'http://localhost:{}'.format(warcprox_.proxy.server_port)
return {'http':archiving_proxy, 'https':archiving_proxy} return {'http':archiving_proxy, 'https':archiving_proxy}
@pytest.fixture() @pytest.fixture(scope="module")
def playback_proxies(warcprox_): def playback_proxies(warcprox_):
playback_proxy = 'http://localhost:{}'.format(warcprox_.playback_proxy.server_port) playback_proxy = 'http://localhost:{}'.format(warcprox_.playback_proxy.server_port)
return {'http':playback_proxy, 'https':playback_proxy} return {'http':playback_proxy, 'https':playback_proxy}
@ -417,15 +417,15 @@ def test_dedup_https(https_daemon, warcprox_, archiving_proxies, playback_proxie
# XXX how to check dedup was used? # XXX how to check dedup was used?
def test_limits(http_daemon, archiving_proxies): def test_limits(http_daemon, archiving_proxies):
url = 'http://localhost:{}/a/b'.format(http_daemon.server_port) url = 'http://localhost:{}/i/j'.format(http_daemon.server_port)
request_meta = {"stats":{"buckets":["job1"]},"limits":{"job1.total.urls":10}} request_meta = {"stats":{"buckets":["job1"]},"limits":{"job1.total.urls":10}}
headers = {"Warcprox-Meta": json.dumps(request_meta)} headers = {"Warcprox-Meta": json.dumps(request_meta)}
for i in range(10): for i in range(10):
response = requests.get(url, proxies=archiving_proxies, headers=headers, stream=True) response = requests.get(url, proxies=archiving_proxies, headers=headers, stream=True)
assert response.status_code == 200 assert response.status_code == 200
assert response.headers['warcprox-test-header'] == 'a!' assert response.headers['warcprox-test-header'] == 'i!'
assert response.content == b'I am the warcprox test payload! bbbbbbbbbb!\n' assert response.content == b'I am the warcprox test payload! jjjjjjjjjj!\n'
# XXX give warc writer thread a chance to update stats # XXX give warc writer thread a chance to update stats
time.sleep(2.0) time.sleep(2.0)