1import functools
2
3import pytest
4
5import aioftp
6
7
8@pytest.mark.asyncio
9async def test_multiply_connections_no_limits(pair_factory):
10    Client = functools.partial(aioftp.Client,
11                               path_io_factory=aioftp.MemoryPathIO)
12    async with pair_factory() as pair:
13        s = pair.server
14        clients = [Client() for _ in range(4)]
15        for c in clients:
16            await c.connect(s.server_host, s.server_port)
17            await c.login()
18        for c in clients:
19            await c.quit()
20
21
22@pytest.mark.asyncio
23async def test_multiply_connections_limited_error(pair_factory, Server,
24                                                  expect_codes_in_exception):
25    Client = functools.partial(aioftp.Client,
26                               path_io_factory=aioftp.MemoryPathIO)
27    s = Server(maximum_connections=4)
28    async with pair_factory(None, s) as pair:
29        s = pair.server
30        clients = [Client() for _ in range(4)]
31        for c in clients[:-1]:
32            await c.connect(s.server_host, s.server_port)
33            await c.login()
34        with expect_codes_in_exception("421"):
35            await clients[-1].connect(s.server_host, s.server_port)
36        for c in clients[:-1]:
37            await c.quit()
38
39
40@pytest.mark.asyncio
41async def test_multiply_user_commands(pair_factory, Server):
42    s = Server(maximum_connections=1)
43    async with pair_factory(None, s) as pair:
44        for _ in range(10):
45            await pair.client.login()
46
47
48@pytest.mark.asyncio
49async def test_multiply_connections_with_user_limited_error(
50        pair_factory, Server, expect_codes_in_exception):
51    Client = functools.partial(aioftp.Client,
52                               path_io_factory=aioftp.MemoryPathIO)
53    s = Server([aioftp.User("foo", maximum_connections=4)])
54    async with pair_factory(None, s, connected=False) as pair:
55        s = pair.server
56        clients = [Client() for _ in range(5)]
57        for c in clients[:-1]:
58            await c.connect(s.server_host, s.server_port)
59            await c.login("foo")
60        await clients[-1].connect(s.server_host, s.server_port)
61        with expect_codes_in_exception("530"):
62            await clients[-1].login("foo")
63        for c in clients[:-1]:
64            await c.quit()
65
66
67@pytest.mark.asyncio
68async def test_multiply_connections_relogin_balanced(
69        pair_factory, Server, expect_codes_in_exception):
70    Client = functools.partial(aioftp.Client,
71                               path_io_factory=aioftp.MemoryPathIO)
72    s = Server(maximum_connections=4)
73    async with pair_factory(None, s, connected=False) as pair:
74        s = pair.server
75        clients = [Client() for _ in range(5)]
76        for c in clients[:-1]:
77            await c.connect(s.server_host, s.server_port)
78            await c.login()
79        await clients[0].quit()
80        await clients[-1].connect(s.server_host, s.server_port)
81        await clients[-1].login()
82        for c in clients[1:]:
83            await c.quit()
84