



'''simple docstring'''

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


__lowerCAmelCase						      =       "\\n    Text data.\n    Second line of data."

__lowerCAmelCase						      =       "file"



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



@pytest.fixture
def 					__lowerCamelCase					(				lowerCAmelCase_  )      ->				Any:





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



@pytest.mark.parametrize('compression_format'						,      ['gzip', 'xz', 'zstd']  )
def 					__lowerCamelCase					(				lowerCAmelCase_						,      lowerCAmelCase_						,      lowerCAmelCase_						,      lowerCAmelCase_						,      lowerCAmelCase_						,      lowerCAmelCase_  )      ->				Dict:
		_a					:				str				=							{'gzip': gz_file, 'xz': xz_file, 'zstd': zstd_path}
		_a					:				Union[str, Any]				=							input_paths[compression_format]
		_a					:				List[str]				=							tmp_path / 'cache'
		_a					:				Tuple				=							DownloadConfig(cache_dir=lowerCAmelCase_						,      extract_compressed_file=lowerCAmelCase_  )
		_a					:				Any				=							cached_path(lowerCAmelCase_						,      download_config=lowerCAmelCase_  )
		with open(lowerCAmelCase_  ) as f:
				_a					:				Tuple				=							f.read()
		with open(lowerCAmelCase_  ) as f:
				_a					:				Dict				=							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 					__lowerCamelCase					(				lowerCAmelCase_						,      lowerCAmelCase_						,      lowerCAmelCase_						,      lowerCAmelCase_						,      lowerCAmelCase_  )      ->				Optional[Any]:
		_a					:				int				=							'custom_cache'
		_a					:				List[str]				=							'custom_extracted_dir'
		_a					:				Optional[int]				=							tmp_path / 'custom_extracted_path'
		if default_extracted:
				_a					:				str				=							('downloads' if default_cache_dir else custom_cache_dir, 'extracted')
		else:
				monkeypatch.setattr('datasets.config.EXTRACTED_DATASETS_DIR'						,      lowerCAmelCase_  )
				monkeypatch.setattr('datasets.config.EXTRACTED_DATASETS_PATH'						,      str(lowerCAmelCase_  )  )
				_a					:				Tuple				=							custom_extracted_path.parts[-2:] if default_cache_dir else (custom_cache_dir, custom_extracted_dir)

		_a					:				Optional[Any]				=							xz_file
		_a					:				List[str]				=							(
		    DownloadConfig(extract_compressed_file=lowerCAmelCase_  )
		    if default_cache_dir
		    else DownloadConfig(cache_dir=tmp_path / custom_cache_dir						,      extract_compressed_file=lowerCAmelCase_  )
		)
		_a					:				Optional[Any]				=							cached_path(lowerCAmelCase_						,      download_config=lowerCAmelCase_  )
		assert Path(lowerCAmelCase_  ).parent.parts[-2:] == expected



def 					__lowerCamelCase					(				lowerCAmelCase_  )      ->				Optional[Any]:
		_a					:				str				=							str(Path(lowerCAmelCase_  ).resolve()  )
		assert cached_path(lowerCAmelCase_  ) == text_file
		# relative path
		_a					:				Tuple				=							str(Path(lowerCAmelCase_  ).resolve().relative_to(Path(os.getcwd()  )  )  )
		assert cached_path(lowerCAmelCase_  ) == text_file



def 					__lowerCamelCase					(				lowerCAmelCase_  )      ->				Dict:
		_a					:				int				=							str(tmp_path.resolve() / '__missing_file__.txt'  )
		with pytest.raises(lowerCAmelCase_  ):
				cached_path(lowerCAmelCase_  )
		# relative path
		_a					:				int				=							'./__missing_file__.txt'
		with pytest.raises(lowerCAmelCase_  ):
				cached_path(lowerCAmelCase_  )



def 					__lowerCamelCase					(				lowerCAmelCase_  )      ->				Union[str, Any]:
		_a					:				Optional[int]				=							get_from_cache(f"""tmp://{tmpfs_file}"""  )
		with open(lowerCAmelCase_  ) as f:
				_a					:				List[str]				=							f.read()
		assert output_file_content == FILE_CONTENT



@patch('datasets.config.HF_DATASETS_OFFLINE'						,      lowerCAmelCase_  )
def 					__lowerCamelCase					(				)      ->				Optional[Any]:





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



