import unittest

from transformers import DebertaVaConfig, is_torch_available
from transformers.testing_utils import require_sentencepiece, require_tokenizers, require_torch, slow, torch_device

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


if is_torch_available():
					import torch

					from transformers import (
					    DebertaVaForMaskedLM,
					    DebertaVaForMultipleChoice,
					    DebertaVaForQuestionAnswering,
					    DebertaVaForSequenceClassification,
					    DebertaVaForTokenClassification,
					    DebertaVaModel,
					)
					from transformers.models.deberta_va.modeling_deberta_va import DEBERTA_V2_PRETRAINED_MODEL_ARCHIVE_LIST





class       lowercase__		(     _snake_case):


						def __init__( self			:       int					,	UpperCamelCase__			:       List[str]					,	UpperCamelCase__			:       Optional[int]=13					,	UpperCamelCase__			:       Optional[Any]=7					,	UpperCamelCase__			:       List[Any]=True					,	UpperCamelCase__			:       int=True					,	UpperCamelCase__			:       Dict=True					,	UpperCamelCase__			:       Optional[int]=True					,	UpperCamelCase__			:       Optional[int]=99					,	UpperCamelCase__			:       Tuple=32					,	UpperCamelCase__			:       List[str]=5					,	UpperCamelCase__			:       int=4					,	UpperCamelCase__			:       Optional[Any]=37					,	UpperCamelCase__			:       List[Any]="gelu"					,	UpperCamelCase__			:       Any=0.1					,	UpperCamelCase__			:       int=0.1					,	UpperCamelCase__			:       Optional[Any]=512					,	UpperCamelCase__			:       Optional[int]=16					,	UpperCamelCase__			:       str=2					,	UpperCamelCase__			:       Union[str, Any]=0.02					,	UpperCamelCase__			:       int=False					,	UpperCamelCase__			:       Dict=True					,	UpperCamelCase__			:       Any="None"					,	UpperCamelCase__			:       Optional[int]=3					,	UpperCamelCase__			:       List[str]=4					,	UpperCamelCase__			:       Union[str, Any]=None					,	):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:					List[str]           =							parent
												SCREAMING_SNAKE_CASE							:					List[Any]           =							batch_size
												SCREAMING_SNAKE_CASE							:					Tuple           =							seq_length
												SCREAMING_SNAKE_CASE							:					List[Any]           =							is_training
												SCREAMING_SNAKE_CASE							:					Dict           =							use_input_mask
												SCREAMING_SNAKE_CASE							:					int           =							use_token_type_ids
												SCREAMING_SNAKE_CASE							:					Any           =							use_labels
												SCREAMING_SNAKE_CASE							:					List[Any]           =							vocab_size
												SCREAMING_SNAKE_CASE							:					Optional[Any]           =							hidden_size
												SCREAMING_SNAKE_CASE							:					Optional[int]           =							num_hidden_layers
												SCREAMING_SNAKE_CASE							:					int           =							num_attention_heads
												SCREAMING_SNAKE_CASE							:					Optional[Any]           =							intermediate_size
												SCREAMING_SNAKE_CASE							:					List[Any]           =							hidden_act
												SCREAMING_SNAKE_CASE							:					Tuple           =							hidden_dropout_prob
												SCREAMING_SNAKE_CASE							:					List[str]           =							attention_probs_dropout_prob
												SCREAMING_SNAKE_CASE							:					Any           =							max_position_embeddings
												SCREAMING_SNAKE_CASE							:					Optional[Any]           =							type_vocab_size
												SCREAMING_SNAKE_CASE							:					Union[str, Any]           =							type_sequence_label_size
												SCREAMING_SNAKE_CASE							:					Optional[Any]           =							initializer_range
												SCREAMING_SNAKE_CASE							:					Optional[int]           =							num_labels
												SCREAMING_SNAKE_CASE							:					List[str]           =							num_choices
												SCREAMING_SNAKE_CASE							:					Optional[Any]           =							relative_attention
												SCREAMING_SNAKE_CASE							:					Optional[Any]           =							position_biased_input
												SCREAMING_SNAKE_CASE							:					Dict           =							pos_att_type
												SCREAMING_SNAKE_CASE							:					List[Any]           =							scope


						def  __A ( self			:       Dict		):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:					int           =							ids_tensor([self.batch_size, self.seq_length]					,	self.vocab_size		)

												SCREAMING_SNAKE_CASE							:					List[Any]           =							None
												if self.use_input_mask:
																		SCREAMING_SNAKE_CASE							:					str           =							ids_tensor([self.batch_size, self.seq_length]					,	vocab_size=2		)

												SCREAMING_SNAKE_CASE							:					List[str]           =							None
												if self.use_token_type_ids:
																		SCREAMING_SNAKE_CASE							:					str           =							ids_tensor([self.batch_size, self.seq_length]					,	self.type_vocab_size		)

												SCREAMING_SNAKE_CASE							:					Any           =							None
												SCREAMING_SNAKE_CASE							:					Optional[int]           =							None
												SCREAMING_SNAKE_CASE							:					Tuple           =							None
												if self.use_labels:
																		SCREAMING_SNAKE_CASE							:					Optional[int]           =							ids_tensor([self.batch_size]					,	self.type_sequence_label_size		)
																		SCREAMING_SNAKE_CASE							:					Union[str, Any]           =							ids_tensor([self.batch_size, self.seq_length]					,	self.num_labels		)
																		SCREAMING_SNAKE_CASE							:					List[str]           =							ids_tensor([self.batch_size]					,	self.num_choices		)

												SCREAMING_SNAKE_CASE							:					Dict           =							self.get_config()

												return config, input_ids, token_type_ids, input_mask, sequence_labels, token_labels, choice_labels


						def  __A ( self			:       Optional[int]		):





												'''simple docstring'''





												return DebertaVaConfig(
												    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					,	type_vocab_size=self.type_vocab_size					,	initializer_range=self.initializer_range					,	relative_attention=self.relative_attention					,	position_biased_input=self.position_biased_input					,	pos_att_type=self.pos_att_type					,	)


						def  __A ( self			:       List[Any]					,	UpperCamelCase__			:       List[str]		):





												'''simple docstring'''





												self.parent.assertListEqual(list(result.loss.size()		)					,	[]		)


						def  __A ( self			:       str					,	UpperCamelCase__			:       Dict					,	UpperCamelCase__			:       Optional[int]					,	UpperCamelCase__			:       Tuple					,	UpperCamelCase__			:       Dict					,	UpperCamelCase__			:       Tuple					,	UpperCamelCase__			:       Union[str, Any]					,	UpperCamelCase__			:       Union[str, Any]		):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:					List[Any]           =							DebertaVaModel(config=UpperCamelCase__		)
												model.to(UpperCamelCase__		)
												model.eval()
												SCREAMING_SNAKE_CASE							:					Optional[Any]           =							model(UpperCamelCase__					,	attention_mask=UpperCamelCase__					,	token_type_ids=UpperCamelCase__		)[0]
												SCREAMING_SNAKE_CASE							:					int           =							model(UpperCamelCase__					,	token_type_ids=UpperCamelCase__		)[0]
												SCREAMING_SNAKE_CASE							:					Optional[Any]           =							model(UpperCamelCase__		)[0]

												self.parent.assertListEqual(list(sequence_output.size()		)					,	[self.batch_size, self.seq_length, self.hidden_size]		)


						def  __A ( self			:       Union[str, Any]					,	UpperCamelCase__			:       Any					,	UpperCamelCase__			:       List[Any]					,	UpperCamelCase__			:       List[Any]					,	UpperCamelCase__			:       Any					,	UpperCamelCase__			:       List[str]					,	UpperCamelCase__			:       Any					,	UpperCamelCase__			:       Optional[Any]		):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:					Optional[int]           =							DebertaVaForMaskedLM(config=UpperCamelCase__		)
												model.to(UpperCamelCase__		)
												model.eval()
												SCREAMING_SNAKE_CASE							:					Any           =							model(UpperCamelCase__					,	attention_mask=UpperCamelCase__					,	token_type_ids=UpperCamelCase__					,	labels=UpperCamelCase__		)

												self.parent.assertEqual(result.logits.shape					,	(self.batch_size, self.seq_length, self.vocab_size)		)


						def  __A ( self			:       List[Any]					,	UpperCamelCase__			:       Optional[int]					,	UpperCamelCase__			:       List[Any]					,	UpperCamelCase__			:       Any					,	UpperCamelCase__			:       Optional[int]					,	UpperCamelCase__			:       int					,	UpperCamelCase__			:       Optional[Any]					,	UpperCamelCase__			:       List[Any]		):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:					Optional[int]           =							self.num_labels
												SCREAMING_SNAKE_CASE							:					Union[str, Any]           =							DebertaVaForSequenceClassification(UpperCamelCase__		)
												model.to(UpperCamelCase__		)
												model.eval()
												SCREAMING_SNAKE_CASE							:					Optional[Any]           =							model(UpperCamelCase__					,	attention_mask=UpperCamelCase__					,	token_type_ids=UpperCamelCase__					,	labels=UpperCamelCase__		)
												self.parent.assertListEqual(list(result.logits.size()		)					,	[self.batch_size, self.num_labels]		)
												self.check_loss_output(UpperCamelCase__		)


						def  __A ( self			:       Any					,	UpperCamelCase__			:       Dict					,	UpperCamelCase__			:       Optional[int]					,	UpperCamelCase__			:       List[Any]					,	UpperCamelCase__			:       Tuple					,	UpperCamelCase__			:       Dict					,	UpperCamelCase__			:       List[Any]					,	UpperCamelCase__			:       Optional[Any]		):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:					Any           =							self.num_labels
												SCREAMING_SNAKE_CASE							:					Tuple           =							DebertaVaForTokenClassification(config=UpperCamelCase__		)
												model.to(UpperCamelCase__		)
												model.eval()
												SCREAMING_SNAKE_CASE							:					Any           =							model(UpperCamelCase__					,	attention_mask=UpperCamelCase__					,	token_type_ids=UpperCamelCase__					,	labels=UpperCamelCase__		)
												self.parent.assertEqual(result.logits.shape					,	(self.batch_size, self.seq_length, self.num_labels)		)


						def  __A ( self			:       Union[str, Any]					,	UpperCamelCase__			:       int					,	UpperCamelCase__			:       Optional[int]					,	UpperCamelCase__			:       Tuple					,	UpperCamelCase__			:       Optional[int]					,	UpperCamelCase__			:       Optional[int]					,	UpperCamelCase__			:       Dict					,	UpperCamelCase__			:       Tuple		):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:					List[str]           =							DebertaVaForQuestionAnswering(config=UpperCamelCase__		)
												model.to(UpperCamelCase__		)
												model.eval()
												SCREAMING_SNAKE_CASE							:					Tuple           =							model(
												    UpperCamelCase__					,	attention_mask=UpperCamelCase__					,	token_type_ids=UpperCamelCase__					,	start_positions=UpperCamelCase__					,	end_positions=UpperCamelCase__					,	)
												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			:       Union[str, Any]					,	UpperCamelCase__			:       int					,	UpperCamelCase__			:       Dict					,	UpperCamelCase__			:       Optional[Any]					,	UpperCamelCase__			:       int					,	UpperCamelCase__			:       int					,	UpperCamelCase__			:       Dict					,	UpperCamelCase__			:       str		):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:					Optional[int]           =							DebertaVaForMultipleChoice(config=UpperCamelCase__		)
												model.to(UpperCamelCase__		)
												model.eval()
												SCREAMING_SNAKE_CASE							:					Optional[int]           =							input_ids.unsqueeze(1		).expand(-1					,	self.num_choices					,	-1		).contiguous()
												SCREAMING_SNAKE_CASE							:					List[Any]           =							token_type_ids.unsqueeze(1		).expand(-1					,	self.num_choices					,	-1		).contiguous()
												SCREAMING_SNAKE_CASE							:					Dict           =							input_mask.unsqueeze(1		).expand(-1					,	self.num_choices					,	-1		).contiguous()
												SCREAMING_SNAKE_CASE							:					List[Any]           =							model(
												    UpperCamelCase__					,	attention_mask=UpperCamelCase__					,	token_type_ids=UpperCamelCase__					,	labels=UpperCamelCase__					,	)
												self.parent.assertEqual(result.logits.shape					,	(self.batch_size, self.num_choices)		)



						def  __A ( self			:       List[str]		):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:					List[Any]           =							self.prepare_config_and_inputs()
												(
												    (
												    SCREAMING_SNAKE_CASE     
												)       ,						(
												    SCREAMING_SNAKE_CASE     
												)       ,						(
												    SCREAMING_SNAKE_CASE     
												)       ,						(
												    SCREAMING_SNAKE_CASE     
												)       ,						(
												    SCREAMING_SNAKE_CASE     
												)       ,						(
												    SCREAMING_SNAKE_CASE     
												)       ,						(
												    SCREAMING_SNAKE_CASE     
												)       ,						     
												)							:					Tuple           =							config_and_inputs
												SCREAMING_SNAKE_CASE							:					Any           =							{'''input_ids''': input_ids, '''token_type_ids''': token_type_ids, '''attention_mask''': input_mask}
												return config, inputs_dict






