
import inspect
import os

import torch
from transformers import AutoModel
from transformers.testing_utils import mockenv_context
from transformers.trainer_utils import set_seed

import accelerate
from accelerate.accelerator import Accelerator
from accelerate.state import AcceleratorState
from accelerate.test_utils.testing import (
    AccelerateTestCase,
    TempDirTestCase,
    execute_subprocess_async,
    require_cuda,
    require_fsdp,
    require_multi_gpu,
    slow,
)
from accelerate.utils.constants import (
    FSDP_AUTO_WRAP_POLICY,
    FSDP_BACKWARD_PREFETCH,
    FSDP_SHARDING_STRATEGY,
    FSDP_STATE_DICT_TYPE,
)
from accelerate.utils.dataclasses import FullyShardedDataParallelPlugin
from accelerate.utils.other import patch_environment


set_seed(42)

SCREAMING_SNAKE_CASE__   :			Union[str, Any]												=    "bert-base-cased"
SCREAMING_SNAKE_CASE__   :			int												=    "fp16"
SCREAMING_SNAKE_CASE__   :			List[Any]												=    "bf16"
SCREAMING_SNAKE_CASE__   :			Optional[int]												=    [FPaa, BFaa]

@require_fsdp
@require_cuda
class    lowerCAmelCase__     (	__lowercase			):






		def  __A       (		self		:							Any  )     ->							Optional[int]:
						super().setUp()

						__lowerCamelCase    					=	dict(
						    ACCELERATE_USE_FSDP='''true'''							,			MASTER_ADDR='''localhost'''							,			MASTER_PORT='''10999'''							,			RANK='''0'''							,			LOCAL_RANK='''0'''							,			WORLD_SIZE='''1'''							,			)






		def  __A       (		self		:							List[Any]  )     ->							Union[str, Any]:
						from torch.distributed.fsdp.fully_sharded_data_parallel import ShardingStrategy

						for i, strategy in enumerate(SCREAMING_SNAKE_CASE__  ):
										__lowerCamelCase    					=	self.dist_env.copy()
										__lowerCamelCase    					=	f'''{i + 1}'''
										__lowerCamelCase    					=	strategy
										with mockenv_context(**SCREAMING_SNAKE_CASE__  ):
														__lowerCamelCase    					=	FullyShardedDataParallelPlugin()
														self.assertEqual(fsdp_plugin.sharding_strategy							,			ShardingStrategy(i + 1  )  )






		def  __A       (		self		:							Union[str, Any]  )     ->							Optional[Any]:
						from torch.distributed.fsdp.fully_sharded_data_parallel import BackwardPrefetch

						for i, prefetch_policy in enumerate(SCREAMING_SNAKE_CASE__  ):
										__lowerCamelCase    					=	self.dist_env.copy()
										__lowerCamelCase    					=	prefetch_policy
										with mockenv_context(**SCREAMING_SNAKE_CASE__  ):
														__lowerCamelCase    					=	FullyShardedDataParallelPlugin()
														if prefetch_policy == "NO_PREFETCH":
																		self.assertIsNone(fsdp_plugin.backward_prefetch  )
														else:
																		self.assertEqual(fsdp_plugin.backward_prefetch							,			BackwardPrefetch(i + 1  )  )






		def  __A       (		self		:							Union[str, Any]  )     ->							Optional[int]:
						from torch.distributed.fsdp.fully_sharded_data_parallel import StateDictType

						for i, state_dict_type in enumerate(SCREAMING_SNAKE_CASE__  ):
										__lowerCamelCase    					=	self.dist_env.copy()
										__lowerCamelCase    					=	state_dict_type
										with mockenv_context(**SCREAMING_SNAKE_CASE__  ):
														__lowerCamelCase    					=	FullyShardedDataParallelPlugin()
														self.assertEqual(fsdp_plugin.state_dict_type							,			StateDictType(i + 1  )  )
														if state_dict_type == "FULL_STATE_DICT":
																		self.assertTrue(fsdp_plugin.state_dict_config.offload_to_cpu  )
																		self.assertTrue(fsdp_plugin.state_dict_config.ranka_only  )






		def  __A       (		self		:							List[str]  )     ->							Dict:
						__lowerCamelCase    					=	AutoModel.from_pretrained(SCREAMING_SNAKE_CASE__  )
						for policy in FSDP_AUTO_WRAP_POLICY:
										__lowerCamelCase    					=	self.dist_env.copy()
										__lowerCamelCase    					=	policy
										if policy == "TRANSFORMER_BASED_WRAP":
														__lowerCamelCase    					=	'''BertLayer'''
										elif policy == "SIZE_BASED_WRAP":
														__lowerCamelCase    					=	'''2000'''
										with mockenv_context(**SCREAMING_SNAKE_CASE__  ):
														__lowerCamelCase    					=	FullyShardedDataParallelPlugin()
														fsdp_plugin.set_auto_wrap_policy(SCREAMING_SNAKE_CASE__  )
														if policy == "NO_WRAP":
																		self.assertIsNone(fsdp_plugin.auto_wrap_policy  )
														else:
																		self.assertIsNotNone(fsdp_plugin.auto_wrap_policy  )

						__lowerCamelCase    					=	self.dist_env.copy()
						__lowerCamelCase    					=	'''TRANSFORMER_BASED_WRAP'''
						__lowerCamelCase    					=	'''T5Layer'''
						with mockenv_context(**SCREAMING_SNAKE_CASE__  ):
										__lowerCamelCase    					=	FullyShardedDataParallelPlugin()
										with self.assertRaises(SCREAMING_SNAKE_CASE__  ) as cm:
														fsdp_plugin.set_auto_wrap_policy(SCREAMING_SNAKE_CASE__  )
										self.assertTrue('''Could not find the transformer layer class to wrap in the model.''' in str(cm.exception  )  )

						__lowerCamelCase    					=	self.dist_env.copy()
						__lowerCamelCase    					=	'''SIZE_BASED_WRAP'''
						__lowerCamelCase    					=	'''0'''
						with mockenv_context(**SCREAMING_SNAKE_CASE__  ):
										__lowerCamelCase    					=	FullyShardedDataParallelPlugin()
										fsdp_plugin.set_auto_wrap_policy(SCREAMING_SNAKE_CASE__  )
										self.assertIsNone(fsdp_plugin.auto_wrap_policy  )






		def  __A       (		self		:							Tuple  )     ->							Union[str, Any]:
						from torch.distributed.fsdp.fully_sharded_data_parallel import MixedPrecision
						from torch.distributed.fsdp.sharded_grad_scaler import ShardedGradScaler

						for mp_dtype in dtypes:
										__lowerCamelCase    					=	self.dist_env.copy()
										__lowerCamelCase    					=	mp_dtype
										with mockenv_context(**SCREAMING_SNAKE_CASE__  ):
														__lowerCamelCase    					=	Accelerator()
														if mp_dtype == "fp16":
																		__lowerCamelCase    					=	torch.floataa
														elif mp_dtype == "bf16":
																		__lowerCamelCase    					=	torch.bfloataa
														__lowerCamelCase    					=	MixedPrecision(param_dtype=SCREAMING_SNAKE_CASE__							,			reduce_dtype=SCREAMING_SNAKE_CASE__							,			buffer_dtype=SCREAMING_SNAKE_CASE__  )
														self.assertEqual(accelerator.state.fsdp_plugin.mixed_precision_policy							,			SCREAMING_SNAKE_CASE__  )
														if mp_dtype == FPaa:
																		self.assertTrue(isinstance(accelerator.scaler							,			SCREAMING_SNAKE_CASE__  )  )
														elif mp_dtype == BFaa:
																		self.assertIsNone(accelerator.scaler  )
														AcceleratorState._reset_state(SCREAMING_SNAKE_CASE__  )







		def  __A       (		self		:							Tuple  )     ->							Optional[int]:
						from torch.distributed.fsdp.fully_sharded_data_parallel import CPUOffload

						for flag in [True, False]:
										__lowerCamelCase    					=	self.dist_env.copy()
										__lowerCamelCase    					=	str(SCREAMING_SNAKE_CASE__  ).lower()
										with mockenv_context(**SCREAMING_SNAKE_CASE__  ):
														__lowerCamelCase    					=	FullyShardedDataParallelPlugin()
														self.assertEqual(fsdp_plugin.cpu_offload							,			CPUOffload(offload_params=SCREAMING_SNAKE_CASE__  )  )





