Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
"""Unit test of logging_manager (batch)."""
caplog.set_level(logging.INFO)
emmental.init()
Meta.update_config(
config={
"logging_config": {
"counter_unit": "batch",
"evaluation_freq": 2,
"checkpointing": True,
"checkpointer_config": {"checkpoint_freq": 2},
}
}
)
logging_manager = LoggingManager(n_batches_per_epoch=5)
logging_manager.update(5)
assert logging_manager.trigger_evaluation() is False
assert logging_manager.trigger_checkpointing() is False
logging_manager.update(5)
assert logging_manager.trigger_evaluation() is True
assert logging_manager.trigger_checkpointing() is False
logging_manager.update(10)
assert logging_manager.trigger_evaluation() is False
assert logging_manager.trigger_checkpointing() is False
logging_manager.update(5)
assert logging_manager.trigger_evaluation() is True
assert logging_manager.trigger_checkpointing() is True
caplog.set_level(logging.INFO)
emmental.init()
Meta.update_config(
config={
"logging_config": {
"counter_unit": "epoch",
"evaluation_freq": 1,
"checkpointing": False,
"checkpointer_config": {"checkpoint_freq": 2},
"writer_config": {"writer": "json"},
}
}
)
logging_manager = LoggingManager(n_batches_per_epoch=2)
logging_manager.update(5)
assert type(logging_manager.writer) == LogWriter
caplog.set_level(logging.INFO)
emmental.init()
Meta.update_config(
config={
"logging_config": {
"counter_unit": "epoch",
"evaluation_freq": 1,
"checkpointing": False,
"checkpointer_config": {"checkpoint_freq": 2},
"writer_config": {"writer": "tensorboard"},
}
}
)
logging_manager = LoggingManager(n_batches_per_epoch=2)
logging_manager.update(5)
assert type(logging_manager.writer) == TensorBoardWriter
emmental.init()
Meta.update_config(
config={
"logging_config": {
"counter_unit": "epoch",
"evaluation_freq": 1,
"checkpointing": False,
"checkpointer_config": {"checkpoint_freq": 2},
"writer_config": {"writer": "a"},
}
}
)
with pytest.raises(ValueError):
logging_manager = LoggingManager(n_batches_per_epoch=2)
logging_manager.update(5)
caplog.set_level(logging.INFO)
emmental.init()
Meta.update_config(
config={
"logging_config": {
"counter_unit": "epochs",
"evaluation_freq": 1,
"checkpointing": False,
"checkpointer_config": {"checkpoint_freq": 2},
}
}
)
with pytest.raises(ValueError):
logging_manager = LoggingManager(n_batches_per_epoch=2)
logging_manager.update(5)
caplog.set_level(logging.INFO)
emmental.init()
Meta.update_config(
config={
"logging_config": {
"counter_unit": "epoch",
"evaluation_freq": 1,
"checkpointing": False,
"checkpointer_config": {"checkpoint_freq": 2},
"writer_config": {"writer": None},
}
}
)
logging_manager = LoggingManager(n_batches_per_epoch=2)
logging_manager.update(5)
assert logging_manager.writer is None
emmental.init()
Meta.update_config(
config={
"meta_config": {"verbose": False},
"logging_config": {
"counter_unit": "epoch",
"evaluation_freq": 1,
"checkpointing": False,
"checkpointer_config": {"checkpoint_freq": 2},
"writer_config": {"writer": "json"},
},
}
)
logging_manager = LoggingManager(n_batches_per_epoch=2)
logging_manager.update(5)
assert logging_manager.trigger_evaluation() is False
assert logging_manager.trigger_checkpointing() is False
logging_manager.update(5)
assert logging_manager.trigger_evaluation() is True
assert logging_manager.trigger_checkpointing() is False
logging_manager.update(10)
assert logging_manager.trigger_evaluation() is False
assert logging_manager.trigger_checkpointing() is False
logging_manager.update(5)
assert logging_manager.trigger_evaluation() is True
assert logging_manager.trigger_checkpointing() is False
caplog.set_level(logging.INFO)
emmental.init()
Meta.update_config(
config={
"meta_config": {"verbose": False},
"logging_config": {
"counter_unit": "epoch",
"evaluation_freq": 1,
"checkpointing": True,
"checkpointer_config": {"checkpoint_freq": 2},
},
}
)
logging_manager = LoggingManager(n_batches_per_epoch=2)
logging_manager.update(5)
assert logging_manager.trigger_evaluation() is False
assert logging_manager.trigger_checkpointing() is False
logging_manager.update(5)
assert logging_manager.trigger_evaluation() is True
assert logging_manager.trigger_checkpointing() is False
logging_manager.update(10)
assert logging_manager.trigger_evaluation() is False
assert logging_manager.trigger_checkpointing() is False
logging_manager.update(5)
assert logging_manager.trigger_evaluation() is True
assert logging_manager.trigger_checkpointing() is True
Meta.reset()
emmental.init()
Meta.update_config(
config={
"logging_config": {
"counter_unit": "sample",
"evaluation_freq": 10,
"checkpointing": True,
"checkpointer_config": {"checkpoint_freq": 2},
}
}
)
logging_manager = LoggingManager(n_batches_per_epoch=10)
logging_manager.update(5)
assert logging_manager.trigger_evaluation() is False
assert logging_manager.trigger_checkpointing() is False
logging_manager.update(5)
assert logging_manager.trigger_evaluation() is True
assert logging_manager.trigger_checkpointing() is False
logging_manager.update(10)
assert logging_manager.trigger_evaluation() is True
assert logging_manager.trigger_checkpointing() is True
logging_manager.update(5)
assert logging_manager.trigger_evaluation() is False
assert logging_manager.trigger_checkpointing() is False