



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,
)


UpperCAmelCase_		 =					"""\
    Text data.
    Second line of data."""

UpperCAmelCase_		 =					"""file"""

@pytest.fixture(scope='''session'''				)
def 	SCREAMING_SNAKE_CASE_							(	_snake_case	:List[Any]				)  ->     Optional[int]:
   _A      =       tmp_path_factory.mktemp('''data'''				) / (FILE_PATH + '''.zstd''')
   _A      =       bytes(_snake_case	,   '''utf-8'''				)
   with zstd.open(_snake_case	,   '''wb'''				) as f:
      f.write(_snake_case				)
   return path

@pytest.fixture
def 	SCREAMING_SNAKE_CASE_							(	_snake_case	:Tuple				)  ->     Any:
   with open(os.path.join(tmpfs.local_root_dir	,   _snake_case				)	,   '''w'''				) as f:
      f.write(_snake_case				)
   return FILE_PATH

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

def 	SCREAMING_SNAKE_CASE_							(	_snake_case	:Tuple				)  ->     Tuple:
   # absolute path
   _A      =       str(Path(_snake_case				).resolve()				)
   assert cached_path(_snake_case				) == text_file
   # relative path
   _A      =       str(Path(_snake_case				).resolve().relative_to(Path(os.getcwd()				)				)				)
   assert cached_path(_snake_case				) == text_file

def 	SCREAMING_SNAKE_CASE_							(	_snake_case	:List[Any]				)  ->     List[Any]:
   # absolute path
   _A      =       str(tmp_path.resolve() / '''__missing_file__.txt'''				)
   with pytest.raises(_snake_case				):
      cached_path(_snake_case				)
   # relative path
   _A      =       '''./__missing_file__.txt'''
   with pytest.raises(_snake_case				):
      cached_path(_snake_case				)

def 	SCREAMING_SNAKE_CASE_							(	_snake_case	:Dict				)  ->     Any:
   _A      =       get_from_cache(F'''tmp://{tmpfs_file}'''				)
   with open(_snake_case				) as f:
      _A      =       f.read()
   assert output_file_content == FILE_CONTENT

@patch('''datasets.config.HF_DATASETS_OFFLINE'''	,   _snake_case				)
def 	SCREAMING_SNAKE_CASE_							(	)  ->     List[str]:
   with pytest.raises(_snake_case				):
      cached_path('''https://huggingface.co'''				)

@patch('''datasets.config.HF_DATASETS_OFFLINE'''	,   _snake_case				)
def 	SCREAMING_SNAKE_CASE_							(	_snake_case	:int				)  ->     Dict:
   _A      =       tmp_path_factory.mktemp('''data'''				) / '''file.html'''
   with pytest.raises(_snake_case				):
      http_get('''https://huggingface.co'''	,   temp_file=_snake_case				)
   with pytest.raises(_snake_case				):
      http_head('''https://huggingface.co'''				)

@patch('''datasets.config.HF_DATASETS_OFFLINE'''	,   _snake_case				)
def 	SCREAMING_SNAKE_CASE_							(	_snake_case	:Optional[int]				)  ->     Union[str, Any]:
   _A      =       tmp_path_factory.mktemp('''data'''				) / '''file.html'''
   with pytest.raises(_snake_case				):
      ftp_get('''ftp://huggingface.co'''	,   temp_file=_snake_case				)
   with pytest.raises(_snake_case				):
      ftp_head('''ftp://huggingface.co'''				)


@patch('''datasets.config.HF_DATASETS_OFFLINE'''	,   _snake_case				)
def 	SCREAMING_SNAKE_CASE_							(	_snake_case	:Any				)  ->     Optional[int]:
   _A      =       tmp_path_factory.mktemp('''data'''				) / '''file.html'''
   with pytest.raises(_snake_case				):
      fsspec_get('''s3://huggingface.co'''	,   temp_file=_snake_case				)
   with pytest.raises(_snake_case				):
      fsspec_head('''s3://huggingface.co'''				)








import warnings

from ...configuration_utils import PretrainedConfig
from ...utils import logging


UpperCAmelCase_		 =					logging.get_logger(__name__)

UpperCAmelCase_		 =					{
    """xlnet-base-cased""": """https://huggingface.co/xlnet-base-cased/resolve/main/config.json""",
    """xlnet-large-cased""": """https://huggingface.co/xlnet-large-cased/resolve/main/config.json""",
}


