FTPServer fixture example

test_server.py

import re
from pyftpdlib.authorizers import DummyAuthorizer
from pyftpdlib.handlers import FTPHandler
from pyftpdlib.servers import FTPServer
import pytest


@pytest.yield_fixture
def ftp_server(tmpdir, request):
    authorizer = DummyAuthorizer()

    # Define a new user having full r/w permissions and a read-only
    # anonymous user
    directory = tmpdir.join('_ftp_server_dir').ensure(dir=1)
    authorizer.add_anonymous(unicode(directory), perm=DummyAuthorizer.read_perms + DummyAuthorizer.write_perms)

    # Instantiate FTP handler class
    handler = FTPHandler
    handler.authorizer = authorizer

    # Define a customized banner (string returned when client connects)
    handler.banner = "pyftpdlib based ftpd ready."

    # Instantiate FTP server class and listen on 0.0.0.0:2121
    is_slave = hasattr(request.config, 'slaveinput')
    port = 46000
    if is_slave:
        slave_id = request.config.slaveinput['slaveid']
        slave_num = re.match(r'.*(\d+)', slave_id).group(1)
        port += slave_num

    address = ('', port)
    server = FTPServer(address, handler)

    # start ftp server
    import threading
    t = threading.Thread(target=server.serve_forever)
    t.start()

    class FTPFixture:
        pass

    fixture = FTPFixture()
    fixture.url = address
    fixture.directory = directory
    yield fixture
    server.close_all()
    t.join()


def test_upload_package(ftp_server, tmpdir):
    local_cache_dir = tmpdir.join('local_cache').ensure(dir=1)
    root_dir = tmpdir.join('project/coilib/build/tests').ensure(dir=1)
    root_dir.join('tests.xml').write('tests xml contents')

    cache = ArtifactsCache(
        'tag',
        ftp_server.url,
        unicode(local_cache_dir),
        unicode(root_dir),
        'package info',
    )
    hash_value = ''
    assert cache.push(['tests.xml'], 'annotation contents')

    package_path = ftp_server.directory.join('/tag-%s.zip' % hash_value)