

import argparse
import glob
import inspect
import os
import re
import unittest

from get_test_info import get_test_classes

from tests.test_pipeline_mixin import pipeline_test_mapping


A__										=      {}
for task, _ in pipeline_test_mapping.items():
  A__										=      {"pt": None, "tf": None}


# DO **NOT** add item to this set (unless the reason is approved)
A__										=      {
    "tests/models/esm/test_modeling_esmfold.py",  # The pipeline test mapping is added to `test_modeling_esm.py`
}





def 				__a							(       __A						):

    if "ModelTesterMixin" in [x.__name__ for x in test_class.__bases__]:
        return "pt"
    elif "TFModelTesterMixin" in [x.__name__ for x in test_class.__bases__]:
        return "tf"
    elif "FlaxModelTesterMixin" in [x.__name__ for x in test_class.__bases__]:
        return "flax"
    else:
        return None





def 				__a							(       __A			,							__A						):
    # Use the cached results
    if PIPELINE_TEST_MAPPING[task].get(_snake_case			,							_snake_case						) is not None:
        return PIPELINE_TEST_MAPPING[task][framework]

    SCREAMING_SNAKE_CASE__         =    pipeline_test_mapping[task]["test"]
    SCREAMING_SNAKE_CASE__         =    None

    if framework == "pt":
        SCREAMING_SNAKE_CASE__         =    getattr(_snake_case			,							"model_mapping"			,							_snake_case						)
    elif framework == "tf":
        SCREAMING_SNAKE_CASE__         =    getattr(_snake_case			,							"tf_model_mapping"			,							_snake_case						)

    if mapping is not None:
        SCREAMING_SNAKE_CASE__         =    dict(mapping.items()						)

    # cache the results
    SCREAMING_SNAKE_CASE__         =    mapping
    return mapping





def 				__a							(       __A			,							__A						):
    SCREAMING_SNAKE_CASE__         =    get_framework(_snake_case						)
    if framework is None:
        return None
    SCREAMING_SNAKE_CASE__         =    get_mapping_for_task(_snake_case			,							_snake_case						)
    if mapping is None:
        return None

    SCREAMING_SNAKE_CASE__         =    list({model_class.config_class for model_class in test_class.all_model_classes}						)
    if len(_snake_case						) != 1:
        raise ValueError("There should be exactly one configuration class from `test_class.all_model_classes`."						)

    # This could be a list/tuple of model classes, but it's rare.
    SCREAMING_SNAKE_CASE__         =    mapping.get(config_classes[0]			,							_snake_case						)
    if isinstance(_snake_case			,							(tuple, list)						):
        SCREAMING_SNAKE_CASE__         =    sorted(_snake_case			,							key=lambda __A						: x.__name__						)

    return model_class





def 				__a							(       __A						):
    SCREAMING_SNAKE_CASE__         =    [(task, get_model_for_pipeline_test(_snake_case			,							_snake_case						)) for task in pipeline_test_mapping]
    SCREAMING_SNAKE_CASE__         =    sorted([(task, model) for task, model in mapping if model is not None]			,							key=lambda __A						: x[0]						)

    return dict(_snake_case						)





def 				__a							(       __A						):
    SCREAMING_SNAKE_CASE__         =    get_framework(_snake_case						)
    if framework == "pt":
        SCREAMING_SNAKE_CASE__         =    "torch"
    SCREAMING_SNAKE_CASE__         =    "{}"

    SCREAMING_SNAKE_CASE__         =    get_pipeline_model_mapping(_snake_case						)
    if len(_snake_case						) == 0:
        return ""

    SCREAMING_SNAKE_CASE__         =    []
    for task, model_classes in mapping.items():
        if isinstance(_snake_case			,							(tuple, list)						):
            # A list/tuple of model classes
            SCREAMING_SNAKE_CASE__         =    "(" + ", ".join([x.__name__ for x in model_classes]						) + ")"
        else:
            # A single model class
            SCREAMING_SNAKE_CASE__         =    model_classes.__name__
        texts.append(f"\"{task}\": {value}"						)
    SCREAMING_SNAKE_CASE__         =    "{" + ", ".join(_snake_case						) + "}"
    SCREAMING_SNAKE_CASE__         =    f"pipeline_model_mapping = {text} if is_{framework}_available() else {default_value}"

    return text





