



"""simple docstring"""




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


_UpperCamelCase:					Dict     				=	{}
for task, _ in pipeline_test_mapping.items():
     _UpperCamelCase:					Any     				=	{'pt': None, 'tf': None}


# DO **NOT** add item to this set (unless the reason is approved)
_UpperCamelCase:					List[str]     				=	{
    'tests/models/esm/test_modeling_esmfold.py',  # The pipeline test mapping is added to `test_modeling_esm.py`
}


def 	lowercase__  (      _UpperCAmelCase					)  ->      Tuple:






      '''simple docstring'''


      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 	lowercase__  (      _UpperCAmelCase						,   _UpperCAmelCase					)  ->      Union[str, Any]:






      '''simple docstring'''


      if PIPELINE_TEST_MAPPING[task].get(_UpperCAmelCase						,   _UpperCAmelCase					) is not None:
            return PIPELINE_TEST_MAPPING[task][framework]

      lowercase	:   Optional[int]											= pipeline_test_mapping[task]['test']
      lowercase	:   Optional[Any]											= None

      if framework == "pt":
            lowercase	:   List[str]											= getattr(_UpperCAmelCase						,   'model_mapping'						,   _UpperCAmelCase					)
      elif framework == "tf":
            lowercase	:   Optional[Any]											= getattr(_UpperCAmelCase						,   'tf_model_mapping'						,   _UpperCAmelCase					)

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

      # cache the results
      lowercase	:   List[str]											= mapping
      return mapping


def 	lowercase__  (      _UpperCAmelCase						,   _UpperCAmelCase					)  ->      Dict:






      '''simple docstring'''


      lowercase	:   Dict											= get_framework(_UpperCAmelCase					)
      if framework is None:
            return None
      lowercase	:   Union[str, Any]											= get_mapping_for_task(_UpperCAmelCase						,   _UpperCAmelCase					)
      if mapping is None:
            return None

      lowercase	:   List[str]											= list({model_class.config_class for model_class in test_class.all_model_classes}					)
      if len(_UpperCAmelCase					) != 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.
      lowercase	:   Union[str, Any]											= mapping.get(config_classes[0]						,   _UpperCAmelCase					)
      if isinstance(_UpperCAmelCase						,   (tuple, list)					):
            lowercase	:   Optional[int]											= sorted(_UpperCAmelCase						,   key=lambda _UpperCAmelCase					: x.__name__					)

      return model_class


def 	lowercase__  (      _UpperCAmelCase					)  ->      int:






      '''simple docstring'''


      lowercase	:   Union[str, Any]											= [(task, get_model_for_pipeline_test(_UpperCAmelCase						,   _UpperCAmelCase					)) for task in pipeline_test_mapping]
      lowercase	:   List[str]											= sorted([(task, model) for task, model in mapping if model is not None]						,   key=lambda _UpperCAmelCase					: x[0]					)

      return dict(_UpperCAmelCase					)


def 	lowercase__  (      _UpperCAmelCase					)  ->      Tuple:






      '''simple docstring'''


      lowercase	:   Optional[Any]											= get_framework(_UpperCAmelCase					)
      if framework == "pt":
            lowercase	:   Tuple											= 'torch'
      lowercase	:   Optional[Any]											= '{}'

      lowercase	:   List[Any]											= get_pipeline_model_mapping(_UpperCAmelCase					)
      if len(_UpperCAmelCase					) == 0:
            return ""

      lowercase	:   Union[str, Any]											= []
      for task, model_classes in mapping.items():
            if isinstance(_UpperCAmelCase						,   (tuple, list)					):
                  # A list/tuple of model classes
                  lowercase	:   Optional[int]											= '(' + ', '.join([x.__name__ for x in model_classes]					) + ')'
            else:
                  # A single model class
                  lowercase	:   Tuple											= model_classes.__name__
            texts.append(f'''"{task}": {value}'''					)
      lowercase	:   Optional[Any]											= '{' + ', '.join(_UpperCAmelCase					) + '}'
      lowercase	:   Any											= f'''pipeline_model_mapping = {text} if is_{framework}_available() else {default_value}'''

      return text


def 	lowercase__  (      _UpperCAmelCase					)  ->      Tuple:






      '''simple docstring'''


      lowercase	:   List[str]											= {'ModelTesterMixin', 'TFModelTesterMixin', 'FlaxModelTesterMixin'}
      if not issubclass(_UpperCAmelCase						,   unittest.TestCase					):
            return False
      return len(base_class_names.intersection([x.__name__ for x in test_class.__bases__]					)					) > 0


