




import unittest

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

from ...generation.test_utils import GenerationTesterMixin
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 (
					    OPENAI_GPT_PRETRAINED_MODEL_ARCHIVE_LIST,
					    OpenAIGPTConfig,
					    OpenAIGPTDoubleHeadsModel,
					    OpenAIGPTForSequenceClassification,
					    OpenAIGPTLMHeadModel,
					    OpenAIGPTModel,
					)







class     __lowerCAmelCase :

			"""simple docstring"""







			def __init__(      self			,						lowerCamelCase__			,						lowerCamelCase__=13			,						lowerCamelCase__=7			,						lowerCamelCase__=True			,						lowerCamelCase__=True			,						lowerCamelCase__=True			,						lowerCamelCase__=99			,						lowerCamelCase__=32			,						lowerCamelCase__=5			,						lowerCamelCase__=4			,						lowerCamelCase__=37			,						lowerCamelCase__="gelu"			,						lowerCamelCase__=0.1			,						lowerCamelCase__=0.1			,						lowerCamelCase__=512			,						lowerCamelCase__=16			,						lowerCamelCase__=2			,						lowerCamelCase__=0.02			,						lowerCamelCase__=3			,						lowerCamelCase__=4			,						lowerCamelCase__=None			,						)    ->      List[str]:



							'''simple docstring'''
							__lowerCamelCase		       =    parent
							__lowerCamelCase		       =    batch_size
							__lowerCamelCase		       =    seq_length
							__lowerCamelCase		       =    is_training
							__lowerCamelCase		       =    use_token_type_ids
							__lowerCamelCase		       =    use_labels
							__lowerCamelCase		       =    vocab_size
							__lowerCamelCase		       =    hidden_size
							__lowerCamelCase		       =    num_hidden_layers
							__lowerCamelCase		       =    num_attention_heads
							__lowerCamelCase		       =    intermediate_size
							__lowerCamelCase		       =    hidden_act
							__lowerCamelCase		       =    hidden_dropout_prob
							__lowerCamelCase		       =    attention_probs_dropout_prob
							__lowerCamelCase		       =    max_position_embeddings
							__lowerCamelCase		       =    type_vocab_size
							__lowerCamelCase		       =    type_sequence_label_size
							__lowerCamelCase		       =    initializer_range
							__lowerCamelCase		       =    num_labels
							__lowerCamelCase		       =    num_choices
							__lowerCamelCase		       =    scope
							__lowerCamelCase		       =    self.vocab_size - 1


			def       lowercase_    (      self							)    ->      str:



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

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

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

							__lowerCamelCase		       =    OpenAIGPTConfig(
							    vocab_size=self.vocab_size			,						n_embd=self.hidden_size			,						n_layer=self.num_hidden_layers			,						n_head=self.num_attention_heads			,						n_positions=self.max_position_embeddings			,						pad_token_id=self.pad_token_id			,						)

							__lowerCamelCase		       =    ids_tensor([self.num_hidden_layers, self.num_attention_heads]			,						2							)

							return (
							    config,
							    input_ids,
							    head_mask,
							    token_type_ids,
							    sequence_labels,
							    token_labels,
							    choice_labels,
							)


			def       lowercase_    (      self			,						lowerCamelCase__			,						lowerCamelCase__			,						lowerCamelCase__			,						lowerCamelCase__			,						*lowerCamelCase__							)    ->      Union[str, Any]:



							'''simple docstring'''
							__lowerCamelCase		       =    OpenAIGPTModel(config=lowerCamelCase__							)
							model.to(lowerCamelCase__							)
							model.eval()

							__lowerCamelCase		       =    model(lowerCamelCase__			,						token_type_ids=lowerCamelCase__			,						head_mask=lowerCamelCase__							)
							__lowerCamelCase		       =    model(lowerCamelCase__			,						token_type_ids=lowerCamelCase__							)
							__lowerCamelCase		       =    model(lowerCamelCase__							)

							self.parent.assertEqual(result.last_hidden_state.shape			,						(self.batch_size, self.seq_length, self.hidden_size)							)


			def       lowercase_    (      self			,						lowerCamelCase__			,						lowerCamelCase__			,						lowerCamelCase__			,						lowerCamelCase__			,						*lowerCamelCase__							)    ->      List[str]:



							'''simple docstring'''
							__lowerCamelCase		       =    OpenAIGPTLMHeadModel(lowerCamelCase__							)
							model.to(lowerCamelCase__							)
							model.eval()

							__lowerCamelCase		       =    model(lowerCamelCase__			,						token_type_ids=lowerCamelCase__			,						labels=lowerCamelCase__							)
							self.parent.assertEqual(result.loss.shape			,						()							)
							self.parent.assertEqual(result.logits.shape			,						(self.batch_size, self.seq_length, self.vocab_size)							)


			def       lowercase_    (      self			,						lowerCamelCase__			,						lowerCamelCase__			,						lowerCamelCase__			,						lowerCamelCase__			,						*lowerCamelCase__							)    ->      Tuple:



							'''simple docstring'''
							__lowerCamelCase		       =    OpenAIGPTDoubleHeadsModel(lowerCamelCase__							)
							model.to(lowerCamelCase__							)
							model.eval()

							__lowerCamelCase		       =    model(lowerCamelCase__			,						token_type_ids=lowerCamelCase__			,						labels=lowerCamelCase__							)
							self.parent.assertEqual(result.loss.shape			,						()							)
							self.parent.assertEqual(result.logits.shape			,						(self.batch_size, self.seq_length, self.vocab_size)							)


			def       lowercase_    (      self			,						lowerCamelCase__			,						lowerCamelCase__			,						lowerCamelCase__			,						lowerCamelCase__			,						*lowerCamelCase__							)    ->      int:



							'''simple docstring'''
							__lowerCamelCase		       =    self.num_labels
							__lowerCamelCase		       =    OpenAIGPTForSequenceClassification(lowerCamelCase__							)
							model.to(lowerCamelCase__							)
							model.eval()

							__lowerCamelCase		       =    ids_tensor([self.batch_size]			,						self.type_sequence_label_size							)
							__lowerCamelCase		       =    model(lowerCamelCase__			,						token_type_ids=lowerCamelCase__			,						labels=lowerCamelCase__							)
							self.parent.assertEqual(result.logits.shape			,						(self.batch_size, self.num_labels)							)






			def       lowercase_    (      self							)    ->      Optional[int]:



							'''simple docstring'''
							__lowerCamelCase		       =    self.prepare_config_and_inputs()
							(
							    (
							    __lowerCamelCase
							)       ,						(
							    __lowerCamelCase
							)       ,						(
							    __lowerCamelCase
							)       ,						(
							    __lowerCamelCase
							)       ,						(
							    __lowerCamelCase
							)       ,						(
							    __lowerCamelCase
							)       ,						(
							    __lowerCamelCase
							)       ,						
							)		       =    config_and_inputs
							__lowerCamelCase		       =    {
							    'input_ids': input_ids,
							    'token_type_ids': token_type_ids,
							    'head_mask': head_mask,
							}

							return config, inputs_dict







