
"""simple docstring"""






import json
import os
import shutil
import tempfile
import unittest

import numpy as np
import pytest

from transformers import BertTokenizer, BertTokenizerFast
from transformers.models.bert.tokenization_bert import VOCAB_FILES_NAMES
from transformers.testing_utils import require_vision
from transformers.utils import FEATURE_EXTRACTOR_NAME, is_vision_available


if is_vision_available():
					from PIL import Image

					from transformers import ChineseCLIPImageProcessor, ChineseCLIPProcessor



@require_vision
class 			lowercase					( unittest.TestCase		):






			def 					UpperCAmelCase							(self  :	Optional[int]			)       ->	Any:





				"""simple docstring"""


				lowerCAmelCase												=  tempfile.mkdtemp()

				lowerCAmelCase												=  [
				    """[UNK]""",
				    """[CLS]""",
				    """[SEP]""",
				    """[PAD]""",
				    """[MASK]""",
				    """的""",
				    """价""",
				    """格""",
				    """是""",
				    """15""",
				    """便""",
				    """alex""",
				    """##andra""",
				    """，""",
				    """。""",
				    """-""",
				    """t""",
				    """shirt""",
				]
				lowerCAmelCase												=  os.path.join(self.tmpdirname      ,VOCAB_FILES_NAMES['''vocab_file''']			)
				with open(self.vocab_file      ,'''w'''      ,encoding='''utf-8'''			) as vocab_writer:
					vocab_writer.write(''''''.join([x + '''\n''' for x in vocab_tokens]			)			)

				lowerCAmelCase												=  {
				    """do_resize""": True,
				    """size""": {"""height""": 224, """width""": 224},
				    """do_center_crop""": True,
				    """crop_size""": {"""height""": 18, """width""": 18},
				    """do_normalize""": True,
				    """image_mean""": [0.48_14_54_66, 0.4_57_82_75, 0.40_82_10_73],
				    """image_std""": [0.26_86_29_54, 0.26_13_02_58, 0.27_57_77_11],
				    """do_convert_rgb""": True,
				}
				lowerCAmelCase												=  os.path.join(self.tmpdirname      ,__snake_case			)
				with open(self.image_processor_file      ,'''w'''      ,encoding='''utf-8'''			) as fp:
					json.dump(__snake_case      ,__snake_case			)






			def 					UpperCAmelCase							(self  :	int      ,**SCREAMING_SNAKE_CASE_  :	str			)       ->	Optional[Any]:





				"""simple docstring"""


				return BertTokenizer.from_pretrained(self.tmpdirname      ,**__snake_case			)






			def 					UpperCAmelCase							(self  :	Optional[Any]      ,**SCREAMING_SNAKE_CASE_  :	List[str]			)       ->	str:





				"""simple docstring"""


				return BertTokenizerFast.from_pretrained(self.tmpdirname      ,**__snake_case			)






			def 					UpperCAmelCase							(self  :	Union[str, Any]      ,**SCREAMING_SNAKE_CASE_  :	int			)       ->	List[Any]:





				"""simple docstring"""


				return ChineseCLIPImageProcessor.from_pretrained(self.tmpdirname      ,**__snake_case			)






			def 					UpperCAmelCase							(self  :	List[Any]			)       ->	Any:





				"""simple docstring"""


				shutil.rmtree(self.tmpdirname			)






			def 					UpperCAmelCase							(self  :	int			)       ->	int:





				"""simple docstring"""


				lowerCAmelCase												=  [np.random.randint(255      ,size=(3, 30, 400)      ,dtype=np.uinta			)]

				lowerCAmelCase												=  [Image.fromarray(np.moveaxis(__snake_case      ,0      ,-1			)			) for x in image_inputs]

				return image_inputs






			def 					UpperCAmelCase							(self  :	Any			)       ->	Optional[int]:





				"""simple docstring"""


				lowerCAmelCase												=  self.get_tokenizer()
				lowerCAmelCase												=  self.get_rust_tokenizer()
				lowerCAmelCase												=  self.get_image_processor()

				lowerCAmelCase												=  ChineseCLIPProcessor(tokenizer=__snake_case      ,image_processor=__snake_case			)
				processor_slow.save_pretrained(self.tmpdirname			)
				lowerCAmelCase												=  ChineseCLIPProcessor.from_pretrained(self.tmpdirname      ,use_fast=__snake_case			)

				lowerCAmelCase												=  ChineseCLIPProcessor(tokenizer=__snake_case      ,image_processor=__snake_case			)
				processor_fast.save_pretrained(self.tmpdirname			)
				lowerCAmelCase												=  ChineseCLIPProcessor.from_pretrained(self.tmpdirname			)

				self.assertEqual(processor_slow.tokenizer.get_vocab()      ,tokenizer_slow.get_vocab()			)
				self.assertEqual(processor_fast.tokenizer.get_vocab()      ,tokenizer_fast.get_vocab()			)
				self.assertEqual(tokenizer_slow.get_vocab()      ,tokenizer_fast.get_vocab()			)
				self.assertIsInstance(processor_slow.tokenizer      ,__snake_case			)
				self.assertIsInstance(processor_fast.tokenizer      ,__snake_case			)

				self.assertEqual(processor_slow.image_processor.to_json_string()      ,image_processor.to_json_string()			)
				self.assertEqual(processor_fast.image_processor.to_json_string()      ,image_processor.to_json_string()			)
				self.assertIsInstance(processor_slow.image_processor      ,__snake_case			)
				self.assertIsInstance(processor_fast.image_processor      ,__snake_case			)






			def 					UpperCAmelCase							(self  :	Dict			)       ->	Optional[Any]:





				"""simple docstring"""


				lowerCAmelCase												=  ChineseCLIPProcessor(tokenizer=self.get_tokenizer()      ,image_processor=self.get_image_processor()			)
				processor.save_pretrained(self.tmpdirname			)

				lowerCAmelCase												=  self.get_tokenizer(cls_token='''(CLS)'''      ,sep_token='''(SEP)'''			)
				lowerCAmelCase												=  self.get_image_processor(do_normalize=__snake_case			)

				lowerCAmelCase												=  ChineseCLIPProcessor.from_pretrained(
				    self.tmpdirname      ,cls_token='''(CLS)'''      ,sep_token='''(SEP)'''      ,do_normalize=__snake_case			)

				self.assertEqual(processor.tokenizer.get_vocab()      ,tokenizer_add_kwargs.get_vocab()			)
				self.assertIsInstance(processor.tokenizer      ,__snake_case			)

				self.assertEqual(processor.image_processor.to_json_string()      ,image_processor_add_kwargs.to_json_string()			)
				self.assertIsInstance(processor.image_processor      ,__snake_case			)






			def 					UpperCAmelCase							(self  :	Optional[int]			)       ->	List[str]:





				"""simple docstring"""


				lowerCAmelCase												=  self.get_image_processor()
				lowerCAmelCase												=  self.get_tokenizer()

				lowerCAmelCase												=  ChineseCLIPProcessor(tokenizer=__snake_case      ,image_processor=__snake_case			)

				lowerCAmelCase												=  self.prepare_image_inputs()

				lowerCAmelCase												=  image_processor(__snake_case      ,return_tensors='''np'''			)
				lowerCAmelCase												=  processor(images=__snake_case      ,return_tensors='''np'''			)

				for key in input_feat_extract.keys():
					self.assertAlmostEqual(input_feat_extract[key].sum()      ,input_processor[key].sum()      ,delta=1e-2			)






			def 					UpperCAmelCase							(self  :	Any			)       ->	List[Any]:





				"""simple docstring"""


				lowerCAmelCase												=  self.get_image_processor()
				lowerCAmelCase												=  self.get_tokenizer()

				lowerCAmelCase												=  ChineseCLIPProcessor(tokenizer=__snake_case      ,image_processor=__snake_case			)

				lowerCAmelCase												=  """Alexandra，T-shirt的价格是15便士。"""

				lowerCAmelCase												=  processor(text=__snake_case			)

				lowerCAmelCase												=  tokenizer(__snake_case			)

				for key in encoded_tok.keys():
					self.assertListEqual(encoded_tok[key]      ,encoded_processor[key]			)






			def 					UpperCAmelCase							(self  :	List[str]			)       ->	List[Any]:





				"""simple docstring"""


				lowerCAmelCase												=  self.get_image_processor()
				lowerCAmelCase												=  self.get_tokenizer()

				lowerCAmelCase												=  ChineseCLIPProcessor(tokenizer=__snake_case      ,image_processor=__snake_case			)

				lowerCAmelCase												=  """Alexandra，T-shirt的价格是15便士。"""
				lowerCAmelCase												=  self.prepare_image_inputs()

				lowerCAmelCase												=  processor(text=__snake_case      ,images=__snake_case			)

				self.assertListEqual(list(inputs.keys()			)      ,['''input_ids''', '''token_type_ids''', '''attention_mask''', '''pixel_values''']			)

				# test if it raises when no input is passed
				with pytest.raises(__snake_case			):
					processor()






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





				"""simple docstring"""


				lowerCAmelCase												=  self.get_image_processor()
				lowerCAmelCase												=  self.get_tokenizer()

				lowerCAmelCase												=  ChineseCLIPProcessor(tokenizer=__snake_case      ,image_processor=__snake_case			)

				lowerCAmelCase												=  [[1, 4, 5, 8, 1, 0, 8], [3, 4, 3, 1, 1, 8, 9]]

				lowerCAmelCase												=  processor.batch_decode(__snake_case			)
				lowerCAmelCase												=  tokenizer.batch_decode(__snake_case			)

				self.assertListEqual(__snake_case      ,__snake_case			)






			def 					UpperCAmelCase							(self  :	Dict			)       ->	List[Any]:





				"""simple docstring"""


				lowerCAmelCase												=  self.get_image_processor()
				lowerCAmelCase												=  self.get_tokenizer()

				lowerCAmelCase												=  ChineseCLIPProcessor(tokenizer=__snake_case      ,image_processor=__snake_case			)

				lowerCAmelCase												=  """Alexandra，T-shirt的价格是15便士。"""
				lowerCAmelCase												=  self.prepare_image_inputs()

				lowerCAmelCase												=  processor(text=__snake_case      ,images=__snake_case			)

				self.assertListEqual(list(inputs.keys()			)      ,processor.model_input_names			)






