codingutf8 2023 the huggingface inc team licensed under the apache license version 2 0 the license you may not use this file except in compliance with the license you may obtain a copy of the license at http www apache orglicenseslicense2 0 unless required by applicable law or agreed to in writing software distributed under the license is distributed on an as is basis without warranties or conditions of any kind either express or implied see the license for the specific language governing permissions and limitations under the license the default value hfinternaltesting is for running the pipeline testing against the tiny models on the hub for debugging purpose we can specify a local path which is the outputpath argument of a previous run of utilscreatedummymodels py dynamically import the transformers module to grab the attribute classes of the processor form their names run pipeline tests for a specific task args task str a task name this should be a key in the mapping pipelinetestmapping get the canonical name adding none if empty so we can generate tests run pipeline tests for a specific task with the give model class and tokenizerprocessor class names args task str a task name this should be a key in the mapping pipelinetestmapping reponame str a model repository id on the hub modelarchitecture type a subclass of pretrainedmodel or pretrainedmodel tokenizernames liststr a list of names of a subclasses of pretrainedtokenizerfast or pretrainedtokenizer processornames liststr a list of names of subclasses of baseimageprocessor or featureextractionmixin get an instance of the corresponding class xxxpipelinetests in order to use gettestpipeline and runpipelinetest run pipeline tests for a specific task with the give model class and tokenizerprocessor class name the model will be loaded from a model repository on the hub args task str a task name this should be a key in the mapping pipelinetestmapping reponame str a model repository id on the hub modelarchitecture type a subclass of pretrainedmodel or pretrainedmodel tokenizername str the name of a subclass of pretrainedtokenizerfast or pretrainedtokenizer processorname str the name of a subclass of baseimageprocessor or featureextractionmixin if the required packages like pillow or torchaudio are not installed this will fail todo maybe not upload such problematic tiny models to hub todo we should check if a model file is on the hub repo instead validate get an instance of the corresponding class xxxpipelinetests in order to use gettestpipeline and runpipelinetest the test can disable itself but it should be very marginal concerns wav2vec2forctc without tokenizer test fasttokenizer don t exist need to copy because conversation are stateful 10 examples with batch size 4 means there needs to be a unfinished batch which is important for the unbatcher need to copy because conversation object is mutated this contains the test cases to be skipped without model architecture being involved skip some tests based on the classes or their names without the instantiated objects this is to avoid calling frompretrained so reducing the runtime if we already know the tests will fail no fix is required for this case documentquestionansweringpipelinetests requires a fast tokenizer skip some more tests based on the information from the instantiated objects no fix is required for this case if pipelinetestcassename qapipelinetests and tokenizer is not none and getattrtokenizer padtoken none is none and not tokenizer class name endswithfast qapipelinetests doesn t work with a slow tokenizer that has no pad token return true return false def validatetestcomponentstestcase task model tokenizer processor todo move this to tiny model creation script headspecific within a model type necessary changes to the config 1 for blenderbotforcausallm if model class name blenderbotforcausallm model config encodernorepeatngramsize 0 todo change the tiny model creation script don t create models with problematic tokenizers avoid indexerror in embedding layers configwithoutvocabsize canineconfig if tokenizer is not none configvocabsize getattrmodel config vocabsize none for cliplike models if configvocabsize is none if hasattrmodel config textconfig configvocabsize getattrmodel config textconfig vocabsize none elif hasattrmodel config textencoder configvocabsize getattrmodel config textencoder vocabsize none if configvocabsize is none and model config class name not in configwithoutvocabsize raise valueerror could not determine vocabsize from model configuration while tokenizer is not none coding utf 8 2023 the huggingface inc team licensed under the apache license version 2 0 the license you may not use this file except in compliance with the license you may obtain a copy of the license at http www apache org licenses license 2 0 unless required by applicable law or agreed to in writing software distributed under the license is distributed on an as is basis without warranties or conditions of any kind either express or implied see the license for the specific language governing permissions and limitations under the license the default value hf internal testing is for running the pipeline testing against the tiny models on the hub for debugging purpose we can specify a local path which is the output_path argument of a previous run of utils create_dummy_models py dynamically import the transformers module to grab the attribute classes of the processor form their names run pipeline tests for a specific task args task str a task name this should be a key in the mapping pipeline_test_mapping get the canonical name adding none if empty so we can generate tests run pipeline tests for a specific task with the give model class and tokenizer processor class names args task str a task name this should be a key in the mapping pipeline_test_mapping repo_name str a model repository id on the hub model_architecture type a subclass of pretrainedmodel or pretrainedmodel tokenizer_names list str a list of names of a subclasses of pretrainedtokenizerfast or pretrainedtokenizer processor_names list str a list of names of subclasses of baseimageprocessor or featureextractionmixin get an instance of the corresponding class xxxpipelinetests in order to use get_test_pipeline and run_pipeline_test run pipeline tests for a specific task with the give model class and tokenizer processor class name the model will be loaded from a model repository on the hub args task str a task name this should be a key in the mapping pipeline_test_mapping repo_name str a model repository id on the hub model_architecture type a subclass of pretrainedmodel or pretrainedmodel tokenizer_name str the name of a subclass of pretrainedtokenizerfast or pretrainedtokenizer processor_name str the name of a subclass of baseimageprocessor or featureextractionmixin if the required packages like pillow or torchaudio are not installed this will fail todo maybe not upload such problematic tiny models to hub todo we should check if a model file is on the hub repo instead validate get an instance of the corresponding class xxxpipelinetests in order to use get_test_pipeline and run_pipeline_test the test can disable itself but it should be very marginal concerns wav2vec2forctc without tokenizer test fasttokenizer don t exist need to copy because conversation are stateful no batching for this and it s ok 10 examples with batch size 4 means there needs to be a unfinished batch which is important for the unbatcher need to copy because conversation object is mutated this contains the test cases to be skipped without model architecture being involved skip some tests based on the classes or their names without the instantiated objects this is to avoid calling from_pretrained so reducing the runtime if we already know the tests will fail no fix is required for this case documentquestionansweringpipelinetests requires a fast tokenizer noqa skip some more tests based on the information from the instantiated objects no fix is required for this case qapipelinetests doesn t work with a slow tokenizer that has no pad token todo move this to tiny model creation script head specific within a model type necessary changes to the config 1 for blenderbotforcausallm todo change the tiny model creation script don t create models with problematic tokenizers avoid indexerror in embedding layers for clip like models
import copy
import json
import os
import random
import unittest
from pathlib import Path

