Function test_nonwindows_functionality

src/trio/_tests/test_fakenet.py:143–193

()

    reason="functions not in socket on windows",
)
async def test_nonwindows_functionality() -> None:
    # mypy doesn't support a good way of aborting typechecking on different platforms
    if sys.platform != "win32":  # pragma: no branch
        fn()
        s1 = trio.socket.socket(type=trio.socket.SOCK_DGRAM)
        s2 = trio.socket.socket(type=trio.socket.SOCK_DGRAM)
        await s2.bind(("127.0.0.1", 0))

        # sendmsg
        with pytest.raises(OSError, match=ENOTCONN_MSG) as exc:
            await s2.sendmsg([b"mno"])
        assert exc.value.errno == errno.ENOTCONN

        assert await s1.sendmsg([b"jkl"], (), 0, s2.getsockname()) == 3
        data, ancdata, msg_flags, addr = await s2.recvmsg(10)
        assert data == b"jkl"
        assert ancdata == []
        assert msg_flags == 0
        assert addr == s1.getsockname()

        # TODO: recvmsg

        # recvmsg_into
        assert await s1.sendto(b"xyzw", s2.getsockname()) == 4
        buf1 = bytearray(2)
        buf2 = bytearray(3)
        ret = await s2.recvmsg_into([buf1, buf2])
        nbytes, ancdata, msg_flags, addr = ret
        assert nbytes == 4
        assert buf1 == b"xy"
        assert buf2 == b"zw" + b"\x00"
        assert ancdata == []
        assert msg_flags == 0
        assert addr == s1.getsockname()

        # recvmsg_into with MSG_TRUNC set
        assert await s1.sendto(b"xyzwv", s2.getsockname()) == 5
        buf1 = bytearray(2)
        ret = await s2.recvmsg_into([buf1])
        nbytes, ancdata, msg_flags, addr = ret
        assert nbytes == 2
        assert buf1 == b"xy"
        assert ancdata == []
        assert msg_flags == socket.MSG_TRUNC
        assert addr == s1.getsockname()

        with pytest.raises(
            AttributeError,
            match=r"^'FakeSocket' object has no attribute 'share'$",
        ):
            await s1.share(0)  # type: ignore[attr-defined]


@pytest.mark.skipif(