o

�N�g��@s�dd
lZdd
l
Z
dd
lZdd
lZdd
lZdd
lZdd
lZdd
lZddl	m
Z
mZ
dd
lZz
ddl
mZmZ
WneyG


ddlmZmZ
Yn
we
�e
�d�
dfg
�
Zdd�Zdd	�Zd
d�Zdd
�ZGdd�de�ZGdd�dej�Zdd�Zd
S)�N)�
renormalizing�
setupstack)�Mock�patchzzc\.lockfile\.LockError:z
LockError:cCs�	ztj
�d�
}Wntj
jy


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

|d
7}|
�
d�

|
�d|�d�
�

|
�
�
|�
�
dS)	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�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)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()
    Nrrrrr�pid_in_lockfileKrrc
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):
      ...
    LockError: Couldn't lock 'f.lock'

    >>> f = open("f.lock")
    >>> _ = f.seek(1)
    >>> f.read().rstrip()
    'myhostname'
    >>> f.close()

    >>> lock.close()
    Nrrrrr�hostname_in_lockfilegrr 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@s0eZ
dZd
Zdd�Zdd�Zdd�Zdd	�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
rrr
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
rr�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_runningN)r+r,r-�__doc__r6r8r;rHrrrrr.�s

r.c
CsTt�
�}|�tjd
ttjtjd��

|�tj	tjtjtd��

|�tj
�t�
�

|S)Nz
README.txt)�checkerr6r8)r6r8rJ)
�unittestZ	TestSuiteZaddTest�doctestZDocFileSuiterJrZsetUpDirectoryr8ZDocTestSuiteZdefaultTestLoaderZloadTestsFromNamer+)
Zsuiterrr�
test_suite�s



�

�
rM)r0�re�sysrKrLZzc.lockfiler
r�	threadingZzope.testingrrr3Z
unittest.mockrr�ImportErrorZmockZ
RENormalizing�compilerJrrrr �objectr!ZTestCaser.rMrrrr�<module>
s*(







�
��)