def 				__a							(       __A						):
    SCREAMING_SNAKE_CASE__         =    {"ModelTesterMixin", "TFModelTesterMixin", "FlaxModelTesterMixin"}
    if not issubclass(_snake_case			,							unittest.TestCase						):
        return False
    return len(base_class_names.intersection([x.__name__ for x in test_class.__bases__]						)						) > 0





def 				__a							(       __A						):
    SCREAMING_SNAKE_CASE__         =    [x for x in get_test_classes(_snake_case						) if is_valid_test_class(_snake_case						)]

    SCREAMING_SNAKE_CASE__         =    None
    for test_class in test_classes:
        # If a test class has defined `pipeline_model_mapping`, let's take it
        if getattr(_snake_case			,							"pipeline_model_mapping"			,							_snake_case						) is not None:
            SCREAMING_SNAKE_CASE__         =    test_class
            break
    # Take the test class with the shortest name (just a heuristic)
    if target_test_class is None and len(_snake_case						) > 0:
        SCREAMING_SNAKE_CASE__         =    sorted(_snake_case			,							key=lambda __A						: (len(x.__name__						), x.__name__)						)[0]

    return target_test_class





def 				__a							(       __A			,							__A			,							__A						):
    SCREAMING_SNAKE_CASE__         =    start_idx
    for idx, line in enumerate(lines[start_idx:]						):
        SCREAMING_SNAKE_CASE__         =    len(_snake_case						) - len(line.lstrip()						)
        if idx == 0 or indent > indent_level or (indent == indent_level and line.strip() == ")"):
            SCREAMING_SNAKE_CASE__         =    start_idx + idx
        elif idx > 0 and indent <= indent_level:
            # Outside the definition block of `pipeline_model_mapping`
            break

    return end_idx





