


import logging
import os
import threading
import time


try:
						import warnings
except ImportError:
						__UpperCamelCase     		=							None

try:
						import msvcrt
except ImportError:
						__UpperCamelCase     		=							None

try:
						import fcntl
except ImportError:
						__UpperCamelCase     		=							None


# Backward compatibility
# ------------------------------------------------
try:
						TimeoutError
except NameError:
						__UpperCamelCase     		=							OSError


# Data
# ------------------------------------------------
__UpperCamelCase     		=							[
    'Timeout',
    'BaseFileLock',
    'WindowsFileLock',
    'UnixFileLock',
    'SoftFileLock',
    'FileLock',
]

__UpperCamelCase     		=							'3.0.12'


__UpperCamelCase     		=							None
def 							UpperCamelCase_(					)->		Optional[int]:
			global _logger
			UpperCamelCase__       =							_logger or logging.getLogger(__name__	)
			return _logger







class 							lowerCamelCase__						(				UpperCAmelCase   ):






			"""simple docstring"""






			def __init__(				self   ,    snake_case  ):

						'''simple docstring'''


						UpperCamelCase__       =							lock_file
						return None







			def __str__(				self  ):

						'''simple docstring'''


						UpperCamelCase__       =							F'''The file lock \'{self.lock_file}\' could not be acquired.'''
						return temp







class 							lowerCamelCase__						:






			"""simple docstring"""






			def __init__(				self   ,    snake_case  ):

						'''simple docstring'''


						UpperCamelCase__       =							lock
						return None




			def __enter__(				self  ):

						'''simple docstring'''


						return self.lock







			def __exit__(				self   ,    snake_case   ,    snake_case   ,    snake_case  ):

						'''simple docstring'''


						self.lock.release()
						return None







class 							lowerCamelCase__						:






			"""simple docstring"""






			def __init__(				self   ,    snake_case   ,    snake_case=-1   ,    snake_case=None  ):

						'''simple docstring'''


						UpperCamelCase__       =							max_filename_length if max_filename_length is not None else 255
						# Hash the filename if it's too long
						UpperCamelCase__       =							self.hash_filename_if_too_long(snake_case   ,    snake_case  )
						# The path to the lock file.
						UpperCamelCase__       =							lock_file

						# The file descriptor for the *_lock_file* as it is returned by the
						# os.open() function.
						# This file lock is only NOT None, if the object currently holds the
						# lock.
						UpperCamelCase__       =							None

						# The default timeout value.
						UpperCamelCase__       =							timeout

						# We use this lock primarily for the lock counter.
						UpperCamelCase__       =							threading.Lock()

						# The lock counter is used for implementing the nested locking
						# mechanism. Whenever the lock is acquired, the counter is increased and
						# the lock is only released, when this value is 0 again.
						UpperCamelCase__       =							0
						return None




			@property
			def 					snake_case__					(				self  ):

						'''simple docstring'''


						return self._lock_file




			@property
			def 					snake_case__					(				self  ):

						'''simple docstring'''


						return self._timeout




			@timeout.setter
			def 					snake_case__					(				self   ,    snake_case  ):

						'''simple docstring'''


						UpperCamelCase__       =							float(snake_case  )
						return None




			def 					snake_case__					(				self  ):

						'''simple docstring'''


						raise NotImplementedError()




			def 					snake_case__					(				self  ):

						'''simple docstring'''


						raise NotImplementedError()




			@property
			def 					snake_case__					(				self  ):

						'''simple docstring'''


						return self._lock_file_fd is not None




			def 					snake_case__					(				self   ,    snake_case=None   ,    snake_case=0.05  ):

						'''simple docstring'''


						if timeout is None:
									UpperCamelCase__       =							self.timeout

						# Increment the number right at the beginning.
						# We can still undo it, if something fails.
						with self._thread_lock:
									self._lock_counter += 1

						UpperCamelCase__       =							id(self  )
						UpperCamelCase__       =							self._lock_file
						UpperCamelCase__       =							time.time()
						try:
									while True:
												with self._thread_lock:
															if not self.is_locked:
																		logger().debug(F'''Attempting to acquire lock {lock_id} on {lock_filename}'''  )
																		self._acquire()

												if self.is_locked:
															logger().debug(F'''Lock {lock_id} acquired on {lock_filename}'''  )
															break
												elif timeout >= 0 and time.time() - start_time > timeout:
															logger().debug(F'''Timeout on acquiring lock {lock_id} on {lock_filename}'''  )
															raise Timeout(self._lock_file  )
												else:
															logger().debug(
															    F'''Lock {lock_id} not acquired on {lock_filename}, waiting {poll_intervall} seconds ...'''  )
															time.sleep(snake_case  )
						except:  # noqa
									# Something did go wrong, so decrement the counter.
									with self._thread_lock:
												UpperCamelCase__       =							max(0   ,    self._lock_counter - 1  )

									raise
						return _Acquire_ReturnProxy(lock=self  )




			def 					snake_case__					(				self   ,    snake_case=False  ):

						'''simple docstring'''


						with self._thread_lock:
									if self.is_locked:
												self._lock_counter -= 1

												if self._lock_counter == 0 or force:
															UpperCamelCase__       =							id(self  )
															UpperCamelCase__       =							self._lock_file

															logger().debug(F'''Attempting to release lock {lock_id} on {lock_filename}'''  )
															self._release()
															UpperCamelCase__       =							0
															logger().debug(F'''Lock {lock_id} released on {lock_filename}'''  )

						return None




			def __enter__(				self  ):

						'''simple docstring'''


						self.acquire()
						return self




			def __exit__(				self   ,    snake_case   ,    snake_case   ,    snake_case  ):

						'''simple docstring'''


						self.release()
						return None




			def __del__(				self  ):

						'''simple docstring'''


						self.release(force=snake_case  )
						return None







			def 					snake_case__					(				self   ,    snake_case   ,    snake_case  ):

						'''simple docstring'''


						UpperCamelCase__       =							os.path.basename(snake_case  )
						if len(snake_case  ) > max_length and max_length > 0:
									UpperCamelCase__       =							os.path.dirname(snake_case  )
									UpperCamelCase__       =							str(hash(snake_case  )  )
									UpperCamelCase__       =							filename[: max_length - len(snake_case  ) - 8] + "..." + hashed_filename + ".lock"
									return os.path.join(snake_case   ,    snake_case  )
						else:
									return path







