






import json
import os
import shutil
import tempfile
import unittest
from multiprocessing import get_context
from pathlib import Path

import datasets
import numpy as np
from datasets import load_dataset
from parameterized import parameterized

from transformers import AutoProcessor
from transformers.models.wavaveca import WavaVecaCTCTokenizer, WavaVecaFeatureExtractor
from transformers.models.wavaveca.tokenization_wavaveca import VOCAB_FILES_NAMES
from transformers.testing_utils import require_pyctcdecode, require_torch, require_torchaudio, slow
from transformers.utils import FEATURE_EXTRACTOR_NAME, is_pyctcdecode_available, is_torch_available

from ..wavaveca.test_feature_extraction_wavaveca import floats_list


if is_pyctcdecode_available():
				from huggingface_hub import snapshot_download
				from pyctcdecode import BeamSearchDecoderCTC

				from transformers.models.wavaveca_with_lm import WavaVecaProcessorWithLM
				from transformers.models.wavaveca_with_lm.processing_wavaveca_with_lm import WavaVecaDecoderWithLMOutput

if is_torch_available():
				from transformers import WavaVecaForCTC





@require_pyctcdecode
class 						SCREAMING_SNAKE_CASE				( unittest.TestCase					):





							"""simple docstring"""

							def       __A (  self:   Tuple   )    ->				Union[str, Any]:
												_A								= '''| <pad> <unk> <s> </s> a b c d e f g h i j k'''.split()
												_A								= dict(zip(__A		,     range(len(__A   )   )   )   )

												_A								= {
												    '''unk_token''': '''<unk>''',
												    '''bos_token''': '''<s>''',
												    '''eos_token''': '''</s>''',
												}
												_A								= {
												    '''feature_size''': 1,
												    '''padding_value''': 0.0,
												    '''sampling_rate''': 1_60_00,
												    '''return_attention_mask''': False,
												    '''do_normalize''': True,
												}

												_A								= tempfile.mkdtemp()
												_A								= os.path.join(self.tmpdirname		,     VOCAB_FILES_NAMES['''vocab_file''']   )
												_A								= os.path.join(self.tmpdirname		,     __A   )
												with open(self.vocab_file		,     '''w'''		,     encoding='''utf-8'''   ) as fp:
																	fp.write(json.dumps(__A   ) + '''\n'''   )

												with open(self.feature_extraction_file		,     '''w'''		,     encoding='''utf-8'''   ) as fp:
																	fp.write(json.dumps(__A   ) + '''\n'''   )

												# load decoder from hub
												_A								= '''hf-internal-testing/ngram-beam-search-decoder'''

							def       __A (  self:   Any		,     **__A:   str   )    ->				Union[str, Any]:
												_A								= self.add_kwargs_tokens_map.copy()
												kwargs.update(__A   )
												return WavaVecaCTCTokenizer.from_pretrained(self.tmpdirname		,     **__A   )

							def       __A (  self:   List[str]		,     **__A:   Optional[Any]   )    ->				Dict:
												return WavaVecaFeatureExtractor.from_pretrained(self.tmpdirname		,     **__A   )

							def       __A (  self:   Optional[Any]		,     **__A:   str   )    ->				Optional[Any]:
												return BeamSearchDecoderCTC.load_from_hf_hub(self.decoder_name		,     **__A   )

							def       __A (  self:   Tuple   )    ->				Dict:
												shutil.rmtree(self.tmpdirname   )

							def       __A (  self:   Dict   )    ->				Optional[int]:
												_A								= self.get_tokenizer()
												_A								= self.get_feature_extractor()
												_A								= self.get_decoder()

												_A								= WavaVecaProcessorWithLM(tokenizer=__A		,     feature_extractor=__A		,     decoder=__A   )

												processor.save_pretrained(self.tmpdirname   )
												_A								= WavaVecaProcessorWithLM.from_pretrained(self.tmpdirname   )

												# tokenizer
												self.assertEqual(processor.tokenizer.get_vocab()		,     tokenizer.get_vocab()   )
												self.assertIsInstance(processor.tokenizer		,     __A   )

												# feature extractor
												self.assertEqual(processor.feature_extractor.to_json_string()		,     feature_extractor.to_json_string()   )
												self.assertIsInstance(processor.feature_extractor		,     __A   )

												# decoder
												self.assertEqual(processor.decoder._alphabet.labels		,     decoder._alphabet.labels   )
												self.assertEqual(
												    processor.decoder.model_container[decoder._model_key]._unigram_set		,     decoder.model_container[decoder._model_key]._unigram_set		,     )
												self.assertIsInstance(processor.decoder		,     __A   )

							def       __A (  self:   Any   )    ->				List[str]:
												_A								= WavaVecaProcessorWithLM(
												    tokenizer=self.get_tokenizer()		,     feature_extractor=self.get_feature_extractor()		,     decoder=self.get_decoder()   )
												processor.save_pretrained(self.tmpdirname   )

												# make sure that error is thrown when decoder alphabet doesn't match
												_A								= WavaVecaProcessorWithLM.from_pretrained(
												    self.tmpdirname		,     alpha=5.0		,     beta=3.0		,     score_boundary=-7.0		,     unk_score_offset=3   )

												# decoder
												self.assertEqual(processor.language_model.alpha		,     5.0   )
												self.assertEqual(processor.language_model.beta		,     3.0   )
												self.assertEqual(processor.language_model.score_boundary		,     -7.0   )
												self.assertEqual(processor.language_model.unk_score_offset		,     3   )

							def       __A (  self:   str   )    ->				int:
												_A								= self.get_tokenizer()
												# add token to trigger raise
												tokenizer.add_tokens(['''xx''']   )
												with self.assertRaisesRegex(__A		,     '''include'''   ):
																	WavaVecaProcessorWithLM(
																	    tokenizer=__A		,     feature_extractor=self.get_feature_extractor()		,     decoder=self.get_decoder()   )

							def       __A (  self:   int   )    ->				Dict:
												_A								= self.get_feature_extractor()
												_A								= self.get_tokenizer()
												_A								= self.get_decoder()

												_A								= WavaVecaProcessorWithLM(tokenizer=__A		,     feature_extractor=__A		,     decoder=__A   )

												_A								= floats_list((3, 10_00)   )

												_A								= feature_extractor(__A		,     return_tensors='''np'''   )
												_A								= processor(__A		,     return_tensors='''np'''   )

												for key in input_feat_extract.keys():
																	self.assertAlmostEqual(input_feat_extract[key].sum()		,     input_processor[key].sum()		,     delta=1e-2   )

							def       __A (  self:   Optional[int]   )    ->				Optional[int]:
												_A								= self.get_feature_extractor()
												_A								= self.get_tokenizer()
												_A								= self.get_decoder()

												_A								= WavaVecaProcessorWithLM(tokenizer=__A		,     feature_extractor=__A		,     decoder=__A   )

												_A								= '''This is a test string'''

												_A								= processor(text=__A   )

												_A								= tokenizer(__A   )

												for key in encoded_tok.keys():
																	self.assertListEqual(encoded_tok[key]		,     encoded_processor[key]   )

							def       __A (  self:   List[Any]		,     __A:   int=(2, 10, 16)		,     __A:   List[Any]=77   )    ->				Dict:
												np.random.seed(__A   )
												return np.random.rand(*__A   )

							def       __A (  self:   List[Any]   )    ->				int:
												_A								= self.get_feature_extractor()
												_A								= self.get_tokenizer()
												_A								= self.get_decoder()

												_A								= WavaVecaProcessorWithLM(tokenizer=__A		,     feature_extractor=__A		,     decoder=__A   )

												_A								= self._get_dummy_logits(shape=(10, 16)		,     seed=13   )

												_A								= processor.decode(__A   )

												_A								= decoder.decode_beams(__A   )[0]

												self.assertEqual(decoded_decoder[0]		,     decoded_processor.text   )
												self.assertEqual('''</s> <s> </s>'''		,     decoded_processor.text   )
												self.assertEqual(decoded_decoder[-2]		,     decoded_processor.logit_score   )
												self.assertEqual(decoded_decoder[-1]		,     decoded_processor.lm_score   )

							@parameterized.expand([[None], ['''fork'''], ['''spawn''']]   )
							def       __A (  self:   List[Any]		,     __A:   Any   )    ->				Any:
												_A								= self.get_feature_extractor()
												_A								= self.get_tokenizer()
												_A								= self.get_decoder()

												_A								= WavaVecaProcessorWithLM(tokenizer=__A		,     feature_extractor=__A		,     decoder=__A   )

												_A								= self._get_dummy_logits()

												# note: pool should be instantiated *after* Wav2Vec2ProcessorWithLM.
												#       otherwise, the LM won't be available to the pool's sub-processes.
												# manual logic used to allow parameterized test for both pool=None and pool=Pool(...)
												if pool_context is None:
																	_A								= processor.batch_decode(__A   )
												else:
																	with get_context(__A   ).Pool() as pool:
																						_A								= processor.batch_decode(__A		,     __A   )

												_A								= list(__A   )

												with get_context('''fork'''   ).Pool() as p:
																	_A								= decoder.decode_beams_batch(__A		,     __A   )

												_A							,_A							,_A								= [], [], []
												for beams in decoded_beams:
																	texts_decoder.append(beams[0][0]   )
																	logit_scores_decoder.append(beams[0][-2]   )
																	lm_scores_decoder.append(beams[0][-1]   )

												self.assertListEqual(__A		,     decoded_processor.text   )
												self.assertListEqual(['''<s> <s> </s>''', '''<s> <s> <s>''']		,     decoded_processor.text   )
												self.assertListEqual(__A		,     decoded_processor.logit_score   )
												self.assertListEqual(__A		,     decoded_processor.lm_score   )

							def       __A (  self:   Tuple   )    ->				Union[str, Any]:
												_A								= self.get_feature_extractor()
												_A								= self.get_tokenizer()
												_A								= self.get_decoder()

												_A								= WavaVecaProcessorWithLM(tokenizer=__A		,     feature_extractor=__A		,     decoder=__A   )

												_A								= self._get_dummy_logits()

												_A								= 15
												_A								= -20.0
												_A								= -4.0

												_A								= processor.batch_decode(
												    __A		,     beam_width=__A		,     beam_prune_logp=__A		,     token_min_logp=__A		,     )
												_A								= decoded_processor_out.text

												_A								= list(__A   )

												with get_context('''fork'''   ).Pool() as pool:
																	_A								= decoder.decode_beams_batch(
																	    __A		,     __A		,     beam_width=__A		,     beam_prune_logp=__A		,     token_min_logp=__A		,     )

												_A								= [d[0][0] for d in decoded_decoder_out]
												_A								= [d[0][2] for d in decoded_decoder_out]
												_A								= [d[0][3] for d in decoded_decoder_out]

												self.assertListEqual(__A		,     __A   )
												self.assertListEqual(['''</s> <s> <s>''', '''<s> <s> <s>''']		,     __A   )

												self.assertTrue(np.array_equal(__A		,     decoded_processor_out.logit_score   )   )
												self.assertTrue(np.allclose([-20.054, -18.447]		,     __A		,     atol=1e-3   )   )

												self.assertTrue(np.array_equal(__A		,     decoded_processor_out.lm_score   )   )
												self.assertTrue(np.allclose([-15.554, -13.9_474]		,     __A		,     atol=1e-3   )   )

							def       __A (  self:   Dict   )    ->				Any:
												_A								= self.get_feature_extractor()
												_A								= self.get_tokenizer()
												_A								= self.get_decoder()

												_A								= WavaVecaProcessorWithLM(tokenizer=__A		,     feature_extractor=__A		,     decoder=__A   )

												_A								= self._get_dummy_logits()

												_A								= 2.0
												_A								= 5.0
												_A								= -20.0
												_A								= True

												_A								= processor.batch_decode(
												    __A		,     alpha=__A		,     beta=__A		,     unk_score_offset=__A		,     lm_score_boundary=__A		,     )
												_A								= decoded_processor_out.text

												_A								= list(__A   )
												decoder.reset_params(
												    alpha=__A		,     beta=__A		,     unk_score_offset=__A		,     lm_score_boundary=__A		,     )

												with get_context('''fork'''   ).Pool() as pool:
																	_A								= decoder.decode_beams_batch(
																	    __A		,     __A		,     )

												_A								= [d[0][0] for d in decoded_decoder_out]

												self.assertListEqual(__A		,     __A   )
												self.assertListEqual(['''<s> </s> <s> </s> </s>''', '''</s> </s> <s> </s> </s>''']		,     __A   )
												_A								= processor.decoder.model_container[processor.decoder._model_key]
												self.assertEqual(lm_model.alpha		,     2.0   )
												self.assertEqual(lm_model.beta		,     5.0   )
												self.assertEqual(lm_model.unk_score_offset		,     -20.0   )
												self.assertEqual(lm_model.score_boundary		,     __A   )

							def       __A (  self:   Union[str, Any]   )    ->				int:
												_A								= WavaVecaProcessorWithLM.from_pretrained('''hf-internal-testing/processor_with_lm'''   )

												_A								= processor.decoder.model_container[processor.decoder._model_key]
												_A								= Path(language_model._kenlm_model.path.decode('''utf-8'''   )   ).parent.parent.absolute()

												_A								= os.listdir(__A   )
												_A								= ['''alphabet.json''', '''language_model''']

												downloaded_decoder_files.sort()
												expected_decoder_files.sort()

												# test that only decoder relevant files from
												# https://huggingface.co/hf-internal-testing/processor_with_lm/tree/main
												# are downloaded and none of the rest (e.g. README.md, ...)
												self.assertListEqual(__A		,     __A   )

							def       __A (  self:   Optional[int]   )    ->				Dict:
												_A								= snapshot_download('''hf-internal-testing/processor_with_lm'''   )

												_A								= WavaVecaProcessorWithLM.from_pretrained(__A   )

												_A								= processor.decoder.model_container[processor.decoder._model_key]
												_A								= Path(language_model._kenlm_model.path.decode('''utf-8'''   )   ).parent.parent.absolute()

												_A								= os.listdir(__A   )
												_A								= os.listdir(__A   )

												local_decoder_files.sort()
												expected_decoder_files.sort()

												# test that both decoder form hub and local files in cache are the same
												self.assertListEqual(__A		,     __A   )

							def       __A (  self:   Tuple   )    ->				Optional[Any]:
												_A								= WavaVecaProcessorWithLM.from_pretrained('''hf-internal-testing/processor_with_lm'''   )
												_A								= AutoProcessor.from_pretrained('''hf-internal-testing/processor_with_lm'''   )

												_A								= floats_list((3, 10_00)   )

												_A								= processor_wavaveca(__A		,     return_tensors='''np'''   )
												_A								= processor_auto(__A		,     return_tensors='''np'''   )

												for key in input_wavaveca.keys():
																	self.assertAlmostEqual(input_wavaveca[key].sum()		,     input_auto[key].sum()		,     delta=1e-2   )

												_A								= self._get_dummy_logits()

												_A								= processor_wavaveca.batch_decode(__A   )
												_A								= processor_auto.batch_decode(__A   )

												self.assertListEqual(decoded_wavaveca.text		,     decoded_auto.text   )

							def       __A (  self:   Optional[Any]   )    ->				int:
												_A								= self.get_feature_extractor()
												_A								= self.get_tokenizer()
												_A								= self.get_decoder()

												_A								= WavaVecaProcessorWithLM(tokenizer=__A		,     feature_extractor=__A		,     decoder=__A   )

												self.assertListEqual(
												    processor.model_input_names		,     feature_extractor.model_input_names		,     msg='''`processor` and `feature_extractor` model input names do not match'''		,     )

							@staticmethod
							def       __A (  __A:   Any		,     __A:   Tuple   )    ->				int:
												_A								= [d[key] for d in offsets]
												return retrieved_list

							def       __A (  self:   Dict   )    ->				List[str]:
												_A								= WavaVecaProcessorWithLM.from_pretrained('''hf-internal-testing/processor_with_lm'''   )
												_A								= self._get_dummy_logits()[0]

												_A								= processor.decode(__A		,     output_word_offsets=__A   )
												# check Wav2Vec2CTCTokenizerOutput keys for word
												self.assertEqual(len(outputs.keys()   )		,     4   )
												self.assertTrue('''text''' in outputs   )
												self.assertTrue('''word_offsets''' in outputs   )
												self.assertTrue(isinstance(__A		,     __A   )   )

												self.assertEqual(''' '''.join(self.get_from_offsets(outputs['''word_offsets''']		,     '''word'''   )   )		,     outputs.text   )
												self.assertListEqual(self.get_from_offsets(outputs['''word_offsets''']		,     '''word'''   )		,     ['''<s>''', '''<s>''', '''</s>''']   )
												self.assertListEqual(self.get_from_offsets(outputs['''word_offsets''']		,     '''start_offset'''   )		,     [0, 2, 4]   )
												self.assertListEqual(self.get_from_offsets(outputs['''word_offsets''']		,     '''end_offset'''   )		,     [1, 3, 5]   )

							def       __A (  self:   List[str]   )    ->				List[str]:
												_A								= WavaVecaProcessorWithLM.from_pretrained('''hf-internal-testing/processor_with_lm'''   )
												_A								= self._get_dummy_logits()

												_A								= processor.batch_decode(__A		,     output_word_offsets=__A   )

												# check Wav2Vec2CTCTokenizerOutput keys for word
												self.assertEqual(len(outputs.keys()   )		,     4   )
												self.assertTrue('''text''' in outputs   )
												self.assertTrue('''word_offsets''' in outputs   )
												self.assertTrue(isinstance(__A		,     __A   )   )

												self.assertListEqual(
												    [''' '''.join(self.get_from_offsets(__A		,     '''word'''   )   ) for o in outputs['''word_offsets''']]		,     outputs.text   )
												self.assertListEqual(self.get_from_offsets(outputs['''word_offsets'''][0]		,     '''word'''   )		,     ['''<s>''', '''<s>''', '''</s>''']   )
												self.assertListEqual(self.get_from_offsets(outputs['''word_offsets'''][0]		,     '''start_offset'''   )		,     [0, 2, 4]   )
												self.assertListEqual(self.get_from_offsets(outputs['''word_offsets'''][0]		,     '''end_offset'''   )		,     [1, 3, 5]   )



							@slow
							@require_torch
							@require_torchaudio
							def       __A (  self:   List[str]   )    ->				Optional[int]:
												import torch

												_A								= load_dataset('''common_voice'''		,     '''en'''		,     split='''train'''		,     streaming=__A   )
												_A								= ds.cast_column('''audio'''		,     datasets.Audio(sampling_rate=1_60_00   )   )
												_A								= iter(__A   )
												_A								= next(__A   )

												_A								= AutoProcessor.from_pretrained('''patrickvonplaten/wav2vec2-base-100h-with-lm'''   )
												_A								= WavaVecaForCTC.from_pretrained('''patrickvonplaten/wav2vec2-base-100h-with-lm'''   )

												# compare to filename `common_voice_en_100038.mp3` of dataset viewer on https://huggingface.co/datasets/common_voice/viewer/en/train
												_A								= processor(sample['''audio''']['''array''']		,     return_tensors='''pt'''   ).input_values

												with torch.no_grad():
																	_A								= model(__A   ).logits.cpu().numpy()

												_A								= processor.decode(logits[0]		,     output_word_offsets=__A   )

												_A								= model.config.inputs_to_logits_ratio / processor.feature_extractor.sampling_rate
												_A								= [
												    {
												        '''start_time''': d['''start_offset'''] * time_offset,
												        '''end_time''': d['''end_offset'''] * time_offset,
												        '''word''': d['''word'''],
												    }
												    for d in output['''word_offsets''']
												]

												_A								= '''WHY DOES MILISANDRA LOOK LIKE SHE WANTS TO CONSUME JOHN SNOW ON THE RIVER AT THE WALL'''

												# output words
												self.assertEqual(''' '''.join(self.get_from_offsets(__A		,     '''word'''   )   )		,     __A   )
												self.assertEqual(''' '''.join(self.get_from_offsets(__A		,     '''word'''   )   )		,     output.text   )

												# output times
												_A								= torch.tensor(self.get_from_offsets(__A		,     '''start_time'''   )   )
												_A								= torch.tensor(self.get_from_offsets(__A		,     '''end_time'''   )   )

												# fmt: off
												_A								= torch.tensor([1.4_199, 1.6_599, 2.2_599, 3.0, 3.24, 3.5_999, 3.7_999, 4.0_999, 4.26, 4.94, 5.28, 5.6_599, 5.78, 5.94, 6.32, 6.5_399, 6.6_599]   )
												_A								= torch.tensor([1.5_399, 1.8_999, 2.9, 3.16, 3.5_399, 3.72, 4.0_199, 4.1_799, 4.76, 5.1_599, 5.5_599, 5.6_999, 5.86, 6.1_999, 6.38, 6.6_199, 6.94]   )
												# fmt: on

												self.assertTrue(torch.allclose(__A		,     __A		,     atol=0.01   )   )
												self.assertTrue(torch.allclose(__A		,     __A		,     atol=0.01   )   )











