


import unittest
from dataclasses import dataclass

import pytest

from accelerate.commands.config.config_args import SageMakerConfig
from accelerate.utils import ComputeEnvironment
from accelerate.utils.launch import _convert_nargs_to_dict






@dataclass
class _lowerCamelCase(		_a     ):
       lowercase_				:			Tuple									=    ComputeEnvironment.AMAZON_SAGEMAKER
       lowercase_				:			str									=    True
       lowercase_				:			Any									=    """ml.p3.2xlarge"""
       lowercase_				:			str									=    """accelerate_sagemaker_execution_role"""
       lowercase_				:			Optional[int]									=    """hf-sm"""
       lowercase_				:			List[str]									=    """us-east-1"""
       lowercase_				:			int									=    1
       lowercase_				:			str									=    """accelerate-sagemaker-1"""
       lowercase_				:			Any									=    """1.6"""
       lowercase_				:			List[str]									=    """4.4"""
       lowercase_				:			str									=    """train.py"""
       lowercase_				:			Union[str, Any]									=    [
           """--model_name_or_path""",
           """bert""",
           """--do_train""",
           """False""",
           """--epochs""",
           """3""",
           """--learning_rate""",
           """5e-5""",
           """--max_steps""",
           """50.5""",
       ]
       lowercase_				:			Tuple									=    [
           """--model_name_or_path""",
           """bert""",
           """--do_train""",
           """--do_test""",
           """False""",
           """--do_predict""",
           """--epochs""",
           """3""",
           """--learning_rate""",
           """5e-5""",
           """--max_steps""",
           """50.5""",
       ]






class _lowerCamelCase(		unittest.TestCase     ):
       def     UpperCamelCase  (      self) ->     Optional[Any]:




           """simple docstring"""

           _lowercase :							Dict   	=	_convert_nargs_to_dict(MockLaunchConfig.success_training_script_args)
           assert isinstance(converted_args['model_name_or_path'],   lowerCamelCase)
           assert isinstance(converted_args['do_train'],   lowerCamelCase)
           assert isinstance(converted_args['epochs'],   lowerCamelCase)
           assert isinstance(converted_args['learning_rate'],   lowerCamelCase)
           assert isinstance(converted_args['max_steps'],   lowerCamelCase)

           with pytest.raises(lowerCamelCase):
               _convert_nargs_to_dict(MockLaunchConfig.fail_training_script_args)






import subprocess
import sys

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






class _lowerCamelCase(		_a     ):
       @require_torch
       def     UpperCamelCase  (      self) ->     int:




           """simple docstring"""

           _lowercase :							Optional[Any]   	=	'\nfrom transformers import BertConfig, BertModel, BertTokenizer, pipeline\n        '

           _lowercase :							Union[str, 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        '

           _lowercase :							List[str]   	=	'\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
           _lowercase :							Union[str, 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
           _lowercase :							Any   	=	[sys.executable, '-c', '\n'.join([load, run, mock])]

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




           """simple docstring"""

           _lowercase :							int   	=	'\nfrom transformers import BertConfig, BertModel, BertTokenizer, pipeline\n        '

           _lowercase :							Union[str, 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        '

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

           # should succeed
           _lowercase :							Any   	=	self.get_env()
           _lowercase :							str   	=	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     UpperCamelCase  (      self) ->     Union[str, Any]:




           """simple docstring"""

           _lowercase :							Tuple   	=	'\nfrom transformers import BertConfig, BertModel, BertTokenizer\n        '

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

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

           # should succeed
           _lowercase :							List[str]   	=	self.get_env()
           _lowercase :							Dict   	=	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
           _lowercase :							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
           _lowercase :							Optional[Any]   	=	'1'
           _lowercase :							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())
       @require_torch
       def     UpperCamelCase  (      self) ->     Optional[Any]:




           """simple docstring"""

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

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




           """simple docstring"""

           _lowercase :							Dict   	=	'\nfrom transformers import AutoModel\n        '
           _lowercase :							int   	=	'\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
           _lowercase :							Optional[int]   	=	[sys.executable, '-c', '\n'.join([load, run])]

           # should succeed
           _lowercase :							int   	=	self.get_env()
           _lowercase :							List[str]   	=	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
           _lowercase :							Tuple   	=	'1'
           _lowercase :							Dict   	=	subprocess.run(lowerCamelCase,   env=lowerCamelCase,   check=lowerCamelCase,   capture_output=lowerCamelCase)
           self.assertEqual(result.returncode,   0,   result.stderr)
           self.assertIn('success',   result.stdout.decode())