@require_torch
class       lowercase__		(     _snake_case			,							_snake_case			,							unittest.TestCase):
						UpperCamelCase_  =	(
						    (
						        DebertaVaModel,
						        DebertaVaForMaskedLM,
						        DebertaVaForSequenceClassification,
						        DebertaVaForTokenClassification,
						        DebertaVaForQuestionAnswering,
						        DebertaVaForMultipleChoice,
						    )
						    if is_torch_available()
						    else ()
						)
						UpperCamelCase_  =	(
						    {
						        'feature-extraction': DebertaVaModel,
						        'fill-mask': DebertaVaForMaskedLM,
						        'question-answering': DebertaVaForQuestionAnswering,
						        'text-classification': DebertaVaForSequenceClassification,
						        'token-classification': DebertaVaForTokenClassification,
						        'zero-shot': DebertaVaForSequenceClassification,
						    }
						    if is_torch_available()
						    else {}
						)

						UpperCamelCase_  =	True
						UpperCamelCase_  =	False
						UpperCamelCase_  =	False
						UpperCamelCase_  =	False
						UpperCamelCase_  =	False


						def  __A ( self			:       Optional[int]		):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:					Optional[int]           =							DebertaVaModelTester(self		)
												SCREAMING_SNAKE_CASE							:					Optional[Any]           =							ConfigTester(self					,	config_class=UpperCamelCase__					,	hidden_size=37		)


						def  __A ( self			:       Dict		):





												'''simple docstring'''





												self.config_tester.run_common_tests()


						def  __A ( self			:       List[Any]		):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:					int           =							self.model_tester.prepare_config_and_inputs()
												self.model_tester.create_and_check_deberta_model(*UpperCamelCase__		)


						def  __A ( self			:       Optional[Any]		):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:					Any           =							self.model_tester.prepare_config_and_inputs()
												self.model_tester.create_and_check_deberta_for_sequence_classification(*UpperCamelCase__		)


						def  __A ( self			:       List[Any]		):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:					Tuple           =							self.model_tester.prepare_config_and_inputs()
												self.model_tester.create_and_check_deberta_for_masked_lm(*UpperCamelCase__		)


						def  __A ( self			:       Any		):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:					Dict           =							self.model_tester.prepare_config_and_inputs()
												self.model_tester.create_and_check_deberta_for_question_answering(*UpperCamelCase__		)


						def  __A ( self			:       int		):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:					Dict           =							self.model_tester.prepare_config_and_inputs()
												self.model_tester.create_and_check_deberta_for_token_classification(*UpperCamelCase__		)


						def  __A ( self			:       List[Any]		):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:					Union[str, Any]           =							self.model_tester.prepare_config_and_inputs()
												self.model_tester.create_and_check_deberta_for_multiple_choice(*UpperCamelCase__		)



						@slow
						def  __A ( self			:       Tuple		):





												'''simple docstring'''





												for model_name in DEBERTA_V2_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
																		SCREAMING_SNAKE_CASE							:					Any           =							DebertaVaModel.from_pretrained(UpperCamelCase__		)
																		self.assertIsNotNone(UpperCamelCase__		)






