"""simple docstring"""







import unittest
from typing import Tuple

import torch

from diffusers.utils import floats_tensor, randn_tensor, torch_all_close, torch_device
from diffusers.utils.testing_utils import require_torch

@require_torch
class 		UpperCamelCase      :



				@property
				def  _lowercase					(self    :				Optional[int])   ->     List[str]:
					return self.get_dummy_input()



				@property
				def  _lowercase					(self    :				Any)   ->     Union[str, Any]:
					if self.block_type == "down":
						return (4, 32, 16, 16)
					elif self.block_type == "mid":
						return (4, 32, 32, 32)
					elif self.block_type == "up":
						return (4, 32, 64, 64)

					raise ValueError(f"'{self.block_type}' is not a supported block_type. Set it to 'up', 'mid', or 'down'.")



				def  _lowercase					(self    :				Optional[int]     ,  _A    :				Optional[int]=True     ,  _A    :				Optional[int]=False     ,  _A    :				Tuple=False     ,  _A    :				str=False     ,  )   ->     List[str]:
					__snake_case      :							int						=			4
					__snake_case      :							int						=			32
					__snake_case      :							List[Any]						=			(32, 32)

					__snake_case      :							str						=			torch.manual_seed(0)
					__snake_case      :							str						=			torch.device(_A)
					__snake_case      :							Tuple						=			(batch_size, num_channels) + sizes
					__snake_case      :							Optional[Any]						=			randn_tensor(_A     ,  generator=_A     ,  device=_A)
					__snake_case      :							List[str]						=			{'hidden_states': hidden_states}

					if include_temb:
						__snake_case      :							Any						=			1_28
						__snake_case      :							Optional[Any]						=			randn_tensor((batch_size, temb_channels)     ,  generator=_A     ,  device=_A)

					if include_res_hidden_states_tuple:
						__snake_case      :							List[str]						=			torch.manual_seed(1)
						__snake_case      :							Optional[int]						=			(randn_tensor(_A     ,  generator=_A     ,  device=_A),)

					if include_encoder_hidden_states:
						__snake_case      :							Optional[Any]						=			floats_tensor((batch_size, 32, 32)).to(_A)

					if include_skip_sample:
						__snake_case      :							Optional[Any]						=			randn_tensor(((batch_size, 3) + sizes)     ,  generator=_A     ,  device=_A)

					return dummy_input



				def  _lowercase					(self    :				int)   ->     Dict:
					__snake_case      :							Union[str, Any]						=			{
					    'in_channels': 32,
					    'out_channels': 32,
					    'temb_channels': 1_28,
					}
					if self.block_type == "up":
						__snake_case      :							int						=			32

					if self.block_type == "mid":
						init_dict.pop('out_channels')

					__snake_case      :							str						=			self.dummy_input
					return init_dict, inputs_dict



				def  _lowercase					(self    :				List[str]     ,  _A    :				Optional[int])   ->     str:
					__snake_case		,    __snake_case      :							int						=			self.prepare_init_args_and_inputs_for_common()
					__snake_case      :							Any						=			self.block_class(**_A)
					unet_block.to(_A)
					unet_block.eval()

					with torch.no_grad():
						__snake_case      :							Any						=			unet_block(**_A)

					if isinstance(_A     ,  _A):
						__snake_case      :							List[Any]						=			output[0]

					self.assertEqual(output.shape     ,  self.output_shape)

					__snake_case      :							Optional[int]						=			output[0, -1, -3:, -3:]
					__snake_case      :							Union[str, Any]						=			torch.tensor(_A).to(_A)
					assert torch_all_close(output_slice.flatten()     ,  _A     ,  atol=5E-3)






				@unittest.skipIf(torch_device == 'mps'     ,  'Training is not supported in mps')
				def  _lowercase					(self    :				str)   ->     List[Any]:
					__snake_case		,    __snake_case      :							int						=			self.prepare_init_args_and_inputs_for_common()
					__snake_case      :							Tuple						=			self.block_class(**_A)
					model.to(_A)
					model.train()
					__snake_case      :							Any						=			model(**_A)

					if isinstance(_A     ,  _A):
						__snake_case      :							Any						=			output[0]

					__snake_case      :							Tuple						=			torch.device(_A)
					__snake_case      :							List[str]						=			randn_tensor(output.shape     ,  device=_A)
					__snake_case      :							Union[str, Any]						=			torch.nn.functional.mse_loss(_A     ,  _A)
					loss.backward()






"""simple docstring"""







import subprocess
import sys

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

class 		UpperCamelCase      (   lowercase						):



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

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

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

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



				@require_torch
				def  _lowercase					(self    :				Union[str, Any])   ->     Union[str, Any]:
					# python one-liner segments
					# this must be loaded before socket.socket is monkey-patched
					__snake_case      :							Optional[Any]						=			'\nfrom transformers import BertConfig, BertModel, BertTokenizer, pipeline\n        '

					__snake_case      :							Optional[Any]						=			'\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      :							Union[str, 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
					__snake_case      :							str						=			'hf-internal-testing/tiny-random-bert'
					BertConfig.from_pretrained(_A)
					BertModel.from_pretrained(_A)
					BertTokenizer.from_pretrained(_A)
					pipeline(task='fill-mask'     ,  model=_A)

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

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



				@require_torch
				def  _lowercase					(self    :				int)   ->     Any:
					# 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
					__snake_case      :							int						=			'\nfrom transformers import BertConfig, BertModel, BertTokenizer\n        '

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

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

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

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



				@require_torch
				def  _lowercase					(self    :				str)   ->     Dict:
					__snake_case      :							Dict						=			'\nfrom transformers import pipeline\n        '
					__snake_case      :							List[Any]						=			'\nmname = "hf-internal-testing/tiny-random-bert"\npipe = pipeline(model=mname)\n        '

					__snake_case      :							List[str]						=			'\nimport socket\ndef offline_socket(*args, **kwargs): raise socket.error("Offline mode is enabled")\nsocket.socket = offline_socket\n        '
					__snake_case      :							str						=			self.get_env()
					__snake_case      :							Tuple						=			'1'
					__snake_case      :							Tuple						=			[sys.executable, '-c', '\n'.join([load, mock, run])]
					__snake_case      :							Optional[Any]						=			subprocess.run(_A     ,  env=_A     ,  check=_A     ,  capture_output=_A)
					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)   ->     Optional[Any]:
					__snake_case      :							int						=			'\nfrom transformers import AutoModel\n        '
					__snake_case      :							str						=			'\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      :							str						=			[sys.executable, '-c', '\n'.join([load, run])]

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





