
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


lowercase						:      List[str]							=							{}
for task, _ in pipeline_test_mapping.items():
  lowercase						:      List[Any]							=							{'pt': None, 'tf': None}


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







def 		_snake_case(						SCREAMING_SNAKE_CASE__       )				->					int:







       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 		_snake_case(						SCREAMING_SNAKE_CASE__      ,			SCREAMING_SNAKE_CASE__       )				->					Optional[int]:







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

       lowercase							:  Dict 			=   pipeline_test_mapping[task]["""test"""]
       lowercase							:  Tuple 			=   None

       if framework == "pt":
              lowercase							:  int 			=   getattr(lowercase_      ,			"""model_mapping"""      ,			lowercase_       )
       elif framework == "tf":
              lowercase							:  Union[str, Any] 			=   getattr(lowercase_      ,			"""tf_model_mapping"""      ,			lowercase_       )

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

       # cache the results
       lowercase							:  Optional[Any] 			=   mapping
       return mapping







def 		_snake_case(						SCREAMING_SNAKE_CASE__      ,			SCREAMING_SNAKE_CASE__       )				->					Tuple:
       lowercase							:  Dict 			=   get_framework(lowercase_       )
       if framework is None:
              return None
       lowercase							:  Dict 			=   get_mapping_for_task(lowercase_      ,			lowercase_       )
       if mapping is None:
              return None

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

       return model_class







def 		_snake_case(						SCREAMING_SNAKE_CASE__       )				->					str:
       lowercase							:  List[Any] 			=   [(task, get_model_for_pipeline_test(lowercase_      ,			lowercase_       )) for task in pipeline_test_mapping]
       lowercase							:  Optional[int] 			=   sorted([(task, model) for task, model in mapping if model is not None]      ,			key=lambda SCREAMING_SNAKE_CASE__       : x[0]       )

       return dict(lowercase_       )







def 		_snake_case(						SCREAMING_SNAKE_CASE__       )				->					Union[str, Any]:
       lowercase							:  Dict 			=   get_framework(lowercase_       )
       if framework == "pt":
              lowercase							:  int 			=   """torch"""
       lowercase							:  str 			=   """{}"""

       lowercase							:  List[str] 			=   get_pipeline_model_mapping(lowercase_       )
       if len(lowercase_       ) == 0:
              return ""

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

       return text







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







def 		_snake_case(						SCREAMING_SNAKE_CASE__       )				->					Optional[int]:
       lowercase							:  Any 			=   [x for x in get_test_classes(lowercase_       ) if is_valid_test_class(lowercase_       )]

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

       return target_test_class







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

       return end_idx







def 		_snake_case(						SCREAMING_SNAKE_CASE__      ,			SCREAMING_SNAKE_CASE__=False       )				->					Optional[int]:







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

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

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

       if start_idx is None:
              return "", -1
       # Find the ending index (inclusive) of the above found block.
       lowercase							:  List[Any] 			=   find_block_ending(lowercase_      ,			lowercase_      ,			lowercase_       )

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

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

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

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

       return line_to_add







def 		_snake_case(						SCREAMING_SNAKE_CASE__      ,			SCREAMING_SNAKE_CASE__=False       )				->					str:
       lowercase							:  List[str] 			=   find_test_class(lowercase_       )
       if test_class:
              add_pipeline_model_mapping(lowercase_      ,			overwrite=lowercase_       )


