


import argparse
import logging
import os
from pathlib import Path
from typing import Any, Dict

import pytorch_lightning as pl
from pytorch_lightning.utilities import rank_zero_info

from transformers import (
    AdamW,
    AutoConfig,
    AutoModel,
    AutoModelForPreTraining,
    AutoModelForQuestionAnswering,
    AutoModelForSeqaSeqLM,
    AutoModelForSequenceClassification,
    AutoModelForTokenClassification,
    AutoModelWithLMHead,
    AutoTokenizer,
    PretrainedConfig,
    PreTrainedTokenizer,
)
from transformers.optimization import (
    Adafactor,
    get_cosine_schedule_with_warmup,
    get_cosine_with_hard_restarts_schedule_with_warmup,
    get_linear_schedule_with_warmup,
    get_polynomial_decay_schedule_with_warmup,
)
from transformers.utils.versions import require_version


__UpperCamelCase     :    Union[str, Any] 	=				logging.getLogger(__name__)

require_version('pytorch_lightning>=1.0.4')

__UpperCamelCase     :    List[Any] 	=				{
    'base': AutoModel,
    'sequence-classification': AutoModelForSequenceClassification,
    'question-answering': AutoModelForQuestionAnswering,
    'pretraining': AutoModelForPreTraining,
    'token-classification': AutoModelForTokenClassification,
    'language-modeling': AutoModelWithLMHead,
    'summarization': AutoModelForSeqaSeqLM,
    'translation': AutoModelForSeqaSeqLM,
}


# update this and the import above to support new schedulers from transformers.optimization
__UpperCamelCase     :    Optional[int] 	=				{
    'linear': get_linear_schedule_with_warmup,
    'cosine': get_cosine_schedule_with_warmup,
    'cosine_w_restarts': get_cosine_with_hard_restarts_schedule_with_warmup,
    'polynomial': get_polynomial_decay_schedule_with_warmup,
    # '': get_constant_schedule,             # not supported for now
    # '': get_constant_schedule_with_warmup, # not supported for now
}
__UpperCamelCase     :    Tuple 	=				sorted(arg_to_scheduler.keys())
__UpperCamelCase     :    int 	=				'{' + ', '.join(arg_to_scheduler_choices) + '}'