import os
from pathlib import Path
from unittest.mock import patch

import pytest
import zstandard as zstd

from datasets.download.download_config import DownloadConfig
from datasets.utils.file_utils import (
    OfflineModeIsEnabled,
    cached_path,
    fsspec_get,
    fsspec_head,
    ftp_get,
    ftp_head,
    get_from_cache,
    http_get,
    http_head,
)


__A      =				'\\n    Text data.\n    Second line of data.'

__A      =				'file'




@pytest.fixture(scope='''session''' )
def      __A					(     _lowercase ):

					'''simple docstring'''







					_A								= tmp_path_factory.mktemp('''data''' ) / (FILE_PATH + '''.zstd''')
					_A								= bytes(_lowercase					,			'''utf-8''' )
					with zstd.open(_lowercase					,			'''wb''' ) as f:
										f.write(_lowercase )
					return path




@pytest.fixture
def      __A					(     _lowercase ):

					'''simple docstring'''







					with open(os.path.join(tmpfs.local_root_dir					,			_lowercase )					,			'''w''' ) as f:
										f.write(_lowercase )
					return FILE_PATH




@pytest.mark.parametrize('''compression_format'''					,			['''gzip''', '''xz''', '''zstd'''] )
def      __A					(     _lowercase					,			_lowercase					,			_lowercase					,			_lowercase					,			_lowercase					,			_lowercase ):

					'''simple docstring'''







					_A								= {'''gzip''': gz_file, '''xz''': xz_file, '''zstd''': zstd_path}
					_A								= input_paths[compression_format]
					_A								= tmp_path / '''cache'''
					_A								= DownloadConfig(cache_dir=_lowercase					,			extract_compressed_file=_lowercase )
					_A								= cached_path(_lowercase					,			download_config=_lowercase )
					with open(_lowercase ) as f:
										_A								= f.read()
					with open(_lowercase ) as f:
										_A								= f.read()
					assert extracted_file_content == expected_file_content




