
'''simple docstring'''



import os
import zipfile

import requests
from get_ci_error_statistics import download_artifact, get_artifacts_links
def  lowerCAmelCase_						(					SCREAMING_SNAKE_CASE_     ,   SCREAMING_SNAKE_CASE_=7    )  ->  List[Any]:




   """simple docstring"""
   _SCREAMING_SNAKE_CASE  			=					None
   if token is not None:
      _SCREAMING_SNAKE_CASE  			=					{"""Accept""": """application/vnd.github+json""", """Authorization""": F"Bearer {token}"}

   # The id of a workflow (not of a workflow run)
   _SCREAMING_SNAKE_CASE  			=					"""636036"""

   _SCREAMING_SNAKE_CASE  			=					F"https://api.github.com/repos/huggingface/transformers/actions/workflows/{workflow_id}/runs"
   # On `main` branch + event being `schedule` + not returning PRs + only `num_runs` results
   url += F"?branch=main&event=schedule&exclude_pull_requests=true&per_page={num_runs}"

   _SCREAMING_SNAKE_CASE  			=					requests.get(SCREAMING_SNAKE_CASE_     ,   headers=SCREAMING_SNAKE_CASE_    ).json()

   return result["workflow_runs"]
def  lowerCAmelCase_						(					SCREAMING_SNAKE_CASE_    )  ->  List[str]:




   """simple docstring"""
   _SCREAMING_SNAKE_CASE  			=					get_daily_ci_runs(SCREAMING_SNAKE_CASE_    )
   _SCREAMING_SNAKE_CASE  			=					None
   for workflow_run in workflow_runs:
      if workflow_run["status"] == "completed":
         _SCREAMING_SNAKE_CASE  			=					workflow_run["""id"""]
         break

   return workflow_run_id
def  lowerCAmelCase_						(					SCREAMING_SNAKE_CASE_     ,   SCREAMING_SNAKE_CASE_     ,   SCREAMING_SNAKE_CASE_    )  ->  List[Any]:




   """simple docstring"""
   _SCREAMING_SNAKE_CASE  			=					get_last_daily_ci_runs(SCREAMING_SNAKE_CASE_    )
   if workflow_run_id is not None:
      _SCREAMING_SNAKE_CASE  			=					get_artifacts_links(worflow_run_id=SCREAMING_SNAKE_CASE_     ,   token=SCREAMING_SNAKE_CASE_    )
      for artifact_name in artifact_names:
         if artifact_name in artifacts_links:
            _SCREAMING_SNAKE_CASE  			=					artifacts_links[artifact_name]
            download_artifact(
                artifact_name=SCREAMING_SNAKE_CASE_     ,   artifact_url=SCREAMING_SNAKE_CASE_     ,   output_dir=SCREAMING_SNAKE_CASE_     ,   token=SCREAMING_SNAKE_CASE_    )







def  lowerCAmelCase_						(					SCREAMING_SNAKE_CASE_     ,   SCREAMING_SNAKE_CASE_     ,   SCREAMING_SNAKE_CASE_    )  ->  int:




   """simple docstring"""
   get_last_daily_ci_artifacts(SCREAMING_SNAKE_CASE_     ,   SCREAMING_SNAKE_CASE_     ,   SCREAMING_SNAKE_CASE_    )

   _SCREAMING_SNAKE_CASE  			=					{}
   for artifact_name in artifact_names:
      _SCREAMING_SNAKE_CASE  			=					os.path.join(SCREAMING_SNAKE_CASE_     ,   F"{artifact_name}.zip"    )
      if os.path.isfile(SCREAMING_SNAKE_CASE_    ):
         _SCREAMING_SNAKE_CASE  			=					{}
         with zipfile.ZipFile(SCREAMING_SNAKE_CASE_    ) as z:
            for filename in z.namelist():
               if not os.path.isdir(SCREAMING_SNAKE_CASE_    ):
                  # read the file
                  with z.open(SCREAMING_SNAKE_CASE_    ) as f:
                     _SCREAMING_SNAKE_CASE  			=					f.read().decode("""UTF-8"""    )

   return results









'''simple docstring'''




import subprocess
import sys

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



class 							lowercase_						(		a			):

  '''simple docstring'''





  @require_torch
  def     snake_case_      (  self  )				->					Optional[Any]:


    """simple docstring"""
    # 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
    UpperCAmelCase    				=						'\nfrom transformers import BertConfig, BertModel, BertTokenizer, pipeline\n        '

    UpperCAmelCase    				=						'\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        '

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

    # should succeed
    UpperCAmelCase    				=						self.get_env()
    # should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
    UpperCAmelCase    				=						'1'
    UpperCAmelCase    				=						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     snake_case_      (  self  )				->					Union[str, Any]:


    """simple docstring"""
    # python one-liner segments
    # this must be loaded before socket.socket is monkey-patched
    UpperCAmelCase    				=						'\nfrom transformers import BertConfig, BertModel, BertTokenizer, pipeline\n        '

    UpperCAmelCase    				=						'\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        '

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

    # should succeed
    UpperCAmelCase    				=						self.get_env()
    UpperCAmelCase    				=						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     snake_case_      (  self  )				->					List[str]:


    """simple docstring"""
    # 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
    UpperCAmelCase    				=						'\nfrom transformers import BertConfig, BertModel, BertTokenizer\n        '

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

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

    # should succeed
    UpperCAmelCase    				=						self.get_env()
    UpperCAmelCase    				=						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
    UpperCAmelCase    				=						[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
    UpperCAmelCase    				=						'1'
    UpperCAmelCase    				=						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     snake_case_      (  self  )				->					Optional[Any]:


    """simple docstring"""
    UpperCAmelCase    				=						'\nfrom transformers import pipeline\n        '
    UpperCAmelCase    				=						'\nmname = "hf-internal-testing/tiny-random-bert"\npipe = pipeline(model=mname)\n        '

    UpperCAmelCase    				=						'\nimport socket\ndef offline_socket(*args, **kwargs): raise socket.error("Offline mode is enabled")\nsocket.socket = offline_socket\n        '
    UpperCAmelCase    				=						self.get_env()
    UpperCAmelCase    				=						'1'
    UpperCAmelCase    				=						[sys.executable, '-c', '\n'.join([load, mock, run]  )]
    UpperCAmelCase    				=						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     snake_case_      (  self  )				->					str:


    """simple docstring"""
    UpperCAmelCase    				=						'\nfrom transformers import AutoModel\n        '
    UpperCAmelCase    				=						'\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
    UpperCAmelCase    				=						[sys.executable, '-c', '\n'.join([load, run]  )]

    # should succeed
    UpperCAmelCase    				=						self.get_env()
    UpperCAmelCase    				=						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
    UpperCAmelCase    				=						'1'
    UpperCAmelCase    				=						subprocess.run(a_ ,			env=a_ ,			check=a_ ,			capture_output=a_  )
    self.assertEqual(result.returncode ,			0 ,			result.stderr  )
    self.assertIn('success' ,			result.stdout.decode()  )