@patch('datasets.config.HF_DATASETS_OFFLINE'						,      lowerCAmelCase_  )
def 					__lowerCamelCase					(				lowerCAmelCase_  )      ->				int:
		_a					:				Optional[Any]				=							tmp_path_factory.mktemp('data'  ) / 'file.html'
		with pytest.raises(lowerCAmelCase_  ):
				http_get('https://huggingface.co'						,      temp_file=lowerCAmelCase_  )
		with pytest.raises(lowerCAmelCase_  ):
				http_head('https://huggingface.co'  )



@patch('datasets.config.HF_DATASETS_OFFLINE'						,      lowerCAmelCase_  )
def 					__lowerCamelCase					(				lowerCAmelCase_  )      ->				List[Any]:
		_a					:				int				=							tmp_path_factory.mktemp('data'  ) / 'file.html'
		with pytest.raises(lowerCAmelCase_  ):
				ftp_get('ftp://huggingface.co'						,      temp_file=lowerCAmelCase_  )
		with pytest.raises(lowerCAmelCase_  ):
				ftp_head('ftp://huggingface.co'  )



@patch('datasets.config.HF_DATASETS_OFFLINE'						,      lowerCAmelCase_  )
def 					__lowerCamelCase					(				lowerCAmelCase_  )      ->				int:
		_a					:				Any				=							tmp_path_factory.mktemp('data'  ) / 'file.html'
		with pytest.raises(lowerCAmelCase_  ):
				fsspec_get('s3://huggingface.co'						,      temp_file=lowerCAmelCase_  )
		with pytest.raises(lowerCAmelCase_  ):
				fsspec_head('s3://huggingface.co'  )









'''simple docstring'''

import os
import random
import sys

from . import cryptomath_module as cryptomath
from . import rabin_miller

__lowerCAmelCase						      =       3



def 					__lowerCamelCase					(				lowerCAmelCase_  )      ->				int:
		print('Generating primitive root of p'  )
		while True:
				_a					:				List[Any]				=							random.randrange(3						,      lowerCAmelCase_  )
				if pow(lowerCAmelCase_						,      2						,      lowerCAmelCase_  ) == 1:
						continue
				if pow(lowerCAmelCase_						,      lowerCAmelCase_						,      lowerCAmelCase_  ) == 1:
						continue
				return g



def 					__lowerCamelCase					(				lowerCAmelCase_  )      ->				tuple[tuple[int, int, int, int], tuple[int, int]]:
		print('Generating prime p...'  )
		_a					:				int				=							rabin_miller.generate_large_prime(lowerCAmelCase_  )  # select large prime number.
		_a					:				List[str]				=							primitive_root(lowerCAmelCase_  )  # one primitive root on modulo p.
		_a					:				Any				=							random.randrange(3						,      lowerCAmelCase_  )  # private_key -> have to be greater than 2 for safety.
		_a					:				List[Any]				=							cryptomath.find_mod_inverse(pow(lowerCAmelCase_						,      lowerCAmelCase_						,      lowerCAmelCase_  )						,      lowerCAmelCase_  )

		_a					:				Tuple				=							(key_size, e_a, e_a, p)
		_a					:				str				=							(key_size, d)

		return public_key, private_key



def 					__lowerCamelCase					(				lowerCAmelCase_						,      lowerCAmelCase_  )      ->				None:
		if os.path.exists(f"""{name}_pubkey.txt"""  ) or os.path.exists(f"""{name}_privkey.txt"""  ):
				print('\nWARNING:'  )
				print(
				    f"""\"{name}_pubkey.txt\" or \"{name}_privkey.txt\" already exists. \n"""
				    'Use a different name or delete these files and re-run this program.'  )
				sys.exit()

		_a	, _a					:				Dict				=							generate_key(lowerCAmelCase_  )
		print(f"""\nWriting public key to file {name}_pubkey.txt..."""  )
		with open(f"""{name}_pubkey.txt"""						,      'w'  ) as fo:
				fo.write(f"""{public_key[0]},{public_key[1]},{public_key[2]},{public_key[3]}"""  )

		print(f"""Writing private key to file {name}_privkey.txt..."""  )
		with open(f"""{name}_privkey.txt"""						,      'w'  ) as fo:
				fo.write(f"""{private_key[0]},{private_key[1]}"""  )



def 					__lowerCamelCase					(				)      ->				None:
		print('Making key files...'  )
		make_key_files('elgamal'						,      2048  )
		print('Key files generation successful'  )


if __name__ == "__main__":
							main()




