1import pytest
2import asyncio
3
4from aioredis import ConnectionClosedError, ReplyError
5from aioredis.pool import ConnectionsPool
6from aioredis import Redis
7from _testutils import redis_version
8
9
10async def test_repr(create_redis, server):
11    redis = await create_redis(server.tcp_address, db=1)
12    assert repr(redis) in {
13        '<Redis <RedisConnection [db:1]>>',
14        '<Redis <ConnectionsPool [db:1, size:[1:10], free:1]>>',
15        }
16
17    redis = await create_redis(server.tcp_address, db=0)
18    assert repr(redis) in {
19        '<Redis <RedisConnection [db:0]>>',
20        '<Redis <ConnectionsPool [db:0, size:[1:10], free:1]>>',
21        }
22
23
24async def test_auth(redis):
25    expected_message = "ERR Client sent AUTH, but no password is set"
26    with pytest.raises(ReplyError, match=expected_message):
27        await redis.auth('')
28
29
30async def test_echo(redis):
31    resp = await redis.echo('ECHO')
32    assert resp == b'ECHO'
33
34    with pytest.raises(TypeError):
35        await redis.echo(None)
36
37
38async def test_ping(redis):
39    assert await redis.ping() == b'PONG'
40
41
42async def test_quit(redis):
43    expected = (ConnectionClosedError, ConnectionError)
44    try:
45        assert b'OK' == await redis.quit()
46    except expected:
47        pass
48
49    if not isinstance(redis.connection, ConnectionsPool):
50        # reader task may not yet been cancelled and _do_close not called
51        #   so the ConnectionClosedError may be raised (or ConnectionError)
52        with pytest.raises(expected):
53            try:
54                await redis.ping()
55            except asyncio.CancelledError:
56                assert False, "Cancelled error must not be raised"
57
58        # wait one loop iteration until it get surely closed
59        await asyncio.sleep(0)
60        assert redis.connection.closed
61
62        with pytest.raises(ConnectionClosedError):
63            await redis.ping()
64
65
66async def test_select(redis):
67    assert redis.db == 0
68
69    resp = await redis.select(1)
70    assert resp is True
71    assert redis.db == 1
72    assert redis.connection.db == 1
73
74
75async def test_encoding(create_redis, server):
76    redis = await create_redis(server.tcp_address, db=1, encoding='utf-8')
77    assert redis.encoding == 'utf-8'
78
79
80async def test_yield_from_backwards_compatibility(create_redis, server):
81    redis = await create_redis(server.tcp_address)
82
83    assert isinstance(redis, Redis)
84    # TODO: there should not be warning
85    # with pytest.warns(UserWarning):
86    with await redis as client:
87        assert isinstance(client, Redis)
88        assert client is not redis
89        assert await client.ping()
90
91
92@redis_version(4, 0, 0, reason="SWAPDB is available since redis>=4.0.0")
93async def test_swapdb(create_redis, start_server):
94    server = start_server('swapdb_1')
95    cli1 = await create_redis(server.tcp_address, db=0)
96    cli2 = await create_redis(server.tcp_address, db=1)
97
98    await cli1.flushall()
99    assert await cli1.set('key', 'val') is True
100    assert await cli1.exists('key')
101    assert not await cli2.exists('key')
102
103    assert await cli1.swapdb(0, 1) is True
104    assert not await cli1.exists('key')
105    assert await cli2.exists('key')
106