diff --git a/.coveragerc b/.coveragerc
deleted file mode 100644
index 92cab5e..0000000
--- a/.coveragerc
+++ /dev/null
@@ -1,5 +0,0 @@
-[run]
-include = ./packages/dataflow/*
-
-[expect_tests]
-expected_coverage_min = 70
diff --git a/README.md b/README.md
deleted file mode 100644
index 5179635..0000000
--- a/README.md
+++ /dev/null
@@ -1,135 +0,0 @@
-# packages/dataflow
-
-The purpose of this package is to simplify the development of Dataflow
-workflows.
-
-See the modules in [common](./common/README.md) for some generally useful
-abstractions.
-
-You'll notice that workflows are included in this package. Workflows are located
-here to simplify job execution, as all
-non-[standard](https://beam.apache.org/documentation/) and
-non-[beam](https://beam.apache.org/documentation/) modules must be packaged
-together for job execution.
-
-See
-[Scheduling a Dataflow workflow](https://chromium.googlesource.com/infra/infra/+/master/doc/users/event_pipeline.md#scheduling-a-dataflow-workflow)
-for more information on automating a workflow you'd like to run regularly.
-
-It's possible that you may only care about running your pipeline locally. In
-that case, you can simply import the common modules.
-
-Note that Dataflow supports continuous pipelines. Chrome Operations hasn't
-experimented with these yet, but they are worth exploring!
-
-[TOC]
-
-# References
-
-[Beam Docs](https://beam.apache.org/documentation/)
-
-# Unit Testing
-
-From the root of the infra repository, run the command `./test.py test
-packages/dataflow`.
-
-# Workflow Testing
-
-There are a couple requirements to testing your Dataflow workflow.
-
-First, you must activate the infra Python environment. Assuming you have that
-set up already, run `source ENV/bin/activate` from the root of your infra
-checkout. If you need to set up or update your environment, see
-[bootstrap/README](../../bootstrap/README.md) for more info.
-
-Next, you must have Google Storage buckets to pass with the `--staging_location`
-and `--temp_location` options. The name is not important, but for example you
-could use `gs://my-dataflow-job/staging`.
-[Create these](https://cloud.google.com/storage/docs/creating-buckets) if you
-don't have them already.
-
-Next, you must have permission within the project to schedule a Dataflow job,
-and be authenticated to do so.
-
-To check your authentication status, ensure that you have the Cloud SDK
-[installed](https://cloud.google.com/sdk/docs/quickstarts), then run `gcloud
-info`.
-
-If you don't see the correct project ID, reach out to an
-[editor](https://pantheon.corp.google.com/iam-admin/iam) of that project to
-request access.
-
-Finally, run the command below to test your workflow as a remote job. Note: Job
-names should match the regular expression `[a-z]\([-a-z0-9]{0,38}[a-z0-9])`.
-
-```
-python <path-to-dataflow-job> --job_name <pick-a-job-name> \
---project <project> --runner DataflowRunner \
---setup_file <infra-checkout-path>/packages/dataflow/setup.py \
---staging_location <staging bucket> \
---temp_location <temp bucket> --save_main_session
-```
-
-Navigate to the [Dataflow console](https://console.cloud.google.com/project) in
-your browser and you should see your job running. Wait until it succeeds.
-
-Running the test will leave behind a directory,
-`packages/dataflow/dataflow.egg-info`, that you must manually clean up.
-
-To run the workflow locally, first set credentials using `export
-GOOGLE_APPLICATION_CREDENTIALS=<path_to_credentials>`
-
-Then `python cq_attempts.py --output <dummy_path> --project
-<name_of_test_project>`
-
-# Updating the package
-
-Changes to this directory are automatically mirrored in a synthesized
-[repo](https://chromium.googlesource.com/infra/infra/packages/dataflow/). To
-deploy changes to this repository:
-
- * Land the changes.
- * Submit a separate CL that updates the version in `setup.py`.
- * Build and upload a new wheel.
- * Submit a single CL that updates the remote execution recipe and deps.pyl.
-
-Jobs scheduled with the
-[`remote_execute_dataflow_workflow`](../../recipes/recipes/remote_execute_dataflow_workflow.py)
-recipe use the version of the job at HEAD but the version of the package pinned
-in [bootstrap/deps.pyl](../../bootstrap/deps.pyl). So, if you make a breaking
-change to the package, submit the update first (which will automatically be
-picked up by the
-[package mirror](https://chromium.googlesource.com/infra/infra/packages/dataflow/)),
-then submit the change to the job along with the ref update in `deps.pyl` together
-in one commit. Be sure to follow the instructions in
-[bootstrap/README.md](../../bootstrap/README.md) to build and upload the new
-wheel before submitting the change to `deps.pyl`.
-
-# Limits
-
-Please see the [Dataflow docs](https://cloud.google.com/dataflow/quotas) for the
-most up to date information on quotas and limits.
-
-At the time of writing, there are limits on Dataflow requests per minute, number
-of GCE instances (`--numWorkers`), number of concurrent jobs, monitoring requests,
-job creation request size, and number of side input shards.
-
-Some of these limits are per user, others are per project, others are per
-organization. In general, users and project owners are responsible for ensuring
-they do not hit user and project limits. Currently these limits are, in general,
-not restrictive for us, and we are not concerned about hitting them.
-
-## Interaction with other cloud services
-
-If you run a job on GCE, GCE limits apply. If you use BigQuery as a source or
-sink, BigQuery limits apply. Project owners should monitor usage and be mindful
-of these limits.
-
-### BigQuery
-
-The relevant limits are query and insert limits.
-[Query](https://cloud.google.com/bigquery/quotas#queries) and
-[insert](https://cloud.google.com/bigquery/quotas#streaminginserts) limits can
-be found in the Dataflow documentation. As query and insert rates are highly
-project-specific, project owners should be responsible for monitoring this
-limit.
diff --git a/__init__.py b/__init__.py
deleted file mode 100644
index e69de29..0000000
--- a/__init__.py
+++ /dev/null
diff --git a/common/README.md b/common/README.md
deleted file mode 100644
index 56b1c6b..0000000
--- a/common/README.md
+++ /dev/null
@@ -1,30 +0,0 @@
-The `common` module provides reusable classes for writing Dataflow workflows.
-
-## chops\_beam
-
-For easily constructing readable pipelines with standard defaults.
-
-```
-q = ('SELECT blah FROM `example_project.example_dataset.example_table`')
-p = chops_beam.EventsPipeline()
-_ = (p
-     | chops_beam.BQRead(q)
-     | ... # do some transforms
-     | chops_beam.BQWrite('example_project', 'destination_table'))
-p.run()
-```
-
-## objects
-
-Convenient classes for BigQuery tables. Can be used for reading from or writing
-to BigQuery.
-
-```
-for row in input_rows:
-  event = objects.CQEvent.from_bigquery_row(row)
-```
-
-## combine\_fns
-
-Generally useful [Combine
-Functions](https://beam.apache.org/documentation/programming-guide/#transforms-combine).
diff --git a/common/__init__.py b/common/__init__.py
deleted file mode 100644
index e69de29..0000000
--- a/common/__init__.py
+++ /dev/null
diff --git a/common/chops_beam.py b/common/chops_beam.py
deleted file mode 100644
index 4a9036c..0000000
--- a/common/chops_beam.py
+++ /dev/null
@@ -1,44 +0,0 @@
-# Copyright 2017 The Chromium Authors. All rights reserved.
-# Use of this source code is governed by a BSD-style license that can be
-# found in the LICENSE file.
-
-import apache_beam as beam
-
-from apache_beam.pipeline import PipelineOptions
-
-
-class EventsPipeline(beam.Pipeline):
-  """Pipeline that reads options from the command line."""
-  def __init__(self):
-    super(EventsPipeline, self).__init__(options=PipelineOptions())
-
-
-class BQRead(beam.io.iobase.Read):
-  """Read transform created from a BigQuerySource with convenient defaults."""
-
-  def __init__(self, query, validate=True, coder=None, use_standard_sql=True,
-               flatten_results=False):
-    """
-    Args:
-      query: The query to be run. Should specify table in
-        `project.dataset.table` form for standard SQL and
-        [project:dataset.table] form is use_standard_sql is False.
-      See beam.io.BigQuerySource for explanation of remaining arguments.
-    """
-    source = beam.io.BigQuerySource(query=query, validate=validate, coder=coder,
-                                    flatten_results=flatten_results,
-                                    use_standard_sql=use_standard_sql)
-    super(BQRead, self).__init__(source)
-
-
-class BQWrite(beam.io.Write):
-  """Write transform created from a BigQuerySink with convenient defaults.
-
-  beam.io.BigQuerySink will automatically add unique insert IDs to rows,
-  which BigQuery uses to prevent duplicate inserts.
-  """
-  def __init__(self, project, table, dataset='aggregated',
-               write_disposition=beam.io.BigQueryDisposition.WRITE_TRUNCATE):
-    sink = beam.io.BigQuerySink(table, dataset, project,
-                                write_disposition=write_disposition)
-    super(BQWrite, self).__init__(sink)
diff --git a/common/combine_fns.py b/common/combine_fns.py
deleted file mode 100644
index 2ecd6fc..0000000
--- a/common/combine_fns.py
+++ /dev/null
@@ -1,51 +0,0 @@
-# Copyright 2017 The Chromium Authors. All rights reserved.
-# Use of this source code is governed by a BSD-style license that can be
-# found in the LICENSE file
-
-import apache_beam as beam
-
-
-class ConvertToCSV(beam.CombineFn):
-  """Convert elements to CSV format to be written out.
-
-  Transform for writing elements out in a CSV format. Can process elements
-  of type dictonary or list. This transform only supports consistent elements,
-  meaning dictionaries must all have the same keys and lists must have the
-  same length and order. If provided a header, a list must already be in that
-  order and a dictionary must have at least those fields.
-  """
-  def __init__(self, header=None):
-    self.header = header
-
-  def create_accumulator(self):
-    return []
-
-  def iterable(self, obj):
-   """Returns an iterable for a dictionary or list.
-
-   Sorting dictionary keys assures that the CSV is in the same order
-   for all dictionaries. If given a header, use the header fields as
-   the iterable for the dictionary to preserve that order.
-   """
-   if isinstance(obj, dict):
-     if self.header:
-       return self.header
-     keys = obj.keys()
-     return sorted(keys)
-   else:
-     return range(len(obj))
-
-  def add_input(self, accumulator, element):
-    element_string = []
-    for i in self.iterable(element):
-      element_string.append(str(element[i]))
-    accumulator.append(','.join(element_string) + '\n')
-    return accumulator
-
-  def merge_accumulators(self, accumulators):
-    return sum(accumulators, [])
-
-  def extract_output(self, accumulator):
-    if self.header:
-      accumulator.insert(0, ','.join(self.header) + '\n')
-    return ''.join(accumulator)
diff --git a/common/objects.py b/common/objects.py
deleted file mode 100644
index 8045790..0000000
--- a/common/objects.py
+++ /dev/null
@@ -1,172 +0,0 @@
-# Copyright 2017 The Chromium Authors. All rights reserved.
-# Use of this source code is governed by a BSD-style license that can be
-# found in the LICENSE file.
-
-
-class BigQueryObject(object):
-  """A BigQueryObject holds data that will be read from/written to BigQuery."""
-
-  def __eq__(self, other):
-    return self.__dict__ == other.__dict__
-
-  @staticmethod
-  def get_bigquery_attributes():
-    """Returns a list of attributes that exist in the BigQuery schema.
-
-       These attributes should be strings.
-    """
-    raise NotImplementedError()
-
-  def as_bigquery_row(self):
-    """Returns data in a suitable format for writing to BigQuery.
-
-       The default behavior constructs a dictionary from the attributes listed
-       by get_bigquery_attributes and their values. This behavior can be
-       overridden.
-    """
-    return {attr: self.__dict__.get(attr)
-            for attr in self.get_bigquery_attributes()}
-
-  @classmethod
-  def from_bigquery_row(cls, row):
-    """Creates an instance of cls from a BigQuery row.
-
-       Args:
-         row: dictionary in the form {field: value} where field is in
-         get_bigquery_attributes().
-    """
-    obj = cls()
-    for field, value in row.items():
-      obj.__dict__[field] = value
-    return obj
-
-
-class CQAttempt(BigQueryObject):
-  """A CQAttempt represents a single CQ attempt.
-
-     It is created by aggregating all CQEvents for a given attempt.
-  """
-  def __init__(self):
-    # Consistent between events for a given attempt
-    self.attempt_start_msec = None
-    self.cq_name = None
-    self.issue = None
-    self.patchset = None
-    self.dry_run = False
-
-    # Patch event timestamps
-    self.first_start_msec = None
-    self.last_start_msec = None
-    self.first_stop_msec = None
-    self.last_stop_msec = None
-    self.patch_committed_msec = None
-    self.patch_started_to_commit_msec = None
-    self.patch_failed_msec = None
-    self.vcs_commit_latency_sec = None
-    self.click_to_failure_sec = None
-    self.click_to_patch_committed_sec = None
-    self.click_to_result_sec = None
-
-    # Patch event bools
-    self.committed = False
-    self.was_throttled = False
-    self.waited_for_tree = False
-    self.failed = False
-
-    # Verifier event timestamps
-    self.first_verifier_trigger_msec = None
-    self.patch_verifier_pass_msec = None
-    self.cq_launch_latency_sec = None
-    self.verifier_pass_latency_sec = None
-    self.tree_check_and_throttle_latency_sec = None
-
-    # Verifier event bools
-    self.no_tryjobs_launched = False
-    self.custom_trybots = False
-
-    self.failure_reason = None
-    self.max_failure_msec = None
-    self.fail_type = None
-
-    self.infra_failures = 0
-    self.compile_failures = 0
-    self.test_failures = 0
-    self.invalid_test_results_failures = 0
-    self.patch_failures = 0
-    self.total_failures = 0
-
-    self.contributing_bbucket_ids = None
-    self.earliest_equivalent_patchset = None
-    self.attempt_key = None
-
-  @staticmethod
-  def get_bigquery_attributes():
-    return [
-        'attempt_start_msec',
-        'first_start_msec',
-        'last_start_msec',
-        'cq_name',
-        'first_stop_msec',
-        'last_stop_msec',
-        'committed',
-        'was_throttled',
-        'waited_for_tree',
-        'issue',
-        'patchset',
-        'dry_run',
-        'cq_launch_latency_sec',
-        'verifier_pass_latency_sec',
-        'tree_check_and_throttle_latency_sec',
-        'no_tryjobs_launched',
-        'custom_trybots',
-        'failed',
-        'infra_failures',
-        'compile_failures',
-        'test_failures',
-        'invalid_test_results_failures',
-        'patch_failures',
-        'total_failures',
-        'fail_type',
-        'contributing_bbucket_ids',
-        'vcs_commit_latency_sec',
-        'click_to_patch_committed_sec',
-        'click_to_failure_sec',
-        'click_to_result_sec',
-        'earliest_equivalent_patchset',
-        'attempt_key',
-    ]
-
-
-class CQEvent(BigQueryObject):
-  """A CQEvent represents event data reported to BigQuery from CQ.
-
-     CQEvents are aggregated to make CQAttempts.
-  """
-  def __init__(self):
-    self.timestamp_millis = None
-    self.action = None
-    self.attempt_start_usec = None
-    self.cq_name = None
-    self.issue = None
-    self.patchset = None
-    self.failure_reason = None
-    self.dry_run = False
-    self.contributing_buildbucket_ids = None
-    self.earliest_equivalent_patchset = None
-    self.attempt_key = None
-
-  @staticmethod
-  def get_bigquery_attributes():
-    return [
-        'timestamp_millis',
-        'action',
-        'attempt_start_usec',
-        'cq_name',
-        'issue',
-        'patchset',
-        'dry_run',
-        'failure_reason',
-        'contributing_buildbucket_ids',
-        'earliest_equivalent_patchset',
-        'attempt_key',
-    ]
diff --git a/common/test/__init__.py b/common/test/__init__.py
deleted file mode 100644
index e69de29..0000000
--- a/common/test/__init__.py
+++ /dev/null
diff --git a/common/test/combine_fns_test.py b/common/test/combine_fns_test.py
deleted file mode 100644
index a5833dc..0000000
--- a/common/test/combine_fns_test.py
+++ /dev/null
@@ -1,44 +0,0 @@
-# Copyright 2017 The Chromium Authors. All rights reserved.
-# Use of this source code is governed by a BSD-style license that can be
-# found in the LICENSE file.
-
-import unittest
-
-import apache_beam as beam
-
-from apache_beam.testing import test_pipeline
-from apache_beam.testing import util
-from dataflow.common import combine_fns
-
-
-class TestCombineFns(unittest.TestCase):
-  def test_convert_to_CSV_with_dicts(self):
-    elements = {'a': 1, 'b': 2, 'c': 3}
-    pipeline = test_pipeline.TestPipeline()
-    result = (pipeline
-              | beam.Create([elements, elements])
-              | beam.CombineGlobally(combine_fns.ConvertToCSV())
-    )
-    util.assert_that(result, util.equal_to(['1,2,3\n1,2,3\n']))
-    pipeline.run()
-
-  def test_convert_to_CSV_with_lists(self):
-    elements = [1, 2, 3]
-    pipeline = test_pipeline.TestPipeline()
-    result = (pipeline
-              | beam.Create([elements, elements])
-              | beam.CombineGlobally(combine_fns.ConvertToCSV())
-    )
-    util.assert_that(result, util.equal_to(['1,2,3\n1,2,3\n']))
-    pipeline.run()
-
-  def test_convert_to_CSV_with_header(self):
-    elements = {'a': 1, 'b': 2, 'c': 3}
-    header = ['a', 'b', 'c']
-    pipeline = test_pipeline.TestPipeline()
-    result = (pipeline
-              | beam.Create([elements, elements])
-              | beam.CombineGlobally(combine_fns.ConvertToCSV(header))
-    )
-    util.assert_that(result, util.equal_to(['a,b,c\n1,2,3\n1,2,3\n']))
-    pipeline.run()
diff --git a/cq_attempts.py b/cq_attempts.py
deleted file mode 100644
index 05775e6..0000000
--- a/cq_attempts.py
+++ /dev/null
@@ -1,248 +0,0 @@
-# Copyright 2017 The Chromium Authors. All rights reserved.
-# Use of this source code is governed by a BSD-style license that can be
-# found in the LICENSE file.
-
-import logging
-import time
-
-import apache_beam as beam
-
-from dataflow.common import chops_beam
-from dataflow.common import objects
-
-
-class CombineEventsToAttempt(beam.CombineFn):
-  ACTION_PATCH_START = 'PATCH_START'
-  ACTION_PATCH_COMMITTED = 'PATCH_COMMITTED'
-  ACTION_PATCH_COMMITTING = 'PATCH_COMMITTING'
-  ACTION_PATCH_STOP = 'PATCH_STOP'
-  ACTION_PATCH_THROTTLED = 'PATCH_THROTTLED'
-  ACTION_PATCH_TREE_CLOSED = 'PATCH_TREE_CLOSED'
-  ACTION_VERIFIER_TRIGGER = 'VERIFIER_TRIGGER'
-  ACTION_VERIFIER_PASS = 'VERIFIER_PASS'
-  ACTION_VERIFIER_NOTRY = 'VERIFIER_NOTRY'
-  ACTION_VERIFIER_CUSTOM_TRYBOTS = 'VERIFIER_CUSTOM_TRYBOTS'
-  ACTION_PATCH_FAILED = 'PATCH_FAILED'
-  # Try job fail types
-  FAIL_TYPE_PATCH = 'FAIL_TYPE_PATCH'
-  FAIL_TYPE_INFRA = 'FAIL_TYPE_INFRA'
-  FAIL_TYPE_COMPILE = 'FAIL_TYPE_COMPILE'
-  FAIL_TYPE_TEST = 'FAIL_TYPE_TEST'
-  FAIL_TYPE_INVALID = 'FAIL_TYPE_INVALID'
-
-  def __init__(self):
-    super(CombineEventsToAttempt, self).__init__()
-    self.action_affects_fields = {
-        self.ACTION_PATCH_START: set(['first_start_msec', 'last_start_msec']),
-        self.ACTION_PATCH_STOP: set(['first_stop_msec', 'last_stop_msec']),
-        self.ACTION_PATCH_COMMITTED: set(['patch_committed_msec', 'committed']),
-        self.ACTION_PATCH_COMMITTING: set(['patch_started_to_commit_msec']),
-        self.ACTION_PATCH_THROTTLED: set(['was_throttled']),
-        self.ACTION_PATCH_TREE_CLOSED: set(['waited_for_tree']),
-        self.ACTION_VERIFIER_TRIGGER: set(['first_verifier_trigger_msec']),
-        self.ACTION_VERIFIER_PASS: set(['patch_verifier_pass_msec']),
-        self.ACTION_VERIFIER_NOTRY: set(['no_tryjobs_launched']),
-        self.ACTION_VERIFIER_CUSTOM_TRYBOTS: set(['custom_trybots']),
-        self.ACTION_PATCH_FAILED: set(['patch_failed_msec', 'failed',
-                                       'failure_reason']),
-    }
-    self.min_timestamp_fields = set([
-        'first_start_msec',
-        'first_stop_msec',
-        'patch_committed_msec',
-        'patch_started_to_commit_msec',
-        'first_verifier_trigger_msec',
-        'patch_verifier_pass_msec',
-        'patch_failed_msec'
-    ])
-    self.max_timestamp_fields = set([
-        'last_start_msec',
-        'last_stop_msec',
-    ])
-    self.logical_or_fields = set([
-        'committed',
-        'was_throttled',
-        'waited_for_tree',
-        'failed',
-        'custom_trybots',
-    ])
-    # Fields that are copied from event to attempt. Values for these fields are
-    # the same for all events for a given attempt.
-    self.consistent_fields = set([
-        'cq_name',
-        'issue',
-        'patchset',
-        'dry_run',
-    ])
-
-  @staticmethod
-  def choose_min(old, new):
-    if new is not None and (old is None or new < old):
-      return new
-    return old
-
-  @staticmethod
-  def compute_difference(minuend, subtrahend):
-    if minuend is None or subtrahend is None:
-      return None
-    return minuend - subtrahend
-
-  @staticmethod
-  def ms_to_sec(ms):
-    return ms / 1000.0 if ms is not None else None
-
-  def create_accumulator(self):
-    return []
-
-  def add_input(self, accumulator, input_rows):
-    for row in input_rows:
-      event = objects.CQEvent.from_bigquery_row(row)
-      if event.attempt_start_usec is None:
-        logging.warn('recieved row with null attempt_start_usec: %s', row)
-        continue
-
-      if event.timestamp_millis is None:
-        logging.warn('recieved raw with null timestamp: %s', row)
-        continue
-
-      accumulator.append(event)
-    return accumulator
-
-  def merge_accumulators(self, accumulators):
-    merged = self.create_accumulator()
-    for a in list(accumulators):
-      merged += a
-    return merged
-
-  def extract_output(self, accumulator):
-    attempt = objects.CQAttempt()
-
-    for event in accumulator:
-      attempt_start_msec = float(event.attempt_start_usec) / 1000
-      if (attempt.attempt_start_msec and
-          attempt.attempt_start_msec != attempt_start_msec):
-        logging.error(('tried to combine events with different '
-                       'attempt_start_msec'))
-        return
-
-      attempt.attempt_start_msec = attempt_start_msec
-
-      # Here we search for the last-reported value of some field.
-      if (event.failure_reason and (attempt.max_failure_msec is None
-           or event.timestamp_millis > attempt.max_failure_msec)):
-        attempt.failure_reason = event.failure_reason
-        attempt.max_failure_msec = event.timestamp_millis
-        attempt.fail_type = attempt.failure_reason['fail_type']
-      if event.contributing_buildbucket_ids:
-        attempt_bb_ids = set(attempt.contributing_bbucket_ids or [])
-        attempt_bb_ids.update(event.contributing_buildbucket_ids)
-        attempt.contributing_bbucket_ids = sorted(attempt_bb_ids)
-      if event.earliest_equivalent_patchset:
-        attempt.earliest_equivalent_patchset = (
-            event.earliest_equivalent_patchset)
-      if event.attempt_key:
-        attempt.attempt_key = event.attempt_key
-
-      for field in self.consistent_fields:
-        attempt_value = attempt.__dict__.get(field)
-        event_value = event.__dict__.get(field)
-        if attempt_value and attempt_value  != event_value:
-          logging.error('tried to combine events with inconsistent %s', field)
-          return
-        attempt.__dict__[field] = event_value
-
-      affected_fields = self.action_affects_fields.get(event.action, [])
-      for field in affected_fields:
-        if field in self.min_timestamp_fields:
-          attempt.__dict__[field] = self.choose_min(attempt.__dict__.get(field),
-                                                    event.timestamp_millis)
-        if field in self.max_timestamp_fields:
-          attempt.__dict__[field] = max(attempt.__dict__.get(field),
-                                        event.timestamp_millis)
-        if field in self.logical_or_fields:
-          attempt.__dict__[field] = True
-
-    attempt.cq_launch_latency_sec = self.ms_to_sec(
-        self.compute_difference(attempt.first_verifier_trigger_msec,
-                                attempt.attempt_start_msec))
-
-    attempt.verifier_pass_latency_sec = self.ms_to_sec(
-        self.compute_difference(attempt.patch_verifier_pass_msec,
-                                attempt.attempt_start_msec))
-
-    attempt.tree_check_and_throttle_latency_sec = self.ms_to_sec(
-        self.compute_difference(attempt.patch_started_to_commit_msec,
-                                attempt.patch_verifier_pass_msec))
-
-    attempt.vcs_commit_latency_sec = self.ms_to_sec(
-        self.compute_difference(attempt.patch_committed_msec,
-                                attempt.patch_started_to_commit_msec))
-
-    attempt.click_to_failure_sec = self.ms_to_sec(
-        self.compute_difference(attempt.patch_failed_msec,
-                                attempt.attempt_start_msec))
-
-    attempt.click_to_patch_committed_sec = self.ms_to_sec(
-        self.compute_difference(attempt.patch_committed_msec,
-                                attempt.attempt_start_msec))
-
-    attempt.click_to_result_sec = self.ms_to_sec(
-        self.compute_difference(attempt.last_stop_msec,
-                                attempt.attempt_start_msec))
-
-    # TODO: Deprecate. Now that we have contributing buildbucket ids, we can
-    # join with completed_builds to get this information.
-    if attempt.failure_reason:
-      for job in attempt.failure_reason.get('failed_try_jobs', []):
-        fail_type = job['fail_type']
-        attempt.total_failures += 1
-        if fail_type == self.FAIL_TYPE_INFRA:
-          attempt.infra_failures += 1
-        if fail_type == self.FAIL_TYPE_COMPILE:
-          attempt.compile_failures += 1
-        if fail_type == self.FAIL_TYPE_TEST:
-          attempt.test_failures += 1
-        if fail_type == self.FAIL_TYPE_INVALID:
-          attempt.invalid_test_results_failures += 1
-        if fail_type == self.FAIL_TYPE_PATCH:
-          attempt.patch_failures += 1
-    return attempt.as_bigquery_row()
-
-
-class ComputeAttempts(beam.PTransform):
-  @staticmethod
-  def key(event):
-    parts = [event.get('attempt_start_usec'), event.get('cq_name'),
-             event.get('issue'), event.get('patchset')]
-    return ':'.join([str(part) or '' for part in parts])
-
-  @staticmethod
-  def filter_incomplete_attempts(attempt):
-    if (attempt is not None and attempt.get('first_start_msec')
-        and attempt.get('last_stop_msec')):
-      yield attempt
-
-  def expand(self, pcoll):
-    return (pcoll
-            | beam.Map(lambda e: (self.key(e), e))
-            | beam.GroupByKey()
-            | beam.CombinePerKey(CombineEventsToAttempt())
-            | beam.Map(lambda (k, v): v)
-            | beam.FlatMap(self.filter_incomplete_attempts))
-
-
-def main():
-  q = ('SELECT timestamp_millis, action, attempt_start_usec, cq_name, issue,'
-       '  patchset, dry_run, failure_reason, contributing_buildbucket_ids, '
-       '  earliest_equivalent_patchset, attempt_key '
-       'FROM `chrome-infra-events.raw_events.cq`')
-  p = chops_beam.EventsPipeline()
-  _ = (p
-       | chops_beam.BQRead(q)
-       | ComputeAttempts()
-       | chops_beam.BQWrite('chrome-infra-events', 'cq_attempts'))
-  p.run()
-
-
-if __name__ == '__main__':
-  main()
diff --git a/new_cq_attempts.py b/new_cq_attempts.py
deleted file mode 100644
index fd32303..0000000
--- a/new_cq_attempts.py
+++ /dev/null
@@ -1,126 +0,0 @@
-# Copyright 2019 The Chromium Authors. All rights reserved.
-# Use of this source code is governed by a BSD-style license that can be
-# found in the LICENSE file.
-
-import apache_beam as beam
-
-from dataflow import cq_attempts as sanitize_cq_attempts
-from dataflow.common import chops_beam
-
-
-class ExtractBuildBucketIdFn(beam.DoFn):
-  def process(self, cq_attempt_with_key):
-    # For a CQ attempt, we create one row for each contributing BuildBucket id.
-    key = cq_attempt_with_key[0]
-    cq_attempt_dict = cq_attempt_with_key[1]
-
-    bb_ids = cq_attempt_dict.get('contributing_bbucket_ids')
-    if bb_ids:
-      for bb_id in bb_ids:
-        yield str(bb_id), key
-
-class FilterJoinedBuildBucketCQAttempt(beam.DoFn):
-  def process(self, joined_result):
-    # The key is BuildBucket ID. We expect there to be exactly 1 cq_attempt, and
-    # up to 1 BuildBucket entry.
-    cq_attempt_key = joined_result[1]['cq_attempt_key']
-    bb_entry = joined_result[1]['bb_entries']
-    if len(bb_entry) != 1 or len(cq_attempt_key) != 1:
-      return
-    yield cq_attempt_key[0], bb_entry[0]
-
-def update_with_presubmit_failure(input_tuple):
-  value = input_tuple[1]
-  cq_attempts = value['cq_attempts']
-  assert len(cq_attempts) == 1, "There must be 1 cq_attempt."
-  cq_attempt = cq_attempts[0]
-
-  if cq_attempt['fail_type'] == 'FAILED_JOBS':
-    buildbucket_results = value['bb_entries']
-    presubmit_failures = 0
-    other_failures = 0
-    for bb_result in buildbucket_results:
-      if (bb_result['status'] == 'FAILURE' and
-          bb_result['builder'] == 'chromium_presubmit'):
-        presubmit_failures += 1
-      elif bb_result['status'] != 'SUCCESS':
-        other_failures += 1
-    if presubmit_failures >= 1 and other_failures == 0:
-      cq_attempt['fail_type'] = 'FAILED_PRESUBMIT_BOT'
-
-  # Dictionaries are supposed to be returned in a single element list.
-  return [cq_attempt]
-
-def process_input(cq_events_pcol, bb_entries_pcol):
-  """Sets up the pipeline stages to return aggregated cq attempts pcol.
-
-  This function performs two tasks:
-    1) Computes CQ attempts from raw CQ events. This includes data sanitization.
-    2) If a CQ attempt fails only because of 'chromium_presubmit' builder, sets
-       the failure status to 'FAILED_PRESUBMIT_BOT'.
-  """
-  # Pcol of cq_attempt_as_dict
-  sanitized_cq_attempts = (
-      cq_events_pcol | sanitize_cq_attempts.ComputeAttempts())
-
-  # Create Pcol of tuples: (cq_attempt_key, cq_attempt_as_dict)
-  def extract_key(cq_attempt_dict):
-    key_parts = [
-      cq_attempt_dict.get('attempt_start_msec'),
-      cq_attempt_dict.get('cq_name'),
-      cq_attempt_dict.get('issue'),
-      cq_attempt_dict.get('patchset')
-    ]
-    key = ':'.join([str(part) or '' for part in key_parts])
-    return key, cq_attempt_dict
-  cq_attempts_with_key = sanitized_cq_attempts | beam.Map(extract_key)
-
-  # Create Pcol of tuples: (build_bucket_id, cq_attempt_key)
-  cq_attempt_key_keyed_by_bb_id = cq_attempts_with_key | beam.ParDo(
-      ExtractBuildBucketIdFn())
-
-  # Create Pcol of tuples: (build_bucket_id, build_bucket_entry)
-  bb_entry_keyed_by_bb_id = bb_entries_pcol | beam.Map(
-      lambda e: (str(e.get('id')), e))
-
-  # Create Pcol of tuples: (cq_attempt_key, BuildBucket entry)
-  bb_entries_keyed_by_cq_attempt_key = ({
-    'bb_entries' : bb_entry_keyed_by_bb_id,
-    'cq_attempt_key': cq_attempt_key_keyed_by_bb_id
-  } | 'Join BuildBucket with cq attempts' >> beam.CoGroupByKey()
-    | beam.ParDo(FilterJoinedBuildBucketCQAttempt())
-  )
-
-  # Uses BuildBucket entries associated with a CQ attempt to potentially change
-  # the failure reason to FAILED_PRESUBMIT_BOT. Creates a Pcol of
-  # cq_attempt_as_dict.
-  results = ({
-    'cq_attempts' : cq_attempts_with_key,
-    'bb_entries' : bb_entries_keyed_by_cq_attempt_key
-  } | beam.CoGroupByKey()
-    | beam.FlatMap(update_with_presubmit_failure)
-  )
-  return results
-
-def main():
-  p = chops_beam.EventsPipeline()
-  q = ('SELECT timestamp_millis, action, attempt_start_usec, cq_name, issue,'
-       '  patchset, dry_run, failure_reason, contributing_buildbucket_ids, '
-       '  earliest_equivalent_patchset '
-       'FROM `chrome-infra-events.raw_events.cq`')
-  cq_events_pcol = p | 'read raw CQ events' >> chops_beam.BQRead(q)
-
-  q = ('SELECT id, builder.builder, status from '
-       '`cr-buildbucket.chromium.builds`')
-  bb_entries_pcol = p | 'read BuildBucket' >> chops_beam.BQRead(q)
-
-  results = process_input(cq_events_pcol, bb_entries_pcol)
-
-  # pylint: disable=expression-not-assigned
-  results | chops_beam.BQWrite('chrome-infra-events', 'cq_attempts')
-
-  p.run()
-
-
-if __name__ == '__main__':
-  main()
diff --git a/setup.py b/setup.py
deleted file mode 100644
index fc596b9..0000000
--- a/setup.py
+++ /dev/null
@@ -1,18 +0,0 @@
-# Copyright 2017 The Chromium Authors. All rights reserved.
-# Use of this source code is governed by a BSD-style license that can be
-# found in the LICENSE file.
-
-from setuptools import setup
-
-setup(
-    name='dataflow',
-    version='0.0.6',
-    description='Chrome Infra Dataflow Workflows',
-    long_description=('This package includes Chrome Infra workflows as well as '
-                      'common modules.'),
-    classifiers=[
-        'Programming Language :: Python :: 2.7',
-    ],
-    package_dir={'dataflow': ''},
-    packages=['dataflow', 'dataflow.common'],
-)
diff --git a/test/__init__.py b/test/__init__.py
deleted file mode 100644
index e69de29..0000000
--- a/test/__init__.py
+++ /dev/null
diff --git a/test/cq_attempts_test.py b/test/cq_attempts_test.py
deleted file mode 100644
index c93bb9b..0000000
--- a/test/cq_attempts_test.py
+++ /dev/null
@@ -1,301 +0,0 @@
-# Copyright 2017 The Chromium Authors. All rights reserved.
-# Use of this source code is governed by a BSD-style license that can be
-# found in the LICENSE file.
-
-import unittest
-
-import apache_beam as beam
-from dataflow import cq_attempts as job
-
-from apache_beam.testing import test_pipeline
-from apache_beam.testing import util
-from dataflow.common import objects
-
-
-class TestCQAttemptAccumulator(unittest.TestCase):
-  def setUp(self):
-    self.attempt_start_usec = 1493833887566000
-    self.attempt_start_msec = self.attempt_start_usec / 1000
-    self.timestamp_msec = 1493833887688
-    self.earlier_timestamp = self.timestamp_msec - 1
-    self.later_timestamp = self.timestamp_msec + 1
-    self.combFn = job.CombineEventsToAttempt()
-
-  @staticmethod
-  def construct_attempt_values(attempt_start_usec, actions,
-                               failure_reasons=None):
-    cq_name = 'test_cq'
-    issue = '1'
-    patchset = '1'
-
-    event_basic = {
-      'attempt_start_usec': attempt_start_usec,
-      'cq_name': cq_name,
-      'issue': issue,
-      'patchset': patchset,
-    }
-
-    events = []
-    for i, action in enumerate(actions):
-      event = event_basic.copy()
-      event.update({
-        'action': action[0],
-        'timestamp_millis': action[1],
-        'failure_reason': failure_reasons[i] if failure_reasons else None,
-      })
-      events.append(event)
-
-    attempt = objects.CQAttempt()
-    attempt.cq_name = cq_name
-    attempt.attempt_start_msec = float(attempt_start_usec) / 1000
-    attempt.issue = issue
-    attempt.patchset = patchset
-
-    return (events, attempt)
-
-
-  def failed_attempt_values(self, attempt_start_usec):
-    actions = [
-      (self.combFn.ACTION_PATCH_START, 1000),
-      (self.combFn.ACTION_VERIFIER_CUSTOM_TRYBOTS, 2000),
-      (self.combFn.ACTION_PATCH_FAILED, 5000),
-      (self.combFn.ACTION_PATCH_FAILED, 6000),
-      (self.combFn.ACTION_PATCH_STOP, 9000),
-    ]
-
-    failure_reasons = [
-      None,
-      None,
-      {
-        'fail_type': 'FAIL_TYPE_1',
-        'failed_try_jobs': [
-          {'fail_type': self.combFn.FAIL_TYPE_TEST},
-          {'fail_type': self.combFn.FAIL_TYPE_TEST},
-          {'fail_type': self.combFn.FAIL_TYPE_TEST},
-          {'fail_type': self.combFn.FAIL_TYPE_PATCH},
-        ]
-      },
-      {
-        'fail_type': 'FAIL_TYPE_2',
-        'failed_try_jobs': [
-          {'fail_type': self.combFn.FAIL_TYPE_COMPILE},
-          {'fail_type': self.combFn.FAIL_TYPE_INVALID},
-        ]
-      },
-      None,
-    ]
-
-    events, attempt = self.construct_attempt_values(attempt_start_usec, actions,
-                                                    failure_reasons)
-
-    attempt.first_start_msec = 1000
-    attempt.last_start_msec = 1000
-    attempt.last_stop_msec = 9000
-    attempt.first_stop_msec = 9000
-    attempt.fail_type = 'FAIL_TYPE_2'
-    attempt.failed = True
-    attempt.patch_failed_msec = 5000
-    attempt.invalid_test_results_failures = 1
-    attempt.compile_failures = 1
-    attempt.total_failures = 2
-    attempt.custom_trybots = True
-    attempt.click_to_failure_sec = 4.0
-    attempt.click_to_result_sec = 8.0
-
-    return (events, attempt.as_bigquery_row())
-
-
-  def complete_attempt_values(self, attempt_start_usec):
-    actions = [
-      (self.combFn.ACTION_PATCH_START, 1000),
-      (self.combFn.ACTION_VERIFIER_TRIGGER, 3000),
-      (self.combFn.ACTION_VERIFIER_PASS, 4000),
-      (self.combFn.ACTION_PATCH_COMMITTING, 5000),
-      (self.combFn.ACTION_PATCH_COMMITTED, 6000),
-      (self.combFn.ACTION_PATCH_STOP, 9000),
-    ]
-
-    events, attempt = self.construct_attempt_values(attempt_start_usec, actions)
-
-    events[-2]['contributing_buildbucket_ids'] = [1]
-    events[-1]['contributing_buildbucket_ids'] = [2, 3]
-
-    attempt.first_start_msec = 1000
-    attempt.last_start_msec = 1000
-    attempt.last_stop_msec = 9000
-    attempt.first_stop_msec = 9000
-    attempt.cq_launch_latency_sec = 3.0
-    attempt.verifier_pass_latency_sec = 4.0
-    attempt.tree_check_and_throttle_latency_sec = 1.0
-    attempt.vcs_commit_latency_sec = 1.0
-    attempt.click_to_patch_committed_sec = 6.0
-    attempt.click_to_result_sec = 9.0
-    attempt.committed = True
-    attempt.contributing_bbucket_ids = [1, 2, 3]
-
-    return (events, attempt.as_bigquery_row())
-
-  def test_compute_attempts(self):
-    complete_attempt_events, complete_attempt = self.complete_attempt_values(
-        attempt_start_usec=0)
-    failed_attempt_events, failed_attempt = self.failed_attempt_values(
-        attempt_start_usec=1000000) # 1 second
-
-    incomplete_attempt_events = [
-        {
-            'timestamp_millis': 2,
-            'action': self.combFn.ACTION_PATCH_START,
-            'attempt_start_usec': 1,
-            'cq_name': 'test_cq',
-            'issue': '2',
-            'patchset': '1',
-        },
-    ]
-
-    events = (complete_attempt_events + failed_attempt_events +
-              incomplete_attempt_events)
-    expected_attempts = [complete_attempt, failed_attempt]
-
-    p = test_pipeline.TestPipeline()
-    pcoll = (p
-             | beam.Create(events)
-             | job.ComputeAttempts())
-    util.assert_that(pcoll, util.equal_to(expected_attempts))
-    p.run()
-
-  def basic_event(self, action=None, timestamp_millis=None,
-                  attempt_start_usec=None, cq_name=None):
-    event = objects.CQEvent()
-    event.attempt_start_usec = (attempt_start_usec if attempt_start_usec else
-                                self.attempt_start_usec)
-    event.timestamp_millis = (timestamp_millis if timestamp_millis else
-                              self.timestamp_msec)
-    event.action = action if action else self.combFn.ACTION_PATCH_START
-    event.cq_name = cq_name if cq_name else 'test_cq'
-    event.issue = '123'
-    event.patchset = '456'
-    event.dry_run = True
-    return event
-
-  def test_null_attempt_start_not_included(self):
-    accumulator = self.combFn.add_input(self.combFn.create_accumulator(),
-                                        [{'attempt_start_usec': None}])
-    self.assertEqual(accumulator, [])
-
-  def test_null_timestamp_not_included(self):
-    accumulator = self.combFn.add_input(self.combFn.create_accumulator(),
-                                        [{'timestamp_millis': None}])
-    self.assertEqual(accumulator, [])
-
-  def test_add_input(self):
-    row = {
-        'attempt_start_usec': self.attempt_start_usec,
-        'timestamp_millis': self.timestamp_msec,
-        'action': self.combFn.ACTION_PATCH_START,
-    }
-    event = objects.CQEvent.from_bigquery_row(row)
-    accumulator = self.combFn.add_input(self.combFn.create_accumulator(), [row])
-    self.assertEqual(accumulator, [event])
-
-  def test_extract_min_timestamp_one_timestamp(self):
-    accumulator = [self.basic_event()]
-    attempt = self.combFn.extract_output(accumulator)
-    self.assertEqual(attempt['first_start_msec'], self.timestamp_msec)
-
-  def test_extract_min_timestamp(self):
-    accumulator = [
-      self.basic_event(timestamp_millis=self.timestamp_msec),
-      self.basic_event(timestamp_millis=self.earlier_timestamp),
-      self.basic_event(timestamp_millis=self.later_timestamp),
-    ]
-    attempt = self.combFn.extract_output(accumulator)
-    self.assertEqual(attempt['first_start_msec'], self.earlier_timestamp)
-
-  def test_extract_max_timestamp(self):
-    accumulator = [
-      self.basic_event(timestamp_millis=self.timestamp_msec),
-      self.basic_event(timestamp_millis=self.later_timestamp),
-      self.basic_event(timestamp_millis=self.earlier_timestamp),
-    ]
-    attempt = self.combFn.extract_output(accumulator)
-    self.assertEqual(attempt['last_start_msec'], self.later_timestamp)
-
-  def test_extract_attempt_start(self):
-    accumulator = [self.basic_event()]
-    attempt = self.combFn.extract_output(accumulator)
-    self.assertEqual(attempt['attempt_start_msec'], self.attempt_start_msec)
-
-  def test_extract_different_attempt_start(self):
-    accumulator = [
-        self.basic_event(attempt_start_usec=self.attempt_start_usec),
-        self.basic_event(attempt_start_usec=self.attempt_start_usec+1000)
-    ]
-    self.assertIsNone(self.combFn.extract_output(accumulator))
-
-  def test_earliest_equivalent_patchset(self):
-    event = self.basic_event()
-    event.earliest_equivalent_patchset = 455
-    attempt = self.combFn.extract_output([event])
-    self.assertEqual(attempt['earliest_equivalent_patchset'], 455)
-
-  def test_attempt_key(self):
-    event = self.basic_event()
-    event.attempt_key = 'deadbeef512'
-    attempt = self.combFn.extract_output([event])
-    self.assertEqual(attempt['attempt_key'], 'deadbeef512')
-
-  def test_extract_consistent_field(self):
-    event = self.basic_event()
-    attempt = self.combFn.extract_output([event])
-    for field in self.combFn.consistent_fields:
-      self.assertEqual(attempt[field], event.__dict__[field])
-
-  def test_extract_different_consistent_field(self):
-    accumulator = [
-        self.basic_event(),
-        self.basic_event(cq_name='different_cq_name')
-    ]
-    self.assertIsNone(self.combFn.extract_output(accumulator))
-
-  def test_extract_logical_or(self):
-    accumulator = [self.basic_event(action=self.combFn.ACTION_PATCH_COMMITTED)]
-    attempt = self.combFn.extract_output(accumulator)
-    self.assertTrue(attempt['committed'])
-
-  def test_filter_incomplete_attempts(self):
-    test_cases = [
-        {
-          'attempt': {
-            'first_start_msec': self.timestamp_msec,
-            'last_stop_msec': self.timestamp_msec,
-          },
-          'filtered_expected': False
-        },
-        {
-          'attempt': {
-            'first_start_msec': None,
-            'last_stop_msec': self.timestamp_msec,
-          },
-          'filtered_expected': True
-        },
-        {
-          'attempt': {
-            'first_start_msec': self.timestamp_msec,
-            'last_stop_msec': None,
-          },
-          'filtered_expected': True
-        }
-    ]
-    for test_case in test_cases:
-      attempt = test_case['attempt']
-      filter_attempts = job.ComputeAttempts.filter_incomplete_attempts
-      if test_case['filtered_expected']:
-        with self.assertRaises(StopIteration):
-          filtered_attempt = filter_attempts(attempt).next()
-      else:
-        filtered_attempt = filter_attempts(attempt).next()
-        self.assertEqual(filtered_attempt, attempt)
-
-
-if __name__ == '__main__':
-  unittest.main()
diff --git a/test/new_cq_attempts_test.py b/test/new_cq_attempts_test.py
deleted file mode 100644
index b585328..0000000
--- a/test/new_cq_attempts_test.py
+++ /dev/null
@@ -1,151 +0,0 @@
-# Copyright 2019 The Chromium Authors. All rights reserved.
-# Use of this source code is governed by a BSD-style license that can be
-# found in the LICENSE file.
-
-
-import unittest
-
-import apache_beam as beam
-from apache_beam.testing import test_pipeline
-from apache_beam.testing import util
-
-from dataflow import cq_attempts as sanitize_cq_attempts
-from dataflow import new_cq_attempts
-
-
-def _beam_equals(expected, actual):
-  if expected != actual:
-    raise util.BeamAssertException(
-        'Expected {} to equal {}'.format(expected, actual))
-
-
-class IntegrationTest(unittest.TestCase):
-  def setUp(self):
-    self.pipeline = test_pipeline.TestPipeline()
-
-  def construct_cq_events_pcol(self, failure_reason=None):
-    attempt_start_usec = 1000000
-    cq_name = 'test_cq'
-    issue = '1'
-    patchset = '1'
-    combine_class = sanitize_cq_attempts.CombineEventsToAttempt
-    actions = [
-      combine_class.ACTION_PATCH_START,
-      combine_class.ACTION_VERIFIER_CUSTOM_TRYBOTS,
-      combine_class.ACTION_PATCH_FAILED,
-      combine_class.ACTION_PATCH_FAILED,
-      combine_class.ACTION_PATCH_STOP,
-    ]
-
-    event_basic = {
-      'attempt_start_usec': attempt_start_usec,
-      'cq_name': cq_name,
-      'issue': issue,
-      'patchset': patchset,
-    }
-
-    events = []
-    for i, action in enumerate(actions):
-      event = event_basic.copy()
-      event_failure_reason = (failure_reason if
-          action == combine_class.ACTION_PATCH_FAILED else None)
-      event.update({
-        'action': action,
-        'timestamp_millis': 1000 * (i + 1),
-        'failure_reason': event_failure_reason,
-      })
-      if action == combine_class.ACTION_VERIFIER_CUSTOM_TRYBOTS:
-        event['contributing_buildbucket_ids'] = [11, 12]
-      events.append(event)
-
-    return self.pipeline | beam.Create(events)
-
-  def construct_bb_entries_pcol(self, builder_name, status):
-    bb_entries = [{'id':11, 'builder': builder_name, 'status':status}]
-    return self.pipeline | 'Construct BuildBucket entries' >> beam.Create(
-        bb_entries)
-
-  # One CQ attempt, with matching BuildBucket entries that all pass.
-  def test_basic_pass(self):
-    cq_events_pcol = self.construct_cq_events_pcol()
-    bb_entries_pcol = self.construct_bb_entries_pcol('random_builder',
-                                                     'SUCCESS')
-    results = new_cq_attempts.process_input(cq_events_pcol, bb_entries_pcol)
-
-    # There should be exactly 1 CQ attempt, with no fail type.
-    def expectation_checker(cq_attempts):
-      _beam_equals(len(cq_attempts), 1)
-      _beam_equals(cq_attempts[0]['fail_type'], None)
-
-    util.assert_that(results, expectation_checker)
-    self.pipeline.run()
-
-  def test_basic_failure_random_builder(self):
-    failure_reason = {'fail_type': 'FAILED_JOBS'}
-    cq_events_pcol = self.construct_cq_events_pcol(failure_reason)
-    bb_entries_pcol = self.construct_bb_entries_pcol('random_builder',
-                                                     'FAILURE')
-    results = new_cq_attempts.process_input(cq_events_pcol, bb_entries_pcol)
-
-    # There should be exactly 1 CQ attempt, with no fail type.
-    def expectation_checker(cq_attempts):
-      _beam_equals(len(cq_attempts), 1)
-      _beam_equals(cq_attempts[0]['fail_type'], 'FAILED_JOBS')
-
-    util.assert_that(results, expectation_checker)
-    self.pipeline.run()
-
-  def test_basic_failure_chromium_presubmit(self):
-    failure_reason = {'fail_type': 'FAILED_JOBS'}
-    cq_events_pcol = self.construct_cq_events_pcol(failure_reason)
-    bb_entries_pcol = self.construct_bb_entries_pcol(
-        'chromium_presubmit', 'FAILURE')
-    results = new_cq_attempts.process_input(cq_events_pcol, bb_entries_pcol)
-
-    # There should be exactly 1 CQ attempt, with no fail type.
-    def expectation_checker(cq_attempts):
-      _beam_equals(len(cq_attempts), 1)
-      _beam_equals(cq_attempts[0]['fail_type'], 'FAILED_PRESUBMIT_BOT')
-
-    util.assert_that(results, expectation_checker)
-    self.pipeline.run()
-
-  def test_two_failures(self):
-    failure_reason = {'fail_type': 'FAILED_JOBS'}
-    cq_events_pcol = self.construct_cq_events_pcol(failure_reason)
-    bb_entries = [
-        {'id':11, 'builder': 'chromium_presubmit', 'status':'FAILURE'},
-        {'id':12, 'builder': 'win7-rel', 'status':'INFRA_FAILURE'},
-    ]
-    bb_entries_pcol = (self.pipeline | 'Construct BuildBucket entries' >>
-        beam.Create(bb_entries))
-    results = new_cq_attempts.process_input(cq_events_pcol, bb_entries_pcol)
-
-    # There should be exactly 1 CQ attempt, with no fail type.
-    def expectation_checker(cq_attempts):
-      _beam_equals(len(cq_attempts), 1)
-      _beam_equals(cq_attempts[0]['fail_type'], 'FAILED_JOBS')
-
-    util.assert_that(results, expectation_checker)
-    self.pipeline.run()
-
-  def test_cq_attempt_no_bb_entries(self):
-    cq_events_pcol = self.construct_cq_events_pcol()
-    bb_entries_pcol = (
-        self.pipeline | 'Construct 0 BuildBucket entries' >> beam.Create([]))
-
-    results = new_cq_attempts.process_input(cq_events_pcol, bb_entries_pcol)
-
-    # There should be exactly 1 CQ attempt, with no fail type.
-    def expectation_checker(cq_attempts):
-      _beam_equals(len(cq_attempts), 1)
-      cq_attempt = cq_attempts[0]
-      _beam_equals(cq_attempt['fail_type'], None)
-      _beam_equals(cq_attempt['issue'], '1')
-
-    util.assert_that(results, expectation_checker)
-    self.pipeline.run()
-
-
-if __name__ == '__main__':
-  unittest.main()