@require_torch
class     __lowerCAmelCase (       __magic_name__			,				__magic_name__			,				__magic_name__			,				unittest.TestCase ):

			"""simple docstring"""







			snake_case_   		=       (
			    (OpenAIGPTModel, OpenAIGPTLMHeadModel, OpenAIGPTDoubleHeadsModel, OpenAIGPTForSequenceClassification)
			    if is_torch_available()
			    else ()
			)
			snake_case_   		=       (
			    (OpenAIGPTLMHeadModel,) if is_torch_available() else ()
			)  # TODO (PVP): Add Double HeadsModel when generate() function is changed accordingly
			snake_case_   		=       (
			    {
			        '''feature-extraction''': OpenAIGPTModel,
			        '''text-classification''': OpenAIGPTForSequenceClassification,
			        '''text-generation''': OpenAIGPTLMHeadModel,
			        '''zero-shot''': OpenAIGPTForSequenceClassification,
			    }
			    if is_torch_available()
			    else {}
			)


			def       lowercase_    (      self			,						lowerCamelCase__			,						lowerCamelCase__			,						lowerCamelCase__			,						lowerCamelCase__			,						lowerCamelCase__							)    ->      Optional[int]:



							'''simple docstring'''
							if pipeline_test_casse_name == "ZeroShotClassificationPipelineTests":
											# Get `tokenizer does not have a padding token` error for both fast/slow tokenizers.
											# `OpenAIGPTConfig` was never used in pipeline tests, either because of a missing checkpoint or because a
											# tiny config could not be created.
											return True

							return False


			def       lowercase_    (      self			,						lowerCamelCase__			,						lowerCamelCase__			,						lowerCamelCase__=False							)    ->      Union[str, Any]:



							'''simple docstring'''
							__lowerCamelCase		       =    super()._prepare_for_class(lowerCamelCase__			,						lowerCamelCase__			,						return_labels=lowerCamelCase__							)

							if return_labels:
											if model_class.__name__ == "OpenAIGPTDoubleHeadsModel":
															__lowerCamelCase		       =    torch.zeros(
															    (self.model_tester.batch_size, self.model_tester.num_choices, self.model_tester.seq_length)			,						dtype=torch.long			,						device=lowerCamelCase__			,						)
															__lowerCamelCase		       =    inputs_dict['labels']
															__lowerCamelCase		       =    inputs_dict['labels']
															__lowerCamelCase		       =    torch.zeros(
															    (self.model_tester.batch_size, self.model_tester.num_choices)			,						dtype=torch.long			,						device=lowerCamelCase__			,						)
															__lowerCamelCase		       =    torch.zeros(
															    self.model_tester.batch_size			,						dtype=torch.long			,						device=lowerCamelCase__							)
							return inputs_dict


			def       lowercase_    (      self							)    ->      Tuple:



							'''simple docstring'''
							__lowerCamelCase		       =    OpenAIGPTModelTester(self							)
							__lowerCamelCase		       =    ConfigTester(self			,						config_class=lowerCamelCase__			,						n_embd=37							)


			def       lowercase_    (      self							)    ->      Any:



							'''simple docstring'''
							self.config_tester.run_common_tests()


			def       lowercase_    (      self							)    ->      Dict:



							'''simple docstring'''
							__lowerCamelCase		       =    self.model_tester.prepare_config_and_inputs()
							self.model_tester.create_and_check_openai_gpt_model(*lowerCamelCase__							)


			def       lowercase_    (      self							)    ->      List[str]:



							'''simple docstring'''
							__lowerCamelCase		       =    self.model_tester.prepare_config_and_inputs()
							self.model_tester.create_and_check_lm_head_model(*lowerCamelCase__							)


			def       lowercase_    (      self							)    ->      Union[str, Any]:



							'''simple docstring'''
							__lowerCamelCase		       =    self.model_tester.prepare_config_and_inputs()
							self.model_tester.create_and_check_double_lm_head_model(*lowerCamelCase__							)


			def       lowercase_    (      self							)    ->      str:



							'''simple docstring'''
							__lowerCamelCase		       =    self.model_tester.prepare_config_and_inputs()
							self.model_tester.create_and_check_openai_gpt_for_sequence_classification(*lowerCamelCase__							)






			@slow
			def       lowercase_    (      self							)    ->      List[Any]:



							'''simple docstring'''
							for model_name in OPENAI_GPT_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
											__lowerCamelCase		       =    OpenAIGPTModel.from_pretrained(lowerCamelCase__							)
											self.assertIsNotNone(lowerCamelCase__							)







