# -*- coding: utf-8 -*-
import os
import socket
import pytest
from rpython.tool.udir import udir
from rpython.rlib import rsocket
from rpython.rtyper.lltypesystem import lltype, rffi

@pytest.fixture
def spaceconfig():
    return {'usemodules': ['_socket', 'array', 'struct', 'unicodedata']}

@pytest.fixture
def w_socket(space):
    return space.appexec([], "(): import _socket as m; return m")

def test_gethostname(space, w_socket):
    host = space.appexec([w_socket], "(_socket): return _socket.gethostname()")
    assert space.unwrap(host) == socket.gethostname()

def test_gethostbyname(space, w_socket):
    for host in ["localhost", "127.0.0.1"]:
        ip = space.appexec([w_socket, space.wrap(host)],
                           "(_socket, host): return _socket.gethostbyname(host)")
        assert space.unwrap(ip) == socket.gethostbyname(host)

def test_gethostbyname_ex(space, w_socket):
    for host in ["localhost", "127.0.0.1"]:
        ip = space.appexec([w_socket, space.wrap(host)],
                           "(_socket, host): return _socket.gethostbyname_ex(host)")
        assert space.unwrap(ip) == socket.gethostbyname_ex(host)

def test_gethostbyaddr(space, w_socket):
    try:
        socket.gethostbyaddr("::1")
    except socket.herror:
        ipv6 = False
    else:
        ipv6 = True
    for host in ["localhost", "127.0.0.1", "::1"]:
        if host == "::1" and not ipv6:
            from pypy.interpreter.error import OperationError
            with pytest.raises(OperationError):
                space.appexec([w_socket, space.wrap(host)],
                              "(_socket, host): return _socket.gethostbyaddr(host)")
            continue
        ip = space.appexec([w_socket, space.wrap(host)],
                           "(_socket, host): return _socket.gethostbyaddr(host)")
        assert space.unwrap(ip) == socket.gethostbyaddr(host)

def test_getservbyname(space, w_socket):
    name = "smtp"
    # 2 args version
    port = space.appexec([w_socket, space.wrap(name)],
                        "(_socket, name): return _socket.getservbyname(name, 'tcp')")
    assert space.unwrap(port) == 25
    # 1 arg version
    port = space.appexec([w_socket, space.wrap(name)],
                        "(_socket, name): return _socket.getservbyname(name)")
    assert space.unwrap(port) == 25

def test_getservbyport(space, w_socket):
    port = 25
    # 2 args version
    name = space.appexec([w_socket, space.wrap(port)],
                         "(_socket, port): return _socket.getservbyport(port, 'tcp')")
    assert space.unwrap(name) == "smtp"
    name = space.appexec([w_socket, space.wrap(port)],
                         """(_socket, port):
                         try:
                             return _socket.getservbyport(port, 42)
                         except TypeError:
                             return 'OK'
                         """)
    assert space.unwrap(name) == 'OK'
    # 1 arg version
    name = space.appexec([w_socket, space.wrap(port)],
                         "(_socket, port): return _socket.getservbyport(port)")
    assert space.unwrap(name) == "smtp"

def test_getprotobyname(space, w_socket):
    name = "tcp"
    w_n = space.appexec([w_socket, space.wrap(name)],
                        "(_socket, name): return _socket.getprotobyname(name)")
    assert space.unwrap(w_n) == socket.IPPROTO_TCP

@pytest.mark.skipif("not hasattr(socket, 'fromfd')")
@pytest.mark.skipif("sys.platform=='win32'")
def test_ntohs(space, w_socket):
    w_n = space.appexec([w_socket, space.wrap(125)],
                        "(_socket, x): return _socket.ntohs(x)")
    assert space.unwrap(w_n) == socket.ntohs(125)

def test_ntohl(space, w_socket):
    w_n = space.appexec([w_socket, space.wrap(125)],
                        "(_socket, x): return _socket.ntohl(x)")
    assert space.unwrap(w_n) == socket.ntohl(125)
    w_n = space.appexec([w_socket, space.wrap(0x89abcdef)],
                        "(_socket, x): return _socket.ntohl(x)")
    assert space.unwrap(w_n) in (0x89abcdef, 0xefcdab89)
    space.raises_w(space.w_OverflowError, space.appexec,
                   [w_socket, space.wrap(1 << 32)],
                   "(_socket, x): return _socket.ntohl(x)")

def test_htons(space, w_socket):
    w_n = space.appexec([w_socket, space.wrap(125)],
                        "(_socket, x): return _socket.htons(x)")
    assert space.unwrap(w_n) == socket.htons(125)

def test_htonl(space, w_socket):
    w_n = space.appexec([w_socket, space.wrap(125)],
                        "(_socket, x): return _socket.htonl(x)")
    assert space.unwrap(w_n) == socket.htonl(125)
    w_n = space.appexec([w_socket, space.wrap(0x89abcdef)],
                        "(_socket, x): return _socket.htonl(x)")
    assert space.unwrap(w_n) in (0x89abcdef, 0xefcdab89)
    space.raises_w(space.w_OverflowError, space.appexec,
                   [w_socket, space.wrap(1 << 32)],
                   "(_socket, x): return _socket.htonl(x)")

def test_aton_ntoa(space, w_socket):
    ip = '123.45.67.89'
    packed = socket.inet_aton(ip)
    w_p = space.appexec([w_socket, space.wrap(ip)],
                        "(_socket, ip): return _socket.inet_aton(ip)")
    assert space.bytes_w(w_p) == packed
    w_ip = space.appexec([w_socket, w_p],
                         "(_socket, p): return _socket.inet_ntoa(p)")
    assert space.utf8_w(w_ip) == ip

