



"""simple docstring"""







import unittest
from queue import Empty
from threading import Thread

from transformers import AutoTokenizer, TextIteratorStreamer, TextStreamer, is_torch_available
from transformers.testing_utils import CaptureStdout, require_torch, torch_device

from ..test_modeling_common import ids_tensor


if is_torch_available():
			import torch

			from transformers import AutoModelForCausalLM

@require_torch
class       _a  ( unittest.TestCase):



			def 				__lowercase					(			self							:				Tuple		)							->					Union[str, Any]:
									snake_case   :			List[str]						    = AutoTokenizer.from_pretrained("hf-internal-testing/tiny-random-gpt2"		)
									snake_case   :			List[Any]						    = AutoModelForCausalLM.from_pretrained("hf-internal-testing/tiny-random-gpt2"		).to(_lowercase		)
									snake_case   :			List[str]						    = -1

									snake_case   :			List[Any]						    = ids_tensor((1, 5)   ,	vocab_size=model.config.vocab_size		).to(_lowercase		)
									snake_case   :			str						    = model.generate(_lowercase   ,	max_new_tokens=10   ,	do_sample=_lowercase		)
									snake_case   :			List[str]						    = tokenizer.decode(greedy_ids[0]		)

									with CaptureStdout() as cs:
															snake_case   :			Any						    = TextStreamer(_lowercase		)
															model.generate(_lowercase   ,	max_new_tokens=10   ,	do_sample=_lowercase   ,	streamer=_lowercase		)
									# The greedy text should be printed to stdout, except for the final "\n" in the streamer
									snake_case   :			Union[str, Any]						    = cs.out[:-1]

									self.assertEqual(_lowercase   ,	_lowercase		)



			def 				__lowercase					(			self							:				List[str]		)							->					Union[str, Any]:
									snake_case   :			Optional[Any]						    = AutoTokenizer.from_pretrained("hf-internal-testing/tiny-random-gpt2"		)
									snake_case   :			str						    = AutoModelForCausalLM.from_pretrained("hf-internal-testing/tiny-random-gpt2"		).to(_lowercase		)
									snake_case   :			str						    = -1

									snake_case   :			List[str]						    = ids_tensor((1, 5)   ,	vocab_size=model.config.vocab_size		).to(_lowercase		)
									snake_case   :			Any						    = model.generate(_lowercase   ,	max_new_tokens=10   ,	do_sample=_lowercase		)
									snake_case   :			List[Any]						    = tokenizer.decode(greedy_ids[0]		)

									snake_case   :			Union[str, Any]						    = TextIteratorStreamer(_lowercase		)
									snake_case   :			List[Any]						    = {"input_ids": input_ids, "max_new_tokens": 10, "do_sample": False, "streamer": streamer}
									snake_case   :			List[Any]						    = Thread(target=model.generate   ,	kwargs=_lowercase		)
									thread.start()
									snake_case   :			Union[str, Any]						    = ""
									for new_text in streamer:
															streamer_text += new_text

									self.assertEqual(_lowercase   ,	_lowercase		)



			def 				__lowercase					(			self							:				Union[str, Any]		)							->					Optional[Any]:
									snake_case   :			Optional[Any]						    = AutoTokenizer.from_pretrained("hf-internal-testing/tiny-random-gpt2"		)
									snake_case   :			Dict						    = AutoModelForCausalLM.from_pretrained("hf-internal-testing/tiny-random-gpt2"		).to(_lowercase		)
									snake_case   :			int						    = -1

									snake_case   :			Union[str, Any]						    = ids_tensor((1, 5)   ,	vocab_size=model.config.vocab_size		).to(_lowercase		)
									snake_case   :			List[Any]						    = model.generate(_lowercase   ,	max_new_tokens=10   ,	do_sample=_lowercase		)
									snake_case   :			List[Any]						    = greedy_ids[:, input_ids.shape[1] :]
									snake_case   :			Dict						    = tokenizer.decode(new_greedy_ids[0]		)

									with CaptureStdout() as cs:
															snake_case   :			Optional[Any]						    = TextStreamer(_lowercase   ,	skip_prompt=_lowercase		)
															model.generate(_lowercase   ,	max_new_tokens=10   ,	do_sample=_lowercase   ,	streamer=_lowercase		)
									# The greedy text should be printed to stdout, except for the final "\n" in the streamer
									snake_case   :			Optional[int]						    = cs.out[:-1]

									self.assertEqual(_lowercase   ,	_lowercase		)



			def 				__lowercase					(			self							:				str		)							->					str:
									# Tests that we can pass `decode_kwargs` to the streamer to control how the tokens are decoded. Must be tested
									# with actual models -- the dummy models' tokenizers are not aligned with their models, and
									# `skip_special_tokens=True` has no effect on them
									snake_case   :			List[str]						    = AutoTokenizer.from_pretrained("distilgpt2"		)
									snake_case   :			Dict						    = AutoModelForCausalLM.from_pretrained("distilgpt2"		).to(_lowercase		)
									snake_case   :			Optional[Any]						    = -1

									snake_case   :			Union[str, Any]						    = torch.ones((1, 5)   ,	device=_lowercase		).long() * model.config.bos_token_id
									with CaptureStdout() as cs:
															snake_case   :			Tuple						    = TextStreamer(_lowercase   ,	skip_special_tokens=_lowercase		)
															model.generate(_lowercase   ,	max_new_tokens=1   ,	do_sample=_lowercase   ,	streamer=_lowercase		)

									# The prompt contains a special token, so the streamer should not print it. As such, the output text, when
									# re-tokenized, must only contain one token
									snake_case   :			Optional[int]						    = cs.out[:-1]  # Remove the final "\n"
									snake_case   :			str						    = tokenizer(_lowercase   ,	return_tensors="pt"		)
									self.assertEqual(streamer_text_tokenized.input_ids.shape   ,	(1, 1)		)



			def 				__lowercase					(			self							:				List[str]		)							->					Tuple:
									snake_case   :			Tuple						    = AutoTokenizer.from_pretrained("hf-internal-testing/tiny-random-gpt2"		)
									snake_case   :			Dict						    = AutoModelForCausalLM.from_pretrained("hf-internal-testing/tiny-random-gpt2"		).to(_lowercase		)
									snake_case   :			int						    = -1

									snake_case   :			int						    = ids_tensor((1, 5)   ,	vocab_size=model.config.vocab_size		).to(_lowercase		)
									snake_case   :			int						    = TextIteratorStreamer(_lowercase   ,	timeout=0.001		)
									snake_case   :			Dict						    = {"input_ids": input_ids, "max_new_tokens": 10, "do_sample": False, "streamer": streamer}
									snake_case   :			int						    = Thread(target=model.generate   ,	kwargs=_lowercase		)
									thread.start()

									# The streamer will timeout after 0.001 seconds, so an exception will be raised
									with self.assertRaises(_lowercase		):
															snake_case   :			Union[str, Any]						    = ""
															for new_text in streamer:
																					streamer_text += new_text