def 	lowercase__  (      _UpperCAmelCase					)  ->      List[str]:






      '''simple docstring'''


      lowercase	:   int											= [x for x in get_test_classes(_UpperCAmelCase					) if is_valid_test_class(_UpperCAmelCase					)]

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

      return target_test_class


def 	lowercase__  (      _UpperCAmelCase						,   _UpperCAmelCase						,   _UpperCAmelCase					)  ->      Tuple:






      '''simple docstring'''


      lowercase	:   List[str]											= start_idx
      for idx, line in enumerate(lines[start_idx:]					):
            lowercase	:   Optional[int]											= len(_UpperCAmelCase					) - len(line.lstrip()					)
            if idx == 0 or indent > indent_level or (indent == indent_level and line.strip() == ")"):
                  lowercase	:   Any											= start_idx + idx
            elif idx > 0 and indent <= indent_level:
                  # Outside the definition block of `pipeline_model_mapping`
                  break

      return end_idx


def 	lowercase__  (      _UpperCAmelCase						,   _UpperCAmelCase=False					)  ->      Dict:






      '''simple docstring'''


      if getattr(_UpperCAmelCase						,   'pipeline_model_mapping'						,   _UpperCAmelCase					) is not None:
            if not overwrite:
                  return "", -1

      lowercase	:   Tuple											= get_pipeline_model_mapping_string(_UpperCAmelCase					)
      if len(_UpperCAmelCase					) == 0:
            return "", -1
      lowercase	:   int											= line_to_add + '\n'

      # The code defined the class `test_class`
      lowercase						,     lowercase	:   Dict											= inspect.getsourcelines(_UpperCAmelCase					)
      # `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(_UpperCAmelCase					):
            if line.lstrip().startswith('class '					):
                  lowercase	:   List[str]											= class_lines[idx:]
                  class_start_line_no += idx
                  break
      lowercase	:   Optional[int]											= class_start_line_no + len(_UpperCAmelCase					) - 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`.
      lowercase	:   List[str]											= None
      # The indent level of the line at `class_lines[start_idx]` (if defined)
      lowercase	:   Optional[int]											= 0
      # To record if `pipeline_model_mapping` is found in `test_class`.
      lowercase	:   List[str]											= None
      for idx, line in enumerate(_UpperCAmelCase					):
            if line.strip().startswith('all_model_classes = '					):
                  lowercase	:   Any											= len(_UpperCAmelCase					) - len(line.lstrip()					)
                  lowercase	:   Dict											= idx
            elif line.strip().startswith('all_generative_model_classes = '					):
                  lowercase	:   Tuple											= len(_UpperCAmelCase					) - len(line.lstrip()					)
                  lowercase	:   Dict											= idx
            elif line.strip().startswith('pipeline_model_mapping = '					):
                  lowercase	:   str											= len(_UpperCAmelCase					) - len(line.lstrip()					)
                  lowercase	:   Union[str, Any]											= idx
                  lowercase	:   Optional[int]											= line
                  break

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

      # 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
      lowercase	:   Any											= re.compile(R'\s(is_\S+?_available\(\))\s'					)
      for line in class_lines[start_idx : end_idx + 1]:
            lowercase	:   List[str]											= r.search(_UpperCAmelCase					)
            if backend_condition is not None:
                  # replace the leading and trailing whitespaces to the space character " ".
                  lowercase	:   Union[str, Any]											= ' ' + backend_condition[0][1:-1] + ' '
                  lowercase	:   Optional[Any]											= r.sub(_UpperCAmelCase						,   _UpperCAmelCase					)
                  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`.
            lowercase	:   int											= end_idx
      else:
            # `pipeline_model_mapping` is defined. The target index is set to be one **BEFORE** its start index.
            lowercase	:   int											= start_idx - 1
            # mark the lines of the currently existing `pipeline_model_mapping` to be removed.
            for idx in range(_UpperCAmelCase						,   end_idx + 1					):
                  # These lines are going to be removed before writing to the test file.
                  lowercase	:   Union[str, Any]											= None  # noqa

    # Make sure the test class is a subclass of `PipelineTesterMixin`.
      lowercase	:   List[Any]											= [x.__name__ for x in test_class.__bases__]
      if "PipelineTesterMixin" not in parent_classes:
            # Put `PipelineTesterMixin` just before `unittest.TestCase`
            lowercase	:   int											= [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'					)
            lowercase	:   Dict											= ', '.join(_parent_classes					)
            for idx, line in enumerate(_UpperCAmelCase					):
                  # 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					):
                              lowercase	:   str											= None  # noqa
                        break
        # Add the new, one-line, class declaration for `test_class`
            lowercase	:   Any											= f'''class {test_class.__name__}({parent_classes}):\n'''

      # Add indentation
      lowercase	:   Optional[int]											= ' ' * indent_level + line_to_add
      # Insert `pipeline_model_mapping` to `class_lines`.
      # (The line at `target_idx` should be kept by definition!)
      lowercase	:   Optional[int]											= class_lines[: target_idx + 1] + [line_to_add] + class_lines[target_idx + 1 :]
      # Remove the lines that are marked to be removed
      lowercase	:   List[Any]											= [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)
      lowercase	:   Optional[Any]											= inspect.getsourcelines(inspect.getmodule(_UpperCAmelCase					)					)[0]
      # Be careful with the 1-off between line numbers and array indices
      lowercase	:   Any											= module_lines[: class_start_line_no - 1] + class_lines + module_lines[class_end_line_no:]
      lowercase	:   Dict											= ''.join(_UpperCAmelCase					)

      lowercase	:   Optional[int]											= inspect.getsourcefile(_UpperCAmelCase					)
      with open(_UpperCAmelCase						,   'w'						,   encoding='UTF-8'						,   newline='\n'					) as fp:
            fp.write(_UpperCAmelCase					)

      return line_to_add


def 	lowercase__  (      _UpperCAmelCase						,   _UpperCAmelCase=False					)  ->      Any:






      '''simple docstring'''


      lowercase	:   str											= find_test_class(_UpperCAmelCase					)
      if test_class:
            add_pipeline_model_mapping(_UpperCAmelCase						,   overwrite=_UpperCAmelCase					)


if __name__ == "__main__":
     _UpperCamelCase:					int     				=	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`.',
     )
     _UpperCamelCase:					Tuple     				=	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.')

     _UpperCamelCase:					List[Any]     				=	[]
     if args.test_file:
          _UpperCamelCase:					Optional[int]     				=	[args.test_file]
     else:
          _UpperCamelCase:					Optional[int]     				=	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)





"""simple docstring"""




import unittest
from pathlib import Path

from transformers import is_vision_available, load_tool
from transformers.testing_utils import get_tests_dir

from .test_tools_common import ToolTesterMixin


if is_vision_available():
     from PIL import Image


class 	a__		(    unittest.TestCase,	SCREAMING_SNAKE_CASE__  ):


   def 							lowercase					(				self   :				Dict      )     ->							List[str]:
         lowercase	:   str											= load_tool('image-captioning'      )
         self.tool.setup()
         lowercase	:   List[str]											= load_tool('image-captioning',  remote=lowerCAmelCase      )


   def 							lowercase					(				self   :				str      )     ->							int:
         lowercase	:   Union[str, Any]											= Image.open(Path(get_tests_dir('fixtures/tests_samples/COCO'      )      ) / '000000039769.png'      )
         lowercase	:   List[str]											= self.tool(lowerCAmelCase      )
         self.assertEqual(lowerCAmelCase,  'two cats sleeping on a couch'      )


   def 							lowercase					(				self   :				str      )     ->							List[str]:
         lowercase	:   str											= Image.open(Path(get_tests_dir('fixtures/tests_samples/COCO'      )      ) / '000000039769.png'      )
         lowercase	:   Tuple											= self.remote_tool(lowerCAmelCase      )
         self.assertEqual(lowerCAmelCase,  'two cats sleeping on a couch'      )


   def 							lowercase					(				self   :				Tuple      )     ->							Union[str, Any]:
         lowercase	:   Tuple											= Image.open(Path(get_tests_dir('fixtures/tests_samples/COCO'      )      ) / '000000039769.png'      )
         lowercase	:   Union[str, Any]											= self.tool(image=lowerCAmelCase      )
         self.assertEqual(lowerCAmelCase,  'two cats sleeping on a couch'      )






   def 							lowercase					(				self   :				str      )     ->							Optional[int]:
         lowercase	:   List[str]											= Image.open(Path(get_tests_dir('fixtures/tests_samples/COCO'      )      ) / '000000039769.png'      )
         lowercase	:   str											= self.remote_tool(image=lowerCAmelCase      )
         self.assertEqual(lowerCAmelCase,  'two cats sleeping on a couch'      )
