




import tempfile
import unittest

import numpy as np
from huggingface_hub import HfFolder, delete_repo
from requests.exceptions import HTTPError

from transformers import BertConfig, is_flax_available
from transformers.testing_utils import TOKEN, USER, is_staging_test, require_flax


if is_flax_available():
  import os

  from flax.core.frozen_dict import unfreeze
  from flax.traverse_util import flatten_dict

  from transformers import FlaxBertModel

  _snake_case										=       '0.12'  # assumed parallelism: 8







@require_flax
@is_staging_test
class 						lowercase							(   unittest.TestCase		):

     @classmethod
     def 					a__	(  cls				)						->							int:
           _A							:     Optional[int]			=						TOKEN
           HfFolder.save_token(lowercase_				)

     @classmethod
     def 					a__	(  cls				)						->							List[Any]:
           try:
                 delete_repo(token=cls._token		,     repo_id="""test-model-flax"""				)
           except HTTPError:
                 pass

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

     def 					a__	(  self				)						->							List[Any]:
           _A							:     Tuple			=						BertConfig(
               vocab_size=99		,     hidden_size=32		,     num_hidden_layers=5		,     num_attention_heads=4		,     intermediate_size=37				)
           _A							:     Dict			=						FlaxBertModel(lowercase_				)
           model.push_to_hub("""test-model-flax"""		,     use_auth_token=self._token				)

           _A							:     List[Any]			=						FlaxBertModel.from_pretrained(F'''{USER}/test-model-flax'''				)

           _A							:     List[Any]			=						flatten_dict(unfreeze(model.params				)				)
           _A							:     Optional[int]			=						flatten_dict(unfreeze(new_model.params				)				)

           for key in base_params.keys():
                 _A							:     List[Any]			=						(base_params[key] - new_params[key]).sum().item()
                 self.assertLessEqual(lowercase_		,     1e-3		,     msg=F'''{key} not identical'''				)

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

           # Push to hub via save_pretrained
           with tempfile.TemporaryDirectory() as tmp_dir:
                 model.save_pretrained(lowercase_		,     repo_id="""test-model-flax"""		,     push_to_hub=lowercase_		,     use_auth_token=self._token				)

           _A							:     Optional[int]			=						FlaxBertModel.from_pretrained(F'''{USER}/test-model-flax'''				)

           _A							:     Optional[int]			=						flatten_dict(unfreeze(model.params				)				)
           _A							:     Any			=						flatten_dict(unfreeze(new_model.params				)				)

           for key in base_params.keys():
                 _A							:     Tuple			=						(base_params[key] - new_params[key]).sum().item()
                 self.assertLessEqual(lowercase_		,     1e-3		,     msg=F'''{key} not identical'''				)



     def 					a__	(  self				)						->							Optional[int]:
           _A							:     Any			=						BertConfig(
               vocab_size=99		,     hidden_size=32		,     num_hidden_layers=5		,     num_attention_heads=4		,     intermediate_size=37				)
           _A							:     Union[str, Any]			=						FlaxBertModel(lowercase_				)
           model.push_to_hub("""valid_org/test-model-flax-org"""		,     use_auth_token=self._token				)

           _A							:     List[str]			=						FlaxBertModel.from_pretrained("""valid_org/test-model-flax-org"""				)

           _A							:     int			=						flatten_dict(unfreeze(model.params				)				)
           _A							:     Optional[int]			=						flatten_dict(unfreeze(new_model.params				)				)

           for key in base_params.keys():
                 _A							:     Union[str, Any]			=						(base_params[key] - new_params[key]).sum().item()
                 self.assertLessEqual(lowercase_		,     1e-3		,     msg=F'''{key} not identical'''				)

           # Reset repo
           delete_repo(token=self._token		,     repo_id="""valid_org/test-model-flax-org"""				)

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

           _A							:     List[Any]			=						FlaxBertModel.from_pretrained("""valid_org/test-model-flax-org"""				)

           _A							:     Optional[Any]			=						flatten_dict(unfreeze(model.params				)				)
           _A							:     str			=						flatten_dict(unfreeze(new_model.params				)				)

           for key in base_params.keys():
                 _A							:     List[str]			=						(base_params[key] - new_params[key]).sum().item()
                 self.assertLessEqual(lowercase_		,     1e-3		,     msg=F'''{key} not identical'''				)







def 				lowerCAmelCase_	(     snake_case_,snake_case_     ):
      _A							:     List[str]			=						True
      _A							:     Optional[Any]			=						flatten_dict(modela.params     )
      _A							:     Union[str, Any]			=						flatten_dict(modela.params     )
      for key in flat_params_a.keys():
            if np.sum(np.abs(flat_params_a[key] - flat_params_a[key]     )     ) > 1e-4:
                  _A							:     Tuple			=						False

      return models_are_equal







