


'''simple docstring'''






import subprocess
import sys

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






class        __snake_case	(       _SCREAMING_SNAKE_CASE):







    """simple docstring"""



    @require_torch
    def __lowercase       (       self   :  Optional[Any]		)       ->  Tuple:
      # 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
      lowerCAmelCase_      :				Dict      =       """
from transformers import BertConfig, BertModel, BertTokenizer, pipeline
        """

      lowerCAmelCase_      :				Tuple      =       """
mname = \"hf-internal-testing/tiny-random-bert\"
BertConfig.from_pretrained(mname)
BertModel.from_pretrained(mname)
BertTokenizer.from_pretrained(mname)
pipe = pipeline(task=\"fill-mask\", model=mname)
print(\"success\")
        """

      lowerCAmelCase_      :				Any      =       """
import socket
def offline_socket(*args, **kwargs): raise RuntimeError(\"Offline mode is enabled, we shouldn't access internet\")
socket.socket = offline_socket
        """

      # Force fetching the files so that we can use the cache
      lowerCAmelCase_      :				List[Any]      =       """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
      lowerCAmelCase_      :				Union[str, Any]      =       [sys.executable, """-c""", """\n""".join([load, run, mock]		)]

      # should succeed
      lowerCAmelCase_      :				int      =       self.get_env()
      # should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
      lowerCAmelCase_      :				Union[str, Any]      =       """1"""
      lowerCAmelCase_      :				List[Any]      =       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 __lowercase       (       self   :  List[str]		)       ->  List[Any]:
      # python one-liner segments
      # this must be loaded before socket.socket is monkey-patched
      lowerCAmelCase_      :				List[str]      =       """
from transformers import BertConfig, BertModel, BertTokenizer, pipeline
        """

      lowerCAmelCase_      :				int      =       """