@require_torch
@require_sentencepiece
@require_tokenizers
class       lowercase__		(     unittest.TestCase):


						@unittest.skip(reason='''Model not available yet'''		)
						def  __A ( self			:       Dict		):





												'''simple docstring'''





												pass



						@slow
						def  __A ( self			:       Optional[Any]		):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:					Optional[Any]           =							DebertaVaModel.from_pretrained('''microsoft/deberta-v2-xlarge'''		)

												SCREAMING_SNAKE_CASE							:					Dict           =							torch.tensor([[0, 3_1414, 232, 328, 740, 1140, 1_2695, 69, 4_6078, 1588, 2]]		)
												SCREAMING_SNAKE_CASE							:					List[Any]           =							torch.tensor([[0, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1]]		)
												with torch.no_grad():
																		SCREAMING_SNAKE_CASE							:					Union[str, Any]           =							model(UpperCamelCase__					,	attention_mask=UpperCamelCase__		)[0]
												# compare the actual values for a slice.
												SCREAMING_SNAKE_CASE							:					Optional[int]           =							torch.tensor(
												    [[[0.2356, 0.1948, 0.0369], [-0.1063, 0.3586, -0.5152], [-0.6399, -0.0259, -0.2525]]]		)
												self.assertTrue(torch.allclose(output[:, 1:4, 1:4]					,	UpperCamelCase__					,	atol=1E-4		)					,	f"""{output[:, 1:4, 1:4]}"""		)






