


"""simple docstring"""




import json
import os
from pathlib import Path

import pytest

from datasets.download.download_config import DownloadConfig
from datasets.download.download_manager import DownloadManager
from datasets.utils.file_utils import hash_url_to_filename


__magic_name__						: Optional[Any]    =				'http://www.mocksite.com/file1.txt'
__magic_name__						: Tuple    =				'"text": ["foo", "foo"]'
__magic_name__						: str    =				'6d8ce9aa78a471c7477201efbeabd3bb01ac2e7d100a6dc024ba1608361f90a8'




class 			__snake_case   :
							__a					 =       200
							__a					 =       {'''Content-Length''': '''100'''}
							__a					 =       {}




							def   __a   (     self:    List[str]     ,							**A_:    List[Any]   ):
										return [bytes(A_     ,							"""utf-8"""   )]





def       a_ (	*lowercase__ :List[Any],			**lowercase__ :Any     ):
			return MockResponse()





@pytest.mark.parametrize("""urls_type""",			[str, list, dict]     )
def       a_ (	lowercase__ :Optional[int],			lowercase__ :Any,			lowercase__ :Optional[int]     ):
			import requests

			monkeypatch.setattr(lowercase__,			"""request""",			lowercase__     )

			__lowerCamelCase          =      URL
			if issubclass(lowercase__,			lowercase__     ):
						__lowerCamelCase          =      url
			elif issubclass(lowercase__,			lowercase__     ):
						__lowerCamelCase          =      [url]
			elif issubclass(lowercase__,			lowercase__     ):
						__lowerCamelCase          =      {"""train""": url}
			__lowerCamelCase          =      """dummy"""
			__lowerCamelCase          =      """downloads"""
			__lowerCamelCase          =      tmp_path
			__lowerCamelCase          =      DownloadConfig(
			    cache_dir=os.path.join(lowercase__,			lowercase__     ),			use_etag=lowercase__,			)
			__lowerCamelCase          =      DownloadManager(dataset_name=lowercase__,			download_config=lowercase__     )
			__lowerCamelCase          =      dl_manager.download(lowercase__     )
			__lowerCamelCase          =      urls
			for downloaded_paths in [downloaded_paths]:
						if isinstance(lowercase__,			lowercase__     ):
									__lowerCamelCase          =      [downloaded_paths]
									__lowerCamelCase          =      [urls]
						elif isinstance(lowercase__,			lowercase__     ):
									assert "train" in downloaded_paths.keys()
									__lowerCamelCase          =      downloaded_paths.values()
									__lowerCamelCase          =      urls.values()
						assert downloaded_paths
						for downloaded_path, input_url in zip(lowercase__,			lowercase__     ):
									assert downloaded_path == dl_manager.downloaded_paths[input_url]
									__lowerCamelCase          =      Path(lowercase__     )
									__lowerCamelCase          =      downloaded_path.parts
									assert parts[-1] == HASH
									assert parts[-2] == cache_subdir
									assert downloaded_path.exists()
									__lowerCamelCase          =      downloaded_path.read_text()
									assert content == CONTENT
									__lowerCamelCase          =      downloaded_path.with_suffix(""".json"""     )
									assert metadata_downloaded_path.exists()
									__lowerCamelCase          =      json.loads(metadata_downloaded_path.read_text()     )
									assert metadata_content == {"url": URL, "etag": None}





@pytest.mark.parametrize("""paths_type""",			[str, list, dict]     )
def       a_ (	lowercase__ :Dict,			lowercase__ :Optional[Any],			lowercase__ :Dict     ):
			__lowerCamelCase          =      str(lowercase__     )
			if issubclass(lowercase__,			lowercase__     ):
						__lowerCamelCase          =      filename
			elif issubclass(lowercase__,			lowercase__     ):
						__lowerCamelCase          =      [filename]
			elif issubclass(lowercase__,			lowercase__     ):
						__lowerCamelCase          =      {"""train""": filename}
			__lowerCamelCase          =      """dummy"""
			__lowerCamelCase          =      xz_file.parent
			__lowerCamelCase          =      """extracted"""
			__lowerCamelCase          =      DownloadConfig(
			    cache_dir=lowercase__,			use_etag=lowercase__,			)
			__lowerCamelCase          =      DownloadManager(dataset_name=lowercase__,			download_config=lowercase__     )
			__lowerCamelCase          =      dl_manager.extract(lowercase__     )
			__lowerCamelCase          =      paths
			for extracted_paths in [extracted_paths]:
						if isinstance(lowercase__,			lowercase__     ):
									__lowerCamelCase          =      [extracted_paths]
									__lowerCamelCase          =      [paths]
						elif isinstance(lowercase__,			lowercase__     ):
									assert "train" in extracted_paths.keys()
									__lowerCamelCase          =      extracted_paths.values()
									__lowerCamelCase          =      paths.values()
						assert extracted_paths
						for extracted_path, input_path in zip(lowercase__,			lowercase__     ):
									assert extracted_path == dl_manager.extracted_paths[input_path]
									__lowerCamelCase          =      Path(lowercase__     )
									__lowerCamelCase          =      extracted_path.parts
									assert parts[-1] == hash_url_to_filename(lowercase__,			etag=lowercase__     )
									assert parts[-2] == extracted_subdir
									assert extracted_path.exists()
									__lowerCamelCase          =      extracted_path.read_text()
									__lowerCamelCase          =      text_file.read_text()
									assert extracted_file_content == expected_file_content





