import sys import unittest import pytest from socketio import asyncio_redis_manager @unittest.skipIf(sys.version_info < (3, 5), 'only for Python 3.5+') class TestAsyncRedisManager(unittest.TestCase): def test_default_url(self): assert asyncio_redis_manager._parse_redis_url('redis://') == ( 'localhost', 6379, None, 0, False, ) def test_only_host_url(self): assert asyncio_redis_manager._parse_redis_url( 'redis://redis.host' ) == ('redis.host', 6379, None, 0, False) def test_no_db_url(self): assert asyncio_redis_manager._parse_redis_url( 'redis://redis.host:123/1' ) == ('redis.host', 123, None, 1, False) def test_no_port_url(self): assert asyncio_redis_manager._parse_redis_url( 'redis://redis.host/1' ) == ('redis.host', 6379, None, 1, False) def test_password(self): assert asyncio_redis_manager._parse_redis_url( 'redis://:pw@redis.host/1' ) == ('redis.host', 6379, 'pw', 1, False) def test_no_host_url(self): assert asyncio_redis_manager._parse_redis_url('redis://:123/1') == ( 'localhost', 123, None, 1, False, ) def test_no_host_password_url(self): assert asyncio_redis_manager._parse_redis_url( 'redis://:pw@:123/1' ) == ('localhost', 123, 'pw', 1, False) def test_bad_port_url(self): with pytest.raises(ValueError): asyncio_redis_manager._parse_redis_url('redis://localhost:abc/1') def test_bad_db_url(self): with pytest.raises(ValueError): asyncio_redis_manager._parse_redis_url('redis://localhost:abc/z') def test_bad_scheme_url(self): with pytest.raises(ValueError): asyncio_redis_manager._parse_redis_url('http://redis.host:123/1') def test_ssl_scheme(self): assert asyncio_redis_manager._parse_redis_url('rediss://') == ( 'localhost', 6379, None, 0, True, )