FM / test

/usr/lib/python3/dist-packages/twisted/python/test/

/usr/lib/python3/dist-packages/twisted/python/test

Name	Size	Mode	Actions
__pycache__/	-	0755	rm
deprecatedattributes.py	505	0644	edit dl rm
modules_helpers.py	1805	0644	edit dl rm
pullpipe.py	1271	0644	edit dl rm
test_appdirs.py	1075	0644	edit dl rm
test_components.py	25826	0644	edit dl rm
test_constants.py	37824	0644	edit dl rm
test_deprecate.py	42024	0644	edit dl rm
test_fakepwd.py	15195	0644	edit dl rm
test_htmlizer.py	1273	0644	edit dl rm
test_inotify.py	3710	0644	edit dl rm
test_pydoctor.py	6214	0644	edit dl rm
test_release.py	41971	0644	edit dl rm
test_runtime.py	7992	0644	edit dl rm
test_sendmsg.py	10274	0644	edit dl rm
test_shellcomp.py	21167	0644	edit dl rm
test_syslog.py	5165	0644	edit dl rm
test_systemd.py	6272	0644	edit dl rm
test_textattributes.py	662	0644	edit dl rm
test_tzhelper.py	3800	0644	edit dl rm
test_url.py	28960	0644	edit dl rm
test_urlpath.py	10213	0644	edit dl rm
test_util.py	34888	0644	edit dl rm
test_versions.py	5238	0644	edit dl rm
test_win32.py	2017	0644	edit dl rm
test_zippath.py	3373	0644	edit dl rm
test_zipstream.py	12079	0644	edit dl rm
__init__.py	42	0644	edit dl rm

Edit: /usr/lib/python3/dist-packages/twisted/python/test/pullpipe.py (1271B)

# -*- test-case-name: twisted.python.test.test_sendmsg -*-
# Copyright (c) Twisted Matrix Laboratories.
# See LICENSE for details.

import os
import socket
import sys
from struct import unpack
from typing import Tuple

from twisted.python.sendmsg import recvmsg

def recvfd(socketfd: int) -> Tuple[int, bytes]:
    """
    Receive a file descriptor from a L{sendmsg} message on the given C{AF_UNIX}
    socket.

@param socketfd: An C{AF_UNIX} socket, attached to another process waiting
        to send sockets via the ancillary data mechanism in L{send1msg}.

@param fd: C{int}

@return: a 2-tuple of (new file descriptor, description).
    @rtype: 2-tuple of (C{int}, C{bytes})
    """
    ourSocket = socket.fromfd(socketfd, socket.AF_UNIX, socket.SOCK_STREAM)
    data, ancillary, flags = recvmsg(ourSocket)
    [(cmsgLevel, cmsgType, packedFD)] = ancillary
    # cmsgLevel and cmsgType really need to be SOL_SOCKET / SCM_RIGHTS, but
    # since those are the *only* standard values, there's not much point in
    # checking.
    [unpackedFD] = unpack("i", packedFD)
    return (unpackedFD, data)

if __name__ == "__main__":
    fd, description = recvfd(int(sys.argv[1]))
    os.write(fd, b"Test fixture data: " + description + b".\n")
    os.close(fd)

# -*- test-case-name: twisted.python.test.test_sendmsg -*- # Copyright (c) Twisted Matrix Laboratories. # See LICENSE for details. import os import socket import sys from struct import unpack from typing import Tuple from twisted.python.sendmsg import recvmsg def recvfd(socketfd: int) -> Tuple[int, bytes]: """ Receive a file descriptor from a L{sendmsg} message on the given C{AF_UNIX} socket. @param socketfd: An C{AF_UNIX} socket, attached to another process waiting to send sockets via the ancillary data mechanism in L{send1msg}. @param fd: C{int} @return: a 2-tuple of (new file descriptor, description). @rtype: 2-tuple of (C{int}, C{bytes}) """ ourSocket = socket.fromfd(socketfd, socket.AF_UNIX, socket.SOCK_STREAM) data, ancillary, flags = recvmsg(ourSocket) [(cmsgLevel, cmsgType, packedFD)] = ancillary # cmsgLevel and cmsgType really need to be SOL_SOCKET / SCM_RIGHTS, but # since those are the *only* standard values, there's not much point in # checking. [unpackedFD] = unpack("i", packedFD) return (unpackedFD, data) if __name__ == "__main__": fd, description = recvfd(int(sys.argv[1])) os.write(fd, b"Test fixture data: " + description + b".\n") os.close(fd)