File: //usr/lib/python3/dist-packages/twisted/internet/test/__pycache__/test_unix.cpython-310.pyc
o
�b&� � @ s` d Z ddlmZ ddlmZmZmZmZmZ ddl m
Z
ddlmZm
Z
mZmZ ddlmZ ddlmZ ddlmZmZ dd lmZmZmZ dd
lmZ zddlmZ W n ey_ dZY nw eZdd
lmZm Z ddl!m"Z"m#Z# ddl$m%Z% ddl&m'Z'm(Z(m)Z) ddl*m+Z+m,Z, ddl-m.Z.m/Z/m0Z0 ddl1m2Z2m3Z3m4Z4m5Z5 ddl6m7Z7m8Z8m9Z9 ddl:m;Z; ddl<m=Z=m>Z>m?Z?m@Z@mAZA ddlBmCZC ddlDmEZEmFZFmGZGmHZH ddlImJZJ ddlKmLZL ddlMmNZN ddlOmPZPmQZQmRZR ddlSmTZT ddlUmVZV eTd�ZWd ZXeTd�du�rd!ZXG d"d#� d#�ZYd$d%� ZZG d&d'� d'e?�Z[G d(d)� d)e=�Z\e e2�G d*d+� d+e=��Z]G d,d-� d-eYeCe>�Z^G d.d/� d/eYeC�Z_G d0d1� d1�Z`G d2d3� d3�ZaG d4d5� d5�ZbG d6d7� d7eaebeCeG�ZcG d8d9� d9e`ebeCeG�ZdG d:d;� d;eHeC�Zeef� �ge^�h� � ef� �ge_�h� � ef� �gec�h� � ef� �ged�h� � ef� �gee�h� � G d<d=� d=eCe@�Zief� �gei�h� � dS )>z/
Tests for implementations of L{IReactorUNIX}.
� )�md5)�close�fstat�stat�unlink�urandom)�pformat)�AF_INET�SOCK_STREAM�
SOL_SOCKET�socket)�S_IMODE)�pack)�mkstemp�mktemp)�Optional�Sequence�Type)�skipIf)�AF_UNIXN)� Interface�implementer)�base�
interfaces)�UNIXAddress)�Deferred�fail�
gatherResults)�UNIXClientEndpoint�UNIXServerEndpoint)�CannotListenError�ConnectionClosed�FileDescriptorOverrun)�IFileDescriptorReceiver�
IReactorFDSet�IReactorSocket�IReactorUNIX)�
ClientFactory�DatagramProtocol�
ServerFactory)�LoopingCall)�ConnectableProtocol�ConnectionTestsMixin�EndpointCreator�StreamClientTestsMixin�runProtocolsWithReactor)�ReactorBuilder)�MyClientFactory�MyServerFactory�StreamTransportTestsMixin�WriteSequenceTestsMixin)�nativeString)�Failure)�_coerceToFilesystemEncoding)�addObserver�err�removeObserver)�
requireModule)�platformztwisted.python.sendmsg� z>sendmsg extension unavailable, extended UNIX features disabledc @ s e Zd ZdZdd� ZdS )�UNIXFamilyMixinzK
Test-helper defining mixin for things related to AF_UNIX sockets.
c C sB d}| � � }t||�|||d�}|�� | �tt|�j�|� dS )z}
Assert that the mode of the created unix socket is set to the mode
specified to the reactor method.
i� )�modeN)�buildReactor�getattr�
stopListening�assertEqualr
r �st_mode)�self�
methodName�path�factoryr? �reactor�unixPort� rK �A/usr/lib/python3/dist-packages/twisted/internet/test/test_unix.py� _modeTestO s
zUNIXFamilyMixin._modeTestN)�__name__�
__module__�__qualname__�__doc__rM rK rK rK rL r>