
'''simple docstring'''







import subprocess
import sys

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


class 						__UpperCAmelCase      (					_lowerCamelCase						):


							@require_torch
							def  lowerCamelCase				(       self  ):





								"""simple docstring"""






								_snake_case   						=		'\nfrom transformers import BertConfig, BertModel, BertTokenizer, pipeline\n        '

								_snake_case   						=		'\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        '

								_snake_case   						=		'\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
								_snake_case   						=		'hf-internal-testing/tiny-random-bert'
								BertConfig.from_pretrained(lowerCAmelCase_  )
								BertModel.from_pretrained(lowerCAmelCase_  )
								BertTokenizer.from_pretrained(lowerCAmelCase_  )
								pipeline(task='fill-mask'    ,   model=lowerCAmelCase_  )

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

								# should succeed
								_snake_case   						=		self.get_env()
								# should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
								_snake_case   						=		'1'
								_snake_case   						=		subprocess.run(lowerCAmelCase_    ,   env=lowerCAmelCase_    ,   check=lowerCAmelCase_    ,   capture_output=lowerCAmelCase_  )
								self.assertEqual(result.returncode    ,   0    ,   result.stderr  )
								self.assertIn('success'    ,   result.stdout.decode()  )


							@require_torch
							def  lowerCamelCase				(       self  ):





								"""simple docstring"""






								_snake_case   						=		'\nfrom transformers import BertConfig, BertModel, BertTokenizer, pipeline\n        '

								_snake_case   						=		'\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        '

								_snake_case   						=		'\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
								_snake_case   						=		'hf-internal-testing/tiny-random-bert'
								BertConfig.from_pretrained(lowerCAmelCase_  )
								BertModel.from_pretrained(lowerCAmelCase_  )
								BertTokenizer.from_pretrained(lowerCAmelCase_  )
								pipeline(task='fill-mask'    ,   model=lowerCAmelCase_  )

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

								# should succeed
								_snake_case   						=		self.get_env()
								_snake_case   						=		subprocess.run(lowerCAmelCase_    ,   env=lowerCAmelCase_    ,   check=lowerCAmelCase_    ,   capture_output=lowerCAmelCase_  )
								self.assertEqual(result.returncode    ,   0    ,   result.stderr  )
								self.assertIn('success'    ,   result.stdout.decode()  )


							@require_torch
							def  lowerCamelCase				(       self  ):





								"""simple docstring"""






								_snake_case   						=		'\nfrom transformers import BertConfig, BertModel, BertTokenizer\n        '

								_snake_case   						=		'\nmname = "hf-internal-testing/tiny-random-bert-sharded"\nBertConfig.from_pretrained(mname)\nBertModel.from_pretrained(mname)\nprint("success")\n        '

								_snake_case   						=		'\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
								_snake_case   						=		[sys.executable, '-c', '\n'.join([load, run]  )]

								# should succeed
								_snake_case   						=		self.get_env()
								_snake_case   						=		subprocess.run(lowerCAmelCase_    ,   env=lowerCAmelCase_    ,   check=lowerCAmelCase_    ,   capture_output=lowerCAmelCase_  )
								self.assertEqual(result.returncode    ,   0    ,   result.stderr  )
								self.assertIn('success'    ,   result.stdout.decode()  )

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


							@require_torch
							def  lowerCamelCase				(       self  ):





								"""simple docstring"""






								_snake_case   						=		'\nfrom transformers import pipeline\n        '
								_snake_case   						=		'\nmname = "hf-internal-testing/tiny-random-bert"\npipe = pipeline(model=mname)\n        '

								_snake_case   						=		'\nimport socket\ndef offline_socket(*args, **kwargs): raise socket.error("Offline mode is enabled")\nsocket.socket = offline_socket\n        '
								_snake_case   						=		self.get_env()
								_snake_case   						=		'1'
								_snake_case   						=		[sys.executable, '-c', '\n'.join([load, mock, run]  )]
								_snake_case   						=		subprocess.run(lowerCAmelCase_    ,   env=lowerCAmelCase_    ,   check=lowerCAmelCase_    ,   capture_output=lowerCAmelCase_  )
								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  lowerCamelCase				(       self  ):





								"""simple docstring"""






								_snake_case   						=		'\nfrom transformers import AutoModel\n        '
								_snake_case   						=		'\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
								_snake_case   						=		[sys.executable, '-c', '\n'.join([load, run]  )]

								# should succeed
								_snake_case   						=		self.get_env()
								_snake_case   						=		subprocess.run(lowerCAmelCase_    ,   env=lowerCAmelCase_    ,   check=lowerCAmelCase_    ,   capture_output=lowerCAmelCase_  )
								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
								_snake_case   						=		'1'
								_snake_case   						=		subprocess.run(lowerCAmelCase_    ,   env=lowerCAmelCase_    ,   check=lowerCAmelCase_    ,   capture_output=lowerCAmelCase_  )
								self.assertEqual(result.returncode    ,   0    ,   result.stderr  )
								self.assertIn('success'    ,   result.stdout.decode()  )



