'''simple docstring'''




from __future__ import annotations

import unittest

from transformers import LEDConfig, is_tf_available
from transformers.testing_utils import require_tf, slow

from ...test_configuration_common import ConfigTester
from ...test_modeling_tf_common import TFModelTesterMixin, ids_tensor
from ...test_pipeline_mixin import PipelineTesterMixin


if is_tf_available():
  import tensorflow as tf

  from transformers import TFLEDForConditionalGeneration, TFLEDModel




@require_tf
class     a__      :
     lowerCamelCase  :     Tuple							=LEDConfig
     lowerCamelCase  :     Tuple							={}
     lowerCamelCase  :     int							="gelu"






     def __init__( self		:		Union[str, Any]  ,				a		:		str  ,				a		:		str=13  ,				a		:		Tuple=7  ,				a		:		Any=True  ,				a		:		str=False  ,				a		:		str=99  ,				a		:		str=32  ,				a		:		str=2  ,				a		:		int=4  ,				a		:		Dict=37  ,				a		:		Any=0.1  ,				a		:		int=0.1  ,				a		:		List[Any]=20  ,				a		:		Optional[Any]=2  ,				a		:		int=1  ,				a		:		Union[str, Any]=0  ,				a		:		List[Any]=4  ,				):





         """simple docstring"""



         __lowerCamelCase		=  parent
         __lowerCamelCase		=  batch_size
         __lowerCamelCase		=  seq_length
         __lowerCamelCase		=  is_training
         __lowerCamelCase		=  use_labels
         __lowerCamelCase		=  vocab_size
         __lowerCamelCase		=  hidden_size
         __lowerCamelCase		=  num_hidden_layers
         __lowerCamelCase		=  num_attention_heads
         __lowerCamelCase		=  intermediate_size

         __lowerCamelCase		=  hidden_dropout_prob
         __lowerCamelCase		=  attention_probs_dropout_prob
         __lowerCamelCase		=  max_position_embeddings
         __lowerCamelCase		=  eos_token_id
         __lowerCamelCase		=  pad_token_id
         __lowerCamelCase		=  bos_token_id
         __lowerCamelCase		=  attention_window

         # `ModelTesterMixin.test_attention_outputs` is expecting attention tensors to be of size
         # [num_attention_heads, encoder_seq_length, encoder_key_length], but TFLongformerSelfAttention
         # returns attention of shape [num_attention_heads, encoder_seq_length, self.attention_window + 1]
         # because its local attention only attends to `self.attention_window` and one before and one after
         __lowerCamelCase		=  self.attention_window + 2

         # because of padding `encoder_seq_length`, is different from `seq_length`. Relevant for
         # the `test_attention_outputs` and `test_hidden_states_output` tests
         __lowerCamelCase		=  (
             self.seq_length + (self.attention_window - self.seq_length % self.attention_window) % self.attention_window
         )






     def    SCREAMING_SNAKE_CASE__   ( self		:		str    ):





         """simple docstring"""



         __lowerCamelCase		=  ids_tensor([self.batch_size, self.seq_length - 1]  ,				self.vocab_size    )
         __lowerCamelCase		=  tf.expand_dims(tf.constant([self.eos_token_id] * self.batch_size    )  ,				1    )
         __lowerCamelCase		=  tf.concat([input_ids, eos_tensor]  ,				axis=1    )

         __lowerCamelCase		=  ids_tensor([self.batch_size, self.seq_length]  ,				self.vocab_size    )

         __lowerCamelCase		=  self.config_cls(
             vocab_size=self.vocab_size  ,				d_model=self.hidden_size  ,				encoder_layers=self.num_hidden_layers  ,				decoder_layers=self.num_hidden_layers  ,				encoder_attention_heads=self.num_attention_heads  ,				decoder_attention_heads=self.num_attention_heads  ,				encoder_ffn_dim=self.intermediate_size  ,				decoder_ffn_dim=self.intermediate_size  ,				dropout=self.hidden_dropout_prob  ,				attention_dropout=self.attention_probs_dropout_prob  ,				max_position_embeddings=self.max_position_embeddings  ,				eos_token_ids=[2]  ,				bos_token_id=self.bos_token_id  ,				pad_token_id=self.pad_token_id  ,				decoder_start_token_id=self.pad_token_id  ,				attention_window=self.attention_window  ,				**self.config_updates  ,				)
         __lowerCamelCase		=  prepare_led_inputs_dict(a  ,				a  ,				a    )
         __lowerCamelCase		=  tf.concat(
             [tf.zeros_like(a    )[:, :-1], tf.ones_like(a    )[:, -1:]]  ,				axis=-1  ,				)
         __lowerCamelCase		=  global_attention_mask
         return config, inputs_dict






     def    SCREAMING_SNAKE_CASE__   ( self		:		Optional[int]  ,				a		:		Dict  ,				a		:		Optional[Any]    ):





         """simple docstring"""



         __lowerCamelCase		=  TFLEDModel(config=a    ).get_decoder()
         __lowerCamelCase		=  inputs_dict["""input_ids"""]

         __lowerCamelCase		=  input_ids[:1, :]
         __lowerCamelCase		=  inputs_dict["""attention_mask"""][:1, :]
         __lowerCamelCase		=  1

         # first forward pass
         __lowerCamelCase		=  model(a  ,				attention_mask=a  ,				use_cache=a    )

         __lowerCamelCase		=  outputs.to_tuple()

         # create hypothetical next token and extent to next_input_ids
         __lowerCamelCase		=  ids_tensor((self.batch_size, 3)  ,				config.vocab_size    )
         __lowerCamelCase		=  tf.cast(ids_tensor((self.batch_size, 3)  ,				2    )  ,				tf.inta    )

         # append to next input_ids and
         __lowerCamelCase		=  tf.concat([input_ids, next_tokens]  ,				axis=-1    )
         __lowerCamelCase		=  tf.concat([attention_mask, next_attn_mask]  ,				axis=-1    )

         __lowerCamelCase		=  model(a  ,				attention_mask=a    )[0]
         __lowerCamelCase		=  model(a  ,				attention_mask=a  ,				past_key_values=a    )[0]

         self.parent.assertEqual(next_tokens.shape[1]  ,				output_from_past.shape[1]    )

         # select random slice
         __lowerCamelCase		=  int(ids_tensor((1,)  ,				output_from_past.shape[-1]    )    )
         __lowerCamelCase		=  output_from_no_past[:, -3:, random_slice_idx]
         __lowerCamelCase		=  output_from_past[:, :, random_slice_idx]

         # test that outputs are equal for slice
         tf.debugging.assert_near(a  ,				a  ,				rtol=1e-3    )






