(self, iteration_log)
| 82 | |
| 83 | @parameterized.expand([[True], [get_event_filter([1, 3])]]) |
| 84 | def test_loss_print(self, iteration_log): |
| 85 | log_stream = StringIO() |
| 86 | log_handler = logging.StreamHandler(log_stream) |
| 87 | log_handler.setLevel(logging.INFO) |
| 88 | key_to_handler = "test_logging" |
| 89 | key_to_print = "myLoss" |
| 90 | |
| 91 | # set up engine |
| 92 | def _train_func(engine, batch): |
| 93 | return [torch.tensor(0.0)] |
| 94 | |
| 95 | engine = Engine(_train_func) |
| 96 | |
| 97 | # set up testing handler |
| 98 | logger = logging.getLogger(key_to_handler) |
| 99 | logger.setLevel(logging.INFO) |
| 100 | logger.addHandler(log_handler) |
| 101 | stats_handler = StatsHandler( |
| 102 | iteration_log=iteration_log, epoch_log=False, name=key_to_handler, tag_name=key_to_print |
| 103 | ) |
| 104 | stats_handler.attach(engine) |
| 105 | |
| 106 | num_iters = 3 |
| 107 | max_epochs = 2 |
| 108 | engine.run(range(num_iters), max_epochs=max_epochs) |
| 109 | |
| 110 | # check logging output |
| 111 | output_str = log_stream.getvalue() |
| 112 | log_handler.close() |
| 113 | has_key_word = re.compile(f".*{key_to_print}.*") |
| 114 | content_count = 0 |
| 115 | for line in output_str.split("\n"): |
| 116 | if has_key_word.match(line): |
| 117 | content_count += 1 |
| 118 | if iteration_log is True: |
| 119 | self.assertEqual(content_count, num_iters * max_epochs) |
| 120 | else: |
| 121 | self.assertEqual(content_count, 2) # 2 = len([1, 3]) from event_filter |
| 122 | |
| 123 | def test_loss_dict(self): |
| 124 | log_stream = StringIO() |
nothing calls this directly
no test coverage detected