# 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

def test_args_to_string():
    assert arq.utils.args_to_string((), {'d': 4}) == 'd=4'
    assert arq.utils.args_to_string((1, 2, 3), {}) == '1, 2, 3'
    assert arq.utils.args_to_string((1, 2, 3), {'d': 4}) == '1, 2, 3, d=4'


@pytest.mark.parametrize(
    'input,output', [(timedelta(days=1), 86_400_000), (42, 42000), (42.123, 42123), (42.123_987, 42124), (None, None)]
)
def test_to_ms(input, output):
    assert arq.utils.to_ms(input) == output


@pytest.mark.parametrize('input,output', [(timedelta(days=1), 86400), (42, 42), (42.123, 42.123), (None, None)])
def test_to_seconds(input, output):
    assert arq.utils.to_seconds(input) == output


def test_typing():
    assert 'OptionType' in arq.typing.__all__


def test_redis_settings_validation():
    class Settings(BaseModel):
        redis_settings: RedisSettings

        @validator('redis_settings', always=True, pre=True)
        def parse_redis_settings(cls, v):
            if isinstance(v, str):

# 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


def test_args_to_string():
    assert arq.utils.args_to_string((), {'d': 4}) == 'd=4'
    assert arq.utils.args_to_string((1, 2, 3), {}) == '1, 2, 3'
    assert arq.utils.args_to_string((1, 2, 3), {'d': 4}) == '1, 2, 3, d=4'


@pytest.mark.parametrize(
    'input,output', [(timedelta(days=1), 86_400_000), (42, 42000), (42.123, 42123), (42.123_987, 42124), (None, None)]
)
def test_to_ms(input, output):
    assert arq.utils.to_ms(input) == output


@pytest.mark.parametrize('input,output', [(timedelta(days=1), 86400), (42, 42), (42.123, 42.123), (None, None)])
def test_to_seconds(input, output):
    assert arq.utils.to_seconds(input) == output


def test_typing():
    assert 'OptionType' in arq.typing.__all__


def test_redis_settings_validation():
    class Settings(BaseModel):
        redis_settings: RedisSettings

        @validator('redis_settings', always=True, pre=True)
        def parse_redis_settings(cls, v):
            if isinstance(v, str):

                return RedisSettings.from_dsn(v)