import json
import os
import subprocess
import unittest
from ast import literal_eval

import pytest
from parameterized import parameterized, parameterized_class

from . import is_sagemaker_available


if is_sagemaker_available():
					from sagemaker import Session, TrainingJobAnalytics
					from sagemaker.huggingface import HuggingFace





@pytest.mark.skipif(
    literal_eval(os.getenv("""TEST_SAGEMAKER"""			,							"""False""")) is not True			,							reason="""Skipping test because should only be run when releasing minor transformers version"""			,							)
@pytest.mark.usefixtures("""sm_env""")
@parameterized_class(
    [
        {
            """framework""": """pytorch""",
            """script""": """run_glue.py""",
            """model_name_or_path""": """distilbert-base-cased""",
            """instance_type""": """ml.p3.16xlarge""",
            """results""": {"""train_runtime""": 650, """eval_accuracy""": 0.7, """eval_loss""": 0.6},
        },
        {
            """framework""": """pytorch""",
            """script""": """run_ddp.py""",
            """model_name_or_path""": """distilbert-base-cased""",
            """instance_type""": """ml.p3.16xlarge""",
            """results""": {"""train_runtime""": 600, """eval_accuracy""": 0.7, """eval_loss""": 0.6},
        },
        {
            """framework""": """tensorflow""",
            """script""": """run_tf_dist.py""",
            """model_name_or_path""": """distilbert-base-cased""",
            """instance_type""": """ml.p3.16xlarge""",
            """results""": {"""train_runtime""": 600, """eval_accuracy""": 0.6, """eval_loss""": 0.7},
        },
    ])