def       a_ (	lowercase__ :List[str],			lowercase__ :int     ):
			assert path.endswith(""".jsonl"""     )
			for num_items, line in enumerate(lowercase__,			start=1     ):
						__lowerCamelCase          =      json.loads(line.decode("""utf-8"""     )     )
						assert item.keys() == {"col_1", "col_2", "col_3"}
			assert num_items == 4





@pytest.mark.parametrize("""archive_jsonl""",			["""tar_jsonl_path""", """zip_jsonl_path"""]     )
def       a_ (	lowercase__ :Optional[int],			lowercase__ :Union[str, Any]     ):
			__lowerCamelCase          =      request.getfixturevalue(lowercase__     )
			__lowerCamelCase          =      DownloadManager()
			for num_jsonl, (path, file) in enumerate(dl_manager.iter_archive(lowercase__     ),			start=1     ):
						_test_jsonl(lowercase__,			lowercase__     )
			assert num_jsonl == 2





@pytest.mark.parametrize("""archive_nested_jsonl""",			["""tar_nested_jsonl_path""", """zip_nested_jsonl_path"""]     )
def       a_ (	lowercase__ :Optional[int],			lowercase__ :List[Any]     ):
			__lowerCamelCase          =      request.getfixturevalue(lowercase__     )
			__lowerCamelCase          =      DownloadManager()
			for num_tar, (path, file) in enumerate(dl_manager.iter_archive(lowercase__     ),			start=1     ):
						for num_jsonl, (subpath, subfile) in enumerate(dl_manager.iter_archive(lowercase__     ),			start=1     ):
									_test_jsonl(lowercase__,			lowercase__     )
			assert num_tar == 1
			assert num_jsonl == 2





def       a_ (	lowercase__ :Tuple     ):
			__lowerCamelCase          =      DownloadManager()
			for num_file, file in enumerate(dl_manager.iter_files(lowercase__     ),			start=1     ):
						assert os.path.basename(lowercase__     ) == ("test.txt" if num_file == 1 else "train.txt")
			assert num_file == 2








"""simple docstring"""




import json
import os
import tempfile
import unittest
import unittest.mock as mock
from pathlib import Path

from requests.exceptions import HTTPError

from transformers.utils import (
    CONFIG_NAME,
    FLAX_WEIGHTS_NAME,
    TF2_WEIGHTS_NAME,
    TRANSFORMERS_CACHE,
    WEIGHTS_NAME,
    cached_file,
    get_file_from_repo,
    has_file,
)


__magic_name__						: Tuple    =				'hf-internal-testing/tiny-random-bert'
__magic_name__						: Dict    =				os.path.join(TRANSFORMERS_CACHE, 'models--hf-internal-testing--tiny-random-bert')
__magic_name__						: Optional[Any]    =				'9b8c223d42b2188cb49d29af482996f9d0f3e5a6'