@require_flax
class 						lowercase							(   unittest.TestCase		):

     def 					a__	(  self				)						->							Optional[int]:
           _A							:     int			=						BertConfig.from_pretrained("""hf-internal-testing/tiny-bert-flax-only"""				)
           _A							:     Dict			=						FlaxBertModel(lowercase_				)

           _A							:     Any			=						"""bert"""
           with tempfile.TemporaryDirectory() as tmp_dir:
                 model.save_pretrained(os.path.join(lowercase_		,     lowercase_				)				)

                 with self.assertRaises(lowercase_				):
                       _A							:     List[Any]			=						FlaxBertModel.from_pretrained(lowercase_				)

                 _A							:     Any			=						FlaxBertModel.from_pretrained(lowercase_		,     subfolder=lowercase_				)

           self.assertTrue(check_models_equal(lowercase_		,     lowercase_				)				)

     def 					a__	(  self				)						->							Optional[Any]:
           _A							:     Dict			=						BertConfig.from_pretrained("""hf-internal-testing/tiny-bert-flax-only"""				)
           _A							:     int			=						FlaxBertModel(lowercase_				)

           _A							:     int			=						"""bert"""
           with tempfile.TemporaryDirectory() as tmp_dir:
                 model.save_pretrained(os.path.join(lowercase_		,     lowercase_				)		,     max_shard_size="""10KB"""				)

                 with self.assertRaises(lowercase_				):
                       _A							:     Any			=						FlaxBertModel.from_pretrained(lowercase_				)

                 _A							:     Optional[Any]			=						FlaxBertModel.from_pretrained(lowercase_		,     subfolder=lowercase_				)

           self.assertTrue(check_models_equal(lowercase_		,     lowercase_				)				)

     def 					a__	(  self				)						->							List[Any]:
           _A							:     Any			=						"""bert"""
           _A							:     List[str]			=						"""hf-internal-testing/tiny-random-bert-subfolder"""

           with self.assertRaises(lowercase_				):
                 _A							:     Dict			=						FlaxBertModel.from_pretrained(lowercase_				)

           _A							:     int			=						FlaxBertModel.from_pretrained(lowercase_		,     subfolder=lowercase_				)

           self.assertIsNotNone(lowercase_				)



     def 					a__	(  self				)						->							Tuple:
           _A							:     Dict			=						"""bert"""
           _A							:     Optional[Any]			=						"""hf-internal-testing/tiny-random-bert-sharded-subfolder"""
           with self.assertRaises(lowercase_				):
                 _A							:     Optional[int]			=						FlaxBertModel.from_pretrained(lowercase_				)

           _A							:     List[Any]			=						FlaxBertModel.from_pretrained(lowercase_		,     subfolder=lowercase_				)

           self.assertIsNotNone(lowercase_				)







'''simple docstring'''

import unittest

from transformers import MPNetConfig, is_torch_available
from transformers.testing_utils import require_torch, slow, torch_device

from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, ids_tensor, random_attention_mask
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
							import torch

							from transformers import (
							    MPNetForMaskedLM,
							    MPNetForMultipleChoice,
							    MPNetForQuestionAnswering,
							    MPNetForSequenceClassification,
							    MPNetForTokenClassification,
							    MPNetModel,
							)