def 				__a							(       __A			,							__A=False						):
    if getattr(_snake_case			,							"pipeline_model_mapping"			,							_snake_case						) is not None:
        if not overwrite:
            return "", -1

    SCREAMING_SNAKE_CASE__         =    get_pipeline_model_mapping_string(_snake_case						)
    if len(_snake_case						) == 0:
        return "", -1
    SCREAMING_SNAKE_CASE__         =    line_to_add + "\n"

    # The code defined the class `test_class`
    SCREAMING_SNAKE_CASE__  ,		SCREAMING_SNAKE_CASE__         =    inspect.getsourcelines(_snake_case						)
    # `inspect` gives the code for an object, including decorator(s) if any.
    # We (only) need the exact line of the class definition.
    for idx, line in enumerate(_snake_case						):
        if line.lstrip().startswith("class "						):
            SCREAMING_SNAKE_CASE__         =    class_lines[idx:]
            class_start_line_no += idx
            break
    SCREAMING_SNAKE_CASE__         =    class_start_line_no + len(_snake_case						) - 1

    # The index in `class_lines` that starts the definition of `all_model_classes`, `all_generative_model_classes` or
    # `pipeline_model_mapping`. This assumes they are defined in such order, and we take the start index of the last
    # block that appears in a `test_class`.
    SCREAMING_SNAKE_CASE__         =    None
    # The indent level of the line at `class_lines[start_idx]` (if defined)
    SCREAMING_SNAKE_CASE__         =    0
    # To record if `pipeline_model_mapping` is found in `test_class`.
    SCREAMING_SNAKE_CASE__         =    None
    for idx, line in enumerate(_snake_case						):
        if line.strip().startswith("all_model_classes = "						):
            SCREAMING_SNAKE_CASE__         =    len(_snake_case						) - len(line.lstrip()						)
            SCREAMING_SNAKE_CASE__         =    idx
        elif line.strip().startswith("all_generative_model_classes = "						):
            SCREAMING_SNAKE_CASE__         =    len(_snake_case						) - len(line.lstrip()						)
            SCREAMING_SNAKE_CASE__         =    idx
        elif line.strip().startswith("pipeline_model_mapping = "						):
            SCREAMING_SNAKE_CASE__         =    len(_snake_case						) - len(line.lstrip()						)
            SCREAMING_SNAKE_CASE__         =    idx
            SCREAMING_SNAKE_CASE__         =    line
            break

    if start_idx is None:
        return "", -1
    # Find the ending index (inclusive) of the above found block.
    SCREAMING_SNAKE_CASE__         =    find_block_ending(_snake_case			,							_snake_case			,							_snake_case						)

    # Extract `is_xxx_available()` from existing blocks: some models require specific libraries like `timm` and use
    # `is_timm_available()` instead of `is_torch_available()`.
    # Keep leading and trailing whitespaces
    SCREAMING_SNAKE_CASE__         =    re.compile(r"\s(is_\S+?_available\(\))\s"						)
    for line in class_lines[start_idx : end_idx + 1]:
        SCREAMING_SNAKE_CASE__         =    r.search(_snake_case						)
        if backend_condition is not None:
            # replace the leading and trailing whitespaces to the space character " ".
            SCREAMING_SNAKE_CASE__         =    " " + backend_condition[0][1:-1] + " "
            SCREAMING_SNAKE_CASE__         =    r.sub(_snake_case			,							_snake_case						)
            break

    if def_line is None:
        # `pipeline_model_mapping` is not defined. The target index is set to the ending index (inclusive) of
        # `all_model_classes` or `all_generative_model_classes`.
        SCREAMING_SNAKE_CASE__         =    end_idx
    else:
        # `pipeline_model_mapping` is defined. The target index is set to be one **BEFORE** its start index.
        SCREAMING_SNAKE_CASE__         =    start_idx - 1
        # mark the lines of the currently existing `pipeline_model_mapping` to be removed.
        for idx in range(_snake_case			,							end_idx + 1						):
            # These lines are going to be removed before writing to the test file.
            SCREAMING_SNAKE_CASE__         =    None  # noqa

    # Make sure the test class is a subclass of `PipelineTesterMixin`.
    SCREAMING_SNAKE_CASE__         =    [x.__name__ for x in test_class.__bases__]
    if "PipelineTesterMixin" not in parent_classes:
        # Put `PipelineTesterMixin` just before `unittest.TestCase`
        SCREAMING_SNAKE_CASE__         =    [x for x in parent_classes if x != "TestCase"] + ["PipelineTesterMixin"]
        if "TestCase" in parent_classes:
            # Here we **assume** the original string is always with `unittest.TestCase`.
            _parent_classes.append("unittest.TestCase"						)
        SCREAMING_SNAKE_CASE__         =    ", ".join(_parent_classes						)
        for idx, line in enumerate(_snake_case						):
            # Find the ending of the declaration of `test_class`
            if line.strip().endswith("):"						):
                # mark the lines of the declaration of `test_class` to be removed
                for _idx in range(idx + 1						):
                    SCREAMING_SNAKE_CASE__         =    None  # noqa
                break
        # Add the new, one-line, class declaration for `test_class`
        SCREAMING_SNAKE_CASE__         =    f"class {test_class.__name__}({parent_classes}):\n"

    # Add indentation
    SCREAMING_SNAKE_CASE__         =    " " * indent_level + line_to_add
    # Insert `pipeline_model_mapping` to `class_lines`.
    # (The line at `target_idx` should be kept by definition!)
    SCREAMING_SNAKE_CASE__         =    class_lines[: target_idx + 1] + [line_to_add] + class_lines[target_idx + 1 :]
    # Remove the lines that are marked to be removed
    SCREAMING_SNAKE_CASE__         =    [x for x in class_lines if x is not None]

    # Move from test class to module (in order to write to the test file)
    SCREAMING_SNAKE_CASE__         =    inspect.getsourcelines(inspect.getmodule(_snake_case						)						)[0]
    # Be careful with the 1-off between line numbers and array indices
    SCREAMING_SNAKE_CASE__         =    module_lines[: class_start_line_no - 1] + class_lines + module_lines[class_end_line_no:]
    SCREAMING_SNAKE_CASE__         =    "".join(_snake_case						)

    SCREAMING_SNAKE_CASE__         =    inspect.getsourcefile(_snake_case						)
    with open(_snake_case			,							"w"			,							encoding="UTF-8"			,							newline="\n"						) as fp:
        fp.write(_snake_case						)

    return line_to_add





