# Path: arq/connections.py
# class RedisSettings:
#     """
#     No-Op class used to hold redis connection redis_settings.
# 
#     Used by :func:`arq.connections.create_pool` and :class:`arq.worker.Worker`.
#     """
# 
#     host: Union[str, List[Tuple[str, int]]] = 'localhost'
#     port: int = 6379
#     database: int = 0
#     password: Optional[str] = None
#     ssl: Union[bool, None, SSLContext] = None
#     conn_timeout: int = 1
#     conn_retries: int = 5
#     conn_retry_delay: int = 1
# 
#     sentinel: bool = False
#     sentinel_master: str = 'mymaster'
# 
#     @classmethod
#     def from_dsn(cls, dsn: str) -> 'RedisSettings':
#         conf = urlparse(dsn)
#         assert conf.scheme in {'redis', 'rediss'}, 'invalid DSN scheme'
#         return RedisSettings(
#             host=conf.hostname or 'localhost',
#             port=conf.port or 6379,
#             ssl=conf.scheme == 'rediss',
#             password=conf.password,
#             database=int((conf.path or '0').strip('/')),
#         )
# 
#     def __repr__(self) -> str:
#         return 'RedisSettings({})'.format(', '.join(f'{k}={v!r}' for k, v in self.__dict__.items()))
# 
# async def log_redis_info(redis: Redis, log_func: Callable[[str], Any]) -> None:
#     async with redis.pipeline(transaction=True) as pipe:
#         pipe.info(section='Server')
#         pipe.info(section='Memory')
#         pipe.info(section='Clients')
#         pipe.dbsize()
#         info_server, info_memory, info_clients, key_count = await pipe.execute()
# 
#     redis_version = info_server.get('redis_version', '?')
#     mem_usage = info_memory.get('used_memory_human', '?')
#     clients_connected = info_clients.get('connected_clients', '?')
# 
#     log_func(
#         f'redis_version={redis_version} '
#         f'mem_usage={mem_usage} '
#         f'clients_connected={clients_connected} '
#         f'db_keys={key_count}'
#     )


import logging
import re
import pytest
import arq.typing
import arq.utils
from datetime import timedelta
from pydantic import BaseModel, validator
from redis.asyncio import ConnectionError, ResponseError
from arq.connections import RedisSettings, log_redis_info
    with pytest.raises(ConnectionError):
        await create_pool(RedisSettings(port=0, conn_retry_delay=0))
    assert arq.utils.asyncio.sleep.call_count == 5


@pytest.mark.skip(reason='this breaks many other tests as low level connections remain after failed connection')
async def test_redis_sentinel_failure(create_pool, cancel_remaining_task, mocker):
    settings = RedisSettings()
    settings.host = [('localhost', 6379), ('localhost', 6379)]
    settings.sentinel = True
    with pytest.raises(ResponseError, match='unknown command `SENTINEL`'):
        await create_pool(settings)


async def test_redis_success_log(caplog, create_pool):
    caplog.set_level(logging.INFO)
    settings = RedisSettings()
    pool = await create_pool(settings)
    assert 'redis connection successful' not in [r.message for r in caplog.records]
    await pool.close(close_connection_pool=True)

    pool = await create_pool(settings, retry=1)
    assert 'redis connection successful' in [r.message for r in caplog.records]
    await pool.close(close_connection_pool=True)


async def test_redis_log(create_pool):
    redis = await create_pool(RedisSettings())
    await redis.flushall()
    await redis.set(b'a', b'1')

# Path: arq/connections.py
# class RedisSettings:
#     """
#     No-Op class used to hold redis connection redis_settings.
# 
#     Used by :func:`arq.connections.create_pool` and :class:`arq.worker.Worker`.
#     """
# 
#     host: Union[str, List[Tuple[str, int]]] = 'localhost'
#     port: int = 6379
#     database: int = 0
#     password: Optional[str] = None
#     ssl: Union[bool, None, SSLContext] = None
#     conn_timeout: int = 1
#     conn_retries: int = 5
#     conn_retry_delay: int = 1
# 
#     sentinel: bool = False
#     sentinel_master: str = 'mymaster'
# 
#     @classmethod
#     def from_dsn(cls, dsn: str) -> 'RedisSettings':
#         conf = urlparse(dsn)
#         assert conf.scheme in {'redis', 'rediss'}, 'invalid DSN scheme'
#         return RedisSettings(
#             host=conf.hostname or 'localhost',
#             port=conf.port or 6379,
#             ssl=conf.scheme == 'rediss',
#             password=conf.password,
#             database=int((conf.path or '0').strip('/')),
#         )
# 
#     def __repr__(self) -> str:
#         return 'RedisSettings({})'.format(', '.join(f'{k}={v!r}' for k, v in self.__dict__.items()))
# 
# async def log_redis_info(redis: Redis, log_func: Callable[[str], Any]) -> None:
#     async with redis.pipeline(transaction=True) as pipe:
#         pipe.info(section='Server')
#         pipe.info(section='Memory')
#         pipe.info(section='Clients')
#         pipe.dbsize()
#         info_server, info_memory, info_clients, key_count = await pipe.execute()
# 
#     redis_version = info_server.get('redis_version', '?')
#     mem_usage = info_memory.get('used_memory_human', '?')
#     clients_connected = info_clients.get('connected_clients', '?')
# 
#     log_func(
#         f'redis_version={redis_version} '
#         f'mem_usage={mem_usage} '
#         f'clients_connected={clients_connected} '
#         f'db_keys={key_count}'
#     )


# Path: tests/test_utils.py
import logging
import re
import pytest
import arq.typing
import arq.utils
from datetime import timedelta
from pydantic import BaseModel, validator
from redis.asyncio import ConnectionError, ResponseError
from arq.connections import RedisSettings, log_redis_info

    with pytest.raises(ConnectionError):
        await create_pool(RedisSettings(port=0, conn_retry_delay=0))
    assert arq.utils.asyncio.sleep.call_count == 5


@pytest.mark.skip(reason='this breaks many other tests as low level connections remain after failed connection')
async def test_redis_sentinel_failure(create_pool, cancel_remaining_task, mocker):
    settings = RedisSettings()
    settings.host = [('localhost', 6379), ('localhost', 6379)]
    settings.sentinel = True
    with pytest.raises(ResponseError, match='unknown command `SENTINEL`'):
        await create_pool(settings)


async def test_redis_success_log(caplog, create_pool):
    caplog.set_level(logging.INFO)
    settings = RedisSettings()
    pool = await create_pool(settings)
    assert 'redis connection successful' not in [r.message for r in caplog.records]
    await pool.close(close_connection_pool=True)

    pool = await create_pool(settings, retry=1)
    assert 'redis connection successful' in [r.message for r in caplog.records]
    await pool.close(close_connection_pool=True)


async def test_redis_log(create_pool):
    redis = await create_pool(RedisSettings())
    await redis.flushall()
    await redis.set(b'a', b'1')

    await redis.set(b'b', b'2')