class 						a				:


				def __init__(		self    :					List[Any]       ,				lowercase_    :					Optional[Any]       ,				lowercase_    :					Any=13       ,				lowercase_    :					Optional[Any]=7       ,				lowercase_    :					Optional[Any]=True       ,				lowercase_    :					Dict=True       ,				lowercase_    :					Tuple=False       ,				lowercase_    :					Optional[Any]=True       ,				lowercase_    :					Any=99       ,				lowercase_    :					Union[str, Any]=64       ,				lowercase_    :					str=5       ,				lowercase_    :					int=4       ,				lowercase_    :					List[Any]=64       ,				lowercase_    :					Dict="gelu"       ,				lowercase_    :					Optional[int]=0.1       ,				lowercase_    :					Optional[Any]=0.1       ,				lowercase_    :					Tuple=512       ,				lowercase_    :					List[Any]=16       ,				lowercase_    :					str=2       ,				lowercase_    :					List[str]=0.02       ,				lowercase_    :					Optional[Any]=3       ,				lowercase_    :					Optional[Any]=4       ,				lowercase_    :					List[Any]=None       ,				):
									snake_case_									=				parent
									snake_case_									=				batch_size
									snake_case_									=				seq_length
									snake_case_									=				is_training
									snake_case_									=				use_input_mask
									snake_case_									=				use_token_type_ids
									snake_case_									=				use_labels
									snake_case_									=				vocab_size
									snake_case_									=				hidden_size
									snake_case_									=				num_hidden_layers
									snake_case_									=				num_attention_heads
									snake_case_									=				intermediate_size
									snake_case_									=				hidden_act
									snake_case_									=				hidden_dropout_prob
									snake_case_									=				attention_probs_dropout_prob
									snake_case_									=				max_position_embeddings
									snake_case_									=				type_vocab_size
									snake_case_									=				type_sequence_label_size
									snake_case_									=				initializer_range
									snake_case_									=				num_labels
									snake_case_									=				num_choices
									snake_case_									=				scope


				def   A_ (		self    :					List[str]					):
									return MPNetConfig.from_pretrained('''microsoft/mpnet-base'''					)


				def   A_ (		self    :					str					):
									snake_case_									=				ids_tensor([self.batch_size, self.seq_length]       ,				self.vocab_size					)

									snake_case_									=				None
									if self.use_input_mask:
														snake_case_									=				random_attention_mask([self.batch_size, self.seq_length]					)

									snake_case_									=				None
									snake_case_									=				None
									snake_case_									=				None
									if self.use_labels:
														snake_case_									=				ids_tensor([self.batch_size]       ,				self.type_sequence_label_size					)
														snake_case_									=				ids_tensor([self.batch_size, self.seq_length]       ,				self.num_labels					)
														snake_case_									=				ids_tensor([self.batch_size]       ,				self.num_choices					)

									snake_case_									=				self.get_config()
									return config, input_ids, input_mask, sequence_labels, token_labels, choice_labels


				def   A_ (		self    :					Tuple					):
									return MPNetConfig(
									    vocab_size=self.vocab_size       ,				hidden_size=self.hidden_size       ,				num_hidden_layers=self.num_hidden_layers       ,				num_attention_heads=self.num_attention_heads       ,				intermediate_size=self.intermediate_size       ,				hidden_act=self.hidden_act       ,				hidden_dropout_prob=self.hidden_dropout_prob       ,				attention_probs_dropout_prob=self.attention_probs_dropout_prob       ,				max_position_embeddings=self.max_position_embeddings       ,				initializer_range=self.initializer_range       ,				)


				def   A_ (		self    :					Any       ,				lowercase_    :					Dict       ,				lowercase_    :					str       ,				lowercase_    :					Optional[Any]       ,				lowercase_    :					Union[str, Any]       ,				lowercase_    :					Dict       ,				lowercase_    :					Optional[int]					):
									snake_case_									=				MPNetModel(config=lowercase_					)
									model.to(lowercase_					)
									model.eval()
									snake_case_									=				model(lowercase_       ,				lowercase_					)
									snake_case_									=				model(lowercase_					)
									self.parent.assertEqual(result.last_hidden_state.shape       ,				(self.batch_size, self.seq_length, self.hidden_size)					)
									self.parent.assertEqual(result.pooler_output.shape       ,				(self.batch_size, self.hidden_size)					)


				def   A_ (		self    :					str       ,				lowercase_    :					Optional[Any]       ,				lowercase_    :					Any       ,				lowercase_    :					str       ,				lowercase_    :					Tuple       ,				lowercase_    :					Tuple       ,				lowercase_    :					Optional[int]					):
									snake_case_									=				MPNetForQuestionAnswering(config=lowercase_					)
									model.to(lowercase_					)
									model.eval()
									snake_case_									=				model(
									    lowercase_       ,				attention_mask=lowercase_       ,				start_positions=lowercase_       ,				end_positions=lowercase_       ,				)
									self.parent.assertEqual(result.start_logits.shape       ,				(self.batch_size, self.seq_length)					)
									self.parent.assertEqual(result.end_logits.shape       ,				(self.batch_size, self.seq_length)					)


				def   A_ (		self    :					Tuple       ,				lowercase_    :					Tuple       ,				lowercase_    :					List[str]       ,				lowercase_    :					str       ,				lowercase_    :					Optional[int]       ,				lowercase_    :					Optional[Any]       ,				lowercase_    :					Any					):
									snake_case_									=				self.num_labels
									snake_case_									=				MPNetForSequenceClassification(lowercase_					)
									model.to(lowercase_					)
									model.eval()
									snake_case_									=				model(lowercase_       ,				attention_mask=lowercase_       ,				labels=lowercase_					)
									self.parent.assertEqual(result.logits.shape       ,				(self.batch_size, self.num_labels)					)


				def   A_ (		self    :					Any       ,				lowercase_    :					Any       ,				lowercase_    :					str       ,				lowercase_    :					Tuple       ,				lowercase_    :					List[str]       ,				lowercase_    :					Union[str, Any]       ,				lowercase_    :					Dict					):
									snake_case_									=				self.num_choices
									snake_case_									=				MPNetForMultipleChoice(config=lowercase_					)
									model.to(lowercase_					)
									model.eval()
									snake_case_									=				input_ids.unsqueeze(1					).expand(-1       ,				self.num_choices       ,				-1					).contiguous()
									snake_case_									=				input_mask.unsqueeze(1					).expand(-1       ,				self.num_choices       ,				-1					).contiguous()
									snake_case_									=				model(
									    lowercase_       ,				attention_mask=lowercase_       ,				labels=lowercase_       ,				)
									self.parent.assertEqual(result.logits.shape       ,				(self.batch_size, self.num_choices)					)


				def   A_ (		self    :					Union[str, Any]       ,				lowercase_    :					List[Any]       ,				lowercase_    :					Optional[Any]       ,				lowercase_    :					Optional[int]       ,				lowercase_    :					str       ,				lowercase_    :					int       ,				lowercase_    :					List[str]					):
									snake_case_									=				self.num_labels
									snake_case_									=				MPNetForTokenClassification(config=lowercase_					)
									model.to(lowercase_					)
									model.eval()
									snake_case_									=				model(lowercase_       ,				attention_mask=lowercase_       ,				labels=lowercase_					)
									self.parent.assertEqual(result.logits.shape       ,				(self.batch_size, self.seq_length, self.num_labels)					)


				def   A_ (		self    :					Union[str, Any]					):
									snake_case_									=				self.prepare_config_and_inputs()
									((snake_case_)			,(snake_case_)			,(snake_case_)			,(snake_case_)			,(snake_case_)			,(snake_case_))									=				config_and_inputs
									snake_case_									=				{'''input_ids''': input_ids, '''attention_mask''': input_mask}
									return config, inputs_dict