@require_torch
class     __lowerCAmelCase (       unittest.TestCase ):

			"""simple docstring"""







			@slow
			def       lowercase_    (      self							)    ->      List[Any]:



							'''simple docstring'''
							__lowerCamelCase		       =    OpenAIGPTLMHeadModel.from_pretrained('openai-gpt'							)
							model.to(lowerCamelCase__							)
							__lowerCamelCase		       =    torch.tensor([[481, 4_735, 544]]			,						dtype=torch.long			,						device=lowerCamelCase__							)  # the president is
							__lowerCamelCase		       =    [
							    481,
							    4_735,
							    544,
							    246,
							    963,
							    870,
							    762,
							    239,
							    244,
							    40_477,
							    244,
							    249,
							    719,
							    881,
							    487,
							    544,
							    240,
							    244,
							    603,
							    481,
							]  # the president is a very good man. " \n " i\'m sure he is, " said the

							__lowerCamelCase		       =    model.generate(lowerCamelCase__			,						do_sample=lowerCamelCase__							)
							self.assertListEqual(output_ids[0].tolist()			,						lowerCamelCase__							)





import argparse
import json
import os
import time
import zipfile

from get_ci_error_statistics import download_artifact, get_artifacts_links

from transformers import logging


snake_case_			  = logging.get_logger(__name__)