@pytest.mark.parametrize('''default_extracted'''					,			[True, False] )
@pytest.mark.parametrize('''default_cache_dir'''					,			[True, False] )
def      __A					(     _lowercase					,			_lowercase					,			_lowercase					,			_lowercase					,			_lowercase ):

					'''simple docstring'''







					_A								= '''custom_cache'''
					_A								= '''custom_extracted_dir'''
					_A								= tmp_path / '''custom_extracted_path'''
					if default_extracted:
										_A								= ('''downloads''' if default_cache_dir else custom_cache_dir, '''extracted''')
					else:
										monkeypatch.setattr('''datasets.config.EXTRACTED_DATASETS_DIR'''					,			_lowercase )
										monkeypatch.setattr('''datasets.config.EXTRACTED_DATASETS_PATH'''					,			str(_lowercase ) )
										_A								= custom_extracted_path.parts[-2:] if default_cache_dir else (custom_cache_dir, custom_extracted_dir)

					_A								= xz_file
					_A								= (
					    DownloadConfig(extract_compressed_file=_lowercase )
					    if default_cache_dir
					    else DownloadConfig(cache_dir=tmp_path / custom_cache_dir					,			extract_compressed_file=_lowercase )
					)
					_A								= cached_path(_lowercase					,			download_config=_lowercase )
					assert Path(_lowercase ).parent.parts[-2:] == expected




