



"""simple docstring"""

import tempfile

import torch

from diffusers import (
    DEISMultistepScheduler,
    DPMSolverMultistepScheduler,
    DPMSolverSinglestepScheduler,
    UniPCMultistepScheduler,
)

from .test_schedulers import SchedulerCommonTest




class        _UpperCAmelCase       (			lowerCAmelCase__):
 _lowerCAmelCase   :			int							=		(UniPCMultistepScheduler,)
 _lowerCAmelCase   :			Optional[int]							=		(("""num_inference_steps""", 2_5),)

 def 						_snake_case					(     self		:       Union[str, Any]  ,     **lowercase_		:       List[Any]				):
     snake_case_ :     Dict				=	{
         '''num_train_timesteps''': 1000,
         '''beta_start''': 0.00_01,
         '''beta_end''': 0.02,
         '''beta_schedule''': '''linear''',
         '''solver_order''': 2,
         '''solver_type''': '''bh2''',
     }

     config.update(**lowercase_				)
     return config

 def 						_snake_case					(     self		:       Union[str, Any]  ,     lowercase_		:       str=0  ,     **lowercase_		:       int				):
     snake_case_ :     Tuple				=	dict(self.forward_default_kwargs				)
     snake_case_ :     Optional[int]				=	kwargs.pop('''num_inference_steps'''  ,     lowercase_				)
     snake_case_ :     Optional[int]				=	self.dummy_sample
     snake_case_ :     List[str]				=	0.1 * sample
     snake_case_ :     Tuple				=	[residual + 0.2, residual + 0.15, residual + 0.10]

     for scheduler_class in self.scheduler_classes:
         snake_case_ :     Any				=	self.get_scheduler_config(**lowercase_				)
         snake_case_ :     List[str]				=	scheduler_class(**lowercase_				)
         scheduler.set_timesteps(lowercase_				)
         # copy over dummy past residuals
         snake_case_ :     Optional[Any]				=	dummy_past_residuals[: scheduler.config.solver_order]

         with tempfile.TemporaryDirectory() as tmpdirname:
             scheduler.save_config(lowercase_				)
             snake_case_ :     int				=	scheduler_class.from_pretrained(lowercase_				)
             new_scheduler.set_timesteps(lowercase_				)
             # copy over dummy past residuals
             snake_case_ :     int				=	dummy_past_residuals[: new_scheduler.config.solver_order]

         snake_case_,	snake_case_ :     Optional[int]				=	sample, sample
         for t in range(lowercase_  ,     time_step + scheduler.config.solver_order + 1				):
             snake_case_ :     int				=	scheduler.step(lowercase_  ,     lowercase_  ,     lowercase_  ,     **lowercase_				).prev_sample
             snake_case_ :     Dict				=	new_scheduler.step(lowercase_  ,     lowercase_  ,     lowercase_  ,     **lowercase_				).prev_sample

             assert torch.sum(torch.abs(output - new_output				)				) < 1E-5, "Scheduler outputs are not identical"

 def 						_snake_case					(     self		:       List[str]  ,     lowercase_		:       Optional[Any]=0  ,     **lowercase_		:       int				):
     snake_case_ :     List[str]				=	dict(self.forward_default_kwargs				)
     snake_case_ :     List[Any]				=	kwargs.pop('''num_inference_steps'''  ,     lowercase_				)
     snake_case_ :     List[Any]				=	self.dummy_sample
     snake_case_ :     List[Any]				=	0.1 * sample
     snake_case_ :     Optional[int]				=	[residual + 0.2, residual + 0.15, residual + 0.10]

     for scheduler_class in self.scheduler_classes:
         snake_case_ :     Optional[Any]				=	self.get_scheduler_config()
         snake_case_ :     Union[str, Any]				=	scheduler_class(**lowercase_				)
         scheduler.set_timesteps(lowercase_				)

         # copy over dummy past residuals (must be after setting timesteps)
         snake_case_ :     Optional[int]				=	dummy_past_residuals[: scheduler.config.solver_order]

         with tempfile.TemporaryDirectory() as tmpdirname:
             scheduler.save_config(lowercase_				)
             snake_case_ :     List[Any]				=	scheduler_class.from_pretrained(lowercase_				)
             # copy over dummy past residuals
             new_scheduler.set_timesteps(lowercase_				)

             # copy over dummy past residual (must be after setting timesteps)
             snake_case_ :     int				=	dummy_past_residuals[: new_scheduler.config.solver_order]

         snake_case_ :     Optional[int]				=	scheduler.step(lowercase_  ,     lowercase_  ,     lowercase_  ,     **lowercase_				).prev_sample
         snake_case_ :     Tuple				=	new_scheduler.step(lowercase_  ,     lowercase_  ,     lowercase_  ,     **lowercase_				).prev_sample

         assert torch.sum(torch.abs(output - new_output				)				) < 1E-5, "Scheduler outputs are not identical"

 def 						_snake_case					(     self		:       List[str]  ,     lowercase_		:       Any=None  ,     **lowercase_		:       str				):
     if scheduler is None:
         snake_case_ :     Optional[int]				=	self.scheduler_classes[0]
         snake_case_ :     Tuple				=	self.get_scheduler_config(**lowercase_				)
         snake_case_ :     Optional[int]				=	scheduler_class(**lowercase_				)

     snake_case_ :     Dict				=	self.scheduler_classes[0]
     snake_case_ :     List[str]				=	self.get_scheduler_config(**lowercase_				)
     snake_case_ :     Optional[int]				=	scheduler_class(**lowercase_				)

     snake_case_ :     Dict				=	10
     snake_case_ :     str				=	self.dummy_model()
     snake_case_ :     List[str]				=	self.dummy_sample_deter
     scheduler.set_timesteps(lowercase_				)

     for i, t in enumerate(scheduler.timesteps				):
         snake_case_ :     List[Any]				=	model(lowercase_  ,     lowercase_				)
         snake_case_ :     Optional[int]				=	scheduler.step(lowercase_  ,     lowercase_  ,     lowercase_				).prev_sample

     return sample

 def 						_snake_case					(     self		:       int				):
     snake_case_ :     str				=	dict(self.forward_default_kwargs				)

     snake_case_ :     Optional[int]				=	kwargs.pop('''num_inference_steps'''  ,     lowercase_				)

     for scheduler_class in self.scheduler_classes:
         snake_case_ :     Tuple				=	self.get_scheduler_config()
         snake_case_ :     Dict				=	scheduler_class(**lowercase_				)

         snake_case_ :     List[str]				=	self.dummy_sample
         snake_case_ :     Optional[int]				=	0.1 * sample

         if num_inference_steps is not None and hasattr(lowercase_  ,     '''set_timesteps'''				):
             scheduler.set_timesteps(lowercase_				)
         elif num_inference_steps is not None and not hasattr(lowercase_  ,     '''set_timesteps'''				):
             snake_case_ :     Tuple				=	num_inference_steps

         # copy over dummy past residuals (must be done after set_timesteps)
         snake_case_ :     str				=	[residual + 0.2, residual + 0.15, residual + 0.10]
         snake_case_ :     List[str]				=	dummy_past_residuals[: scheduler.config.solver_order]

         snake_case_ :     List[str]				=	scheduler.timesteps[5]
         snake_case_ :     int				=	scheduler.timesteps[6]

         snake_case_ :     Optional[int]				=	scheduler.step(lowercase_  ,     lowercase_  ,     lowercase_  ,     **lowercase_				).prev_sample
         snake_case_ :     Dict				=	scheduler.step(lowercase_  ,     lowercase_  ,     lowercase_  ,     **lowercase_				).prev_sample

         self.assertEqual(output_a.shape  ,     sample.shape				)
         self.assertEqual(output_a.shape  ,     output_a.shape				)

 def 						_snake_case					(     self		:       List[str]				):
     # make sure that iterating over schedulers with same config names gives same results
     # for defaults
     snake_case_ :     Tuple				=	UniPCMultistepScheduler(**self.get_scheduler_config()				)
     snake_case_ :     int				=	self.full_loop(scheduler=lowercase_				)
     snake_case_ :     Union[str, Any]				=	torch.mean(torch.abs(lowercase_				)				)

     assert abs(result_mean.item() - 0.24_64				) < 1E-3

     snake_case_ :     str				=	DPMSolverSinglestepScheduler.from_config(scheduler.config				)
     snake_case_ :     Tuple				=	DEISMultistepScheduler.from_config(scheduler.config				)
     snake_case_ :     Union[str, Any]				=	DPMSolverMultistepScheduler.from_config(scheduler.config				)
     snake_case_ :     int				=	UniPCMultistepScheduler.from_config(scheduler.config				)

     snake_case_ :     List[Any]				=	self.full_loop(scheduler=lowercase_				)
     snake_case_ :     Any				=	torch.mean(torch.abs(lowercase_				)				)

     assert abs(result_mean.item() - 0.24_64				) < 1E-3

 def 						_snake_case					(     self		:       Union[str, Any]				):
     for timesteps in [25, 50, 100, 999, 1000]:
         self.check_over_configs(num_train_timesteps=lowercase_				)

 def 						_snake_case					(     self		:       Tuple				):
     self.check_over_configs(thresholding=lowercase_				)
     for order in [1, 2, 3]:
         for solver_type in ["bh1", "bh2"]:
             for threshold in [0.5, 1.0, 2.0]:
                 for prediction_type in ["epsilon", "sample"]:
                     self.check_over_configs(
                         thresholding=lowercase_  ,     prediction_type=lowercase_  ,     sample_max_value=lowercase_  ,     solver_order=lowercase_  ,     solver_type=lowercase_  ,     )

 def 						_snake_case					(     self		:       List[str]				):
     for prediction_type in ["epsilon", "v_prediction"]:
         self.check_over_configs(prediction_type=lowercase_				)

 def 						_snake_case					(     self		:       Any				):
     for solver_type in ["bh1", "bh2"]:
         for order in [1, 2, 3]:
             for prediction_type in ["epsilon", "sample"]:
                 self.check_over_configs(
                     solver_order=lowercase_  ,     solver_type=lowercase_  ,     prediction_type=lowercase_  ,     )
                 snake_case_ :     Tuple				=	self.full_loop(
                     solver_order=lowercase_  ,     solver_type=lowercase_  ,     prediction_type=lowercase_  ,     )
                 assert not torch.isnan(lowercase_				).any(), "Samples have nan numbers"

 def 						_snake_case					(     self		:       List[Any]				):
     self.check_over_configs(lower_order_final=lowercase_				)
     self.check_over_configs(lower_order_final=lowercase_				)

 def 						_snake_case					(     self		:       Dict				):
     for num_inference_steps in [1, 2, 3, 5, 10, 50, 100, 999, 1000]:
         self.check_over_forward(num_inference_steps=lowercase_  ,     time_step=0				)

 def 						_snake_case					(     self		:       str				):
     snake_case_ :     Union[str, Any]				=	self.full_loop()
     snake_case_ :     Tuple				=	torch.mean(torch.abs(lowercase_				)				)

     assert abs(result_mean.item() - 0.24_64				) < 1E-3

 def 						_snake_case					(     self		:       Union[str, Any]				):
     snake_case_ :     Dict				=	self.full_loop(prediction_type='''v_prediction'''				)
     snake_case_ :     Optional[Any]				=	torch.mean(torch.abs(lowercase_				)				)

     assert abs(result_mean.item() - 0.10_14				) < 1E-3

 def 						_snake_case					(     self		:       List[str]				):
     snake_case_ :     Union[str, Any]				=	self.scheduler_classes[0]
     snake_case_ :     int				=	self.get_scheduler_config(thresholding=lowercase_  ,     dynamic_thresholding_ratio=0				)
     snake_case_ :     Union[str, Any]				=	scheduler_class(**lowercase_				)

     snake_case_ :     Optional[Any]				=	10
     snake_case_ :     Any				=	self.dummy_model()
     snake_case_ :     Optional[int]				=	self.dummy_sample_deter.half()
     scheduler.set_timesteps(lowercase_				)

     for i, t in enumerate(scheduler.timesteps				):
         snake_case_ :     Tuple				=	model(lowercase_  ,     lowercase_				)
         snake_case_ :     Dict				=	scheduler.step(lowercase_  ,     lowercase_  ,     lowercase_				).prev_sample

     assert sample.dtype == torch.floataa






 def 						_snake_case					(     self		:       List[Any]  ,     **lowercase_		:       Any				):
     for scheduler_class in self.scheduler_classes:
         snake_case_ :     Tuple				=	self.get_scheduler_config(**lowercase_				)
         snake_case_ :     int				=	scheduler_class(**lowercase_				)

         scheduler.set_timesteps(scheduler.config.num_train_timesteps				)
         assert len(scheduler.timesteps.unique()				) == scheduler.num_inference_steps










