o

;j��@s�dd
lZdd
l
Z
dd
lZdd
lZdd
lZdd
lZddlmZ
ddlmZ
ddl	m
Z

dd
lZdd�Z
dd�Zd	d
�Zdd�ZGd
d�d�ZGdd�dej�Zdd�Zd
S)�N)
�Mock)
�patch)
�
setupstackcCs�	ztj
�d�
}Wntj
jy


Yqw	tdd�}
t|
�����
}t�	d�

|d7}|
�
d�

|
�d|�d	�
�

|
�
�
|�
�
dS)
NT�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�E/opt/saltstack/salt/lib/python3.10/site-packages/zc/lockfile/tests.py�incs




�










rc
C�d
S)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')

    Nrrrrr�many_threads_read_and_write.�rc
Cr)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()
    Nrrrrr�pid_in_lockfileLrrc
Cr)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()
    Nrrrrr�hostname_in_lockfilehrr c@seZ
dZd
d�Zdd�ZdS)�
TestLoggerc

Cs
g|_dS�
N)
�log_entries�
�selfrrr�__init__�s

zTestLogger.__init__cGs|j�
|
f
|�

dSr")r#�append)r%�msg�argsrrr�	exception�s
zTestLogger.exceptionN)�__name__�
__module__�__qualname__r&r*rrrrr!�s
r!c@s8eZ
dZd
Zdd�Zdd�Zdd�Zdd	�Zd
d�ZdS)
�LockFileLogEntryTestCasez)Tests for logging in case of lock failurec

Cs(t�
�|_tjd
d�
|_t�|j�

dS)Nzzc.lockfile-test-)
�prefix)�os�getcwd�here�tempfileZmkdtemp�tmp�chdirr$rrr�setUp�s



zLockFileLogEntryTestCase.setUpc

Cst�
|j�

t�|j�

dSr")r0r5r2r�rmtreer4r$rrr�tearDown�s

z!LockFileLogEntryTestCase.tearDownc

Cs�td
t
dd�
��Q
tdt
dd�
��0
tjjddd�}
td�
�}|�d	|���
Wd�
n1s1w



Y
|
��
Wd�
n1sDw



Y
Wd�
dSWd�
dS1s\w



Y
dS)
Nz	os.getpid�{)
Zreturn_valuezsocket.gethostnameZ
myhostnamerz{pid}/{hostname})
Zcontent_templatez 123/myhostname
)	rrr	r
rr
ZassertEqual�readr)r%rrrrr�test_log_formatting�s


�

�
��
"�z,LockFileLogEntryTestCase.test_log_formattingc
Csxd
dl}
t
j�d�
}|
��}|
j|jd�
}d|_|��
|�	�
t
j�d�
}|�
|���

|�d
�

|�	�
|�
�
dS)Nr
�
l)
�targetT)�multiprocessingr	r
r�QueueZProcess�get�daemon�startrZ
assertTrue�is_alive�put�join)r%r>r�
q�
prrr�:test_unlock_and_lock_while_multiprocessing_process_running�s









zSLockFileLogEntryTestCase.test_unlock_and_lock_while_multiprocessing_process_runningc
Csttt
jjt�s	J�
t
j�d
�
}
|�t
jj�
�
t
j�d
�

Wd�
n1s'w



Y
|
��
t
j�d
�
��
dS)N�
s)�
isinstancer	r
ZSimpleLockFile�typeZassertRaisesrr)r%rrrr�test_simple_lock�s



�
z)LockFileLogEntryTestCase.test_simple_lockN)	r+r,r-�__doc__r6r8r;rHrLrrrrr.�s
r.c
CsPt�
�}|�tjd
tjtjd��

|�tjtjtjd��

|�tj	�
t�
�

|S)Nz
README.txt)r6r8)�unittestZ	TestSuiteZaddTest�doctestZDocFileSuiterZsetUpDirectoryr8ZDocTestSuiteZdefaultTestLoaderZloadTestsFromNamer+)
Zsuiterrr�
test_suite�s



�
�
rP)rOr0r3�	threadingrrNZ
unittest.mockrrZzope.testingrZzc.lockfiler	rrrr r!ZTestCaser.rPrrrr�<module>
s"







 2