@require_torch
class 						a				(   _lowerCamelCase   ,     _lowerCamelCase   ,     unittest.TestCase     ):
				snake_case_      =		(
				    (
				        MPNetForMaskedLM,
				        MPNetForMultipleChoice,
				        MPNetForQuestionAnswering,
				        MPNetForSequenceClassification,
				        MPNetForTokenClassification,
				        MPNetModel,
				    )
				    if is_torch_available()
				    else ()
				)
				snake_case_      =		(
				    {
				        "feature-extraction": MPNetModel,
				        "fill-mask": MPNetForMaskedLM,
				        "question-answering": MPNetForQuestionAnswering,
				        "text-classification": MPNetForSequenceClassification,
				        "token-classification": MPNetForTokenClassification,
				        "zero-shot": MPNetForSequenceClassification,
				    }
				    if is_torch_available()
				    else {}
				)
				snake_case_      =		False
				snake_case_      =		True


				def   A_ (		self    :					Tuple					):
									snake_case_									=				MPNetModelTester(self					)
									snake_case_									=				ConfigTester(self       ,				config_class=lowercase_       ,				hidden_size=37					)


				def   A_ (		self    :					Union[str, Any]					):
									self.config_tester.run_common_tests()


				def   A_ (		self    :					Tuple					):
									snake_case_									=				self.model_tester.prepare_config_and_inputs()
									self.model_tester.create_and_check_mpnet_model(*lowercase_					)


				def   A_ (		self    :					List[Any]					):
									snake_case_									=				self.model_tester.prepare_config_and_inputs()
									self.model_tester.create_and_check_mpnet_for_sequence_classification(*lowercase_					)


				def   A_ (		self    :					List[Any]					):
									snake_case_									=				self.model_tester.prepare_config_and_inputs()
									self.model_tester.create_and_check_mpnet_for_multiple_choice(*lowercase_					)


				def   A_ (		self    :					Union[str, Any]					):
									snake_case_									=				self.model_tester.prepare_config_and_inputs()
									self.model_tester.create_and_check_mpnet_for_token_classification(*lowercase_					)


				def   A_ (		self    :					Tuple					):
									snake_case_									=				self.model_tester.prepare_config_and_inputs()
									self.model_tester.create_and_check_mpnet_for_question_answering(*lowercase_					)







@require_torch
class 						a				(   unittest.TestCase     ):


				@slow
				def   A_ (		self    :					List[Any]					):
									snake_case_									=				MPNetModel.from_pretrained('''microsoft/mpnet-base'''					)
									snake_case_									=				torch.tensor([[0, 345, 232, 328, 740, 140, 1695, 69, 6078, 1588, 2]]					)
									snake_case_									=				model(lowercase_					)[0]
									snake_case_									=				torch.Size((1, 11, 768)					)
									self.assertEqual(output.shape       ,				lowercase_					)
									snake_case_									=				torch.tensor(
									    [[[-0.0550, 0.1943, -0.0740], [-0.0562, 0.2211, -0.0579], [-0.0437, 0.3337, -0.0641]]]					)
									# compare the actual values for a slice.
									self.assertTrue(torch.allclose(output[:, :3, :3]       ,				lowercase_       ,				atol=1e-4					)					)