class 				_UpperCamelCase			(    pl.LightningModule    ):

   '''simple docstring'''







   def __init__(	self	:			List[Any]							,	_lowerCamelCase	:			Tuple							,	_lowerCamelCase	:			Optional[Any]=None							,	_lowerCamelCase	:			Tuple="base"							,	_lowerCamelCase	:			Any=None							,	_lowerCamelCase	:			List[Any]=None							,	_lowerCamelCase	:			Union[str, Any]=None							,	**_lowerCamelCase	:			str							,	):






         '''simple docstring'''



         super().__init__()
         # TODO: move to self.save_hyperparameters()
         # self.save_hyperparameters()
         # can also expand arguments into trainer signature for easier reading

         self.save_hyperparameters(_lowerCamelCase				)
         __lowerCamelCase  :   Optional[Any]							    =    0
         __lowerCamelCase  :   Optional[Any]							    =    Path(self.hparams.output_dir				)
         __lowerCamelCase  :   Optional[Any]							    =    self.hparams.cache_dir if self.hparams.cache_dir else None
         if config is None:
               __lowerCamelCase  :   Tuple							    =    AutoConfig.from_pretrained(
                   self.hparams.config_name if self.hparams.config_name else self.hparams.model_name_or_path							,	**({"""num_labels""": num_labels} if num_labels is not None else {})							,	cache_dir=_lowerCamelCase							,	**_lowerCamelCase							,	)
         else:
               __lowerCamelCase  :   List[Any]							    =    config

         __lowerCamelCase  :   int							    =    ("""encoder_layerdrop""", """decoder_layerdrop""", """dropout""", """attention_dropout""")
         for p in extra_model_params:
               if getattr(self.hparams							,	_lowerCamelCase							,	_lowerCamelCase				):
                     assert hasattr(self.config							,	_lowerCamelCase				), F"""model config doesn't have a `{p}` attribute"""
                     setattr(self.config							,	_lowerCamelCase							,	getattr(self.hparams							,	_lowerCamelCase				)				)

         if tokenizer is None:
               __lowerCamelCase  :   Union[str, Any]							    =    AutoTokenizer.from_pretrained(
                   self.hparams.tokenizer_name if self.hparams.tokenizer_name else self.hparams.model_name_or_path							,	cache_dir=_lowerCamelCase							,	)
         else:
               __lowerCamelCase  :   Optional[Any]							    =    tokenizer
         __lowerCamelCase  :   Optional[int]							    =    MODEL_MODES[mode]
         if model is None:
               __lowerCamelCase  :   List[str]							    =    self.model_type.from_pretrained(
                   self.hparams.model_name_or_path							,	from_tf=bool(""".ckpt""" in self.hparams.model_name_or_path				)							,	config=self.config							,	cache_dir=_lowerCamelCase							,	)
         else:
               __lowerCamelCase  :   Tuple							    =    model

   def 	_snake_case     (	self	:			Dict							,	*_lowerCamelCase	:			str							,	**_lowerCamelCase	:			Tuple				):






         '''simple docstring'''



         __lowerCamelCase  :   Union[str, Any]							    =    self.model_type.from_pretrained(*_lowerCamelCase							,	**_lowerCamelCase				)

   def 	_snake_case     (	self	:			Optional[int]				):






         '''simple docstring'''



         __lowerCamelCase  :   Optional[Any]							    =    arg_to_scheduler[self.hparams.lr_scheduler]
         __lowerCamelCase  :   Optional[int]							    =    get_schedule_func(
             self.opt							,	num_warmup_steps=self.hparams.warmup_steps							,	num_training_steps=self.total_steps()				)
         __lowerCamelCase  :   Dict							    =    {"""scheduler""": scheduler, """interval""": """step""", """frequency""": 1}
         return scheduler

   def 	_snake_case     (	self	:			Dict				):






         '''simple docstring'''



         __lowerCamelCase  :   List[Any]							    =    self.model
         __lowerCamelCase  :   Union[str, Any]							    =    ["""bias""", """LayerNorm.weight"""]
         __lowerCamelCase  :   List[Any]							    =    [
             {
                 """params""": [
                     p for n, p in model.named_parameters() if not any(nd in n for nd in no_decay				)
                 ],  # check this named paramters
                 """weight_decay""": self.hparams.weight_decay,
             },
             {
                 """params""": [p for n, p in model.named_parameters() if any(nd in n for nd in no_decay				)],
                 """weight_decay""": 0.0,
             },
         ]
         if self.hparams.adafactor:
               __lowerCamelCase  :   Any							    =    Adafactor(
                   _lowerCamelCase							,	lr=self.hparams.learning_rate							,	scale_parameter=_lowerCamelCase							,	relative_step=_lowerCamelCase				)

         else:
               __lowerCamelCase  :   str							    =    AdamW(
                   _lowerCamelCase							,	lr=self.hparams.learning_rate							,	eps=self.hparams.adam_epsilon				)
         __lowerCamelCase  :   str							    =    optimizer

         __lowerCamelCase  :   Optional[Any]							    =    self.get_lr_scheduler()

         return [optimizer], [scheduler]

   def 	_snake_case     (	self	:			Any							,	_lowerCamelCase	:			Union[str, Any]							,	_lowerCamelCase	:			List[Any]				):






         '''simple docstring'''



         return self.validation_step(_lowerCamelCase							,	_lowerCamelCase				)

   def 	_snake_case     (	self	:			int							,	_lowerCamelCase	:			str				):






         '''simple docstring'''



         return self.validation_end(_lowerCamelCase				)

   def 	_snake_case     (	self	:			List[Any]				):






         '''simple docstring'''



         __lowerCamelCase  :   List[Any]							    =    max(1							,	self.hparams.gpus				)  # TODO: consider num_tpu_cores
         __lowerCamelCase  :   Tuple							    =    self.hparams.train_batch_size * self.hparams.accumulate_grad_batches * num_devices
         return (self.dataset_size / effective_batch_size) * self.hparams.max_epochs

   def 	_snake_case     (	self	:			List[str]							,	_lowerCamelCase	:			Any				):






         '''simple docstring'''



         if stage == "test":
               __lowerCamelCase  :   Any							    =    len(self.test_dataloader().dataset				)
         else:
               __lowerCamelCase  :   Optional[Any]							    =    self.get_dataloader("""train"""							,	self.hparams.train_batch_size							,	shuffle=_lowerCamelCase				)
               __lowerCamelCase  :   List[Any]							    =    len(self.train_dataloader().dataset				)

   def 	_snake_case     (	self	:			Any							,	_lowerCamelCase	:			Any							,	_lowerCamelCase	:			List[str]							,	_lowerCamelCase	:			int = False				):






         '''simple docstring'''



         raise NotImplementedError("""You must implement this for your task"""				)

   def 	_snake_case     (	self	:			Union[str, Any]				):






         '''simple docstring'''



         return self.train_loader

   def 	_snake_case     (	self	:			Union[str, Any]				):






         '''simple docstring'''



         return self.get_dataloader("""dev"""							,	self.hparams.eval_batch_size							,	shuffle=_lowerCamelCase				)

   def 	_snake_case     (	self	:			Tuple				):






         '''simple docstring'''



         return self.get_dataloader("""test"""							,	self.hparams.eval_batch_size							,	shuffle=_lowerCamelCase				)

   def 	_snake_case     (	self	:			Optional[int]							,	_lowerCamelCase	:			str				):






         '''simple docstring'''



         return os.path.join(
             self.hparams.data_dir							,	"""cached_{}_{}_{}""".format(
                 _lowerCamelCase							,	list(filter(_lowerCamelCase							,	self.hparams.model_name_or_path.split("""/"""				)				)				).pop()							,	str(self.hparams.max_seq_length				)							,	)							,	)

   @pl.utilities.rank_zero_only
   def 	_snake_case     (	self	:			Optional[Any]							,	_lowerCamelCase	:			str				):






         '''simple docstring'''



         __lowerCamelCase  :   Union[str, Any]							    =    self.output_dir.joinpath("""best_tfmr"""				)
         __lowerCamelCase  :   Union[str, Any]							    =    self.step_count
         self.model.save_pretrained(_lowerCamelCase				)
         self.tokenizer.save_pretrained(_lowerCamelCase				)


   @staticmethod
   def 	_snake_case     (	_lowerCamelCase	:			Tuple							,	_lowerCamelCase	:			List[Any]				):






         '''simple docstring'''



         parser.add_argument(
             """--model_name_or_path"""							,	default=_lowerCamelCase							,	type=_lowerCamelCase							,	required=_lowerCamelCase							,	help="""Path to pretrained model or model identifier from huggingface.co/models"""							,	)
         parser.add_argument(
             """--config_name"""							,	default=""""""							,	type=_lowerCamelCase							,	help="""Pretrained config name or path if not the same as model_name"""				)
         parser.add_argument(
             """--tokenizer_name"""							,	default=_lowerCamelCase							,	type=_lowerCamelCase							,	help="""Pretrained tokenizer name or path if not the same as model_name"""							,	)
         parser.add_argument(
             """--cache_dir"""							,	default=str(Path(_lowerCamelCase				).parent / """test_run""" / """cache"""				)							,	type=_lowerCamelCase							,	help="""Where do you want to store the pre-trained models downloaded from huggingface.co"""							,	)
         parser.add_argument(
             """--encoder_layerdrop"""							,	type=_lowerCamelCase							,	help="""Encoder layer dropout probability (Optional). Goes into model.config"""							,	)
         parser.add_argument(
             """--decoder_layerdrop"""							,	type=_lowerCamelCase							,	help="""Decoder layer dropout probability (Optional). Goes into model.config"""							,	)
         parser.add_argument(
             """--dropout"""							,	type=_lowerCamelCase							,	help="""Dropout probability (Optional). Goes into model.config"""							,	)
         parser.add_argument(
             """--attention_dropout"""							,	type=_lowerCamelCase							,	help="""Attention dropout probability (Optional). Goes into model.config"""							,	)
         parser.add_argument("""--learning_rate"""							,	default=5E-5							,	type=_lowerCamelCase							,	help="""The initial learning rate for Adam."""				)
         parser.add_argument(
             """--lr_scheduler"""							,	default="""linear"""							,	choices=_lowerCamelCase							,	metavar=_lowerCamelCase							,	type=_lowerCamelCase							,	help="""Learning rate scheduler"""							,	)
         parser.add_argument("""--weight_decay"""							,	default=0.0							,	type=_lowerCamelCase							,	help="""Weight decay if we apply some."""				)
         parser.add_argument("""--adam_epsilon"""							,	default=1E-8							,	type=_lowerCamelCase							,	help="""Epsilon for Adam optimizer."""				)
         parser.add_argument("""--warmup_steps"""							,	default=0							,	type=_lowerCamelCase							,	help="""Linear warmup over warmup_steps."""				)
         parser.add_argument("""--num_workers"""							,	default=4							,	type=_lowerCamelCase							,	help="""kwarg passed to DataLoader"""				)
         parser.add_argument("""--num_train_epochs"""							,	dest="""max_epochs"""							,	default=3							,	type=_lowerCamelCase				)
         parser.add_argument("""--train_batch_size"""							,	default=3_2							,	type=_lowerCamelCase				)
         parser.add_argument("""--eval_batch_size"""							,	default=3_2							,	type=_lowerCamelCase				)
         parser.add_argument("""--adafactor"""							,	action="""store_true"""				)



