@ -1,3 +1,4 @@
# -*- coding: utf-8 -*-
import io
import os
import pytest
@ -8,18 +9,36 @@ 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
if sys . version_info . major > = 3 :
from pathlib import Path
from urllib . parse import quote
connrefused_exc = ConnectionRefusedError
else :
from pathlib2 import Path
from urllib import quote
connrefused_exc = socket . error
def _get_port_from_env ( var_name , default ) :
port = int ( os . environ . get ( var_name , default ) )
if port == 0 :
# do a one-time port selection for a free port, use it for all tests
s = socket . socket ( )
s . bind ( ( ' ' , 0 ) )
port = s . getsockname ( ) [ 1 ]
s . close ( )
return port
SERVEFILE_DEFAULT_PORT = _get_port_from_env ( ' SERVEFILE_DEFAULT_PORT ' , 0 )
SERVEFILE_SECONDARY_PORT = _get_port_from_env ( ' SERVEFILE_SECONDARY_PORT ' , 0 )
@pytest.fixture
def run_servefile ( ) :
instances = [ ]
@ -34,9 +53,12 @@ def run_servefile():
# call servefile as python module
servefile_path = [ ' -m ' , ' servefile ' ]
# use non-default default port, if one is given via env (and none via args)
if ' -p ' not in args and ' --port ' not in args :
args . extend ( [ ' -p ' , str ( SERVEFILE_DEFAULT_PORT ) ] )
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
@ -62,22 +84,26 @@ def datadir(tmp_path):
_datadir ( v , new_path )
else :
if hasattr ( v , ' decode ' ) :
v = v . decode ( ) # python2 compability
v = v . decode ( ' utf-8 ' ) # python2 compability
( path / k ) . write_text ( v )
return path
return _datadir
def make_request ( path = ' / ' , host = ' localhost ' , port = 8080 , method = ' get ' , protocol = ' http ' , * * kwargs ) :
def make_request ( path = ' / ' , host = ' localhost ' , port = SERVEFILE_DEFAULT_PORT , method = ' get ' , protocol = ' http ' ,
encoding = ' utf-8 ' , * * kwargs ) :
url = ' {} :// {} : {} {} ' . format ( protocol , host , port , path )
print ( ' Calling {} on {} with {} ' . format ( method , url , kwargs ) )
r = getattr ( requests , method ) ( url , * * kwargs )
if r . encoding is None and encoding :
r . encoding = encoding
return r
def check_download ( expected_data = None , path = ' / ' , fname = None , status_code = 200 , * * kwargs ) :
def check_download ( expected_data = None , path = ' / ' , fname = None , * * kwargs ) :
if fname is None :
fname = os . path . basename ( path )
r = make_request ( path , * * kwargs )
@ -91,6 +117,22 @@ def check_download(expected_data=None, path='/', fname=None, status_code=200, **
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 )
@ -104,7 +146,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. 1 '
assert version == ' servefile 0.5. 4 '
def test_version ( run_servefile ) :
@ -121,7 +163,7 @@ def test_correct_headers(run_servefile, datadir):
p = datadir ( { ' testfile ' : data } ) / ' testfile '
run_servefile ( str ( p ) )
r = make_request( )
r = _retry_while( ConnectionError , make_request) ( )
assert r . status_code == 200
assert r . headers . get ( ' Content-Type ' ) == ' application/octet-stream '
assert r . headers . get ( ' Content-Disposition ' ) == ' attachment; filename= " testfile " '
@ -134,7 +176,7 @@ def test_redirect_and_download(run_servefile, datadir):
run_servefile ( str ( p ) )
# redirect
r = make_request( allow_redirects = False )
r = _retry_while( ConnectionError , make_request) ( allow_redirects = False )
assert r . status_code == 302
assert r . headers . get ( ' Location ' ) == ' /testfile '
@ -142,12 +184,29 @@ def test_redirect_and_download(run_servefile, datadir):
check_download ( data , fname = ' testfile ' )
def test_redirect_and_download_with_umlaut ( run_servefile , datadir ) :
data = " NÖÖT NÖÖT "
filename = " tästføile "
p = datadir ( { filename : data } ) / filename
run_servefile ( str ( p ) )
# redirect
r = _retry_while ( ConnectionError , make_request ) ( allow_redirects = False )
assert r . status_code == 302
assert r . headers . get ( ' Location ' ) == ' / {} ' . format ( quote ( filename ) )
# normal download
if sys . version_info . major < 3 :
data = unicode ( data , ' utf-8 ' )
check_download ( data , fname = filename )
def test_specify_port ( run_servefile , datadir ) :
data = " NOOT NOOT "
p = datadir ( { ' testfile ' : data } ) / ' testfile '
run_servefile ( [ str ( p ) , ' -p ' , ' 8081 ' ] )
run_servefile ( [ str ( p ) , ' -p ' , str ( SERVEFILE_SECONDARY_PORT ) ] )
check_download ( data , fname = ' testfile ' , port = 8081 )
_retry_while( ConnectionError , check_download) ( data , fname = ' testfile ' , port = SERVEFILE_SECONDARY_PORT )
def test_ipv4_only ( run_servefile , datadir ) :
@ -155,11 +214,11 @@ def test_ipv4_only(run_servefile, datadir):
p = datadir ( { ' testfile ' : data } ) / ' testfile '
run_servefile ( [ str ( p ) , ' -4 ' ] )
check_download( data , fname = ' testfile ' , host = ' 127.0.0.1 ' )
_retry_while( ConnectionError , check_download) ( data , fname = ' testfile ' , host = ' 127.0.0.1 ' )
sock = socket . socket ( socket . AF_INET6 )
with pytest . raises ( connrefused_exc ) :
sock . connect ( ( " ::1 " , 8080 ) )
sock . connect ( ( " ::1 " , SERVEFILE_DEFAULT_PORT ) )
def test_big_download ( run_servefile , datadir ) :
@ -168,7 +227,7 @@ def test_big_download(run_servefile, datadir):
p = datadir ( { ' testfile ' : data } ) / ' testfile '
run_servefile ( str ( p ) )
check_download( data , fname = ' testfile ' )
_retry_while( ConnectionError , check_download) ( data , fname = ' testfile ' )
def test_authentication ( run_servefile , datadir ) :
@ -177,11 +236,11 @@ def test_authentication(run_servefile, datadir):
run_servefile ( [ str ( p ) , ' -a ' , ' user:password ' ] )
for auth in [ ( ' foo ' , ' bar ' ) , ( ' user ' , ' wrong ' ) , ( ' unknown ' , ' password ' ) ] :
r = make_request( auth = auth )
r = _retry_while( ConnectionError , make_request) ( auth = auth )
assert ' 401 - Unauthorized ' in r . text
assert r . status_code == 401
check_download( data , fname = ' testfile ' , auth = ( ' user ' , ' password ' ) )
_retry_while( ConnectionError , check_download) ( data , fname = ' testfile ' , auth = ( ' user ' , ' password ' ) )
def test_serve_directory ( run_servefile , datadir ) :
@ -190,6 +249,7 @@ def test_serve_directory(run_servefile, datadir):
' bar ' : { ' thisisaverylongfilenamefortestingthatthisstillworksproperly ' : ' jup! ' } ,
' noot ' : ' still data in here ' ,
' bigfile ' : ' x ' * ( 10 * 1024 * * 2 ) ,
' möwe ' : ' KRAKRAKRAKA ' ,
}
p = datadir ( d )
run_servefile ( [ str ( p ) , ' -l ' ] )
@ -197,12 +257,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 = make_request( path )
r = _retry_while( ConnectionError , make_request) ( path )
for k in d :
assert k in r . text
assert quote( k) in r . text
for fname , content in d [ ' foo ' ] . items ( ) :
check_download( content , ' /foo/ ' + fname )
_retry_while( ConnectionError , check_download) ( content , ' /foo/ ' + fname )
r = make_request ( ' /unknown ' )
assert r . status_code == 404
@ -224,7 +284,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 = make_request( path )
r = _retry_while( ConnectionError , make_request) ( path )
for k in d :
assert k in r . text
@ -248,14 +308,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 = make_request( )
r = _retry_while ( ConnectionError , make_request ) ( )
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 )
@ -271,6 +331,13 @@ def test_upload(run_servefile, tmp_path):
with open ( str ( uploaddir / ' haiku.txt(1) ' ) ) as f :
assert f . read ( ) == data
# upload file using PUT
r = make_request ( " /haiku.txt " , method = ' put ' , data = data )
assert r . status_code == 201
assert ' OK! ' in r . text
with open ( str ( uploaddir / ' haiku.txt(2) ' ) ) as f :
assert f . read ( ) == data
def test_upload_size_limit ( run_servefile , tmp_path ) :
uploaddir = tmp_path / ' upload '
@ -278,7 +345,7 @@ def test_upload_size_limit(run_servefile, tmp_path):
# upload file that is too big
files = { ' file ' : ( ' toobig ' , " x " * 2049 ) }
r = make_request( method = ' post ' , files = files )
r = _retry_while( ConnectionError , make_request) ( method = ' post ' , files = files )
assert ' Your file was too big ' in r . text
assert r . status_code == 413
assert not ( uploaddir / ' toobig ' ) . exists ( )
@ -290,6 +357,20 @@ 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 ' : {
@ -303,7 +384,7 @@ def test_tar_mode(run_servefile, datadir):
# test redirect?
# test contents of tar file
r = make_request( )
r = _retry_while( ConnectionError , make_request) ( )
assert r . status_code == 200
tar = tarfile . open ( fileobj = io . BytesIO ( r . content ) )
assert len ( tar . getmembers ( ) ) == 3
@ -319,7 +400,7 @@ def test_tar_compression(run_servefile, datadir):
p = datadir ( d )
run_servefile ( [ ' -c ' , ' gzip ' , ' -t ' , str ( p / ' foo ' ) ] )
r = make_request( )
r = _retry_while( ConnectionError , make_request) ( )
assert r . status_code == 200
tar = tarfile . open ( fileobj = io . BytesIO ( r . content ) , mode = ' r:gz ' )
assert len ( tar . getmembers ( ) ) == 1
@ -329,7 +410,6 @@ 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:
@ -344,7 +424,7 @@ def test_https(run_servefile, datadir):
# assert fingerprint
urllib3 . disable_warnings ( )
check_download( data , protocol = ' https ' , verify = False )
_retry_while( ConnectionError , check_download) ( data , protocol = ' https ' , verify = False )
def test_https_big_download ( run_servefile , datadir ) :
@ -352,10 +432,9 @@ 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 ( )
check_download( data , protocol = ' https ' , verify = False )
_retry_while( ConnectionError , check_download) ( data , protocol = ' https ' , verify = False )
def test_abort_download ( run_servefile , datadir ) :
@ -368,7 +447,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 )
sock. connect ( ( " localhost " , 8080 ) )
_retry_while( connrefused_exc , sock. connect ) ( ( " localhost " , SERVEFILE_DEFAULT_PORT ) )
sock . send ( b " GET /testfile HTTP/1.0 \n \n " )
resp = sock . recv ( 100 )
assert resp != b ' '