





"""simple docstring"""
from __future__ import annotations

import unittest

from transformers import BlenderbotSmallConfig, BlenderbotSmallTokenizer, is_tf_available
from transformers.testing_utils import require_tf, require_tokenizers, slow
from transformers.utils import cached_property

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 TFAutoModelForSeqaSeqLM, TFBlenderbotSmallForConditionalGeneration, TFBlenderbotSmallModel







@require_tf
class       a							:
 snake_case__             =       BlenderbotSmallConfig
 snake_case__             =       {}
 snake_case__             =       '''gelu'''



 def __init__(     self				,       _snake_case				,       _snake_case=13				,       _snake_case=7				,       _snake_case=True				,       _snake_case=False				,       _snake_case=99				,       _snake_case=32				,       _snake_case=2				,       _snake_case=4				,       _snake_case=37				,       _snake_case=0.1				,       _snake_case=0.1				,       _snake_case=20				,       _snake_case=2				,       _snake_case=1				,       _snake_case=0				,       ):







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



 def    UpperCamelCase__  (     self     ):







  """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				,       **self.config_updates				,       )
  lowerCAmelCase    				= prepare_blenderbot_small_inputs_dict(_snake_case				,       _snake_case				,       _snake_case     )
  return config, inputs_dict



 def    UpperCamelCase__  (     self				,       _snake_case				,       _snake_case     ):







  """simple docstring"""






  lowerCAmelCase    				= TFBlenderbotSmallModel(config=_snake_case     ).get_decoder()
  lowerCAmelCase    				= inputs_dict['input_ids']

  lowerCAmelCase    				= input_ids[:1, :]
  lowerCAmelCase    				= inputs_dict['attention_mask'][:1, :]
  lowerCAmelCase    				= inputs_dict['head_mask']
  lowerCAmelCase    				= 1

  # first forward pass
  lowerCAmelCase    				= model(_snake_case				,       attention_mask=_snake_case				,       head_mask=_snake_case				,       use_cache=_snake_case     )

  lowerCAmelCase				,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(_snake_case				,       attention_mask=_snake_case     )[0]
  lowerCAmelCase    				= model(_snake_case				,       attention_mask=_snake_case				,       past_key_values=_snake_case     )[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(_snake_case				,       _snake_case				,       rtol=1E-3     )


def 		_SCREAMING_SNAKE_CASE			(_UpperCAmelCase					:    str  ,     _UpperCAmelCase					:    List[Any]  ,     _UpperCAmelCase					:    Optional[Any]  ,     _UpperCAmelCase					:    Optional[int]=None  ,     _UpperCAmelCase					:    Dict=None  ,     _UpperCAmelCase					:    Optional[int]=None  ,     _UpperCAmelCase					:    Union[str, Any]=None  ,     _UpperCAmelCase					:    str=None  ,     ):
 if attention_mask is None:
  lowerCAmelCase    				= tf.cast(tf.math.not_equal(_UpperCAmelCase  ,     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)			)
 if cross_attn_head_mask is None:
  lowerCAmelCase    				= tf.ones((config.decoder_layers, config.decoder_attention_heads)			)
 return {
     "input_ids": input_ids,
     "decoder_input_ids": decoder_input_ids,
     "attention_mask": attention_mask,
     "decoder_attention_mask": decoder_attention_mask,
     "head_mask": head_mask,
     "decoder_head_mask": decoder_head_mask,
     "cross_attn_head_mask": cross_attn_head_mask,
 }







@require_tf
class       a							(   a__   , a__   , unittest.TestCase  ):
 snake_case__             =       (
     (TFBlenderbotSmallForConditionalGeneration, TFBlenderbotSmallModel) if is_tf_available() else ()
 )
 snake_case__             =       (TFBlenderbotSmallForConditionalGeneration,) if is_tf_available() else ()
 snake_case__             =       (
     {
         '''conversational''': TFBlenderbotSmallForConditionalGeneration,
         '''feature-extraction''': TFBlenderbotSmallModel,
         '''summarization''': TFBlenderbotSmallForConditionalGeneration,
         '''text2text-generation''': TFBlenderbotSmallForConditionalGeneration,
         '''translation''': TFBlenderbotSmallForConditionalGeneration,
     }
     if is_tf_available()
     else {}
 )
 snake_case__             =       True
 snake_case__             =       False
 snake_case__             =       False



 def    UpperCamelCase__  (     self     ):







  """simple docstring"""






  lowerCAmelCase    				= TFBlenderbotSmallModelTester(self     )
  lowerCAmelCase    				= ConfigTester(self				,       config_class=_snake_case     )



 def    UpperCamelCase__  (     self     ):







  """simple docstring"""






  self.config_tester.run_common_tests()



 def    UpperCamelCase__  (     self     ):







  """simple docstring"""






  lowerCAmelCase    				= self.model_tester.prepare_config_and_inputs_for_common()
  self.model_tester.check_decoder_model_past_large_inputs(*_snake_case     )







@require_tokenizers
@require_tf
class       a							(   unittest.TestCase  ):
 snake_case__             =       [
     '''Social anxiety\nWow, I am never shy. Do you have anxiety?\nYes. I end up sweating and blushing and feel like '''
     '''  i\'m going to throw up.\nand why is that?'''
 ]
 snake_case__             =       '''facebook/blenderbot_small-90M'''



 @cached_property
 def    UpperCamelCase__  (     self     ):







  """simple docstring"""






  return BlenderbotSmallTokenizer.from_pretrained('facebook/blenderbot-90M'     )



 @cached_property
 def    UpperCamelCase__  (     self     ):







  """simple docstring"""






  lowerCAmelCase    				= TFAutoModelForSeqaSeqLM.from_pretrained(self.model_name     )
  return model



 @slow
 def    UpperCamelCase__  (     self     ):







  """simple docstring"""






  lowerCAmelCase    				= self.tokenizer(self.src_text				,       return_tensors='tf'     )
  lowerCAmelCase    				= self.model.generate(
      model_inputs.input_ids				,       attention_mask=model_inputs.attention_mask				,       num_beams=2				,       use_cache=_snake_case				,       )
  lowerCAmelCase    				= self.tokenizer.batch_decode(generated_ids.numpy()				,       skip_special_tokens=_snake_case     )[0]
  assert generated_words in (
      "i don't know. i just feel like i'm going to throw up. it's not fun.",
      "i'm not sure. i just feel like i've been feeling like i have to be in a certain place",
      "i'm not sure. i just feel like i've been in a bad situation.",
  )










'''simple docstring'''

import shutil
import tempfile
import unittest
from unittest.mock import patch

from transformers import (
    DefaultFlowCallback,
    IntervalStrategy,
    PrinterCallback,
    ProgressCallback,
    Trainer,
    TrainerCallback,
    TrainingArguments,
    is_torch_available,
)
from transformers.testing_utils import require_torch


if is_torch_available():
			from transformers.trainer import DEFAULT_CALLBACKS

			from .test_trainer import RegressionDataset, RegressionModelConfig, RegressionPreTrainedModel

class 							a__( snake_case__     ):






			def __init__(     self						)							->							Dict:
					snake_case__	     =[]






			def 						_lowercase  (     self					,   _UpperCAmelCase					,   _UpperCAmelCase					,   _UpperCAmelCase					,   **_UpperCAmelCase						)							->							List[str]:
					self.events.append('on_init_end'						)






			def 						_lowercase  (     self					,   _UpperCAmelCase					,   _UpperCAmelCase					,   _UpperCAmelCase					,   **_UpperCAmelCase						)							->							Dict:
					self.events.append('on_train_begin'						)






			def 						_lowercase  (     self					,   _UpperCAmelCase					,   _UpperCAmelCase					,   _UpperCAmelCase					,   **_UpperCAmelCase						)							->							Optional[Any]:
					self.events.append('on_train_end'						)






			def 						_lowercase  (     self					,   _UpperCAmelCase					,   _UpperCAmelCase					,   _UpperCAmelCase					,   **_UpperCAmelCase						)							->							int:
					self.events.append('on_epoch_begin'						)






			def 						_lowercase  (     self					,   _UpperCAmelCase					,   _UpperCAmelCase					,   _UpperCAmelCase					,   **_UpperCAmelCase						)							->							List[str]:
					self.events.append('on_epoch_end'						)






			def 						_lowercase  (     self					,   _UpperCAmelCase					,   _UpperCAmelCase					,   _UpperCAmelCase					,   **_UpperCAmelCase						)							->							Optional[int]:
					self.events.append('on_step_begin'						)






			def 						_lowercase  (     self					,   _UpperCAmelCase					,   _UpperCAmelCase					,   _UpperCAmelCase					,   **_UpperCAmelCase						)							->							str:
					self.events.append('on_step_end'						)






			def 						_lowercase  (     self					,   _UpperCAmelCase					,   _UpperCAmelCase					,   _UpperCAmelCase					,   **_UpperCAmelCase						)							->							List[Any]:
					self.events.append('on_evaluate'						)






			def 						_lowercase  (     self					,   _UpperCAmelCase					,   _UpperCAmelCase					,   _UpperCAmelCase					,   **_UpperCAmelCase						)							->							Dict:
					self.events.append('on_predict'						)






			def 						_lowercase  (     self					,   _UpperCAmelCase					,   _UpperCAmelCase					,   _UpperCAmelCase					,   **_UpperCAmelCase						)							->							Any:
					self.events.append('on_save'						)






			def 						_lowercase  (     self					,   _UpperCAmelCase					,   _UpperCAmelCase					,   _UpperCAmelCase					,   **_UpperCAmelCase						)							->							Tuple:
					self.events.append('on_log'						)






			def 						_lowercase  (     self					,   _UpperCAmelCase					,   _UpperCAmelCase					,   _UpperCAmelCase					,   **_UpperCAmelCase						)							->							List[Any]:
					self.events.append('on_prediction_step'						)





@require_torch
class 							a__( unittest.TestCase     ):






			def 						_lowercase  (     self						)							->							Optional[Any]:
					snake_case__	     =tempfile.mkdtemp()






			def 						_lowercase  (     self						)							->							List[str]:
					shutil.rmtree(self.output_dir						)






			def 						_lowercase  (     self					,   _UpperCAmelCase=0					,   _UpperCAmelCase=0					,   _UpperCAmelCase=64					,   _UpperCAmelCase=64					,   _UpperCAmelCase=None					,   _UpperCAmelCase=False					,   **_UpperCAmelCase						)							->							int:
					# disable_tqdm in TrainingArguments has a flaky default since it depends on the level of logging. We make sure
					# its set to False since the tests later on depend on its value.
					snake_case__	     =RegressionDataset(length=_UpperCAmelCase						)
					snake_case__	     =RegressionDataset(length=_UpperCAmelCase						)
					snake_case__	     =RegressionModelConfig(a=_UpperCAmelCase					,   b=_UpperCAmelCase						)
					snake_case__	     =RegressionPreTrainedModel(_UpperCAmelCase						)

					snake_case__	     =TrainingArguments(self.output_dir					,   disable_tqdm=_UpperCAmelCase					,   report_to=[]					,   **_UpperCAmelCase						)
					return Trainer(
					    _UpperCAmelCase					,   _UpperCAmelCase					,   train_dataset=_UpperCAmelCase					,   eval_dataset=_UpperCAmelCase					,   callbacks=_UpperCAmelCase					,   )






			def 						_lowercase  (     self					,   _UpperCAmelCase					,   _UpperCAmelCase						)							->							Optional[Any]:
					self.assertEqual(len(_UpperCAmelCase						)					,   len(_UpperCAmelCase						)						)

					# Order doesn't matter
					snake_case__	     =sorted(_UpperCAmelCase					,   key=lambda _UpperCAmelCase						: cb.__name__ if isinstance(_UpperCAmelCase					,   _UpperCAmelCase						) else cb.__class__.__name__						)
					snake_case__	     =sorted(_UpperCAmelCase					,   key=lambda _UpperCAmelCase						: cb.__name__ if isinstance(_UpperCAmelCase					,   _UpperCAmelCase						) else cb.__class__.__name__						)

					for cba, cba in zip(_UpperCAmelCase					,   _UpperCAmelCase						):
							if isinstance(_UpperCAmelCase					,   _UpperCAmelCase						) and isinstance(_UpperCAmelCase					,   _UpperCAmelCase						):
									self.assertEqual(_UpperCAmelCase					,   _UpperCAmelCase						)
							elif isinstance(_UpperCAmelCase					,   _UpperCAmelCase						) and not isinstance(_UpperCAmelCase					,   _UpperCAmelCase						):
									self.assertEqual(_UpperCAmelCase					,   cba.__class__						)
							elif not isinstance(_UpperCAmelCase					,   _UpperCAmelCase						) and isinstance(_UpperCAmelCase					,   _UpperCAmelCase						):
									self.assertEqual(cba.__class__					,   _UpperCAmelCase						)
							else:
									self.assertEqual(_UpperCAmelCase					,   _UpperCAmelCase						)






			def 						_lowercase  (     self					,   _UpperCAmelCase						)							->							Any:
					snake_case__	     =['on_init_end', 'on_train_begin']
					snake_case__	     =0
					snake_case__	     =len(trainer.get_eval_dataloader()						)
					snake_case__	     =['on_prediction_step'] * len(trainer.get_eval_dataloader()						) + ['on_log', 'on_evaluate']
					for _ in range(trainer.state.num_train_epochs						):
							expected_events.append('on_epoch_begin'						)
							for _ in range(_UpperCAmelCase						):
									step += 1
									expected_events += ["on_step_begin", "on_step_end"]
									if step % trainer.args.logging_steps == 0:
											expected_events.append('on_log'						)
									if trainer.args.evaluation_strategy == IntervalStrategy.STEPS and step % trainer.args.eval_steps == 0:
											expected_events += evaluation_events.copy()
									if step % trainer.args.save_steps == 0:
											expected_events.append('on_save'						)
							expected_events.append('on_epoch_end'						)
							if trainer.args.evaluation_strategy == IntervalStrategy.EPOCH:
									expected_events += evaluation_events.copy()
					expected_events += ["on_log", "on_train_end"]
					return expected_events






			def 						_lowercase  (     self						)							->							Optional[int]:
					snake_case__	     =self.get_trainer()
					snake_case__	     =DEFAULT_CALLBACKS.copy() + [ProgressCallback]
					self.check_callbacks_equality(trainer.callback_handler.callbacks					,   _UpperCAmelCase						)

					# Callbacks passed at init are added to the default callbacks
					snake_case__	     =self.get_trainer(callbacks=[MyTestTrainerCallback]						)
					expected_callbacks.append(_UpperCAmelCase						)
					self.check_callbacks_equality(trainer.callback_handler.callbacks					,   _UpperCAmelCase						)

					# TrainingArguments.disable_tqdm controls if use ProgressCallback or PrinterCallback
					snake_case__	     =self.get_trainer(disable_tqdm=_UpperCAmelCase						)
					snake_case__	     =DEFAULT_CALLBACKS.copy() + [PrinterCallback]
					self.check_callbacks_equality(trainer.callback_handler.callbacks					,   _UpperCAmelCase						)






			def 						_lowercase  (     self						)							->							Optional[Any]:
					snake_case__	     =DEFAULT_CALLBACKS.copy() + [ProgressCallback]
					snake_case__	     =self.get_trainer()

					# We can add, pop, or remove by class name
					trainer.remove_callback(_UpperCAmelCase						)
					expected_callbacks.remove(_UpperCAmelCase						)
					self.check_callbacks_equality(trainer.callback_handler.callbacks					,   _UpperCAmelCase						)

					snake_case__	     =self.get_trainer()
					snake_case__	     =trainer.pop_callback(_UpperCAmelCase						)
					self.assertEqual(cb.__class__					,   _UpperCAmelCase						)
					self.check_callbacks_equality(trainer.callback_handler.callbacks					,   _UpperCAmelCase						)

					trainer.add_callback(_UpperCAmelCase						)
					expected_callbacks.insert(0					,   _UpperCAmelCase						)
					self.check_callbacks_equality(trainer.callback_handler.callbacks					,   _UpperCAmelCase						)

					# We can also add, pop, or remove by instance
					snake_case__	     =self.get_trainer()
					snake_case__	     =trainer.callback_handler.callbacks[0]
					trainer.remove_callback(_UpperCAmelCase						)
					expected_callbacks.remove(_UpperCAmelCase						)
					self.check_callbacks_equality(trainer.callback_handler.callbacks					,   _UpperCAmelCase						)

					snake_case__	     =self.get_trainer()
					snake_case__	     =trainer.callback_handler.callbacks[0]
					snake_case__	     =trainer.pop_callback(_UpperCAmelCase						)
					self.assertEqual(_UpperCAmelCase					,   _UpperCAmelCase						)
					self.check_callbacks_equality(trainer.callback_handler.callbacks					,   _UpperCAmelCase						)

					trainer.add_callback(_UpperCAmelCase						)
					expected_callbacks.insert(0					,   _UpperCAmelCase						)
					self.check_callbacks_equality(trainer.callback_handler.callbacks					,   _UpperCAmelCase						)






			def 						_lowercase  (     self						)							->							Dict:
					import warnings

					# XXX: for now ignore scatter_gather warnings in this test since it's not relevant to what's being tested
					warnings.simplefilter(action='ignore'					,   category=_UpperCAmelCase						)

					snake_case__	     =self.get_trainer(callbacks=[MyTestTrainerCallback]						)
					trainer.train()
					snake_case__	     =trainer.callback_handler.callbacks[-2].events
					self.assertEqual(_UpperCAmelCase					,   self.get_expected_events(_UpperCAmelCase						)						)

					# Independent log/save/eval
					snake_case__	     =self.get_trainer(callbacks=[MyTestTrainerCallback]					,   logging_steps=5						)
					trainer.train()
					snake_case__	     =trainer.callback_handler.callbacks[-2].events
					self.assertEqual(_UpperCAmelCase					,   self.get_expected_events(_UpperCAmelCase						)						)

					snake_case__	     =self.get_trainer(callbacks=[MyTestTrainerCallback]					,   save_steps=5						)
					trainer.train()
					snake_case__	     =trainer.callback_handler.callbacks[-2].events
					self.assertEqual(_UpperCAmelCase					,   self.get_expected_events(_UpperCAmelCase						)						)

					snake_case__	     =self.get_trainer(callbacks=[MyTestTrainerCallback]					,   eval_steps=5					,   evaluation_strategy='steps'						)
					trainer.train()
					snake_case__	     =trainer.callback_handler.callbacks[-2].events
					self.assertEqual(_UpperCAmelCase					,   self.get_expected_events(_UpperCAmelCase						)						)

					snake_case__	     =self.get_trainer(callbacks=[MyTestTrainerCallback]					,   evaluation_strategy='epoch'						)
					trainer.train()
					snake_case__	     =trainer.callback_handler.callbacks[-2].events
					self.assertEqual(_UpperCAmelCase					,   self.get_expected_events(_UpperCAmelCase						)						)

					# A bit of everything
					snake_case__	     =self.get_trainer(
					    callbacks=[MyTestTrainerCallback]					,   logging_steps=3					,   save_steps=10					,   eval_steps=5					,   evaluation_strategy='steps'					,   )
					trainer.train()
					snake_case__	     =trainer.callback_handler.callbacks[-2].events
					self.assertEqual(_UpperCAmelCase					,   self.get_expected_events(_UpperCAmelCase						)						)

					# warning should be emitted for duplicated callbacks
					with patch('transformers.trainer_callback.logger.warning'						) as warn_mock:
							snake_case__	     =self.get_trainer(
							    callbacks=[MyTestTrainerCallback, MyTestTrainerCallback]					,   )
							assert str(_UpperCAmelCase						) in warn_mock.call_args[0][0]