@pytest.mark.skipif("not hasattr(socket, 'inet_pton')")
def test_pton_ntop_ipv4(space, w_socket):
    tests = [
        ("123.45.67.89", "\x7b\x2d\x43\x59"),
        ("0.0.0.0", "\x00" * 4),
        ("255.255.255.255", "\xff" * 4),
    ]
    for ip, packed in tests:
        w_p = space.appexec([w_socket, space.wrap(ip)], """(_socket, ip):
            return _socket.inet_pton(_socket.AF_INET, ip)""")
        assert space.unwrap(w_p) == packed
        w_ip = space.appexec([w_socket, w_p], """(_socket, p):
            return _socket.inet_ntop(_socket.AF_INET, p)""")
        assert space.unwrap(w_ip) == ip

def test_ntop_ipv6(space, w_socket):
    if not hasattr(socket, 'inet_pton'):
        pytest.skip('No socket.inet_pton on this platform')
    if not socket.has_ipv6:
        pytest.skip("No IPv6 on this platform")
    tests = [
        (b"\x00" * 16, "::"),
        (b"\x01" * 16, ":".join(["101"] * 8)),
        (b"\x00\x00\x10\x10" * 4, None),  # "::1010:" + ":".join(["0:1010"] * 3)),
        (b"\x00" * 12 + "\x01\x02\x03\x04", "::1.2.3.4"),
        (b"\x00" * 10 + "\xff\xff\x01\x02\x03\x04", "::ffff:1.2.3.4"),
    ]
    for packed, ip in tests:
        w_ip = space.appexec([w_socket, space.newbytes(packed)],
            "(_socket, packed): return _socket.inet_ntop(_socket.AF_INET6, packed)")
        if ip is not None:   # else don't check for the precise representation
            assert space.unwrap(w_ip) == ip
        w_packed = space.appexec([w_socket, w_ip],
            "(_socket, ip): return _socket.inet_pton(_socket.AF_INET6, ip)")
        assert space.unwrap(w_packed) == packed

def test_pton_ipv6(space, w_socket):
    import sys
    if not hasattr(socket, 'inet_pton'):
        pytest.skip('No socket.inet_pton on this platform')
    if not socket.has_ipv6:
        pytest.skip("No IPv6 on this platform")
    tests = [
        ("\x00" * 16, "::"),
        ("\x01" * 16, ":".join(["101"] * 8)),
        ("\x00\x01" + "\x00" * 12 + "\x00\x02", "1::2"),
        ("\x00" * 4 + "\x00\x01" * 6, "::1:1:1:1:1:1"),
        ("\x00\x01" * 6 + "\x00" * 4, "1:1:1:1:1:1::"),
        ("\xab\xcd\xef\00" + "\x00" * 12, "ABCD:EF00::"),
        ("\xab\xcd\xef\00" + "\x00" * 12, "abcd:ef00::"),
        ("\x00" * 12 + "\x01\x02\x03\x04", "::1.2.3.4"),
        ("\x00" * 10 + "\xff\xff\x01\x02\x03\x04", "::ffff:1.2.3.4"),
    ]
    if sys.platform != 'win32':
        tests.append(
            ("\x00\x00\x10\x10" * 4, "::1010:" + ":".join(["0:1010"] * 3))
        )
    for packed, ip in tests:
        w_packed = space.appexec([w_socket, space.wrap(ip)],
            "(_socket, ip): return _socket.inet_pton(_socket.AF_INET6, ip)")
        assert space.unwrap(w_packed) == packed

def test_getaddrinfo(space, w_socket):
    host = b"localhost"
    port = 25
    info = socket.getaddrinfo(host, port)
    w_l = space.appexec([w_socket, space.newbytes(host), space.wrap(port)],
                        "(_socket, host, port): return _socket.getaddrinfo(host, port)")
    assert space.unwrap(w_l) == info
    w_l = space.appexec([w_socket, space.wrap(host), space.wrap(port)],
                        "(_socket, host, port): return _socket.getaddrinfo(host, port)")
    assert space.unwrap(w_l) == info
    w_l = space.appexec([w_socket, space.newbytes(host), space.wrap('smtp')],
                        "(_socket, host, port): return _socket.getaddrinfo(host, port)")
    assert space.unwrap(w_l) == socket.getaddrinfo(host, 'smtp')
    w_l = space.appexec([w_socket, space.newbytes(host), space.wrap(u'\uD800')], '''

       (_socket, host, port):
            try:
                info = _socket.getaddrinfo(host, port)
            except Exception as e:
                return e.reason == 'surrogates not allowed'
            return -1
        ''')
    assert space.unwrap(w_l) == True

def test_getaddrinfo_ipv6(space, w_socket):
    host = 'fe80::1%1'
    port = 80
    w_l = space.appexec([w_socket, space.newtext(host), space.newint(port)],
                        "(_socket, host, port): return _socket.getaddrinfo(host, port)")
    w_tup = space.getitem(w_l, space.newint(0))
    w_tup2 = space.getitem(w_tup, space.newint(4))
    w_canon = space.getitem(w_tup2, space.newint(0))
    canon_name = space.text_w(w_canon)
    # Make sure the scope ID (the `%1` part) is removed (issue 3628, 3938)
    assert '%' not in canon_name

@pytest.mark.skipif("not hasattr(socket, 'sethostname')")
def test_sethostname(space, w_socket):
    space.raises_w(space.w_OSError, space.appexec,
                   [w_socket],
                   "(_socket): _socket.sethostname(_socket.gethostname())")


@pytest.mark.skipif("not hasattr(socket, 'sethostname')")
def test_sethostname_bytes(space, w_socket):
    space.raises_w(space.w_OSError, space.appexec,
                   [w_socket],
                   "(_socket): _socket.sethostname(_socket.gethostname().encode())")


def test_unknown_addr_as_object(space, ):
    from pypy.module._socket.interp_socket import addr_as_object
    c_addr = lltype.malloc(rsocket._c.sockaddr, flavor='raw', track_allocation=False)
    c_addr.c_sa_data[0] = 'c'
    rffi.setintfield(c_addr, 'c_sa_family', 15)
    # XXX what size to pass here? for the purpose of this test it has
    #     to be short enough so we have some data, 1 sounds good enough
    #     + sizeof USHORT
    w_obj = addr_as_object(rsocket.Address(c_addr, 1 + 2), -1, space)
    assert space.isinstance_w(w_obj, space.w_tuple)
    assert space.int_w(space.getitem(w_obj, space.wrap(0))) == 15
    assert space.text_w(space.getitem(w_obj, space.wrap(1))) == 'c'