"""simple docstring"""







import contextlib
import importlib
import io
import unittest

import transformers

# Try to import everything from transformers to ensure every object can be loaded.
from transformers import *  # noqa F406
from transformers.testing_utils import DUMMY_UNKNOWN_IDENTIFIER, require_flax, require_tf, require_torch
from transformers.utils import ContextManagers, find_labels, is_flax_available, is_tf_available, is_torch_available


if is_torch_available():
			from transformers import BertForPreTraining, BertForQuestionAnswering, BertForSequenceClassification

if is_tf_available():
			from transformers import TFBertForPreTraining, TFBertForQuestionAnswering, TFBertForSequenceClassification

if is_flax_available():
			from transformers import FlaxBertForPreTraining, FlaxBertForQuestionAnswering, FlaxBertForSequenceClassification


A		=						DUMMY_UNKNOWN_IDENTIFIER
# An actual model hosted on huggingface.co

A		=						'main'
# Default branch name
A		=						'f2c752cfc5c0ab6f4bdec59acea69eefbee381c2'
# One particular commit (not the top of `main`)
A		=						'aaaaaaa'
# This commit does not exist, so we should 404.

A		=						'd9e9f15bc825e4b2c9249e9578f884bbcb5e3684'
# Sha-1 of config.json on the top of `main`, for checking purposes
A		=						'4b243c475af8d0a7754e87d7d096c92e5199ec2fe168a2ee7998e3b8e9bcb1d3'

