How to use aioftp - 10 common examples

async def worker(fname):

        _client = aioftp.Client(loop=loop, ssl=client.ssl)
        await _client.connect("127.0.0.1", PORT)
        await _client.login()
        await _client.download(
            "tests/foo/foo.txt",
            str.format("tests/foo/{}", fname),
            write_into=True
        )
        await _client.quit()

async def test_server_side_throttle(pair_factory, skip_sleep, times, users,
                                    throttle_direction, data_direction,
                                    throttle_level):
    async with pair_factory() as pair:
        names = []
        for i in range(users):
            name = f"foo{i}"
            names.append(name)
            await pair.make_server_files(name, size=SIZE)
        throttle = reduce(getattr, [throttle_level, throttle_direction],
                          pair.server)
        throttle.limit = SIZE / times
        clients = []
        for name in names:
            c = aioftp.Client(path_io_factory=aioftp.MemoryPathIO)
            async with c.path_io.open(Path(name), "wb") as f:
                await f.write(b"-" * SIZE)
            await c.connect(pair.server.server_host, pair.server.server_port)
            await c.login()
            clients.append(c)
        coros = [getattr(c, data_direction)(n) for c, n in zip(clients, names)]
        await asyncio.gather(*coros)
        await asyncio.gather(*[c.quit() for c in clients])
    throttled = {("read", "upload"), ("write", "download")}
    if (throttle_direction, data_direction) not in throttled:
        assert skip_sleep.is_close(0)
    else:
        t = times
        if throttle_level == "throttle":  # global
            t *= users
        assert skip_sleep.is_close(t)

async def worker(fname):

        _client = aioftp.Client(loop=loop, ssl=client.ssl)
        await _client.connect("127.0.0.1", PORT)
        await _client.login()
        await _client.upload(
            "tests/foo/foo.txt",
            str.format("tests/foo/{}", fname),
            write_into=True
        )
        await _client.quit()

    server_args=([(aioftp.User(maximum_connections=4),)], {}))
@expect_codes_in_exception("530")
@with_connection
async def test_multiply_connections_limited_error(loop, client, server):

    clients = [aioftp.Client(loop=loop, ssl=client.ssl) for _ in range(5)]
    for client in clients:

        await client.connect("127.0.0.1", PORT)
        await client.login()

    for client in clients:

        await client.quit()

writer.write(b)
        await writer.drain()

    await client.command(None, "2xx", "1xx")
    await client.quit()

    with f.open("rb") as fin:

        rb = fin.read()

    f.unlink()

    nose.tools.eq_(b, rb)


class SlowServer(aioftp.Server):

    async def dispatcher(self, reader, writer):

        await asyncio.sleep(10, loop=self.loop)


@nose.tools.raises(asyncio.TimeoutError)
def test_client_socket_timeout():
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(None)
    server = SlowServer(loop=loop)
    client = aioftp.Client(loop=loop, socket_timeout=1)

    async def coro():
        try:
            await server.start(None, 8888)

def test_user_manager_timeout():
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(None)
    server = aioftp.Server(SlowUserManager(None, timeout=1, loop=loop),
                           loop=loop)
    client = aioftp.Client(loop=loop)

    async def coro():
        try:
            await server.start(None, 8888)
            await client.connect("127.0.0.1", 8888)
            await client.login()
        finally:
            await server.close()

    loop.run_until_complete(coro())

def test_parse_list_line_failed():
    with pytest.raises(ValueError):
        aioftp.Client(encoding="utf-8").parse_list_line(b"what a hell?!")

def run_in_loop(s_args, s_kwargs, c_args, c_kwargs, s_ssl=None, c_ssl=None):
                logging.basicConfig(
                    level=logging.INFO,
                    format="%(asctime)s [%(name)s] %(message)s",
                    datefmt="[%H:%M:%S]:",
                )
                loop = asyncio.new_event_loop()
                asyncio.set_event_loop(None)
                server = aioftp.Server(*s_args, loop=loop, ssl=s_ssl, **s_kwargs)
                client = aioftp.Client(*c_args, loop=loop, ssl=c_ssl, **c_kwargs)
                try:
                    loop.run_until_complete(f(loop, client, server))
                finally:
                    if hasattr(server, "server"):
                        loop.run_until_complete(server.close())
                    if hasattr(client, "writer"):
                        client.close()
                    loop.close()

        [[aioftp.User(read_speed_limit_per_connection=200 * 1024)]],
        {},
    ))
@with_connection
@with_tmp_dir("foo")
async def test_server_user_per_connection_read_throttle_multi_users(loop,
                                                                    client,
                                                                    server, *,
                                                                    tmp_dir):

    async def worker(fname):

        _client = aioftp.Client(loop=loop, ssl=client.ssl)
        await _client.connect("127.0.0.1", PORT)
        await _client.login()
        await _client.upload(
            "tests/foo/foo.txt",

        [(aioftp.User(base_path="tests/foo", home_path="/"),)],
        {}))
@with_connection
@expect_codes_in_exception("503")
@with_tmp_dir("foo")
async def test_wait_pasv_timeout_fail_long(loop, client, server, *, tmp_dir):

    f = tmp_dir / "foo.txt"
    b = b"foobar"

    await client.login()
    await client.command("STOR " + f.name)
    await asyncio.sleep(2, loop=loop)
    reader, writer = await client.get_passive_connection("I")

    with contextlib.closing(writer) as writer: