'''simple docstring'''




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 __lowerCAmelCase (							)	->    List[str]:
    global _logger
    __lowerCamelCase		=  _logger or logging.getLogger(__name__				)
    return _logger




class     a__      (   _SCREAMING_SNAKE_CASE ):






     def __init__( self		:		Dict  ,				a		:		List[Any]    ):





         """simple docstring"""



         __lowerCamelCase		=  lock_file
         return None






     def __str__( self		:		str    ):





         """simple docstring"""



         __lowerCamelCase		=  f"""The file lock \'{self.lock_file}\' could not be acquired."""
         return temp




class     a__      :






     def __init__( self		:		Dict  ,				a		:		Union[str, Any]    ):





         """simple docstring"""



         __lowerCamelCase		=  lock
         return None






     def __enter__( self		:		Union[str, Any]    ):





         """simple docstring"""



         return self.lock






     def __exit__( self		:		Any  ,				a		:		int  ,				a		:		str  ,				a		:		Tuple    ):





         """simple docstring"""



         self.lock.release()
         return None




class     a__      :






     def __init__( self		:		Optional[Any]  ,				a		:		int  ,				a		:		Optional[int]=-1  ,				a		:		Optional[Any]=None    ):





         """simple docstring"""



         __lowerCamelCase		=  max_filename_length if max_filename_length is not None else 2_55
         # Hash the filename if it's too long
         __lowerCamelCase		=  self.hash_filename_if_too_long(a  ,				a    )
         # The path to the lock file.
         __lowerCamelCase		=  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.
         __lowerCamelCase		=  None

         # The default timeout value.
         __lowerCamelCase		=  timeout

         # We use this lock primarily for the lock counter.
         __lowerCamelCase		=  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.
         __lowerCamelCase		=  0
         return None






     @property
     def    SCREAMING_SNAKE_CASE__   ( self		:		Dict    ):





         """simple docstring"""



         return self._lock_file






     @property
     def    SCREAMING_SNAKE_CASE__   ( self		:		Dict    ):





         """simple docstring"""



         return self._timeout






     @timeout.setter
     def    SCREAMING_SNAKE_CASE__   ( self		:		Optional[int]  ,				a		:		int    ):





         """simple docstring"""



         __lowerCamelCase		=  float(a    )
         return None






     def    SCREAMING_SNAKE_CASE__   ( self		:		Optional[int]    ):





         """simple docstring"""



         raise NotImplementedError()






     def    SCREAMING_SNAKE_CASE__   ( self		:		Dict    ):





         """simple docstring"""



         raise NotImplementedError()






     @property
     def    SCREAMING_SNAKE_CASE__   ( self		:		Optional[int]    ):





         """simple docstring"""



         return self._lock_file_fd is not None






     def    SCREAMING_SNAKE_CASE__   ( self		:		Any  ,				a		:		List[Any]=None  ,				a		:		List[str]=0.05    ):





         """simple docstring"""



         if timeout is None:
             __lowerCamelCase		=  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

         __lowerCamelCase		=  id(self    )
         __lowerCamelCase		=  self._lock_file
         __lowerCamelCase		=  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(a    )
         except:  # noqa
             # Something did go wrong, so decrement the counter.
             with self._thread_lock:
                 __lowerCamelCase		=  max(0  ,				self._lock_counter - 1    )

             raise
         return _Acquire_ReturnProxy(lock=self    )






     def    SCREAMING_SNAKE_CASE__   ( self		:		str  ,				a		:		int=False    ):





         """simple docstring"""



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

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

                     logger().debug(f"""Attempting to release lock {lock_id} on {lock_filename}"""    )
                     self._release()
                     __lowerCamelCase		=  0
                     logger().debug(f"""Lock {lock_id} released on {lock_filename}"""    )

         return None






     def __enter__( self		:		Union[str, Any]    ):





         """simple docstring"""



         self.acquire()
         return self






     def __exit__( self		:		Tuple  ,				a		:		List[str]  ,				a		:		Any  ,				a		:		Optional[Any]    ):





         """simple docstring"""



         self.release()
         return None






     def __del__( self		:		str    ):





         """simple docstring"""



         self.release(force=a    )
         return None






     def    SCREAMING_SNAKE_CASE__   ( self		:		Tuple  ,				a		:		str  ,				a		:		int    ):





         """simple docstring"""



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




