test: out_kafka_buffered: Add testcases for out_kafka_buffered

Signed-off-by: Hiroshi Hatake <hatake@calyptia.com>

Author: cosmo0920

test/helper.rb

@@ -23,6 +23,7 @@ def method_missing(method, *args)
 end
 
 require 'fluent/plugin/out_kafka'
+require 'fluent/plugin/out_kafka_buffered'
 require 'fluent/plugin/out_kafka2'
 require 'fluent/plugin/in_kafka'
 require 'fluent/plugin/in_kafka_group'

test/plugin/test_out_kafka_buffered.rb

@@ -0,0 +1,68 @@
+require 'helper'
+require 'fluent/output'
+
+class KafkaBufferedOutputTest < Test::Unit::TestCase
+ def setup
+ Fluent::Test.setup
+ end
+
+ BASE_CONFIG = %[
+ type kafka_buffered
+ ]
+
+ CONFIG = BASE_CONFIG + %[
+ default_topic kitagawakeiko
+ brokers localhost:9092
+ ]
+
+ def create_driver(conf = CONFIG, tag='test')
+ Fluent::Test::BufferedOutputTestDriver.new(Fluent::KafkaOutputBuffered, tag).configure(conf)
+ end
+
+ def test_configure
+ assert_nothing_raised(Fluent::ConfigError) {
+ create_driver(BASE_CONFIG)
+ }
+
+ assert_nothing_raised(Fluent::ConfigError) {
+ create_driver(CONFIG)
+ }
+
+ assert_nothing_raised(Fluent::ConfigError) {
+ create_driver(CONFIG + %[
+ buffer_type memory
+ ])
+ }
+
+ d = create_driver
+ assert_equal 'kitagawakeiko', d.instance.default_topic
+ assert_equal 'localhost:9092', d.instance.brokers
+ end
+
+ def test_format
+ d = create_driver
+ end
+
+ data("crc32" => "crc32",
+ "murmur2" => "murmur2")
+ def test_partitioner_hash_function(data)
+ hash_type = data
+ d = create_driver(CONFIG + %[partitioner_hash_function #{hash_type}])
+ assert_nothing_raised do
+ d.instance.refresh_client
+ end
+ end
+
+ def test_mutli_worker_support
+ d = create_driver
+ assert_equal true, d.instance.multi_workers_ready?
+
+ end
+
+ def test_write
+ d = create_driver
+ time = Time.parse("2011-01-02 13:14:15 UTC").to_i
+ d.emit({"a"=>1}, time)
+ d.emit({"a"=>2}, time)
+ end
+end