class 			__snake_case   (unittest.TestCase  ):




							def   __a   (     self:    List[Any]   ):
										__lowerCamelCase          =      cached_file(A_     ,							A_   )
										# Should have downloaded the file in here
										self.assertTrue(os.path.isdir(A_   )   )
										# Cache should contain at least those three subfolders:
										for subfolder in ["blobs", "refs", "snapshots"]:
													self.assertTrue(os.path.isdir(os.path.join(A_     ,							A_   )   )   )
										with open(os.path.join(A_     ,							"""refs"""     ,							"""main"""   )   ) as f:
													__lowerCamelCase          =      f.read()
										self.assertEqual(A_     ,							os.path.join(A_     ,							"""snapshots"""     ,							A_     ,							A_   )   )
										self.assertTrue(os.path.isfile(A_   )   )

										# File is cached at the same place the second time.
										__lowerCamelCase          =      cached_file(A_     ,							A_   )
										self.assertEqual(A_     ,							A_   )

										# Using a specific revision to test the full commit hash.
										__lowerCamelCase          =      cached_file(A_     ,							A_     ,							revision="""9b8c223"""   )
										self.assertEqual(A_     ,							os.path.join(A_     ,							"""snapshots"""     ,							A_     ,							A_   )   )




							def   __a   (     self:    int   ):
										with self.assertRaisesRegex(A_     ,							"""is not a valid model identifier"""   ):
													__lowerCamelCase          =      cached_file("""tiny-random-bert"""     ,							A_   )

										with self.assertRaisesRegex(A_     ,							"""is not a valid git identifier"""   ):
													__lowerCamelCase          =      cached_file(A_     ,							A_     ,							revision="""aaaa"""   )

										with self.assertRaisesRegex(A_     ,							"""does not appear to have a file named"""   ):
													__lowerCamelCase          =      cached_file(A_     ,							"""conf"""   )




							def   __a   (     self:    Optional[int]   ):
										with self.assertRaisesRegex(A_     ,							"""does not appear to have a file named"""   ):
													__lowerCamelCase          =      cached_file(A_     ,							"""conf"""   )

										with open(os.path.join(A_     ,							"""refs"""     ,							"""main"""   )   ) as f:
													__lowerCamelCase          =      f.read()
										self.assertTrue(os.path.isfile(os.path.join(A_     ,							""".no_exist"""     ,							A_     ,							"""conf"""   )   )   )

										__lowerCamelCase          =      cached_file(A_     ,							"""conf"""     ,							_raise_exceptions_for_missing_entries=A_   )
										self.assertIsNone(A_   )

										__lowerCamelCase          =      cached_file(A_     ,							"""conf"""     ,							local_files_only=A_     ,							_raise_exceptions_for_missing_entries=A_   )
										self.assertIsNone(A_   )

										__lowerCamelCase          =      mock.Mock()
										__lowerCamelCase          =      5_00
										__lowerCamelCase          =      {}
										__lowerCamelCase          =      HTTPError
										__lowerCamelCase          =      {}

										# Under the mock environment we get a 500 error when trying to reach the tokenizer.
										with mock.patch("""requests.Session.request"""     ,							return_value=A_   ) as mock_head:
													__lowerCamelCase          =      cached_file(A_     ,							"""conf"""     ,							_raise_exceptions_for_connection_errors=A_   )
													self.assertIsNone(A_   )
													# This check we did call the fake head request
													mock_head.assert_called()




							def   __a   (     self:    str   ):
										self.assertTrue(has_file("""hf-internal-testing/tiny-bert-pt-only"""     ,							A_   )   )
										self.assertFalse(has_file("""hf-internal-testing/tiny-bert-pt-only"""     ,							A_   )   )
										self.assertFalse(has_file("""hf-internal-testing/tiny-bert-pt-only"""     ,							A_   )   )




							def   __a   (     self:    str   ):
										# `get_file_from_repo` returns None if the file does not exist
										self.assertIsNone(get_file_from_repo("""bert-base-cased"""     ,							"""ahah.txt"""   )   )

										# The function raises if the repository does not exist.
										with self.assertRaisesRegex(A_     ,							"""is not a valid model identifier"""   ):
													get_file_from_repo("""bert-base-case"""     ,							A_   )

										# The function raises if the revision does not exist.
										with self.assertRaisesRegex(A_     ,							"""is not a valid git identifier"""   ):
													get_file_from_repo("""bert-base-cased"""     ,							A_     ,							revision="""ahaha"""   )

										__lowerCamelCase          =      get_file_from_repo("""bert-base-cased"""     ,							A_   )
										# The name is the cached name which is not very easy to test, so instead we load the content.
										__lowerCamelCase          =      json.loads(open(A_     ,							"""r"""   ).read()   )
										self.assertEqual(config["""hidden_size"""]     ,							7_68   )




							def   __a   (     self:    Union[str, Any]   ):
										with tempfile.TemporaryDirectory() as tmp_dir:
													__lowerCamelCase          =      Path(A_   ) / """a.txt"""
													filename.touch()
													self.assertEqual(get_file_from_repo(A_     ,							"""a.txt"""   )     ,							str(A_   )   )

													self.assertIsNone(get_file_from_repo(A_     ,							"""b.txt"""   )   )




