loader_test.py 文件源码

def testMarkWithShrinkingBatchSize_raisesValueError(self):
    id_ = db.RUN_ROWID.create(1, 1)
    event1 = tf.Event(step=123)
    event2 = tf.Event(step=456)
    path1 = self._save_records('events.out.tfevents.1.localhost',
                               [event1.SerializeToString()])
    path2 = self._save_records('events.out.tfevents.2.localhost',
                               [event2.SerializeToString()])
    with self.connect_db() as db_conn:
      with self.EventLog(path1) as log1, self.EventLog(path2) as log2:
        with loader.RunReader(id_, 'doodle') as run:
          run.add_event_log(db_conn, log1)
          run.add_event_log(db_conn, log2)
          run.mark()
          self.assertEqual(event1, run.get_next_event())
          self.assertEqual(event2, run.get_next_event())
          self.assertIsNone(run.get_next_event())
          run.reset()
          self.assertEqual(event1, run.get_next_event())
          with six.assertRaisesRegex(self, ValueError, r'monotonic'):
            run.mark()