servefile/tests/test_servefile.py

import io
import os
import pytest
import requests
import subprocess
import tarfile
import time
import urllib3

# crudly written to learn more about pytest and to have a base for refactoring


@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.3))
        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:
                if hasattr(v, 'decode'):
                    v = v.decode()  # python2 compability
                (path / k).write_text(v)

        return path
    return _datadir


def make_request(path='/', host='localhost', port=8080, method='get', protocol='http', **kwargs):
    url = '{}://{}:{}{}'.format(protocol, host, port, path)
    print('Calling {} on {} with {}'.format(method, url, kwargs))
    r = getattr(requests, method)(url, **kwargs)

    return r


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)
    assert r.status_code == 200
    assert r.text == expected_data
    assert r.headers.get('Content-Type') == 'application/octet-stream'
    if fname:
        assert r.headers.get('Content-Disposition') == 'attachment; filename="{}"'.format(fname)
    assert r.headers.get('Content-Transfer-Encoding') == 'binary'

    return r  # for additional tests


def test_version(run_servefile):
    # we expect the version on stdout (python3.4+) or stderr(python2.6-3.3)
    s = run_servefile('--version', stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
    s.wait()
    version = s.stdout.readline().decode().strip()

    # hardcode version as string until servefile is a module
    assert version == 'servefile 0.4.4'


def test_correct_headers(run_servefile, datadir):
    data = "NOOT NOOT"
    p = datadir({'testfile': data}) / 'testfile'
    run_servefile(str(p))

    r = 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"'
    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))

    # redirect
    r = make_request(allow_redirects=False)
    assert r.status_code == 302
    assert r.headers.get('Location') == '/testfile'

    # normal download
    check_download(data, fname='testfile')


def test_specify_port(run_servefile, datadir):
    data = "NOOT NOOT"
    p = datadir({'testfile': data}) / 'testfile'
    run_servefile([str(p), '-p', '8081'])

    check_download(data, fname='testfile', port=8081)


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))

    check_download(data, fname='testfile')


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 = make_request(auth=auth)
        assert '401 - Unauthorized' in r.text
        assert r.status_code == 401

    check_download(data, fname='testfile', auth=('user', 'password'))


def test_serve_directory(run_servefile, datadir):
    d = {
        'foo': {'kratzbaum': 'cat', 'I like Cats!': 'kitteh', '&&&&&&&': '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 = make_request(path)
        for k in d:
            assert k in r.text

    for fname, content in d['foo'].items():
        check_download(content, '/foo/' + fname)

    r = make_request('/unknown')
    assert r.status_code == 404

    # download
    check_download('jup!', '/bar/thisisaverylongfilenamefortestingthatthisstillworksproperly')


def test_upload(run_servefile, tmp_path):
    data = ('this is my live now\n'
            'uploading strings to servers\n'
            'so very joyful')
    uploaddir = tmp_path / 'upload'
    # check that uploaddir does not exist before servefile is started
    assert not uploaddir.is_dir()

    run_servefile(['-u', str(uploaddir)])

    # check that servefile created the directory
    assert uploaddir.is_dir()

    # check upload form present
    r = make_request()
    assert r.status_code == 200
    assert 'multipart/form-data' in r.text

    # upload file
    files = {'file': ('haiku.txt', data)}
    r = make_request(method='post', files=files)
    assert 'Thanks' in r.text
    assert r.status_code == 200
    with open(str(uploaddir / 'haiku.txt')) as f:
        assert f.read() == data

    # upload file AGAIN!! (and check it is available unter a different name)
    files = {'file': ('haiku.txt', data)}
    r = make_request(method='post', files=files)
    assert r.status_code == 200
    with open(str(uploaddir / 'haiku.txt(1)')) as f:
        assert f.read() == data


def test_upload_size_limit(run_servefile, tmp_path):
    uploaddir = tmp_path / 'upload'
    run_servefile(['-s', '2kb', '-u', str(uploaddir)])

    # upload file that is too big
    files = {'file': ('toobig', "x" * 2049)}
    r = 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()

    # upload file that should fit
    # the size has to be smaller than 2kb, as the sent size also includes mime-headers
    files = {'file': ('justright', "x" * 1900)}
    r = make_request(method='post', files=files)
    assert r.status_code == 200


def test_tar_mode(run_servefile, datadir):
    d = {
        'foo': {
            'bar': 'hello testmode my old friend',
            'baz': 'you came to test me once again',
        }
    }
    p = datadir(d)
    run_servefile(['-t', str(p / 'foo')])

    # test redirect?

    # test contents of tar file
    r = make_request()
    assert r.status_code == 200
    tar = tarfile.open(fileobj=io.BytesIO(r.content))
    assert len(tar.getmembers()) == 3
    assert tar.getmember('foo').isdir()
    for filename, content in d['foo'].items():
        info = tar.getmember('foo/{}'.format(filename))
        assert info.isfile
        assert tar.extractfile(info.path).read().decode() == content


def test_tar_compression(run_servefile, datadir):
    d = {'foo': 'blubb'}
    p = datadir(d)
    run_servefile(['-c', 'gzip', '-t', str(p / 'foo')])

    r = make_request()
    assert r.status_code == 200
    tar = tarfile.open(fileobj=io.BytesIO(r.content), mode='r:gz')
    assert len(tar.getmembers()) == 1


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:
    #     line = s.stdout.readline()
    #     print(line)
    #     # if we find this line we went too far...
    #     assert not line.startswith("Some addresses this file will be available at")

    #     if line.startswith("SHA1 fingerprint"):
    #         fingerprint = line.replace("SHA1 fingerprint: ", "").strip()
    #         break

    # assert fingerprint
    urllib3.disable_warnings()
    check_download(data, protocol='https', verify=False)

def test_https_big_download(run_servefile, datadir):
    # test with about 10 mb of data
    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)