def      __A					(     _lowercase ):

					'''simple docstring'''







					_A								= str(Path(_lowercase ).resolve() )
					assert cached_path(_lowercase ) == text_file
					# relative path
					_A								= str(Path(_lowercase ).resolve().relative_to(Path(os.getcwd() ) ) )
					assert cached_path(_lowercase ) == text_file




def      __A					(     _lowercase ):

					'''simple docstring'''







					_A								= str(tmp_path.resolve() / '''__missing_file__.txt''' )
					with pytest.raises(_lowercase ):
										cached_path(_lowercase )
					# relative path
					_A								= '''./__missing_file__.txt'''
					with pytest.raises(_lowercase ):
										cached_path(_lowercase )




def      __A					(     _lowercase ):

					'''simple docstring'''







					_A								= get_from_cache(f"""tmp://{tmpfs_file}""" )
					with open(_lowercase ) as f:
										_A								= f.read()
					assert output_file_content == FILE_CONTENT




@patch('''datasets.config.HF_DATASETS_OFFLINE'''					,			_lowercase )
def      __A					(     ):

					'''simple docstring'''







					with pytest.raises(_lowercase ):
										cached_path('''https://huggingface.co''' )




@patch('''datasets.config.HF_DATASETS_OFFLINE'''					,			_lowercase )
def      __A					(     _lowercase ):

					'''simple docstring'''







					_A								= tmp_path_factory.mktemp('''data''' ) / '''file.html'''
					with pytest.raises(_lowercase ):
										http_get('''https://huggingface.co'''					,			temp_file=_lowercase )
					with pytest.raises(_lowercase ):
										http_head('''https://huggingface.co''' )




@patch('''datasets.config.HF_DATASETS_OFFLINE'''					,			_lowercase )
def      __A					(     _lowercase ):

					'''simple docstring'''







					_A								= tmp_path_factory.mktemp('''data''' ) / '''file.html'''
					with pytest.raises(_lowercase ):
										ftp_get('''ftp://huggingface.co'''					,			temp_file=_lowercase )
					with pytest.raises(_lowercase ):
										ftp_head('''ftp://huggingface.co''' )




@patch('''datasets.config.HF_DATASETS_OFFLINE'''					,			_lowercase )
def      __A					(     _lowercase ):

					'''simple docstring'''







					_A								= tmp_path_factory.mktemp('''data''' ) / '''file.html'''
					with pytest.raises(_lowercase ):
										fsspec_get('''s3://huggingface.co'''					,			temp_file=_lowercase )
					with pytest.raises(_lowercase ):
										fsspec_head('''s3://huggingface.co''' )