class 				_UpperCamelCase			(    pl.Callback    ):

   '''simple docstring'''







   def 	_snake_case     (	self	:			List[str]							,	_lowerCamelCase	:			Union[str, Any]							,	_lowerCamelCase	:			Union[str, Any]				):






         '''simple docstring'''



         if (
             trainer.is_global_zero and trainer.global_rank == 0
         ):  # we initialize the retriever only on master worker with RAY. In new pytorch-lightning accelorators are removed.
               pl_module.model.rag.retriever.init_retrieval()  # better to use hook functions.



class 				_UpperCamelCase			(    pl.Callback    ):

   '''simple docstring'''







   def 	_snake_case     (	self	:			Optional[int]							,	_lowerCamelCase	:			Any							,	_lowerCamelCase	:			Optional[int]				):






         '''simple docstring'''



         for name, param in pl_module.model.rag.named_parameters():
               if param.grad is None:
                     print(_lowerCamelCase				)



class 				_UpperCamelCase			(    pl.Callback    ):

   '''simple docstring'''







   def 	_snake_case     (	self	:			Tuple							,	_lowerCamelCase	:			Union[str, Any]							,	_lowerCamelCase	:			Optional[int]				):






         '''simple docstring'''



         __lowerCamelCase  :   Tuple							    =    trainer.lr_schedulers[0]["""scheduler"""]
         __lowerCamelCase  :   List[Any]							    =    {F"""lr_group_{i}""": lr for i, lr in enumerate(lr_scheduler.get_lr()				)}
         pl_module.logger.log_metrics(_lowerCamelCase				)

   def 	_snake_case     (	self	:			Any							,	_lowerCamelCase	:			Union[str, Any]							,	_lowerCamelCase	:			Tuple				):






         '''simple docstring'''



         rank_zero_info("""***** Validation results *****"""				)
         __lowerCamelCase  :   Dict							    =    trainer.callback_metrics
         # Log results
         for key in sorted(_lowerCamelCase				):
               if key not in ["log", "progress_bar"]:
                     rank_zero_info("""{} = {}\n""".format(_lowerCamelCase							,	str(metrics[key]				)				)				)


   def 	_snake_case     (	self	:			Dict							,	_lowerCamelCase	:			Union[str, Any]							,	_lowerCamelCase	:			int				):






         '''simple docstring'''



         rank_zero_info("""***** Test results *****"""				)
         __lowerCamelCase  :   List[str]							    =    trainer.callback_metrics
         # Log and save results to file
         __lowerCamelCase  :   Tuple							    =    os.path.join(pl_module.hparams.output_dir							,	"""test_results.txt"""				)
         with open(_lowerCamelCase							,	"""w"""				) as writer:
               for key in sorted(_lowerCamelCase				):
                     if key not in ["log", "progress_bar"]:
                           rank_zero_info("""{} = {}\n""".format(_lowerCamelCase							,	str(metrics[key]				)				)				)
                           writer.write("""{} = {}\n""".format(_lowerCamelCase							,	str(metrics[key]				)				)				)





