





import collections
import json
import math
import os
import re
import time
from fnmatch import fnmatch
from typing import Dict

import requests
from slack_sdk import WebClient


__A     =    WebClient(token=os.environ["CI_SLACK_BOT_TOKEN"])





def 					lowerCAmelCase_			(	__a		)					->		str:







  """simple docstring"""





  lowerCamelCase__:				str        =test_results.split(" "		)

  lowerCamelCase__:				Optional[Any]        =0
  lowerCamelCase__:				Union[str, Any]        =0

  # When the output is short enough, the output is surrounded by = signs: "== OUTPUT =="
  # When it is too long, those signs are not present.
  lowerCamelCase__:				Optional[int]        =expressions[-2] if "=" in expressions[-1] else expressions[-1]

  for i, expression in enumerate(__a		):
    if "failed" in expression:
      failed += int(expressions[i - 1]		)
    if "passed" in expression:
      success += int(expressions[i - 1]		)

  return failed, success, time_spent





def 					lowerCAmelCase_			(	__a		)					->		Any:







  """simple docstring"""





  lowerCamelCase__:				str        ={}
  lowerCamelCase__:				Union[str, Any]        =None
  lowerCamelCase__:				Optional[Any]        =False
  for line in failures_short_lines.split("\n"		):
    if re.search(R"_ \[doctest\]"     ,   __a		):
      lowerCamelCase__:				List[str]        =True
      lowerCamelCase__:				int        =line.split(" "		)[2]
    elif in_error and not line.split(" "		)[0].isdigit():
      lowerCamelCase__:				Union[str, Any]        =line
      lowerCamelCase__:				Union[str, Any]        =False

  return failures