@contextlib.contextmanager
def  __SCREAMING_SNAKE_CASE		(			):
						"""simple docstring"""






						print("Welcome!" )
						yield
						print("Bye!" )

@contextlib.contextmanager
def  __SCREAMING_SNAKE_CASE		(			):
						"""simple docstring"""






						print("Bonjour!" )
						yield
						print("Au revoir!" )

class       _a  ( unittest.TestCase):



			def 				__lowercase					(			self							:				Dict		)							->					Optional[int]:
									# If the spec is missing, importlib would not be able to import the module dynamically.
									assert transformers.__spec__ is not None
									assert importlib.util.find_spec("transformers"		) is not None



class       _a  ( unittest.TestCase):



			@unittest.mock.patch("sys.stdout"   ,	new_callable=io.StringIO		)
			def 				__lowercase					(			self							:				Union[str, Any]   ,	_lowercase							:				Optional[int]		)							->					Optional[Any]:
									with ContextManagers([]		):
															print("Transformers are awesome!"		)
									# The print statement adds a new line at the end of the output
									self.assertEqual(mock_stdout.getvalue()   ,	"Transformers are awesome!\n"		)



			@unittest.mock.patch("sys.stdout"   ,	new_callable=io.StringIO		)
			def 				__lowercase					(			self							:				Any   ,	_lowercase							:				Any		)							->					Dict:
									with ContextManagers([context_en()]		):
															print("Transformers are awesome!"		)
									# The output should be wrapped with an English welcome and goodbye
									self.assertEqual(mock_stdout.getvalue()   ,	"Welcome!\nTransformers are awesome!\nBye!\n"		)



			@unittest.mock.patch("sys.stdout"   ,	new_callable=io.StringIO		)
			def 				__lowercase					(			self							:				str   ,	_lowercase							:				List[Any]		)							->					Optional[int]:
									with ContextManagers([context_fr(), context_en()]		):
															print("Transformers are awesome!"		)
									# The output should be wrapped with an English and French welcome and goodbye
									self.assertEqual(mock_stdout.getvalue()   ,	"Bonjour!\nWelcome!\nTransformers are awesome!\nBye!\nAu revoir!\n"		)



			@require_torch
			def 				__lowercase					(			self							:				Any		)							->					Union[str, Any]:
									self.assertEqual(find_labels(_lowercase		)   ,	["labels"]		)
									self.assertEqual(find_labels(_lowercase		)   ,	["labels", "next_sentence_label"]		)
									self.assertEqual(find_labels(_lowercase		)   ,	["start_positions", "end_positions"]		)

									class       _a  ( SCREAMING_SNAKE_CASE__):
												pass



									self.assertEqual(find_labels(_lowercase		)   ,	["labels"]		)



			@require_tf
			def 				__lowercase					(			self							:				Optional[Any]		)							->					int:
									self.assertEqual(find_labels(_lowercase		)   ,	["labels"]		)
									self.assertEqual(find_labels(_lowercase		)   ,	["labels", "next_sentence_label"]		)
									self.assertEqual(find_labels(_lowercase		)   ,	["start_positions", "end_positions"]		)

									class       _a  ( SCREAMING_SNAKE_CASE__):
												pass



									self.assertEqual(find_labels(_lowercase		)   ,	["labels"]		)



			@require_flax
			def 				__lowercase					(			self							:				Tuple		)							->					List[Any]:
									# Flax models don't have labels
									self.assertEqual(find_labels(_lowercase		)   ,	[]		)
									self.assertEqual(find_labels(_lowercase		)   ,	[]		)
									self.assertEqual(find_labels(_lowercase		)   ,	[]		)

									class       _a  ( SCREAMING_SNAKE_CASE__):
												pass



									self.assertEqual(find_labels(_lowercase		)   ,	[]		)



