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