Файловый менеджер - Редактировать - /opt/imh-python/lib/python3.9/site-packages/zc/lockfile/__pycache__/tests.cpython-39.pyc
Ðазад
a i�h� � @ s� d dl Z d dlZd dlZd dlZd dlZd dlZd dlmZ d dlmZ d dl m Z d dl Z dd� Z dd� Zd d � Zd d � ZG d d� d�ZG dd� dej�Zdd� ZdS )� N)�Mock)�patch)� setupstackc C s� zt j�d�} W q0 t jjy* Y q Y q 0 q0q tdd�}t|�� �� �}t� d� |d7 }|� d� |� d| � d�� |� � | � � d S ) N�f.lock�fzr+bg{�G�z�?� r z%d �ASCII)�zc�lockfile�LockFile� LockError�open�int�readline�strip�time�sleep�seek�write�encode�close)�lockr �v� r ��/root/rpmbuild/BUILDROOT/imh-python39-modules-3.9.7-92.el8.x86_64/opt/imh-python/lib/python3.9/site-packages/zc/lockfile/tests.py�inc s r c C s dS )a� >>> with open('f', 'w+b') as file: ... _ = file.write(b'0\n') >>> with open('f.lock', 'w+b') as file: ... _ = file.write(b'0\n') >>> n = 50 >>> threads = [threading.Thread(target=inc) for i in range(n)] >>> _ = [thread.start() for thread in threads] >>> _ = [thread.join() for thread in threads] >>> with open('f', 'rb') as file: ... saved = int(file.read().strip()) >>> saved == n True >>> os.remove('f') We should only have one pid in the lock file: >>> f = open('f.lock') >>> len(f.read().strip().split()) 1 >>> f.close() >>> os.remove('f.lock') Nr r r r r �many_threads_read_and_write. s r c C s dS )a9 >>> import os, zc.lockfile >>> pid = os.getpid() >>> lock = zc.lockfile.LockFile("f.lock") >>> f = open("f.lock") >>> _ = f.seek(1) >>> f.read().strip() == str(pid) True >>> f.close() Make sure that locking twice does not overwrite the old pid: >>> lock = zc.lockfile.LockFile("f.lock") Traceback (most recent call last): ... zc.lockfile.LockError: Couldn't lock 'f.lock' >>> f = open("f.lock") >>> _ = f.seek(1) >>> f.read().strip() == str(pid) True >>> f.close() >>> lock.close() Nr r r r r �pid_in_lockfileL s r c C s dS )a' hostname is correctly written into the lock file when it's included in the lock file content template >>> import zc.lockfile >>> with patch('socket.gethostname', Mock(return_value='myhostname')): ... lock = zc.lockfile.LockFile( ... "f.lock", content_template='{hostname}') >>> f = open("f.lock") >>> _ = f.seek(1) >>> f.read().rstrip() 'myhostname' >>> f.close() Make sure that locking twice does not overwrite the old hostname: >>> lock = zc.lockfile.LockFile("f.lock", content_template='{hostname}') Traceback (most recent call last): ... zc.lockfile.LockError: Couldn't lock 'f.lock' >>> f = open("f.lock") >>> _ = f.seek(1) >>> f.read().rstrip() 'myhostname' >>> f.close() >>> lock.close() Nr r r r r �hostname_in_lockfileh s r c @ s e Zd Zdd� Zdd� ZdS )� TestLoggerc C s g | _ d S �N)� log_entries��selfr r r �__init__� s zTestLogger.__init__c G s | j �|f| � d S r )r! �append)r# �msg�argsr r r � exception� s zTestLogger.exceptionN)�__name__� __module__� __qualname__r$ r( r r r r r � s r c @ s8 e Zd ZdZdd� Zdd� Zdd� Zdd � Zd d � Zd S ) �LockFileLogEntryTestCasez)Tests for logging in case of lock failurec C s( t �� | _tjdd�| _t �| j� d S )Nzzc.lockfile-test-)�prefix)�os�getcwd�here�tempfile�mkdtemp�tmp�chdirr" r r r �setUp� s z LockFileLogEntryTestCase.setUpc C s t �| j� t�| j� d S r )r. r4 r0 r �rmtreer3 r" r r r �tearDown� s z!LockFileLogEntryTestCase.tearDownc C s� t dtdd���� t dtdd���` tjjddd�}td�� }| �d |�� � W d � n1 sb0 Y |�� W d � n1 s�0 Y W d � n1 s�0 Y d S ) Nz os.getpid�{ )� return_valuezsocket.gethostnameZ myhostnamer z{pid}/{hostname})Zcontent_templatez 123/myhostname ) r r r r r r Z assertEqual�readr )r# r r r r r �test_log_formatting� s � .z,LockFileLogEntryTestCase.test_log_formattingc C sx dd l }tj�d�}|�� }|j|jd�}d|_|�� |� � tj�d�}| � |� � � |� d� |� � |� � d S )Nr �l)�targetT)�multiprocessingr r r �Queue�Process�get�daemon�startr Z assertTrue�is_alive�put�join)r# r> r �q�pr r r �:test_unlock_and_lock_while_multiprocessing_process_running� s zSLockFileLogEntryTestCase.test_unlock_and_lock_while_multiprocessing_process_runningc C st t tjjt�sJ �tj�d�}| �tjj�� tj�d� W d � n1 sN0 Y |�� tj�d��� d S )N�s)� isinstancer r ZSimpleLockFile�typeZ assertRaisesr r )r# r r r r �test_simple_lock� s *z)LockFileLogEntryTestCase.test_simple_lockN) r) r* r+ �__doc__r5 r7 r; rI rM r r r r r, � s r, c C sP t �� } | �tjdtjtjd�� | �tjtjtjd�� | �t j � t �� | S )Nz README.txt)r5 r7 ) �unittestZ TestSuiteZaddTest�doctestZ DocFileSuiter ZsetUpDirectoryr7 Z DocTestSuiteZdefaultTestLoaderZloadTestsFromNamer) )Zsuiter r r � test_suite� s ��rQ )rP r. r1 � threadingr rO Z unittest.mockr r Z zope.testingr Z zc.lockfiler r r r r r ZTestCaser, rQ r r r r �<module> s 2
| ver. 1.1 | |
.
| PHP 8.3.30 | Ð“ÐµÐ½ÐµÑ€Ð°Ñ†Ð¸Ñ Ñтраницы: 0.01 |
proxy
|
phpinfo
|
ÐаÑтройка