



'''simple docstring'''

from dataclasses import dataclass
from enum import Enum
from typing import List, Optional, Union

import numpy as np
import PIL
from PIL import Image

from ...utils import BaseOutput, is_torch_available, is_transformers_available

@dataclass
class 	__magic_name__						(    _UpperCamelCase   ):
							lowerCAmelCase       :    Union[List[PIL.Image.Image], np.ndarray]
							lowerCAmelCase       :    Optional[List[bool]]


if is_transformers_available() and is_torch_available():
							from .pipeline_semantic_stable_diffusion import SemanticStableDiffusionPipeline









'''simple docstring'''

import subprocess
import sys

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

class 	__magic_name__						(    _UpperCamelCase   ):







							@require_torch
							def 							__lowercase			(							self					:       Tuple		):
									# 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
									_a     :      Optional[int]        =							'\nfrom transformers import BertConfig, BertModel, BertTokenizer, pipeline\n        '

									_a     :      List[str]        =							'\nmname = "hf-internal-testing/tiny-random-bert"\nBertConfig.from_pretrained(mname)\nBertModel.from_pretrained(mname)\nBertTokenizer.from_pretrained(mname)\npipe = pipeline(task="fill-mask", model=mname)\nprint("success")\n        '

									_a     :      Tuple        =							'\nimport socket\ndef offline_socket(*args, **kwargs): raise RuntimeError("Offline mode is enabled, we shouldn\'t access internet")\nsocket.socket = offline_socket\n        '

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

									# baseline - just load from_pretrained with normal network
									_a     :      Optional[int]        =							[sys.executable, '-c', '\n'.join([load, run, mock]		)]

									# should succeed
									_a     :      Tuple        =							self.get_env()
									# should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
									_a     :      int        =							'1'
									_a     :      List[Any]        =							subprocess.run(_UpperCAmelCase	,env=_UpperCAmelCase	,check=_UpperCAmelCase	,capture_output=_UpperCAmelCase		)
									self.assertEqual(result.returncode	,0	,result.stderr		)
									self.assertIn('success'	,result.stdout.decode()		)







							@require_torch
							def 							__lowercase			(							self					:       Any		):
									# python one-liner segments
									# this must be loaded before socket.socket is monkey-patched
									_a     :      Dict        =							'\nfrom transformers import BertConfig, BertModel, BertTokenizer, pipeline\n        '

									_a     :      Optional[int]        =							'\nmname = "hf-internal-testing/tiny-random-bert"\nBertConfig.from_pretrained(mname)\nBertModel.from_pretrained(mname)\nBertTokenizer.from_pretrained(mname)\npipe = pipeline(task="fill-mask", model=mname)\nprint("success")\n        '

									_a     :      Optional[Any]        =							'\nimport socket\ndef offline_socket(*args, **kwargs): raise socket.error("Faking flaky internet")\nsocket.socket = offline_socket\n        '

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

									# baseline - just load from_pretrained with normal network
									_a     :      Optional[int]        =							[sys.executable, '-c', '\n'.join([load, run, mock]		)]

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







							@require_torch
							def 							__lowercase			(							self					:       List[str]		):
									# 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
									_a     :      Union[str, Any]        =							'\nfrom transformers import BertConfig, BertModel, BertTokenizer\n        '

									_a     :      Optional[Any]        =							'\nmname = "hf-internal-testing/tiny-random-bert-sharded"\nBertConfig.from_pretrained(mname)\nBertModel.from_pretrained(mname)\nprint("success")\n        '

									_a     :      str        =							'\nimport socket\ndef offline_socket(*args, **kwargs): raise ValueError("Offline mode is enabled")\nsocket.socket = offline_socket\n        '

									# baseline - just load from_pretrained with normal network
									_a     :      Optional[Any]        =							[sys.executable, '-c', '\n'.join([load, run]		)]

									# should succeed
									_a     :      Dict        =							self.get_env()
									_a     :      int        =							subprocess.run(_UpperCAmelCase	,env=_UpperCAmelCase	,check=_UpperCAmelCase	,capture_output=_UpperCAmelCase		)
									self.assertEqual(result.returncode	,0	,result.stderr		)
									self.assertIn('success'	,result.stdout.decode()		)

									# next emulate no network
									_a     :      List[Any]        =							[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
									_a     :      int        =							'1'
									_a     :      Any        =							subprocess.run(_UpperCAmelCase	,env=_UpperCAmelCase	,check=_UpperCAmelCase	,capture_output=_UpperCAmelCase		)
									self.assertEqual(result.returncode	,0	,result.stderr		)
									self.assertIn('success'	,result.stdout.decode()		)







							@require_torch
							def 							__lowercase			(							self					:       int		):
									_a     :      Optional[Any]        =							'\nfrom transformers import pipeline\n        '
									_a     :      str        =							'\nmname = "hf-internal-testing/tiny-random-bert"\npipe = pipeline(model=mname)\n        '

									_a     :      List[str]        =							'\nimport socket\ndef offline_socket(*args, **kwargs): raise socket.error("Offline mode is enabled")\nsocket.socket = offline_socket\n        '
									_a     :      List[Any]        =							self.get_env()
									_a     :      Dict        =							'1'
									_a     :      Dict        =							[sys.executable, '-c', '\n'.join([load, mock, run]		)]
									_a     :      str        =							subprocess.run(_UpperCAmelCase	,env=_UpperCAmelCase	,check=_UpperCAmelCase	,capture_output=_UpperCAmelCase		)
									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 							__lowercase			(							self					:       int		):
									_a     :      Optional[int]        =							'\nfrom transformers import AutoModel\n        '
									_a     :      List[Any]        =							'\nmname = "hf-internal-testing/test_dynamic_model"\nAutoModel.from_pretrained(mname, trust_remote_code=True)\nprint("success")\n        '

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

									# should succeed
									_a     :      Tuple        =							self.get_env()
									_a     :      List[str]        =							subprocess.run(_UpperCAmelCase	,env=_UpperCAmelCase	,check=_UpperCAmelCase	,capture_output=_UpperCAmelCase		)
									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
									_a     :      Optional[Any]        =							'1'
									_a     :      Any        =							subprocess.run(_UpperCAmelCase	,env=_UpperCAmelCase	,check=_UpperCAmelCase	,capture_output=_UpperCAmelCase		)
									self.assertEqual(result.returncode	,0	,result.stderr		)
									self.assertIn('success'	,result.stdout.decode()		)




