Skip to content

Commit

Permalink
Add more placeholers test cases
Browse files Browse the repository at this point in the history
  • Loading branch information
@cosmo0920
cosmo0920 committed Feb 25, 2019
1 parent 738064e commit 09f1acc
Show file tree
Hide file tree
Showing 2 changed files with 134 additions and 36 deletions.
85 changes: 67 additions & 18 deletions test/plugin/test_out_kinesis_streams.rb
View file
Original file line number Diff line number Diff line change
Expand Up @@ -165,26 +165,75 @@ def test_multibyte_input
assert_equal (record.to_json + "\n").b, @server.records.first
end

def test_placeholder
d = create_driver(
Fluent::Config::Element.new('ROOT', '', {
"stream_name" => "stream-placeholder-${tag}",
"@log_level" => "error",
"retries_on_batch_request" => 10,
"endpoint" => "https://localhost:#{@server.port}",
"ssl_verify_peer" => false,
class PlaceholdersTest < self
def test_tag_placeholder
d = create_driver(
Fluent::Config::Element.new('ROOT', '', {
"stream_name" => "stream-placeholder-${tag}",
"@log_level" => "error",
"retries_on_batch_request" => 10,
"endpoint" => "https://localhost:#{@server.port}",
"ssl_verify_peer" => false,
},[
Fluent::Config::Element.new('buffer', 'tag', {
'@type' => 'memory',
}, [])
])
)
Fluent::Config::Element.new('buffer', 'tag', {
'@type' => 'memory',
}, [])
])
)

record = {"a" => "test"}
driver_run(d, [record])
assert_equal("stream-placeholder-test", @server.detailed_records.first[:stream_name])
assert_equal 0, d.instance.log.out.logs.size
assert_equal (record.to_json + "\n").b, @server.records.first
record = {"a" => "test"}
driver_run(d, [record])
assert_equal("stream-placeholder-test", @server.detailed_records.first[:stream_name])
assert_equal 0, d.instance.log.out.logs.size
assert_equal (record.to_json + "\n").b, @server.records.first
end

def test_time_placeholder
d = create_driver(
Fluent::Config::Element.new('ROOT', '', {
"stream_name" => "stream-placeholder-${tag}-%Y%m%d",
"@log_level" => "error",
"retries_on_batch_request" => 10,
"endpoint" => "https://localhost:#{@server.port}",
"ssl_verify_peer" => false,
},[
Fluent::Config::Element.new('buffer', 'tag, time', {
'@type' => 'memory',
'timekey' => 3600
}, [])
])
)

record = {"a" => "test"}
time = event_time
driver_run(d, [record], time: time)
assert_equal("stream-placeholder-test-#{Time.now.strftime("%Y%m%d")}",
@server.detailed_records.first[:stream_name])
assert_equal 0, d.instance.log.out.logs.size
assert_equal (record.to_json + "\n").b, @server.records.first
end

def test_custom_placeholder
d = create_driver(
Fluent::Config::Element.new('ROOT', '', {
"stream_name" => "stream-placeholder-${$.key.nested}",
"@log_level" => "error",
"retries_on_batch_request" => 10,
"endpoint" => "https://localhost:#{@server.port}",
"ssl_verify_peer" => false,
},[
Fluent::Config::Element.new('buffer', '$.key.nested', {
'@type' => 'memory',
}, [])
])
)

record = {"key" => {"nested" => "nested-value"}}
driver_run(d, [record])
assert_equal("stream-placeholder-nested-value", @server.detailed_records.first[:stream_name])
assert_equal 0, d.instance.log.out.logs.size
assert_equal (record.to_json + "\n").b, @server.records.first
end
end

def test_record_count
Expand Down
85 changes: 67 additions & 18 deletions test/plugin/test_out_kinesis_streams_aggregated.rb
View file
Original file line number Diff line number Diff line change
Expand Up @@ -212,26 +212,75 @@ def test_record_count
assert @server.raw_records.size < count
end

def test_placeholder
d = create_driver(
Fluent::Config::Element.new('ROOT', '', {
"stream_name" => "stream-placeholder-${tag}",
"@log_level" => "error",
"retries_on_batch_request" => 10,
"endpoint" => "https://localhost:#{@server.port}",
"ssl_verify_peer" => false,
class PlaceholdersTest < self
def test_tag_placeholder
d = create_driver(
Fluent::Config::Element.new('ROOT', '', {
"stream_name" => "stream-placeholder-${tag}",
"@log_level" => "error",
"retries_on_batch_request" => 10,
"endpoint" => "https://localhost:#{@server.port}",
"ssl_verify_peer" => false,
},[
Fluent::Config::Element.new('buffer', 'tag', {
'@type' => 'memory',
}, [])
])
)
Fluent::Config::Element.new('buffer', 'tag', {
'@type' => 'memory',
}, [])
])
)

record = {"a" => "test"}
driver_run(d, [record])
assert_equal("stream-placeholder-test", @server.detailed_records.first[:stream_name])
assert_equal 0, d.instance.log.out.logs.size
assert_equal (record.to_json + "\n").b, @server.records.first
record = {"a" => "test"}
driver_run(d, [record])
assert_equal("stream-placeholder-test", @server.detailed_records.first[:stream_name])
assert_equal 0, d.instance.log.out.logs.size
assert_equal (record.to_json + "\n").b, @server.records.first
end

def test_time_placeholder
d = create_driver(
Fluent::Config::Element.new('ROOT', '', {
"stream_name" => "stream-placeholder-${tag}-%Y%m%d",
"@log_level" => "error",
"retries_on_batch_request" => 10,
"endpoint" => "https://localhost:#{@server.port}",
"ssl_verify_peer" => false,
},[
Fluent::Config::Element.new('buffer', 'tag, time', {
'@type' => 'memory',
'timekey' => 3600
}, [])
])
)

record = {"a" => "test"}
time = event_time
driver_run(d, [record], time: time)
assert_equal("stream-placeholder-test-#{Time.now.strftime("%Y%m%d")}",
@server.detailed_records.first[:stream_name])
assert_equal 0, d.instance.log.out.logs.size
assert_equal (record.to_json + "\n").b, @server.records.first
end

def test_custom_placeholder
d = create_driver(
Fluent::Config::Element.new('ROOT', '', {
"stream_name" => "stream-placeholder-${$.key.nested}",
"@log_level" => "error",
"retries_on_batch_request" => 10,
"endpoint" => "https://localhost:#{@server.port}",
"ssl_verify_peer" => false,
},[
Fluent::Config::Element.new('buffer', '$.key.nested', {
'@type' => 'memory',
}, [])
])
)

record = {"key" => {"nested" => "nested-value"}}
driver_run(d, [record])
assert_equal("stream-placeholder-nested-value", @server.detailed_records.first[:stream_name])
assert_equal 0, d.instance.log.out.logs.size
assert_equal (record.to_json + "\n").b, @server.records.first
end
end

# Debug test case for the issue that it fails to flush the buffer
Expand Down

0 comments on commit 09f1acc

Please sign in to comment.