def testPartialRecords(self):
    """Test merging into partial key values."""
    try:
      self._prev_max_values_count = shuffler._MergePipeline._MAX_VALUES_COUNT
      # force max values count to extremely low value.
      shuffler._MergePipeline._MAX_VALUES_COUNT = 1

      input_data = [("1", "a"), ("2", "b"), ("3", "c")]
      input_data.sort()

      bucket_name = "testbucket"
      test_filename = "testfile"
      full_filename = "/%s/%s" % (bucket_name, test_filename)

      with cloudstorage.open(full_filename, mode="w") as f:
        with records.RecordsWriter(f) as w:
          for (k, v) in input_data:
            proto = kv_pb.KeyValue()
            proto.set_key(k)
            proto.set_value(v)
            w.write(proto.Encode())

      p = TestMergePipeline(bucket_name,
                            [full_filename, full_filename, full_filename])
      p.start()
      test_support.execute_until_empty(self.taskqueue)
      p = TestMergePipeline.from_id(p.pipeline_id)

      output_file = p.outputs.default.value[0]
      output_data = []
      with cloudstorage.open(output_file) as f:
        for record in records.RecordsReader(f):
          output_data.append(record)

      expected_data = [
          ("1", ["a"], True),
          ("1", ["a"], True),
          ("1", ["a"], False),
          ("2", ["b"], True),
          ("2", ["b"], True),
          ("2", ["b"], False),
          ("3", ["c"], True),
          ("3", ["c"], True),
          ("3", ["c"], False),
          ]
      self.assertEquals([str(e) for e in expected_data], output_data)
    finally:
      shuffler._MergePipeline._MAX_VALUES_COUNT = self._prev_max_values_count
    self.assertEquals(1, len(self.emails))
dataset/ETHPy150Open GoogleCloudPlatform/appengine-mapreduce/python/test/mapreduce/shuffler_end_to_end_test.py/MergingReaderEndToEndTest.testPartialRecords/original