'''simple docstring'''







import gc
import tempfile
import unittest

import numpy as np
import torch

from diffusers import VersatileDiffusionTextToImagePipeline
from diffusers.utils.testing_utils import nightly, require_torch_gpu, torch_device


lowercase    :							Optional[Any]      =					False


class 						__UpperCAmelCase      (					unittest.TestCase						):
							pass


@nightly
@require_torch_gpu
class 						__UpperCAmelCase      (					unittest.TestCase						):


							def  lowerCamelCase				(       self  ):





								"""simple docstring"""






								super().tearDown()
								gc.collect()
								torch.cuda.empty_cache()


							def  lowerCamelCase				(       self  ):





								"""simple docstring"""






								_snake_case   						=		VersatileDiffusionTextToImagePipeline.from_pretrained('shi-labs/versatile-diffusion'  )
								# remove text_unet
								pipe.remove_unused_weights()
								pipe.to(lowerCAmelCase_  )
								pipe.set_progress_bar_config(disable=lowerCAmelCase_  )

								_snake_case   						=		'A painting of a squirrel eating a burger '
								_snake_case   						=		torch.manual_seed(0  )
								_snake_case   						=		pipe(
								    prompt=lowerCAmelCase_    ,   generator=lowerCAmelCase_    ,   guidance_scale=7.5    ,   num_inference_steps=2    ,   output_type='numpy'  ).images

								with tempfile.TemporaryDirectory() as tmpdirname:
									pipe.save_pretrained(lowerCAmelCase_  )
									_snake_case   						=		VersatileDiffusionTextToImagePipeline.from_pretrained(lowerCAmelCase_  )
								pipe.to(lowerCAmelCase_  )
								pipe.set_progress_bar_config(disable=lowerCAmelCase_  )

								_snake_case   						=		generator.manual_seed(0  )
								_snake_case   						=		pipe(
								    prompt=lowerCAmelCase_    ,   generator=lowerCAmelCase_    ,   guidance_scale=7.5    ,   num_inference_steps=2    ,   output_type='numpy'  ).images

								assert np.abs(image - new_image  ).sum() < 1E-5, "Models don't have the same forward pass"




							def  lowerCamelCase				(       self  ):





								"""simple docstring"""






								_snake_case   						=		VersatileDiffusionTextToImagePipeline.from_pretrained(
								    'shi-labs/versatile-diffusion'    ,   torch_dtype=torch.floataa  )
								pipe.to(lowerCAmelCase_  )
								pipe.set_progress_bar_config(disable=lowerCAmelCase_  )

								_snake_case   						=		'A painting of a squirrel eating a burger '
								_snake_case   						=		torch.manual_seed(0  )
								_snake_case   						=		pipe(
								    prompt=lowerCAmelCase_    ,   generator=lowerCAmelCase_    ,   guidance_scale=7.5    ,   num_inference_steps=50    ,   output_type='numpy'  ).images

								_snake_case   						=		image[0, 2_53:2_56, 2_53:2_56, -1]

								assert image.shape == (1, 5_12, 5_12, 3)
								_snake_case   						=		np.array([0.3367, 0.3169, 0.2656, 0.3870, 0.4790, 0.3796, 0.4009, 0.4878, 0.4778]  )

								assert np.abs(image_slice.flatten() - expected_slice  ).max() < 1E-2

