1 # Copyright 2015 The TensorFlow Authors. All Rights Reserved. 2 # 3 # Licensed under the Apache License, Version 2.0 (the "License"); 4 # you may not use this file except in compliance with the License. 5 # You may obtain a copy of the License at 6 # 7 # http://www.apache.org/licenses/LICENSE-2.0 8 # 9 # Unless required by applicable law or agreed to in writing, software 10 # distributed under the License is distributed on an "AS IS" BASIS, 11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 12 # See the License for the specific language governing permissions and 13 # limitations under the License. 14 # ============================================================================== 15 """Tests for tensorflow.python.framework.tensorboard_logging.""" 16 17 from __future__ import absolute_import 18 from __future__ import division 19 from __future__ import print_function 20 21 import glob 22 import os 23 import shutil 24 import tempfile 25 import time 26 27 from tensorflow.core.util import event_pb2 28 from tensorflow.python.platform import test 29 from tensorflow.python.platform import tf_logging as logging 30 from tensorflow.python.summary import summary_iterator 31 from tensorflow.python.summary.writer import writer 32 from tensorflow.python.training import tensorboard_logging 33 34 35 class EventLoggingTest(test.TestCase): 36 37 def setUp(self): 38 self._work_dir = tempfile.mkdtemp(dir=self.get_temp_dir()) 39 self._sw = writer.FileWriter(self._work_dir) 40 tensorboard_logging.set_summary_writer(self._sw) 41 self.addCleanup(shutil.rmtree, self._work_dir) 42 43 # Stop the clock to avoid test flakiness. 44 now = time.time() 45 time._real_time = time.time 46 time.time = lambda: now 47 48 # Mock out logging calls so we can verify that the right number of messages 49 # get logged. 50 self.logged_message_count = 0 51 self._actual_log = logging.log 52 53 def mockLog(*args, **kwargs): 54 self.logged_message_count += 1 55 self._actual_log(*args, **kwargs) 56 57 logging.log = mockLog 58 59 def tearDown(self): 60 time.time = time._real_time 61 logging.log = self._actual_log 62 63 def assertLoggedMessagesAre(self, expected_messages): 64 self._sw.close() 65 event_paths = glob.glob(os.path.join(self._work_dir, "event*")) 66 # If the tests runs multiple time in the same directory we can have 67 # more than one matching event file. We only want to read the last one. 68 self.assertTrue(event_paths) 69 event_reader = summary_iterator.summary_iterator(event_paths[-1]) 70 # Skip over the version event. 71 next(event_reader) 72 73 for level, message in expected_messages: 74 event = next(event_reader) 75 self.assertEqual(event.wall_time, time.time()) 76 self.assertEqual(event.log_message.level, level) 77 self.assertEqual(event.log_message.message, message) 78 79 def testBasic(self): 80 tensorboard_logging.set_summary_writer(self._sw) 81 tensorboard_logging.error("oh no!") 82 tensorboard_logging.error("for%s", "mat") 83 84 self.assertLoggedMessagesAre([(event_pb2.LogMessage.ERROR, "oh no!"), 85 (event_pb2.LogMessage.ERROR, "format")]) 86 self.assertEqual(2, self.logged_message_count) 87 88 def testVerbosity(self): 89 tensorboard_logging.set_summary_writer(self._sw) 90 tensorboard_logging.set_verbosity(tensorboard_logging.ERROR) 91 tensorboard_logging.warn("warn") 92 tensorboard_logging.error("error") 93 tensorboard_logging.set_verbosity(tensorboard_logging.DEBUG) 94 tensorboard_logging.debug("debug") 95 96 self.assertLoggedMessagesAre([(event_pb2.LogMessage.ERROR, "error"), 97 (event_pb2.LogMessage.DEBUGGING, "debug")]) 98 # All message should be logged because tensorboard_logging verbosity doesn't 99 # affect logging verbosity. 100 self.assertEqual(3, self.logged_message_count) 101 102 def testBadVerbosity(self): 103 with self.assertRaises(ValueError): 104 tensorboard_logging.set_verbosity("failure") 105 106 with self.assertRaises(ValueError): 107 tensorboard_logging.log("bad", "dead") 108 109 def testNoSummaryWriter(self): 110 """Test that logging without a SummaryWriter succeeds.""" 111 tensorboard_logging.set_summary_writer(None) 112 tensorboard_logging.warn("this should work") 113 self.assertEqual(1, self.logged_message_count) 114 115 def testSummaryWriterFailsAfterClear(self): 116 tensorboard_logging._clear_summary_writer() 117 with self.assertRaises(RuntimeError): 118 tensorboard_logging.log(tensorboard_logging.ERROR, "failure") 119 120 121 if __name__ == "__main__": 122 test.main() 123