from transformers.testing_utils import (
    is_pipeline_test,
    require_decord,
    require_pytesseract,
    require_timm,
    require_torch,
    require_torch_or_tf,
    require_vision,
)
from transformers.utils import direct_transformers_import, logging

from .pipelines.test_pipelines_audio_classification import AudioClassificationPipelineTests
from .pipelines.test_pipelines_automatic_speech_recognition import AutomaticSpeechRecognitionPipelineTests
from .pipelines.test_pipelines_conversational import ConversationalPipelineTests
from .pipelines.test_pipelines_depth_estimation import DepthEstimationPipelineTests
from .pipelines.test_pipelines_document_question_answering import DocumentQuestionAnsweringPipelineTests
from .pipelines.test_pipelines_feature_extraction import FeatureExtractionPipelineTests
from .pipelines.test_pipelines_fill_mask import FillMaskPipelineTests
from .pipelines.test_pipelines_image_classification import ImageClassificationPipelineTests
from .pipelines.test_pipelines_image_segmentation import ImageSegmentationPipelineTests
from .pipelines.test_pipelines_image_to_image import ImageToImagePipelineTests
from .pipelines.test_pipelines_image_to_text import ImageToTextPipelineTests
from .pipelines.test_pipelines_mask_generation import MaskGenerationPipelineTests
from .pipelines.test_pipelines_object_detection import ObjectDetectionPipelineTests
from .pipelines.test_pipelines_question_answering import QAPipelineTests
from .pipelines.test_pipelines_summarization import SummarizationPipelineTests
from .pipelines.test_pipelines_table_question_answering import TQAPipelineTests
from .pipelines.test_pipelines_text2text_generation import Text2TextGenerationPipelineTests
from .pipelines.test_pipelines_text_classification import TextClassificationPipelineTests
from .pipelines.test_pipelines_text_generation import TextGenerationPipelineTests
from .pipelines.test_pipelines_text_to_audio import TextToAudioPipelineTests
from .pipelines.test_pipelines_token_classification import TokenClassificationPipelineTests
from .pipelines.test_pipelines_translation import TranslationPipelineTests
from .pipelines.test_pipelines_video_classification import VideoClassificationPipelineTests
from .pipelines.test_pipelines_visual_question_answering import VisualQuestionAnsweringPipelineTests
from .pipelines.test_pipelines_zero_shot import ZeroShotClassificationPipelineTests
from .pipelines.test_pipelines_zero_shot_audio_classification import ZeroShotAudioClassificationPipelineTests
from .pipelines.test_pipelines_zero_shot_image_classification import ZeroShotImageClassificationPipelineTests
from .pipelines.test_pipelines_zero_shot_object_detection import ZeroShotObjectDetectionPipelineTests