class 	lowerCamelCase__	(       _A):



 """simple docstring"""

 a__					: Any =  "xlnet"
 a__					: Dict =  ["mems"]
 a__					: List[str] =  {
     "n_token": "vocab_size",  # Backward compatibility
     "hidden_size": "d_model",
     "num_attention_heads": "n_head",
     "num_hidden_layers": "n_layer",
 }

 def __init__(    self					:					int			,						__lowerCAmelCase					:					Dict=3_20_00			,						__lowerCAmelCase					:					List[str]=10_24			,						__lowerCAmelCase					:					Dict=24			,						__lowerCAmelCase					:					Optional[Any]=16			,						__lowerCAmelCase					:					Dict=40_96			,						__lowerCAmelCase					:					Any="gelu"			,						__lowerCAmelCase					:					int=True			,						__lowerCAmelCase					:					List[str]="bi"			,						__lowerCAmelCase					:					Dict=0.02			,						__lowerCAmelCase					:					Union[str, Any]=1E-12			,						__lowerCAmelCase					:					Optional[Any]=0.1			,						__lowerCAmelCase					:					Optional[Any]=5_12			,						__lowerCAmelCase					:					List[str]=None			,						__lowerCAmelCase					:					Tuple=True			,						__lowerCAmelCase					:					Tuple=False			,						__lowerCAmelCase					:					Optional[Any]=False			,						__lowerCAmelCase					:					Union[str, Any]=-1			,						__lowerCAmelCase					:					Optional[Any]=False			,						__lowerCAmelCase					:					Any="last"			,						__lowerCAmelCase					:					List[Any]=True			,						__lowerCAmelCase					:					Tuple="tanh"			,						__lowerCAmelCase					:					int=0.1			,						__lowerCAmelCase					:					str=5			,						__lowerCAmelCase					:					str=5			,						__lowerCAmelCase					:					List[str]=5			,						__lowerCAmelCase					:					List[str]=1			,						__lowerCAmelCase					:					Optional[int]=2			,						**__lowerCAmelCase					:					List[str]			,						)   ->	Tuple:
    _A      =       vocab_size
    _A      =       d_model
    _A      =       n_layer
    _A      =       n_head
    if d_model % n_head != 0:
       raise ValueError(f'''\'d_model % n_head\' ({d_model % n_head}) should be equal to 0'''      )
    if "d_head" in kwargs:
       if kwargs["d_head"] != d_model // n_head:
          raise ValueError(
              f'''`d_head` ({kwargs['d_head']}) should be equal to `d_model // n_head` ({d_model // n_head})'''      )
    _A      =       d_model // n_head
    _A      =       ff_activation
    _A      =       d_inner
    _A      =       untie_r
    _A      =       attn_type

    _A      =       initializer_range
    _A      =       layer_norm_eps

    _A      =       dropout
    _A      =       mem_len
    _A      =       reuse_len
    _A      =       bi_data
    _A      =       clamp_len
    _A      =       same_length

    _A      =       summary_type
    _A      =       summary_use_proj
    _A      =       summary_activation
    _A      =       summary_last_dropout
    _A      =       start_n_top
    _A      =       end_n_top

    _A      =       bos_token_id
    _A      =       pad_token_id
    _A      =       eos_token_id

    if "use_cache" in kwargs:
       warnings.warn(
           '''The `use_cache` argument is deprecated and will be removed in a future version, use `use_mems_eval`'''
           ''' instead.'''			,						__lowerCAmelCase			,						)
       _A      =       kwargs['''use_cache''']

    _A      =       use_mems_eval
    _A      =       use_mems_train
    super().__init__(pad_token_id=__lowerCAmelCase			,						bos_token_id=__lowerCAmelCase			,						eos_token_id=__lowerCAmelCase			,						**__lowerCAmelCase      )

 @property
 def     snake_case_      (    self					:					Optional[Any]      )   ->	Union[str, Any]:
    logger.info(f'''The model {self.model_type} is one of the few models that has no sequence length limit.'''      )
    return -1

 @max_position_embeddings.setter
 def     snake_case_      (    self					:					Tuple			,						__lowerCAmelCase					:					Optional[Any]      )   ->	Dict:
    # Message copied from Transformer-XL documentation
    raise NotImplementedError(
        f'''The model {self.model_type} is one of the few models that has no sequence length limit.'''      )