def 				__a							(       __A			,							__A=False						):
    SCREAMING_SNAKE_CASE__         =    find_test_class(_snake_case						)
    if test_class:
        add_pipeline_model_mapping(_snake_case			,							overwrite=_snake_case						)


if __name__ == "__main__":
  A__										=      argparse.ArgumentParser()
  parser.add_argument(
      "--test_file", type=str, help="A path to the test file, starting with the repository's `tests` directory."
  )
  parser.add_argument(
      "--all",
      action="store_true",
      help="If to check and modify all test files.",
  )
  parser.add_argument(
      "--overwrite",
      action="store_true",
      help="If to overwrite a test class if it has already defined `pipeline_model_mapping`.",
  )
  A__										=      parser.parse_args()

  if not args.all and not args.test_file:
    raise ValueError("Please specify either `test_file` or pass `--all` to check/modify all test files.")
  elif args.all and args.test_file:
    raise ValueError("Only one of `--test_file` and `--all` could be specified.")

  A__										=      []
  if args.test_file:
    A__										=      [args.test_file]
  else:
    A__										=      os.path.join("tests", "models", "**", "test_modeling_*.py")
    for test_file in glob.glob(pattern):
      # `Flax` is not concerned at this moment
      if not test_file.startswith("test_modeling_flax_"):
        test_files.append(test_file)

  for test_file in test_files:
    if test_file in TEST_FILE_TO_IGNORE:
      print(f'''[SKIPPED] {test_file} is skipped as it is in `TEST_FILE_TO_IGNORE` in the file {__file__}.''')
      continue
    add_pipeline_model_mapping_to_test_file(test_file, overwrite=args.overwrite)








import os
import sys

from . import rsa_key_generator as rkg

UpperCAmelCase_    =128
UpperCAmelCase_    =256



def    UpperCAmelCase							(  _snake_case    ,			_snake_case = DEFAULT_BLOCK_SIZE				):
							lowerCAmelCase =		message.encode('''ascii'''				)

							lowerCAmelCase =		[]
							for block_start in range(0    ,			len(_snake_case				)    ,			_snake_case				):
														lowerCAmelCase =		0
														for i in range(_snake_case    ,			min(block_start + block_size    ,			len(_snake_case				)				)				):
																					block_int += message_bytes[i] * (BYTE_SIZE ** (i % block_size))
														block_ints.append(_snake_case				)
							return block_ints



def    UpperCAmelCase							(  _snake_case    ,			_snake_case    ,			_snake_case = DEFAULT_BLOCK_SIZE				):
							lowerCAmelCase =		[]
							for block_int in block_ints:
														lowerCAmelCase =		[]
														for i in range(block_size - 1    ,			-1    ,			-1				):
																					if len(_snake_case				) + i < message_length:
																												lowerCAmelCase =		block_int // (BYTE_SIZE**i)
																												lowerCAmelCase =		block_int % (BYTE_SIZE**i)
																												block_message.insert(0    ,			chr(_snake_case				)				)
														message.extend(_snake_case				)
							return "".join(_snake_case				)



def    UpperCAmelCase							(  _snake_case    ,			_snake_case    ,			_snake_case = DEFAULT_BLOCK_SIZE				):
							lowerCAmelCase =		[]
							lowerCAmelCase ,  lowerCAmelCase =		key

							for block in get_blocks_from_text(_snake_case    ,			_snake_case				):
														encrypted_blocks.append(pow(_snake_case    ,			_snake_case    ,			_snake_case				)				)
							return encrypted_blocks



def    UpperCAmelCase							(  _snake_case    ,			_snake_case    ,			_snake_case    ,			_snake_case = DEFAULT_BLOCK_SIZE    ,			):
							lowerCAmelCase =		[]
							lowerCAmelCase ,  lowerCAmelCase =		key
							for block in encrypted_blocks:
														decrypted_blocks.append(pow(_snake_case    ,			_snake_case    ,			_snake_case				)				)
							return get_text_from_blocks(_snake_case    ,			_snake_case    ,			_snake_case				)



