


"""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


lowerCamelCase_       :       Optional[Any]      						=  get_tests_dir("""fixtures""")






class 				__A    (	unittest.TestCase      ):






 """simple docstring"""




 def 				SCREAMING_SNAKE_CASE							(		self						) ->					int:
     # A mock response for an HTTP head request to emulate server down
     a					=mock.Mock()
     a					=500
     a					={}
     a					=HTTPError
     a					={}

     # Download this model to make sure it's in the cache.
     a					=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=__A						) as mock_head:
         a					=WavaVecaFeatureExtractor.from_pretrained('''hf-internal-testing/tiny-random-wav2vec2'''						)
         # This check we did call the fake head request
         mock_head.assert_called()





 def 				SCREAMING_SNAKE_CASE							(		self						) ->					int:
     # This test is for deprecated behavior and can be removed in v5
     a					=WavaVecaFeatureExtractor.from_pretrained(
         '''https://huggingface.co/hf-internal-testing/tiny-random-wav2vec2/resolve/main/preprocessor_config.json'''						)






@is_staging_test
class 				__A    (	unittest.TestCase      ):






 """simple docstring"""




 @classmethod
 def 				SCREAMING_SNAKE_CASE							(		cls						) ->					List[Any]:
     a					=TOKEN
     HfFolder.save_token(__A						)


 @classmethod
 def 				SCREAMING_SNAKE_CASE							(		cls						) ->					List[str]:
     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 				SCREAMING_SNAKE_CASE							(		self						) ->					int:
     a					=WavaVecaFeatureExtractor.from_pretrained(__A						)
     feature_extractor.push_to_hub('''test-feature-extractor'''			,   use_auth_token=self._token						)

     a					=WavaVecaFeatureExtractor.from_pretrained(f'''{USER}/test-feature-extractor'''						)
     for k, v in feature_extractor.__dict__.items():
         self.assertEqual(__A			,   getattr(__A			,   __A						)						)

     # 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(
             __A			,   repo_id='''test-feature-extractor'''			,   push_to_hub=__A			,   use_auth_token=self._token						)

     a					=WavaVecaFeatureExtractor.from_pretrained(f'''{USER}/test-feature-extractor'''						)
     for k, v in feature_extractor.__dict__.items():
         self.assertEqual(__A			,   getattr(__A			,   __A						)						)


 def 				SCREAMING_SNAKE_CASE							(		self						) ->					Any:
     a					=WavaVecaFeatureExtractor.from_pretrained(__A						)
     feature_extractor.push_to_hub('''valid_org/test-feature-extractor'''			,   use_auth_token=self._token						)

     a					=WavaVecaFeatureExtractor.from_pretrained('''valid_org/test-feature-extractor'''						)
     for k, v in feature_extractor.__dict__.items():
         self.assertEqual(__A			,   getattr(__A			,   __A						)						)

     # 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(
             __A			,   repo_id='''valid_org/test-feature-extractor-org'''			,   push_to_hub=__A			,   use_auth_token=self._token						)

     a					=WavaVecaFeatureExtractor.from_pretrained('''valid_org/test-feature-extractor-org'''						)
     for k, v in feature_extractor.__dict__.items():
         self.assertEqual(__A			,   getattr(__A			,   __A						)						)





 def 				SCREAMING_SNAKE_CASE							(		self						) ->					Tuple:
     CustomFeatureExtractor.register_for_auto_class()
     a					=CustomFeatureExtractor.from_pretrained(__A						)

     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'''}			,   )

     a					=AutoFeatureExtractor.from_pretrained(
         f'''{USER}/test-dynamic-feature-extractor'''			,   trust_remote_code=__A						)
     # 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'''						)



"""simple docstring"""







from __future__ import annotations

from typing import Generic, TypeVar

lowerCamelCase_       :       List[Any]      						=  TypeVar("""T""")






class 				__A    (	Generic[T]      ):






 """simple docstring"""




 def __init__(		self			,   __A						) ->					None:
     a					=data
     a					=self
     a					=0






class 				__A    (	Generic[T]      ):






 """simple docstring"""




 def __init__(		self						) ->					None:
     # map from node name to the node object
     a					={}


 def 				SCREAMING_SNAKE_CASE							(		self			,   __A						) ->					None:
     # create a new set with x as its member
     a					=DisjointSetTreeNode(__A						)


 def 				SCREAMING_SNAKE_CASE							(		self			,   __A						) ->					DisjointSetTreeNode[T]:
     # find the set x belongs to (with path-compression)
     a					=self.map[data]
     if elem_ref != elem_ref.parent:
         a					=self.find_set(elem_ref.parent.data						)
     return elem_ref.parent


 def 				SCREAMING_SNAKE_CASE							(		self			,   __A			,   __A						) ->					None:
     # helper function for union operation
     if nodea.rank > nodea.rank:
         a					=nodea
     else:
         a					=nodea
         if nodea.rank == nodea.rank:
             nodea.rank += 1





 def 				SCREAMING_SNAKE_CASE							(		self			,   __A			,   __A						) ->					None:
     # merge 2 disjoint sets
     self.link(self.find_set(__A						)			,   self.find_set(__A						)						)






class 				__A    (	Generic[T]      ):






 """simple docstring"""




 def __init__(		self						) ->					None:
     # connections: map from the node to the neighbouring nodes (with weights)
     a					={}


 def 				SCREAMING_SNAKE_CASE							(		self			,   __A						) ->					None:
     # add a node ONLY if its not present in the graph
     if node not in self.connections:
         a					={}


 def 				SCREAMING_SNAKE_CASE							(		self			,   __A			,   __A			,   __A						) ->					None:
     # add an edge with the given weight
     self.add_node(__A						)
     self.add_node(__A						)
     a					=weight
     a					=weight





 def 				SCREAMING_SNAKE_CASE							(		self						) ->					GraphUndirectedWeighted[T]:
     a					=[]
     a					=set()
     for start in self.connections:
         for end in self.connections[start]:
             if (start, end) not in seen:
                 seen.add((end, start)						)
                 edges.append((start, end, self.connections[start][end])						)
     edges.sort(key=lambda __A						: x[2]						)

     # creating the disjoint set
     a					=DisjointSetTree[T]()
     for node in self.connections:
         disjoint_set.make_set(__A						)

     # MST generation
     a					=0
     a					=0
     a					=GraphUndirectedWeighted[T]()
     while num_edges < len(self.connections						) - 1:
         a		,				a		,				a					=edges[index]
         index += 1
         a					=disjoint_set.find_set(__A						)
         a					=disjoint_set.find_set(__A						)
         if parent_u != parent_v:
             num_edges += 1
             graph.add_edge(__A			,   __A			,   __A						)
             disjoint_set.union(__A			,   __A						)
     return graph