class     a__      (   _SCREAMING_SNAKE_CASE ):






     def __init__( self		:		List[Any]  ,				a		:		Dict  ,				a		:		Union[str, Any]=-1  ,				a		:		Optional[int]=None    ):





         """simple docstring"""



         from .file_utils import relative_to_absolute_path

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






     def    SCREAMING_SNAKE_CASE__   ( self		:		List[str]    ):





         """simple docstring"""



         __lowerCamelCase		=  os.O_RDWR | os.O_CREAT | os.O_TRUNC

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






     def    SCREAMING_SNAKE_CASE__   ( self		:		List[Any]    ):





         """simple docstring"""



         __lowerCamelCase		=  self._lock_file_fd
         __lowerCamelCase		=  None
         msvcrt.locking(a  ,				msvcrt.LK_UNLCK  ,				1    )
         os.close(a    )

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




class     a__      (   _SCREAMING_SNAKE_CASE ):






     def __init__( self		:		Any  ,				a		:		Optional[int]  ,				a		:		int=-1  ,				a		:		str=None    ):





         """simple docstring"""



         __lowerCamelCase		=  os.statvfs(os.path.dirname(a    )    ).f_namemax
         super().__init__(a  ,				timeout=a  ,				max_filename_length=a    )






     def    SCREAMING_SNAKE_CASE__   ( self		:		Any    ):





         """simple docstring"""



         __lowerCamelCase		=  os.O_RDWR | os.O_CREAT | os.O_TRUNC
         __lowerCamelCase		=  os.open(self._lock_file  ,				a    )

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






     def    SCREAMING_SNAKE_CASE__   ( self		:		Optional[Any]    ):





         """simple docstring"""



         __lowerCamelCase		=  self._lock_file_fd
         __lowerCamelCase		=  None
         fcntl.flock(a  ,				fcntl.LOCK_UN    )
         os.close(a    )
         return None




class     a__      (   _SCREAMING_SNAKE_CASE ):






     def    SCREAMING_SNAKE_CASE__   ( self		:		Optional[Any]    ):





         """simple docstring"""



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






     def    SCREAMING_SNAKE_CASE__   ( self		:		List[str]    ):





         """simple docstring"""



         os.close(self._lock_file_fd    )
         __lowerCamelCase		=  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")




'''simple docstring'''






import sys
import tempfile
import unittest
import unittest.mock as mock
from pathlib import Path

from huggingface_hub import HfFolder, delete_repo
from requests.exceptions import HTTPError

from transformers import AutoFeatureExtractor, WavaVecaFeatureExtractor
from transformers.testing_utils import TOKEN, USER, get_tests_dir, is_staging_test


sys.path.append(str(Path(__file__).parent.parent / "utils"))

from test_module.custom_feature_extraction import CustomFeatureExtractor  # noqa E402


__A					:  str											=      get_tests_dir("fixtures")






class    __snake_case       (   unittest.TestCase):







    """simple docstring"""



    def __lowercase   (   self    :  int		)   ->  List[str]:
      # A mock response for an HTTP head request to emulate server down
      lowerCAmelCase_			:      str			=   mock.Mock()
      lowerCAmelCase_			:      Optional[Any]			=   5_00
      lowerCAmelCase_			:      List[str]			=   {}
      lowerCAmelCase_			:      str			=   HTTPError
      lowerCAmelCase_			:      Any			=   {}

      # Download this model to make sure it's in the cache.
      lowerCAmelCase_			:      Optional[Any]			=   WavaVecaFeatureExtractor.from_pretrained("""hf-internal-testing/tiny-random-wav2vec2"""		)
      # Under the mock environment we get a 500 error when trying to reach the model.
      with mock.patch("""requests.Session.request"""    ,						return_value=lowerCamelCase		) as mock_head:
        lowerCAmelCase_			:      Union[str, Any]			=   WavaVecaFeatureExtractor.from_pretrained("""hf-internal-testing/tiny-random-wav2vec2"""		)
        # This check we did call the fake head request
        mock_head.assert_called()





    def __lowercase   (   self    :  Dict		)   ->  Any:
      # This test is for deprecated behavior and can be removed in v5
      lowerCAmelCase_			:      List[str]			=   WavaVecaFeatureExtractor.from_pretrained(
          """https://huggingface.co/hf-internal-testing/tiny-random-wav2vec2/resolve/main/preprocessor_config.json"""		)







