


import random
import unittest

import torch

from diffusers import IFInpaintingPipeline
from diffusers.utils import floats_tensor
from diffusers.utils.import_utils import is_xformers_available
from diffusers.utils.testing_utils import skip_mps, torch_device

from ..pipeline_params import (
    TEXT_GUIDED_IMAGE_INPAINTING_BATCH_PARAMS,
    TEXT_GUIDED_IMAGE_INPAINTING_PARAMS,
)
from ..test_pipelines_common import PipelineTesterMixin
from . import IFPipelineTesterMixin
@skip_mps
class 			SCREAMING_SNAKE_CASE__  (    lowercase__						,   lowercase__						,   unittest.TestCase      ):
      snake_case__    :					Optional[Any]					       =						IFInpaintingPipeline
      snake_case__    :					Dict					       =						TEXT_GUIDED_IMAGE_INPAINTING_PARAMS - {'''width''', '''height'''}
      snake_case__    :					Optional[int]					       =						TEXT_GUIDED_IMAGE_INPAINTING_BATCH_PARAMS
      snake_case__    :					str					       =						PipelineTesterMixin.required_optional_params - {'''latents'''}
      def 	SCREAMING_SNAKE_CASE							(	self    :	List[Any]					)		->      int:
             return self._get_dummy_components()
      def 	SCREAMING_SNAKE_CASE							(	self    :	str					,							SCREAMING_SNAKE_CASE__    :	Any					,							SCREAMING_SNAKE_CASE__    :	List[Any]=0					)		->      int:
             if str(SCREAMING_SNAKE_CASE__					).startswith('mps'					):
                    a_ :		List[Any]							=  torch.manual_seed(SCREAMING_SNAKE_CASE__					)
             else:
                    a_ :		Optional[Any]							=  torch.Generator(device=SCREAMING_SNAKE_CASE__					).manual_seed(SCREAMING_SNAKE_CASE__					)

             a_ :		Any							=  floats_tensor((1, 3, 3_2, 3_2)					,							rng=random.Random(SCREAMING_SNAKE_CASE__					)					).to(SCREAMING_SNAKE_CASE__					)
             a_ :		List[Any]							=  floats_tensor((1, 3, 3_2, 3_2)					,							rng=random.Random(SCREAMING_SNAKE_CASE__					)					).to(SCREAMING_SNAKE_CASE__					)

             a_ :		Optional[int]							=  {
                 'prompt': 'A painting of a squirrel eating a burger',
                 'image': image,
                 'mask_image': mask_image,
                 'generator': generator,
                 'num_inference_steps': 2,
                 'output_type': 'numpy',
             }

             return inputs
      @unittest.skipIf(
          torch_device != 'cuda' or not is_xformers_available()					,							reason='XFormers attention is only available with CUDA and `xformers` installed'					,							)
      def 	SCREAMING_SNAKE_CASE							(	self    :	List[str]					)		->      Optional[Any]:
             self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=1E-3					)
      def 	SCREAMING_SNAKE_CASE							(	self    :	str					)		->      Dict:
             self._test_save_load_optional_components()
      @unittest.skipIf(torch_device != 'cuda'					,							reason='float16 requires CUDA'					)
      def 	SCREAMING_SNAKE_CASE							(	self    :	Dict					)		->      List[str]:
             # Due to non-determinism in save load of the hf-internal-testing/tiny-random-t5 text encoder
             super().test_save_load_floataa(expected_max_diff=1E-1					)
      def 	SCREAMING_SNAKE_CASE							(	self    :	Union[str, Any]					)		->      Any:
             self._test_attention_slicing_forward_pass(expected_max_diff=1E-2					)
      def 	SCREAMING_SNAKE_CASE							(	self    :	str					)		->      List[str]:
             self._test_save_load_local()

      def 	SCREAMING_SNAKE_CASE							(	self    :	Optional[Any]					)		->      int:
             self._test_inference_batch_single_identical(
                 expected_max_diff=1E-2					,							)






import subprocess
import sys

from transformers import BertConfig, BertModel, BertTokenizer, pipeline
from transformers.testing_utils import TestCasePlus, require_torch
class 			SCREAMING_SNAKE_CASE__  (    lowercase__      ):
      @require_torch
      def 	SCREAMING_SNAKE_CASE							(	self    :	Optional[Any]					)		->      List[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
             a_ :		str							=  '\nfrom transformers import BertConfig, BertModel, BertTokenizer, pipeline\n        '

             a_ :		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        '

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

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

             # should succeed
             a_ :		List[str]							=  self.get_env()
             # should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
             a_ :		Any							=  '1'
             a_ :		Optional[Any]							=  subprocess.run(SCREAMING_SNAKE_CASE__					,							env=SCREAMING_SNAKE_CASE__					,							check=SCREAMING_SNAKE_CASE__					,							capture_output=SCREAMING_SNAKE_CASE__					)
             self.assertEqual(result.returncode					,							0					,							result.stderr					)
             self.assertIn('success'					,							result.stdout.decode()					)
      @require_torch
      def 	SCREAMING_SNAKE_CASE							(	self    :	str					)		->      List[Any]:
             # python one-liner segments
             # this must be loaded before socket.socket is monkey-patched
             a_ :		Tuple							=  '\nfrom transformers import BertConfig, BertModel, BertTokenizer, pipeline\n        '

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

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

             # should succeed
             a_ :		Dict							=  self.get_env()
             a_ :		Dict							=  subprocess.run(SCREAMING_SNAKE_CASE__					,							env=SCREAMING_SNAKE_CASE__					,							check=SCREAMING_SNAKE_CASE__					,							capture_output=SCREAMING_SNAKE_CASE__					)
             self.assertEqual(result.returncode					,							0					,							result.stderr					)
             self.assertIn('success'					,							result.stdout.decode()					)
      @require_torch
      def 	SCREAMING_SNAKE_CASE							(	self    :	int					)		->      Union[str, 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
             a_ :		Any							=  '\nfrom transformers import BertConfig, BertModel, BertTokenizer\n        '

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

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

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

             # next emulate no network
             a_ :		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
             a_ :		Dict							=  '1'
             a_ :		int							=  subprocess.run(SCREAMING_SNAKE_CASE__					,							env=SCREAMING_SNAKE_CASE__					,							check=SCREAMING_SNAKE_CASE__					,							capture_output=SCREAMING_SNAKE_CASE__					)
             self.assertEqual(result.returncode					,							0					,							result.stderr					)
             self.assertIn('success'					,							result.stdout.decode()					)
      @require_torch
      def 	SCREAMING_SNAKE_CASE							(	self    :	Optional[int]					)		->      Union[str, Any]:
             a_ :		Union[str, Any]							=  '\nfrom transformers import pipeline\n        '
             a_ :		Optional[Any]							=  '\nmname = "hf-internal-testing/tiny-random-bert"\npipe = pipeline(model=mname)\n        '

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

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


