


'''simple docstring'''

import json
import os
import tempfile
import unittest
import unittest.mock as mock
from pathlib import Path

from requests.exceptions import HTTPError

from transformers.utils import (
    CONFIG_NAME,
    FLAX_WEIGHTS_NAME,
    TF2_WEIGHTS_NAME,
    TRANSFORMERS_CACHE,
    WEIGHTS_NAME,
    cached_file,
    get_file_from_repo,
    has_file,
)


lowercase__				     =		"hf-internal-testing/tiny-random-bert"
lowercase__				     =		os.path.join(TRANSFORMERS_CACHE, "models--hf-internal-testing--tiny-random-bert")
lowercase__				     =		"9b8c223d42b2188cb49d29af482996f9d0f3e5a6"





class 						__SCREAMING_SNAKE_CASE					(      unittest.TestCase ):






					def  a_					(			self		)       ->    Tuple:
										_a						=		cached_file(_lowerCamelCase     ,							_lowerCamelCase		)
										# Should have downloaded the file in here
										self.assertTrue(os.path.isdir(_lowerCamelCase		)		)
										# Cache should contain at least those three subfolders:
										for subfolder in ["blobs", "refs", "snapshots"]:
															self.assertTrue(os.path.isdir(os.path.join(_lowerCamelCase     ,							_lowerCamelCase		)		)		)
										with open(os.path.join(_lowerCamelCase     ,							"refs"     ,							"main"		)		) as f:
															_a						=		f.read()
										self.assertEqual(_lowerCamelCase     ,							os.path.join(_lowerCamelCase     ,							"snapshots"     ,							_lowerCamelCase     ,							_lowerCamelCase		)		)
										self.assertTrue(os.path.isfile(_lowerCamelCase		)		)

										# File is cached at the same place the second time.
										_a						=		cached_file(_lowerCamelCase     ,							_lowerCamelCase		)
										self.assertEqual(_lowerCamelCase     ,							_lowerCamelCase		)

										# Using a specific revision to test the full commit hash.
										_a						=		cached_file(_lowerCamelCase     ,							_lowerCamelCase     ,							revision="9b8c223"		)
										self.assertEqual(_lowerCamelCase     ,							os.path.join(_lowerCamelCase     ,							"snapshots"     ,							_lowerCamelCase     ,							_lowerCamelCase		)		)






					def  a_					(			self		)       ->    Tuple:
										with self.assertRaisesRegex(_lowerCamelCase     ,							"is not a valid model identifier"		):
															_a						=		cached_file("tiny-random-bert"     ,							_lowerCamelCase		)

										with self.assertRaisesRegex(_lowerCamelCase     ,							"is not a valid git identifier"		):
															_a						=		cached_file(_lowerCamelCase     ,							_lowerCamelCase     ,							revision="aaaa"		)

										with self.assertRaisesRegex(_lowerCamelCase     ,							"does not appear to have a file named"		):
															_a						=		cached_file(_lowerCamelCase     ,							"conf"		)






					def  a_					(			self		)       ->    Tuple:
										with self.assertRaisesRegex(_lowerCamelCase     ,							"does not appear to have a file named"		):
															_a						=		cached_file(_lowerCamelCase     ,							"conf"		)

										with open(os.path.join(_lowerCamelCase     ,							"refs"     ,							"main"		)		) as f:
															_a						=		f.read()
										self.assertTrue(os.path.isfile(os.path.join(_lowerCamelCase     ,							".no_exist"     ,							_lowerCamelCase     ,							"conf"		)		)		)

										_a						=		cached_file(_lowerCamelCase     ,							"conf"     ,							_raise_exceptions_for_missing_entries=_lowerCamelCase		)
										self.assertIsNone(_lowerCamelCase		)

										_a						=		cached_file(_lowerCamelCase     ,							"conf"     ,							local_files_only=_lowerCamelCase     ,							_raise_exceptions_for_missing_entries=_lowerCamelCase		)
										self.assertIsNone(_lowerCamelCase		)

										_a						=		mock.Mock()
										_a						=		500
										_a						=		{}
										_a						=		HTTPError
										_a						=		{}

										# Under the mock environment we get a 500 error when trying to reach the tokenizer.
										with mock.patch("requests.Session.request"     ,							return_value=_lowerCamelCase		) as mock_head:
															_a						=		cached_file(_lowerCamelCase     ,							"conf"     ,							_raise_exceptions_for_connection_errors=_lowerCamelCase		)
															self.assertIsNone(_lowerCamelCase		)
															# This check we did call the fake head request
															mock_head.assert_called()






					def  a_					(			self		)       ->    Tuple:
										self.assertTrue(has_file("hf-internal-testing/tiny-bert-pt-only"     ,							_lowerCamelCase		)		)
										self.assertFalse(has_file("hf-internal-testing/tiny-bert-pt-only"     ,							_lowerCamelCase		)		)
										self.assertFalse(has_file("hf-internal-testing/tiny-bert-pt-only"     ,							_lowerCamelCase		)		)






					def  a_					(			self		)       ->    str:
										# `get_file_from_repo` returns None if the file does not exist
										self.assertIsNone(get_file_from_repo("bert-base-cased"     ,							"ahah.txt"		)		)

										# The function raises if the repository does not exist.
										with self.assertRaisesRegex(_lowerCamelCase     ,							"is not a valid model identifier"		):
															get_file_from_repo("bert-base-case"     ,							_lowerCamelCase		)

										# The function raises if the revision does not exist.
										with self.assertRaisesRegex(_lowerCamelCase     ,							"is not a valid git identifier"		):
															get_file_from_repo("bert-base-cased"     ,							_lowerCamelCase     ,							revision="ahaha"		)

										_a						=		get_file_from_repo("bert-base-cased"     ,							_lowerCamelCase		)
										# The name is the cached name which is not very easy to test, so instead we load the content.
										_a						=		json.loads(open(_lowerCamelCase     ,							"r"		).read()		)
										self.assertEqual(config["hidden_size"]     ,							768		)






					def  a_					(			self		)       ->    Optional[int]:
										with tempfile.TemporaryDirectory() as tmp_dir:
															_a						=		Path(_lowerCamelCase		) / "a.txt"
															filename.touch()
															self.assertEqual(get_file_from_repo(_lowerCamelCase     ,							"a.txt"		)     ,							str(_lowerCamelCase		)		)

															self.assertIsNone(get_file_from_repo(_lowerCamelCase     ,							"b.txt"		)		)