def test_addr_raw_packet(space, ):
    from pypy.module._socket.interp_socket import addr_as_object
    if not hasattr(rsocket._c, 'sockaddr_ll'):
        pytest.skip("posix specific test")
    # HACK: To get the correct interface number of lo, which in most cases is 1,
    # but can be anything (i.e. 39), we need to call the libc function
    # if_nametoindex to get the correct index
    import ctypes
    libc = ctypes.CDLL(ctypes.util.find_library('c'))
    ifnum = libc.if_nametoindex('lo')

    c_addr_ll = lltype.malloc(rsocket._c.sockaddr_ll, flavor='raw')
    addrlen = rffi.sizeof(rsocket._c.sockaddr_ll)
    c_addr = rffi.cast(lltype.Ptr(rsocket._c.sockaddr), c_addr_ll)
    rffi.setintfield(c_addr_ll, 'c_sll_ifindex', ifnum)
    rffi.setintfield(c_addr_ll, 'c_sll_protocol', 8)
    rffi.setintfield(c_addr_ll, 'c_sll_pkttype', 13)
    rffi.setintfield(c_addr_ll, 'c_sll_hatype', 0)
    rffi.setintfield(c_addr_ll, 'c_sll_halen', 3)
    c_addr_ll.c_sll_addr[0] = 'a'
    c_addr_ll.c_sll_addr[1] = 'b'
    c_addr_ll.c_sll_addr[2] = 'c'
    rffi.setintfield(c_addr, 'c_sa_family', socket.AF_PACKET)
    # fd needs to be somehow valid
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    fd = s.fileno()
    w_obj = addr_as_object(rsocket.make_address(c_addr, addrlen), fd, space)
    lltype.free(c_addr_ll, flavor='raw')
    assert space.is_true(space.eq(w_obj, space.newtuple([
        space.newtext('lo'),
        space.newint(socket.ntohs(8)),
        space.newint(13),
        space.newbool(False),
        space.newbytes("abc"),
    ])))

def test_getnameinfo(space, w_socket):
    from pypy.module._socket.interp_socket import get_error
    host = "127.0.0.1"
    port = 25
    info = socket.getnameinfo((host, port), 0)
    w_l = space.appexec([w_socket, space.wrap(host), space.wrap(port)],
                        "(_socket, host, port): return _socket.getnameinfo((host, port), 0)")
    assert space.unwrap(w_l) == info
    sockaddr = space.newtuple([space.wrap('mail.python.org'), space.wrap(0)])
    space.raises_w(get_error(space, 'error'), space.appexec,
                   [w_socket, sockaddr, space.wrap(0)],
                   "(_socket, sockaddr, flags): return _socket.getnameinfo(sockaddr, flags)")
    if socket.has_ipv6:
        sockaddr = space.newtuple([space.wrap('::1'), space.wrap(0),
                                   space.wrap(0xffffffff)])
        space.raises_w(space.w_OverflowError, space.appexec,
                       [w_socket, sockaddr, space.wrap(0)],
                       "(_socket, sockaddr, flags): return _socket.getnameinfo(sockaddr, flags)")

def test_timeout(space, w_socket):
    space.appexec([w_socket, space.wrap(25.4)],
                  "(_socket, timeout): _socket.setdefaulttimeout(timeout)")
    w_t = space.appexec([w_socket],
                  "(_socket): return _socket.getdefaulttimeout()")
    assert space.unwrap(w_t) == 25.4

    space.appexec([w_socket, space.w_None],
                  "(_socket, timeout): _socket.setdefaulttimeout(timeout)")
    w_t = space.appexec([w_socket],
                  "(_socket): return _socket.getdefaulttimeout()")
    assert space.unwrap(w_t) is None


# XXX also need tests for other connection and timeout errors

def test_type(space, w_socket):
    w_bool = space.appexec([w_socket],
                    """(_socket,):
                    if not hasattr(_socket, 'SOCK_CLOEXEC'):
                        return -1
                    s = _socket.socket(_socket.AF_INET,
                                    _socket.SOCK_STREAM | _socket.SOCK_CLOEXEC)
                    return s.type == _socket.SOCK_STREAM
                    """)
    assert(space.bool_w(w_bool))