if __name__ == "__main__":
  lowercase						:      List[str]							=							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`.""",
  )
  lowercase						:      Union[str, Any]							=							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.""")

  lowercase						:      Optional[Any]							=							[]
  if args.test_file:
    lowercase						:      List[Any]							=							[args.test_file]
  else:
    lowercase						:      Union[str, Any]							=							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 unittest
from unittest.util import safe_repr

from transformers import AutoTokenizer, RwkvConfig, is_torch_available
from transformers.testing_utils import require_torch, slow, torch_device

from ...generation.test_utils import GenerationTesterMixin
from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, floats_tensor, ids_tensor, random_attention_mask
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
  import torch

  from transformers import (
      RWKV_PRETRAINED_MODEL_ARCHIVE_LIST,
      RwkvForCausalLM,
      RwkvModel,
  )
  from transformers.pytorch_utils import is_torch_greater_or_equal_than_2_0
else:
  lowercase						:      Any							=							False

class __snake_case	:
    def __init__(						self	,snake_case	,snake_case=14	,snake_case=7	,snake_case=True	,snake_case=False	,snake_case=True	,snake_case=True	,snake_case=True	,snake_case=99	,snake_case=32	,snake_case=5	,snake_case=37	,snake_case="gelu"	,snake_case=0.1	,snake_case=0.1	,snake_case=512	,snake_case=16	,snake_case=2	,snake_case=3	,snake_case=4	,snake_case=None	,):



           '''simple docstring'''







           lowercase							:  Optional[Any] 			=   parent
           lowercase							:  str 			=   batch_size
           lowercase							:  str 			=   seq_length
           lowercase							:  int 			=   is_training
           lowercase							:  Optional[Any] 			=   use_token_type_ids
           lowercase							:  List[str] 			=   use_input_mask
           lowercase							:  List[Any] 			=   use_labels
           lowercase							:  List[Any] 			=   use_mc_token_ids
           lowercase							:  int 			=   vocab_size
           lowercase							:  Any 			=   hidden_size
           lowercase							:  Optional[int] 			=   num_hidden_layers
           lowercase							:  Optional[int] 			=   intermediate_size
           lowercase							:  Optional[int] 			=   hidden_act
           lowercase							:  Optional[int] 			=   hidden_dropout_prob
           lowercase							:  List[str] 			=   attention_probs_dropout_prob
           lowercase							:  Dict 			=   max_position_embeddings
           lowercase							:  Optional[Any] 			=   type_vocab_size
           lowercase							:  str 			=   type_sequence_label_size
           lowercase							:  int 			=   num_labels
           lowercase							:  List[Any] 			=   num_choices
           lowercase							:  int 			=   scope
           lowercase							:  Any 			=   vocab_size - 1
           lowercase							:  str 			=   vocab_size - 1
           lowercase							:  int 			=   vocab_size - 1
    def    _SCREAMING_SNAKE_CASE (						self   ):



           '''simple docstring'''







           return RwkvConfig.from_pretrained("""sgugger/rwkv-4-pile-7b"""   )
    def    _SCREAMING_SNAKE_CASE (						self	,snake_case=False	,snake_case=False	,snake_case=False   ):



           '''simple docstring'''







           lowercase							:  Optional[Any] 			=   ids_tensor([self.batch_size, self.seq_length]	,self.vocab_size   )

           lowercase							:  Tuple 			=   None
           if self.use_input_mask:
                  lowercase							:  List[str] 			=   random_attention_mask([self.batch_size, self.seq_length]   )

           lowercase							:  List[Any] 			=   None
           if self.use_token_type_ids:
                  lowercase							:  str 			=   ids_tensor([self.batch_size, self.seq_length]	,self.type_vocab_size   )

           lowercase							:  Any 			=   None
           if self.use_mc_token_ids:
                  lowercase							:  Tuple 			=   ids_tensor([self.batch_size, self.num_choices]	,self.seq_length   )

           lowercase							:  Union[str, Any] 			=   None
           lowercase							:  List[Any] 			=   None
           lowercase							:  Tuple 			=   None
           if self.use_labels:
                  lowercase							:  int 			=   ids_tensor([self.batch_size]	,self.type_sequence_label_size   )
                  lowercase							:  str 			=   ids_tensor([self.batch_size, self.seq_length]	,self.num_labels   )
                  lowercase							:  Tuple 			=   ids_tensor([self.batch_size]	,self.num_choices   )

           lowercase							:  List[str] 			=   self.get_config(
               gradient_checkpointing=snake_case	,scale_attn_by_inverse_layer_idx=snake_case	,reorder_and_upcast_attn=snake_case	,)

           return (
               config,
               input_ids,
               input_mask,
               None,
               token_type_ids,
               mc_token_ids,
               sequence_labels,
               token_labels,
               choice_labels,
           )
    def    _SCREAMING_SNAKE_CASE (						self	,snake_case=False	,snake_case=False	,snake_case=False   ):



           '''simple docstring'''







           return RwkvConfig(
               vocab_size=self.vocab_size	,hidden_size=self.hidden_size	,num_hidden_layers=self.num_hidden_layers	,intermediate_size=self.intermediate_size	,activation_function=self.hidden_act	,resid_pdrop=self.hidden_dropout_prob	,attn_pdrop=self.attention_probs_dropout_prob	,n_positions=self.max_position_embeddings	,type_vocab_size=self.type_vocab_size	,use_cache=snake_case	,bos_token_id=self.bos_token_id	,eos_token_id=self.eos_token_id	,pad_token_id=self.pad_token_id	,gradient_checkpointing=snake_case	,scale_attn_by_inverse_layer_idx=snake_case	,reorder_and_upcast_attn=snake_case	,)
    def    _SCREAMING_SNAKE_CASE (						self   ):



           '''simple docstring'''







           lowercase							:  List[Any] 			=   self.get_config()
           lowercase							:  str 			=   300
           return config
    def    _SCREAMING_SNAKE_CASE (						self   ):



           '''simple docstring'''







           (
               (
               lowercase
           )			,    (
               lowercase
           )			,    (
               lowercase
           )			,    (
               lowercase
           )			,    (
               lowercase
           )			,    (
               lowercase
           )			,    (
               lowercase
           )			,    (
               lowercase
           )			,    (
               lowercase
           )			,    
           )							:  Dict 			=   self.prepare_config_and_inputs()

           lowercase							:  List[str] 			=   floats_tensor([self.batch_size, self.seq_length, self.hidden_size]   )
           lowercase							:  Tuple 			=   ids_tensor([self.batch_size, self.seq_length]	,vocab_size=2   )

           return (
               config,
               input_ids,
               input_mask,
               head_mask,
               token_type_ids,
               sequence_labels,
               token_labels,
               choice_labels,
               encoder_hidden_states,
               encoder_attention_mask,
           )
    def    _SCREAMING_SNAKE_CASE (						self	,snake_case	,snake_case	,snake_case	,snake_case	,snake_case	,*snake_case   ):



           '''simple docstring'''







           lowercase							:  Optional[Any] 			=   True
           lowercase							:  Dict 			=   RwkvModel(config=snake_case   )
           model.to(snake_case   )
           model.eval()

           lowercase							:  Optional[Any] 			=   model(snake_case   )

           self.parent.assertEqual(result.last_hidden_state.shape	,(self.batch_size, self.seq_length, self.hidden_size)   )
           self.parent.assertEqual(len(result.hidden_states   )	,config.num_hidden_layers + 1   )
    def    _SCREAMING_SNAKE_CASE (						self	,snake_case	,snake_case	,snake_case	,snake_case	,snake_case	,*snake_case   ):



           '''simple docstring'''







           lowercase							:  int 			=   RwkvForCausalLM(snake_case   )
           model.to(snake_case   )
           model.eval()

           lowercase							:  Optional[int] 			=   model(snake_case	,labels=snake_case   )
           self.parent.assertEqual(result.loss.shape	,()   )
           self.parent.assertEqual(result.logits.shape	,(self.batch_size, self.seq_length, self.vocab_size)   )
    def    _SCREAMING_SNAKE_CASE (						self	,snake_case	,snake_case	,snake_case	,snake_case	,snake_case	,*snake_case   ):



           '''simple docstring'''







           lowercase							:  str 			=   RwkvModel(config=snake_case   )
           model.to(snake_case   )
           model.eval()

           lowercase							:  int 			=   model(snake_case   )
           lowercase							:  int 			=   outputs.last_hidden_state

           lowercase							:  Dict 			=   model(input_ids[:, :2]   )
           lowercase							:  Optional[int] 			=   outputs.last_hidden_state

           # Using the state computed on the first inputs, we will get the same output
           lowercase							:  int 			=   model(input_ids[:, 2:]	,state=outputs.state   )
           lowercase							:  Union[str, Any] 			=   outputs.last_hidden_state

           self.parent.assertTrue(torch.allclose(torch.cat([output_one, output_two]	,dim=1   )	,snake_case	,atol=1e-5   )   )
    def    _SCREAMING_SNAKE_CASE (						self	,snake_case	,snake_case	,snake_case	,snake_case	,snake_case	,*snake_case	,snake_case=False   ):



           '''simple docstring'''







           lowercase							:  Dict 			=   RwkvForCausalLM(snake_case   )
           model.to(snake_case   )
           if gradient_checkpointing:
                  model.gradient_checkpointing_enable()

           lowercase							:  List[str] 			=   model(snake_case	,labels=snake_case   )
           self.parent.assertEqual(result.loss.shape	,()   )
           self.parent.assertEqual(result.logits.shape	,(self.batch_size, self.seq_length, self.vocab_size)   )
           result.loss.backward()
    def    _SCREAMING_SNAKE_CASE (						self   ):



           '''simple docstring'''







           lowercase							:  List[Any] 			=   self.prepare_config_and_inputs()

           (
               (
               lowercase
           )			,    (
               lowercase
           )			,    (
               lowercase
           )			,    (
               lowercase
           )			,    (
               lowercase
           )			,    (
               lowercase
           )			,    (
               lowercase
           )			,    (
               lowercase
           )			,    (
               lowercase
           )			,    
           )							:  Union[str, Any] 			=   config_and_inputs

           lowercase							:  Optional[Any] 			=   {"""input_ids""": input_ids}

           return config, inputs_dict






@unittest.skipIf(
    not is_torch_greater_or_equal_than_2_0      ,			reason="See https://github.com/huggingface/transformers/pull/24204"  )
@require_torch
class __snake_case	(   lowerCAmelCase      ,			lowerCAmelCase      ,			lowerCAmelCase      ,			unittest.TestCase  ):
    _a	:		str=       (RwkvModel, RwkvForCausalLM) if is_torch_available() else ()
    _a	:		Tuple=       (
        {"feature-extraction": RwkvModel, "text-generation": RwkvForCausalLM} if is_torch_available() else {}
    )
    # all_generative_model_classes = (RwkvForCausalLM,) if is_torch_available() else ()
    _a	:		List[str]=       False
    _a	:		int=       False
    _a	:		Optional[Any]=       False
    _a	:		List[Any]=       False
    _a	:		int=       False  # Rwkv does not support head masking
    def    _SCREAMING_SNAKE_CASE (						self   ):



           '''simple docstring'''







           lowercase							:  Dict 			=   RwkvModelTester(self   )
           lowercase							:  Any 			=   ConfigTester(
               self	,config_class=snake_case	,n_embd=37	,common_properties=["""hidden_size""", """num_hidden_layers"""]   )
    def    _SCREAMING_SNAKE_CASE (						self	,snake_case	,snake_case	,snake_case=None   ):



           '''simple docstring'''







           if isinstance(snake_case	,torch.Tensor   ):
                  lowercase			,    lowercase							:  Tuple 			=   member.max().item(), member.min().item()
           elif isinstance(snake_case	,snake_case   ) or isinstance(snake_case	,snake_case   ):
                  lowercase			,    lowercase							:  Dict 			=   max(snake_case   ), min(snake_case   )

           if not isinstance(snake_case	,snake_case   ):
                  raise TypeError("""container should be a list or tuple"""   )
           elif len(snake_case   ) != 2:
                  raise ValueError("""container should have 2 elements"""   )

           lowercase			,    lowercase							:  int 			=   container

           lowercase							:  Union[str, Any] 			=   (min_value >= expected_min) and (max_value <= expected_max)

           if not is_inside_interval:
                  lowercase							:  Optional[int] 			=   """%s not found in %s""" % (safe_repr(snake_case   ), safe_repr(snake_case   ))
                  self.fail(self._formatMessage(snake_case	,snake_case   )   )
    def    _SCREAMING_SNAKE_CASE (						self   ):



           '''simple docstring'''







           self.config_tester.run_common_tests()
    def    _SCREAMING_SNAKE_CASE (						self   ):



           '''simple docstring'''







           lowercase							:  Any 			=   self.model_tester.prepare_config_and_inputs()
           self.model_tester.create_and_check_rwkv_model(*snake_case   )
    def    _SCREAMING_SNAKE_CASE (						self   ):



           '''simple docstring'''







           lowercase							:  Any 			=   self.model_tester.prepare_config_and_inputs()
           self.model_tester.create_and_check_causl_lm(*snake_case   )
    def    _SCREAMING_SNAKE_CASE (						self   ):



           '''simple docstring'''







           lowercase							:  Any 			=   self.model_tester.prepare_config_and_inputs()
           self.model_tester.create_and_check_state_equivalency(*snake_case   )
    def    _SCREAMING_SNAKE_CASE (						self   ):



           '''simple docstring'''







           lowercase			,    lowercase							:  Union[str, Any] 			=   self.model_tester.prepare_config_and_inputs_for_common()

           for model_class in self.all_model_classes:
                  lowercase							:  List[Any] 			=   model_class(config=snake_case   )
                  for name, param in model.named_parameters():
                         if "time_decay" in name:
                                if param.requires_grad:
                                       self.assertTrue(param.data.max().item() == 3.0   )
                                       self.assertTrue(param.data.min().item() == -5.0   )
                         elif "time_first" in name:
                                if param.requires_grad:
                                       # check if it's a ones like
                                       self.assertTrue(torch.allclose(param.data	,torch.ones_like(param.data   )	,atol=1e-5	,rtol=1e-5   )   )
                         elif any(x in name for x in ["""time_mix_key""", """time_mix_receptance"""]   ):
                                if param.requires_grad:
                                       self.assertInterval(
                                           param.data	,[0.0, 1.0]	,msg=f"Parameter {name} of model {model_class} seems not properly initialized"	,)
                         elif "time_mix_value" in name:
                                if param.requires_grad:
                                       self.assertInterval(
                                           param.data	,[0.0, 1.3]	,msg=f"Parameter {name} of model {model_class} seems not properly initialized"	,)
    def    _SCREAMING_SNAKE_CASE (						self   ):



           '''simple docstring'''







           lowercase			,    lowercase							:  str 			=   self.model_tester.prepare_config_and_inputs_for_common()
           lowercase							:  List[Any] 			=   True

           lowercase							:  int 			=   getattr(self.model_tester	,"""seq_length"""	,snake_case   )

           for model_class in self.all_model_classes:
                  lowercase							:  Union[str, Any] 			=   True
                  lowercase							:  str 			=   False
                  lowercase							:  int 			=   True
                  lowercase							:  List[str] 			=   model_class(snake_case   )
                  model.to(snake_case   )
                  model.eval()

                  lowercase							:  Tuple 			=   self._prepare_for_class(snake_case	,snake_case   )
                  lowercase							:  Optional[Any] 			=   inputs["""input_ids"""].shape[0]
                  with torch.no_grad():
                         lowercase							:  Any 			=   model(**snake_case   )
                  lowercase							:  List[Any] 			=   outputs.attentions
                  self.assertEqual(len(snake_case   )	,self.model_tester.num_hidden_layers   )

                  # check that output_attentions also work using config
                  del inputs_dict["output_attentions"]
                  lowercase							:  str 			=   True
                  lowercase							:  Any 			=   model_class(snake_case   )
                  model.to(snake_case   )
                  model.eval()

                  lowercase							:  Optional[int] 			=   self._prepare_for_class(snake_case	,snake_case   )
                  lowercase							:  Tuple 			=   inputs["""input_ids"""].shape[0]
                  with torch.no_grad():
                         lowercase							:  List[Any] 			=   model(**snake_case   )
                  lowercase							:  str 			=   outputs.attentions
                  self.assertEqual(len(snake_case   )	,self.model_tester.num_hidden_layers   )

                  self.assertListEqual(
                      list(attentions[0].shape[-3:]   )	,[batch_size, seq_len, config.hidden_size]	,)
                  lowercase							:  Dict 			=   len(snake_case   )

                  # Check attention is always last and order is fine
                  lowercase							:  Any 			=   True
                  lowercase							:  List[str] 			=   True
                  lowercase							:  Optional[Any] 			=   model_class(snake_case   )
                  model.to(snake_case   )
                  model.eval()

                  lowercase							:  Optional[int] 			=   self._prepare_for_class(snake_case	,snake_case   )
                  lowercase							:  Optional[Any] 			=   inputs["""input_ids"""].shape[0]
                  with torch.no_grad():
                         lowercase							:  Union[str, Any] 			=   model(**snake_case   )

                  lowercase							:  Tuple 			=   1
                  self.assertEqual(out_len + added_hidden_states	,len(snake_case   )   )

                  lowercase							:  Optional[Any] 			=   outputs.attentions

                  self.assertEqual(len(snake_case   )	,self.model_tester.num_hidden_layers   )
                  self.assertListEqual(
                      list(self_attentions[0].shape[-3:]   )	,[batch_size, seq_len, config.hidden_size]	,)
    @slow
    def    _SCREAMING_SNAKE_CASE (						self   ):



           '''simple docstring'''







           for model_name in RWKV_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
                  lowercase							:  Dict 			=   RwkvModel.from_pretrained(snake_case   )
                  self.assertIsNotNone(snake_case   )






@unittest.skipIf(
    not is_torch_greater_or_equal_than_2_0      ,			reason="See https://github.com/huggingface/transformers/pull/24204"  )
@slow
class __snake_case	(   unittest.TestCase  ):
    def    _SCREAMING_SNAKE_CASE (						self   ):



           '''simple docstring'''







           lowercase							:  List[Any] 			=   """RWKV/rwkv-4-169m-pile"""
           lowercase							:  str 			=   AutoTokenizer.from_pretrained(self.model_id   )
    def    _SCREAMING_SNAKE_CASE (						self   ):



           '''simple docstring'''







           lowercase							:  List[Any] 			=   """Hello my name is Jasmine and I am a newbie to the"""
           lowercase							:  List[Any] 			=   RwkvForCausalLM.from_pretrained(self.model_id   ).to(snake_case   )

           lowercase							:  Union[str, Any] 			=   self.tokenizer("""Hello my name is"""	,return_tensors="""pt"""   ).input_ids.to(snake_case   )
           lowercase							:  Optional[Any] 			=   model.generate(snake_case	,max_new_tokens=10   )
           lowercase							:  List[Any] 			=   self.tokenizer.decode(output[0].tolist()   )

           self.assertEqual(snake_case	,snake_case   )
    def    _SCREAMING_SNAKE_CASE (						self   ):



           '''simple docstring'''







           lowercase							:  Any 			=   """Hello my name is Jasmine and I am a newbie to the"""

           lowercase							:  Optional[Any] 			=   self.tokenizer("""Hello my name is"""	,return_tensors="""pt"""   ).input_ids.to(snake_case   )
           lowercase							:  Union[str, Any] 			=   RwkvForCausalLM.from_pretrained(self.model_id	,torch_dtype=torch.bfloataa   ).to(snake_case   )

           lowercase							:  Tuple 			=   model.generate(snake_case	,max_new_tokens=10   )
           lowercase							:  Optional[int] 			=   self.tokenizer.decode(output[0].tolist()   )

           self.assertEqual(snake_case	,snake_case   )