@require_fsdp
@require_multi_gpu
@slow
class    lowerCAmelCase__     (	__lowercase			):






		def  __A       (		self		:							str  )     ->							Union[str, Any]:
						super().setUp()
						__lowerCamelCase    					=	0.82
						__lowerCamelCase    					=	[
						    '''fsdp_shard_grad_op_transformer_based_wrap''',
						    '''fsdp_full_shard_transformer_based_wrap''',
						]
						__lowerCamelCase    					=	{
						    '''multi_gpu_fp16''': 32_00,
						    '''fsdp_shard_grad_op_transformer_based_wrap_fp16''': 20_00,
						    '''fsdp_full_shard_transformer_based_wrap_fp16''': 19_00,
						    # Disabling below test as it overwhelms the RAM memory usage
						    # on CI self-hosted runner leading to tests getting killed.
						    # "fsdp_full_shard_cpu_offload_transformer_based_wrap_fp32": 1500,  # fp16 was leading to indefinite hang
						}
						__lowerCamelCase    					=	1_60
						__lowerCamelCase    					=	1_60

						__lowerCamelCase    					=	inspect.getfile(accelerate.test_utils  )
						__lowerCamelCase    					=	os.path.sep.join(mod_file.split(os.path.sep  )[:-1] + ['''scripts''', '''external_deps''']  )






		def  __A       (		self		:							Optional[int]  )     ->							Any:
						__lowerCamelCase    					=	os.path.join(self.test_scripts_folder							,			'''test_performance.py'''  )
						__lowerCamelCase    					=	['''accelerate''', '''launch''', '''--num_processes=2''', '''--num_machines=1''', '''--machine_rank=0''', '''--use_fsdp''']
						for config in self.performance_configs:
										__lowerCamelCase    					=	cmd.copy()
										for i, strategy in enumerate(SCREAMING_SNAKE_CASE__  ):
														if strategy.lower() in config:
																		cmd_config.append(f'''--fsdp_sharding_strategy={i+1}'''  )
																		break

										if "fp32" in config:
														cmd_config.append('''--mixed_precision=no'''  )
										else:
														cmd_config.append('''--mixed_precision=fp16'''  )

										if "cpu_offload" in config:
														cmd_config.append('''--fsdp_offload_params=True'''  )

										for policy in FSDP_AUTO_WRAP_POLICY:
														if policy.lower() in config:
																		cmd_config.append(f'''--fsdp_auto_wrap_policy={policy}'''  )
																		break

										if policy == "TRANSFORMER_BASED_WRAP":
														cmd_config.append('''--fsdp_transformer_layer_cls_to_wrap=BertLayer'''  )
										elif policy == "SIZE_BASED_WRAP":
														cmd_config.append('''--fsdp_min_num_params=2000'''  )

										cmd_config.extend(
										    [
										        self.test_file_path,
										        f'''--output_dir={self.tmpdir}''',
										        f'''--performance_lower_bound={self.performance_lower_bound}''',
										    ]  )
										with patch_environment(omp_num_threads=1  ):
														execute_subprocess_async(SCREAMING_SNAKE_CASE__							,			env=os.environ.copy()  )






		def  __A       (		self		:							Tuple  )     ->							List[str]:
						__lowerCamelCase    					=	os.path.join(self.test_scripts_folder							,			'''test_checkpointing.py'''  )
						__lowerCamelCase    					=	[
						    '''accelerate''',
						    '''launch''',
						    '''--num_processes=2''',
						    '''--num_machines=1''',
						    '''--machine_rank=0''',
						    '''--use_fsdp''',
						    '''--mixed_precision=fp16''',
						    '''--fsdp_transformer_layer_cls_to_wrap=BertLayer''',
						]

						for i, strategy in enumerate(SCREAMING_SNAKE_CASE__  ):
										__lowerCamelCase    					=	cmd.copy()
										cmd_config.append(f'''--fsdp_sharding_strategy={i+1}'''  )
										if strategy != "FULL_SHARD":
														continue
										__lowerCamelCase    					=	len(SCREAMING_SNAKE_CASE__  )
										for state_dict_type in FSDP_STATE_DICT_TYPE:
														__lowerCamelCase    					=	cmd_config[:state_dict_config_index]
														cmd_config.append(f'''--fsdp_state_dict_type={state_dict_type}'''  )
														cmd_config.extend(
														    [
														        self.test_file_path,
														        f'''--output_dir={self.tmpdir}''',
														        '''--partial_train_epoch=1''',
														    ]  )
														with patch_environment(omp_num_threads=1  ):
																		execute_subprocess_async(SCREAMING_SNAKE_CASE__							,			env=os.environ.copy()  )

														__lowerCamelCase    					=	cmd_config[:-1]
														__lowerCamelCase    					=	os.path.join(self.tmpdir							,			'''epoch_0'''  )
														cmd_config.extend(
														    [
														        f'''--resume_from_checkpoint={resume_from_checkpoint}''',
														    ]  )
														with patch_environment(omp_num_threads=1  ):
																		execute_subprocess_async(SCREAMING_SNAKE_CASE__							,			env=os.environ.copy()  )







		def  __A       (		self		:							Union[str, Any]  )     ->							Any:
						__lowerCamelCase    					=	os.path.join(self.test_scripts_folder							,			'''test_peak_memory_usage.py'''  )
						__lowerCamelCase    					=	[
						    '''accelerate''',
						    '''launch''',
						    '''--num_processes=2''',
						    '''--num_machines=1''',
						    '''--machine_rank=0''',
						]
						for spec, peak_mem_upper_bound in self.peak_memory_usage_upper_bound.items():
										__lowerCamelCase    					=	cmd.copy()
										if "fp16" in spec:
														cmd_config.extend(['''--mixed_precision=fp16''']  )
										else:
														cmd_config.extend(['''--mixed_precision=no''']  )

										if "multi_gpu" in spec:
														continue
										else:
														cmd_config.extend(['''--use_fsdp''']  )
														for i, strategy in enumerate(SCREAMING_SNAKE_CASE__  ):
																		if strategy.lower() in spec:
																						cmd_config.append(f'''--fsdp_sharding_strategy={i+1}'''  )
																						break

														if "cpu_offload" in spec:
																		cmd_config.append('''--fsdp_offload_params=True'''  )

														for policy in FSDP_AUTO_WRAP_POLICY:
																		if policy.lower() in spec:
																						cmd_config.append(f'''--fsdp_auto_wrap_policy={policy}'''  )
																						break

														if policy == "TRANSFORMER_BASED_WRAP":
																		cmd_config.append('''--fsdp_transformer_layer_cls_to_wrap=BertLayer'''  )
														elif policy == "SIZE_BASED_WRAP":
																		cmd_config.append('''--fsdp_min_num_params=2000'''  )

										cmd_config.extend(
										    [
										        self.test_file_path,
										        f'''--output_dir={self.tmpdir}''',
										        f'''--peak_memory_upper_bound={peak_mem_upper_bound}''',
										        f'''--n_train={self.n_train}''',
										        f'''--n_val={self.n_val}''',
										    ]  )
										with patch_environment(omp_num_threads=1  ):
														execute_subprocess_async(SCREAMING_SNAKE_CASE__							,			env=os.environ.copy()  )





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






		def __init__(		self		:							int  )     ->							Optional[int]:
						__lowerCamelCase    					=	[]






		def  __A       (		self		:							Any							,			SCREAMING_SNAKE_CASE__		:							List[Any]							,			SCREAMING_SNAKE_CASE__		:							List[Any]							,			SCREAMING_SNAKE_CASE__		:							str							,			**SCREAMING_SNAKE_CASE__		:							List[Any]  )     ->							Tuple:
						self.events.append('''on_init_end'''  )






		def  __A       (		self		:							Tuple							,			SCREAMING_SNAKE_CASE__		:							Optional[int]							,			SCREAMING_SNAKE_CASE__		:							Optional[Any]							,			SCREAMING_SNAKE_CASE__		:							Optional[int]							,			**SCREAMING_SNAKE_CASE__		:							Optional[Any]  )     ->							Any:
						self.events.append('''on_train_begin'''  )






		def  __A       (		self		:							str							,			SCREAMING_SNAKE_CASE__		:							int							,			SCREAMING_SNAKE_CASE__		:							Any							,			SCREAMING_SNAKE_CASE__		:							Dict							,			**SCREAMING_SNAKE_CASE__		:							int  )     ->							Any:
						self.events.append('''on_train_end'''  )






		def  __A       (		self		:							Union[str, Any]							,			SCREAMING_SNAKE_CASE__		:							Dict							,			SCREAMING_SNAKE_CASE__		:							List[Any]							,			SCREAMING_SNAKE_CASE__		:							List[Any]							,			**SCREAMING_SNAKE_CASE__		:							List[Any]  )     ->							int:
						self.events.append('''on_epoch_begin'''  )






		def  __A       (		self		:							Optional[Any]							,			SCREAMING_SNAKE_CASE__		:							int							,			SCREAMING_SNAKE_CASE__		:							Any							,			SCREAMING_SNAKE_CASE__		:							Optional[Any]							,			**SCREAMING_SNAKE_CASE__		:							Tuple  )     ->							Optional[Any]:
						self.events.append('''on_epoch_end'''  )






		def  __A       (		self		:							str							,			SCREAMING_SNAKE_CASE__		:							Tuple							,			SCREAMING_SNAKE_CASE__		:							Optional[Any]							,			SCREAMING_SNAKE_CASE__		:							Union[str, Any]							,			**SCREAMING_SNAKE_CASE__		:							str  )     ->							List[Any]:
						self.events.append('''on_step_begin'''  )






		def  __A       (		self		:							Any							,			SCREAMING_SNAKE_CASE__		:							List[Any]							,			SCREAMING_SNAKE_CASE__		:							int							,			SCREAMING_SNAKE_CASE__		:							Any							,			**SCREAMING_SNAKE_CASE__		:							Optional[int]  )     ->							Tuple:
						self.events.append('''on_step_end'''  )






		def  __A       (		self		:							int							,			SCREAMING_SNAKE_CASE__		:							List[str]							,			SCREAMING_SNAKE_CASE__		:							List[str]							,			SCREAMING_SNAKE_CASE__		:							int							,			**SCREAMING_SNAKE_CASE__		:							List[Any]  )     ->							Union[str, Any]:
						self.events.append('''on_evaluate'''  )






		def  __A       (		self		:							Union[str, Any]							,			SCREAMING_SNAKE_CASE__		:							List[str]							,			SCREAMING_SNAKE_CASE__		:							List[str]							,			SCREAMING_SNAKE_CASE__		:							str							,			**SCREAMING_SNAKE_CASE__		:							str  )     ->							str:
						self.events.append('''on_predict'''  )






		def  __A       (		self		:							Optional[int]							,			SCREAMING_SNAKE_CASE__		:							int							,			SCREAMING_SNAKE_CASE__		:							List[Any]							,			SCREAMING_SNAKE_CASE__		:							Optional[int]							,			**SCREAMING_SNAKE_CASE__		:							Optional[Any]  )     ->							Tuple:
						self.events.append('''on_save'''  )






		def  __A       (		self		:							Optional[int]							,			SCREAMING_SNAKE_CASE__		:							Dict							,			SCREAMING_SNAKE_CASE__		:							Union[str, Any]							,			SCREAMING_SNAKE_CASE__		:							Union[str, Any]							,			**SCREAMING_SNAKE_CASE__		:							List[str]  )     ->							List[str]:
						self.events.append('''on_log'''  )







		def  __A       (		self		:							int							,			SCREAMING_SNAKE_CASE__		:							List[Any]							,			SCREAMING_SNAKE_CASE__		:							List[str]							,			SCREAMING_SNAKE_CASE__		:							List[Any]							,			**SCREAMING_SNAKE_CASE__		:							Optional[int]  )     ->							str:
						self.events.append('''on_prediction_step'''  )





