def testSortFile(self):
    """Test sorting a file."""
    bucket_name = "testbucket"
    test_filename = "testfile"
    full_filename = (bucket_name, test_filename) % "/%s/%s"

    input_data = [
        (str(i), "_" + str(i)) for i in range(100)]

    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._SortChunksPipeline("testjob", bucket_name, [[full_filename]])
    p.start()
    test_support.execute_until_empty(self.taskqueue)
    p = shuffler._SortChunksPipeline.from_id(p.pipeline_id)

    input_data.sort()
    output_files = p.outputs.default.value[0]
    output_data = []
    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()))

    self.assertEquals(input_data, output_data)
    self.assertEquals(1, len(self.emails))


# pylint: disable=invalid-name
dataset/ETHPy150Open GoogleCloudPlatform/appengine-mapreduce/python/test/mapreduce/shuffler_end_to_end_test.py/SortFileEndToEndTest.testSortFile/swapped operands of `%`