class AppTestSocket:
    spaceconfig = dict(usemodules=['_socket', '_weakref', 'struct', 'select',
                                   'unicodedata'])

    def setup_class(cls):
        space = cls.space
        cls.w_udir = space.wrap(str(udir))

    def teardown_class(cls):
        if not cls.runappdirect:
            cls.space.sys.getmodule('_socket').shutdown(cls.space)

    def test_module(self):
        import _socket
        assert _socket.socket.__name__ == 'socket'
        assert _socket.socket.__module__ == '_socket'

    def test_overflow_errors(self):
        import _socket
        raises(OverflowError, _socket.getservbyport, -1)
        raises(OverflowError, _socket.getservbyport, 65536)

    def test_ntoa_exception(self):
        import _socket
        raises(_socket.error, _socket.inet_ntoa, b"ab")

    def test_aton_exceptions(self):
        import _socket
        tests = ["127.0.0.256", "127.0.0.255555555555555555", "127.2b.0.0",
            "127.2.0.0.1", "127.2.0."]
        for ip in tests:
            raises(_socket.error, _socket.inet_aton, ip)

    def test_ntop_exceptions(self):
        import _socket
        if not hasattr(_socket, 'inet_ntop'):
            skip('No socket.inet_pton on this platform')
        for family, packed, exception in \
                    [(_socket.AF_INET + _socket.AF_INET6, b"", _socket.error),
                     (_socket.AF_INET, b"a", ValueError),
                     (_socket.AF_INET6, b"a", ValueError),
                     (_socket.AF_INET, "aa\u2222a", TypeError)]:
            raises(exception, _socket.inet_ntop, family, packed)

    def test_pton_exceptions(self):
        import _socket
        if not hasattr(_socket, 'inet_pton'):
            skip('No socket.inet_pton on this platform')
        tests = [
            (_socket.AF_INET + _socket.AF_INET6, ""),
            (_socket.AF_INET, "127.0.0.256"),
            (_socket.AF_INET, "127.0.0.255555555555555555"),
            (_socket.AF_INET, "127.2b.0.0"),
            (_socket.AF_INET, "127.2.0.0.1"),
            (_socket.AF_INET, "127.2..0"),
            (_socket.AF_INET6, "127.0.0.1"),
            (_socket.AF_INET6, "1::2::3"),
            (_socket.AF_INET6, "1:1:1:1:1:1:1:1:1"),
            (_socket.AF_INET6, "1:1:1:1:1:1:1:1::"),
            (_socket.AF_INET6, "1:1:1::1:1:1:1:1"),
            (_socket.AF_INET6, "1::22222:1"),
            (_socket.AF_INET6, "1::eg"),
        ]
        for family, ip in tests:
            raises(_socket.error, _socket.inet_pton, family, ip)

    def test_newsocket_error(self):
        import _socket
        raises(_socket.error, _socket.socket, 10001, _socket.SOCK_STREAM, 0)

    def test_socket_fileno(self):
        import _socket
        s = _socket.socket(_socket.AF_INET, _socket.SOCK_STREAM, 0)
        assert s.fileno() > -1
        assert isinstance(s.fileno(), int)

    def test_socket_repr(self):
        import _socket
        s = _socket.socket(_socket.AF_INET, _socket.SOCK_STREAM)
        try:
            expected = ('<socket object, fd=%s, family=%s, type=%s, proto=%s>'
                        % (s.fileno(), s.family, s.type, s.proto))
            assert repr(s) == expected
        finally:
            s.close()
        expected = ('<socket object, fd=-1, family=%s, type=%s, proto=%s>'
                    % (s.family, s.type, s.proto))
        assert repr(s) == expected

    def test_socket_close(self):
        import _socket, os
        s = _socket.socket(_socket.AF_INET, _socket.SOCK_STREAM, 0)
        fileno = s.fileno()
        assert s.fileno() >= 0
        s.close()
        assert s.fileno() < 0
        s.close()
        if os.name != 'nt':
            raises(OSError, os.close, fileno)

    @pytest.mark.skipif("sys.platform == 'win32'")
    def test_socket_close_exception(self):
        import _socket, errno
        s = _socket.socket(_socket.AF_INET, _socket.SOCK_STREAM, 0)
        _socket.socket(fileno=s.fileno()).close()
        e = raises(OSError, s.close)
        assert e.value.errno in (errno.EBADF, errno.ENOTSOCK)

    @pytest.mark.skipif("sys.platform == 'win32'")
    def test_setblocking_invalidfd(self):
        import _socket
        s = _socket.socket(_socket.AF_INET, _socket.SOCK_STREAM, 0)
        _socket.socket(fileno=s.fileno()).close()
        raises(OSError, s.setblocking, False)

    def test_socket_connect(self):
        import _socket
        s = _socket.socket(_socket.AF_INET, _socket.SOCK_STREAM, 0)
        # it would be nice to have a test which works even if there is no
        # network connection. However, this one is "good enough" for now. Skip
        # it if there is no connection.
        try:
            s.connect(("www.python.org", 80))
        except _socket.gaierror as ex:
            skip("GAIError - probably no connection: %s" % str(ex.args))
        except ConnectionRefusedError as ex:
            skip("Connection Refused - probably no connection: %s" % str(ex.args))
        name = s.getpeername() # Will raise socket.error if not connected
        assert name[1] == 80
        s.close()

    def test_socket_connect_ex(self):
        import _socket
        s = _socket.socket(_socket.AF_INET, _socket.SOCK_STREAM, 0)
        # The following might fail if the DNS redirects failed requests to a
        # catch-all address (i.e. opendns).
        # Make sure we get an app-level error, not an interp one.
        raises(_socket.gaierror, s.connect_ex, ("wrong.invalid", 80))
        s.close()

    def test_socket_connect_typeerrors(self):
        tests = [
            "",
            "80",
            ("80",),
            ("80", "80"),
            (80, 80),
        ]
        import _socket
        s = _socket.socket(_socket.AF_INET, _socket.SOCK_STREAM, 0)
        for args in tests:
            raises((TypeError, ValueError), s.connect, args)
        s.close()

    def test_bigport(self):
        import _socket
        s = _socket.socket()
        exc = raises(OverflowError, s.connect, ("localhost", -1))
        assert "port must be 0-65535." in str(exc.value)
        exc = raises(OverflowError, s.connect, ("localhost", 1000000))
        assert "port must be 0-65535." in str(exc.value)
        s = _socket.socket(_socket.AF_INET6)
        exc = raises(OverflowError, s.connect, ("::1", 1234, 1048576))
        assert "flowinfo must be 0-1048575." in str(exc.value)

    def test_NtoH(self):
        import _socket as socket
        # This checks that htons etc. are their own inverse,
        # when looking at the lower 16 or 32 bits.  It also
        # checks that we get OverflowErrors when calling with -1,
        # or (for XtoXl()) with too large values.  For XtoXs()
        # large values are silently truncated instead, like CPython.
        sizes = {socket.htonl: 32, socket.ntohl: 32,
                 socket.htons: 16, socket.ntohs: 16}
        for func, size in sizes.items():
            mask = (1 << size) - 1
            for i in (0, 1, 0xffff, 0xffff0000, 2, 0x01234567, 0x76543210):
                assert i & mask == func(func(i&mask)) & mask

            swapped = func(mask)
            assert swapped & mask == mask
            raises(OverflowError, func, -1)
            if size > 16:    # else, values too large are ignored
                raises(OverflowError, func, 2 ** size)

    def test_newsocket(self):
        import _socket
        s = _socket.socket()

    def test_subclass(self):
        from _socket import socket
        class MySock(socket):
            blah = 123
        s = MySock()
        assert s.blah == 123

    def test_getsetsockopt(self):
        import _socket as socket
        import struct
        # A socket should start with reuse == 0
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        reuse = s.getsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR)
        assert reuse == 0
        #
        raises(TypeError, s.setsockopt, socket.SOL_SOCKET,
                          socket.SO_REUSEADDR, 2 ** 31)
        raises(TypeError, s.setsockopt, socket.SOL_SOCKET,
                          socket.SO_REUSEADDR, 2 ** 32 + 1)
        assert s.getsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR) == 0
        #
        s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        reuse = s.getsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR)
        assert reuse != 0
        # String case
        intsize = struct.calcsize('i')
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        reusestr = s.getsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR,
                                intsize)
        (reuse,) = struct.unpack('i', reusestr)
        assert reuse == 0
        reusestr = struct.pack('i', 1)
        s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, reusestr)
        reusestr = s.getsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR,
                                intsize)
        (reuse,) = struct.unpack('i', reusestr)
        assert reuse != 0
        # try to call setsockopt() with a buffer argument
        reusestr = struct.pack('i', 0)
        s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, memoryview(reusestr))
        reusestr = s.getsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR,
                                intsize)
        (reuse,) = struct.unpack('i', reusestr)
        assert reuse == 0

    def test_getsetsockopt_zero(self):
        # related to issue #2561: when specifying the buffer size param:
        # if 0 or None, should return the setted value,
        # otherwise an empty buffer of the specified size
        import _socket
        s = _socket.socket()
        assert s.getsockopt(_socket.IPPROTO_TCP, _socket.TCP_NODELAY, 0) == 0
        ret = s.getsockopt(_socket.IPPROTO_TCP, _socket.TCP_NODELAY, 2)
        if len(ret) == 1:
            # win32 returns a byte-as-bool
            assert ret == b'\x00'
        else:
            assert ret == b'\x00\x00'
        s.setsockopt(_socket.IPPROTO_TCP, _socket.TCP_NODELAY, True)
        assert s.getsockopt(_socket.IPPROTO_TCP, _socket.TCP_NODELAY, 0) != 0
        s.setsockopt(_socket.IPPROTO_TCP, _socket.TCP_NODELAY, 1)
        assert s.getsockopt(_socket.IPPROTO_TCP, _socket.TCP_NODELAY, 0) != 0

    def test_getsockopt_bad_length(self):
        import _socket
        s = _socket.socket()
        buf = s.getsockopt(_socket.IPPROTO_TCP, _socket.TCP_NODELAY, 1024)
        if len(buf) == 1:
            # win32 returns a byte-as-bool
            assert buf == b'\x00'
        else:
            assert buf == b'\x00' * 4
        raises(_socket.error, s.getsockopt,
               _socket.IPPROTO_TCP, _socket.TCP_NODELAY, 1025)
        raises(_socket.error, s.getsockopt,
               _socket.IPPROTO_TCP, _socket.TCP_NODELAY, -1)

    def test_socket_ioctl(self):
        import _socket, sys
        if sys.platform != 'win32':
            skip("win32 only")
        assert hasattr(_socket.socket, 'ioctl')
        assert hasattr(_socket, 'SIO_RCVALL')
        assert hasattr(_socket, 'RCVALL_ON')
        assert hasattr(_socket, 'RCVALL_OFF')
        assert hasattr(_socket, 'SIO_KEEPALIVE_VALS')
        s = _socket.socket()
        raises(ValueError, s.ioctl, -1, None)
        s.ioctl(_socket.SIO_KEEPALIVE_VALS, (1, 100, 100))

    @pytest.mark.skipif(os.name != 'nt', reason="win32 only")
    def test_socket_sharelocal(self):
        import _socket, sys, os
        assert hasattr(_socket.socket, 'share')
        s = _socket.socket(_socket.AF_INET, _socket.SOCK_STREAM)
        with raises(OSError):
            s.listen()
        s.bind(('localhost', 0))
        s.listen(1)
        data = s.share(os.getpid())
        # emulate socket.fromshare
        s2 = _socket.socket(0, 0, 0, data)
        try:
            assert s.gettimeout() == s2.gettimeout()
            assert s.family == s2.family
            assert s.type == s2.type
            if s.proto != 0:
                assert s.proto == s2.proto
        finally:
            s.close()
            s2.close()

    def test_dup(self):
        import _socket as socket, os
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        s.bind(('localhost', 0))
        fd = socket.dup(s.fileno())
        assert s.fileno() != fd
        if os.name != 'nt':
            assert os.get_inheritable(s.fileno()) is False
            assert os.get_inheritable(fd) is False
        s_dup = socket.socket(fileno=fd)
        s_dup.close()
        s.close()

    def test_dup_error(self):
        import _socket
        raises(_socket.error, _socket.dup, 123456)

    @pytest.mark.skipif(os.name=='nt', reason="no recvmsg on win32")
    def test_recvmsg_issue2649(self):
        import _socket as socket
        listener = socket.socket(family=socket.AF_INET, type=socket.SOCK_DGRAM)
        listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        listener.bind(('127.0.0.1', 1234))

        s = socket.socket(family=socket.AF_INET, type=socket.SOCK_DGRAM)
        s.sendto(b'x', ('127.0.0.1', 1234))
        with raises(BlockingIOError):
            queue = s.recvmsg(1024, 1024, socket.MSG_ERRQUEUE)

    def test_buffer(self):
        # Test that send/sendall/sendto accept a buffer as arg
        import _socket
        s = _socket.socket(_socket.AF_INET, _socket.SOCK_STREAM, 0)
        # XXX temporarily we use python.org to test, will have more robust tests
        # in the absence of a network connection later when more parts of the
        # socket API are implemented.  Currently skip the test if there is no
        # connection.
        try:
            s.connect(("www.python.org", 80))
        except _socket.gaierror as ex:
            skip("GAIError - probably no connection: %s" % str(ex.args))
        except ConnectionRefusedError as ex:
            skip("Connection Refused - probably no connection: %s" % str(ex.args))
        exc = raises(TypeError, s.send, None)
        assert str(exc.value).startswith("a bytes-like object is required,")
        assert s.send(memoryview(b'')) == 0
        assert s.sendall(memoryview(b'')) is None
        exc = raises(TypeError, s.send, '')
        assert str(exc.value).startswith("a bytes-like object is required,")
        raises(TypeError, s.sendall, '')
        s.close()
        s = _socket.socket(_socket.AF_INET, _socket.SOCK_DGRAM, 0)
        s.sendto(memoryview(b''), ('localhost', 9))  # Send to discard port.
        s.close()

    def test_listen_default(self):
        import _socket, sys
        if sys.platform == 'win32':
            with raises(OSError):
                _socket.socket().listen()
        else:
            _socket.socket().listen()
        assert isinstance(_socket.SOMAXCONN, int)

    def test_unix_socket_connect(self):
        import _socket, os
        if not hasattr(_socket, 'AF_UNIX'):
            skip('AF_UNIX not supported.')
        oldcwd = os.getcwd()
        os.chdir(self.udir)
        try:
          for sockpath in ['app_test_unix_socket_connect',
                           b'b_app_test_unix_socket_connect',
                           bytearray(b'ba_app_test_unix_socket_connect')]:

            serversock = _socket.socket(_socket.AF_UNIX)
            serversock.bind(sockpath)
            serversock.listen(1)

            clientsock = _socket.socket(_socket.AF_UNIX)
            clientsock.connect(sockpath)
            fileno, addr = serversock._accept()
            s = _socket.socket(fileno=fileno)
            assert not addr

            s.send(b'X')
            data = clientsock.recv(100)
            assert data == b'X'
            clientsock.send(b'Y')
            data = s.recv(100)
            assert data == b'Y'

            clientsock.close()
            s.close()
        finally:
            os.chdir(oldcwd)

    def test_automatic_shutdown(self):
        # doesn't really test anything, but at least should not explode
        # in close_all_sockets()
        import _socket
        self.foo = _socket.socket()

    def test_subclass_init(self):
        # Socket is not created in __new__, but in __init__.
        import _socket as socket
        class Socket_IPV6(socket.socket):
            def __init__(self):
                socket.socket.__init__(self, family=socket.AF_INET6)
        assert Socket_IPV6().family == socket.AF_INET6

    def test_subclass_noinit(self):
        from _socket import socket
        class MySock(socket):
            def __init__(self, *args):
                pass  # don't call super
        s = MySock()
        assert s.type == 0
        assert s.proto == 0
        assert s.family == 0
        assert s.fileno() < 0
        raises(OSError, s.bind, ('localhost', 0))

    def test_dealloc_warn(self):
        import _socket
        import gc
        import warnings

        s = _socket.socket(_socket.AF_INET, _socket.SOCK_STREAM)
        r = repr(s)
        gc.collect()
        with warnings.catch_warnings(record=True) as w:
            warnings.simplefilter('always')
            s = None
            gc.collect()
        assert len(w) == 1, [str(warning) for warning in w]
        assert r in str(w[0])

    def test_invalid_fd(self):
        import _socket
        raises(ValueError, _socket.socket, fileno=-1)
        raises(TypeError, _socket.socket, fileno=42.5)

    def test_socket_non_inheritable(self):
        import _socket, os
        s1 = _socket.socket()
        if os.name == 'nt':
            with raises(OSError):
                os.get_inheritable(s1.fileno())
        else:
            assert os.get_inheritable(s1.fileno()) is False
        s1.close()

    def test_socketpair_non_inheritable(self):
        import _socket, os
        if not hasattr(_socket, 'socketpair'):
            skip("no socketpair")
        s1, s2 = _socket.socketpair()
        assert os.get_inheritable(s1.fileno()) is False
        assert os.get_inheritable(s2.fileno()) is False
        s1.close()
        s2.close()

    def test_hostname_unicode(self):
        import _socket
        domain = u"испытание.pythontest.net"
        try:
            _socket.gethostbyname(domain)
            _socket.gethostbyname_ex(domain)
            _socket.getaddrinfo(domain, 0, _socket.AF_UNSPEC, _socket.SOCK_STREAM)
        except _socket.gaierror as ex:
            skip("GAIError - probably no connection: %s" % str(ex.args))
        s = _socket.socket(_socket.AF_INET, _socket.SOCK_STREAM)
        try:
            s.connect((domain, 80))
        except ConnectionRefusedError as ex:
            skip("Connection Refused - probably no connection: %s" % str(ex.args))
        s.close()
        raises(TypeError, s.connect, (domain + '\x00', 80))

    def test_socket_close(self):
        import _socket
        sock = _socket.socket()
        try:
            sock.bind(('localhost', 0))
            _socket.close(sock.fileno())
            with raises(OSError):
                sock.listen(1)
            with raises(OSError):
                _socket.close(sock.fileno())
        finally:
            with raises(OSError):
                sock.close()

    def test_socket_get_values_from_fd(self):
        import _socket
        if hasattr(_socket, "SOCK_DGRAM"):
            s = _socket.socket(_socket.AF_INET, _socket.SOCK_DGRAM)
            try:
                s.bind(('localhost', 0))
                fd = s.fileno()
                s2 = _socket.socket(fileno=fd)
                try:
                    # detach old fd to avoid double close
                    s.detach()
                    assert s2.fileno() == fd
                    assert s2.family == _socket.AF_INET
                    assert s2.type == _socket.SOCK_DGRAM

                finally:
                    s2.close()
            finally:
                s.close()

    def test_socket_init_non_blocking(self):
        import _socket
        if not hasattr(_socket, "SOCK_NONBLOCK"):
            skip("no SOCK_NONBLOCK")
        s = _socket.socket(_socket.AF_INET, _socket.SOCK_STREAM |
                                            _socket.SOCK_NONBLOCK)
        assert s.getblocking() == False
        assert s.gettimeout() == 0.0

    def test_socket_consistent_sock_type(self):
        import _socket
        SOCK_NONBLOCK = getattr(_socket, 'SOCK_NONBLOCK', 0)
        SOCK_CLOEXEC = getattr(_socket, 'SOCK_CLOEXEC', 0)
        sock_type = _socket.SOCK_STREAM | SOCK_NONBLOCK | SOCK_CLOEXEC

        s = _socket.socket(_socket.AF_INET, sock_type)
        try:
            assert s.type == _socket.SOCK_STREAM
            s.settimeout(1)
            assert s.type == _socket.SOCK_STREAM
            s.settimeout(0)
            assert s.type == _socket.SOCK_STREAM
            s.setblocking(True)
            assert s.type == _socket.SOCK_STREAM
            s.setblocking(False)
            assert s.type == _socket.SOCK_STREAM
        finally:
            s.close()

