def testHashingMultipleFiles(self):
    """Test hashing files."""
    input_data = [(str(i), str(i)) for i in range(100)]
    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 = shuffler._HashPipeline("testjob", bucket_name,
                               [full_filename, full_filename, full_filename])
    p.start()
    test_support.execute_until_empty(self.taskqueue)
    p = shuffler._HashPipeline.from_id(p.pipeline_id)

    list_of_output_files = p.outputs.default.value
    output_data = []
    for output_files in list_of_output_files:
      for output_file in output_files:
        with cloudstorage.open(output_file) as f:
          for binary_record in records.RecordsReader(f):
            proto = kv_pb.KeyValue()
            proto.ParseFromString(binary_record)
            output_data.append((proto.key(), proto.value()))

    output_data.sort()
    self.assertEquals(300, len(output_data))
    for i in range(len(input_data)):
      self.assertEquals(input_data[i], output_data[(3 * i)])
      self.assertEquals(input_data[i], output_data[(3 * i) + 1])
      self.assertEquals(input_data[i], output_data[(3 * i) + 2])
    self.assertEquals(1, len(self.emails))
dataset/ETHPy150Open GoogleCloudPlatform/appengine-mapreduce/python/test/mapreduce/shuffler_end_to_end_test.py/HashEndToEndTest.testHashingMultipleFiles/original