class       lowercase__		(     unittest.TestCase):


						def  __A ( self			:       Any		):





												'''simple docstring'''





												if self.framework == "pytorch":
																		subprocess.run(
																		    f"""cp ./examples/pytorch/text-classification/run_glue.py {self.env.test_path}/run_glue.py""".split()					,	encoding='''utf-8'''					,	check=UpperCamelCase__					,	)
												assert hasattr(self					,	'''env'''		)


						def  __A ( self			:       str					,	UpperCamelCase__			:       Dict		):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:					Any           =							f"""{self.env.base_job_name}-{instance_count}-{'ddp' if 'ddp' in self.script else 'smd'}"""
												# distributed data settings
												SCREAMING_SNAKE_CASE							:					Any           =							{'''smdistributed''': {'''dataparallel''': {'''enabled''': True}}} if self.script != '''run_ddp.py''' else None

												# creates estimator
												return HuggingFace(
												    entry_point=self.script					,	source_dir=self.env.test_path					,	role=self.env.role					,	image_uri=self.env.image_uri					,	base_job_name=UpperCamelCase__					,	instance_count=UpperCamelCase__					,	instance_type=self.instance_type					,	debugger_hook_config=UpperCamelCase__					,	hyperparameters={**self.env.distributed_hyperparameters, '''model_name_or_path''': self.model_name_or_path}					,	metric_definitions=self.env.metric_definitions					,	distribution=UpperCamelCase__					,	py_version='''py36'''					,	)


						def  __A ( self			:       Optional[Any]					,	UpperCamelCase__			:       List[str]		):





												'''simple docstring'''





												TrainingJobAnalytics(UpperCamelCase__		).export_csv(f"""{self.env.test_path}/{job_name}_metrics.csv"""		)



						@parameterized.expand([(2,)]		)
						def  __A ( self			:       Tuple					,	UpperCamelCase__			:       Optional[int]		):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:					List[str]           =							self.create_estimator(UpperCamelCase__		)

												# run training
												estimator.fit()

												# result dataframe
												SCREAMING_SNAKE_CASE							:					Optional[Any]           =							TrainingJobAnalytics(estimator.latest_training_job.name		).dataframe()

												# extract kpis
												SCREAMING_SNAKE_CASE							:					List[str]           =							list(result_metrics_df[result_metrics_df.metric_name == '''eval_accuracy''']['''value''']		)
												SCREAMING_SNAKE_CASE							:					Dict           =							list(result_metrics_df[result_metrics_df.metric_name == '''eval_loss''']['''value''']		)
												# get train time from SageMaker job, this includes starting, preprocessing, stopping
												SCREAMING_SNAKE_CASE							:					List[Any]           =							(
												    Session().describe_training_job(estimator.latest_training_job.name		).get('''TrainingTimeInSeconds'''					,	99_9999		)
												)

												# assert kpis
												assert train_runtime <= self.results["train_runtime"]
												assert all(t >= self.results['''eval_accuracy'''] for t in eval_accuracy		)
												assert all(t <= self.results['''eval_loss'''] for t in eval_loss		)

												# dump tests result into json file to share in PR
												with open(f"""{estimator.latest_training_job.name}.json"""					,	'''w'''		) as outfile:
																		json.dump({'''train_time''': train_runtime, '''eval_accuracy''': eval_accuracy, '''eval_loss''': eval_loss}					,	UpperCamelCase__		)





