
import os

import pytest
import yaml

from datasets.features.features import Features, Value
from datasets.info import DatasetInfo, DatasetInfosDict







@pytest.mark.parametrize(
    """files"""  ,     [
        ["""full:README.md""", """dataset_infos.json"""],
        ["""empty:README.md""", """dataset_infos.json"""],
        ["""dataset_infos.json"""],
        ["""full:README.md"""],
    ]  ,     )
def 							A__       (    lowerCamelCase  ,     lowerCamelCase    )	->						int:
				UpperCamelCase_:				Optional[int]						=    tmp_path_factory.mktemp("""dset_infos_dir"""    )
				if "full:README.md" in files:
								with open(dataset_infos_dir / """README.md"""  ,     """w"""    ) as f:
												f.write("""---\ndataset_info:\n  dataset_size: 42\n---"""    )
				if "empty:README.md" in files:
								with open(dataset_infos_dir / """README.md"""  ,     """w"""    ) as f:
												f.write(""""""    )
    # we want to support dataset_infos.json for backward compatibility
				if "dataset_infos.json" in files:
								with open(dataset_infos_dir / """dataset_infos.json"""  ,     """w"""    ) as f:
												f.write("""{\"default\": {\"dataset_size\": 42}}"""    )
				UpperCamelCase_:				Optional[int]						=    DatasetInfosDict.from_directory(lowerCamelCase    )
				assert dataset_infos
				assert dataset_infos["default"].dataset_size == 42







@pytest.mark.parametrize(
    """dataset_info"""  ,     [
        DatasetInfo(),
        DatasetInfo(
            description="""foo"""  ,     features=Features({"""a""": Value("""int32"""    )}    )  ,     builder_name="""builder"""  ,     config_name="""config"""  ,     version="""1.0.0"""  ,     splits=[{"""name""": """train"""}]  ,     download_size=42  ,     ),
    ]  ,     )
def 							A__       (    lowerCamelCase  ,     lowerCamelCase    )	->						List[str]:
				UpperCamelCase_:				Tuple						=    str(lowerCamelCase    )
				dataset_info.write_to_directory(lowerCamelCase    )
				UpperCamelCase_:				List[Any]						=    DatasetInfo.from_directory(lowerCamelCase    )
				assert dataset_info == reloaded
				assert os.path.exists(os.path.join(lowerCamelCase  ,     """dataset_info.json"""    )    )







def 							A__       (    )	->						List[Any]:
				UpperCamelCase_:				str						=    DatasetInfo(
				    description="""foo"""  ,     citation="""bar"""  ,     homepage="""https://foo.bar"""  ,     license="""CC0"""  ,     features=Features({"""a""": Value("""int32"""    )}    )  ,     post_processed={}  ,     supervised_keys=()  ,     task_templates=[]  ,     builder_name="""builder"""  ,     config_name="""config"""  ,     version="""1.0.0"""  ,     splits=[{"""name""": """train""", """num_examples""": 42}]  ,     download_checksums={}  ,     download_size=13_37  ,     post_processing_size=4_42  ,     dataset_size=12_34  ,     size_in_bytes=13_37 + 4_42 + 12_34  ,     )
				UpperCamelCase_:				Dict						=    dataset_info._to_yaml_dict()
				assert sorted(lowerCamelCase    ) == sorted(DatasetInfo._INCLUDED_INFO_IN_YAML    )
				for key in DatasetInfo._INCLUDED_INFO_IN_YAML:
								assert key in dataset_info_yaml_dict
								assert isinstance(dataset_info_yaml_dict[key]  ,     (list, dict, int, str)    )
				UpperCamelCase_:				List[str]						=    yaml.safe_dump(lowerCamelCase    )
				UpperCamelCase_:				str						=    yaml.safe_load(lowerCamelCase    )
				assert dataset_info_yaml_dict == reloaded







def 							A__       (    )	->						Any:
				UpperCamelCase_:				Tuple						=    DatasetInfo()
				UpperCamelCase_:				int						=    dataset_info._to_yaml_dict()
				assert dataset_info_yaml_dict == {}







