No Description
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

test_servefile.py 11KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358
  1. import io
  2. import os
  3. import pytest
  4. import requests
  5. import socket
  6. import subprocess
  7. import sys
  8. import tarfile
  9. import time
  10. import urllib3
  11. # crudly written to learn more about pytest and to have a base for refactoring
  12. if sys.version_info.major >= 3:
  13. from pathlib import Path
  14. connrefused_exc = ConnectionRefusedError
  15. else:
  16. from pathlib2 import Path
  17. connrefused_exc = socket.error
  18. @pytest.fixture
  19. def run_servefile():
  20. instances = []
  21. def _run_servefile(args, **kwargs):
  22. if not isinstance(args, list):
  23. args = [args]
  24. if kwargs.pop('standalone', None):
  25. # directly call servefile.py
  26. servefile_path = [str(Path(__file__).parent.parent / 'servefile' / 'servefile.py')]
  27. else:
  28. # call servefile as python module
  29. servefile_path = ['-m', 'servefile']
  30. print("running {} with args {}".format(", ".join(servefile_path), args))
  31. p = subprocess.Popen([sys.executable] + servefile_path + args, **kwargs)
  32. time.sleep(kwargs.get('timeout', 0.3))
  33. instances.append(p)
  34. return p
  35. yield _run_servefile
  36. for instance in instances:
  37. try:
  38. instance.terminate()
  39. except OSError:
  40. pass
  41. instance.wait()
  42. @pytest.fixture
  43. def datadir(tmp_path):
  44. def _datadir(data, path=None):
  45. path = path or tmp_path
  46. for k, v in data.items():
  47. if isinstance(v, dict):
  48. new_path = path / k
  49. new_path.mkdir()
  50. _datadir(v, new_path)
  51. else:
  52. if hasattr(v, 'decode'):
  53. v = v.decode() # python2 compability
  54. (path / k).write_text(v)
  55. return path
  56. return _datadir
  57. def make_request(path='/', host='localhost', port=8080, method='get', protocol='http', **kwargs):
  58. url = '{}://{}:{}{}'.format(protocol, host, port, path)
  59. print('Calling {} on {} with {}'.format(method, url, kwargs))
  60. r = getattr(requests, method)(url, **kwargs)
  61. return r
  62. def check_download(expected_data=None, path='/', fname=None, status_code=200, **kwargs):
  63. if fname is None:
  64. fname = os.path.basename(path)
  65. r = make_request(path, **kwargs)
  66. assert r.status_code == 200
  67. assert r.text == expected_data
  68. assert r.headers.get('Content-Type') == 'application/octet-stream'
  69. if fname:
  70. assert r.headers.get('Content-Disposition') == 'attachment; filename="{}"'.format(fname)
  71. assert r.headers.get('Content-Transfer-Encoding') == 'binary'
  72. return r # for additional tests
  73. def _test_version(run_servefile, standalone):
  74. # we expect the version on stdout (python3.4+) or stderr(python2.6-3.3)
  75. s = run_servefile('--version', standalone=standalone, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
  76. s.wait()
  77. version = s.stdout.readline().decode().strip()
  78. # python2 is deprecated, but we still want our tests to run for it
  79. # CryptographyDeprecationWarnings get in the way for this
  80. if 'CryptographyDeprecationWarning' in version:
  81. s.stdout.readline() # ignore "from x import y" line
  82. version = s.stdout.readline().decode().strip()
  83. # hardcode version as string until servefile is a module
  84. assert version == 'servefile 0.5.0'
  85. def test_version(run_servefile):
  86. _test_version(run_servefile, standalone=False)
  87. def test_version_standalone(run_servefile):
  88. # test if servefile also works by calling servefile.py directly
  89. _test_version(run_servefile, standalone=True)
  90. def test_correct_headers(run_servefile, datadir):
  91. data = "NOOT NOOT"
  92. p = datadir({'testfile': data}) / 'testfile'
  93. run_servefile(str(p))
  94. r = make_request()
  95. assert r.status_code == 200
  96. assert r.headers.get('Content-Type') == 'application/octet-stream'
  97. assert r.headers.get('Content-Disposition') == 'attachment; filename="testfile"'
  98. assert r.headers.get('Content-Transfer-Encoding') == 'binary'
  99. def test_redirect_and_download(run_servefile, datadir):
  100. data = "NOOT NOOT"
  101. p = datadir({'testfile': data}) / 'testfile'
  102. run_servefile(str(p))
  103. # redirect
  104. r = make_request(allow_redirects=False)
  105. assert r.status_code == 302
  106. assert r.headers.get('Location') == '/testfile'
  107. # normal download
  108. check_download(data, fname='testfile')
  109. def test_specify_port(run_servefile, datadir):
  110. data = "NOOT NOOT"
  111. p = datadir({'testfile': data}) / 'testfile'
  112. run_servefile([str(p), '-p', '8081'])
  113. check_download(data, fname='testfile', port=8081)
  114. def test_ipv4_only(run_servefile, datadir):
  115. data = "NOOT NOOT"
  116. p = datadir({'testfile': data}) / 'testfile'
  117. run_servefile([str(p), '-4'])
  118. check_download(data, fname='testfile', host='127.0.0.1')
  119. sock = socket.socket(socket.AF_INET6)
  120. with pytest.raises(connrefused_exc):
  121. sock.connect(("::1", 8080))
  122. def test_big_download(run_servefile, datadir):
  123. # test with about 10 mb of data
  124. data = "x" * (10 * 1024 ** 2)
  125. p = datadir({'testfile': data}) / 'testfile'
  126. run_servefile(str(p))
  127. check_download(data, fname='testfile')
  128. def test_authentication(run_servefile, datadir):
  129. data = "NOOT NOOT"
  130. p = datadir({'testfile': data}) / 'testfile'
  131. run_servefile([str(p), '-a', 'user:password'])
  132. for auth in [('foo', 'bar'), ('user', 'wrong'), ('unknown', 'password')]:
  133. r = make_request(auth=auth)
  134. assert '401 - Unauthorized' in r.text
  135. assert r.status_code == 401
  136. check_download(data, fname='testfile', auth=('user', 'password'))
  137. def test_serve_directory(run_servefile, datadir):
  138. d = {
  139. 'foo': {'kratzbaum': 'cat', 'I like Cats!': 'kitteh', '&&&&&&&': 'wheee'},
  140. 'bar': {'thisisaverylongfilenamefortestingthatthisstillworksproperly': 'jup!'},
  141. 'noot': 'still data in here',
  142. 'bigfile': 'x' * (10 * 1024 ** 2),
  143. }
  144. p = datadir(d)
  145. run_servefile([str(p), '-l'])
  146. # check if all files are in directory listing
  147. # (could be made more sophisticated with beautifulsoup)
  148. for path in '/', '/../':
  149. r = make_request(path)
  150. for k in d:
  151. assert k in r.text
  152. for fname, content in d['foo'].items():
  153. check_download(content, '/foo/' + fname)
  154. r = make_request('/unknown')
  155. assert r.status_code == 404
  156. # download
  157. check_download('jup!', '/bar/thisisaverylongfilenamefortestingthatthisstillworksproperly')
  158. def test_serve_relative_directory(run_servefile, datadir):
  159. d = {
  160. 'foo': {'kratzbaum': 'cat', 'I like Cats!': 'kitteh', '&&&&&&&': 'wheee'},
  161. 'bar': {'thisisaverylongfilenamefortestingthatthisstillworksproperly': 'jup!'},
  162. 'noot': 'still data in here',
  163. 'bigfile': 'x' * (10 * 1024 ** 2),
  164. }
  165. p = datadir(d)
  166. run_servefile(['../', '-l'], cwd=os.path.join(str(p), 'foo'))
  167. # check if all files are in directory listing
  168. # (could be made more sophisticated with beautifulsoup)
  169. for path in '/', '/../':
  170. r = make_request(path)
  171. for k in d:
  172. assert k in r.text
  173. for fname, content in d['foo'].items():
  174. check_download(content, '/foo/' + fname)
  175. r = make_request('/unknown')
  176. assert r.status_code == 404
  177. # download
  178. check_download('jup!', '/bar/thisisaverylongfilenamefortestingthatthisstillworksproperly')
  179. def test_upload(run_servefile, tmp_path):
  180. data = ('this is my live now\n'
  181. 'uploading strings to servers\n'
  182. 'so very joyful')
  183. uploaddir = tmp_path / 'upload'
  184. # check that uploaddir does not exist before servefile is started
  185. assert not uploaddir.is_dir()
  186. run_servefile(['-u', str(uploaddir)])
  187. # check that servefile created the directory
  188. assert uploaddir.is_dir()
  189. # check upload form present
  190. r = make_request()
  191. assert r.status_code == 200
  192. assert 'multipart/form-data' in r.text
  193. # upload file
  194. files = {'file': ('haiku.txt', data)}
  195. r = make_request(method='post', files=files)
  196. assert 'Thanks' in r.text
  197. assert r.status_code == 200
  198. with open(str(uploaddir / 'haiku.txt')) as f:
  199. assert f.read() == data
  200. # upload file AGAIN!! (and check it is available unter a different name)
  201. files = {'file': ('haiku.txt', data)}
  202. r = make_request(method='post', files=files)
  203. assert r.status_code == 200
  204. with open(str(uploaddir / 'haiku.txt(1)')) as f:
  205. assert f.read() == data
  206. def test_upload_size_limit(run_servefile, tmp_path):
  207. uploaddir = tmp_path / 'upload'
  208. run_servefile(['-s', '2kb', '-u', str(uploaddir)])
  209. # upload file that is too big
  210. files = {'file': ('toobig', "x" * 2049)}
  211. r = make_request(method='post', files=files)
  212. assert 'Your file was too big' in r.text
  213. assert r.status_code == 413
  214. assert not (uploaddir / 'toobig').exists()
  215. # upload file that should fit
  216. # the size has to be smaller than 2kb, as the sent size also includes mime-headers
  217. files = {'file': ('justright', "x" * 1900)}
  218. r = make_request(method='post', files=files)
  219. assert r.status_code == 200
  220. def test_tar_mode(run_servefile, datadir):
  221. d = {
  222. 'foo': {
  223. 'bar': 'hello testmode my old friend',
  224. 'baz': 'you came to test me once again',
  225. }
  226. }
  227. p = datadir(d)
  228. run_servefile(['-t', str(p / 'foo')])
  229. # test redirect?
  230. # test contents of tar file
  231. r = make_request()
  232. assert r.status_code == 200
  233. tar = tarfile.open(fileobj=io.BytesIO(r.content))
  234. assert len(tar.getmembers()) == 3
  235. assert tar.getmember('foo').isdir()
  236. for filename, content in d['foo'].items():
  237. info = tar.getmember('foo/{}'.format(filename))
  238. assert info.isfile
  239. assert tar.extractfile(info.path).read().decode() == content
  240. def test_tar_compression(run_servefile, datadir):
  241. d = {'foo': 'blubb'}
  242. p = datadir(d)
  243. run_servefile(['-c', 'gzip', '-t', str(p / 'foo')])
  244. r = make_request()
  245. assert r.status_code == 200
  246. tar = tarfile.open(fileobj=io.BytesIO(r.content), mode='r:gz')
  247. assert len(tar.getmembers()) == 1
  248. def test_https(run_servefile, datadir):
  249. data = "NOOT NOOT"
  250. p = datadir({'testfile': data}) / 'testfile'
  251. run_servefile(['--ssl', str(p)])
  252. time.sleep(0.2) # time for generating ssl certificates
  253. # fingerprint = None
  254. # while not fingerprint:
  255. # line = s.stdout.readline()
  256. # print(line)
  257. # # if we find this line we went too far...
  258. # assert not line.startswith("Some addresses this file will be available at")
  259. # if line.startswith("SHA1 fingerprint"):
  260. # fingerprint = line.replace("SHA1 fingerprint: ", "").strip()
  261. # break
  262. # assert fingerprint
  263. urllib3.disable_warnings()
  264. check_download(data, protocol='https', verify=False)
  265. def test_https_big_download(run_servefile, datadir):
  266. # test with about 10 mb of data
  267. data = "x" * (10 * 1024 ** 2)
  268. p = datadir({'testfile': data}) / 'testfile'
  269. run_servefile(['--ssl', str(p)])
  270. time.sleep(0.2) # time for generating ssl certificates
  271. urllib3.disable_warnings()
  272. check_download(data, protocol='https', verify=False)