

import unittest

from transformers import (
    MODEL_FOR_CAUSAL_LM_MAPPING,
    TF_MODEL_FOR_CAUSAL_LM_MAPPING,
    TextGenerationPipeline,
    logging,
    pipeline,
)
from transformers.testing_utils import (
    CaptureLogger,
    is_pipeline_test,
    require_accelerate,
    require_tf,
    require_torch,
    require_torch_gpu,
    require_torch_or_tf,
)

from .test_pipelines_common import ANY






@is_pipeline_test
@require_torch_or_tf
class 	A_		(	unittest.TestCase    ):
							_UpperCAmelCase   :  Tuple										=				MODEL_FOR_CAUSAL_LM_MAPPING
							_UpperCAmelCase   :  List[str]										=				TF_MODEL_FOR_CAUSAL_LM_MAPPING
							@require_torch
							def    lowerCAmelCase					(		self		:      str):
								__lowerCamelCase    :    Optional[int] 	=			pipeline(task='text-generation'       ,model='sshleifer/tiny-ctrl'       ,framework='pt')
								# Using `do_sample=False` to force deterministic output
								__lowerCamelCase    :    Optional[Any] 	=			text_generator('This is a test'       ,do_sample=SCREAMING_SNAKE_CASE__)
								self.assertEqual(
								    SCREAMING_SNAKE_CASE__       ,[
								        {
								            'generated_text': (
								                'This is a test ☃ ☃ segmental segmental segmental 议议eski eski flutter flutter Lacy oscope.'
								                ' oscope. FiliFili@@'
								            )
								        }
								    ]       ,)

								__lowerCamelCase    :    int 	=			text_generator(['This is a test', 'This is a second test'])
								self.assertEqual(
								    SCREAMING_SNAKE_CASE__       ,[
								        [
								            {
								                'generated_text': (
								                    'This is a test ☃ ☃ segmental segmental segmental 议议eski eski flutter flutter Lacy oscope.'
								                    ' oscope. FiliFili@@'
								                )
								            }
								        ],
								        [
								            {
								                'generated_text': (
								                    'This is a second test ☃ segmental segmental segmental 议议eski eski flutter flutter Lacy'
								                    ' oscope. oscope. FiliFili@@'
								                )
								            }
								        ],
								    ]       ,)

								__lowerCamelCase    :    Union[str, Any] 	=			text_generator('This is a test'       ,do_sample=SCREAMING_SNAKE_CASE__       ,num_return_sequences=2       ,return_tensors=SCREAMING_SNAKE_CASE__)
								self.assertEqual(
								    SCREAMING_SNAKE_CASE__       ,[
								        {'generated_token_ids': ANY(SCREAMING_SNAKE_CASE__)},
								        {'generated_token_ids': ANY(SCREAMING_SNAKE_CASE__)},
								    ]       ,)
								__lowerCamelCase    :    List[Any] 	=			text_generator.model.config.eos_token_id
								__lowerCamelCase    :    List[str] 	=			'<pad>'
								__lowerCamelCase    :    Optional[Any] 	=			text_generator(
								    ['This is a test', 'This is a second test']       ,do_sample=SCREAMING_SNAKE_CASE__       ,num_return_sequences=2       ,batch_size=2       ,return_tensors=SCREAMING_SNAKE_CASE__       ,)
								self.assertEqual(
								    SCREAMING_SNAKE_CASE__       ,[
								        [
								            {'generated_token_ids': ANY(SCREAMING_SNAKE_CASE__)},
								            {'generated_token_ids': ANY(SCREAMING_SNAKE_CASE__)},
								        ],
								        [
								            {'generated_token_ids': ANY(SCREAMING_SNAKE_CASE__)},
								            {'generated_token_ids': ANY(SCREAMING_SNAKE_CASE__)},
								        ],
								    ]       ,)
							@require_tf
							def    lowerCAmelCase					(		self		:      Optional[int]):
								__lowerCamelCase    :    Union[str, Any] 	=			pipeline(task='text-generation'       ,model='sshleifer/tiny-ctrl'       ,framework='tf')

								# Using `do_sample=False` to force deterministic output
								__lowerCamelCase    :    Union[str, Any] 	=			text_generator('This is a test'       ,do_sample=SCREAMING_SNAKE_CASE__)
								self.assertEqual(
								    SCREAMING_SNAKE_CASE__       ,[
								        {
								            'generated_text': (
								                'This is a test FeyFeyFey(Croatis.), s.), Cannes Cannes Cannes 閲閲Cannes Cannes Cannes 攵'
								                ' please,'
								            )
								        }
								    ]       ,)

								__lowerCamelCase    :    Optional[int] 	=			text_generator(['This is a test', 'This is a second test']       ,do_sample=SCREAMING_SNAKE_CASE__)
								self.assertEqual(
								    SCREAMING_SNAKE_CASE__       ,[
								        [
								            {
								                'generated_text': (
								                    'This is a test FeyFeyFey(Croatis.), s.), Cannes Cannes Cannes 閲閲Cannes Cannes Cannes 攵'
								                    ' please,'
								                )
								            }
								        ],
								        [
								            {
								                'generated_text': (
								                    'This is a second test Chieftain Chieftain prefecture prefecture prefecture Cannes Cannes'
								                    ' Cannes 閲閲Cannes Cannes Cannes 攵 please,'
								                )
								            }
								        ],
								    ]       ,)
							def    lowerCAmelCase					(		self		:      Dict       ,SCREAMING_SNAKE_CASE__		:      List[str]       ,SCREAMING_SNAKE_CASE__		:      Dict       ,SCREAMING_SNAKE_CASE__		:      int):
								__lowerCamelCase    :    List[str] 	=			TextGenerationPipeline(model=SCREAMING_SNAKE_CASE__       ,tokenizer=SCREAMING_SNAKE_CASE__)
								return text_generator, ["This is a test", "Another test"]
							def    lowerCAmelCase					(		self		:      Union[str, Any]):
								__lowerCamelCase    :    Optional[int] 	=			'Hello I believe in'
								__lowerCamelCase    :    Optional[int] 	=			pipeline('text-generation'       ,model='hf-internal-testing/tiny-random-gpt2')
								__lowerCamelCase    :    Union[str, Any] 	=			text_generator(SCREAMING_SNAKE_CASE__)
								self.assertEqual(
								    SCREAMING_SNAKE_CASE__       ,[{'generated_text': 'Hello I believe in fe fe fe fe fe fe fe fe fe fe fe fe'}]       ,)

								__lowerCamelCase    :    Dict 	=			text_generator(SCREAMING_SNAKE_CASE__       ,stop_sequence=' fe')
								self.assertEqual(SCREAMING_SNAKE_CASE__       ,[{'generated_text': 'Hello I believe in fe'}])
							def    lowerCAmelCase					(		self		:      Optional[Any]       ,SCREAMING_SNAKE_CASE__		:      str       ,SCREAMING_SNAKE_CASE__		:      Any):
								__lowerCamelCase    :    List[Any] 	=			text_generator.model
								__lowerCamelCase    :    Optional[Any] 	=			text_generator.tokenizer

								__lowerCamelCase    :    Dict 	=			text_generator('This is a test')
								self.assertEqual(SCREAMING_SNAKE_CASE__       ,[{'generated_text': ANY(SCREAMING_SNAKE_CASE__)}])
								self.assertTrue(outputs[0]['generated_text'].startswith('This is a test'))

								__lowerCamelCase    :    Union[str, Any] 	=			text_generator('This is a test'       ,return_full_text=SCREAMING_SNAKE_CASE__)
								self.assertEqual(SCREAMING_SNAKE_CASE__       ,[{'generated_text': ANY(SCREAMING_SNAKE_CASE__)}])
								self.assertNotIn('This is a test'       ,outputs[0]['generated_text'])

								__lowerCamelCase    :    List[Any] 	=			pipeline(task='text-generation'       ,model=SCREAMING_SNAKE_CASE__       ,tokenizer=SCREAMING_SNAKE_CASE__       ,return_full_text=SCREAMING_SNAKE_CASE__)
								__lowerCamelCase    :    List[str] 	=			text_generator('This is a test')
								self.assertEqual(SCREAMING_SNAKE_CASE__       ,[{'generated_text': ANY(SCREAMING_SNAKE_CASE__)}])
								self.assertNotIn('This is a test'       ,outputs[0]['generated_text'])

								__lowerCamelCase    :    int 	=			text_generator('This is a test'       ,return_full_text=SCREAMING_SNAKE_CASE__)
								self.assertEqual(SCREAMING_SNAKE_CASE__       ,[{'generated_text': ANY(SCREAMING_SNAKE_CASE__)}])
								self.assertTrue(outputs[0]['generated_text'].startswith('This is a test'))

								__lowerCamelCase    :    List[str] 	=			text_generator(['This is great !', 'Something else']       ,num_return_sequences=2       ,do_sample=SCREAMING_SNAKE_CASE__)
								self.assertEqual(
								    SCREAMING_SNAKE_CASE__       ,[
								        [{'generated_text': ANY(SCREAMING_SNAKE_CASE__)}, {'generated_text': ANY(SCREAMING_SNAKE_CASE__)}],
								        [{'generated_text': ANY(SCREAMING_SNAKE_CASE__)}, {'generated_text': ANY(SCREAMING_SNAKE_CASE__)}],
								    ]       ,)

								if text_generator.tokenizer.pad_token is not None:
									__lowerCamelCase    :    Any 	=			text_generator(
									    ['This is great !', 'Something else']       ,num_return_sequences=2       ,batch_size=2       ,do_sample=SCREAMING_SNAKE_CASE__)
									self.assertEqual(
									    SCREAMING_SNAKE_CASE__       ,[
									        [{'generated_text': ANY(SCREAMING_SNAKE_CASE__)}, {'generated_text': ANY(SCREAMING_SNAKE_CASE__)}],
									        [{'generated_text': ANY(SCREAMING_SNAKE_CASE__)}, {'generated_text': ANY(SCREAMING_SNAKE_CASE__)}],
									    ]       ,)

								with self.assertRaises(SCREAMING_SNAKE_CASE__):
									__lowerCamelCase    :    Dict 	=			text_generator('test'       ,return_full_text=SCREAMING_SNAKE_CASE__       ,return_text=SCREAMING_SNAKE_CASE__)
								with self.assertRaises(SCREAMING_SNAKE_CASE__):
									__lowerCamelCase    :    Optional[Any] 	=			text_generator('test'       ,return_full_text=SCREAMING_SNAKE_CASE__       ,return_tensors=SCREAMING_SNAKE_CASE__)
								with self.assertRaises(SCREAMING_SNAKE_CASE__):
									__lowerCamelCase    :    Optional[Any] 	=			text_generator('test'       ,return_text=SCREAMING_SNAKE_CASE__       ,return_tensors=SCREAMING_SNAKE_CASE__)

								# Empty prompt is slighly special
								# it requires BOS token to exist.
								# Special case for Pegasus which will always append EOS so will
								# work even without BOS.
								if (
								    text_generator.tokenizer.bos_token_id is not None
								    or "Pegasus" in tokenizer.__class__.__name__
								    or "Git" in model.__class__.__name__
								):
									__lowerCamelCase    :    Optional[Any] 	=			text_generator('')
									self.assertEqual(SCREAMING_SNAKE_CASE__       ,[{'generated_text': ANY(SCREAMING_SNAKE_CASE__)}])
								else:
									with self.assertRaises((ValueError, AssertionError)):
										__lowerCamelCase    :    List[str] 	=			text_generator('')

								if text_generator.framework == "tf":
									# TF generation does not support max_new_tokens, and it's impossible
									# to control long generation with only max_length without
									# fancy calculation, dismissing tests for now.
									return
								# We don't care about infinite range models.
								# They already work.
								# Skip this test for XGLM, since it uses sinusoidal positional embeddings which are resized on-the-fly.
								__lowerCamelCase    :    Union[str, Any] 	=			['RwkvForCausalLM', 'XGLMForCausalLM', 'GPTNeoXForCausalLM']
								if (
								    tokenizer.model_max_length < 1_0_0_0_0
								    and text_generator.model.__class__.__name__ not in EXTRA_MODELS_CAN_HANDLE_LONG_INPUTS
								):
									# Handling of large generations
									with self.assertRaises((RuntimeError, IndexError, ValueError, AssertionError)):
										text_generator('This is a test' * 5_0_0       ,max_new_tokens=2_0)

									__lowerCamelCase    :    Any 	=			text_generator('This is a test' * 5_0_0       ,handle_long_generation='hole'       ,max_new_tokens=2_0)
									# Hole strategy cannot work
									with self.assertRaises(SCREAMING_SNAKE_CASE__):
										text_generator(
										    'This is a test' * 5_0_0       ,handle_long_generation='hole'       ,max_new_tokens=tokenizer.model_max_length + 1_0       ,)
							@require_torch
							@require_accelerate
							@require_torch_gpu
							def    lowerCAmelCase					(		self		:      Optional[Any]):
								import torch

								# Classic `model_kwargs`
								__lowerCamelCase    :    int 	=			pipeline(
								    model='hf-internal-testing/tiny-random-bloom'       ,model_kwargs={'device_map': 'auto', 'torch_dtype': torch.bfloataa}       ,)
								self.assertEqual(pipe.model.device       ,torch.device(0))
								self.assertEqual(pipe.model.lm_head.weight.dtype       ,torch.bfloataa)
								__lowerCamelCase    :    Optional[int] 	=			pipe('This is a test')
								self.assertEqual(
								    SCREAMING_SNAKE_CASE__       ,[
								        {
								            'generated_text': (
								                'This is a test test test test test test test test test test test test test test test test'
								                ' test'
								            )
								        }
								    ]       ,)

								# Upgraded those two to real pipeline arguments (they just get sent for the model as they're unlikely to mean anything else.)
								__lowerCamelCase    :    Tuple 	=			pipeline(model='hf-internal-testing/tiny-random-bloom'       ,device_map='auto'       ,torch_dtype=torch.bfloataa)
								self.assertEqual(pipe.model.device       ,torch.device(0))
								self.assertEqual(pipe.model.lm_head.weight.dtype       ,torch.bfloataa)
								__lowerCamelCase    :    List[Any] 	=			pipe('This is a test')
								self.assertEqual(
								    SCREAMING_SNAKE_CASE__       ,[
								        {
								            'generated_text': (
								                'This is a test test test test test test test test test test test test test test test test'
								                ' test'
								            )
								        }
								    ]       ,)

								# torch_dtype will be automatically set to float32 if not provided - check: https://github.com/huggingface/transformers/pull/20602
								__lowerCamelCase    :    List[str] 	=			pipeline(model='hf-internal-testing/tiny-random-bloom'       ,device_map='auto')
								self.assertEqual(pipe.model.device       ,torch.device(0))
								self.assertEqual(pipe.model.lm_head.weight.dtype       ,torch.floataa)
								__lowerCamelCase    :    List[Any] 	=			pipe('This is a test')
								self.assertEqual(
								    SCREAMING_SNAKE_CASE__       ,[
								        {
								            'generated_text': (
								                'This is a test test test test test test test test test test test test test test test test'
								                ' test'
								            )
								        }
								    ]       ,)
							@require_torch
							@require_torch_gpu
							def    lowerCAmelCase					(		self		:      Optional[int]):
								import torch

								__lowerCamelCase    :    Any 	=			pipeline(model='hf-internal-testing/tiny-random-bloom'       ,device=0       ,torch_dtype=torch.floataa)
								pipe('This is a test')
							@require_torch
							@require_accelerate
							@require_torch_gpu
							def    lowerCAmelCase					(		self		:      str):
								import torch

								__lowerCamelCase    :    Union[str, Any] 	=			pipeline(model='hf-internal-testing/tiny-random-bloom'       ,device_map='auto'       ,torch_dtype=torch.floataa)
								pipe('This is a test'       ,do_sample=SCREAMING_SNAKE_CASE__       ,top_p=0.5)


							def    lowerCAmelCase					(		self		:      Any):
								__lowerCamelCase    :    Optional[Any] 	=			'Hello world'
								__lowerCamelCase    :    Union[str, Any] 	=			pipeline('text-generation'       ,model='hf-internal-testing/tiny-random-gpt2')
								if text_generator.model.framework == "tf":
									__lowerCamelCase    :    str 	=			logging.get_logger('transformers.generation.tf_utils')
								else:
									__lowerCamelCase    :    Dict 	=			logging.get_logger('transformers.generation.utils')
								__lowerCamelCase    :    Optional[int] 	=			'Both `max_new_tokens`'  # The beggining of the message to be checked in this test

								# Both are set by the user -> log warning
								with CaptureLogger(SCREAMING_SNAKE_CASE__) as cl:
									__lowerCamelCase    :    Dict 	=			text_generator(SCREAMING_SNAKE_CASE__       ,max_length=1_0       ,max_new_tokens=1)
								self.assertIn(SCREAMING_SNAKE_CASE__       ,cl.out)

								# The user only sets one -> no warning
								with CaptureLogger(SCREAMING_SNAKE_CASE__) as cl:
									__lowerCamelCase    :    str 	=			text_generator(SCREAMING_SNAKE_CASE__       ,max_new_tokens=1)
								self.assertNotIn(SCREAMING_SNAKE_CASE__       ,cl.out)

								with CaptureLogger(SCREAMING_SNAKE_CASE__) as cl:
									__lowerCamelCase    :    Any 	=			text_generator(SCREAMING_SNAKE_CASE__       ,max_length=1_0)
								self.assertNotIn(SCREAMING_SNAKE_CASE__       ,cl.out)







# This is the module that test_patching.py uses to test patch_submodule()

import os  # noqa: this is just for tests
import os as renamed_os  # noqa: this is just for tests
from os import path  # noqa: this is just for tests
from os import path as renamed_path  # noqa: this is just for tests
from os.path import join  # noqa: this is just for tests
from os.path import join as renamed_join  # noqa: this is just for tests


a		=open  # noqa: we just need to have a builtin inside this module to test it properly