def __lowerCAmelCase (							UpperCamelCase__				,  UpperCamelCase__				,  UpperCamelCase__				,  UpperCamelCase__=None				,  UpperCamelCase__=None				,  UpperCamelCase__=None				,  UpperCamelCase__=None				,  )	->    Any:





    if attention_mask is None:
        __lowerCamelCase		=  tf.cast(tf.math.not_equal(A__				,  config.pad_token_id				)				,  tf.inta				)
    if decoder_attention_mask is None:
        __lowerCamelCase		=  tf.concat(
            [
                tf.ones(decoder_input_ids[:, :1].shape				,  dtype=tf.inta				),
                tf.cast(tf.math.not_equal(decoder_input_ids[:, 1:]				,  config.pad_token_id				)				,  tf.inta				),
            ]				,  axis=-1				,  )
    if head_mask is None:
        __lowerCamelCase		=  tf.ones((config.encoder_layers, config.encoder_attention_heads)				)
    if decoder_head_mask is None:
        __lowerCamelCase		=  tf.ones((config.decoder_layers, config.decoder_attention_heads)				)
    return {
        "input_ids": input_ids,
        "attention_mask": attention_mask,
        "decoder_input_ids": decoder_input_ids,
        "decoder_attention_mask": decoder_attention_mask,
        "head_mask": head_mask,
        "decoder_head_mask": decoder_head_mask,
    }




