<s> import pytest <EOL> import socket <EOL> from aiohttp . parsers import StreamWriter , CORK <EOL> from unittest import mock <EOL> def test_nodelay_default ( loop ) : <EOL> transport = mock . Mock ( ) <EOL> s = socket . socket ( socket . AF_INET , socket . SOCK_STREAM ) <EOL> transport . get_extra_info . return_value = s <EOL> proto = mock . Mock ( ) <EOL> reader = mock . Mock ( ) <EOL> writer = StreamWriter ( transport , proto , reader , loop ) <EOL> assert not writer . tcp_nodelay <EOL> assert not s . getsockopt ( socket . IPPROTO_TCP , socket . TCP_NODELAY ) <EOL> def test_set_nodelay_no_change ( loop ) : <EOL> transport = mock . Mock ( ) <EOL> s = socket . socket ( socket . AF_INET , socket . SOCK_STREAM ) <EOL> transport . get_extra_info . return_value = s <EOL> proto = mock . Mock ( ) <EOL> reader = mock . Mock ( ) <EOL> writer = StreamWriter ( transport , proto , reader , loop ) <EOL> writer . set_tcp_nodelay ( False ) <EOL> assert not writer . tcp_nodelay <EOL> assert not s . getsockopt ( socket . IPPROTO_TCP , socket . TCP_NODELAY ) <EOL> def test_set_nodelay_enable ( loop ) : <EOL> transport = mock . Mock ( ) <EOL> s = socket . socket ( socket . AF_INET , socket . SOCK_STREAM ) <EOL> transport . get_extra_info . return_value = s <EOL> proto = mock . Mock ( ) <EOL> reader = mock . Mock ( ) <EOL> writer = StreamWriter ( transport , proto , reader , loop ) <EOL> writer . set_tcp_nodelay ( True ) <EOL> assert writer . tcp_nodelay <EOL> assert s . getsockopt ( socket . IPPROTO_TCP , socket . TCP_NODELAY ) <EOL> def test_set_nodelay_enable_and_disable ( loop ) : <EOL> transport = mock . Mock ( ) <EOL> s = socket . socket ( socket . AF_INET , socket . SOCK_STREAM ) <EOL> transport . get_extra_info . return_value = s <EOL> proto = mock . Mock ( ) <EOL> reader = mock . Mock ( ) <EOL> writer = StreamWriter ( transport , proto , reader , loop ) <EOL> writer . set_tcp_nodelay ( True ) <EOL> writer . set_tcp_nodelay ( False ) <EOL> assert not writer . tcp_nodelay <EOL> assert not s . getsockopt ( socket . IPPROTO_TCP , socket . TCP_NODELAY ) <EOL> def test_set_nodelay_enable_ipv6 ( loop ) : <EOL> transport = mock . Mock ( ) <EOL> s = socket . socket ( socket . AF_INET6 , socket . SOCK_STREAM ) <EOL> transport . get_extra_info . return_value = s <EOL> proto = mock . Mock ( ) <EOL> reader = mock . Mock ( ) <EOL> writer = StreamWriter ( transport , proto , reader , loop ) <EOL> writer . set_tcp_nodelay ( True ) <EOL> assert writer . tcp_nodelay <EOL> assert s . getsockopt ( socket . IPPROTO_TCP , socket . TCP_NODELAY ) <EOL> @ pytest . mark . skipif ( not hasattr ( socket , '<STR_LIT>' ) , <EOL> reason = "<STR_LIT>" ) <EOL> def test_set_nodelay_enable_unix ( loop ) : <EOL> transport = mock . Mock ( ) <EOL> s = socket . socket ( socket . AF_UNIX , socket . SOCK_STREAM ) <EOL> transport . get_extra_info . return_value = s <EOL> proto = mock . Mock ( ) <EOL> reader = mock . Mock ( ) <EOL> writer = StreamWriter ( transport , proto , reader , loop ) <EOL> writer . set_tcp_nodelay ( True ) <EOL> assert writer . tcp_nodelay <EOL> def test_set_nodelay_enable_no_socket ( loop ) : <EOL> transport = mock . Mock ( ) <EOL> transport . get_extra_info . return_value = None <EOL> proto = mock . Mock ( ) <EOL> reader = mock . Mock ( ) <EOL> writer = StreamWriter ( transport , proto , reader , loop ) <EOL> writer . set_tcp_nodelay ( True ) <EOL> assert writer . tcp_nodelay <EOL> assert writer . _socket is None <EOL> @ pytest . mark . skipif ( CORK is None , reason = "<STR_LIT>" ) <EOL> def test_cork_default ( loop ) : <EOL> transport = mock . Mock ( ) <EOL> s = socket . socket ( socket . AF_INET , socket . SOCK_STREAM ) <EOL> transport . get_extra_info . return_value = s <EOL> proto = mock . Mock ( ) <EOL> reader = mock . Mock ( ) <EOL> writer = StreamWriter ( transport , proto , reader , loop ) <EOL> assert not writer . tcp_cork <EOL> assert not s . getsockopt ( socket . IPPROTO_TCP , CORK ) <EOL> @ pytest . mark . skipif ( CORK is None , reason = "<STR_LIT>" ) <EOL> def test_set_cork_no_change ( loop ) : <EOL> transport = mock . Mock ( ) <EOL> s = socket . socket ( socket . AF_INET , socket . SOCK_STREAM ) <EOL> transport . get_extra_info . return_value = s <EOL> proto = mock . Mock ( ) <EOL> reader = mock . Mock ( ) <EOL> writer = StreamWriter ( transport , proto , reader , loop ) <EOL> writer . set_tcp_cork ( False ) <EOL> assert not writer . tcp_cork <EOL> assert not s . getsockopt ( socket . IPPROTO_TCP , CORK ) <EOL> @ pytest . mark . skipif ( CORK is None , reason = "<STR_LIT>" ) <EOL> def test_set_cork_enable ( loop ) : <EOL> transport = mock . Mock ( ) <EOL> s = socket . socket ( socket . AF_INET , socket . SOCK_STREAM ) <EOL> transport . get_extra_info . return_value = s <EOL> proto = mock . Mock ( ) <EOL> reader = mock . Mock ( ) <EOL> writer = StreamWriter ( transport , proto , reader , loop ) <EOL> writer . set_tcp_cork ( True ) <EOL> assert writer . tcp_cork <EOL> assert s . getsockopt ( socket . IPPROTO_TCP , CORK ) <EOL> @ pytest . mark . skipif ( CORK is None , reason = "<STR_LIT>" ) <EOL> def test_set_cork_enable_and_disable ( loop ) : <EOL> transport = mock . Mock ( ) <EOL> s = socket . socket ( socket . AF_INET , socket . SOCK_STREAM ) <EOL> transport . get_extra_info . return_value = s <EOL> proto = mock . Mock ( ) <EOL> reader = mock . Mock ( ) <EOL> writer = StreamWriter ( transport , proto , reader , loop ) <EOL> writer . set_tcp_cork ( True ) <EOL> writer . set_tcp_cork ( False ) <EOL> assert not writer . tcp_cork <EOL> assert not s . getsockopt ( socket . IPPROTO_TCP , CORK ) <EOL> @ pytest . mark . skipif ( CORK is None , reason = "<STR_LIT>" ) <EOL> def test_set_cork_enable_ipv6 ( loop ) : <EOL> transport = mock . Mock ( ) <EOL> s = socket . socket ( socket . AF_INET6 , socket . SOCK_STREAM ) <EOL> transport . get_extra_info . return_value = s <EOL> proto = mock . Mock ( ) <EOL> reader = mock . Mock ( ) <EOL> writer = StreamWriter ( transport , proto , reader , loop ) <EOL> writer . set_tcp_cork ( True ) <EOL> assert writer . tcp_cork <EOL> assert s . getsockopt ( socket . IPPROTO_TCP , CORK ) <EOL> @ pytest . mark . skipif ( not hasattr ( socket , '<STR_LIT>' ) , <EOL> reason = "<STR_LIT>" ) <EOL> @ pytest . mark . skipif ( CORK is None , reason = "<STR_LIT>" ) <EOL> def test_set_cork_enable_unix ( loop ) : <EOL> transport = mock . Mock ( ) <EOL> s = socket . socket ( socket . AF_UNIX , socket . SOCK_STREAM ) <EOL> transport . get_extra_info . return_value = s <EOL> proto = mock . Mock ( ) <EOL> reader = mock . Mock ( ) <EOL> writer = StreamWriter ( transport , proto , reader , loop ) <EOL> writer . set_tcp_cork ( True ) <EOL> assert writer . tcp_cork <EOL> @ pytest . mark . skipif ( CORK is None , reason = "<STR_LIT>" ) <EOL> def test_set_cork_enable_no_socket ( loop ) : <EOL> transport = mock . Mock ( ) <EOL> transport . get_extra_info . return_value = None <EOL> proto = mock . Mock ( ) <EOL> reader = mock . Mock ( ) <EOL> writer = StreamWriter ( transport , proto , reader , loop ) <EOL> writer . set_tcp_cork ( True ) <EOL> assert writer . tcp_cork <EOL> assert writer . _socket is None <EOL> @ pytest . mark . skipif ( CORK is None , reason = "<STR_LIT>" ) <EOL> def test_set_enabling_cork_disables_nodelay ( loop ) : <EOL> transport = mock . Mock ( ) <EOL> s = socket . socket ( socket . AF_INET , socket . SOCK_STREAM ) <EOL> transport . get_extra_info . return_value = s <EOL> proto = mock . Mock ( ) <EOL> reader = mock . Mock ( ) <EOL> writer = StreamWriter ( transport , proto , reader , loop ) <EOL> writer . set_tcp_nodelay ( True ) <EOL> writer . set_tcp_cork ( True ) <EOL> assert not writer . tcp_nodelay <EOL> assert not s . getsockopt ( socket . IPPROTO_TCP , socket . TCP_NODELAY ) <EOL> assert writer . tcp_cork <EOL> assert s . getsockopt ( socket . IPPROTO_TCP , CORK ) <EOL> @ pytest . mark . skipif ( CORK is None , reason = "<STR_LIT>" ) <EOL> def test_set_enabling_nodelay_disables_cork ( loop ) : <EOL> transport = mock . Mock ( ) <EOL> s = socket . socket ( socket . AF_INET , socket . SOCK_STREAM ) <EOL> transport . get_extra_info . return_value = s <EOL> proto = mock . Mock ( ) <EOL> reader = mock . Mock ( ) <EOL> writer = StreamWriter ( transport , proto , reader , loop ) <EOL> writer . set_tcp_cork ( True ) <EOL> writer . set_tcp_nodelay ( True ) <EOL> assert writer . tcp_nodelay <EOL> assert s . getsockopt ( socket . IPPROTO_TCP , socket . TCP_NODELAY ) <EOL> assert not writer . tcp_cork <EOL> assert not s . getsockopt ( socket . IPPROTO_TCP , CORK ) </s>
