|
|
|
@ -0,0 +1,159 @@
|
|
|
|
|
import pytest
|
|
|
|
|
import requests
|
|
|
|
|
import time
|
|
|
|
|
import subprocess
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@pytest.fixture
|
|
|
|
|
def run_servefile():
|
|
|
|
|
instances = []
|
|
|
|
|
|
|
|
|
|
def _run_servefile(args, **kwargs):
|
|
|
|
|
if not isinstance(args, list):
|
|
|
|
|
args = [args]
|
|
|
|
|
print("running with args", args)
|
|
|
|
|
p = subprocess.Popen(['servefile'] + args, **kwargs)
|
|
|
|
|
time.sleep(kwargs.get('timeout', 0.1))
|
|
|
|
|
instances.append(p)
|
|
|
|
|
|
|
|
|
|
return p
|
|
|
|
|
|
|
|
|
|
yield _run_servefile
|
|
|
|
|
|
|
|
|
|
for instance in instances:
|
|
|
|
|
try:
|
|
|
|
|
instance.terminate()
|
|
|
|
|
except OSError:
|
|
|
|
|
pass
|
|
|
|
|
instance.wait()
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@pytest.fixture
|
|
|
|
|
def datadir(tmp_path):
|
|
|
|
|
def _datadir(data, path=None):
|
|
|
|
|
path = path or tmp_path
|
|
|
|
|
for k, v in data.items():
|
|
|
|
|
if isinstance(v, dict):
|
|
|
|
|
new_path = path / k
|
|
|
|
|
new_path.mkdir()
|
|
|
|
|
_datadir(v, new_path)
|
|
|
|
|
else:
|
|
|
|
|
(path / k).write_text(v.decode())
|
|
|
|
|
|
|
|
|
|
return path
|
|
|
|
|
return _datadir
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@pytest.fixture(scope='module')
|
|
|
|
|
def servefile_simple(run_servefile, datadir):
|
|
|
|
|
data = "NOOT NOOT"
|
|
|
|
|
p = datadir({'testfile': data}) / 'testfile'
|
|
|
|
|
|
|
|
|
|
run_servefile(str(p))
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def download_ok(r, expected_data, expected_code=200):
|
|
|
|
|
assert r.status_code == expected_code
|
|
|
|
|
assert r.text == expected_data
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def check_download(expected_data, path='/', host='localhost', port=8080, protocol='http', status_code=200):
|
|
|
|
|
url = '{}://{}:{}{}'.format(protocol, host, port, path)
|
|
|
|
|
r = requests.get(url, allow_redirects=False)
|
|
|
|
|
assert r.status_code == 200
|
|
|
|
|
assert r.text == expected_data
|
|
|
|
|
assert r.headers.get('Content-Type') == 'application/octet-stream'
|
|
|
|
|
# assert r.headers.get('Content-Disposition') == 'attachment; filename=""'
|
|
|
|
|
assert r.headers.get('Content-Transfer-Encoding') == 'binary'
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_version(run_servefile):
|
|
|
|
|
s = run_servefile('--version', stderr=subprocess.PIPE)
|
|
|
|
|
s.wait()
|
|
|
|
|
version = s.stderr.readline().strip()
|
|
|
|
|
assert version == 'servefile 0.4.5-unreleased'
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_correct_headers(run_servefile, datadir):
|
|
|
|
|
data = "NOOT NOOT"
|
|
|
|
|
p = datadir({'testfile': data}) / 'testfile'
|
|
|
|
|
run_servefile(str(p))
|
|
|
|
|
r = requests.get("http://127.0.0.1:8080")
|
|
|
|
|
assert r.status_code == 200
|
|
|
|
|
assert r.headers.get('Content-Type') == 'application/octet-stream'
|
|
|
|
|
assert r.headers.get('Content-Disposition') == 'attachment; filename="testfile"'
|
|
|
|
|
assert r.headers.get('Content-Transfer-Encoding') == 'binary'
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_redirect_and_download(run_servefile, datadir):
|
|
|
|
|
data = "NOOT NOOT"
|
|
|
|
|
p = datadir({'testfile': data}) / 'testfile'
|
|
|
|
|
|
|
|
|
|
run_servefile(str(p))
|
|
|
|
|
|
|
|
|
|
r = requests.get("http://127.0.0.1:8080", allow_redirects=False)
|
|
|
|
|
assert r.status_code == 302
|
|
|
|
|
assert r.headers.get('Location') == '/testfile'
|
|
|
|
|
|
|
|
|
|
r = requests.get("http://127.0.0.1:8080")
|
|
|
|
|
download_ok(r, data)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_specify_port(run_servefile, datadir):
|
|
|
|
|
data = "NOOT NOOT"
|
|
|
|
|
p = datadir({'testfile': data}) / 'testfile'
|
|
|
|
|
run_servefile([str(p), '-p', '8081'])
|
|
|
|
|
r = requests.get("http://127.0.0.1:8081")
|
|
|
|
|
download_ok(r, data)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_big_download(run_servefile, datadir):
|
|
|
|
|
# test with about 10 mb of data
|
|
|
|
|
data = "x" * (10 * 1024 ** 2)
|
|
|
|
|
p = datadir({'testfile': data}) / 'testfile'
|
|
|
|
|
|
|
|
|
|
run_servefile(str(p))
|
|
|
|
|
r = requests.get("http://127.0.0.1:8080")
|
|
|
|
|
download_ok(r, data)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_authentication(run_servefile, datadir):
|
|
|
|
|
data = "NOOT NOOT"
|
|
|
|
|
p = datadir({'testfile': data}) / 'testfile'
|
|
|
|
|
|
|
|
|
|
run_servefile([str(p), '-a', 'user:password'])
|
|
|
|
|
for auth in [('foo', 'bar'), ('user', 'wrong'), ('unknown', 'password')]:
|
|
|
|
|
r = requests.get('http://127.0.0.1:8080', auth=auth)
|
|
|
|
|
assert '401 - Unauthorized' in r.text
|
|
|
|
|
assert r.status_code == 401
|
|
|
|
|
|
|
|
|
|
r = requests.get("http://127.0.0.1:8080", auth=('user', 'password'))
|
|
|
|
|
download_ok(r, data)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_serve_directory(run_servefile, datadir):
|
|
|
|
|
d = {
|
|
|
|
|
'foo': {'1': 'cat', '2': 'kitteh', '3': 'wheee'},
|
|
|
|
|
'bar': {'thisisaverylongfilenamefortestingthatthisstillworksproperly': 'jup!'},
|
|
|
|
|
'noot': 'still data in here',
|
|
|
|
|
'bigfile': 'x' * (10 * 1024 ** 2),
|
|
|
|
|
}
|
|
|
|
|
p = datadir(d)
|
|
|
|
|
run_servefile([str(p), '-l'])
|
|
|
|
|
|
|
|
|
|
# check if all files are in directory listing
|
|
|
|
|
# (could be made more sophisticated with beautifulsoup)
|
|
|
|
|
for path in '/', '/../':
|
|
|
|
|
r = requests.get("http://127.0.0.1:8080" + path)
|
|
|
|
|
for k in d:
|
|
|
|
|
assert k in r.text
|
|
|
|
|
|
|
|
|
|
# download
|
|
|
|
|
check_download('cat', '/foo/1')
|
|
|
|
|
check_download('kitteh', '/foo/2')
|
|
|
|
|
check_download('jup!', '/bar/thisisaverylongfilenamefortestingthatthisstillworksproperly')
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
class Foo:
|
|
|
|
|
def noot(servefile_simple):
|
|
|
|
|
|