@require_tf
class     a__      (   _SCREAMING_SNAKE_CASE							,   _SCREAMING_SNAKE_CASE							,   unittest.TestCase ):
     lowerCamelCase  :     Any							=(TFLEDForConditionalGeneration, TFLEDModel) if is_tf_available() else ()
     lowerCamelCase  :     List[Any]							=(TFLEDForConditionalGeneration,) if is_tf_available() else ()
     lowerCamelCase  :     str							=(
         {
             "conversational": TFLEDForConditionalGeneration,
             "feature-extraction": TFLEDModel,
             "summarization": TFLEDForConditionalGeneration,
             "text2text-generation": TFLEDForConditionalGeneration,
             "translation": TFLEDForConditionalGeneration,
         }
         if is_tf_available()
         else {}
     )
     lowerCamelCase  :     Dict							=True
     lowerCamelCase  :     Tuple							=False
     lowerCamelCase  :     List[str]							=False
     lowerCamelCase  :     Any							=False






     def    SCREAMING_SNAKE_CASE__   ( self		:		Any    ):





         """simple docstring"""



         __lowerCamelCase		=  TFLEDModelTester(self    )
         __lowerCamelCase		=  ConfigTester(self  ,				config_class=a    )






     def    SCREAMING_SNAKE_CASE__   ( self		:		List[str]    ):





         """simple docstring"""



         self.config_tester.run_common_tests()






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





         """simple docstring"""



         __lowerCamelCase		=  self.model_tester.prepare_config_and_inputs_for_common()
         self.model_tester.check_decoder_model_past_large_inputs(*a    )






     def    SCREAMING_SNAKE_CASE__   ( self		:		Tuple    ):





         """simple docstring"""



         __lowerCamelCase		=  self.model_tester.prepare_config_and_inputs_for_common()
         __lowerCamelCase		=  tf.zeros_like(inputs_dict['''attention_mask''']    )
         __lowerCamelCase		=  2
         __lowerCamelCase		=  tf.where(
             tf.range(self.model_tester.seq_length    )[None, :] < num_global_attn_indices  ,				1  ,				inputs_dict['''global_attention_mask''']  ,				)

         __lowerCamelCase		=  True
         __lowerCamelCase		=  self.model_tester.seq_length
         __lowerCamelCase		=  self.model_tester.encoder_seq_length

         def check_decoder_attentions_output(a		:		Union[str, Any]    ):
             __lowerCamelCase		=  outputs.decoder_attentions
             self.assertEqual(len(a    )  ,				self.model_tester.num_hidden_layers    )
             self.assertListEqual(
                 list(decoder_attentions[0].shape[-3:]    )  ,				[self.model_tester.num_attention_heads, seq_length, seq_length]  ,				)

         def check_encoder_attentions_output(a		:		int    ):
             __lowerCamelCase		=  [t.numpy() for t in outputs.encoder_attentions]
             __lowerCamelCase		=  [t.numpy() for t in outputs.encoder_global_attentions]
             self.assertEqual(len(a    )  ,				self.model_tester.num_hidden_layers    )
             self.assertEqual(len(a    )  ,				self.model_tester.num_hidden_layers    )
             self.assertListEqual(
                 list(attentions[0].shape[-3:]    )  ,				[self.model_tester.num_attention_heads, seq_length, seq_length]  ,				)
             self.assertListEqual(
                 list(global_attentions[0].shape[-3:]    )  ,				[self.model_tester.num_attention_heads, encoder_seq_length, num_global_attn_indices]  ,				)

         for model_class in self.all_model_classes:
             __lowerCamelCase		=  True
             __lowerCamelCase		=  False
             __lowerCamelCase		=  False
             __lowerCamelCase		=  model_class(a    )
             __lowerCamelCase		=  model(self._prepare_for_class(a  ,				a    )    )
             __lowerCamelCase		=  len(a    )
             self.assertEqual(config.output_hidden_states  ,				a    )
             check_encoder_attentions_output(a    )

             if self.is_encoder_decoder:
                 __lowerCamelCase		=  model_class(a    )
                 __lowerCamelCase		=  model(self._prepare_for_class(a  ,				a    )    )
                 self.assertEqual(config.output_hidden_states  ,				a    )
                 check_decoder_attentions_output(a    )

             # Check that output attentions can also be changed via the config
             del inputs_dict["output_attentions"]
             __lowerCamelCase		=  True
             __lowerCamelCase		=  model_class(a    )
             __lowerCamelCase		=  model(self._prepare_for_class(a  ,				a    )    )
             self.assertEqual(config.output_hidden_states  ,				a    )
             check_encoder_attentions_output(a    )

             # Check attention is always last and order is fine
             __lowerCamelCase		=  True
             __lowerCamelCase		=  True
             __lowerCamelCase		=  model_class(a    )
             __lowerCamelCase		=  model(self._prepare_for_class(a  ,				a    )    )

             self.assertEqual(out_len + (2 if self.is_encoder_decoder else 1)  ,				len(a    )    )
             self.assertEqual(model.config.output_hidden_states  ,				a    )
             check_encoder_attentions_output(a    )






     @unittest.skip('''LED keeps using potentially symbolic tensors in conditionals and breaks tracing.'''    )
     def    SCREAMING_SNAKE_CASE__   ( self		:		List[str]    ):





         """simple docstring"""



         pass






     def    SCREAMING_SNAKE_CASE__   ( self		:		Dict    ):





         """simple docstring"""



         pass






