class_docstr: Tests for verifying ProgressHooks.
      
class_docstr_tok: ["Tests","for","verifying","ProgressHooks","."]
code: def verify_trace(
      self,
      trace,
      pipeline_start_date,
      pipeline_end_date,
      expected_chunks,
      empty=False,
  ):
      # Percent complete should be monotonically increasing through the whole
      # execution.
      for before, after in toolz.sliding_window(2, trace):
          self.assertGreaterEqual(
              after.percent_complete,
              before.percent_complete,
          )
  
      # First publish should come from the start of the first chunk, with no
      # work yet.
      first = trace[0]
      expected_first = TestingProgressPublisher.TraceState(
          state="init",
          percent_complete=0.0,
          execution_bounds=(pipeline_start_date, pipeline_end_date),
          current_chunk_bounds=expected_chunks[0],
          current_work=None,
      )
      self.assertEqual(first, expected_first)
  
      # Last publish should have a state of success and be 100% complete.
      last = trace[-1]
      expected_last = TestingProgressPublisher.TraceState(
          state="success",
          percent_complete=100.0,
          execution_bounds=(pipeline_start_date, pipeline_end_date),
          current_chunk_bounds=expected_chunks[-1],
          # We don't know what the last work item will be, but it must be an
          # instance of a single ComputableTerm, because we only run
          # ComputableTerms one at a time, and a LoadableTerm will only be in
          # the graph if some ComputableTerm depends on it.
          #
          # The one exception to this rule is that, if we run a completely
          # empty pipeline, the final work will be None.
          current_work=None if empty else [instance_of(ComputableTerm)],
      )
      self.assertEqual(last, expected_last)
  
      # Remaining updates should all be loads or computes.
      middle = trace[1:-1]
      for update in middle:
          # For empty pipelines we never leave the 'init' state.
          if empty:
              self.assertEqual(update.state, "init")
              self.assertIs(update.current_work, None)
              continue
  
          if update.state in ("loading", "computing"):
              self.assertIsInstance(update.current_work, list)
          if update.state == "loading":
              for term in update.current_work:
                  self.assertIsInstance(
                      term, (LoadableTerm, AssetExists)
                  )
          elif update.state == "computing":
              for term in update.current_work:
                  self.assertIsInstance(term, ComputableTerm)
          else:
              raise AssertionError(
                  "Unexpected state: {}".format(update.state),
              )
  
      # Break up the remaining updates by chunk.
      all_chunks = []
      grouped = itertools.groupby(
          middle, attrgetter("current_chunk_bounds")
      )
      for (chunk_start, chunk_stop), chunk_trace in grouped:
          all_chunks.append((chunk_start, chunk_stop))
  
          chunk_trace = list(chunk_trace)
          expected_end_progress = self.expected_chunk_progress(
              pipeline_start_date,
              pipeline_end_date,
              chunk_stop,
          )
          end_progress = chunk_trace[-1].percent_complete
          assert_almost_equal(
              end_progress,
              expected_end_progress,
          )
  
      self.assertEqual(all_chunks, expected_chunks)
  
imports: import itertools
  from operator import attrgetter
  import numpy
  import pandas
  import toolz
  from zipline.pipeline import Pipeline
  from zipline.pipeline.classifiers import Everything
  from zipline.pipeline.domain import US_EQUITIES
  from zipline.pipeline.factors import CustomFactor
  from zipline.pipeline.data import Column, DataSet
  from zipline.pipeline.data.testing import TestingDataSet
  from zipline.pipeline.hooks.testing import TestingHooks
  from zipline.pipeline.hooks.progress import ProgressHooks, repr_htmlsafe, TestingProgressPublisher
  from zipline.pipeline.term import AssetExists, ComputableTerm, LoadableTerm
  from zipline.testing import parameter_space
  from zipline.testing.fixtures import ZiplineTestCase, WithSeededRandomPipelineEngine
  from zipline.testing.predicates import assert_almost_equal, instance_of
label_desc: Use image node_id 5 for calling the ProgressHooksTestCase obj's underlying member method code with example usage: obj.verify_trace(trace, pipeline_start_date, pipeline_end_date, expected_chunks, empty) without return types