@is_staging_test
class    __snake_case       (   unittest.TestCase):







    """simple docstring"""



    @classmethod
    def __lowercase   (   cls    :  Tuple		)   ->  str:
      lowerCAmelCase_			:      Dict			=   TOKEN
      HfFolder.save_token(lowerCamelCase		)

    @classmethod
    def __lowercase   (   cls    :  Any		)   ->  Any:
      try:
        delete_repo(token=cls._token    ,						repo_id="""test-feature-extractor"""		)
      except HTTPError:
        pass

      try:
        delete_repo(token=cls._token    ,						repo_id="""valid_org/test-feature-extractor-org"""		)
      except HTTPError:
        pass

      try:
        delete_repo(token=cls._token    ,						repo_id="""test-dynamic-feature-extractor"""		)
      except HTTPError:
        pass

    def __lowercase   (   self    :  int		)   ->  str:
      lowerCAmelCase_			:      Tuple			=   WavaVecaFeatureExtractor.from_pretrained(lowerCamelCase		)
      feature_extractor.push_to_hub("""test-feature-extractor"""    ,						use_auth_token=self._token		)

      lowerCAmelCase_			:      Optional[Any]			=   WavaVecaFeatureExtractor.from_pretrained(F'{USER}/test-feature-extractor'		)
      for k, v in feature_extractor.__dict__.items():
        self.assertEqual(lowerCamelCase    ,						getattr(lowerCamelCase    ,						lowerCamelCase		)		)

      # Reset repo
      delete_repo(token=self._token    ,						repo_id="""test-feature-extractor"""		)

      # Push to hub via save_pretrained
      with tempfile.TemporaryDirectory() as tmp_dir:
        feature_extractor.save_pretrained(
            lowerCamelCase    ,						repo_id="""test-feature-extractor"""    ,						push_to_hub=lowerCamelCase    ,						use_auth_token=self._token		)

      lowerCAmelCase_			:      Any			=   WavaVecaFeatureExtractor.from_pretrained(F'{USER}/test-feature-extractor'		)
      for k, v in feature_extractor.__dict__.items():
        self.assertEqual(lowerCamelCase    ,						getattr(lowerCamelCase    ,						lowerCamelCase		)		)

    def __lowercase   (   self    :  Optional[Any]		)   ->  int:
      lowerCAmelCase_			:      Union[str, Any]			=   WavaVecaFeatureExtractor.from_pretrained(lowerCamelCase		)
      feature_extractor.push_to_hub("""valid_org/test-feature-extractor"""    ,						use_auth_token=self._token		)

      lowerCAmelCase_			:      Optional[int]			=   WavaVecaFeatureExtractor.from_pretrained("""valid_org/test-feature-extractor"""		)
      for k, v in feature_extractor.__dict__.items():
        self.assertEqual(lowerCamelCase    ,						getattr(lowerCamelCase    ,						lowerCamelCase		)		)

      # Reset repo
      delete_repo(token=self._token    ,						repo_id="""valid_org/test-feature-extractor"""		)

      # Push to hub via save_pretrained
      with tempfile.TemporaryDirectory() as tmp_dir:
        feature_extractor.save_pretrained(
            lowerCamelCase    ,						repo_id="""valid_org/test-feature-extractor-org"""    ,						push_to_hub=lowerCamelCase    ,						use_auth_token=self._token		)

      lowerCAmelCase_			:      Optional[int]			=   WavaVecaFeatureExtractor.from_pretrained("""valid_org/test-feature-extractor-org"""		)
      for k, v in feature_extractor.__dict__.items():
        self.assertEqual(lowerCamelCase    ,						getattr(lowerCamelCase    ,						lowerCamelCase		)		)





    def __lowercase   (   self    :  Optional[Any]		)   ->  Any:
      CustomFeatureExtractor.register_for_auto_class()
      lowerCAmelCase_			:      Dict			=   CustomFeatureExtractor.from_pretrained(lowerCamelCase		)

      feature_extractor.push_to_hub("""test-dynamic-feature-extractor"""    ,						use_auth_token=self._token		)

      # This has added the proper auto_map field to the config
      self.assertDictEqual(
          feature_extractor.auto_map    ,						{"""AutoFeatureExtractor""": """custom_feature_extraction.CustomFeatureExtractor"""}    ,						)

      lowerCAmelCase_			:      Union[str, Any]			=   AutoFeatureExtractor.from_pretrained(
          F'{USER}/test-dynamic-feature-extractor'    ,						trust_remote_code=lowerCamelCase		)
      # Can't make an isinstance check because the new_feature_extractor is from the CustomFeatureExtractor class of a dynamic module
      self.assertEqual(new_feature_extractor.__class__.__name__    ,						"""CustomFeatureExtractor"""		)



