@ -9,7 +9,6 @@ import sys
import tarfile
import time
import urllib3
from requests . exceptions import ConnectionError
# crudly written to learn more about pytest and to have a base for refactoring
@ -59,6 +58,7 @@ def run_servefile():
print ( " running {} with args {} " . format ( " , " . join ( servefile_path ) , args ) )
p = subprocess . Popen ( [ sys . executable ] + servefile_path + args , * * kwargs )
time . sleep ( kwargs . get ( ' timeout ' , 0.3 ) )
instances . append ( p )
return p
@ -91,7 +91,7 @@ def datadir(tmp_path):
return _datadir
def make_request ( path = ' / ' , host = ' localhost ' , port = SERVEFILE_DEFAULT_PORT , method = ' get ' , protocol = ' http ' ,
def make_request ( path = ' / ' , host = ' localhost ' , port = SERVEFILE_DEFAULT_PORT , method = ' get ' , protocol = ' http ' , timeout = 5 ,
encoding = ' utf-8 ' , * * kwargs ) :
url = ' {} :// {} : {} {} ' . format ( protocol , host , port , path )
print ( ' Calling {} on {} with {} ' . format ( method , url , kwargs ) )
@ -103,7 +103,7 @@ def make_request(path='/', host='localhost', port=SERVEFILE_DEFAULT_PORT, method
return r
def check_download ( expected_data = None , path = ' / ' , fname = None , * * kwargs ) :
def check_download ( expected_data = None , path = ' / ' , fname = None , status_code = 200 , * * kwargs ) :
if fname is None :
fname = os . path . basename ( path )
r = make_request ( path , * * kwargs )
@ -117,22 +117,6 @@ def check_download(expected_data=None, path='/', fname=None, **kwargs):
return r # for additional tests
def _retry_while ( exception , function , timeout = 2 ) :
now = time . time # float seconds since epoch
def wrapped ( * args , * * kwargs ) :
timeout_after = now ( ) + timeout
while True :
try :
return function ( * args , * * kwargs )
except exception :
if now ( ) > = timeout_after :
raise
time . sleep ( 0.1 )
return wrapped
def _test_version ( run_servefile , standalone ) :
# we expect the version on stdout (python3.4+) or stderr(python2.6-3.3)
s = run_servefile ( ' --version ' , standalone = standalone , stdout = subprocess . PIPE , stderr = subprocess . STDOUT )
@ -146,7 +130,7 @@ def _test_version(run_servefile, standalone):
version = s . stdout . readline ( ) . decode ( ) . strip ( )
# hardcode version as string until servefile is a module
assert version == ' servefile 0.5. 4 '
assert version == ' servefile 0.5. 1 '
def test_version ( run_servefile ) :
@ -163,7 +147,7 @@ def test_correct_headers(run_servefile, datadir):
p = datadir ( { ' testfile ' : data } ) / ' testfile '
run_servefile ( str ( p ) )
r = _retry_while( ConnectionError , make_request) ( )
r = ( )
assert r . status_code == 200
assert r . headers . get ( ' Content-Type ' ) == ' application/octet-stream '
assert r . headers . get ( ' Content-Disposition ' ) == ' attachment; filename= " testfile " '
@ -176,7 +160,7 @@ def test_redirect_and_download(run_servefile, datadir):
run_servefile ( str ( p ) )
# redirect
r = _retry_while( ConnectionError , make_request) ( allow_redirects = False )
r = ( allow_redirects = False )
assert r . status_code == 302
assert r . headers . get ( ' Location ' ) == ' /testfile '
@ -191,7 +175,7 @@ def test_redirect_and_download_with_umlaut(run_servefile, datadir):
run_servefile ( str ( p ) )
# redirect
r = _retry_while( ConnectionError , make_request) ( allow_redirects = False )
r = ( allow_redirects = False )
assert r . status_code == 302
assert r . headers . get ( ' Location ' ) == ' / {} ' . format ( quote ( filename ) )
@ -206,7 +190,7 @@ def test_specify_port(run_servefile, datadir):
p = datadir ( { ' testfile ' : data } ) / ' testfile '
run_servefile ( [ str ( p ) , ' -p ' , str ( SERVEFILE_SECONDARY_PORT ) ] )
_retry_while( ConnectionError , check_download) ( data , fname = ' testfile ' , port = SERVEFILE_SECONDARY_PORT )
( data , fname = ' testfile ' , port = SERVEFILE_SECONDARY_PORT )
def test_ipv4_only ( run_servefile , datadir ) :
@ -214,7 +198,7 @@ def test_ipv4_only(run_servefile, datadir):
p = datadir ( { ' testfile ' : data } ) / ' testfile '
run_servefile ( [ str ( p ) , ' -4 ' ] )
_retry_while( ConnectionError , check_download) ( data , fname = ' testfile ' , host = ' 127.0.0.1 ' )
( data , fname = ' testfile ' , host = ' 127.0.0.1 ' )
sock = socket . socket ( socket . AF_INET6 )
with pytest . raises ( connrefused_exc ) :
@ -227,7 +211,7 @@ def test_big_download(run_servefile, datadir):
p = datadir ( { ' testfile ' : data } ) / ' testfile '
run_servefile ( str ( p ) )
_retry_while( ConnectionError , check_download) ( data , fname = ' testfile ' )
( data , fname = ' testfile ' )
def test_authentication ( run_servefile , datadir ) :
@ -236,11 +220,11 @@ def test_authentication(run_servefile, datadir):
run_servefile ( [ str ( p ) , ' -a ' , ' user:password ' ] )
for auth in [ ( ' foo ' , ' bar ' ) , ( ' user ' , ' wrong ' ) , ( ' unknown ' , ' password ' ) ] :
r = _retry_while( ConnectionError , make_request) ( auth = auth )
r = ( auth = auth )
assert ' 401 - Unauthorized ' in r . text
assert r . status_code == 401
_retry_while( ConnectionError , check_download) ( data , fname = ' testfile ' , auth = ( ' user ' , ' password ' ) )
( data , fname = ' testfile ' , auth = ( ' user ' , ' password ' ) )
def test_serve_directory ( run_servefile , datadir ) :
@ -257,12 +241,12 @@ def test_serve_directory(run_servefile, datadir):
# check if all files are in directory listing
# (could be made more sophisticated with beautifulsoup)
for path in ' / ' , ' /../ ' :
r = _retry_while( ConnectionError , make_request) ( path )
r = ( path )
for k in d :
assert quote ( k ) in r . text
for fname , content in d [ ' foo ' ] . items ( ) :
_retry_while( ConnectionError , check_download) ( content , ' /foo/ ' + fname )
( content , ' /foo/ ' + fname )
r = make_request ( ' /unknown ' )
assert r . status_code == 404
@ -284,7 +268,7 @@ def test_serve_relative_directory(run_servefile, datadir):
# check if all files are in directory listing
# (could be made more sophisticated with beautifulsoup)
for path in ' / ' , ' /../ ' :
r = _retry_while( ConnectionError , make_request) ( path )
r = ( path )
for k in d :
assert k in r . text
@ -308,14 +292,14 @@ def test_upload(run_servefile, tmp_path):
run_servefile ( [ ' -u ' , str ( uploaddir ) ] )
# check that servefile created the directory
assert uploaddir . is_dir ( )
# check upload form present
r = _retry_while ( ConnectionError , make_request ) ( )
r = ( )
assert r . status_code == 200
assert ' multipart/form-data ' in r . text
# check that servefile created the directory
assert uploaddir . is_dir ( )
# upload file
files = { ' file ' : ( ' haiku.txt ' , data ) }
r = make_request ( method = ' post ' , files = files )
@ -345,7 +329,7 @@ def test_upload_size_limit(run_servefile, tmp_path):
# upload file that is too big
files = { ' file ' : ( ' toobig ' , " x " * 2049 ) }
r = _retry_while( ConnectionError , make_request) ( method = ' post ' , files = files )
r = ( method = ' post ' , files = files )
assert ' Your file was too big ' in r . text
assert r . status_code == 413
assert not ( uploaddir / ' toobig ' ) . exists ( )
@ -357,20 +341,6 @@ def test_upload_size_limit(run_servefile, tmp_path):
assert r . status_code == 200
def test_upload_large_file ( run_servefile , tmp_path ) :
# small files end up in BytesIO while large files get temporary files. this
# test makes sure we hit the large file codepath at least once
uploaddir = tmp_path / ' upload '
run_servefile ( [ ' -u ' , str ( uploaddir ) ] )
data = " asdf " * 1024
files = { ' file ' : ( ' more_data.txt ' , data ) }
r = _retry_while ( ConnectionError , make_request ) ( method = ' post ' , files = files )
assert r . status_code == 200
with open ( str ( uploaddir / ' more_data.txt ' ) ) as f :
assert f . read ( ) == data
def test_tar_mode ( run_servefile , datadir ) :
d = {
' foo ' : {
@ -384,7 +354,7 @@ def test_tar_mode(run_servefile, datadir):
# test redirect?
# test contents of tar file
r = _retry_while( ConnectionError , make_request) ( )
r = ( )
assert r . status_code == 200
tar = tarfile . open ( fileobj = io . BytesIO ( r . content ) )
assert len ( tar . getmembers ( ) ) == 3
@ -400,7 +370,7 @@ def test_tar_compression(run_servefile, datadir):
p = datadir ( d )
run_servefile ( [ ' -c ' , ' gzip ' , ' -t ' , str ( p / ' foo ' ) ] )
r = _retry_while( ConnectionError , make_request) ( )
r = ( )
assert r . status_code == 200
tar = tarfile . open ( fileobj = io . BytesIO ( r . content ) , mode = ' r:gz ' )
assert len ( tar . getmembers ( ) ) == 1
@ -410,6 +380,7 @@ def test_https(run_servefile, datadir):
data = " NOOT NOOT "
p = datadir ( { ' testfile ' : data } ) / ' testfile '
run_servefile ( [ ' --ssl ' , str ( p ) ] )
time . sleep ( 0.2 ) # time for generating ssl certificates
# fingerprint = None
# while not fingerprint:
@ -424,7 +395,7 @@ def test_https(run_servefile, datadir):
# assert fingerprint
urllib3 . disable_warnings ( )
_retry_while( ConnectionError , check_download) ( data , protocol = ' https ' , verify = False )
( data , protocol = ' https ' , verify = False )
def test_https_big_download ( run_servefile , datadir ) :
@ -432,9 +403,10 @@ def test_https_big_download(run_servefile, datadir):
data = " x " * ( 10 * 1024 * * 2 )
p = datadir ( { ' testfile ' : data } ) / ' testfile '
run_servefile ( [ ' --ssl ' , str ( p ) ] )
time . sleep ( 0.2 ) # time for generating ssl certificates
urllib3 . disable_warnings ( )
_retry_while( ConnectionError , check_download) ( data , protocol = ' https ' , verify = False )
( data , protocol = ' https ' , verify = False )
def test_abort_download ( run_servefile , datadir ) :
@ -447,7 +419,7 @@ def test_abort_download(run_servefile, datadir):
# provoke a connection abort
# hopefully the buffers will not fill up with all of the 10mb
sock = socket . socket ( socket . AF_INET )
_retry_while( connrefused_exc , sock. connect ) ( ( " localhost " , SERVEFILE_DEFAULT_PORT ) )
. connect ( ( " localhost " , SERVEFILE_DEFAULT_PORT ) )
sock . send ( b " GET /testfile HTTP/1.0 \n \n " )
resp = sock . recv ( 100 )
assert resp != b ' '
Sebastian Lohff