class        _SCREAMING_SNAKE_CASE						:







 '''simple docstring'''

 def __init__(self    :       Dict	,			UpperCAmelCase_    :       str	,			UpperCAmelCase_    :       Dict)	->List[Any]:



   '''simple docstring'''



   lowerCamelCase__:				Optional[int]        =title

   lowerCamelCase__:				Dict        =doc_test_results["time_spent"].split(",")[0]
   lowerCamelCase__:				Any        =doc_test_results["success"]
   lowerCamelCase__:				Union[str, Any]        =doc_test_results["failures"]
   lowerCamelCase__:				int        =self.n_success + self.n_failures

   # Failures and success of the modeling tests
   lowerCamelCase__:				Optional[int]        =doc_test_results

 @property
 def 							SCREAMING_SNAKE_CASE_ (self    :       Optional[Any])	->str:



   '''simple docstring'''



   lowerCamelCase__:				Optional[Any]        =[self._time_spent]
   lowerCamelCase__:				int        =0

   for time in time_spent:
     lowerCamelCase__:				str        =time.split(":")

     # Time can be formatted as xx:xx:xx, as .xx, or as x.xx if the time spent was less than a minute.
     if len(UpperCAmelCase_) == 1:
       lowerCamelCase__:				List[str]        =[0, 0, time_parts[0]]

     lowerCamelCase__  , lowerCamelCase__  , lowerCamelCase__:				Dict        =int(time_parts[0]), int(time_parts[1]), float(time_parts[2])
     total_secs += hours * 3_600 + minutes * 60 + seconds

   lowerCamelCase__  , lowerCamelCase__  , lowerCamelCase__:				Union[str, Any]        =total_secs // 3_600, (total_secs % 3_600) // 60, total_secs % 60
   return F"""{int(UpperCAmelCase_)}h{int(UpperCAmelCase_)}m{int(UpperCAmelCase_)}s"""

 @property
 def 							SCREAMING_SNAKE_CASE_ (self    :       Optional[Any])	->Dict:



   '''simple docstring'''



   return {"type": "header", "text": {"type": "plain_text", "text": self.title}}

 @property
 def 							SCREAMING_SNAKE_CASE_ (self    :       Optional[int])	->Dict:



   '''simple docstring'''



   return {
       "type": "section",
       "text": {
           "type": "plain_text",
           "text": F"""🌞 There were no failures: all {self.n_tests} tests passed. The suite ran in {self.time}.""",
           "emoji": True,
       },
       "accessory": {
           "type": "button",
           "text": {"type": "plain_text", "text": "Check Action results", "emoji": True},
           "url": F"""https://github.com/huggingface/transformers/actions/runs/{os.environ["GITHUB_RUN_ID"]}""",
       },
   }

 @property
 def 							SCREAMING_SNAKE_CASE_ (self    :       Tuple)	->Dict:



   '''simple docstring'''



   return {
       "type": "section",
       "text": {
           "type": "plain_text",
           "text": (
               F"""There were {self.n_failures} failures, out of {self.n_tests} tests.\nThe suite ran in"""
               F""" {self.time}."""
           ),
           "emoji": True,
       },
       "accessory": {
           "type": "button",
           "text": {"type": "plain_text", "text": "Check Action results", "emoji": True},
           "url": F"""https://github.com/huggingface/transformers/actions/runs/{os.environ["GITHUB_RUN_ID"]}""",
       },
   }

 @property
 def 							SCREAMING_SNAKE_CASE_ (self    :       Any)	->Dict:



   '''simple docstring'''



   lowerCamelCase__:				Optional[int]        =40
   lowerCamelCase__:				Union[str, Any]        ={k: v["failed"] for k, v in doc_test_results.items() if isinstance(UpperCAmelCase_	,			UpperCAmelCase_)}

   lowerCamelCase__:				List[Any]        =""
   for category, failures in category_failures.items():
     if len(UpperCAmelCase_) == 0:
       continue

     if report != "":
       report += "\n\n"

     report += F"""*{category} failures*:""".ljust(line_length // 2).rjust(line_length // 2) + "\n"
     report += "`"
     report += "`\n`".join(UpperCAmelCase_)
     report += "`"

   return {
       "type": "section",
       "text": {
           "type": "mrkdwn",
           "text": F"""The following examples had failures:\n\n\n{report}\n""",
       },
   }

 @property
 def 							SCREAMING_SNAKE_CASE_ (self    :       Tuple)	->str:



   '''simple docstring'''



   lowerCamelCase__:				int        =[self.header]

   if self.n_failures > 0:
     blocks.append(self.failures)

   if self.n_failures > 0:
     blocks.extend([self.category_failures])

   if self.n_failures == 0:
     blocks.append(self.no_failures)

   return json.dumps(UpperCAmelCase_)

 @staticmethod
 def 							SCREAMING_SNAKE_CASE_ ()	->Tuple:



   '''simple docstring'''



   lowerCamelCase__:				Any        =[
       {
           "type": "section",
           "text": {
               "type": "plain_text",
               "text": "There was an issue running the tests.",
           },
           "accessory": {
               "type": "button",
               "text": {"type": "plain_text", "text": "Check Action results", "emoji": True},
               "url": F"""https://github.com/huggingface/transformers/actions/runs/{os.environ["GITHUB_RUN_ID"]}""",
           },
       }
   ]

   print("Sending the following payload")
   print(json.dumps({"blocks": json.loads(UpperCAmelCase_)}))

   client.chat_postMessage(
       channel=os.environ["CI_SLACK_CHANNEL_ID_DAILY"]	,			text="There was an issue running the tests."	,			blocks=UpperCAmelCase_	,			)

 def 							SCREAMING_SNAKE_CASE_ (self    :       Optional[Any])	->Optional[Any]:



   '''simple docstring'''



   print("Sending the following payload")
   print(json.dumps({"blocks": json.loads(self.payload)}))

   lowerCamelCase__:				Any        =F"""{self.n_failures} failures out of {self.n_tests} tests,""" if self.n_failures else "All tests passed."

   lowerCamelCase__:				Union[str, Any]        =client.chat_postMessage(
       channel=os.environ["CI_SLACK_CHANNEL_ID_DAILY"]	,			blocks=self.payload	,			text=UpperCAmelCase_	,			)

 def 							SCREAMING_SNAKE_CASE_ (self    :       Tuple	,			UpperCAmelCase_    :       Optional[int]	,			UpperCAmelCase_    :       str	,			UpperCAmelCase_    :       List[str]	,			UpperCAmelCase_    :       Optional[int])	->Any:



   '''simple docstring'''



   lowerCamelCase__:				int        =""
   for key, value in failures.items():
     lowerCamelCase__:				Tuple        =value[:200] + " [Truncated]" if len(UpperCAmelCase_) > 250 else value
     failures_text += F"""*{key}*\n_{value}_\n\n"""

   lowerCamelCase__:				Dict        =job_name
   lowerCamelCase__:				List[str]        ={"type": "section", "text": {"type": "mrkdwn", "text": text}}

   if job_link is not None:
     lowerCamelCase__:				Tuple        ={
         "type": "button",
         "text": {"type": "plain_text", "text": "GitHub Action job", "emoji": True},
         "url": job_link,
     }

   return [
       {"type": "header", "text": {"type": "plain_text", "text": title.upper(), "emoji": True}},
       content,
       {"type": "section", "text": {"type": "mrkdwn", "text": failures_text}},
   ]







 def 							SCREAMING_SNAKE_CASE_ (self    :       Dict)	->Dict:



   '''simple docstring'''



   if self.thread_ts is None:
     raise ValueError("Can only post reply if a post has been made.")

   lowerCamelCase__:				Dict        =self.doc_test_results.pop("job_link")
   self.doc_test_results.pop("failures")
   self.doc_test_results.pop("success")
   self.doc_test_results.pop("time_spent")

   lowerCamelCase__:				Optional[int]        =sorted(self.doc_test_results.items()	,			key=lambda UpperCAmelCase_: t[0])
   for job, job_result in sorted_dict:
     if len(job_result["failures"]):
       lowerCamelCase__:				Optional[Any]        =F"""*Num failures* :{len(job_result["failed"])} \n"""
       lowerCamelCase__:				Dict        =job_result["failures"]
       lowerCamelCase__:				Any        =self.get_reply_blocks(UpperCAmelCase_	,			UpperCAmelCase_	,			UpperCAmelCase_	,			text=UpperCAmelCase_)

       print("Sending the following reply")
       print(json.dumps({"blocks": blocks}))

       client.chat_postMessage(
           channel=os.environ["CI_SLACK_CHANNEL_ID_DAILY"]	,			text=F"""Results for {job}"""	,			blocks=UpperCAmelCase_	,			thread_ts=self.thread_ts["ts"]	,			)

       time.sleep(1)





def 					lowerCAmelCase_			(	)					->		int:







  """simple docstring"""





  lowerCamelCase__:				str        =os.environ["GITHUB_RUN_ID"]
  lowerCamelCase__:				Optional[int]        =F"""https://api.github.com/repos/huggingface/transformers/actions/runs/{run_id}/jobs?per_page=100"""
  lowerCamelCase__:				Optional[Any]        =requests.get(__a		).json()
  lowerCamelCase__:				str        ={}

  try:
    jobs.update({job["name"]: job["html_url"] for job in result["jobs"]}		)
    lowerCamelCase__:				Tuple        =math.ceil((result["total_count"] - 100) / 100		)

    for i in range(__a		):
      lowerCamelCase__:				Optional[Any]        =requests.get(url + F"""&page={i + 2}"""		).json()
      jobs.update({job["name"]: job["html_url"] for job in result["jobs"]}		)

    return jobs
  except Exception as e:
    print("Unknown error, could not fetch links."     ,   __a		)

  return {}





def 					lowerCAmelCase_			(	__a		)					->		List[Any]:







  """simple docstring"""





  lowerCamelCase__:				Dict        ={}

  if os.path.exists(__a		):
    lowerCamelCase__:				Dict        =os.listdir(__a		)
    for file in files:
      try:
        with open(os.path.join(__a     ,   __a		)     ,   encoding="utf-8"		) as f:
          lowerCamelCase__:				List[Any]        =f.read()
      except UnicodeDecodeError as e:
        raise ValueError(F"""Could not open {os.path.join(__a     ,   __a		)}."""		) from e

  return _artifact





def 					lowerCAmelCase_			(	)					->		List[Any]:







  """simple docstring"""





  class        _SCREAMING_SNAKE_CASE						:







   '''simple docstring'''

   def __init__(self    :       Union[str, Any]	,			UpperCAmelCase_    :       str)	->Optional[int]:



     '''simple docstring'''



     lowerCamelCase__:				List[Any]        =name
     lowerCamelCase__:				Tuple        =[]

   def __str__(self    :       Optional[int])	->int:



     '''simple docstring'''



     return self.name







   def 							SCREAMING_SNAKE_CASE_ (self    :       Dict	,			UpperCAmelCase_    :       str)	->str:



     '''simple docstring'''



     self.paths.append({"name": self.name, "path": path})



  lowerCamelCase__:				Dict[str, Artifact]        ={}

  lowerCamelCase__:				Optional[Any]        =filter(os.path.isdir     ,   os.listdir()		)
  for directory in directories:
    lowerCamelCase__:				str        =directory
    if artifact_name not in _available_artifacts:
      lowerCamelCase__:				int        =Artifact(__a		)

      _available_artifacts[artifact_name].add_path(__a		)

  return _available_artifacts


if __name__ == "__main__":
     __A     =    get_job_links()
     __A     =    retrieve_available_artifacts()

     __A     =    collections.OrderedDict(
         [
             ("*.py", "API Examples"),
             ("*.md", "MD Examples"),
         ]
     )

     # This dict will contain all the information relative to each doc test category:
     # - failed: list of failed tests
     # - failures: dict in the format 'test': 'error_message'
     __A     =    {
         v: {
             "failed": [],
             "failures": {},
         }
         for v in docs.values()
     }

     # Link to the GitHub Action job
     __A     =    github_actions_job_links.get("run_doctests")

     __A     =    available_artifacts["doc_tests_gpu_test_reports"].paths[0]
     __A     =    retrieve_artifact(artifact_path["name"])
     if "stats" in artifact:
          __A ,     __A ,     __A     =    handle_test_results(artifact["stats"])
          __A     =    failed
          __A     =    success
          __A     =    time_spent[1:-1] + ", "

          __A     =    extract_first_line_failure(artifact["failures_short"])
          for line in artifact["summary_short"].split("\n"):
               if re.search("FAILED", line):
                    __A     =    line.replace("FAILED ", "")
                    __A     =    line.split()[0].replace("\n", "")

                    if "::" in line:
                         __A ,     __A     =    line.split("::")
                    else:
                         __A ,     __A     =    line, line

                    for file_regex in docs.keys():
                         if fnmatch(file_path, file_regex):
                              __A     =    docs[file_regex]
                              doc_test_results[category]["failed"].append(test)

                              __A     =    all_failures[test] if test in all_failures else "N/A"
                              __A     =    failure
                              break

     __A     =    Message("🤗 Results of the doc tests.", doc_test_results)
     message.post()
     message.post_reply()








import shutil
import tempfile
import unittest
from unittest.mock import patch

from transformers import (
    DefaultFlowCallback,
    IntervalStrategy,
    PrinterCallback,
    ProgressCallback,
    Trainer,
    TrainerCallback,
    TrainingArguments,
    is_torch_available,
)
from transformers.testing_utils import require_torch


if is_torch_available():
     from transformers.trainer import DEFAULT_CALLBACKS

     from .test_trainer import RegressionDataset, RegressionModelConfig, RegressionPreTrainedModel



class        _SCREAMING_SNAKE_CASE						(     __SCREAMING_SNAKE_CASE   ):







 '''simple docstring'''

 def __init__(self    :       str)	->List[Any]:



   '''simple docstring'''



   lowerCamelCase__:				List[Any]        =[]

 def 							SCREAMING_SNAKE_CASE_ (self    :       Any	,			UpperCAmelCase_    :       int	,			UpperCAmelCase_    :       str	,			UpperCAmelCase_    :       Dict	,			**UpperCAmelCase_    :       Any)	->Dict:



   '''simple docstring'''



   self.events.append("on_init_end")

 def 							SCREAMING_SNAKE_CASE_ (self    :       Union[str, Any]	,			UpperCAmelCase_    :       Any	,			UpperCAmelCase_    :       Dict	,			UpperCAmelCase_    :       Dict	,			**UpperCAmelCase_    :       str)	->List[str]:



   '''simple docstring'''



   self.events.append("on_train_begin")

 def 							SCREAMING_SNAKE_CASE_ (self    :       List[str]	,			UpperCAmelCase_    :       Dict	,			UpperCAmelCase_    :       Optional[Any]	,			UpperCAmelCase_    :       List[Any]	,			**UpperCAmelCase_    :       str)	->int:



   '''simple docstring'''



   self.events.append("on_train_end")

 def 							SCREAMING_SNAKE_CASE_ (self    :       Tuple	,			UpperCAmelCase_    :       Union[str, Any]	,			UpperCAmelCase_    :       Any	,			UpperCAmelCase_    :       List[str]	,			**UpperCAmelCase_    :       Optional[int])	->List[Any]:



   '''simple docstring'''



   self.events.append("on_epoch_begin")

 def 							SCREAMING_SNAKE_CASE_ (self    :       Optional[int]	,			UpperCAmelCase_    :       Dict	,			UpperCAmelCase_    :       List[Any]	,			UpperCAmelCase_    :       int	,			**UpperCAmelCase_    :       Any)	->Tuple:



   '''simple docstring'''



   self.events.append("on_epoch_end")

 def 							SCREAMING_SNAKE_CASE_ (self    :       List[Any]	,			UpperCAmelCase_    :       Dict	,			UpperCAmelCase_    :       Any	,			UpperCAmelCase_    :       Any	,			**UpperCAmelCase_    :       List[Any])	->Optional[int]:



   '''simple docstring'''



   self.events.append("on_step_begin")

 def 							SCREAMING_SNAKE_CASE_ (self    :       List[str]	,			UpperCAmelCase_    :       List[str]	,			UpperCAmelCase_    :       List[str]	,			UpperCAmelCase_    :       Optional[int]	,			**UpperCAmelCase_    :       List[str])	->Tuple:



   '''simple docstring'''



   self.events.append("on_step_end")

 def 							SCREAMING_SNAKE_CASE_ (self    :       Union[str, Any]	,			UpperCAmelCase_    :       Optional[Any]	,			UpperCAmelCase_    :       Dict	,			UpperCAmelCase_    :       Optional[Any]	,			**UpperCAmelCase_    :       str)	->Optional[int]:



   '''simple docstring'''



   self.events.append("on_evaluate")

 def 							SCREAMING_SNAKE_CASE_ (self    :       Tuple	,			UpperCAmelCase_    :       List[str]	,			UpperCAmelCase_    :       List[Any]	,			UpperCAmelCase_    :       Union[str, Any]	,			**UpperCAmelCase_    :       Any)	->int:



   '''simple docstring'''



   self.events.append("on_predict")

 def 							SCREAMING_SNAKE_CASE_ (self    :       Union[str, Any]	,			UpperCAmelCase_    :       List[Any]	,			UpperCAmelCase_    :       Optional[int]	,			UpperCAmelCase_    :       Tuple	,			**UpperCAmelCase_    :       List[Any])	->Any:



   '''simple docstring'''



   self.events.append("on_save")

 def 							SCREAMING_SNAKE_CASE_ (self    :       str	,			UpperCAmelCase_    :       Any	,			UpperCAmelCase_    :       List[str]	,			UpperCAmelCase_    :       Any	,			**UpperCAmelCase_    :       Optional[Any])	->str:



   '''simple docstring'''



   self.events.append("on_log")







 def 							SCREAMING_SNAKE_CASE_ (self    :       Optional[Any]	,			UpperCAmelCase_    :       Any	,			UpperCAmelCase_    :       List[Any]	,			UpperCAmelCase_    :       str	,			**UpperCAmelCase_    :       List[str])	->Optional[int]:



   '''simple docstring'''



   self.events.append("on_prediction_step")



@require_torch
class        _SCREAMING_SNAKE_CASE						(     unittest.TestCase   ):







 '''simple docstring'''

 def 							SCREAMING_SNAKE_CASE_ (self    :       List[Any])	->int:



   '''simple docstring'''



   lowerCamelCase__:				Tuple        =tempfile.mkdtemp()

 def 							SCREAMING_SNAKE_CASE_ (self    :       List[str])	->Tuple:



   '''simple docstring'''



   shutil.rmtree(self.output_dir)

 def 							SCREAMING_SNAKE_CASE_ (self    :       List[Any]	,			UpperCAmelCase_    :       int=0	,			UpperCAmelCase_    :       Tuple=0	,			UpperCAmelCase_    :       str=64	,			UpperCAmelCase_    :       List[Any]=64	,			UpperCAmelCase_    :       Union[str, Any]=None	,			UpperCAmelCase_    :       List[str]=False	,			**UpperCAmelCase_    :       Tuple)	->Any:



   '''simple docstring'''



   lowerCamelCase__:				Union[str, Any]        =RegressionDataset(length=UpperCAmelCase_)
   lowerCamelCase__:				int        =RegressionDataset(length=UpperCAmelCase_)
   lowerCamelCase__:				str        =RegressionModelConfig(a=UpperCAmelCase_	,			b=UpperCAmelCase_)
   lowerCamelCase__:				Union[str, Any]        =RegressionPreTrainedModel(UpperCAmelCase_)

   lowerCamelCase__:				int        =TrainingArguments(self.output_dir	,			disable_tqdm=UpperCAmelCase_	,			report_to=[]	,			**UpperCAmelCase_)
   return Trainer(
       UpperCAmelCase_	,			UpperCAmelCase_	,			train_dataset=UpperCAmelCase_	,			eval_dataset=UpperCAmelCase_	,			callbacks=UpperCAmelCase_	,			)

 def 							SCREAMING_SNAKE_CASE_ (self    :       List[str]	,			UpperCAmelCase_    :       str	,			UpperCAmelCase_    :       List[str])	->Dict:



   '''simple docstring'''



   self.assertEqual(len(UpperCAmelCase_)	,			len(UpperCAmelCase_))

   # Order doesn't matter
   lowerCamelCase__:				Dict        =sorted(UpperCAmelCase_	,			key=lambda UpperCAmelCase_: cb.__name__ if isinstance(UpperCAmelCase_	,			UpperCAmelCase_) else cb.__class__.__name__)
   lowerCamelCase__:				Optional[int]        =sorted(UpperCAmelCase_	,			key=lambda UpperCAmelCase_: cb.__name__ if isinstance(UpperCAmelCase_	,			UpperCAmelCase_) else cb.__class__.__name__)

   for cba, cba in zip(UpperCAmelCase_	,			UpperCAmelCase_):
     if isinstance(UpperCAmelCase_	,			UpperCAmelCase_) and isinstance(UpperCAmelCase_	,			UpperCAmelCase_):
       self.assertEqual(UpperCAmelCase_	,			UpperCAmelCase_)
     elif isinstance(UpperCAmelCase_	,			UpperCAmelCase_) and not isinstance(UpperCAmelCase_	,			UpperCAmelCase_):
       self.assertEqual(UpperCAmelCase_	,			cba.__class__)
     elif not isinstance(UpperCAmelCase_	,			UpperCAmelCase_) and isinstance(UpperCAmelCase_	,			UpperCAmelCase_):
       self.assertEqual(cba.__class__	,			UpperCAmelCase_)
     else:
       self.assertEqual(UpperCAmelCase_	,			UpperCAmelCase_)

 def 							SCREAMING_SNAKE_CASE_ (self    :       List[Any]	,			UpperCAmelCase_    :       Optional[Any])	->Optional[int]:



   '''simple docstring'''



   lowerCamelCase__:				Dict        =["on_init_end", "on_train_begin"]
   lowerCamelCase__:				List[str]        =0
   lowerCamelCase__:				List[Any]        =len(trainer.get_eval_dataloader())
   lowerCamelCase__:				Dict        =["on_prediction_step"] * len(trainer.get_eval_dataloader()) + ["on_log", "on_evaluate"]
   for _ in range(trainer.state.num_train_epochs):
     expected_events.append("on_epoch_begin")
     for _ in range(UpperCAmelCase_):
       step += 1
       expected_events += ["on_step_begin", "on_step_end"]
       if step % trainer.args.logging_steps == 0:
         expected_events.append("on_log")
       if trainer.args.evaluation_strategy == IntervalStrategy.STEPS and step % trainer.args.eval_steps == 0:
         expected_events += evaluation_events.copy()
       if step % trainer.args.save_steps == 0:
         expected_events.append("on_save")
     expected_events.append("on_epoch_end")
     if trainer.args.evaluation_strategy == IntervalStrategy.EPOCH:
       expected_events += evaluation_events.copy()
   expected_events += ["on_log", "on_train_end"]
   return expected_events

 def 							SCREAMING_SNAKE_CASE_ (self    :       int)	->Optional[int]:



   '''simple docstring'''



   lowerCamelCase__:				Optional[Any]        =self.get_trainer()
   lowerCamelCase__:				Any        =DEFAULT_CALLBACKS.copy() + [ProgressCallback]
   self.check_callbacks_equality(trainer.callback_handler.callbacks	,			UpperCAmelCase_)

   # Callbacks passed at init are added to the default callbacks
   lowerCamelCase__:				Dict        =self.get_trainer(callbacks=[MyTestTrainerCallback])
   expected_callbacks.append(UpperCAmelCase_)
   self.check_callbacks_equality(trainer.callback_handler.callbacks	,			UpperCAmelCase_)

   # TrainingArguments.disable_tqdm controls if use ProgressCallback or PrinterCallback
   lowerCamelCase__:				int        =self.get_trainer(disable_tqdm=UpperCAmelCase_)
   lowerCamelCase__:				Tuple        =DEFAULT_CALLBACKS.copy() + [PrinterCallback]
   self.check_callbacks_equality(trainer.callback_handler.callbacks	,			UpperCAmelCase_)

 def 							SCREAMING_SNAKE_CASE_ (self    :       List[str])	->Union[str, Any]:



   '''simple docstring'''



   lowerCamelCase__:				Union[str, Any]        =DEFAULT_CALLBACKS.copy() + [ProgressCallback]
   lowerCamelCase__:				Optional[int]        =self.get_trainer()

   # We can add, pop, or remove by class name
   trainer.remove_callback(UpperCAmelCase_)
   expected_callbacks.remove(UpperCAmelCase_)
   self.check_callbacks_equality(trainer.callback_handler.callbacks	,			UpperCAmelCase_)

   lowerCamelCase__:				Dict        =self.get_trainer()
   lowerCamelCase__:				str        =trainer.pop_callback(UpperCAmelCase_)
   self.assertEqual(cb.__class__	,			UpperCAmelCase_)
   self.check_callbacks_equality(trainer.callback_handler.callbacks	,			UpperCAmelCase_)

   trainer.add_callback(UpperCAmelCase_)
   expected_callbacks.insert(0	,			UpperCAmelCase_)
   self.check_callbacks_equality(trainer.callback_handler.callbacks	,			UpperCAmelCase_)

   # We can also add, pop, or remove by instance
   lowerCamelCase__:				List[str]        =self.get_trainer()
   lowerCamelCase__:				List[str]        =trainer.callback_handler.callbacks[0]
   trainer.remove_callback(UpperCAmelCase_)
   expected_callbacks.remove(UpperCAmelCase_)
   self.check_callbacks_equality(trainer.callback_handler.callbacks	,			UpperCAmelCase_)

   lowerCamelCase__:				str        =self.get_trainer()
   lowerCamelCase__:				List[Any]        =trainer.callback_handler.callbacks[0]
   lowerCamelCase__:				Dict        =trainer.pop_callback(UpperCAmelCase_)
   self.assertEqual(UpperCAmelCase_	,			UpperCAmelCase_)
   self.check_callbacks_equality(trainer.callback_handler.callbacks	,			UpperCAmelCase_)

   trainer.add_callback(UpperCAmelCase_)
   expected_callbacks.insert(0	,			UpperCAmelCase_)
   self.check_callbacks_equality(trainer.callback_handler.callbacks	,			UpperCAmelCase_)







 def 							SCREAMING_SNAKE_CASE_ (self    :       List[Any])	->int:



   '''simple docstring'''



   import warnings

   # XXX: for now ignore scatter_gather warnings in this test since it's not relevant to what's being tested
   warnings.simplefilter(action="ignore"	,			category=UpperCAmelCase_)

   lowerCamelCase__:				Union[str, Any]        =self.get_trainer(callbacks=[MyTestTrainerCallback])
   trainer.train()
   lowerCamelCase__:				int        =trainer.callback_handler.callbacks[-2].events
   self.assertEqual(UpperCAmelCase_	,			self.get_expected_events(UpperCAmelCase_))

   # Independent log/save/eval
   lowerCamelCase__:				Dict        =self.get_trainer(callbacks=[MyTestTrainerCallback]	,			logging_steps=5)
   trainer.train()
   lowerCamelCase__:				Optional[int]        =trainer.callback_handler.callbacks[-2].events
   self.assertEqual(UpperCAmelCase_	,			self.get_expected_events(UpperCAmelCase_))

   lowerCamelCase__:				Any        =self.get_trainer(callbacks=[MyTestTrainerCallback]	,			save_steps=5)
   trainer.train()
   lowerCamelCase__:				List[Any]        =trainer.callback_handler.callbacks[-2].events
   self.assertEqual(UpperCAmelCase_	,			self.get_expected_events(UpperCAmelCase_))

   lowerCamelCase__:				int        =self.get_trainer(callbacks=[MyTestTrainerCallback]	,			eval_steps=5	,			evaluation_strategy="steps")
   trainer.train()
   lowerCamelCase__:				str        =trainer.callback_handler.callbacks[-2].events
   self.assertEqual(UpperCAmelCase_	,			self.get_expected_events(UpperCAmelCase_))

   lowerCamelCase__:				Dict        =self.get_trainer(callbacks=[MyTestTrainerCallback]	,			evaluation_strategy="epoch")
   trainer.train()
   lowerCamelCase__:				Tuple        =trainer.callback_handler.callbacks[-2].events
   self.assertEqual(UpperCAmelCase_	,			self.get_expected_events(UpperCAmelCase_))

   # A bit of everything
   lowerCamelCase__:				Tuple        =self.get_trainer(
       callbacks=[MyTestTrainerCallback]	,			logging_steps=3	,			save_steps=10	,			eval_steps=5	,			evaluation_strategy="steps"	,			)
   trainer.train()
   lowerCamelCase__:				int        =trainer.callback_handler.callbacks[-2].events
   self.assertEqual(UpperCAmelCase_	,			self.get_expected_events(UpperCAmelCase_))

   # warning should be emitted for duplicated callbacks
   with patch("transformers.trainer_callback.logger.warning") as warn_mock:
     lowerCamelCase__:				Optional[int]        =self.get_trainer(
         callbacks=[MyTestTrainerCallback, MyTestTrainerCallback]	,			)
     assert str(UpperCAmelCase_) in warn_mock.call_args[0][0]

