





import unittest

import numpy as np

from transformers.file_utils import is_torch_available, is_vision_available
from transformers.testing_utils import require_torch, require_vision

from ...test_image_processing_common import ImageProcessingSavingTestMixin, prepare_image_inputs


if is_torch_available():
  import torch

if is_vision_available():
  from PIL import Image

  from transformers import DPTImageProcessor







class   __lowerCAmelCase			(					unittest.TestCase  ):





   def __init__( self					,			snake_case					,			snake_case=7					,			snake_case=3					,			snake_case=18					,			snake_case=30					,			snake_case=400					,			snake_case=True					,			snake_case=None					,			snake_case=True					,			snake_case=[0.5, 0.5, 0.5]					,			snake_case=[0.5, 0.5, 0.5]					,			)				->  Union[str, Any]:
      """simple docstring"""





      a__	:						Union[str, Any]      		=  size if size is not None else {"""height""": 18, """width""": 18}
      a__	:						Tuple      		=  parent
      a__	:						Any      		=  batch_size
      a__	:						Tuple      		=  num_channels
      a__	:						int      		=  image_size
      a__	:						Union[str, Any]      		=  min_resolution
      a__	:						List[str]      		=  max_resolution
      a__	:						str      		=  do_resize
      a__	:						Dict      		=  size
      a__	:						str      		=  do_normalize
      a__	:						Union[str, Any]      		=  image_mean
      a__	:						str      		=  image_std





   def 	_snake_case ( self      )				->  str:
      """simple docstring"""





      return {
          "image_mean": self.image_mean,
          "image_std": self.image_std,
          "do_normalize": self.do_normalize,
          "do_resize": self.do_resize,
          "size": self.size,
      }







@require_torch
@require_vision
class   __lowerCAmelCase			(					__SCREAMING_SNAKE_CASE		,unittest.TestCase  ):
   _UpperCamelCase :						int     		=				DPTImageProcessor if is_vision_available() else None





   def 	_snake_case ( self      )				->  Tuple:
      """simple docstring"""





      a__	:						Tuple      		=  DPTImageProcessingTester(self      )





   @property
   def 	_snake_case ( self      )				->  List[Any]:
      """simple docstring"""





      return self.image_processor_tester.prepare_image_processor_dict()





   def 	_snake_case ( self      )				->  Any:
      """simple docstring"""





      a__	:						List[Any]      		=  self.image_processing_class(**self.image_processor_dict      )
      self.assertTrue(hasattr(_a					,			"image_mean"      )      )
      self.assertTrue(hasattr(_a					,			"image_std"      )      )
      self.assertTrue(hasattr(_a					,			"do_normalize"      )      )
      self.assertTrue(hasattr(_a					,			"do_resize"      )      )
      self.assertTrue(hasattr(_a					,			"size"      )      )





   def 	_snake_case ( self      )				->  int:
      """simple docstring"""





      a__	:						Dict      		=  self.image_processing_class.from_dict(self.image_processor_dict      )
      self.assertEqual(image_processor.size					,			{"height": 18, "width": 18}      )

      a__	:						Union[str, Any]      		=  self.image_processing_class.from_dict(self.image_processor_dict					,			size=42      )
      self.assertEqual(image_processor.size					,			{"height": 42, "width": 42}      )





   def 	_snake_case ( self      )				->  List[Any]:
      """simple docstring"""





      a__	:						int      		=  self.image_processing_class(**self.image_processor_dict      )
      # create random PIL images
      a__	:						Optional[Any]      		=  prepare_image_inputs(self.image_processor_tester					,			equal_resolution=_a      )
      for image in image_inputs:
         self.assertIsInstance(_a					,			Image.Image      )

      # Test not batched input
      a__	:						Optional[int]      		=  image_processing(image_inputs[0]					,			return_tensors="pt"      ).pixel_values
      self.assertEqual(
          encoded_images.shape					,			(
              1,
              self.image_processor_tester.num_channels,
              self.image_processor_tester.size["height"],
              self.image_processor_tester.size["width"],
          )					,			)

      # Test batched
      a__	:						Union[str, Any]      		=  image_processing(_a					,			return_tensors="pt"      ).pixel_values
      self.assertEqual(
          encoded_images.shape					,			(
              self.image_processor_tester.batch_size,
              self.image_processor_tester.num_channels,
              self.image_processor_tester.size["height"],
              self.image_processor_tester.size["width"],
          )					,			)





   def 	_snake_case ( self      )				->  Optional[int]:
      """simple docstring"""





      a__	:						str      		=  self.image_processing_class(**self.image_processor_dict      )
      # create random numpy tensors
      a__	:						Union[str, Any]      		=  prepare_image_inputs(self.image_processor_tester					,			equal_resolution=_a					,			numpify=_a      )
      for image in image_inputs:
         self.assertIsInstance(_a					,			np.ndarray      )

      # Test not batched input
      a__	:						List[Any]      		=  image_processing(image_inputs[0]					,			return_tensors="pt"      ).pixel_values
      self.assertEqual(
          encoded_images.shape					,			(
              1,
              self.image_processor_tester.num_channels,
              self.image_processor_tester.size["height"],
              self.image_processor_tester.size["width"],
          )					,			)

      # Test batched
      a__	:						Optional[Any]      		=  image_processing(_a					,			return_tensors="pt"      ).pixel_values
      self.assertEqual(
          encoded_images.shape					,			(
              self.image_processor_tester.batch_size,
              self.image_processor_tester.num_channels,
              self.image_processor_tester.size["height"],
              self.image_processor_tester.size["width"],
          )					,			)





   def 	_snake_case ( self      )				->  Any:
      """simple docstring"""





      a__	:						Union[str, Any]      		=  self.image_processing_class(**self.image_processor_dict      )
      # create random PyTorch tensors
      a__	:						str      		=  prepare_image_inputs(self.image_processor_tester					,			equal_resolution=_a					,			torchify=_a      )
      for image in image_inputs:
         self.assertIsInstance(_a					,			torch.Tensor      )

      # Test not batched input
      a__	:						List[str]      		=  image_processing(image_inputs[0]					,			return_tensors="pt"      ).pixel_values
      self.assertEqual(
          encoded_images.shape					,			(
              1,
              self.image_processor_tester.num_channels,
              self.image_processor_tester.size["height"],
              self.image_processor_tester.size["width"],
          )					,			)

      # Test batched
      a__	:						Union[str, Any]      		=  image_processing(_a					,			return_tensors="pt"      ).pixel_values
      self.assertEqual(
          encoded_images.shape					,			(
              self.image_processor_tester.batch_size,
              self.image_processor_tester.num_channels,
              self.image_processor_tester.size["height"],
              self.image_processor_tester.size["width"],
          )					,			)