"""simple docstring"""

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        _UpperCAmelCase       (			lowerCAmelCase__):

 def __init__(     self		:       Optional[int]				):
     snake_case_ :     str				=	[]

 def 						_snake_case					(     self		:       List[Any]  ,     lowercase_		:       Any  ,     lowercase_		:       Union[str, Any]  ,     lowercase_		:       List[str]  ,     **lowercase_		:       Tuple				):
     self.events.append('''on_init_end'''				)

 def 						_snake_case					(     self		:       List[Any]  ,     lowercase_		:       str  ,     lowercase_		:       Optional[int]  ,     lowercase_		:       List[str]  ,     **lowercase_		:       List[str]				):
     self.events.append('''on_train_begin'''				)

 def 						_snake_case					(     self		:       Any  ,     lowercase_		:       List[str]  ,     lowercase_		:       Tuple  ,     lowercase_		:       List[Any]  ,     **lowercase_		:       Optional[int]				):
     self.events.append('''on_train_end'''				)

 def 						_snake_case					(     self		:       str  ,     lowercase_		:       Optional[int]  ,     lowercase_		:       int  ,     lowercase_		:       Optional[Any]  ,     **lowercase_		:       List[Any]				):
     self.events.append('''on_epoch_begin'''				)

 def 						_snake_case					(     self		:       Tuple  ,     lowercase_		:       List[str]  ,     lowercase_		:       Dict  ,     lowercase_		:       Union[str, Any]  ,     **lowercase_		:       Optional[Any]				):
     self.events.append('''on_epoch_end'''				)

 def 						_snake_case					(     self		:       List[str]  ,     lowercase_		:       Optional[Any]  ,     lowercase_		:       Optional[Any]  ,     lowercase_		:       int  ,     **lowercase_		:       Optional[Any]				):
     self.events.append('''on_step_begin'''				)

 def 						_snake_case					(     self		:       int  ,     lowercase_		:       int  ,     lowercase_		:       Union[str, Any]  ,     lowercase_		:       List[Any]  ,     **lowercase_		:       List[str]				):
     self.events.append('''on_step_end'''				)

 def 						_snake_case					(     self		:       str  ,     lowercase_		:       int  ,     lowercase_		:       Dict  ,     lowercase_		:       List[str]  ,     **lowercase_		:       List[str]				):
     self.events.append('''on_evaluate'''				)

 def 						_snake_case					(     self		:       Dict  ,     lowercase_		:       Union[str, Any]  ,     lowercase_		:       Any  ,     lowercase_		:       List[Any]  ,     **lowercase_		:       str				):
     self.events.append('''on_predict'''				)

 def 						_snake_case					(     self		:       List[Any]  ,     lowercase_		:       Union[str, Any]  ,     lowercase_		:       List[Any]  ,     lowercase_		:       int  ,     **lowercase_		:       Union[str, Any]				):
     self.events.append('''on_save'''				)

 def 						_snake_case					(     self		:       str  ,     lowercase_		:       Tuple  ,     lowercase_		:       Optional[int]  ,     lowercase_		:       List[str]  ,     **lowercase_		:       Any				):
     self.events.append('''on_log'''				)






 def 						_snake_case					(     self		:       Dict  ,     lowercase_		:       Optional[int]  ,     lowercase_		:       List[str]  ,     lowercase_		:       Union[str, Any]  ,     **lowercase_		:       Optional[int]				):
     self.events.append('''on_prediction_step'''				)