@pytest.mark.skipif(not hasattr(os, 'getpid'),
    reason="AF_NETLINK needs os.getpid()")
class AppTestNetlink:
    spaceconfig = {'usemodules': ['_socket', 'select']}

    def test_connect_to_kernel_netlink_routing_socket(self):
        import _socket, os
        if not hasattr(_socket, 'AF_NETLINK'):
            skip("no AF_NETLINK on this platform")
        s = _socket.socket(_socket.AF_NETLINK, _socket.SOCK_DGRAM,
                           _socket.NETLINK_ROUTE)
        assert s.getsockname() == (0, 0)
        s.bind((0, 0))
        a, b = s.getsockname()
        assert a == os.getpid()
        assert b == 0


@pytest.mark.skipif(not hasattr(os, 'getuid') or os.getuid() != 0,
    reason="AF_PACKET needs to be root for testing")
class AppTestPacket:
    spaceconfig = {'usemodules': ['_socket', 'select']}

    def test_convert_between_tuple_and_sockaddr_ll(self):
        import _socket
        if not hasattr(_socket, 'AF_PACKET'):
            skip("no AF_PACKET on this platform")
        s = _socket.socket(_socket.AF_PACKET, _socket.SOCK_RAW)
        assert s.getsockname() == ('', 0, 0, 0, b''), 's.getsockname %s' % str(s.getsockname())
        s.bind(('lo', 123))
        a, b, c, d, e = s.getsockname()
        assert (a, b, c) == ('lo', 123, 0)
        assert isinstance(d, int)
        assert isinstance(e, bytes)
        assert 0 <= len(e) <= 8
        s.close()