import subprocess
import sys

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







class   __lowerCAmelCase			(					_UpperCamelCase  ):





   @require_torch
   def 	_snake_case ( self      )				->  str:
      """simple docstring"""





      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__	:						Optional[Any]      		=  "\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__	:						Tuple      		=  "hf-internal-testing/tiny-random-bert"
      BertConfig.from_pretrained(snake_case      )
      BertModel.from_pretrained(snake_case      )
      BertTokenizer.from_pretrained(snake_case      )
      pipeline(task="fill-mask"					,			model=snake_case      )

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

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





   @require_torch
   def 	_snake_case ( self      )				->  List[Any]:
      """simple docstring"""





      a__	:						Tuple      		=  "\nfrom transformers import BertConfig, BertModel, BertTokenizer, pipeline\n        "

      a__	:						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        "

      a__	:						Tuple      		=  "\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__	:						List[Any]      		=  "hf-internal-testing/tiny-random-bert"
      BertConfig.from_pretrained(snake_case      )
      BertModel.from_pretrained(snake_case      )
      BertTokenizer.from_pretrained(snake_case      )
      pipeline(task="fill-mask"					,			model=snake_case      )

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

      # should succeed
      a__	:						Any      		=  self.get_env()
      a__	:						int      		=  subprocess.run(snake_case					,			env=snake_case					,			check=snake_case					,			capture_output=snake_case      )
      self.assertEqual(result.returncode					,			0					,			result.stderr      )
      self.assertIn("success"					,			result.stdout.decode()      )





   @require_torch
   def 	_snake_case ( self      )				->  Tuple:
      """simple docstring"""





      a__	:						List[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__	:						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
      a__	:						List[Any]      		=  [sys.executable, "-c", "\n".join([load, run]      )]

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





   @require_torch
   def 	_snake_case ( self      )				->  Tuple:
      """simple docstring"""





      a__	:						Optional[Any]      		=  "\nfrom transformers import pipeline\n        "
      a__	:						int      		=  "\nmname = \"hf-internal-testing/tiny-random-bert\"\npipe = pipeline(model=mname)\n        "

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





      a__	:						Any      		=  "\nfrom transformers import AutoModel\n        "
      a__	:						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__	:						List[str]      		=  [sys.executable, "-c", "\n".join([load, run]      )]

      # should succeed
      a__	:						Optional[Any]      		=  self.get_env()
      a__	:						Union[str, Any]      		=  subprocess.run(snake_case					,			env=snake_case					,			check=snake_case					,			capture_output=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__	:						Dict      		=  "1"
      a__	:						Union[str, Any]      		=  subprocess.run(snake_case					,			env=snake_case					,			check=snake_case					,			capture_output=snake_case      )
      self.assertEqual(result.returncode					,			0					,			result.stderr      )
      self.assertIn("success"					,			result.stdout.decode()      )

