golden hour
/opt/saltstack/salt/lib/python3.10/site-packages/zc/lockfile/__pycache__
⬆️ Go Up
Upload
File/Folder
Size
Actions
__init__.cpython-310.pyc
2.73 KB
Del
OK
tests.cpython-310.pyc
5.8 KB
Del
OK
Edit: tests.cpython-310.pyc
o �xe� � @ s� d dl Z d dlZd dlZd dlZd dlZd dlZd dlZd dlZd dl m Z m Z d dl Z z d dl mZmZ W n eyG d dlmZmZ Y nw e �e�d�dfg�Zdd� Zdd � Zd d � Zd d � ZG dd� de�ZG dd� dej�Z dd� Z dS )� N)� renormalizing� setupstack)�Mock�patchzzc\.lockfile\.LockError:z LockError:c C s� zt j�d�} W n t jjy Y q w tdd�}t|�� �� �}t� d� |d7 }|� d� |� d| � d�� |� � | � � d S ) N� �f.lock�fzr+bg{�G�z�?r z%d �ASCII)�zc�lockfile�LockFileZ LockError�open�int�readline�strip�time�sleep�seek�write�encode�close)�lockr �v� r �E/opt/saltstack/salt/lib/python3.10/site-packages/zc/lockfile/tests.py�inc s � r c C � 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. � r c C r )a- >>> 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): ... 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_lockfileK r r c C r )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): ... 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_lockfileg r 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 @ s0 e Zd ZdZdd� Zdd� Zdd� Zdd � 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�tempfileZmkdtemp�tmp�chdirr$ r r r �setUp� s z LockFileLogEntryTestCase.setUpc C s t �| j� t�| j� d S r" )r0 r5 r2 r �rmtreer4 r$ r r r �tearDown� s z!LockFileLogEntryTestCase.tearDownc C s� t dtdd���Q t dtdd���0 tjjddd�}td��}| �d |�� � W d � n1 s1w Y |�� W d � n1 sDw Y W d � d S W d � d S 1 s\w Y d S ) Nz os.getpid�{ )Z 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 ZQueueZProcess�get�daemon�startr Z assertTrue�is_aliveZput�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_runningN)r+ r, r- �__doc__r6 r8 r; rF r r r r r. � s r. c C sT t �� } | �tjdttjtjd�� | �tj tjtjtd�� | �t j � t �� | S )Nz README.txt)�checkerr6 r8 )r6 r8 rH ) �unittestZ TestSuiteZaddTest�doctestZ DocFileSuiterH r ZsetUpDirectoryr8 Z DocTestSuiteZdefaultTestLoaderZloadTestsFromNamer+ )Zsuiter r r � test_suite� s ��rK ) r0 �re�sysrI rJ Z zc.lockfiler r � threadingZ zope.testingr r r3 Z unittest.mockr r � ImportErrorZmockZ RENormalizing�compilerH r r r r �objectr! ZTestCaser. rK r r r r �<module> s* ( ��� )
Save