def 	_UpperCAmelCase							(  UpperCAmelCase   :      int    ,						UpperCAmelCase   :      str   ):
      """simple docstring"""




      parser.add_argument(
          """--output_dir"""    ,						default=str(Path(__UpperCAmelCase   ).parent / """test_run""" / """model_checkpoints"""   )    ,						type=__UpperCAmelCase    ,						help="""The output directory where the model predictions and checkpoints will be written."""    ,						)
      parser.add_argument(
          """--fp16"""    ,						action="""store_true"""    ,						help="""Whether to use 16-bit (mixed) precision (through NVIDIA apex) instead of 32-bit"""    ,						)

      parser.add_argument(
          """--fp16_opt_level"""    ,						type=__UpperCAmelCase    ,						default="""O2"""    ,						help=(
              """For fp16: Apex AMP optimization level selected in [\'O0\', \'O1\', \'O2\', and \'O3\']."""
              """See details at https://nvidia.github.io/apex/amp.html"""
          )    ,						)
      parser.add_argument("""--n_tpu_cores"""    ,						dest="""tpu_cores"""    ,						type=__UpperCAmelCase   )
      parser.add_argument("""--max_grad_norm"""    ,						dest="""gradient_clip_val"""    ,						default=1.0    ,						type=__UpperCAmelCase    ,						help="""Max gradient norm"""   )
      parser.add_argument("""--do_train"""    ,						action="""store_true"""    ,						help="""Whether to run training."""   )
      parser.add_argument("""--do_predict"""    ,						action="""store_true"""    ,						help="""Whether to run predictions on the test set."""   )
      parser.add_argument(
          """--gradient_accumulation_steps"""    ,						dest="""accumulate_grad_batches"""    ,						type=__UpperCAmelCase    ,						default=1    ,						help="""Number of updates steps to accumulate before performing a backward/update pass."""    ,						)
      parser.add_argument("""--seed"""    ,						type=__UpperCAmelCase    ,						default=42    ,						help="""random seed for initialization"""   )
      parser.add_argument(
          """--data_dir"""    ,						default=str(Path(__UpperCAmelCase   ).parent / """test_run""" / """dummy-train-data"""   )    ,						type=__UpperCAmelCase    ,						help="""The input data dir. Should contain the training files for the CoNLL-2003 NER task."""    ,						)





