# Path: aredis/connection.py
# class Connection(BaseConnection):
#     description = 'Connection<host={host},port={port},db={db}>'
# 
#     def __init__(self, host='127.0.0.1', port=6379, password=None,
#                  db=0, retry_on_timeout=False, stream_timeout=None, connect_timeout=None,
#                  ssl_context=None, parser_class=DefaultParser, reader_read_size=65535,
#                  encoding='utf-8', decode_responses=False, socket_keepalive=None,
#                  socket_keepalive_options=None, *, loop=None):
#         super(Connection, self).__init__(retry_on_timeout, stream_timeout,
#                                          parser_class, reader_read_size,
#                                          encoding, decode_responses,
#                                          loop=loop)
#         self.host = host
#         self.port = port
#         self.password = password
#         self.db = db
#         self.ssl_context = ssl_context
#         self._connect_timeout = connect_timeout
#         self._description_args = {
#             'host': self.host,
#             'port': self.port,
#             'db': self.db
#         }
#         self.socket_keepalive = socket_keepalive
#         self.socket_keepalive_options = socket_keepalive_options or {}
# 
#     async def _connect(self):
#         reader, writer = await exec_with_timeout(
#             asyncio.open_connection(host=self.host,
#                                     port=self.port,
#                                     ssl=self.ssl_context,
#                                     loop=self.loop),
#             self._connect_timeout,
#             loop=self.loop
#         )
#         self._reader = reader
#         self._writer = writer
#         sock = writer.transport.get_extra_info('socket')
#         if sock is not None:
#             sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
#             try:
#                 # TCP_KEEPALIVE
#                 if self.socket_keepalive:
#                     sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
#                     for k, v in self.socket_keepalive_options.items():
#                         sock.setsockopt(socket.SOL_TCP, k, v)
#             except (socket.error, TypeError):
#                 # `socket_keepalive_options` might contain invalid options
#                 # causing an error. Do not leave the connection open.
#                 writer.close()
#                 raise
#         await self.on_connect()
# 
# class UnixDomainSocketConnection(BaseConnection):
#     description = "UnixDomainSocketConnection<path={path},db={db}>"
# 
#     def __init__(self, path='', password=None,
#                  db=0, retry_on_timeout=False, stream_timeout=None, connect_timeout=None,
#                  ssl_context=None, parser_class=DefaultParser, reader_read_size=65535,
#                  encoding='utf-8', decode_responses=False, *, loop=None):
#         super(UnixDomainSocketConnection, self).__init__(retry_on_timeout, stream_timeout,
#                                                          parser_class, reader_read_size,
#                                                          encoding, decode_responses,
#                                                          loop=loop)
#         self.path = path
#         self.db = db
#         self.password = password
#         self.ssl_context = ssl_context
#         self._connect_timeout = connect_timeout
#         self._description_args = {
#             'path': self.path,
#             'db': self.db
#         }
# 
#     async def _connect(self):
#         reader, writer = await exec_with_timeout(
#             asyncio.open_unix_connection(path=self.path,
#                                          ssl=self.ssl_context,
#                                          loop=self.loop),
#             self._connect_timeout,
#             loop=self.loop
#         )
#         self._reader = reader
#         self._writer = writer
#         await self.on_connect()


import socket
import pytest
from aredis import (Connection,
                    UnixDomainSocketConnection)
#!/usr/bin/python
# -*- coding: utf-8 -*-



@pytest.mark.asyncio(forbid_global_loop=True)
async def test_connect_tcp(event_loop):
    conn = Connection(loop=event_loop)
    assert conn.host == '127.0.0.1'
    assert conn.port == 6379
    assert str(conn) == 'Connection<host=127.0.0.1,port=6379,db=0>'
    await conn.send_command('PING')
    res = await conn.read_response()
    assert res == b'PONG'
    assert (conn._reader is not None) and (conn._writer is not None)
    conn.disconnect()
    assert (conn._reader is None) and (conn._writer is None)


@pytest.mark.asyncio(forbid_global_loop=True)
async def test_connect_tcp_keepalive_options(event_loop):