@require_torch
class        _UpperCAmelCase       (			unittest.TestCase):

 def 						_snake_case					(     self		:       List[str]				):
     snake_case_ :     Tuple				=	tempfile.mkdtemp()

 def 						_snake_case					(     self		:       Tuple				):
     shutil.rmtree(self.output_dir				)

 def 						_snake_case					(     self		:       int  ,     lowercase_		:       Union[str, Any]=0  ,     lowercase_		:       Dict=0  ,     lowercase_		:       List[str]=64  ,     lowercase_		:       Union[str, Any]=64  ,     lowercase_		:       Union[str, Any]=None  ,     lowercase_		:       Any=False  ,     **lowercase_		:       List[Any]				):
     # disable_tqdm in TrainingArguments has a flaky default since it depends on the level of logging. We make sure
     # its set to False since the tests later on depend on its value.
     snake_case_ :     int				=	RegressionDataset(length=lowercase_				)
     snake_case_ :     Any				=	RegressionDataset(length=lowercase_				)
     snake_case_ :     int				=	RegressionModelConfig(a=lowercase_  ,     b=lowercase_				)
     snake_case_ :     Tuple				=	RegressionPreTrainedModel(lowercase_				)

     snake_case_ :     Any				=	TrainingArguments(self.output_dir  ,     disable_tqdm=lowercase_  ,     report_to=[]  ,     **lowercase_				)
     return Trainer(
         lowercase_  ,     lowercase_  ,     train_dataset=lowercase_  ,     eval_dataset=lowercase_  ,     callbacks=lowercase_  ,     )

 def 						_snake_case					(     self		:       Optional[int]  ,     lowercase_		:       Any  ,     lowercase_		:       List[Any]				):
     self.assertEqual(len(lowercase_				)  ,     len(lowercase_				)				)

     # Order doesn't matter
     snake_case_ :     Any				=	sorted(lowercase_  ,     key=lambda lowercase_				: cb.__name__ if isinstance(lowercase_  ,     lowercase_				) else cb.__class__.__name__				)
     snake_case_ :     List[str]				=	sorted(lowercase_  ,     key=lambda lowercase_				: cb.__name__ if isinstance(lowercase_  ,     lowercase_				) else cb.__class__.__name__				)

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

 def 						_snake_case					(     self		:       Optional[Any]  ,     lowercase_		:       Tuple				):
     snake_case_ :     Tuple				=	['''on_init_end''', '''on_train_begin''']
     snake_case_ :     List[Any]				=	0
     snake_case_ :     Union[str, Any]				=	len(trainer.get_eval_dataloader()				)
     snake_case_ :     List[Any]				=	['''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(lowercase_				):
             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 						_snake_case					(     self		:       List[str]				):
     snake_case_ :     Union[str, Any]				=	self.get_trainer()
     snake_case_ :     Dict				=	DEFAULT_CALLBACKS.copy() + [ProgressCallback]
     self.check_callbacks_equality(trainer.callback_handler.callbacks  ,     lowercase_				)

     # Callbacks passed at init are added to the default callbacks
     snake_case_ :     Optional[Any]				=	self.get_trainer(callbacks=[MyTestTrainerCallback]				)
     expected_callbacks.append(lowercase_				)
     self.check_callbacks_equality(trainer.callback_handler.callbacks  ,     lowercase_				)

     # TrainingArguments.disable_tqdm controls if use ProgressCallback or PrinterCallback
     snake_case_ :     Optional[int]				=	self.get_trainer(disable_tqdm=lowercase_				)
     snake_case_ :     List[Any]				=	DEFAULT_CALLBACKS.copy() + [PrinterCallback]
     self.check_callbacks_equality(trainer.callback_handler.callbacks  ,     lowercase_				)

 def 						_snake_case					(     self		:       int				):
     snake_case_ :     int				=	DEFAULT_CALLBACKS.copy() + [ProgressCallback]
     snake_case_ :     List[Any]				=	self.get_trainer()

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

     snake_case_ :     Dict				=	self.get_trainer()
     snake_case_ :     Optional[int]				=	trainer.pop_callback(lowercase_				)
     self.assertEqual(cb.__class__  ,     lowercase_				)
     self.check_callbacks_equality(trainer.callback_handler.callbacks  ,     lowercase_				)

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

     # We can also add, pop, or remove by instance
     snake_case_ :     Optional[int]				=	self.get_trainer()
     snake_case_ :     List[Any]				=	trainer.callback_handler.callbacks[0]
     trainer.remove_callback(lowercase_				)
     expected_callbacks.remove(lowercase_				)
     self.check_callbacks_equality(trainer.callback_handler.callbacks  ,     lowercase_				)

     snake_case_ :     List[Any]				=	self.get_trainer()
     snake_case_ :     Optional[int]				=	trainer.callback_handler.callbacks[0]
     snake_case_ :     Optional[Any]				=	trainer.pop_callback(lowercase_				)
     self.assertEqual(lowercase_  ,     lowercase_				)
     self.check_callbacks_equality(trainer.callback_handler.callbacks  ,     lowercase_				)

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






 def 						_snake_case					(     self		:       List[Any]				):
     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=lowercase_				)

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

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

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

     snake_case_ :     List[Any]				=	self.get_trainer(callbacks=[MyTestTrainerCallback]  ,     eval_steps=5  ,     evaluation_strategy='''steps'''				)
     trainer.train()
     snake_case_ :     Union[str, Any]				=	trainer.callback_handler.callbacks[-2].events
     self.assertEqual(lowercase_  ,     self.get_expected_events(lowercase_				)				)

     snake_case_ :     Union[str, Any]				=	self.get_trainer(callbacks=[MyTestTrainerCallback]  ,     evaluation_strategy='''epoch'''				)
     trainer.train()
     snake_case_ :     Dict				=	trainer.callback_handler.callbacks[-2].events
     self.assertEqual(lowercase_  ,     self.get_expected_events(lowercase_				)				)

     # A bit of everything
     snake_case_ :     str				=	self.get_trainer(
         callbacks=[MyTestTrainerCallback]  ,     logging_steps=3  ,     save_steps=10  ,     eval_steps=5  ,     evaluation_strategy='''steps'''  ,     )
     trainer.train()
     snake_case_ :     str				=	trainer.callback_handler.callbacks[-2].events
     self.assertEqual(lowercase_  ,     self.get_expected_events(lowercase_				)				)

     # warning should be emitted for duplicated callbacks
     with patch('''transformers.trainer_callback.logger.warning'''				) as warn_mock:
         snake_case_ :     Dict				=	self.get_trainer(
             callbacks=[MyTestTrainerCallback, MyTestTrainerCallback]  ,     )
         assert str(lowercase_				) in warn_mock.call_args[0][0]