def    UpperCAmelCase							(  _snake_case				):
							with open(_snake_case				) as fo:
														lowerCAmelCase =		fo.read()
							lowerCAmelCase ,  lowerCAmelCase ,  lowerCAmelCase =		content.split(''','''				)
							return (int(_snake_case				), int(_snake_case				), int(_snake_case				))



def    UpperCAmelCase							(  _snake_case    ,			_snake_case    ,			_snake_case    ,			_snake_case = DEFAULT_BLOCK_SIZE    ,			):
							lowerCAmelCase ,  lowerCAmelCase ,  lowerCAmelCase =		read_key_file(_snake_case				)
							if key_size < block_size * 8:
														sys.exit(
														    '''ERROR: Block size is {} bits and key size is {} bits. The RSA cipher '''
														    '''requires the block size to be equal to or greater than the key size. '''
														    '''Either decrease the block size or use different keys.'''.format(
														        block_size * 8    ,			_snake_case				)				)

							lowerCAmelCase =		[str(_snake_case				) for i in encrypt_message(_snake_case    ,			(n, e)    ,			_snake_case				)]

							lowerCAmelCase =		''','''.join(_snake_case				)
							lowerCAmelCase =		F"""{len(_snake_case				)}_{block_size}_{encrypted_content}"""
							with open(_snake_case    ,			'''w'''				) as fo:
														fo.write(_snake_case				)
							return encrypted_content



def    UpperCAmelCase							(  _snake_case    ,			_snake_case				):
							lowerCAmelCase ,  lowerCAmelCase ,  lowerCAmelCase =		read_key_file(_snake_case				)
							with open(_snake_case				) as fo:
														lowerCAmelCase =		fo.read()
							lowerCAmelCase ,  lowerCAmelCase ,  lowerCAmelCase =		content.split('''_'''				)
							lowerCAmelCase =		int(_snake_case				)
							lowerCAmelCase =		int(_snake_case				)

							if key_size < block_size * 8:
														sys.exit(
														    '''ERROR: Block size is {} bits and key size is {} bits. The RSA cipher '''
														    '''requires the block size to be equal to or greater than the key size. '''
														    '''Did you specify the correct key file and encrypted file?'''.format(
														        block_size * 8    ,			_snake_case				)				)

							lowerCAmelCase =		[]
							for block in encrypted_message.split(''','''				):
														encrypted_blocks.append(int(_snake_case				)				)

							return decrypt_message(_snake_case    ,			_snake_case    ,			(n, d)    ,			_snake_case				)



def    UpperCAmelCase							(  ):
							lowerCAmelCase =		'''encrypted_file.txt'''
							lowerCAmelCase =		input(R'''Encrypt\Decrypt [e\d]: '''				)

							if response.lower().startswith('''e'''				):
														lowerCAmelCase =		'''encrypt'''
							elif response.lower().startswith('''d'''				):
														lowerCAmelCase =		'''decrypt'''

							if mode == "encrypt":
														if not os.path.exists('''rsa_pubkey.txt'''				):
																					rkg.make_key_files('''rsa'''    ,			1024				)

														lowerCAmelCase =		input('''\nEnter message: '''				)
														lowerCAmelCase =		'''rsa_pubkey.txt'''
														print(F"""Encrypting and writing to {filename}..."""				)
														lowerCAmelCase =		encrypt_and_write_to_file(_snake_case    ,			_snake_case    ,			_snake_case				)

														print('''\nEncrypted text:'''				)
														print(_snake_case				)

							elif mode == "decrypt":
														lowerCAmelCase =		'''rsa_privkey.txt'''
														print(F"""Reading from {filename} and decrypting..."""				)
														lowerCAmelCase =		read_from_file_and_decrypt(_snake_case    ,			_snake_case				)
														print('''writing decryption to rsa_decryption.txt...'''				)
														with open('''rsa_decryption.txt'''    ,			'''w'''				) as dec:
																					dec.write(_snake_case				)

														print('''\nDecryption:'''				)
														print(_snake_case				)


if __name__ == "__main__":
		main()