'''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__				     =		False





class 						__SCREAMING_SNAKE_CASE					(      unittest.TestCase ):
     pass







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






     def  a_					(			self		)       ->    Dict:
          # clean up the VRAM after each test
          super().tearDown()
          gc.collect()
          torch.cuda.empty_cache()






     def  a_					(			self		)       ->    List[str]:
          _a						=		VersatileDiffusionTextToImagePipeline.from_pretrained("shi-labs/versatile-diffusion"		)
          # remove text_unet
          pipe.remove_unused_weights()
          pipe.to(__UpperCamelCase		)
          pipe.set_progress_bar_config(disable=__UpperCamelCase		)

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

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

          _a						=		generator.manual_seed(0		)
          _a						=		pipe(
              prompt=__UpperCamelCase     ,							generator=__UpperCamelCase     ,							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  a_					(			self		)       ->    List[Any]:
          _a						=		VersatileDiffusionTextToImagePipeline.from_pretrained(
              "shi-labs/versatile-diffusion"     ,							torch_dtype=torch.floataa		)
          pipe.to(__UpperCamelCase		)
          pipe.set_progress_bar_config(disable=__UpperCamelCase		)

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

          _a						=		image[0, 253:256, 253:256, -1]

          assert image.shape == (1, 512, 512, 3)
          _a						=		np.array([0.33_67, 0.31_69, 0.26_56, 0.38_70, 0.47_90, 0.37_96, 0.40_09, 0.48_78, 0.47_78]		)

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