pipeline_test_mapping = {
    "audio-classification": {"test": AudioClassificationPipelineTests},
    "automatic-speech-recognition": {"test": AutomaticSpeechRecognitionPipelineTests},
    "conversational": {"test": ConversationalPipelineTests},
    "depth-estimation": {"test": DepthEstimationPipelineTests},
    "document-question-answering": {"test": DocumentQuestionAnsweringPipelineTests},
    "feature-extraction": {"test": FeatureExtractionPipelineTests},
    "fill-mask": {"test": FillMaskPipelineTests},
    "image-classification": {"test": ImageClassificationPipelineTests},
    "image-segmentation": {"test": ImageSegmentationPipelineTests},
    "image-to-image": {"test": ImageToImagePipelineTests},
    "image-to-text": {"test": ImageToTextPipelineTests},
    "mask-generation": {"test": MaskGenerationPipelineTests},
    "object-detection": {"test": ObjectDetectionPipelineTests},
    "question-answering": {"test": QAPipelineTests},
    "summarization": {"test": SummarizationPipelineTests},
    "table-question-answering": {"test": TQAPipelineTests},
    "text2text-generation": {"test": Text2TextGenerationPipelineTests},
    "text-classification": {"test": TextClassificationPipelineTests},
    "text-generation": {"test": TextGenerationPipelineTests},
    "text-to-audio": {"test": TextToAudioPipelineTests},
    "token-classification": {"test": TokenClassificationPipelineTests},
    "translation": {"test": TranslationPipelineTests},
    "video-classification": {"test": VideoClassificationPipelineTests},
    "visual-question-answering": {"test": VisualQuestionAnsweringPipelineTests},
    "zero-shot": {"test": ZeroShotClassificationPipelineTests},
    "zero-shot-audio-classification": {"test": ZeroShotAudioClassificationPipelineTests},
    "zero-shot-image-classification": {"test": ZeroShotImageClassificationPipelineTests},
    "zero-shot-object-detection": {"test": ZeroShotObjectDetectionPipelineTests},
}

for task, task_info in pipeline_test_mapping.items():
    test = task_info["test"]
    task_info["mapping"] = {
        "pt": getattr(test, "model_mapping", None),
        "tf": getattr(test, "tf_model_mapping", None),
    }





TRANSFORMERS_TINY_MODEL_PATH = os.environ.get("TRANSFORMERS_TINY_MODEL_PATH", "hf-internal-testing")
if TRANSFORMERS_TINY_MODEL_PATH == "hf-internal-testing":
    TINY_MODEL_SUMMARY_FILE_PATH = os.path.join(Path(__file__).parent.parent, "tests/utils/tiny_model_summary.json")
else:
    TINY_MODEL_SUMMARY_FILE_PATH = os.path.join(TRANSFORMERS_TINY_MODEL_PATH, "reports", "tiny_model_summary.json")