class AppTestSocketTCP:
    HOST = 'localhost'
    spaceconfig = {'usemodules': ['_socket', 'array', 'select']}

    def setup_method(self, method):
        w_HOST = self.space.wrap(self.HOST)
        self.w_serv = self.space.appexec([w_HOST],
            '''(HOST):
            import _socket
            serv = _socket.socket(_socket.AF_INET, _socket.SOCK_STREAM)
            serv.bind((HOST, 0))
            serv.listen(1)
            return serv
            ''')

    def teardown_method(self, method):
        if hasattr(self, 'w_serv'):
            self.space.appexec([self.w_serv], '(serv): serv.close()')
            self.w_serv = None

    def test_timeout(self):
        from _socket import timeout
        def raise_timeout():
            self.serv.settimeout(1.0)
            self.serv._accept()
        raises(timeout, raise_timeout)

    def test_timeout_zero(self):
        from _socket import error
        def raise_error():
            self.serv.settimeout(0.0)
            foo = self.serv._accept()
        raises(error, raise_error)

    @pytest.mark.skipif(os.name == 'nt', reason="win32 has additional buffering")
    def test_recv_send_timeout(self):
        from _socket import socket, timeout, SOL_SOCKET, SO_RCVBUF, SO_SNDBUF
        import sys
        cli = socket()
        cli.settimeout(1.0)
        cli.connect(self.serv.getsockname())
        fileno, addr = self.serv._accept()
        t = socket(fileno=fileno)
        # test recv() timeout
        t.send(b'*')
        buf = cli.recv(100)
        assert buf == b'*'
        raises(timeout, cli.recv, 100)
        # test that send() works
        count = cli.send(b'!')
        assert count == 1
        buf = t.recv(1)
        assert buf == b'!'
        # test that sendall() works
        count = cli.sendall(b'?')
        assert count is None
        buf = t.recv(1)
        assert buf == b'?'
        # speed up filling the buffers
        t.setsockopt(SOL_SOCKET, SO_RCVBUF, 4096)
        cli.setsockopt(SOL_SOCKET, SO_SNDBUF, 4096)
        # test send() timeout
        count = 0
        if sys.platform != 'win32':
            # windows never fills the buffer
            try:
                while 1:
                    count += cli.send(b'foobar' * 70)
                    if sys.platform == 'darwin':
                        # MacOS will auto-tune up to 512k
                        # (net.inet.tcp.doauto{rcv,snd}buf sysctls)
                        assert count < 1000000
                    else:
                        assert count < 100000
            except timeout:
                pass
            t.recv(count)
        # test sendall() timeout
        try:
            while 1:
                cli.sendall(b'foobar' * 70)
        except timeout:
            pass
        # done
        cli.close()
        t.close()

    def test_getblocking(self):
        self.serv.setblocking(True)
        assert self.serv.getblocking()
        self.serv.setblocking(False)
        assert not self.serv.getblocking()

    def test_recv_into(self):
        import _socket as socket
        import array
        import _io
        MSG = b'dupa was here\n'
        cli = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        cli.connect(self.serv.getsockname())
        fileno, addr = self.serv._accept()
        conn = socket.socket(fileno=fileno)
        buf = memoryview(MSG)
        conn.send(buf)
        buf = array.array('b', b' ' * 1024)
        nbytes = cli.recv_into(buf)
        assert nbytes == len(MSG)
        msg = buf.tobytes()[:len(MSG)]
        assert msg == MSG

        conn.send(MSG)
        buf = bytearray(1024)
        nbytes = cli.recv_into(memoryview(buf))
        assert nbytes == len(MSG)
        msg = buf[:len(MSG)]
        assert msg == MSG

        # A case where rwbuffer.get_raw_address() fails
        conn.send(MSG)
        buf = _io.BytesIO(b' ' * 1024)
        m = buf.getbuffer()
        nbytes = cli.recv_into(m)
        assert nbytes == len(MSG)
        msg = buf.getvalue()[:len(MSG)]
        assert msg == MSG
        conn.close()

    def test_recvfrom_into(self):
        import _socket as socket
        import array
        import _io
        MSG = b'dupa was here\n'
        cli = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        cli.connect(self.serv.getsockname())
        fileno, addr = self.serv._accept()
        conn = socket.socket(fileno=fileno)
        buf = memoryview(MSG)
        conn.send(buf)
        buf = array.array('b', b' ' * 1024)
        nbytes, addr = cli.recvfrom_into(buf)
        assert nbytes == len(MSG)
        msg = buf.tobytes()[:len(MSG)]
        assert msg == MSG

        conn.send(MSG)
        buf = bytearray(1024)
        nbytes, addr = cli.recvfrom_into(memoryview(buf))
        assert nbytes == len(MSG)
        msg = buf[:len(MSG)]
        assert msg == MSG

        # A case where rwbuffer.get_raw_address() fails
        conn.send(MSG)
        buf = _io.BytesIO(b' ' * 1024)
        nbytes, addr = cli.recvfrom_into(buf.getbuffer())
        assert nbytes == len(MSG)
        msg = buf.getvalue()[:len(MSG)]
        assert msg == MSG

        conn.send(MSG)
        buf = bytearray(8)
        exc = raises(ValueError, cli.recvfrom_into, buf, 1024)
        assert str(exc.value) == "nbytes is greater than the length of the buffer"
        conn.close()

    @pytest.mark.skipif(os.name == 'nt', reason="no recvmg_into on win32")
    def test_recvmsg_into(self):
        import _socket
        cli = _socket.socket(_socket.AF_INET, _socket.SOCK_STREAM)
        cli.connect(self.serv.getsockname())
        fileno, addr = self.serv._accept()
        conn = _socket.socket(fileno=fileno)
        conn.send(b'Hello World!')
        buf1 = bytearray(5)
        buf2 = bytearray(6)
        rettup = cli.recvmsg_into([memoryview(buf1), memoryview(buf2)])
        nbytes, _, _, addr = rettup
        assert nbytes == 11
        assert buf1 == b'Hello'
        assert buf2 == b' World'
        conn.close()
        cli.close()

    def test_family(self):
        import _socket as socket
        cli = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        assert cli.family == socket.AF_INET

    def test_missing_error_catching(self):
        from _socket import socket, error
        s = socket()
        s.close()
        raises(error, s.settimeout, 1)            # EBADF
        raises(error, s.setblocking, True)        # EBADF
        raises(error, s.getsockopt, 42, 84, 8)    # EBADF

    def test_accept_non_inheritable(self):
        import _socket, os
        cli = _socket.socket()
        cli.connect(self.serv.getsockname())
        fileno, addr = self.serv._accept()
        if os.name == 'nt':
            with raises(OSError):
                os.get_inheritable(fileno)
        else:
            assert os.get_inheritable(fileno) is False
        conn = _socket.socket(fileno=fileno)
        conn.close()
        cli.close()

    def test_recv_into_params(self):
        import os
        import _socket
        cli = _socket.socket()
        cli.connect(self.serv.getsockname())
        fileno, addr = self.serv._accept()
        conn = _socket.socket(fileno=fileno)
        conn.send(b"abcdef")
        #
        m = memoryview(bytearray(5))
        raises(ValueError, cli.recv_into, m, -1)
        raises(ValueError, cli.recv_into, m, 6)
        cli.recv_into(m, 5)
        assert m.tobytes() == b"abcde"
        conn.close()
        cli.close()

    def test_bytearray_name(self):
        import _socket as socket
        if not hasattr(socket, 'AF_UNIX'):
            skip('AF_UNIX not supported.')
        s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        s.bind(bytearray(b"\x00python\x00test\x00"))
        assert s.getsockname() == b"\x00python\x00test\x00"

    def test_no_socket_cloexec_non_block(self):
        import _socket
        #assert not hasattr(_socket, "SOCK_CLOEXEC") # not in py 2
        #assert not hasattr(_socket, "SOCK_NONBLOCK") # 3.7 only

    def test_bind_audit(self):
        import _socket
        import sys
        events = []
        def f(event, args):
            events.append((event, args))
        s = _socket.socket(_socket.AF_INET, _socket.SOCK_STREAM)
        sys.addaudithook(f)
        s.bind(('localhost', 0))
        assert len(events) == 1
        assert events[0][0] == 'socket.bind'
        assert events[0][1][0] is s
        assert events[0][1][1] == ('localhost', 0)

    def test_timeout_is_alias(self):
        import _socket
        assert _socket.timeout is TimeoutError

class AppTestErrno:
    spaceconfig = {'usemodules': ['_socket', 'select']}

    def test_errno(self):
        from _socket import socket, AF_INET, SOCK_STREAM, error
        import errno
        s = socket(AF_INET, SOCK_STREAM)
        exc = raises(error, s._accept)
        assert isinstance(exc.value, error)
        assert isinstance(exc.value, IOError)
        # error is EINVAL, or WSAEINVAL on Windows
        assert exc.value.errno == getattr(errno, 'WSAEINVAL', errno.EINVAL)
        assert isinstance(exc.value.strerror, str)