distributions_plugin_test.py 文件源码

def _test_distributions(self, run_name, tag_name, should_work=True):
    self.set_up_with_runs([self._RUN_WITH_SCALARS,
                           self._RUN_WITH_LEGACY_DISTRIBUTION,
                           self._RUN_WITH_DISTRIBUTION])
    if should_work:
      (data, mime_type) = self.plugin.distributions_impl(tag_name, run_name)
      self.assertEqual('application/json', mime_type)
      self.assertEqual(len(data), self._STEPS)
      for i in xrange(self._STEPS):
        [_unused_wall_time, step, bps_and_icdfs] = data[i]
        self.assertEqual(i, step)
        (bps, _unused_icdfs) = zip(*bps_and_icdfs)
        self.assertEqual(bps, compressor.NORMAL_HISTOGRAM_BPS)
    else:
      with six.assertRaisesRegex(self, ValueError, 'No histogram tag'):
        self.plugin.distributions_impl(self._DISTRIBUTION_TAG, run_name)