with open(TINY_MODEL_SUMMARY_FILE_PATH) as fp:
    tiny_model_summary = json.load(fp)


PATH_TO_TRANSFORMERS = os.path.join(Path(__file__).parent.parent, "src/transformers")



transformers_module = direct_transformers_import(PATH_TO_TRANSFORMERS)

logger = logging.get_logger(__name__)


class PipelineTesterMixin:
    model_tester = None
    pipeline_model_mapping = None
    supported_frameworks = ["pt", "tf"]

    def run_task_tests(self, task):
        
        if task not in self.pipeline_model_mapping:
            self.skipTest(
                f"{self.__class__.__name__}::test_pipeline_{task.replace('-', '_')} is skipped: `{task}` is not in "
                f"`self.pipeline_model_mapping` for `{self.__class__.__name__}`."
            )

        model_architectures = self.pipeline_model_mapping[task]
        if not isinstance(model_architectures, tuple):
            model_architectures = (model_architectures,)
        if not isinstance(model_architectures, tuple):
            raise ValueError(f"`model_architectures` must be a tuple. Got {type(model_architectures)} instead.")

        for model_architecture in model_architectures:
            model_arch_name = model_architecture.__name__

            
            for _prefix in ["Flax", "TF"]:
                if model_arch_name.startswith(_prefix):
                    model_arch_name = model_arch_name[len(_prefix) :]
                    break

            tokenizer_names = []
            processor_names = []
            commit = None
            if model_arch_name in tiny_model_summary:
                tokenizer_names = tiny_model_summary[model_arch_name]["tokenizer_classes"]
                processor_names = tiny_model_summary[model_arch_name]["processor_classes"]
                if "sha" in tiny_model_summary[model_arch_name]:
                    commit = tiny_model_summary[model_arch_name]["sha"]
            
            tokenizer_names = [None] if len(tokenizer_names) == 0 else tokenizer_names
            processor_names = [None] if len(processor_names) == 0 else processor_names

            repo_name = f"tiny-random-{model_arch_name}"
            if TRANSFORMERS_TINY_MODEL_PATH != "hf-internal-testing":
                repo_name = model_arch_name

            self.run_model_pipeline_tests(
                task, repo_name, model_architecture, tokenizer_names, processor_names, commit
            )

    def run_model_pipeline_tests(self, task, repo_name, model_architecture, tokenizer_names, processor_names, commit):
        
        
        
        pipeline_test_class_name = pipeline_test_mapping[task]["test"].__name__

        for tokenizer_name in tokenizer_names:
            for processor_name in processor_names:
                if self.is_pipeline_test_to_skip(
                    pipeline_test_class_name,
                    model_architecture.config_class,
                    model_architecture,
                    tokenizer_name,
                    processor_name,
                ):
                    logger.warning(
                        f"{self.__class__.__name__}::test_pipeline_{task.replace('-', '_')} is skipped: test is "
                        f"currently known to fail for: model `{model_architecture.__name__}` | tokenizer "
                        f"`{tokenizer_name}` | processor `{processor_name}`."
                    )
                    continue
                self.run_pipeline_test(task, repo_name, model_architecture, tokenizer_name, processor_name, commit)

    def run_pipeline_test(self, task, repo_name, model_architecture, tokenizer_name, processor_name, commit):
        
        repo_id = f"{TRANSFORMERS_TINY_MODEL_PATH}/{repo_name}"
        if TRANSFORMERS_TINY_MODEL_PATH != "hf-internal-testing":
            model_type = model_architecture.config_class.model_type
            repo_id = os.path.join(TRANSFORMERS_TINY_MODEL_PATH, model_type, repo_name)

        tokenizer = None
        if tokenizer_name is not None:
            tokenizer_class = getattr(transformers_module, tokenizer_name)
            tokenizer = tokenizer_class.from_pretrained(repo_id, revision=commit)

        processor = None
        if processor_name is not None:
            processor_class = getattr(transformers_module, processor_name)
            
            try:
                processor = processor_class.from_pretrained(repo_id, revision=commit)
            except Exception:
                logger.warning(
                    f"{self.__class__.__name__}::test_pipeline_{task.replace('-', '_')} is skipped: Could not load the "
                    f"processor from `{repo_id}` with `{processor_name}`."
                )
                return

        
        if tokenizer is None and processor is None:
            logger.warning(
                f"{self.__class__.__name__}::test_pipeline_{task.replace('-', '_')} is skipped: Could not find or load "
                f"any tokenizer / processor from `{repo_id}`."
            )
            return

        
        try:
            model = model_architecture.from_pretrained(repo_id, revision=commit)
        except Exception:
            logger.warning(
                f"{self.__class__.__name__}::test_pipeline_{task.replace('-', '_')} is skipped: Could not find or load "
                f"the model from `{repo_id}` with `{model_architecture}`."
            )
            return

        pipeline_test_class_name = pipeline_test_mapping[task]["test"].__name__
        if self.is_pipeline_test_to_skip_more(pipeline_test_class_name, model.config, model, tokenizer, processor):
            logger.warning(
                f"{self.__class__.__name__}::test_pipeline_{task.replace('-', '_')} is skipped: test is "
                f"currently known to fail for: model `{model_architecture.__name__}` | tokenizer "
                f"`{tokenizer_name}` | processor `{processor_name}`."
            )
            return

        
        validate_test_components(self, task, model, tokenizer, processor)

        if hasattr(model, "eval"):
            model = model.eval()

        
        
        task_test = pipeline_test_mapping[task]["test"]()

        pipeline, examples = task_test.get_test_pipeline(model, tokenizer, processor)
        if pipeline is None:
            
            
            logger.warning(
                f"{self.__class__.__name__}::test_pipeline_{task.replace('-', '_')} is skipped: Could not get the "
                "pipeline for testing."
            )
            return

        task_test.run_pipeline_test(pipeline, examples)

        def run_batch_test(pipeline, examples):
            
            if pipeline.tokenizer is not None and pipeline.tokenizer.pad_token_id is None:
                return  

            
            
            def data(n):
                for _ in range(n):
                    
                    yield copy.deepcopy(random.choice(examples))

            out = []
            if task == "conversational":
                for item in pipeline(data(10), batch_size=4, max_new_tokens=5):
                    out.append(item)
            else:
                for item in pipeline(data(10), batch_size=4):
                    out.append(item)
            self.assertEqual(len(out), 10)

        run_batch_test(pipeline, examples)

    @is_pipeline_test
    def test_pipeline_audio_classification(self):
        self.run_task_tests(task="audio-classification")

    @is_pipeline_test
    def test_pipeline_automatic_speech_recognition(self):
        self.run_task_tests(task="automatic-speech-recognition")

    @is_pipeline_test
    def test_pipeline_conversational(self):
        self.run_task_tests(task="conversational")

    @is_pipeline_test
    @require_vision
    @require_timm
    @require_torch
    def test_pipeline_depth_estimation(self):
        self.run_task_tests(task="depth-estimation")

    @is_pipeline_test
    @require_pytesseract
    @require_torch
    @require_vision
    def test_pipeline_document_question_answering(self):
        self.run_task_tests(task="document-question-answering")

    @is_pipeline_test
    def test_pipeline_feature_extraction(self):
        self.run_task_tests(task="feature-extraction")

    @is_pipeline_test
    def test_pipeline_fill_mask(self):
        self.run_task_tests(task="fill-mask")

    @is_pipeline_test
    @require_torch_or_tf
    @require_vision
    def test_pipeline_image_classification(self):
        self.run_task_tests(task="image-classification")

    @is_pipeline_test
    @require_vision
    @require_timm
    @require_torch
    def test_pipeline_image_segmentation(self):
        self.run_task_tests(task="image-segmentation")

    @is_pipeline_test
    @require_vision
    def test_pipeline_image_to_text(self):
        self.run_task_tests(task="image-to-text")

    @unittest.skip(reason="`run_pipeline_test` is currently not implemented.")
    @is_pipeline_test
    @require_vision
    @require_torch
    def test_pipeline_mask_generation(self):
        self.run_task_tests(task="mask-generation")

    @is_pipeline_test
    @require_vision
    @require_timm
    @require_torch
    def test_pipeline_object_detection(self):
        self.run_task_tests(task="object-detection")

    @is_pipeline_test
    def test_pipeline_question_answering(self):
        self.run_task_tests(task="question-answering")

    @is_pipeline_test
    def test_pipeline_summarization(self):
        self.run_task_tests(task="summarization")

    @is_pipeline_test
    def test_pipeline_table_question_answering(self):
        self.run_task_tests(task="table-question-answering")

    @is_pipeline_test
    def test_pipeline_text2text_generation(self):
        self.run_task_tests(task="text2text-generation")

    @is_pipeline_test
    def test_pipeline_text_classification(self):
        self.run_task_tests(task="text-classification")

    @is_pipeline_test
    @require_torch_or_tf
    def test_pipeline_text_generation(self):
        self.run_task_tests(task="text-generation")

    @is_pipeline_test
    @require_torch
    def test_pipeline_text_to_audio(self):
        self.run_task_tests(task="text-to-audio")

    @is_pipeline_test
    def test_pipeline_token_classification(self):
        self.run_task_tests(task="token-classification")

    @is_pipeline_test
    def test_pipeline_translation(self):
        self.run_task_tests(task="translation")

    @is_pipeline_test
    @require_torch_or_tf
    @require_vision
    @require_decord
    def test_pipeline_video_classification(self):
        self.run_task_tests(task="video-classification")

    @is_pipeline_test
    @require_torch
    @require_vision
    def test_pipeline_visual_question_answering(self):
        self.run_task_tests(task="visual-question-answering")

    @is_pipeline_test
    def test_pipeline_zero_shot(self):
        self.run_task_tests(task="zero-shot")

    @is_pipeline_test
    @require_torch
    def test_pipeline_zero_shot_audio_classification(self):
        self.run_task_tests(task="zero-shot-audio-classification")

    @is_pipeline_test
    @require_vision
    def test_pipeline_zero_shot_image_classification(self):
        self.run_task_tests(task="zero-shot-image-classification")

    @is_pipeline_test
    @require_vision
    @require_torch
    def test_pipeline_zero_shot_object_detection(self):
        self.run_task_tests(task="zero-shot-object-detection")

    
    def is_pipeline_test_to_skip(
        self, pipeline_test_casse_name, config_class, model_architecture, tokenizer_name, processor_name
    ):
        
        
        if (
            pipeline_test_casse_name == "DocumentQuestionAnsweringPipelineTests"
            and tokenizer_name is not None
            and not tokenizer_name.endswith("Fast")
        ):
            
            return True

        return False

    def is_pipeline_test_to_skip_more(self, pipeline_test_casse_name, config, model, tokenizer, processor):  
        
        
        if (
            pipeline_test_casse_name == "QAPipelineTests"
            and tokenizer is not None
            and getattr(tokenizer, "pad_token", None) is None
            and not tokenizer.__class__.__name__.endswith("Fast")
        ):
            
            return True

        return False


def validate_test_components(test_case, task, model, tokenizer, processor):
    
    
    
    if model.__class__.__name__ == "BlenderbotForCausalLM":
        model.config.encoder_no_repeat_ngram_size = 0

    
    
    CONFIG_WITHOUT_VOCAB_SIZE = ["CanineConfig"]
    if tokenizer is not None:
        config_vocab_size = getattr(model.config, "vocab_size", None)
        
        if config_vocab_size is None:
            if hasattr(model.config, "text_config"):
                config_vocab_size = getattr(model.config.text_config, "vocab_size", None)
            elif hasattr(model.config, "text_encoder"):
                config_vocab_size = getattr(model.config.text_encoder, "vocab_size", None)

        if config_vocab_size is None and model.config.__class__.__name__ not in CONFIG_WITHOUT_VOCAB_SIZE:
            raise ValueError(
                "Could not determine `vocab_size` from model configuration while `tokenizer` is not `None`."
            )