File manager - Edit - /home/newsbmcs.com/public_html/static/img/logo/runner.zip
Back
PK ���Z�`��� � ! __pycache__/_exit.cpython-310.pycnu �[��� o �b| � @ s� d Z ddlZddlmZ ddlmZmZmZ ddlm Z zddl Z W n e y2 G dd� d�Z Y nw G dd � d e�Z dd e ee f d ed dfdd�ZdS )z System exit support. � N)�IntEnum)�exit�stderr�stdout)�Unionc @ s� e Zd ZdZdZdZeZed Zed Zed Z ed Z ed Z ed Z ed Z ed Zed Zed Zed Zed Zed Zed ZdS )�StatuszT Object to hang C{EX_*} values off of as a substitute for L{posix}. �@ r � � � � � � � � � � � � � � N)�__name__� __module__� __qualname__�__doc__�EX__BASE�EX_OK�EX_USAGE� EX_DATAERR� EX_NOINPUT� EX_NOUSER� EX_NOHOST�EX_UNAVAILABLE� EX_SOFTWARE�EX_OSERR� EX_OSFILE� EX_CANTCREAT�EX_IOERR� EX_TEMPFAIL� EX_PROTOCOL� EX_NOPERM� EX_CONFIG� r, r, �B/usr/lib/python3/dist-packages/twisted/application/runner/_exit.pyr s& r c @ sp e Zd ZdZejZejZejZejZej Z ej Z ej Z ej Z ej Z ejZejZejZejZejZejZejZdS )� ExitStatusa" Standard exit status codes for system programs. @cvar EX_OK: Successful termination. @cvar EX_USAGE: Command line usage error. @cvar EX_DATAERR: Data format error. @cvar EX_NOINPUT: Cannot open input. @cvar EX_NOUSER: Addressee unknown. @cvar EX_NOHOST: Host name unknown. @cvar EX_UNAVAILABLE: Service unavailable. @cvar EX_SOFTWARE: Internal software error. @cvar EX_OSERR: System error (e.g., can't fork). @cvar EX_OSFILE: Critical OS file missing. @cvar EX_CANTCREAT: Can't create (user) output file. @cvar EX_IOERR: Input/output error. @cvar EX_TEMPFAIL: Temporary failure; the user is invited to retry. @cvar EX_PROTOCOL: Remote error in protocol. @cvar EX_NOPERM: Permission denied. @cvar EX_CONFIG: Configuration error. N)r r r r r r r r r r r! r" r# r$ r% r& r' r( r) r* r+ r, r, r, r- r. + s$ r. � �status�message�returnztyping.NoReturnc C s8 |r| t jkr t}nt}|�|� |�d� t| � dS )z� Exit the python interpreter with the given status and an optional message. @param status: An exit status. An appropriate value from L{ExitStatus} is recommended. @param message: An optional message to print. � N)r. r r r �write�sysexit)r0 r1 �outr, r, r- r S s r )r/ )r �typing�enumr �sysr r5 r r r �posixr � ImportErrorr. �int�strr, r, r, r- �<module> s �$(PK ���Z N�K� � $ __pycache__/__init__.cpython-310.pycnu �[��� o �b� � @ s d Z dS )z/ Facilities for running a Twisted application. N)�__doc__� r r �E/usr/lib/python3/dist-packages/twisted/application/runner/__init__.py�<module> s PK ���Z� �$ $ $ __pycache__/_pidfile.cpython-310.pycnu �[��� o �b� � @ s� U d Z ddlZddlmZmZmZ ddlmZ ddl m Z m Z ddl m Z mZ ddlmZ ddlmZ G d d � d e �Zee�G d d � d ��Zee�G d d� d��Ze� Zeed< G dd� de�ZG dd� de�ZG dd� de�ZG dd� de�Z dS )z PID file. � N)�getpid�kill�name)� TracebackType)�Optional�Type)� Interface� implementer)�Logger)�FilePathc @ sx e Zd ZdZdefdd�Zddd�Zddd �Zdefd d �Z dd d �Z de e e de e de e de e fdd�ZdS )�IPIDFilez5 Manages a file that remembers a process ID. �returnc C � dS )a+ Read the process ID stored in this PID file. @return: The contained process ID. @raise NoPIDFound: If this PID file does not exist. @raise EnvironmentError: If this PID file cannot be read. @raise ValueError: If this PID file's content is invalid. N� r r r �E/usr/lib/python3/dist-packages/twisted/application/runner/_pidfile.py�read � z IPIDFile.readNc C r )z� Store the PID of the current process in this PID file. @raise EnvironmentError: If this PID file cannot be written. Nr r r r r �writeRunningPID$ r zIPIDFile.writeRunningPIDc C r )zm Remove this PID file. @raise EnvironmentError: If this PID file cannot be removed. Nr r r r r �remove+ r zIPIDFile.removec C r )a Determine whether there is a running process corresponding to the PID in this PID file. @return: True if this PID file contains a PID and a process with that PID is currently running; false otherwise. @raise EnvironmentError: If this PID file cannot be read. @raise InvalidPIDFileError: If this PID file's content is invalid. @raise StalePIDFileError: If this PID file's content refers to a PID for which there is no corresponding running process. Nr r r r r � isRunning2 r zIPIDFile.isRunningc C r )z� Enter a context using this PIDFile. Writes the PID file with the PID of the running process. @raise AlreadyRunningError: A process corresponding to the PID in this PID file is already running. Nr r r r r � __enter__@ r zIPIDFile.__enter__�excType�excValue� tracebackc C r )zS Exit a context using this PIDFile. Removes the PID file. Nr )r r r r r r �__exit__J r zIPIDFile.__exit__�r N)r r )�__name__� __module__� __qualname__�__doc__�intr r r �boolr r r r � BaseExceptionr r r r r r r s ����r c @ s� e Zd ZdZe� Zededefdd��Z de ddfdd �Z defd d �Z deddfd d �Z d dd�Zd dd�Zdefdd�Zededefdd��Zd dd�Zdeee dee dee ddfdd �ZdS ) �PIDFilez� Concrete implementation of L{IPIDFile}. This implementation is presently not supported on non-POSIX platforms. Specifically, calling L{PIDFile.isRunning} will raise L{NotImplementedError}. �pidr c C s t | �� d��� S )z� Format a PID file's content. @param pid: A process ID. @return: Formatted PID file contents. � )r �encode�r$ r r r �_formatb s zPIDFile._format�filePathNc C s || _ dS )zD @param filePath: The path to the PID file on disk. N)r) )�selfr) r r r �__init__m s zPIDFile.__init__c C s� d}z | j �� � }|D ]} W d � n1 sw Y W n ty5 } z |jtjkr0td��� d }~ww zt|�W S tyI td|����w )N� �PID file does not existz#non-integer PID value in PID file: ) r) �open�OSError�errno�ENOENT� NoPIDFoundr � ValueError�InvalidPIDFileError)r* � pidString�fh�er r r r s s* � �� �� ��z PIDFile.readc C s | j �| j|d�� dS )z� Store a PID in this PID file. @param pid: A PID to store. @raise EnvironmentError: If this PID file cannot be written. r' N)r) � setContentr( �r* r$ r r r �_write� s zPIDFile._writec C s | � t� � d S �N)r: r �r* r r r r � s zPIDFile.writeRunningPIDc C s | j �� d S r; )r) r r< r r r r � � zPIDFile.removec C sB z| � � }W n ty Y dS w tdkr| �|�S tdt� ���)NF�posixz isRunning is not implemented on )r r2 � SYSTEM_NAME�_pidIsRunningPOSIX�NotImplementedErrorr9 r r r r � s � zPIDFile.isRunningc C sZ zt | d� W dS ty, } z|jtjkrtd��|jtjkr'W Y d}~dS � d}~ww )a1 POSIX implementation for running process check. Determine whether there is a running process corresponding to the given PID. @param pid: The PID to check. @return: True if the given PID is currently running; false otherwise. @raise EnvironmentError: If this PID file cannot be read. @raise InvalidPIDFileError: If this PID file's content is invalid. @raise StalePIDFileError: If this PID file's content refers to a PID for which there is no corresponding running process. r z'PID file refers to non-existing processNT)r r/ r0 �ESRCH�StalePIDFileError�EPERM)r$ r7 r r r r@ � s � ��zPIDFile._pidIsRunningPOSIXc C s> z | � � rt� �W n ty | j�d� Y nw | �� | S )Nz&Replacing stale PID file: {log_source})r �AlreadyRunningErrorrC �_log�infor r< r r r r � s � �zPIDFile.__enter__r r r c C s | � � d S r; )r �r* r r r r r r r � s zPIDFile.__exit__r )r r# )r r r r r rF � staticmethodr �bytesr( r r+ r r: r r r! r r@ r r r r" r r r r r r r# V s. ����r# c @ s� e Zd ZdZddd�Zdefdd�Zdeddfd d �Zdd d �Zdd d�Z de fdd�Z ddd�Z de ee de e de e ddfdd�ZdS )� NonePIDFilez� PID file implementation that does nothing. This is meant to be used as a "active None" object in place of a PID file when no PID file is desired. r Nc C � d S r; r r< r r r r+ � � zNonePIDFile.__init__c C s t d��)Nr- )r2 r< r r r r � s zNonePIDFile.readr$ c C � t tjd��)z� Store a PID in this PID file. @param pid: A PID to store. @raise EnvironmentError: If this PID file cannot be written. @note: This implementation always raises an L{EnvironmentError}. zOperation not permitted)r/ r0 rD r9 r r r r: � s zNonePIDFile._writec C s | � d� d S )Nr )r: r< r r r r � r= zNonePIDFile.writeRunningPIDc C rN )NzNo such file or directory)r/ r0 r1 r< r r r r � s zNonePIDFile.removec C r )NFr r< r r r r � rM zNonePIDFile.isRunningc C s | S r; r r< r r r r � rM zNonePIDFile.__enter__r r r c C rL r; r rH r r r r � s zNonePIDFile.__exit__r )r rK )r r r r r+ r r r: r r r! r r r r r" r r r r r r rK � s$ ����rK � nonePIDFilec @ � e Zd ZdZdS )rE z% Process is already running. N�r r r r r r r r rE � rE c @ rP )r4 z( PID file contents are invalid. NrQ r r r r r4 rR r4 c @ rP )rC z[ PID file contents are valid, but there is no process with the referenced PID. NrQ r r r r rC rR rC c @ rP )r2 z# No PID found in PID file. NrQ r r r r r2 rR r2 ) r r0 �osr r r r? �typesr �typingr r �zope.interfacer r �twisted.loggerr �twisted.python.filepathr r r# rK rO �__annotations__� ExceptionrE r4 rC r2 r r r r �<module> s$ By/PK ���Z���� � # __pycache__/_runner.cpython-310.pycnu �[��� o �b� � @ s� d Z ddlmZ ddlmZ ddlmZ ddlmZm Z m Z m Z ddl m Z mZmZ ddlmZ ddlmZ dd lmZmZmZmZmZmZmZ d d l m Z m Z d d l m Z m!Z!m"Z"m#Z# ed d�G dd� d��Z$dS )z Twisted application runner. � )�kill)�SIGTERM)�stderr)�Any�Callable�Mapping�TextIO)�Factory�attrib�attrs)� NamedConstant)� IReactorCore)�FileLogObserver�FilteringLogObserver�Logger�LogLevel�LogLevelFilterPredicate�globalLogBeginner�textFileLogObserver� )� ExitStatus�exit)�AlreadyRunningError�InvalidPIDFileError�IPIDFile� nonePIDFileT)�frozenc @ s� e Zd ZdZe� Zeed�Zee e d�Z ee dd�Z eeejd�Zeeed�Zeeegef ed�Zeed dd� d�Zeeee f e e �d�Z eed dd� d�Z eeee f e e �d�Z!dd d �Z"dd d�Z#ddd�Z$ddd�Z%ddd�Z&ddd�Z'd S )�Runnera Twisted application runner. @cvar _log: The logger attached to this class. @ivar _reactor: The reactor to start and run the application in. @ivar _pidFile: The file to store the running process ID in. @ivar _kill: Whether this runner should kill an existing running instance of the application. @ivar _defaultLogLevel: The default log level to start the logging system with. @ivar _logFile: A file stream to write logging output to. @ivar _fileLogObserverFactory: A factory for the file log observer to use when starting the logging system. @ivar _whenRunning: Hook to call after the reactor is running; this is where the application code that relies on the reactor gets called. @ivar _whenRunningArguments: Keyword arguments to pass to C{whenRunning} when it is called. @ivar _reactorExited: Hook to call after the reactor exits. @ivar _reactorExitedArguments: Keyword arguments to pass to C{reactorExited} when it is called. )�type)r �defaultF).Nc K � d S �N� ��_r" r" �D/usr/lib/python3/dist-packages/twisted/application/runner/_runner.py�<lambda>C � zRunner.<lambda>c K r r! r" r# r" r" r% r& E r' �returnNc C sv | j }| �� z#|� | �� | �� | �� W d � W dS 1 s#w Y W dS ty: ttjd� Y dS w )z# Run this command. NzAlready running.) �_pidFile�killIfRequested� startLogging� startReactor� reactorExitedr r r � EX_CONFIG)�self�pidFiler" r" r% �runH s &� �z Runner.runc C s� | j }| jrP|tu rttjd� dS z|�� }W n ty( ttjd� Y dS t y7 ttj d� Y dS w | � � | j j d|d� t|t� ttj� dS dS )zj If C{self._kill} is true, attempt to kill a running instance of the application. zNo PID file specified.NzUnable to read PID file.zInvalid PID file.zTerminating process: {pid})�pid)r) �_killr r r �EX_USAGE�read�OSError�EX_IOERRr � EX_DATAERRr+ �_log�infor r �EX_OK)r/ r0 r2 r" r" r% r* [ s( � �zRunner.killIfRequestedc C s< | j }| j}||�}t| jd�}t||g�}t�|g� dS )z= Start the L{twisted.logger} logging system. )�defaultLogLevelN)�_logFile�_fileLogObserverFactoryr �_defaultLogLevelr r �beginLoggingTo)r/ �logFile�fileLogObserverFactory�fileLogObserver�logLevelPredicate�filteringObserverr" r" r% r+ | s � zRunner.startLoggingc C s( | j �| j� | j�d� | j �� dS )z� Register C{self._whenRunning} with the reactor so that it is called once the reactor is running, then start the reactor. zStarting reactor...N)�_reactor�callWhenRunning� whenRunningr9 r: r1 �r/ r" r" r% r, � s zRunner.startReactorc C � | j di | j�� dS )z� Call C{self._whenRunning} with C{self._whenRunningArguments}. @note: This method is called after the reactor starts running. Nr" )� _whenRunning�_whenRunningArgumentsrI r" r" r% rH � � zRunner.whenRunningc C rJ )z� Call C{self._reactorExited} with C{self._reactorExitedArguments}. @note: This method is called after the reactor exits. Nr" )�_reactorExited�_reactorExitedArgumentsrI r" r" r% r- � rM zRunner.reactorExited)r( N)(�__name__� __module__� __qualname__�__doc__r r9 r r rF r r r) �boolr3 r r r: r? r r r= r r r r> rK r �strr r �dictrL rN rO r1 r* r+ r, rH r- r" r" r" r% r s* � ! r N)%rS �osr �signalr �sysr �typingr r r r �attrr r r � constantlyr �twisted.internet.interfacesr �twisted.loggerr r r r r r r �_exitr r �_pidfiler r r r r r" r" r" r% �<module> s $ PK ���Z� � � _pidfile.pynu �[��� # -*- test-case-name: twisted.application.runner.test.test_pidfile -*- # Copyright (c) Twisted Matrix Laboratories. # See LICENSE for details. """ PID file. """ import errno from os import getpid, kill, name as SYSTEM_NAME from types import TracebackType from typing import Optional, Type from zope.interface import Interface, implementer from twisted.logger import Logger from twisted.python.filepath import FilePath class IPIDFile(Interface): """ Manages a file that remembers a process ID. """ def read() -> int: """ Read the process ID stored in this PID file. @return: The contained process ID. @raise NoPIDFound: If this PID file does not exist. @raise EnvironmentError: If this PID file cannot be read. @raise ValueError: If this PID file's content is invalid. """ def writeRunningPID() -> None: """ Store the PID of the current process in this PID file. @raise EnvironmentError: If this PID file cannot be written. """ def remove() -> None: """ Remove this PID file. @raise EnvironmentError: If this PID file cannot be removed. """ def isRunning() -> bool: """ Determine whether there is a running process corresponding to the PID in this PID file. @return: True if this PID file contains a PID and a process with that PID is currently running; false otherwise. @raise EnvironmentError: If this PID file cannot be read. @raise InvalidPIDFileError: If this PID file's content is invalid. @raise StalePIDFileError: If this PID file's content refers to a PID for which there is no corresponding running process. """ def __enter__() -> "IPIDFile": """ Enter a context using this PIDFile. Writes the PID file with the PID of the running process. @raise AlreadyRunningError: A process corresponding to the PID in this PID file is already running. """ def __exit__( excType: Optional[Type[BaseException]], excValue: Optional[BaseException], traceback: Optional[TracebackType], ) -> Optional[bool]: """ Exit a context using this PIDFile. Removes the PID file. """ @implementer(IPIDFile) class PIDFile: """ Concrete implementation of L{IPIDFile}. This implementation is presently not supported on non-POSIX platforms. Specifically, calling L{PIDFile.isRunning} will raise L{NotImplementedError}. """ _log = Logger() @staticmethod def _format(pid: int) -> bytes: """ Format a PID file's content. @param pid: A process ID. @return: Formatted PID file contents. """ return f"{int(pid)}\n".encode() def __init__(self, filePath: FilePath) -> None: """ @param filePath: The path to the PID file on disk. """ self.filePath = filePath def read(self) -> int: pidString = b"" try: with self.filePath.open() as fh: for pidString in fh: break except OSError as e: if e.errno == errno.ENOENT: # No such file raise NoPIDFound("PID file does not exist") raise try: return int(pidString) except ValueError: raise InvalidPIDFileError( f"non-integer PID value in PID file: {pidString!r}" ) def _write(self, pid: int) -> None: """ Store a PID in this PID file. @param pid: A PID to store. @raise EnvironmentError: If this PID file cannot be written. """ self.filePath.setContent(self._format(pid=pid)) def writeRunningPID(self) -> None: self._write(getpid()) def remove(self) -> None: self.filePath.remove() def isRunning(self) -> bool: try: pid = self.read() except NoPIDFound: return False if SYSTEM_NAME == "posix": return self._pidIsRunningPOSIX(pid) else: raise NotImplementedError(f"isRunning is not implemented on {SYSTEM_NAME}") @staticmethod def _pidIsRunningPOSIX(pid: int) -> bool: """ POSIX implementation for running process check. Determine whether there is a running process corresponding to the given PID. @param pid: The PID to check. @return: True if the given PID is currently running; false otherwise. @raise EnvironmentError: If this PID file cannot be read. @raise InvalidPIDFileError: If this PID file's content is invalid. @raise StalePIDFileError: If this PID file's content refers to a PID for which there is no corresponding running process. """ try: kill(pid, 0) except OSError as e: if e.errno == errno.ESRCH: # No such process raise StalePIDFileError("PID file refers to non-existing process") elif e.errno == errno.EPERM: # Not permitted to kill return True else: raise else: return True def __enter__(self) -> "PIDFile": try: if self.isRunning(): raise AlreadyRunningError() except StalePIDFileError: self._log.info("Replacing stale PID file: {log_source}") self.writeRunningPID() return self def __exit__( self, excType: Optional[Type[BaseException]], excValue: Optional[BaseException], traceback: Optional[TracebackType], ) -> None: self.remove() return None @implementer(IPIDFile) class NonePIDFile: """ PID file implementation that does nothing. This is meant to be used as a "active None" object in place of a PID file when no PID file is desired. """ def __init__(self) -> None: pass def read(self) -> int: raise NoPIDFound("PID file does not exist") def _write(self, pid: int) -> None: """ Store a PID in this PID file. @param pid: A PID to store. @raise EnvironmentError: If this PID file cannot be written. @note: This implementation always raises an L{EnvironmentError}. """ raise OSError(errno.EPERM, "Operation not permitted") def writeRunningPID(self) -> None: self._write(0) def remove(self) -> None: raise OSError(errno.ENOENT, "No such file or directory") def isRunning(self) -> bool: return False def __enter__(self) -> "NonePIDFile": return self def __exit__( self, excType: Optional[Type[BaseException]], excValue: Optional[BaseException], traceback: Optional[TracebackType], ) -> None: return None nonePIDFile: IPIDFile = NonePIDFile() class AlreadyRunningError(Exception): """ Process is already running. """ class InvalidPIDFileError(Exception): """ PID file contents are invalid. """ class StalePIDFileError(Exception): """ PID file contents are valid, but there is no process with the referenced PID. """ class NoPIDFound(Exception): """ No PID found in PID file. """ PK ���Z�]r��7 �7 - test/__pycache__/test_pidfile.cpython-310.pycnu �[��� o �b;2 � @ s� d Z ddlZddlmZ ddlmZmZ ddlm Z m Z m Z ddl m Z ddlZddlmZ ddlmZ dd lmZ d d lmZ d d lmZmZmZmZmZm Z m Z de de f de de f fdd�Z G dd� dej j j!�Z"G dd� dej j j!�Z#dS )z3 Tests for L{twisted.application.runner._pidfile}. � N)�wraps)�getpid�name)�Any�Callable�Optional)� verifyObject)�FilePath)�platform)�SkipTest� )�_pidfile� )�AlreadyRunningError�InvalidPIDFileError�IPIDFile� NonePIDFile� NoPIDFound�PIDFile�StalePIDFileError�f.�returnc s* t � �dtdtdtdtf� fdd� �}|S )a= Decorator for tests that are not expected to work on all platforms. Calling L{PIDFile.isRunning} currently raises L{NotImplementedError} on non-POSIX platforms. On an unsupported platform, we expect to see any test that calls L{PIDFile.isRunning} to raise either L{NotImplementedError}, L{SkipTest}, or C{self.failureException}. (C{self.failureException} may occur in a test that checks for a specific exception but it gets NotImplementedError instead.) @param f: The test method to decorate. @return: The wrapped callable. �self�args�kwargsr c sp t �� dk}|r� | g|�R i |��S | jtt| jf� | g|�R i |��}t|t�r6| �t|�� d�� d S d S )N�posixz isRunning is not implemented on ) r �getType� assertRaises�NotImplementedErrorr �failureException� isinstance� assertTrue�str� startswith)r r r � supported�e�r � �N/usr/lib/python3/dist-packages/twisted/application/runner/test/test_pidfile.py�wrapper1 s ��� �z$ifPlatformSupported.<locals>.wrapper)r r )r r) r' r& r( �ifPlatformSupported s r* c @ s e Zd ZdZd1dee defdd�Zd2dd�Zd2d d �Z d2d d �Z d2d d�Z d2dd�Z d2dd�Z d2dd�Zd2dd�Zd2dd�Zd2dd�Zd2dd �Zed2d d ��Zed2d d ��Zed2d!d"��Zed2d#d$��Zed2d%d&��Zed2d'd(��Zd2d)d*�Zd2d+d,�Zed2d-d.��Z ed2d/d0��Z dS )3� PIDFileTestsz Tests for L{PIDFile}. N�contentr c C s" t | �� �}|d ur|�|� |S �N)r �mktemp� setContent)r r, �filePathr' r' r( r0 J s zPIDFileTests.filePathc C s t | �� �}tt|� dS )z5 L{PIDFile} conforms to L{IPIDFile}. N)r r0 r r �r �pidFiler' r' r( �test_interfaceP s zPIDFileTests.test_interfacec C s | � tjdd�d� dS )zR L{PIDFile._format} returns the expected format when given a PID. �9 ��pids 1337 N)� assertEqualr �_format)r r' r' r( �test_formatWithPIDW s z PIDFileTests.test_formatWithPIDc C s. d}t | �t j|d���}| �||�� � dS )zK L{PIDFile.read} returns the PID from the given file path. r4 r5 N)r r0 r8 r7 �read�r r6 r2 r' r' r( �test_readWithPID] s z PIDFileTests.test_readWithPIDc C s: d}t | �d��}| �t|j�}| �t|�d|��� dS )�f L{PIDFile.read} raises L{InvalidPIDFileError} when given an empty file path. � �#non-integer PID value in PID file: N�r r0 r r r: r7 r" �r �pidValuer2 r% r' r' r( �test_readEmptyPIDg � z PIDFileTests.test_readEmptyPIDc C s: d}t | �|��}| �t|j�}| �t|�d|��� dS )r= s $foo!r? Nr@ rA r' r' r( �test_readWithBogusPIDr rD z"PIDFileTests.test_readWithBogusPIDc C s. t | �� �}| �t|j�}| �t|�d� dS )zc L{PIDFile.read} raises L{NoPIDFound} when given a non-existing file path. �PID file does not existN)r r0 r r r: r7 r" �r r2 r% r' r' r( �test_readDoesntExist} s z!PIDFileTests.test_readDoesntExistc C sP ddt dtfdd�}| �td|� t| �� �}| �t|j�}| �|j t j � dS ) z� L{PIDFile.read} re-raises L{OSError} if the associated C{errno} is anything other than L{errno.ENOENT}. �r�moder c S � t tjd��)Nz I/O error)�OSError�errno�EIO)rJ r' r' r( �oops� � z>PIDFileTests.test_readOpenRaisesOSErrorNotENOENT.<locals>.oops�openN)rI ) r" r �patchr r0 r rL r: r7 rM rN )r rO r2 �errorr' r' r( �#test_readOpenRaisesOSErrorNotENOENT� s z0PIDFileTests.test_readOpenRaisesOSErrorNotENOENTc C s. d}t | �� �}|�|� | �|�� |� dS )z9 L{PIDFile._write} stores the given PID. i� N)r r0 �_writer7 r: r; r' r' r( � test_writePID� s zPIDFileTests.test_writePIDc C s t | �� �}| �t|jd� dS )zS L{PIDFile._write} raises L{ValueError} when given an invalid PID. �burpN)r r0 r � ValueErrorrU r1 r' r' r( �test_writePIDInvalid� s z!PIDFileTests.test_writePIDInvalidc C s* t | �� �}|�� | �|�� t� � dS )zT L{PIDFile.writeRunningPID} stores the PID for the current process. N)r r0 �writeRunningPIDr7 r: r r1 r' r' r( �test_writeRunningPID� s z!PIDFileTests.test_writeRunningPIDc C s: t | �d��}| �|j�� � |�� | �|j�� � dS )z9 L{PIDFile.remove} removes the PID file. r> N)r r0 r! �exists�remove� assertFalser1 r' r' r( � test_remove� s zPIDFileTests.test_removec C �L t | �� �}|�d� dtdtddfdd�}| �td|� | �|�� � dS ) zR L{PIDFile.isRunning} returns true for a process that does exist. r4 r6 �signalr Nc S � d S r- r' �r6 ra r' r' r( �kill� � z2PIDFileTests.test_isRunningDoesExist.<locals>.killrd �r r0 rU �intrR r r! � isRunning�r r2 rd r' r' r( �test_isRunningDoesExist� s z$PIDFileTests.test_isRunningDoesExistc C s& t | �� �}|�� | �|�� � dS )a@ L{PIDFile.isRunning} returns true for this process (which is running). @note: This differs from L{PIDFileTests.test_isRunningDoesExist} in that it actually invokes the C{kill} system call, which is useful for testing of our chosen method for probing the existence of a process. N)r r0 rZ r! rh r1 r' r' r( �test_isRunningThis� s z PIDFileTests.test_isRunningThisc C sL t | �� �}|�d� dtdtddfdd�}| �td|� | �t|j� dS ) z{ L{PIDFile.isRunning} raises L{StalePIDFileError} for a process that does not exist (errno=ESRCH). r4 r6 ra r Nc S rK �NzNo such process�rL rM �ESRCHrc r' r' r( rd � rP z5PIDFileTests.test_isRunningDoesNotExist.<locals>.killrd ) r r0 rU rg rR r r r rh ri r' r' r( �test_isRunningDoesNotExist� � z'PIDFileTests.test_isRunningDoesNotExistc C r` ) zx L{PIDFile.isRunning} returns true for a process that we are not allowed to kill (errno=EPERM). r4 r6 ra r Nc S rK )NzOperation not permitted)rL rM �EPERMrc r' r' r( rd � rP z3PIDFileTests.test_isRunningNotAllowed.<locals>.killrd rf ri r' r' r( �test_isRunningNotAllowed� rp z%PIDFileTests.test_isRunningNotAllowedc C s8 t dkrtd��t| �� �}|�d� | �|�� � dS )ac L{PIDFile.isRunning} returns true for a process that we are not allowed to kill (errno=EPERM). @note: This differs from L{PIDFileTests.test_isRunningNotAllowed} in that it actually invokes the C{kill} system call, which is useful for testing of our chosen method for probing the existence of a process that we are not allowed to kill. @note: In this case, we try killing C{init}, which is process #1 on POSIX systems, so this test is not portable. C{init} should always be running and should not be killable by non-root users. r zThis test assumes POSIX� N)� SYSTEM_NAMEr r r0 rU r! rh r1 r' r' r( �test_isRunningInit� s z PIDFileTests.test_isRunningInitc C sJ t | �� �}|�� dtdtddfdd�}| �td|� | �t|j� dS )z� L{PIDFile.isRunning} re-raises L{OSError} if the attached C{errno} value from L{os.kill} is not an expected one. r6 ra r Nc S rK )Nz File exists)rL rM �EEXISTrc r' r' r( rd rP z5PIDFileTests.test_isRunningUnknownErrno.<locals>.killrd ) r r0 rZ rg rR r r rL rh ri r' r' r( �test_isRunningUnknownErrno s z'PIDFileTests.test_isRunningUnknownErrnoc C s t | �� �}| �|�� � dS )zS L{PIDFile.isRunning} returns false if the PID file doesn't exist. N)r r0 r^ rh r1 r' r' r( �test_isRunningNoPIDFile! s z$PIDFileTests.test_isRunningNoPIDFilec C sv t | �� �}| �|j�� � |� | �|j�� � | �|�� t� � W d � n1 s,w Y | �|j�� � dS )z� When used as a context manager, a L{PIDFile} will store the current pid on entry, then removes the PID file on exit. N)r r0 r^ r\ r! r7 r: r r1 r' r' r( �test_contextManager) s �z PIDFileTests.test_contextManagerc C s� t | �� �}|�d� dtdtddfdd�}| �td|� | �t|j�}| � t |�d � |� | � |� � t � � W d � dS 1 sCw Y dS ) z� When used as a context manager, a L{PIDFile} will replace the underlying PIDFile rather than raising L{AlreadyRunningError} if the contained PID file exists but refers to a non-running PID. r4 r6 ra r Nc S rK rl rm rc r' r' r( rd A rP z9PIDFileTests.test_contextManagerDoesntExist.<locals>.killrd z'PID file refers to non-existing process) r r0 rU rg rR r r r rh r7 r" r: r )r r2 rd r% r' r' r( � test_contextManagerDoesntExist7 s "�z+PIDFileTests.test_contextManagerDoesntExistc C sZ t | �� �}|�d� dtdtddfdd�}| �td|� | �|�� � | �t |j � dS ) z� When used as a context manager, a L{PIDFile} will raise L{AlreadyRunningError} if the there is already a running process with the contained PID. r4 r6 ra r Nc S rb r- r' rc r' r' r( rd V re z<PIDFileTests.test_contextManagerAlreadyRunning.<locals>.killrd ) r r0 rU rg rR r r! rh r r � __enter__ri r' r' r( �!test_contextManagerAlreadyRunningL s z.PIDFileTests.test_contextManagerAlreadyRunningr- �r N) �__name__� __module__� __qualname__�__doc__r �bytesr r0 r3 r9 r< rC rE rH rT rV rY r[ r_ r* rj rk ro rr ru rw rx ry rz r| r' r' r' r( r+ E s@ r+ c @ sV e Zd ZdZddd�Zddd�Zddd �Zdd d �Zdd d �Zddd�Z ddd�Z dS )�NonePIDFileTestsz# Tests for L{NonePIDFile}. r Nc C s t � }tt|� dS )z9 L{NonePIDFile} conforms to L{IPIDFile}. N)r r r r1 r' r' r( r3 e s z NonePIDFileTests.test_interfacec C s( t � }| �t|j�}| �t|�d� dS )z; L{NonePIDFile.read} raises L{NoPIDFound}. rF N)r r r r: r7 r" rG r' r' r( � test_readl � zNonePIDFileTests.test_readc C s* t � }| �t|jd�}| �|jtj� dS )zZ L{NonePIDFile._write} raises L{OSError} with an errno of L{errno.EPERM}. r N)r r rL rU r7 rM rq �r r2 rS r' r' r( � test_writeu s zNonePIDFileTests.test_writec C �( t � }| �t|j�}| �|jtj� dS )zk L{NonePIDFile.writeRunningPID} raises L{OSError} with an errno of L{errno.EPERM}. N)r r rL rZ r7 rM rq r� r' r' r( r[ ~ s z%NonePIDFileTests.test_writeRunningPIDc C r� )zZ L{NonePIDFile.remove} raises L{OSError} with an errno of L{errno.EPERM}. N)r r rL r] r7 rM �ENOENTr� r' r' r( r_ � r� z NonePIDFileTests.test_removec C s t � }| �|�� d� dS )z< L{NonePIDFile.isRunning} returns L{False}. FN)r r7 rh r1 r' r' r( �test_isRunning� s z NonePIDFileTests.test_isRunningc C s0 t � }|� W d � dS 1 sw Y dS )zo When used as a context manager, a L{NonePIDFile} doesn't raise, despite not existing. N)r r1 r' r' r( ry � s "�z$NonePIDFileTests.test_contextManagerr} ) r~ r r� r� r3 r� r� r[ r_ r� ry r' r' r' r( r� ` s r� )$r� rM � functoolsr �osr r rt �typingr r r �zope.interface.verifyr �twisted.trial.unittest�twisted�twisted.python.filepathr �twisted.python.runtimer r �runnerr r r r r r r r r* �trial�unittest�TestCaser+ r� r' r' r' r( �<module> s" $" & PK ���Z��� � ) test/__pycache__/__init__.cpython-310.pycnu �[��� o �b� � @ s d Z dS )z* Tests for L{twisted.application.runner}. N)�__doc__� r r �J/usr/lib/python3/dist-packages/twisted/application/runner/test/__init__.py�<module> s PK ���Z�{KQ< Q< , test/__pycache__/test_runner.cpython-310.pycnu �[��� o �b�8 � @ sd d Z ddlZddlmZ ddlmZ ddlmZ ddlm Z m Z m Z m Z m Z mZmZmZmZ ddlmZmZmZ ddlZddlmZ dd lmZm Z m Z m Z m Z m Z m!Z! dd l"m#Z# d d l$m%Z% d dl&m'Z' d dl(m)Z)m*Z* d dl%m+Z+ G dd� dej,j-j.�Z/edd�G dd� de+��Z0G dd� de)�Z1G dd� d�Z2G dd � d �Z3G d d � d �Z4G d d � d �Z5dS )!z2 Tests for L{twisted.application.runner._runner}. � N)�StringIO)�SIGTERM)� TracebackType) �Any�Iterable�List�Optional�TextIO�Tuple�Type�Union�cast)�Factory�attrib�attrs)� MemoryReactor)�FileLogObserver�FilteringLogObserver� ILogObserver� LogBeginner�LogLevel�LogLevelFilterPredicate� LogPublisher)�FilePath� )�_runner� )� ExitStatus)� NonePIDFile�PIDFile)�Runnerc @ s� e Zd ZdZd)dee defdd�Zd*dd�Zd*d d �Z d*d d �Z d*d d�Z d*dd�Z d*dd�Z d*dd�Zd*dd�Zd*dd�Zd*dd�Zd*dd �Zd*d d �Zd*d d �Zd*d!d"�Zd*d#d$�Zd)d%ed&ee ddfd'd(�ZdS )+� RunnerTestsz Tests for L{Runner}. N�content�returnc C s" t | �� �}|d ur|�|� |S �N)r �mktemp� setContent)�selfr" �filePath� r) �M/usr/lib/python3/dist-packages/twisted/application/runner/test/test_runner.pyr( ' s zRunnerTests.filePathc C s� t � | _t� | _| �td| j� | �td| j� d| _| j� d��� | _t � | _ t � | _ t | j | j �| _ t� | _t� | _t| j| j j | j | j�| _| �td| j � | �td| j� d S )N�exit�killi9 � �stderr�globalLogBeginner)� DummyExitr+ � DummyKillr, �patchr �pid�encode�pidFileContentr �stdoutr. �DummyStandardIO�stdio�DummyWarningsModule�warningsr �globalLogPublisherr r/ �r' r) r) r* �setUp- s& �zRunnerTests.setUpc C s* t t� d�}|�� | �|jg d�� dS )zD L{Runner.run} calls the expected methods in order. ��reactor)�killIfRequested� startLogging� startReactor� reactorExitedN)� DummyRunnerr �run� assertEqual� calledMethods�r' �runnerr) r) r* �test_runInOrderO s �zRunnerTests.test_runInOrderc C sP t � }tt� |d�}| �|j� | �|j� |�� | �|j� | �|j� dS )z; L{Runner.run} uses the provided PID file. �r? �pidFileN)� DummyPIDFiler r � assertFalse�entered�exitedrE � assertTrue�r' rL rI r) r) r* �test_runUsesPIDFile` s z RunnerTests.test_runUsesPIDFilec C sV t | �| j��}dd� |_tt� |d�}|�� | �| jj t j � | �| jj d� dS )z� L{Runner.run} exits with L{ExitStatus.EX_USAGE} and the expected message if a process is already running that corresponds to the given PID file. c S � dS �NTr) r) r) r) r* �<lambda>w � z4RunnerTests.test_runAlreadyRunning.<locals>.<lambda>rK zAlready running.N) r r( r5 � isRunningr r rE rF r+ �statusr � EX_CONFIG�messagerR r) r) r* �test_runAlreadyRunningp s z"RunnerTests.test_runAlreadyRunningc C s6 t t� d�}|�� | �| jjg � | �| jj� dS )z� L{Runner.killIfRequested} when C{kill} is false doesn't exit and doesn't indiscriminately murder anyone. r> N) r r r@ rF r, �callsrN r+ rP rH r) r) r* �test_killNotRequested s z!RunnerTests.test_killNotRequestedc C sL t t� dd�}|�� | �| jjg � | �| jjtj � | �| jj d� dS )z� L{Runner.killIfRequested} when C{kill} is true but C{pidFile} is L{nonePIDFile} exits with L{ExitStatus.EX_USAGE} and the expected message; and also doesn't indiscriminately murder anyone. T)r? r, zNo PID file specified.N) r r r@ rF r, r] r+ rY r �EX_USAGEr[ rH r) r) r* � test_killRequestedWithoutPIDFile� s z,RunnerTests.test_killRequestedWithoutPIDFilec C sf t | �| j��}tt� d|d�}|�� | �| jj| j t fg� | �| j j t j� | �| j jd� dS )z� L{Runner.killIfRequested} when C{kill} is true and given a C{pidFile} performs a targeted killing of the appropriate process. T�r? r, rL N)r r( r5 r r r@ rF r, r] r3 r r+ rY r �EX_OK�assertIdenticalr[ rR r) r) r* � test_killRequestedWithPIDFile� s z)RunnerTests.test_killRequestedWithPIDFilec C s` t | �d��}dtfdd�}||_tt� d|d�}|�� | �| jj t j � | �| jj d� dS )z� L{Runner.killIfRequested} when C{kill} is true and given a C{pidFile} that it can't read exits with L{ExitStatus.EX_IOERR}. Nr# c S s t tjd��)NzPermission denied)�OSError�errno�EACCESr) r) r) r* �read� s z?RunnerTests.test_killRequestedWithPIDFileCantRead.<locals>.readTra zUnable to read PID file.) r r( �intrh r r r@ rF r+ rY r �EX_IOERRr[ )r' rL rh rI r) r) r* �%test_killRequestedWithPIDFileCantRead� s z1RunnerTests.test_killRequestedWithPIDFileCantReadc C �L t | �d��}tt� d|d�}|�� | �| jjtj � | �| jj d� dS )z� L{Runner.killIfRequested} when C{kill} is true and given a C{pidFile} containing no value exits with L{ExitStatus.EX_DATAERR}. � Tra �Invalid PID file.N� r r( r r r@ rF r+ rY r � EX_DATAERRr[ rR r) r) r* �"test_killRequestedWithPIDFileEmpty� � z.RunnerTests.test_killRequestedWithPIDFileEmptyc C rl )z� L{Runner.killIfRequested} when C{kill} is true and given a C{pidFile} containing a non-integer value exits with L{ExitStatus.EX_DATAERR}. s ** totally not a number, dude **Tra rn Nro rR r) r) r* �%test_killRequestedWithPIDFileNotAnInt� rr z1RunnerTests.test_killRequestedWithPIDFileNotAnIntc s t � }G � fdd�d�� | �td� � � G �fdd�dt��| �td�� G �fdd�dt��tt� tj|�d �}|� � | � t � j �d � | � � j d t� | � t �j�d � | � �jd t� | ��jd jtj� t��j�}| � |�� | �|j|� d S ) z� L{Runner.startLogging} sets up a filtering observer with a log level predicate set to the given log level that contains a file observer of the given type which writes to the given file. c s8 e Zd ZU g Zee ed< dee ddf� fdd� ZdS )z2RunnerTests.test_startLogging.<locals>.LogBeginner� observersr# Nc s t |�� _d S r$ )�listrt )r' rt �r r) r* �beginLoggingTo� � zARunnerTests.test_startLogging.<locals>.LogBeginner.beginLoggingTo) �__name__� __module__� __qualname__rt r r �__annotations__r rw r) rv r) r* r � s r r/ c sZ e Zd ZU dZee ed< g Zee ed< e edd� �fdede e def� fdd� Z dS ) z?RunnerTests.test_startLogging.<locals>.MockFilteringLogObserverN�observer� predicatesc S s d S r$ r) )�eventr) r) r* rV � rW zHRunnerTests.test_startLogging.<locals>.MockFilteringLogObserver.<lambda>�negativeObserverc s$ |� _ t|�� _t�| |||� d S r$ )r} ru r~ r �__init__)r' r} r~ r� ��MockFilteringLogObserverr) r* r� � s �zHRunnerTests.test_startLogging.<locals>.MockFilteringLogObserver.__init__) ry rz r{ r} r r r| r~ r r r r r� r) r� r) r* r� � s ����r� r c s4 e Zd ZU dZee ed< deddf� fdd� ZdS )z:RunnerTests.test_startLogging.<locals>.MockFileLogObserverN�outFiler# c s |� _ t�| |t� d S r$ )r� r r� �str)r' r� ��MockFileLogObserverr) r* r� � s zCRunnerTests.test_startLogging.<locals>.MockFileLogObserver.__init__)ry rz r{ r� r r r| r� r) r� r) r* r� � s r� )r? �defaultLogLevel�logFile�fileLogObserverFactory� r N)r r2 r r r r r r �criticalrA rF �lenrt �assertIsInstancer~ r rc r� r r} r� )r' r� rI r} r) )r r� r� r* �test_startLogging� s2 � �� z RunnerTests.test_startLoggingc C s( t � }t|d�}|�� | �|j� dS )ze L{Runner.startReactor} with the C{reactor} argument runs the given reactor. r> N)r r rB rQ �hasRun)r' r? rI r) r) r* � test_startReactorWithReactor s z(RunnerTests.test_startReactorWithReactorc C s | � dd� dS )z� L{Runner.startReactor} ensures that C{whenRunning} is called with C{whenRunningArguments} when the reactor is running. � whenRunningrB N�� _testHookr< r) r) r* � test_startReactorWhenRunning& s z(RunnerTests.test_startReactorWhenRunningc C � | � d� dS )zb L{Runner.whenRunning} calls C{whenRunning} with C{whenRunningArguments}. r� Nr� r< r) r) r* � test_whenRunningWithArguments- � z)RunnerTests.test_whenRunningWithArgumentsc C r� )zf L{Runner.whenRunning} calls C{reactorExited} with C{reactorExitedArguments}. rC Nr� r< r) r) r* � test_reactorExitedWithArguments4 r� z+RunnerTests.test_reactorExitedWithArguments� methodName� callerNamec s� |du r|}t t� t� t� d�}g � dtddf� fdd� }|||� d�|�� i}td dt� i|��}t||�}|� | �t� �d � | �� d |� dS ) a� Verify that the named hook is run with the expected arguments as specified by the arguments used to create the L{Runner}, when the specified caller is invoked. @param methodName: The name of the hook to verify. @param callerName: The name of the method that is expected to cause the hook to be called. If C{None}, use the L{Runner} method with the same name as the hook. N)�a�b�c� argumentsr# c s � � | � d S r$ )�append)r� �� argumentsSeenr) r* �hookN rx z#RunnerTests._testHook.<locals>.hook� Argumentsr? r� r r) )�dict�object�copyr r �getattrrF r� )r' r� r� r� r� �runnerArgumentsrI � hookCallerr) r� r* r� ; s" ��� zRunnerTests._testHookr$ �r# N)ry rz r{ �__doc__r �bytesr r( r= rJ rS r\ r^ r` rd rk rq rs r� r� r� r� r� r� r� r) r) r) r* r! " s& " M r! T)�frozenc @ sL e Zd ZdZeee ee�d�Z d dd�Z d dd�Z d d d �Z d d d �Z dS )rD zg Stub for L{Runner}. Keep track of calls to some methods without actually doing anything. )�type�defaultr# Nc C � | j �d� d S )Nr@ �rG r� r< r) r) r* r@ j � zDummyRunner.killIfRequestedc C r� )NrA r� r< r) r) r* rA m r� zDummyRunner.startLoggingc C r� )NrB r� r< r) r) r* rB p r� zDummyRunner.startReactorc C r� )NrC r� r< r) r) r* rC s r� zDummyRunner.reactorExitedr� )ry rz r{ r� r r r� r ru rG r@ rA rB rC r) r) r) r* rD ` s rD c @ sN e Zd ZdZd dd�Zddd�Zdeee d ee d ee ddfd d �Z dS )rM z] Stub for L{PIDFile}. Tracks context manager entry/exit without doing anything. r# Nc C s t �| � d| _d| _d S �NF)r r� rO rP r< r) r) r* r� ~ s zDummyPIDFile.__init__c C s d| _ | S rU )rO r< r) r) r* � __enter__� s zDummyPIDFile.__enter__�excType�excValue� tracebackc C � d| _ d S rU �rP )r' r� r� r� r) r) r* �__exit__� s zDummyPIDFile.__exit__r� )r# rM ) ry rz r{ r� r� r� r r � BaseExceptionr r� r) r) r) r* rM w s ����rM c @ s@ e Zd ZdZd dd�Z d deeef dee ddfdd �Z dS ) r0 zy Stub for L{_exit.exit} that remembers whether it's been called and, if it has, what arguments it was given. r# Nc C r� r� r� r< r) r) r* r� � � zDummyExit.__init__rY r[ c C s | j rJ �|| _|| _d| _ d S rU )rP rY r[ )r' rY r[ r) r) r* �__call__� s zDummyExit.__call__r� r$ ) ry rz r{ r� r� r ri r r r� r� r) r) r) r* r0 � s � ���r0 c @ s0 e Zd ZdZd dd�Zdededdfdd �ZdS ) r1 zv Stub for L{os.kill} that remembers whether it's been called and, if it has, what arguments it was given. r# Nc C s g | _ d S r$ )r] r< r) r) r* r� � r� zDummyKill.__init__r3 �sigc C s | j �||f� d S r$ )r] r� )r' r3 r� r) r) r* r� � s zDummyKill.__call__r� )ry rz r{ r� r� ri r� r) r) r) r* r1 � s r1 c @ �& e Zd ZdZdededdfdd�ZdS )r7 zR Stub for L{sys} which provides L{StringIO} streams as stdout and stderr. r6 r. r# Nc C s || _ || _d S r$ )r6 r. )r' r6 r. r) r) r* r� � s zDummyStandardIO.__init__)ry rz r{ r� r r� r) r) r) r* r7 � � r7 c @ r� )r9 zV Stub for L{warnings} which provides a C{showwarning} method that is a no-op. �args�kwargsr# Nc O rT )z\ Do nothing. @param args: ignored. @param kwargs: ignored. Nr) )r� r� r) r) r* � showwarning� rW z DummyWarningsModule.showwarning)ry rz r{ r� r r� r) r) r) r* r9 � r� r9 )6r� rf �ior �signalr �typesr �typingr r r r r r r r r �attrr r r �twisted.trial.unittest�twisted�twisted.internet.testingr �twisted.loggerr r r r r r r �twisted.python.filepathr rI r �_exitr �_pidfiler r r �trial�unittest�TestCaser! rD rM r0 r1 r7 r9 r) r) r) r* �<module> s2 , $ @ PK ���Z��\ � � * test/__pycache__/test_exit.cpython-310.pycnu �[��� o �b� � @ sl d Z ddlmZ ddlmZmZ ddlZddlm Z ddl m Z m Z G d d � d ej j j�ZG d d � d �ZdS ) z0 Tests for L{twisted.application.runner._exit}. � )�StringIO)�Optional�UnionN� )�_exit� )� ExitStatus�exitc @ sB e Zd ZdZddd�Zddd�Zddd �Zdd d �Zdd d �ZdS )� ExitTestsz Tests for L{exit}. �returnNc C s t � | _| �td| j� d S )N�sysexit)� DummyExitr �patchr ��self� r �K/usr/lib/python3/dist-packages/twisted/application/runner/test/test_exit.py�setUp s zExitTests.setUpc C s d}t |� | �| j j|� dS )zR L{exit} given an L{int} status code will pass it to L{sys.exit}. i� N)r � assertEqual�arg�r �statusr r r �test_exitStatusInt s z ExitTests.test_exitStatusIntc C s$ t j}t|� | �| jj|j� dS )zu L{exit} given a L{ValueConstant} status code passes the corresponding value to L{sys.exit}. N)r � EX_CONFIGr r r �valuer r r r �test_exitConstant! s zExitTests.test_exitConstantc C �: t � }| �td|� d}td|� | �|�� |d � dS )zq L{exit} given a status code of zero (C{0}) writes the given message to standard output. �stdout� Hello, world.r � N�r r r r r �getvalue�r �out�messager r r �test_exitMessageZero* � z ExitTests.test_exitMessageZeroc C r )zj L{exit} given a non-zero status code writes the given message to standard error. �stderrr �@ r Nr r" r r r �test_exitMessageNonZero7 r& z!ExitTests.test_exitMessageNonZero�r N) �__name__� __module__� __qualname__�__doc__r r r r% r) r r r r r s r c @ s: e Zd ZdZd dd�Zd deeeef ddfdd�Z dS ) r zv Stub for L{sys.exit} that remembers whether it's been called and, if it has, what argument it was given. r Nc C s d| _ d S )NF)�exitedr r r r �__init__K s zDummyExit.__init__r c C s | j rJ �|| _d| _ d S )NT)r/ r )r r r r r �__call__N s zDummyExit.__call__r* )N) r+ r, r- r. r0 r r �int�strr1 r r r r r E s $r )r. �ior �typingr r �twisted.trial.unittest�twisted�runnerr r r �trial�unittest�TestCaser r r r r r �<module> s 5PK ���ZC�;2 ;2 test/test_pidfile.pynu �[��� # Copyright (c) Twisted Matrix Laboratories. # See LICENSE for details. """ Tests for L{twisted.application.runner._pidfile}. """ import errno from functools import wraps from os import getpid, name as SYSTEM_NAME from typing import Any, Callable, Optional from zope.interface.verify import verifyObject import twisted.trial.unittest from twisted.python.filepath import FilePath from twisted.python.runtime import platform from twisted.trial.unittest import SkipTest from ...runner import _pidfile from .._pidfile import ( AlreadyRunningError, InvalidPIDFileError, IPIDFile, NonePIDFile, NoPIDFound, PIDFile, StalePIDFileError, ) def ifPlatformSupported(f: Callable[..., Any]) -> Callable[..., Any]: """ Decorator for tests that are not expected to work on all platforms. Calling L{PIDFile.isRunning} currently raises L{NotImplementedError} on non-POSIX platforms. On an unsupported platform, we expect to see any test that calls L{PIDFile.isRunning} to raise either L{NotImplementedError}, L{SkipTest}, or C{self.failureException}. (C{self.failureException} may occur in a test that checks for a specific exception but it gets NotImplementedError instead.) @param f: The test method to decorate. @return: The wrapped callable. """ @wraps(f) def wrapper(self: Any, *args: Any, **kwargs: Any) -> Any: supported = platform.getType() == "posix" if supported: return f(self, *args, **kwargs) else: e = self.assertRaises( (NotImplementedError, SkipTest, self.failureException), f, self, *args, **kwargs, ) if isinstance(e, NotImplementedError): self.assertTrue(str(e).startswith("isRunning is not implemented on ")) return wrapper class PIDFileTests(twisted.trial.unittest.TestCase): """ Tests for L{PIDFile}. """ def filePath(self, content: Optional[bytes] = None) -> FilePath: filePath = FilePath(self.mktemp()) if content is not None: filePath.setContent(content) return filePath def test_interface(self) -> None: """ L{PIDFile} conforms to L{IPIDFile}. """ pidFile = PIDFile(self.filePath()) verifyObject(IPIDFile, pidFile) def test_formatWithPID(self) -> None: """ L{PIDFile._format} returns the expected format when given a PID. """ self.assertEqual(PIDFile._format(pid=1337), b"1337\n") def test_readWithPID(self) -> None: """ L{PIDFile.read} returns the PID from the given file path. """ pid = 1337 pidFile = PIDFile(self.filePath(PIDFile._format(pid=pid))) self.assertEqual(pid, pidFile.read()) def test_readEmptyPID(self) -> None: """ L{PIDFile.read} raises L{InvalidPIDFileError} when given an empty file path. """ pidValue = b"" pidFile = PIDFile(self.filePath(b"")) e = self.assertRaises(InvalidPIDFileError, pidFile.read) self.assertEqual(str(e), f"non-integer PID value in PID file: {pidValue!r}") def test_readWithBogusPID(self) -> None: """ L{PIDFile.read} raises L{InvalidPIDFileError} when given an empty file path. """ pidValue = b"$foo!" pidFile = PIDFile(self.filePath(pidValue)) e = self.assertRaises(InvalidPIDFileError, pidFile.read) self.assertEqual(str(e), f"non-integer PID value in PID file: {pidValue!r}") def test_readDoesntExist(self) -> None: """ L{PIDFile.read} raises L{NoPIDFound} when given a non-existing file path. """ pidFile = PIDFile(self.filePath()) e = self.assertRaises(NoPIDFound, pidFile.read) self.assertEqual(str(e), "PID file does not exist") def test_readOpenRaisesOSErrorNotENOENT(self) -> None: """ L{PIDFile.read} re-raises L{OSError} if the associated C{errno} is anything other than L{errno.ENOENT}. """ def oops(mode: str = "r") -> FilePath: raise OSError(errno.EIO, "I/O error") self.patch(FilePath, "open", oops) pidFile = PIDFile(self.filePath()) error = self.assertRaises(OSError, pidFile.read) self.assertEqual(error.errno, errno.EIO) def test_writePID(self) -> None: """ L{PIDFile._write} stores the given PID. """ pid = 1995 pidFile = PIDFile(self.filePath()) pidFile._write(pid) self.assertEqual(pidFile.read(), pid) def test_writePIDInvalid(self) -> None: """ L{PIDFile._write} raises L{ValueError} when given an invalid PID. """ pidFile = PIDFile(self.filePath()) self.assertRaises(ValueError, pidFile._write, "burp") def test_writeRunningPID(self) -> None: """ L{PIDFile.writeRunningPID} stores the PID for the current process. """ pidFile = PIDFile(self.filePath()) pidFile.writeRunningPID() self.assertEqual(pidFile.read(), getpid()) def test_remove(self) -> None: """ L{PIDFile.remove} removes the PID file. """ pidFile = PIDFile(self.filePath(b"")) self.assertTrue(pidFile.filePath.exists()) pidFile.remove() self.assertFalse(pidFile.filePath.exists()) @ifPlatformSupported def test_isRunningDoesExist(self) -> None: """ L{PIDFile.isRunning} returns true for a process that does exist. """ pidFile = PIDFile(self.filePath()) pidFile._write(1337) def kill(pid: int, signal: int) -> None: return # Don't actually kill anything self.patch(_pidfile, "kill", kill) self.assertTrue(pidFile.isRunning()) @ifPlatformSupported def test_isRunningThis(self) -> None: """ L{PIDFile.isRunning} returns true for this process (which is running). @note: This differs from L{PIDFileTests.test_isRunningDoesExist} in that it actually invokes the C{kill} system call, which is useful for testing of our chosen method for probing the existence of a process. """ pidFile = PIDFile(self.filePath()) pidFile.writeRunningPID() self.assertTrue(pidFile.isRunning()) @ifPlatformSupported def test_isRunningDoesNotExist(self) -> None: """ L{PIDFile.isRunning} raises L{StalePIDFileError} for a process that does not exist (errno=ESRCH). """ pidFile = PIDFile(self.filePath()) pidFile._write(1337) def kill(pid: int, signal: int) -> None: raise OSError(errno.ESRCH, "No such process") self.patch(_pidfile, "kill", kill) self.assertRaises(StalePIDFileError, pidFile.isRunning) @ifPlatformSupported def test_isRunningNotAllowed(self) -> None: """ L{PIDFile.isRunning} returns true for a process that we are not allowed to kill (errno=EPERM). """ pidFile = PIDFile(self.filePath()) pidFile._write(1337) def kill(pid: int, signal: int) -> None: raise OSError(errno.EPERM, "Operation not permitted") self.patch(_pidfile, "kill", kill) self.assertTrue(pidFile.isRunning()) @ifPlatformSupported def test_isRunningInit(self) -> None: """ L{PIDFile.isRunning} returns true for a process that we are not allowed to kill (errno=EPERM). @note: This differs from L{PIDFileTests.test_isRunningNotAllowed} in that it actually invokes the C{kill} system call, which is useful for testing of our chosen method for probing the existence of a process that we are not allowed to kill. @note: In this case, we try killing C{init}, which is process #1 on POSIX systems, so this test is not portable. C{init} should always be running and should not be killable by non-root users. """ if SYSTEM_NAME != "posix": raise SkipTest("This test assumes POSIX") pidFile = PIDFile(self.filePath()) pidFile._write(1) # PID 1 is init on POSIX systems self.assertTrue(pidFile.isRunning()) @ifPlatformSupported def test_isRunningUnknownErrno(self) -> None: """ L{PIDFile.isRunning} re-raises L{OSError} if the attached C{errno} value from L{os.kill} is not an expected one. """ pidFile = PIDFile(self.filePath()) pidFile.writeRunningPID() def kill(pid: int, signal: int) -> None: raise OSError(errno.EEXIST, "File exists") self.patch(_pidfile, "kill", kill) self.assertRaises(OSError, pidFile.isRunning) def test_isRunningNoPIDFile(self) -> None: """ L{PIDFile.isRunning} returns false if the PID file doesn't exist. """ pidFile = PIDFile(self.filePath()) self.assertFalse(pidFile.isRunning()) def test_contextManager(self) -> None: """ When used as a context manager, a L{PIDFile} will store the current pid on entry, then removes the PID file on exit. """ pidFile = PIDFile(self.filePath()) self.assertFalse(pidFile.filePath.exists()) with pidFile: self.assertTrue(pidFile.filePath.exists()) self.assertEqual(pidFile.read(), getpid()) self.assertFalse(pidFile.filePath.exists()) @ifPlatformSupported def test_contextManagerDoesntExist(self) -> None: """ When used as a context manager, a L{PIDFile} will replace the underlying PIDFile rather than raising L{AlreadyRunningError} if the contained PID file exists but refers to a non-running PID. """ pidFile = PIDFile(self.filePath()) pidFile._write(1337) def kill(pid: int, signal: int) -> None: raise OSError(errno.ESRCH, "No such process") self.patch(_pidfile, "kill", kill) e = self.assertRaises(StalePIDFileError, pidFile.isRunning) self.assertEqual(str(e), "PID file refers to non-existing process") with pidFile: self.assertEqual(pidFile.read(), getpid()) @ifPlatformSupported def test_contextManagerAlreadyRunning(self) -> None: """ When used as a context manager, a L{PIDFile} will raise L{AlreadyRunningError} if the there is already a running process with the contained PID. """ pidFile = PIDFile(self.filePath()) pidFile._write(1337) def kill(pid: int, signal: int) -> None: return # Don't actually kill anything self.patch(_pidfile, "kill", kill) self.assertTrue(pidFile.isRunning()) self.assertRaises(AlreadyRunningError, pidFile.__enter__) class NonePIDFileTests(twisted.trial.unittest.TestCase): """ Tests for L{NonePIDFile}. """ def test_interface(self) -> None: """ L{NonePIDFile} conforms to L{IPIDFile}. """ pidFile = NonePIDFile() verifyObject(IPIDFile, pidFile) def test_read(self) -> None: """ L{NonePIDFile.read} raises L{NoPIDFound}. """ pidFile = NonePIDFile() e = self.assertRaises(NoPIDFound, pidFile.read) self.assertEqual(str(e), "PID file does not exist") def test_write(self) -> None: """ L{NonePIDFile._write} raises L{OSError} with an errno of L{errno.EPERM}. """ pidFile = NonePIDFile() error = self.assertRaises(OSError, pidFile._write, 0) self.assertEqual(error.errno, errno.EPERM) def test_writeRunningPID(self) -> None: """ L{NonePIDFile.writeRunningPID} raises L{OSError} with an errno of L{errno.EPERM}. """ pidFile = NonePIDFile() error = self.assertRaises(OSError, pidFile.writeRunningPID) self.assertEqual(error.errno, errno.EPERM) def test_remove(self) -> None: """ L{NonePIDFile.remove} raises L{OSError} with an errno of L{errno.EPERM}. """ pidFile = NonePIDFile() error = self.assertRaises(OSError, pidFile.remove) self.assertEqual(error.errno, errno.ENOENT) def test_isRunning(self) -> None: """ L{NonePIDFile.isRunning} returns L{False}. """ pidFile = NonePIDFile() self.assertEqual(pidFile.isRunning(), False) def test_contextManager(self) -> None: """ When used as a context manager, a L{NonePIDFile} doesn't raise, despite not existing. """ pidFile = NonePIDFile() with pidFile: pass PK ���Zkc� � test/test_exit.pynu �[��� # Copyright (c) Twisted Matrix Laboratories. # See LICENSE for details. """ Tests for L{twisted.application.runner._exit}. """ from io import StringIO from typing import Optional, Union import twisted.trial.unittest from ...runner import _exit from .._exit import ExitStatus, exit class ExitTests(twisted.trial.unittest.TestCase): """ Tests for L{exit}. """ def setUp(self) -> None: self.exit = DummyExit() self.patch(_exit, "sysexit", self.exit) def test_exitStatusInt(self) -> None: """ L{exit} given an L{int} status code will pass it to L{sys.exit}. """ status = 1234 exit(status) self.assertEqual(self.exit.arg, status) # type: ignore[unreachable] def test_exitConstant(self) -> None: """ L{exit} given a L{ValueConstant} status code passes the corresponding value to L{sys.exit}. """ status = ExitStatus.EX_CONFIG exit(status) self.assertEqual(self.exit.arg, status.value) # type: ignore[unreachable] def test_exitMessageZero(self) -> None: """ L{exit} given a status code of zero (C{0}) writes the given message to standard output. """ out = StringIO() self.patch(_exit, "stdout", out) message = "Hello, world." exit(0, message) self.assertEqual(out.getvalue(), message + "\n") # type: ignore[unreachable] def test_exitMessageNonZero(self) -> None: """ L{exit} given a non-zero status code writes the given message to standard error. """ out = StringIO() self.patch(_exit, "stderr", out) message = "Hello, world." exit(64, message) self.assertEqual(out.getvalue(), message + "\n") # type: ignore[unreachable] class DummyExit: """ Stub for L{sys.exit} that remembers whether it's been called and, if it has, what argument it was given. """ def __init__(self) -> None: self.exited = False def __call__(self, arg: Optional[Union[int, str]] = None) -> None: assert not self.exited self.arg = arg self.exited = True PK ���Z ud4�8 �8 test/test_runner.pynu �[��� # Copyright (c) Twisted Matrix Laboratories. # See LICENSE for details. """ Tests for L{twisted.application.runner._runner}. """ import errno from io import StringIO from signal import SIGTERM from types import TracebackType from typing import Any, Iterable, List, Optional, TextIO, Tuple, Type, Union, cast from attr import Factory, attrib, attrs import twisted.trial.unittest from twisted.internet.testing import MemoryReactor from twisted.logger import ( FileLogObserver, FilteringLogObserver, ILogObserver, LogBeginner, LogLevel, LogLevelFilterPredicate, LogPublisher, ) from twisted.python.filepath import FilePath from ...runner import _runner from .._exit import ExitStatus from .._pidfile import NonePIDFile, PIDFile from .._runner import Runner class RunnerTests(twisted.trial.unittest.TestCase): """ Tests for L{Runner}. """ def filePath(self, content: Optional[bytes] = None) -> FilePath: filePath = FilePath(self.mktemp()) if content is not None: filePath.setContent(content) return filePath def setUp(self) -> None: # Patch exit and kill so we can capture usage and prevent actual exits # and kills. self.exit = DummyExit() self.kill = DummyKill() self.patch(_runner, "exit", self.exit) self.patch(_runner, "kill", self.kill) # Patch getpid so we get a known result self.pid = 1337 self.pidFileContent = f"{self.pid}\n".encode() # Patch globalLogBeginner so that we aren't trying to install multiple # global log observers. self.stdout = StringIO() self.stderr = StringIO() self.stdio = DummyStandardIO(self.stdout, self.stderr) self.warnings = DummyWarningsModule() self.globalLogPublisher = LogPublisher() self.globalLogBeginner = LogBeginner( self.globalLogPublisher, self.stdio.stderr, self.stdio, self.warnings, ) self.patch(_runner, "stderr", self.stderr) self.patch(_runner, "globalLogBeginner", self.globalLogBeginner) def test_runInOrder(self) -> None: """ L{Runner.run} calls the expected methods in order. """ runner = DummyRunner(reactor=MemoryReactor()) runner.run() self.assertEqual( runner.calledMethods, [ "killIfRequested", "startLogging", "startReactor", "reactorExited", ], ) def test_runUsesPIDFile(self) -> None: """ L{Runner.run} uses the provided PID file. """ pidFile = DummyPIDFile() runner = Runner(reactor=MemoryReactor(), pidFile=pidFile) self.assertFalse(pidFile.entered) self.assertFalse(pidFile.exited) runner.run() self.assertTrue(pidFile.entered) self.assertTrue(pidFile.exited) def test_runAlreadyRunning(self) -> None: """ L{Runner.run} exits with L{ExitStatus.EX_USAGE} and the expected message if a process is already running that corresponds to the given PID file. """ pidFile = PIDFile(self.filePath(self.pidFileContent)) pidFile.isRunning = lambda: True # type: ignore[assignment] runner = Runner(reactor=MemoryReactor(), pidFile=pidFile) runner.run() self.assertEqual(self.exit.status, ExitStatus.EX_CONFIG) self.assertEqual(self.exit.message, "Already running.") def test_killNotRequested(self) -> None: """ L{Runner.killIfRequested} when C{kill} is false doesn't exit and doesn't indiscriminately murder anyone. """ runner = Runner(reactor=MemoryReactor()) runner.killIfRequested() self.assertEqual(self.kill.calls, []) self.assertFalse(self.exit.exited) def test_killRequestedWithoutPIDFile(self) -> None: """ L{Runner.killIfRequested} when C{kill} is true but C{pidFile} is L{nonePIDFile} exits with L{ExitStatus.EX_USAGE} and the expected message; and also doesn't indiscriminately murder anyone. """ runner = Runner(reactor=MemoryReactor(), kill=True) runner.killIfRequested() self.assertEqual(self.kill.calls, []) self.assertEqual(self.exit.status, ExitStatus.EX_USAGE) self.assertEqual(self.exit.message, "No PID file specified.") def test_killRequestedWithPIDFile(self) -> None: """ L{Runner.killIfRequested} when C{kill} is true and given a C{pidFile} performs a targeted killing of the appropriate process. """ pidFile = PIDFile(self.filePath(self.pidFileContent)) runner = Runner(reactor=MemoryReactor(), kill=True, pidFile=pidFile) runner.killIfRequested() self.assertEqual(self.kill.calls, [(self.pid, SIGTERM)]) self.assertEqual(self.exit.status, ExitStatus.EX_OK) self.assertIdentical(self.exit.message, None) def test_killRequestedWithPIDFileCantRead(self) -> None: """ L{Runner.killIfRequested} when C{kill} is true and given a C{pidFile} that it can't read exits with L{ExitStatus.EX_IOERR}. """ pidFile = PIDFile(self.filePath(None)) def read() -> int: raise OSError(errno.EACCES, "Permission denied") pidFile.read = read # type: ignore[assignment] runner = Runner(reactor=MemoryReactor(), kill=True, pidFile=pidFile) runner.killIfRequested() self.assertEqual(self.exit.status, ExitStatus.EX_IOERR) self.assertEqual(self.exit.message, "Unable to read PID file.") def test_killRequestedWithPIDFileEmpty(self) -> None: """ L{Runner.killIfRequested} when C{kill} is true and given a C{pidFile} containing no value exits with L{ExitStatus.EX_DATAERR}. """ pidFile = PIDFile(self.filePath(b"")) runner = Runner(reactor=MemoryReactor(), kill=True, pidFile=pidFile) runner.killIfRequested() self.assertEqual(self.exit.status, ExitStatus.EX_DATAERR) self.assertEqual(self.exit.message, "Invalid PID file.") def test_killRequestedWithPIDFileNotAnInt(self) -> None: """ L{Runner.killIfRequested} when C{kill} is true and given a C{pidFile} containing a non-integer value exits with L{ExitStatus.EX_DATAERR}. """ pidFile = PIDFile(self.filePath(b"** totally not a number, dude **")) runner = Runner(reactor=MemoryReactor(), kill=True, pidFile=pidFile) runner.killIfRequested() self.assertEqual(self.exit.status, ExitStatus.EX_DATAERR) self.assertEqual(self.exit.message, "Invalid PID file.") def test_startLogging(self) -> None: """ L{Runner.startLogging} sets up a filtering observer with a log level predicate set to the given log level that contains a file observer of the given type which writes to the given file. """ logFile = StringIO() # Patch the log beginner so that we don't try to start the already # running (started by trial) logging system. class LogBeginner: observers: List[ILogObserver] = [] def beginLoggingTo(self, observers: Iterable[ILogObserver]) -> None: LogBeginner.observers = list(observers) self.patch(_runner, "globalLogBeginner", LogBeginner()) # Patch FilteringLogObserver so we can capture its arguments class MockFilteringLogObserver(FilteringLogObserver): observer: Optional[ILogObserver] = None predicates: List[LogLevelFilterPredicate] = [] def __init__( self, observer: ILogObserver, predicates: Iterable[LogLevelFilterPredicate], negativeObserver: ILogObserver = cast(ILogObserver, lambda event: None), ): MockFilteringLogObserver.observer = observer MockFilteringLogObserver.predicates = list(predicates) FilteringLogObserver.__init__( self, observer, predicates, negativeObserver ) self.patch(_runner, "FilteringLogObserver", MockFilteringLogObserver) # Patch FileLogObserver so we can capture its arguments class MockFileLogObserver(FileLogObserver): outFile: Optional[TextIO] = None def __init__(self, outFile: TextIO) -> None: MockFileLogObserver.outFile = outFile FileLogObserver.__init__(self, outFile, str) # Start logging runner = Runner( reactor=MemoryReactor(), defaultLogLevel=LogLevel.critical, logFile=logFile, fileLogObserverFactory=MockFileLogObserver, ) runner.startLogging() # Check for a filtering observer self.assertEqual(len(LogBeginner.observers), 1) self.assertIsInstance(LogBeginner.observers[0], FilteringLogObserver) # Check log level predicate with the correct default log level self.assertEqual(len(MockFilteringLogObserver.predicates), 1) self.assertIsInstance( MockFilteringLogObserver.predicates[0], LogLevelFilterPredicate ) self.assertIdentical( MockFilteringLogObserver.predicates[0].defaultLogLevel, LogLevel.critical ) # Check for a file observer attached to the filtering observer observer = cast(MockFileLogObserver, MockFilteringLogObserver.observer) self.assertIsInstance(observer, MockFileLogObserver) # Check for the file we gave it self.assertIdentical(observer.outFile, logFile) def test_startReactorWithReactor(self) -> None: """ L{Runner.startReactor} with the C{reactor} argument runs the given reactor. """ reactor = MemoryReactor() runner = Runner(reactor=reactor) runner.startReactor() self.assertTrue(reactor.hasRun) def test_startReactorWhenRunning(self) -> None: """ L{Runner.startReactor} ensures that C{whenRunning} is called with C{whenRunningArguments} when the reactor is running. """ self._testHook("whenRunning", "startReactor") def test_whenRunningWithArguments(self) -> None: """ L{Runner.whenRunning} calls C{whenRunning} with C{whenRunningArguments}. """ self._testHook("whenRunning") def test_reactorExitedWithArguments(self) -> None: """ L{Runner.whenRunning} calls C{reactorExited} with C{reactorExitedArguments}. """ self._testHook("reactorExited") def _testHook(self, methodName: str, callerName: Optional[str] = None) -> None: """ Verify that the named hook is run with the expected arguments as specified by the arguments used to create the L{Runner}, when the specified caller is invoked. @param methodName: The name of the hook to verify. @param callerName: The name of the method that is expected to cause the hook to be called. If C{None}, use the L{Runner} method with the same name as the hook. """ if callerName is None: callerName = methodName arguments = dict(a=object(), b=object(), c=object()) argumentsSeen = [] def hook(**arguments: object) -> None: argumentsSeen.append(arguments) runnerArguments = { methodName: hook, f"{methodName}Arguments": arguments.copy(), } runner = Runner( reactor=MemoryReactor(), **runnerArguments # type: ignore[arg-type] ) hookCaller = getattr(runner, callerName) hookCaller() self.assertEqual(len(argumentsSeen), 1) self.assertEqual(argumentsSeen[0], arguments) @attrs(frozen=True) class DummyRunner(Runner): """ Stub for L{Runner}. Keep track of calls to some methods without actually doing anything. """ calledMethods = attrib(type=List[str], default=Factory(list)) def killIfRequested(self) -> None: self.calledMethods.append("killIfRequested") def startLogging(self) -> None: self.calledMethods.append("startLogging") def startReactor(self) -> None: self.calledMethods.append("startReactor") def reactorExited(self) -> None: self.calledMethods.append("reactorExited") class DummyPIDFile(NonePIDFile): """ Stub for L{PIDFile}. Tracks context manager entry/exit without doing anything. """ def __init__(self) -> None: NonePIDFile.__init__(self) self.entered = False self.exited = False def __enter__(self) -> "DummyPIDFile": self.entered = True return self def __exit__( self, excType: Optional[Type[BaseException]], excValue: Optional[BaseException], traceback: Optional[TracebackType], ) -> None: self.exited = True class DummyExit: """ Stub for L{_exit.exit} that remembers whether it's been called and, if it has, what arguments it was given. """ def __init__(self) -> None: self.exited = False def __call__( self, status: Union[int, ExitStatus], message: Optional[str] = None ) -> None: assert not self.exited self.status = status self.message = message self.exited = True class DummyKill: """ Stub for L{os.kill} that remembers whether it's been called and, if it has, what arguments it was given. """ def __init__(self) -> None: self.calls: List[Tuple[int, int]] = [] def __call__(self, pid: int, sig: int) -> None: self.calls.append((pid, sig)) class DummyStandardIO: """ Stub for L{sys} which provides L{StringIO} streams as stdout and stderr. """ def __init__(self, stdout: TextIO, stderr: TextIO) -> None: self.stdout = stdout self.stderr = stderr class DummyWarningsModule: """ Stub for L{warnings} which provides a C{showwarning} method that is a no-op. """ def showwarning(*args: Any, **kwargs: Any) -> None: """ Do nothing. @param args: ignored. @param kwargs: ignored. """ PK ���Z�0�� � test/__init__.pynu �[��� # -*- test-case-name: twisted.application.runner.test -*- # Copyright (c) Twisted Matrix Laboratories. # See LICENSE for details. """ Tests for L{twisted.application.runner}. """ PK ���ZNe��� � _runner.pynu �[��� # -*- test-case-name: twisted.application.runner.test.test_runner -*- # Copyright (c) Twisted Matrix Laboratories. # See LICENSE for details. """ Twisted application runner. """ from os import kill from signal import SIGTERM from sys import stderr from typing import Any, Callable, Mapping, TextIO from attr import Factory, attrib, attrs from constantly import NamedConstant # type: ignore[import] from twisted.internet.interfaces import IReactorCore from twisted.logger import ( FileLogObserver, FilteringLogObserver, Logger, LogLevel, LogLevelFilterPredicate, globalLogBeginner, textFileLogObserver, ) from ._exit import ExitStatus, exit from ._pidfile import AlreadyRunningError, InvalidPIDFileError, IPIDFile, nonePIDFile @attrs(frozen=True) class Runner: """ Twisted application runner. @cvar _log: The logger attached to this class. @ivar _reactor: The reactor to start and run the application in. @ivar _pidFile: The file to store the running process ID in. @ivar _kill: Whether this runner should kill an existing running instance of the application. @ivar _defaultLogLevel: The default log level to start the logging system with. @ivar _logFile: A file stream to write logging output to. @ivar _fileLogObserverFactory: A factory for the file log observer to use when starting the logging system. @ivar _whenRunning: Hook to call after the reactor is running; this is where the application code that relies on the reactor gets called. @ivar _whenRunningArguments: Keyword arguments to pass to C{whenRunning} when it is called. @ivar _reactorExited: Hook to call after the reactor exits. @ivar _reactorExitedArguments: Keyword arguments to pass to C{reactorExited} when it is called. """ _log = Logger() _reactor = attrib(type=IReactorCore) _pidFile = attrib(type=IPIDFile, default=nonePIDFile) _kill = attrib(type=bool, default=False) _defaultLogLevel = attrib(type=NamedConstant, default=LogLevel.info) _logFile = attrib(type=TextIO, default=stderr) _fileLogObserverFactory = attrib( type=Callable[[TextIO], FileLogObserver], default=textFileLogObserver ) _whenRunning = attrib(type=Callable[..., None], default=lambda **_: None) _whenRunningArguments = attrib(type=Mapping[str, Any], default=Factory(dict)) _reactorExited = attrib(type=Callable[..., None], default=lambda **_: None) _reactorExitedArguments = attrib(type=Mapping[str, Any], default=Factory(dict)) def run(self) -> None: """ Run this command. """ pidFile = self._pidFile self.killIfRequested() try: with pidFile: self.startLogging() self.startReactor() self.reactorExited() except AlreadyRunningError: exit(ExitStatus.EX_CONFIG, "Already running.") # When testing, patched exit doesn't exit return # type: ignore[unreachable] def killIfRequested(self) -> None: """ If C{self._kill} is true, attempt to kill a running instance of the application. """ pidFile = self._pidFile if self._kill: if pidFile is nonePIDFile: exit(ExitStatus.EX_USAGE, "No PID file specified.") # When testing, patched exit doesn't exit return # type: ignore[unreachable] try: pid = pidFile.read() except OSError: exit(ExitStatus.EX_IOERR, "Unable to read PID file.") # When testing, patched exit doesn't exit return # type: ignore[unreachable] except InvalidPIDFileError: exit(ExitStatus.EX_DATAERR, "Invalid PID file.") # When testing, patched exit doesn't exit return # type: ignore[unreachable] self.startLogging() self._log.info("Terminating process: {pid}", pid=pid) kill(pid, SIGTERM) exit(ExitStatus.EX_OK) # When testing, patched exit doesn't exit return # type: ignore[unreachable] def startLogging(self) -> None: """ Start the L{twisted.logger} logging system. """ logFile = self._logFile fileLogObserverFactory = self._fileLogObserverFactory fileLogObserver = fileLogObserverFactory(logFile) logLevelPredicate = LogLevelFilterPredicate( defaultLogLevel=self._defaultLogLevel ) filteringObserver = FilteringLogObserver(fileLogObserver, [logLevelPredicate]) globalLogBeginner.beginLoggingTo([filteringObserver]) def startReactor(self) -> None: """ Register C{self._whenRunning} with the reactor so that it is called once the reactor is running, then start the reactor. """ self._reactor.callWhenRunning(self.whenRunning) self._log.info("Starting reactor...") self._reactor.run() def whenRunning(self) -> None: """ Call C{self._whenRunning} with C{self._whenRunningArguments}. @note: This method is called after the reactor starts running. """ self._whenRunning(**self._whenRunningArguments) def reactorExited(self) -> None: """ Call C{self._reactorExited} with C{self._reactorExitedArguments}. @note: This method is called after the reactor exits. """ self._reactorExited(**self._reactorExitedArguments) PK ���Z2$�z| | _exit.pynu �[��� # -*- test-case-name: twisted.application.runner.test.test_exit -*- # Copyright (c) Twisted Matrix Laboratories. # See LICENSE for details. """ System exit support. """ import typing from enum import IntEnum from sys import exit as sysexit, stderr, stdout from typing import Union try: import posix as Status except ImportError: class Status: # type: ignore[no-redef] """ Object to hang C{EX_*} values off of as a substitute for L{posix}. """ EX__BASE = 64 EX_OK = 0 EX_USAGE = EX__BASE EX_DATAERR = EX__BASE + 1 EX_NOINPUT = EX__BASE + 2 EX_NOUSER = EX__BASE + 3 EX_NOHOST = EX__BASE + 4 EX_UNAVAILABLE = EX__BASE + 5 EX_SOFTWARE = EX__BASE + 6 EX_OSERR = EX__BASE + 7 EX_OSFILE = EX__BASE + 8 EX_CANTCREAT = EX__BASE + 9 EX_IOERR = EX__BASE + 10 EX_TEMPFAIL = EX__BASE + 11 EX_PROTOCOL = EX__BASE + 12 EX_NOPERM = EX__BASE + 13 EX_CONFIG = EX__BASE + 14 class ExitStatus(IntEnum): """ Standard exit status codes for system programs. @cvar EX_OK: Successful termination. @cvar EX_USAGE: Command line usage error. @cvar EX_DATAERR: Data format error. @cvar EX_NOINPUT: Cannot open input. @cvar EX_NOUSER: Addressee unknown. @cvar EX_NOHOST: Host name unknown. @cvar EX_UNAVAILABLE: Service unavailable. @cvar EX_SOFTWARE: Internal software error. @cvar EX_OSERR: System error (e.g., can't fork). @cvar EX_OSFILE: Critical OS file missing. @cvar EX_CANTCREAT: Can't create (user) output file. @cvar EX_IOERR: Input/output error. @cvar EX_TEMPFAIL: Temporary failure; the user is invited to retry. @cvar EX_PROTOCOL: Remote error in protocol. @cvar EX_NOPERM: Permission denied. @cvar EX_CONFIG: Configuration error. """ EX_OK = Status.EX_OK EX_USAGE = Status.EX_USAGE EX_DATAERR = Status.EX_DATAERR EX_NOINPUT = Status.EX_NOINPUT EX_NOUSER = Status.EX_NOUSER EX_NOHOST = Status.EX_NOHOST EX_UNAVAILABLE = Status.EX_UNAVAILABLE EX_SOFTWARE = Status.EX_SOFTWARE EX_OSERR = Status.EX_OSERR EX_OSFILE = Status.EX_OSFILE EX_CANTCREAT = Status.EX_CANTCREAT EX_IOERR = Status.EX_IOERR EX_TEMPFAIL = Status.EX_TEMPFAIL EX_PROTOCOL = Status.EX_PROTOCOL EX_NOPERM = Status.EX_NOPERM EX_CONFIG = Status.EX_CONFIG def exit(status: Union[int, ExitStatus], message: str = "") -> "typing.NoReturn": """ Exit the python interpreter with the given status and an optional message. @param status: An exit status. An appropriate value from L{ExitStatus} is recommended. @param message: An optional message to print. """ if message: if status == ExitStatus.EX_OK: out = stdout else: out = stderr out.write(message) out.write("\n") sysexit(status) PK ���Z� �� � __init__.pynu �[��� # -*- test-case-name: twisted.application.runner.test -*- # Copyright (c) Twisted Matrix Laboratories. # See LICENSE for details. """ Facilities for running a Twisted application. """ PK ���Z�`��� � ! __pycache__/_exit.cpython-310.pycnu �[��� PK ���Z N�K� � $ 6 __pycache__/__init__.cpython-310.pycnu �[��� PK ���Z� �$ $ $ g __pycache__/_pidfile.cpython-310.pycnu �[��� PK ���Z���� � # �0 __pycache__/_runner.cpython-310.pycnu �[��� PK ���Z� � � �E _pidfile.pynu �[��� PK ���Z�]r��7 �7 - �b test/__pycache__/test_pidfile.cpython-310.pycnu �[��� PK ���Z��� � ) !� test/__pycache__/__init__.cpython-310.pycnu �[��� PK ���Z�{KQ< Q< , W� test/__pycache__/test_runner.cpython-310.pycnu �[��� PK ���Z��\ � � * � test/__pycache__/test_exit.cpython-310.pycnu �[��� PK ���ZC�;2 ;2 F� test/test_pidfile.pynu �[��� PK ���Zkc� � � test/test_exit.pynu �[��� PK ���Z ud4�8 �8 � test/test_runner.pynu �[��� PK ���Z�0�� � X test/__init__.pynu �[��� PK ���ZNe��� � sY _runner.pynu �[��� PK ���Z2$�z| | �o _exit.pynu �[��� PK ���Z� �� � ]{ __init__.pynu �[��� PK � Q|
| ver. 1.4 |
Github
|
.
| PHP 8.2.28 | Generation time: 0.02 |
proxy
|
phpinfo
|
Settings