def 	_UpperCAmelCase							(  UpperCAmelCase   :      Optional[int]    ,						UpperCAmelCase   :      Tuple    ,						UpperCAmelCase   :      int=None    ,						UpperCAmelCase   :      Tuple=True    ,						UpperCAmelCase   :      Any=[]    ,						UpperCAmelCase   :      List[str]=None    ,						UpperCAmelCase   :      Union[str, Any]=None    ,						**UpperCAmelCase   :      Optional[int]    ,						):
      """simple docstring"""




      pl.seed_everything(args.seed   )

      # init model
      __lowerCamelCase  :   Union[str, Any]							    =    Path(model.hparams.output_dir   )
      odir.mkdir(exist_ok=__UpperCAmelCase   )

      # add custom checkpoints
      if checkpoint_callback is None:
            __lowerCamelCase  :   List[str]							    =    pl.callbacks.ModelCheckpoint(
                filepath=args.output_dir    ,						prefix="""checkpoint"""    ,						monitor="""val_loss"""    ,						mode="""min"""    ,						save_top_k=1   )
      if early_stopping_callback:
            extra_callbacks.append(__UpperCAmelCase   )
      if logging_callback is None:
            __lowerCamelCase  :   Tuple							    =    LoggingCallback()

      __lowerCamelCase  :   Dict							    =    {}

      if args.fpaa:
            __lowerCamelCase  :   Dict							    =    16

      if args.gpus > 1:
            __lowerCamelCase  :   List[str]							    =    """auto"""
            __lowerCamelCase  :   Optional[int]							    =    """ddp"""

      __lowerCamelCase  :   Optional[int]							    =    args.accumulate_grad_batches
      __lowerCamelCase  :   Union[str, Any]							    =    None
      __lowerCamelCase  :   Any							    =    """auto"""

      __lowerCamelCase  :   Dict							    =    pl.Trainer.from_argparse_args(
          __UpperCAmelCase    ,						weights_summary=__UpperCAmelCase    ,						callbacks=[logging_callback] + extra_callbacks + [InitCallback()] + [checkpoint_callback]    ,						logger=__UpperCAmelCase    ,						val_check_interval=1    ,						num_sanity_val_steps=2    ,						**__UpperCAmelCase    ,						)

      if args.do_train:
            trainer.fit(__UpperCAmelCase   )

      else:
            print("""RAG modeling tests with new set functions successfuly executed!"""   )
      return trainer








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_      :      int    =     '\\n    Text data.\n    Second line of data.'

a_      :      Optional[Any]    =     'file'






@pytest.fixture(scope='''session'''       )
def     __a					(		__UpperCAmelCase       ):
 a__												= tmp_path_factory.mktemp('''data'''       ) / (FILE_PATH + '''.zstd''')
 a__												= bytes(__UpperCAmelCase							,  '''utf-8'''       )
 with zstd.open(__UpperCAmelCase							,  '''wb'''       ) as f:
  f.write(__UpperCAmelCase       )
 return path






@pytest.fixture
def     __a					(		__UpperCAmelCase       ):
 with open(os.path.join(tmpfs.local_root_dir							,  __UpperCAmelCase       )							,  '''w'''       ) as f:
  f.write(__UpperCAmelCase       )
 return FILE_PATH