mname = \"hf-internal-testing/tiny-random-bert\"
BertConfig.from_pretrained(mname)
BertModel.from_pretrained(mname)
BertTokenizer.from_pretrained(mname)
pipe = pipeline(task=\"fill-mask\", model=mname)
print(\"success\")
        """

      lowerCAmelCase_      :				Union[str, Any]      =       """
import socket
def offline_socket(*args, **kwargs): raise socket.error(\"Faking flaky internet\")
socket.socket = offline_socket
        """

      # Force fetching the files so that we can use the cache
      lowerCAmelCase_      :				Any      =       """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
      lowerCAmelCase_      :				str      =       [sys.executable, """-c""", """\n""".join([load, run, mock]		)]

      # should succeed
      lowerCAmelCase_      :				Union[str, Any]      =       self.get_env()
      lowerCAmelCase_      :				Optional[int]      =       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 __lowercase       (       self   :  Tuple		)       ->  Dict:
      # 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
      lowerCAmelCase_      :				str      =       """
from transformers import BertConfig, BertModel, BertTokenizer
        """

      lowerCAmelCase_      :				Optional[Any]      =       """
mname = \"hf-internal-testing/tiny-random-bert-sharded\"
BertConfig.from_pretrained(mname)
BertModel.from_pretrained(mname)
print(\"success\")
        """

      lowerCAmelCase_      :				int      =       """
import socket
def offline_socket(*args, **kwargs): raise ValueError(\"Offline mode is enabled\")
socket.socket = offline_socket
        """

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

      # should succeed
      lowerCAmelCase_      :				List[str]      =       self.get_env()
      lowerCAmelCase_      :				Union[str, Any]      =       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
      lowerCAmelCase_      :				Union[str, 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
      lowerCAmelCase_      :				int      =       """1"""
      lowerCAmelCase_      :				Union[str, Any]      =       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 __lowercase       (       self   :  List[str]		)       ->  Union[str, Any]:
      lowerCAmelCase_      :				Tuple      =       """
from transformers import pipeline
        """
      lowerCAmelCase_      :				Optional[int]      =       """
mname = \"hf-internal-testing/tiny-random-bert\"
pipe = pipeline(model=mname)
        """

      lowerCAmelCase_      :				List[Any]      =       """
import socket
def offline_socket(*args, **kwargs): raise socket.error(\"Offline mode is enabled\")
socket.socket = offline_socket
        """
      lowerCAmelCase_      :				Optional[int]      =       self.get_env()
      lowerCAmelCase_      :				Union[str, Any]      =       """1"""
      lowerCAmelCase_      :				Dict      =       [sys.executable, """-c""", """\n""".join([load, mock, run]		)]
      lowerCAmelCase_      :				Dict      =       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 __lowercase       (       self   :  int		)       ->  int:
      lowerCAmelCase_      :				Optional[Any]      =       """
from transformers import AutoModel
        """
      lowerCAmelCase_      :				str      =       """
mname = \"hf-internal-testing/test_dynamic_model\"
AutoModel.from_pretrained(mname, trust_remote_code=True)
print(\"success\")
        """

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

      # should succeed
      lowerCAmelCase_      :				Optional[Any]      =       self.get_env()
      lowerCAmelCase_      :				Optional[Any]      =       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
      lowerCAmelCase_      :				str      =       """1"""
      lowerCAmelCase_      :				List[Any]      =       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 unittest

from diffusers.pipelines.pipeline_utils import is_safetensors_compatible






class        __snake_case	(       unittest.TestCase):







    """simple docstring"""



    def __lowercase       (       self   :  Tuple		)       ->  Dict:
      lowerCAmelCase_      :				str      =       [
          """safety_checker/pytorch_model.bin""",
          """safety_checker/model.safetensors""",
          """vae/diffusion_pytorch_model.bin""",
          """vae/diffusion_pytorch_model.safetensors""",
          """text_encoder/pytorch_model.bin""",
          """text_encoder/model.safetensors""",
          """unet/diffusion_pytorch_model.bin""",
          """unet/diffusion_pytorch_model.safetensors""",
      ]
      self.assertTrue(is_safetensors_compatible(lowerCamelCase		)		)

    def __lowercase       (       self   :  List[Any]		)       ->  int:
      lowerCAmelCase_      :				Tuple      =       [
          """unet/diffusion_pytorch_model.bin""",
          """unet/diffusion_pytorch_model.safetensors""",
      ]
      self.assertTrue(is_safetensors_compatible(lowerCamelCase		)		)

    def __lowercase       (       self   :  Optional[int]		)       ->  Optional[Any]:
      lowerCAmelCase_      :				int      =       [
          """safety_checker/pytorch_model.bin""",
          """safety_checker/model.safetensors""",
          """vae/diffusion_pytorch_model.bin""",
          """vae/diffusion_pytorch_model.safetensors""",
          """text_encoder/pytorch_model.bin""",
          """text_encoder/model.safetensors""",
          """unet/diffusion_pytorch_model.bin""",
          # Removed: 'unet/diffusion_pytorch_model.safetensors',
      ]
      self.assertFalse(is_safetensors_compatible(lowerCamelCase		)		)

    def __lowercase       (       self   :  int		)       ->  List[Any]:
      lowerCAmelCase_      :				Dict      =       [
          """text_encoder/pytorch_model.bin""",
          """text_encoder/model.safetensors""",
      ]
      self.assertTrue(is_safetensors_compatible(lowerCamelCase		)		)

    def __lowercase       (       self   :  str		)       ->  List[str]:
      lowerCAmelCase_      :				Union[str, Any]      =       [
          """safety_checker/pytorch_model.bin""",
          """safety_checker/model.safetensors""",
          """vae/diffusion_pytorch_model.bin""",
          """vae/diffusion_pytorch_model.safetensors""",
          """text_encoder/pytorch_model.bin""",
          # Removed: 'text_encoder/model.safetensors',
          """unet/diffusion_pytorch_model.bin""",
          """unet/diffusion_pytorch_model.safetensors""",
      ]
      self.assertFalse(is_safetensors_compatible(lowerCamelCase		)		)

    def __lowercase       (       self   :  List[Any]		)       ->  Tuple:
      lowerCAmelCase_      :				Union[str, Any]      =       [
          """safety_checker/pytorch_model.fp16.bin""",
          """safety_checker/model.fp16.safetensors""",
          """vae/diffusion_pytorch_model.fp16.bin""",
          """vae/diffusion_pytorch_model.fp16.safetensors""",
          """text_encoder/pytorch_model.fp16.bin""",
          """text_encoder/model.fp16.safetensors""",
          """unet/diffusion_pytorch_model.fp16.bin""",
          """unet/diffusion_pytorch_model.fp16.safetensors""",
      ]
      lowerCAmelCase_      :				Union[str, Any]      =       """fp16"""
      self.assertTrue(is_safetensors_compatible(lowerCamelCase   ,						variant=lowerCamelCase		)		)

    def __lowercase       (       self   :  Optional[Any]		)       ->  List[str]:
      lowerCAmelCase_      :				str      =       [
          """unet/diffusion_pytorch_model.fp16.bin""",
          """unet/diffusion_pytorch_model.fp16.safetensors""",
      ]
      lowerCAmelCase_      :				Optional[int]      =       """fp16"""
      self.assertTrue(is_safetensors_compatible(lowerCamelCase   ,						variant=lowerCamelCase		)		)

    def __lowercase       (       self   :  Tuple		)       ->  Optional[Any]:
      # pass variant but use the non-variant filenames
      lowerCAmelCase_      :				Dict      =       [
          """unet/diffusion_pytorch_model.bin""",
          """unet/diffusion_pytorch_model.safetensors""",
      ]
      lowerCAmelCase_      :				str      =       """fp16"""
      self.assertTrue(is_safetensors_compatible(lowerCamelCase   ,						variant=lowerCamelCase		)		)

    def __lowercase       (       self   :  Optional[int]		)       ->  List[str]:
      lowerCAmelCase_      :				str      =       [
          """safety_checker/pytorch_model.fp16.bin""",
          """safety_checker/model.fp16.safetensors""",
          """vae/diffusion_pytorch_model.fp16.bin""",
          """vae/diffusion_pytorch_model.fp16.safetensors""",
          """text_encoder/pytorch_model.fp16.bin""",
          """text_encoder/model.fp16.safetensors""",
          """unet/diffusion_pytorch_model.fp16.bin""",
          # Removed: 'unet/diffusion_pytorch_model.fp16.safetensors',
      ]
      lowerCAmelCase_      :				List[str]      =       """fp16"""
      self.assertFalse(is_safetensors_compatible(lowerCamelCase   ,						variant=lowerCamelCase		)		)

    def __lowercase       (       self   :  Union[str, Any]		)       ->  Optional[int]:
      lowerCAmelCase_      :				str      =       [
          """text_encoder/pytorch_model.fp16.bin""",
          """text_encoder/model.fp16.safetensors""",
      ]
      lowerCAmelCase_      :				Optional[Any]      =       """fp16"""
      self.assertTrue(is_safetensors_compatible(lowerCamelCase   ,						variant=lowerCamelCase		)		)

    def __lowercase       (       self   :  List[Any]		)       ->  List[Any]:
      # pass variant but use the non-variant filenames
      lowerCAmelCase_      :				Dict      =       [
          """text_encoder/pytorch_model.bin""",
          """text_encoder/model.safetensors""",
      ]
      lowerCAmelCase_      :				Any      =       """fp16"""
      self.assertTrue(is_safetensors_compatible(lowerCamelCase   ,						variant=lowerCamelCase		)		)





    def __lowercase       (       self   :  Dict		)       ->  Any:
      lowerCAmelCase_      :				Optional[int]      =       [
          """safety_checker/pytorch_model.fp16.bin""",
          """safety_checker/model.fp16.safetensors""",
          """vae/diffusion_pytorch_model.fp16.bin""",
          """vae/diffusion_pytorch_model.fp16.safetensors""",
          """text_encoder/pytorch_model.fp16.bin""",
          # 'text_encoder/model.fp16.safetensors',
          """unet/diffusion_pytorch_model.fp16.bin""",
          """unet/diffusion_pytorch_model.fp16.safetensors""",
      ]
      lowerCAmelCase_      :				int      =       """fp16"""
      self.assertFalse(is_safetensors_compatible(lowerCamelCase   ,						variant=lowerCamelCase		)		)



