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	571	0644	edit dl rm
modules_helpers.py	1605	0644	edit dl rm
pullpipe.py	1220	0644	edit dl rm
test_appdirs.py	1081	0644	edit dl rm
test_components.py	25974	0644	edit dl rm
test_constants.py	38117	0644	edit dl rm
test_deprecate.py	38930	0644	edit dl rm
test_dist3.py	1858	0644	edit dl rm
test_fakepwd.py	14107	0644	edit dl rm
test_htmlizer.py	1266	0644	edit dl rm
test_inotify.py	3639	0644	edit dl rm
test_release.py	41046	0644	edit dl rm
test_runtime.py	7926	0644	edit dl rm
test_sendmsg.py	25233	0644	edit dl rm
test_setup.py	12413	0644	edit dl rm
test_shellcomp.py	21809	0644	edit dl rm
test_syslog.py	4948	0644	edit dl rm
test_systemd.py	6401	0644	edit dl rm
test_textattributes.py	712	0644	edit dl rm
test_tzhelper.py	3984	0644	edit dl rm
test_url.py	29928	0644	edit dl rm
test_urlpath.py	10328	0644	edit dl rm
test_util.py	36460	0644	edit dl rm
test_versions.py	5400	0644	edit dl rm
test_zippath.py	3474	0644	edit dl rm
test_zipstream.py	12308	0644	edit dl rm
_deprecatetests.py.3only	1814	0644	edit dl rm
__init__.py	42	0644	edit dl rm

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

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

import sys
import os
import socket
from struct import unpack

from twisted.python.sendmsg import recvmsg

def recvfd(socketfd):
    """
    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 sys import os import socket from struct import unpack from twisted.python.sendmsg import recvmsg def recvfd(socketfd): """ 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)