@require_torch
class    lowerCAmelCase__     (	unittest.TestCase			):






		def  __A       (		self		:							Union[str, Any]  )     ->							Optional[Any]:
						__lowerCamelCase    					=	tempfile.mkdtemp()






		def  __A       (		self		:							int  )     ->							List[str]:
						shutil.rmtree(self.output_dir  )






		def  __A       (		self		:							Tuple							,			SCREAMING_SNAKE_CASE__		:							List[Any]=0							,			SCREAMING_SNAKE_CASE__		:							Any=0							,			SCREAMING_SNAKE_CASE__		:							List[str]=64							,			SCREAMING_SNAKE_CASE__		:							Optional[int]=64							,			SCREAMING_SNAKE_CASE__		:							Optional[int]=None							,			SCREAMING_SNAKE_CASE__		:							Optional[int]=False							,			**SCREAMING_SNAKE_CASE__		:							Optional[Any]  )     ->							Union[str, 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.
						__lowerCamelCase    					=	RegressionDataset(length=SCREAMING_SNAKE_CASE__  )
						__lowerCamelCase    					=	RegressionDataset(length=SCREAMING_SNAKE_CASE__  )
						__lowerCamelCase    					=	RegressionModelConfig(a=SCREAMING_SNAKE_CASE__							,			b=SCREAMING_SNAKE_CASE__  )
						__lowerCamelCase    					=	RegressionPreTrainedModel(SCREAMING_SNAKE_CASE__  )

						__lowerCamelCase    					=	TrainingArguments(self.output_dir							,			disable_tqdm=SCREAMING_SNAKE_CASE__							,			report_to=[]							,			**SCREAMING_SNAKE_CASE__  )
						return Trainer(
						    SCREAMING_SNAKE_CASE__							,			SCREAMING_SNAKE_CASE__							,			train_dataset=SCREAMING_SNAKE_CASE__							,			eval_dataset=SCREAMING_SNAKE_CASE__							,			callbacks=SCREAMING_SNAKE_CASE__							,			)






		def  __A       (		self		:							List[Any]							,			SCREAMING_SNAKE_CASE__		:							List[str]							,			SCREAMING_SNAKE_CASE__		:							str  )     ->							Optional[int]:
						self.assertEqual(len(SCREAMING_SNAKE_CASE__  )							,			len(SCREAMING_SNAKE_CASE__  )  )

						# Order doesn't matter
						__lowerCamelCase    					=	sorted(SCREAMING_SNAKE_CASE__							,			key=lambda SCREAMING_SNAKE_CASE__  : cb.__name__ if isinstance(SCREAMING_SNAKE_CASE__							,			SCREAMING_SNAKE_CASE__  ) else cb.__class__.__name__  )
						__lowerCamelCase    					=	sorted(SCREAMING_SNAKE_CASE__							,			key=lambda SCREAMING_SNAKE_CASE__  : cb.__name__ if isinstance(SCREAMING_SNAKE_CASE__							,			SCREAMING_SNAKE_CASE__  ) else cb.__class__.__name__  )

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






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

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

						# TrainingArguments.disable_tqdm controls if use ProgressCallback or PrinterCallback
						__lowerCamelCase    					=	self.get_trainer(disable_tqdm=SCREAMING_SNAKE_CASE__  )
						__lowerCamelCase    					=	DEFAULT_CALLBACKS.copy() + [PrinterCallback]
						self.check_callbacks_equality(trainer.callback_handler.callbacks							,			SCREAMING_SNAKE_CASE__  )






		def  __A       (		self		:							List[Any]  )     ->							str:
						__lowerCamelCase    					=	DEFAULT_CALLBACKS.copy() + [ProgressCallback]
						__lowerCamelCase    					=	self.get_trainer()

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

						__lowerCamelCase    					=	self.get_trainer()
						__lowerCamelCase    					=	trainer.pop_callback(SCREAMING_SNAKE_CASE__  )
						self.assertEqual(cb.__class__							,			SCREAMING_SNAKE_CASE__  )
						self.check_callbacks_equality(trainer.callback_handler.callbacks							,			SCREAMING_SNAKE_CASE__  )

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

						# We can also add, pop, or remove by instance
						__lowerCamelCase    					=	self.get_trainer()
						__lowerCamelCase    					=	trainer.callback_handler.callbacks[0]
						trainer.remove_callback(SCREAMING_SNAKE_CASE__  )
						expected_callbacks.remove(SCREAMING_SNAKE_CASE__  )
						self.check_callbacks_equality(trainer.callback_handler.callbacks							,			SCREAMING_SNAKE_CASE__  )

						__lowerCamelCase    					=	self.get_trainer()
						__lowerCamelCase    					=	trainer.callback_handler.callbacks[0]
						__lowerCamelCase    					=	trainer.pop_callback(SCREAMING_SNAKE_CASE__  )
						self.assertEqual(SCREAMING_SNAKE_CASE__							,			SCREAMING_SNAKE_CASE__  )
						self.check_callbacks_equality(trainer.callback_handler.callbacks							,			SCREAMING_SNAKE_CASE__  )

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







		def  __A       (		self		:							Union[str, Any]  )     ->							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=SCREAMING_SNAKE_CASE__  )

						__lowerCamelCase    					=	self.get_trainer(callbacks=[MyTestTrainerCallback]  )
						trainer.train()
						__lowerCamelCase    					=	trainer.callback_handler.callbacks[-2].events
						self.assertEqual(SCREAMING_SNAKE_CASE__							,			self.get_expected_events(SCREAMING_SNAKE_CASE__  )  )

						# Independent log/save/eval
						__lowerCamelCase    					=	self.get_trainer(callbacks=[MyTestTrainerCallback]							,			logging_steps=5  )
						trainer.train()
						__lowerCamelCase    					=	trainer.callback_handler.callbacks[-2].events
						self.assertEqual(SCREAMING_SNAKE_CASE__							,			self.get_expected_events(SCREAMING_SNAKE_CASE__  )  )

						__lowerCamelCase    					=	self.get_trainer(callbacks=[MyTestTrainerCallback]							,			save_steps=5  )
						trainer.train()
						__lowerCamelCase    					=	trainer.callback_handler.callbacks[-2].events
						self.assertEqual(SCREAMING_SNAKE_CASE__							,			self.get_expected_events(SCREAMING_SNAKE_CASE__  )  )

						__lowerCamelCase    					=	self.get_trainer(callbacks=[MyTestTrainerCallback]							,			eval_steps=5							,			evaluation_strategy='''steps'''  )
						trainer.train()
						__lowerCamelCase    					=	trainer.callback_handler.callbacks[-2].events
						self.assertEqual(SCREAMING_SNAKE_CASE__							,			self.get_expected_events(SCREAMING_SNAKE_CASE__  )  )

						__lowerCamelCase    					=	self.get_trainer(callbacks=[MyTestTrainerCallback]							,			evaluation_strategy='''epoch'''  )
						trainer.train()
						__lowerCamelCase    					=	trainer.callback_handler.callbacks[-2].events
						self.assertEqual(SCREAMING_SNAKE_CASE__							,			self.get_expected_events(SCREAMING_SNAKE_CASE__  )  )

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

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