'''simple docstring'''


import subprocess
import sys

from transformers import BertConfig, BertModel, BertTokenizer, pipeline
from transformers.testing_utils import TestCasePlus, require_torch





class        lowercase__    (		_snake_case   ):





							'''simple docstring'''


							@require_torch
							def UpperCAmelCase_      (       self    ):
											# this test is a bit tricky since TRANSFORMERS_OFFLINE can only be changed before
											# `transformers` is loaded, and it's too late for inside pytest - so we are changing it
											# while running an external program

											# python one-liner segments

											# this must be loaded before socket.socket is monkey-patched
											_SCREAMING_SNAKE_CASE					:			Optional[int]							= """
from transformers import BertConfig, BertModel, BertTokenizer, pipeline
        """

											_SCREAMING_SNAKE_CASE					:			Dict							= """
mname = \"hf-internal-testing/tiny-random-bert\"
BertConfig.from_pretrained(mname)
BertModel.from_pretrained(mname)
BertTokenizer.from_pretrained(mname)
pipe = pipeline(task=\"fill-mask\", model=mname)
print(\"success\")
        """

											_SCREAMING_SNAKE_CASE					:			Optional[Any]							= """
import socket
def offline_socket(*args, **kwargs): raise RuntimeError(\"Offline mode is enabled, we shouldn't access internet\")
socket.socket = offline_socket
        """

											# Force fetching the files so that we can use the cache
											_SCREAMING_SNAKE_CASE					:			Optional[Any]							= """hf-internal-testing/tiny-random-bert"""
											BertConfig.from_pretrained(__snake_case    )
											BertModel.from_pretrained(__snake_case    )
											BertTokenizer.from_pretrained(__snake_case    )
											pipeline(task="""fill-mask"""       ,   model=__snake_case    )

											# baseline - just load from_pretrained with normal network
											_SCREAMING_SNAKE_CASE					:			Union[str, Any]							= [sys.executable, """-c""", """\n""".join([load, run, mock]    )]

											# should succeed
											_SCREAMING_SNAKE_CASE					:			Optional[Any]							= self.get_env()
											# should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
											_SCREAMING_SNAKE_CASE					:			int							= """1"""
											_SCREAMING_SNAKE_CASE					:			str							= subprocess.run(__snake_case       ,   env=__snake_case       ,   check=__snake_case       ,   capture_output=__snake_case    )
											self.assertEqual(result.returncode       ,   0       ,   result.stderr    )
											self.assertIn("""success"""       ,   result.stdout.decode()    )






							@require_torch
							def UpperCAmelCase_      (       self    ):
											# python one-liner segments
											# this must be loaded before socket.socket is monkey-patched
											_SCREAMING_SNAKE_CASE					:			Optional[int]							= """
from transformers import BertConfig, BertModel, BertTokenizer, pipeline
        """

											_SCREAMING_SNAKE_CASE					:			Any							= """
mname = \"hf-internal-testing/tiny-random-bert\"
BertConfig.from_pretrained(mname)
BertModel.from_pretrained(mname)
BertTokenizer.from_pretrained(mname)
pipe = pipeline(task=\"fill-mask\", model=mname)
print(\"success\")
        """

											_SCREAMING_SNAKE_CASE					:			List[Any]							= """
import socket
def offline_socket(*args, **kwargs): raise socket.error(\"Faking flaky internet\")
socket.socket = offline_socket
        """

											# Force fetching the files so that we can use the cache
											_SCREAMING_SNAKE_CASE					:			Tuple							= """hf-internal-testing/tiny-random-bert"""
											BertConfig.from_pretrained(__snake_case    )
											BertModel.from_pretrained(__snake_case    )
											BertTokenizer.from_pretrained(__snake_case    )
											pipeline(task="""fill-mask"""       ,   model=__snake_case    )

											# baseline - just load from_pretrained with normal network
											_SCREAMING_SNAKE_CASE					:			Tuple							= [sys.executable, """-c""", """\n""".join([load, run, mock]    )]

											# should succeed
											_SCREAMING_SNAKE_CASE					:			str							= self.get_env()
											_SCREAMING_SNAKE_CASE					:			Tuple							= subprocess.run(__snake_case       ,   env=__snake_case       ,   check=__snake_case       ,   capture_output=__snake_case    )
											self.assertEqual(result.returncode       ,   0       ,   result.stderr    )
											self.assertIn("""success"""       ,   result.stdout.decode()    )






							@require_torch
							def UpperCAmelCase_      (       self    ):
											# this test is a bit tricky since TRANSFORMERS_OFFLINE can only be changed before
											# `transformers` is loaded, and it's too late for inside pytest - so we are changing it
											# while running an external program

											# python one-liner segments

											# this must be loaded before socket.socket is monkey-patched
											_SCREAMING_SNAKE_CASE					:			Tuple							= """
from transformers import BertConfig, BertModel, BertTokenizer
        """

											_SCREAMING_SNAKE_CASE					:			str							= """
mname = \"hf-internal-testing/tiny-random-bert-sharded\"
BertConfig.from_pretrained(mname)
BertModel.from_pretrained(mname)
print(\"success\")
        """

											_SCREAMING_SNAKE_CASE					:			List[Any]							= """
import socket
def offline_socket(*args, **kwargs): raise ValueError(\"Offline mode is enabled\")
socket.socket = offline_socket
        """

											# baseline - just load from_pretrained with normal network
											_SCREAMING_SNAKE_CASE					:			List[str]							= [sys.executable, """-c""", """\n""".join([load, run]    )]

											# should succeed
											_SCREAMING_SNAKE_CASE					:			List[str]							= self.get_env()
											_SCREAMING_SNAKE_CASE					:			Optional[Any]							= subprocess.run(__snake_case       ,   env=__snake_case       ,   check=__snake_case       ,   capture_output=__snake_case    )
											self.assertEqual(result.returncode       ,   0       ,   result.stderr    )
											self.assertIn("""success"""       ,   result.stdout.decode()    )

											# next emulate no network
											_SCREAMING_SNAKE_CASE					:			List[str]							= [sys.executable, """-c""", """\n""".join([load, mock, run]    )]

											# Doesn't fail anymore since the model is in the cache due to other tests, so commenting this.
											# env["TRANSFORMERS_OFFLINE"] = "0"
											# result = subprocess.run(cmd, env=env, check=False, capture_output=True)
											# self.assertEqual(result.returncode, 1, result.stderr)

											# should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
											_SCREAMING_SNAKE_CASE					:			Dict							= """1"""
											_SCREAMING_SNAKE_CASE					:			int							= subprocess.run(__snake_case       ,   env=__snake_case       ,   check=__snake_case       ,   capture_output=__snake_case    )
											self.assertEqual(result.returncode       ,   0       ,   result.stderr    )
											self.assertIn("""success"""       ,   result.stdout.decode()    )






							@require_torch
							def UpperCAmelCase_      (       self    ):
											_SCREAMING_SNAKE_CASE					:			Optional[int]							= """
from transformers import pipeline
        """
											_SCREAMING_SNAKE_CASE					:			int							= """
mname = \"hf-internal-testing/tiny-random-bert\"
pipe = pipeline(model=mname)
        """

											_SCREAMING_SNAKE_CASE					:			Any							= """
import socket
def offline_socket(*args, **kwargs): raise socket.error(\"Offline mode is enabled\")
socket.socket = offline_socket
        """
											_SCREAMING_SNAKE_CASE					:			Dict							= self.get_env()
											_SCREAMING_SNAKE_CASE					:			Any							= """1"""
											_SCREAMING_SNAKE_CASE					:			str							= [sys.executable, """-c""", """\n""".join([load, mock, run]    )]
											_SCREAMING_SNAKE_CASE					:			List[Any]							= subprocess.run(__snake_case       ,   env=__snake_case       ,   check=__snake_case       ,   capture_output=__snake_case    )
											self.assertEqual(result.returncode       ,   1       ,   result.stderr    )
											self.assertIn(
											    """You cannot infer task automatically within `pipeline` when using offline mode"""       ,   result.stderr.decode().replace("""\n"""       ,   """"""    )       ,   )






							@require_torch
							def UpperCAmelCase_      (       self    ):
											_SCREAMING_SNAKE_CASE					:			Optional[int]							= """
from transformers import AutoModel
        """
											_SCREAMING_SNAKE_CASE					:			Tuple							= """
mname = \"hf-internal-testing/test_dynamic_model\"
AutoModel.from_pretrained(mname, trust_remote_code=True)
print(\"success\")
        """

											# baseline - just load from_pretrained with normal network
											_SCREAMING_SNAKE_CASE					:			List[str]							= [sys.executable, """-c""", """\n""".join([load, run]    )]

											# should succeed
											_SCREAMING_SNAKE_CASE					:			Optional[Any]							= self.get_env()
											_SCREAMING_SNAKE_CASE					:			List[str]							= subprocess.run(__snake_case       ,   env=__snake_case       ,   check=__snake_case       ,   capture_output=__snake_case    )
											self.assertEqual(result.returncode       ,   0       ,   result.stderr    )
											self.assertIn("""success"""       ,   result.stdout.decode()    )

											# should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
											_SCREAMING_SNAKE_CASE					:			Optional[Any]							= """1"""
											_SCREAMING_SNAKE_CASE					:			Any							= subprocess.run(__snake_case       ,   env=__snake_case       ,   check=__snake_case       ,   capture_output=__snake_case    )
											self.assertEqual(result.returncode       ,   0       ,   result.stderr    )
											self.assertIn("""success"""       ,   result.stdout.decode()    )