@pytest.mark.parametrize('''compression_format'''							,  ['''gzip''', '''xz''', '''zstd''']       )
def     __a					(		__UpperCAmelCase							,  __UpperCAmelCase							,  __UpperCAmelCase							,  __UpperCAmelCase							,  __UpperCAmelCase							,  __UpperCAmelCase       ):
 a__												= {'''gzip''': gz_file, '''xz''': xz_file, '''zstd''': zstd_path}
 a__												= input_paths[compression_format]
 a__												= tmp_path / '''cache'''
 a__												= DownloadConfig(cache_dir=__UpperCAmelCase							,  extract_compressed_file=__UpperCAmelCase       )
 a__												= cached_path(__UpperCAmelCase							,  download_config=__UpperCAmelCase       )
 with open(__UpperCAmelCase       ) as f:
  a__												= f.read()
 with open(__UpperCAmelCase       ) 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					(		__UpperCAmelCase							,  __UpperCAmelCase							,  __UpperCAmelCase							,  __UpperCAmelCase							,  __UpperCAmelCase       ):
 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'''							,  __UpperCAmelCase       )
  monkeypatch.setattr('''datasets.config.EXTRACTED_DATASETS_PATH'''							,  str(__UpperCAmelCase       )       )
  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=__UpperCAmelCase       )
     if default_cache_dir
     else DownloadConfig(cache_dir=tmp_path / custom_cache_dir							,  extract_compressed_file=__UpperCAmelCase       )
 )
 a__												= cached_path(__UpperCAmelCase							,  download_config=__UpperCAmelCase       )
 assert Path(__UpperCAmelCase       ).parent.parts[-2:] == expected






def     __a					(		__UpperCAmelCase       ):
 # absolute path
 a__												= str(Path(__UpperCAmelCase       ).resolve()       )
 assert cached_path(__UpperCAmelCase       ) == text_file
 # relative path
 a__												= str(Path(__UpperCAmelCase       ).resolve().relative_to(Path(os.getcwd()       )       )       )
 assert cached_path(__UpperCAmelCase       ) == text_file






def     __a					(		__UpperCAmelCase       ):
 # absolute path
 a__												= str(tmp_path.resolve() / '''__missing_file__.txt'''       )
 with pytest.raises(__UpperCAmelCase       ):
  cached_path(__UpperCAmelCase       )
 # relative path
 a__												= '''./__missing_file__.txt'''
 with pytest.raises(__UpperCAmelCase       ):
  cached_path(__UpperCAmelCase       )






def     __a					(		__UpperCAmelCase       ):
 a__												= get_from_cache(f"tmp://{tmpfs_file}"       )
 with open(__UpperCAmelCase       ) as f:
  a__												= f.read()
 assert output_file_content == FILE_CONTENT






@patch('''datasets.config.HF_DATASETS_OFFLINE'''							,  __UpperCAmelCase       )
def     __a					(		):
 with pytest.raises(__UpperCAmelCase       ):
  cached_path('''https://huggingface.co'''       )






@patch('''datasets.config.HF_DATASETS_OFFLINE'''							,  __UpperCAmelCase       )
def     __a					(		__UpperCAmelCase       ):
 a__												= tmp_path_factory.mktemp('''data'''       ) / '''file.html'''
 with pytest.raises(__UpperCAmelCase       ):
  http_get('''https://huggingface.co'''							,  temp_file=__UpperCAmelCase       )
 with pytest.raises(__UpperCAmelCase       ):
  http_head('''https://huggingface.co'''       )






@patch('''datasets.config.HF_DATASETS_OFFLINE'''							,  __UpperCAmelCase       )
def     __a					(		__UpperCAmelCase       ):
 a__												= tmp_path_factory.mktemp('''data'''       ) / '''file.html'''
 with pytest.raises(__UpperCAmelCase       ):
  ftp_get('''ftp://huggingface.co'''							,  temp_file=__UpperCAmelCase       )
 with pytest.raises(__UpperCAmelCase       ):
  ftp_head('''ftp://huggingface.co'''       )






@patch('''datasets.config.HF_DATASETS_OFFLINE'''							,  __UpperCAmelCase       )
def     __a					(		__UpperCAmelCase       ):
 a__												= tmp_path_factory.mktemp('''data'''       ) / '''file.html'''
 with pytest.raises(__UpperCAmelCase       ):
  fsspec_get('''s3://huggingface.co'''							,  temp_file=__UpperCAmelCase       )
 with pytest.raises(__UpperCAmelCase       ):
  fsspec_head('''s3://huggingface.co'''       )





