class_docstr: Tests for verifying that SimplePipelineEngine calls hooks as expected.
      
class_docstr_tok: ["Tests","for","verifying","that","SimplePipelineEngine","calls","hooks","as","expected","."]
docstr: Verify a trace of a Pipeline execution.
          
docstr_tok: ["Verify","a","trace","of","a","Pipeline","execution","."]
code: def verify_trace(
      self,
      trace,
      pipeline,
      pipeline_start_date,
      pipeline_end_date,
      expected_loads,
      expected_computes,
      expected_chunks,
  ):
      """Verify a trace of a Pipeline execution."""
      # First/last calls should bracket the pipeline execution.
      self.expect_context_pair(trace[0], trace[-1], "running_pipeline")
      self.assertEqual(
          trace[0].args,
          (pipeline, pipeline_start_date, pipeline_end_date),
      )
  
      # Break up the trace into the traces of each chunk.
      chunk_traces = self.split_by_chunk(trace[1:-1])
  
      for ctrace, (chunk_start, chunk_end) in zip(
          chunk_traces, expected_chunks
      ):
          # Next call should bracket compute_chunk
          self.expect_context_pair(
              ctrace[0], ctrace[-1], "computing_chunk"
          )
          self.assertIsInstance(ctrace[0].args[0], list)  # terms
          self.assertEqual(ctrace[0].args[1:], (chunk_start, chunk_end))
  
          # Remainder of calls should be loads and computes. These have to
          # happen in dependency order, but we don't bother to assert that
          # here. We just make sure that we see each expected load/compute
          # exactly once.
          loads_and_computes = ctrace[1:-1]
          loads = set()
          computes = set()
          for enter, exit_ in two_at_a_time(loads_and_computes):
              self.expect_context_pair(enter, exit_, method=None)
  
              if enter.method_name == "loading_terms":
                  for loaded_term in enter.args[0]:
                      # We should only see each term once.
                      self.assertNotIn(loaded_term, loads)
                      # Don't worry about domains here.
                      loads.add(loaded_term.unspecialize())
              elif enter.method_name == "computing_term":
                  computed_term = enter.args[0]
                  self.assertNotIn(computed_term, computes)
                  computes.add(computed_term)
              else:
                  raise ValueError(
                      "Unexpected method: {}".format(enter.method_name)
                  )
  
          self.assertEqual(loads, expected_loads)
          self.assertEqual(computes, expected_computes)
  
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 4 for calling the HooksTestCase obj's underlying member method code with example usage: obj.verify_trace(trace, pipeline, pipeline_start_date, pipeline_end_date, expected_loads, expected_computes, expected_chunks) without return types