@pytest.mark.parametrize(
    """dataset_infos_dict"""  ,     [
        DatasetInfosDict(),
        DatasetInfosDict({"""default""": DatasetInfo()}    ),
        DatasetInfosDict({"""my_config_name""": DatasetInfo()}    ),
        DatasetInfosDict(
            {
                """default""": DatasetInfo(
                    description="""foo"""  ,     features=Features({"""a""": Value("""int32"""    )}    )  ,     builder_name="""builder"""  ,     config_name="""config"""  ,     version="""1.0.0"""  ,     splits=[{"""name""": """train"""}]  ,     download_size=42  ,     )
            }    ),
        DatasetInfosDict(
            {
                """v1""": DatasetInfo(dataset_size=42    ),
                """v2""": DatasetInfo(dataset_size=13_37    ),
            }    ),
    ]  ,     )
def 							A__       (    lowerCamelCase  ,     lowerCamelCase    )	->						List[Any]:
				UpperCamelCase_:				Optional[int]						=    str(lowerCamelCase    )
				dataset_infos_dict.write_to_directory(lowerCamelCase    )
				UpperCamelCase_:				Optional[Any]						=    DatasetInfosDict.from_directory(lowerCamelCase    )

				# the config_name of the dataset_infos_dict take over the attribute
				for config_name, dataset_info in dataset_infos_dict.items():
								UpperCamelCase_:				List[Any]						=    config_name
								# the yaml representation doesn't include fields like description or citation
								# so we just test that we can recover what we can from the yaml
								UpperCamelCase_:				int						=    DatasetInfo._from_yaml_dict(dataset_info._to_yaml_dict()    )
				assert dataset_infos_dict == reloaded

				if dataset_infos_dict:
								assert os.path.exists(os.path.join(lowerCamelCase  ,     """README.md"""    )    )








import subprocess
import sys

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

class 							_UpperCamelCase					(       _A	):






			'''simple docstring'''

			@require_torch
			def    lowerCAmelCase__						(							self			:							List[str]   ):
							# 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_:				Optional[Any]						=    """
from transformers import BertConfig, BertModel, BertTokenizer, pipeline
        """

							UpperCamelCase_:				List[Any]						=    """
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\")
        """

							UpperCamelCase_:				Dict						=    """
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
							UpperCamelCase_:				int						=    """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
							UpperCamelCase_:				Tuple						=    [sys.executable, """-c""", """\n""".join([load, run, mock]   )]

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

							UpperCamelCase_:				Any						=    """
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\")
        """

							UpperCamelCase_:				Optional[int]						=    """
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
							UpperCamelCase_:				Optional[int]						=    """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
							UpperCamelCase_:				Optional[int]						=    [sys.executable, """-c""", """\n""".join([load, run, mock]   )]

							# should succeed
							UpperCamelCase_:				List[str]						=    self.get_env()
							UpperCamelCase_:				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    lowerCAmelCase__						(							self			:							int   ):
							# 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_:				Dict						=    """
from transformers import BertConfig, BertModel, BertTokenizer
        """

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

							UpperCamelCase_:				Dict						=    """
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
							UpperCamelCase_:				Union[str, Any]						=    [sys.executable, """-c""", """\n""".join([load, run]   )]

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

							# next emulate no network
							UpperCamelCase_:				Optional[int]						=    [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_:				Union[str, Any]						=    """1"""
							UpperCamelCase_:				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    lowerCAmelCase__						(							self			:							List[str]   ):
							UpperCamelCase_:				Dict						=    """
from transformers import pipeline
        """
							UpperCamelCase_:				Union[str, Any]						=    """
mname = \"hf-internal-testing/tiny-random-bert\"
pipe = pipeline(model=mname)
        """

							UpperCamelCase_:				Dict						=    """
import socket
def offline_socket(*args, **kwargs): raise socket.error(\"Offline mode is enabled\")
socket.socket = offline_socket
        """
							UpperCamelCase_:				Optional[Any]						=    self.get_env()
							UpperCamelCase_:				int						=    """1"""
							UpperCamelCase_:				int						=    [sys.executable, """-c""", """\n""".join([load, mock, run]   )]
							UpperCamelCase_:				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    lowerCAmelCase__						(							self			:							Optional[int]   ):
							UpperCamelCase_:				str						=    """
from transformers import AutoModel
        """
							UpperCamelCase_:				Dict						=    """
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
							UpperCamelCase_:				Optional[int]						=    [sys.executable, """-c""", """\n""".join([load, run]   )]

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

							# should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
							UpperCamelCase_:				Union[str, Any]						=    """1"""
							UpperCamelCase_:				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()   )