def 	snake_case__			(    SCREAMING_SNAKE_CASE_		:    Tuple	,     SCREAMING_SNAKE_CASE_		:    Union[str, Any]			):



     '''simple docstring'''





     lowercase__    :   List[str]      =       set()
     lowercase__    :   List[str]      =       []

     def parse_line(SCREAMING_SNAKE_CASE_		:    Union[str, Any]			):
          for line in fp:
               if isinstance(SCREAMING_SNAKE_CASE_	,     SCREAMING_SNAKE_CASE_			):
                    lowercase__    :   Optional[int]      =       line.decode('UTF-8'			)
               if "warnings summary (final)" in line:
                    continue
               # This means we are outside the body of a warning
               elif not line.startswith(' '			):
                    # process a single warning and move it to `selected_warnings`.
                    if len(SCREAMING_SNAKE_CASE_			) > 0:
                         lowercase__    :   Optional[Any]      =       '\n'.join(SCREAMING_SNAKE_CASE_			)
                         # Only keep the warnings specified in `targets`
                         if any(f""": {x}: """ in warning for x in targets			):
                              selected_warnings.add(SCREAMING_SNAKE_CASE_			)
                         buffer.clear()
                    continue
               else:
                    lowercase__    :   Optional[Any]      =       line.strip()
                    buffer.append(SCREAMING_SNAKE_CASE_			)

     if from_gh:
          for filename in os.listdir(SCREAMING_SNAKE_CASE_			):
               lowercase__    :   int      =       os.path.join(SCREAMING_SNAKE_CASE_	,     SCREAMING_SNAKE_CASE_			)
               if not os.path.isdir(SCREAMING_SNAKE_CASE_			):
                    # read the file
                    if filename != "warnings.txt":
                         continue
                    with open(SCREAMING_SNAKE_CASE_			) as fp:
                         parse_line(SCREAMING_SNAKE_CASE_			)
     else:
          try:
               with zipfile.ZipFile(SCREAMING_SNAKE_CASE_			) as z:
                    for filename in z.namelist():
                         if not os.path.isdir(SCREAMING_SNAKE_CASE_			):
                              # read the file
                              if filename != "warnings.txt":
                                   continue
                              with z.open(SCREAMING_SNAKE_CASE_			) as fp:
                                   parse_line(SCREAMING_SNAKE_CASE_			)
          except Exception:
               logger.warning(
                   f"""{artifact_path} is either an invalid zip file or something else wrong. This file is skipped."""			)

     return selected_warnings

def 	snake_case__			(    SCREAMING_SNAKE_CASE_		:    List[Any]	,     SCREAMING_SNAKE_CASE_		:    Optional[Any]			):



     '''simple docstring'''





     lowercase__    :   Optional[Any]      =       set()

     lowercase__    :   List[str]      =       [os.path.join(SCREAMING_SNAKE_CASE_	,     SCREAMING_SNAKE_CASE_			) for p in os.listdir(SCREAMING_SNAKE_CASE_			) if (p.endswith('.zip'			) or from_gh)]
     for p in paths:
          selected_warnings.update(extract_warnings_from_single_artifact(SCREAMING_SNAKE_CASE_	,     SCREAMING_SNAKE_CASE_			)			)

     return selected_warnings


if __name__ == "__main__":

       def 	snake_case__			(    SCREAMING_SNAKE_CASE_		:    Any			):



            '''simple docstring'''





            return values.split(','			)

       snake_case_			  = argparse.ArgumentParser()
       # Required parameters
       parser.add_argument('''--workflow_run_id''', type=str, required=True, help='''A GitHub Actions workflow run id.''')
       parser.add_argument(
           '''--output_dir''',
           type=str,
           required=True,
           help='''Where to store the downloaded artifacts and other result files.''',
       )
       parser.add_argument('''--token''', default=None, type=str, help='''A token that has actions:read permission.''')
       # optional parameters
       parser.add_argument(
           '''--targets''',
           default='''DeprecationWarning,UserWarning,FutureWarning''',
           type=list_str,
           help='''Comma-separated list of target warning(s) which we want to extract.''',
       )
       parser.add_argument(
           '''--from_gh''',
           action='''store_true''',
           help='''If running from a GitHub action workflow and collecting warnings from its artifacts.''',
       )

       snake_case_			  = parser.parse_args()

       snake_case_			  = args.from_gh
       if from_gh:
              # The artifacts have to be downloaded using `actions/download-artifact@v3`
              pass
       else:
              os.makedirs(args.output_dir, exist_ok=True)

              # get download links
              snake_case_			  = get_artifacts_links(args.workflow_run_id, token=args.token)
              with open(os.path.join(args.output_dir, '''artifacts.json'''), '''w''', encoding='''UTF-8''') as fp:
                     json.dump(artifacts, fp, ensure_ascii=False, indent=4)

              # download artifacts
              for idx, (name, url) in enumerate(artifacts.items()):
                     print(name)
                     print(url)
                     print('''=''' * 80)
                     download_artifact(name, url, args.output_dir, args.token)
                     # Be gentle to GitHub
                     time.sleep(1)

    # extract warnings from artifacts
       snake_case_			  = extract_warnings(args.output_dir, args.targets)
       snake_case_			  = sorted(selected_warnings)
       with open(os.path.join(args.output_dir, '''selected_warnings.json'''), '''w''', encoding='''UTF-8''') as fp:
              json.dump(selected_warnings, fp, ensure_ascii=False, indent=4)