class 							lowerCamelCase__						(				UpperCAmelCase   ):






			"""simple docstring"""






			def __init__(				self   ,    snake_case   ,    snake_case=-1   ,    snake_case=None  ):

						'''simple docstring'''


						from .file_utils import relative_to_absolute_path

						super().__init__(snake_case   ,    timeout=snake_case   ,    max_filename_length=snake_case  )
						UpperCamelCase__       =							"\\\\?\\" + relative_to_absolute_path(self.lock_file  )




			def 					snake_case__					(				self  ):

						'''simple docstring'''


						UpperCamelCase__       =							os.O_RDWR | os.O_CREAT | os.O_TRUNC

						try:
									UpperCamelCase__       =							os.open(self._lock_file   ,    snake_case  )
						except OSError:
									pass
						else:
									try:
												msvcrt.locking(snake_case   ,    msvcrt.LK_NBLCK   ,    1  )
									except OSError:
												os.close(snake_case  )
									else:
												UpperCamelCase__       =							fd
						return None







			def 					snake_case__					(				self  ):

						'''simple docstring'''


						UpperCamelCase__       =							self._lock_file_fd
						UpperCamelCase__       =							None
						msvcrt.locking(snake_case   ,    msvcrt.LK_UNLCK   ,    1  )
						os.close(snake_case  )

						try:
									os.remove(self._lock_file  )
						# Probably another instance of the application
						# that acquired the file lock.
						except OSError:
									pass
						return None







class 							lowerCamelCase__						(				UpperCAmelCase   ):






			"""simple docstring"""






			def __init__(				self   ,    snake_case   ,    snake_case=-1   ,    snake_case=None  ):

						'''simple docstring'''


						UpperCamelCase__       =							os.statvfs(os.path.dirname(snake_case  )  ).f_namemax
						super().__init__(snake_case   ,    timeout=snake_case   ,    max_filename_length=snake_case  )




			def 					snake_case__					(				self  ):

						'''simple docstring'''


						UpperCamelCase__       =							os.O_RDWR | os.O_CREAT | os.O_TRUNC
						UpperCamelCase__       =							os.open(self._lock_file   ,    snake_case  )

						try:
									fcntl.flock(snake_case   ,    fcntl.LOCK_EX | fcntl.LOCK_NB  )
						except OSError:
									os.close(snake_case  )
						else:
									UpperCamelCase__       =							fd
						return None







			def 					snake_case__					(				self  ):

						'''simple docstring'''


						UpperCamelCase__       =							self._lock_file_fd
						UpperCamelCase__       =							None
						fcntl.flock(snake_case   ,    fcntl.LOCK_UN  )
						os.close(snake_case  )
						return None







class 							lowerCamelCase__						(				UpperCAmelCase   ):






			"""simple docstring"""






			def 					snake_case__					(				self  ):

						'''simple docstring'''


						UpperCamelCase__       =							os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_TRUNC
						try:
									UpperCamelCase__       =							os.open(self._lock_file   ,    snake_case  )
						except OSError:
									pass
						else:
									UpperCamelCase__       =							fd
						return None







			def 					snake_case__					(				self  ):

						'''simple docstring'''


						os.close(self._lock_file_fd  )
						UpperCamelCase__       =							None

						try:
									os.remove(self._lock_file  )
						# The file is already deleted and that's what we want.
						except OSError:
									pass
						return None



__UpperCamelCase     		=							None

if msvcrt:
						__UpperCamelCase     		=							WindowsFileLock
elif fcntl:
						__UpperCamelCase     		=							UnixFileLock
else:
						__UpperCamelCase     		=							SoftFileLock

						if warnings is not None:
												warnings.warn('only soft file lock is available')






from unittest.mock import Mock, patch

from file_transfer.send_file import send_file
@patch("socket.socket"	)
@patch("builtins.open"	)
def 							UpperCamelCase_(					_A :Tuple					,  _A :str	)->		int:
			# ===== initialization =====
			UpperCamelCase__       =							Mock()
			UpperCamelCase__       =							conn, Mock()
			UpperCamelCase__       =							iter([1, None]	)
			UpperCamelCase__       =							lambda _A	: next(_A	)

			# ===== invoke =====
			send_file(filename="mytext.txt"					,  testing=_A	)

			# ===== ensurance =====
			sock.assert_called_once()
			sock.return_value.bind.assert_called_once()
			sock.return_value.listen.assert_called_once()
			sock.return_value.accept.assert_called_once()
			conn.recv.assert_called_once()

			file.return_value.__enter__.assert_called_once()
			file.return_value.__enter__.return_value.read.assert_called()

			conn.send.assert_called_once()
			conn.close.assert_called_once()
			sock.return_value.shutdown.assert_called_once()
			sock.return_value.close.assert_called_once()