def __lowerCAmelCase (							UpperCamelCase__				)	->    Tuple:
    return tf.constant(A__				,  dtype=tf.intaa				)


__UpperCAmelCase     							=1e-4




@slow
@require_tf
class     a__      (   unittest.TestCase ):






     def    SCREAMING_SNAKE_CASE__   ( self		:		Any    ):





         """simple docstring"""



         __lowerCamelCase		=  TFLEDForConditionalGeneration.from_pretrained('''allenai/led-base-16384'''    ).led

         # change to intended input here
         __lowerCamelCase		=  _long_tensor([5_12 * [0, 3_14_14, 2_32, 3_28, 7_40, 11_40, 1_26_95, 69]]    )
         __lowerCamelCase		=  _long_tensor([1_28 * [0, 3_14_14, 2_32, 3_28, 7_40, 11_40, 1_26_95, 69]]    )
         __lowerCamelCase		=  prepare_led_inputs_dict(model.config  ,				a  ,				a    )
         __lowerCamelCase		=  model(**a    )[0]
         __lowerCamelCase		=  (1, 10_24, 7_68)
         self.assertEqual(output.shape  ,				a    )
         # change to expected output here
         __lowerCamelCase		=  tf.convert_to_tensor(
             [[2.30_50, 2.82_79, 0.65_31], [-1.84_57, -0.14_55, -3.56_61], [-1.01_86, 0.45_86, -2.20_43]]  ,				)
         tf.debugging.assert_near(output[:, :3, :3]  ,				a  ,				atol=1e-3    )






     def    SCREAMING_SNAKE_CASE__   ( self		:		List[Any]    ):





         """simple docstring"""



         __lowerCamelCase		=  TFLEDForConditionalGeneration.from_pretrained('''allenai/led-base-16384'''    )

         # change to intended input here
         __lowerCamelCase		=  _long_tensor([5_12 * [0, 3_14_14, 2_32, 3_28, 7_40, 11_40, 1_26_95, 69]]    )
         __lowerCamelCase		=  _long_tensor([1_28 * [0, 3_14_14, 2_32, 3_28, 7_40, 11_40, 1_26_95, 69]]    )
         __lowerCamelCase		=  prepare_led_inputs_dict(model.config  ,				a  ,				a    )
         __lowerCamelCase		=  model(**a    )[0]
         __lowerCamelCase		=  (1, 10_24, model.config.vocab_size)
         self.assertEqual(output.shape  ,				a    )
         # change to expected output here
         __lowerCamelCase		=  tf.convert_to_tensor(
             [[33.65_07, 6.45_72, 16.80_89], [5.87_39, -2.42_38, 11.29_02], [-3.21_39, -4.31_49, 4.27_83]]  ,				)
         tf.debugging.assert_near(output[:, :3, :3]  ,				a  ,				atol=1e-3  ,				rtol=1e-3    )




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