# Path: aredis/connection.py
# class Connection(BaseConnection):
#     description = 'Connection<host={host},port={port},db={db}>'
# 
#     def __init__(self, host='127.0.0.1', port=6379, password=None,
#                  db=0, retry_on_timeout=False, stream_timeout=None, connect_timeout=None,
#                  ssl_context=None, parser_class=DefaultParser, reader_read_size=65535,
#                  encoding='utf-8', decode_responses=False, socket_keepalive=None,
#                  socket_keepalive_options=None, *, loop=None):
#         super(Connection, self).__init__(retry_on_timeout, stream_timeout,
#                                          parser_class, reader_read_size,
#                                          encoding, decode_responses,
#                                          loop=loop)
#         self.host = host
#         self.port = port
#         self.password = password
#         self.db = db
#         self.ssl_context = ssl_context
#         self._connect_timeout = connect_timeout
#         self._description_args = {
#             'host': self.host,
#             'port': self.port,
#             'db': self.db
#         }
#         self.socket_keepalive = socket_keepalive
#         self.socket_keepalive_options = socket_keepalive_options or {}
# 
#     async def _connect(self):
#         reader, writer = await exec_with_timeout(
#             asyncio.open_connection(host=self.host,
#                                     port=self.port,
#                                     ssl=self.ssl_context,
#                                     loop=self.loop),
#             self._connect_timeout,
#             loop=self.loop
#         )
#         self._reader = reader
#         self._writer = writer
#         sock = writer.transport.get_extra_info('socket')
#         if sock is not None:
#             sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
#             try:
#                 # TCP_KEEPALIVE
#                 if self.socket_keepalive:
#                     sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
#                     for k, v in self.socket_keepalive_options.items():
#                         sock.setsockopt(socket.SOL_TCP, k, v)
#             except (socket.error, TypeError):
#                 # `socket_keepalive_options` might contain invalid options
#                 # causing an error. Do not leave the connection open.
#                 writer.close()
#                 raise
#         await self.on_connect()
# 
# class UnixDomainSocketConnection(BaseConnection):
#     description = "UnixDomainSocketConnection<path={path},db={db}>"
# 
#     def __init__(self, path='', password=None,
#                  db=0, retry_on_timeout=False, stream_timeout=None, connect_timeout=None,
#                  ssl_context=None, parser_class=DefaultParser, reader_read_size=65535,
#                  encoding='utf-8', decode_responses=False, *, loop=None):
#         super(UnixDomainSocketConnection, self).__init__(retry_on_timeout, stream_timeout,
#                                                          parser_class, reader_read_size,
#                                                          encoding, decode_responses,
#                                                          loop=loop)
#         self.path = path
#         self.db = db
#         self.password = password
#         self.ssl_context = ssl_context
#         self._connect_timeout = connect_timeout
#         self._description_args = {
#             'path': self.path,
#             'db': self.db
#         }
# 
#     async def _connect(self):
#         reader, writer = await exec_with_timeout(
#             asyncio.open_unix_connection(path=self.path,
#                                          ssl=self.ssl_context,
#                                          loop=self.loop),
#             self._connect_timeout,
#             loop=self.loop
#         )
#         self._reader = reader
#         self._writer = writer
#         await self.on_connect()


# Path: tests/client/test_connection.py
import socket
import pytest
from aredis import (Connection,
                    UnixDomainSocketConnection)

#!/usr/bin/python
# -*- coding: utf-8 -*-



@pytest.mark.asyncio(forbid_global_loop=True)
async def test_connect_tcp(event_loop):
    conn = Connection(loop=event_loop)
    assert conn.host == '127.0.0.1'
    assert conn.port == 6379
    assert str(conn) == 'Connection<host=127.0.0.1,port=6379,db=0>'
    await conn.send_command('PING')
    res = await conn.read_response()
    assert res == b'PONG'
    assert (conn._reader is not None) and (conn._writer is not None)
    conn.disconnect()
    assert (conn._reader is None) and (conn._writer is None)


@pytest.mark.asyncio(forbid_global_loop=True)
async def test_connect_tcp_keepalive_options(event_loop):

    conn = Connection(