require_relative '../charrington_spec_helper' describe LogStash::Outputs::Charrington do include_context 'postgres' include_context 'pipeline' describe 'a new payload with one event that truncates strings correctly' do let(:too_big) do "0123456789" * 30 end let(:config) do <<-CONFIG input { generator { message => '{"app_name": "Web App", "event": "schemaless", "meta": { "type": "XML", "too_big": "#{too_big}" }, "published_at": "2019-07-29T20:09:18Z" }' codec => 'json' count => 1 } } output { charrington { connection_string => '#{@url}' driver_jar_path => '#{driver_path}' schema => '' transformer => 'postgres' } } CONFIG end it 'creates a table and inserts a record' do drop_table('tracks') create_tracks_table drop_table('schemaless') run_pipeline expect(query('SELECT * FROM schemaless')).to match_array([{id: "1", app_name: "Web App", event: "schemaless", inserted_at: a_kind_of(String), :meta_type => "XML", :meta_too_big => too_big[0,254], :published_at => "2019-07-29 20:09:18"}]) expect(query('SELECT COUNT(1) FROM schemaless').first[:count]).to eq("1") end end describe 'removes a trailing plus sign' do let(:config) do <<-CONFIG input { generator { message => '{"app_name": "Web App", "event": "with plus +", "meta": { "type": "XML" } }' codec => 'json' count => 1 } } output { charrington { connection_string => '#{@url}' driver_jar_path => '#{driver_path}' schema => '' transformer => 'postgres' } } CONFIG end it 'creates a table and inserts a record' do drop_table('with_plus') run_pipeline expect(query('SELECT * FROM with_plus')).to match_array([{id: "1", app_name: "Web App", event: "with plus +", inserted_at: a_kind_of(String), :meta_type => "XML"}]) expect(query('SELECT COUNT(1) FROM with_plus').first[:count]).to eq("1") end end describe '2 event payloads with different metadata' do let(:config) do <<-CONFIG input { generator { lines => [ '{"app_name": "Web App", "event": "metadata", "meta": { "type": "XML" } }', '{"app_name": "Web App", "event": "metadata", "meta": { "type": "XML", "file_name": "virus.pdf" } }' ] codec => 'json' count => 1 } } output { charrington { connection_string => '#{@url}' driver_jar_path => '#{driver_path}' schema => '' transformer => 'postgres' } } CONFIG end it 'creates a table and inserts the first record and alters the table for the second record' do drop_table('tracks') create_tracks_table drop_table('metadata') run_pipeline expect(query('SELECT * FROM metadata')).to match_array([ {app_name: 'Web App', event: 'metadata', id: '1', inserted_at: a_kind_of(String), meta_file_name: nil, meta_type: 'XML'}, {app_name: 'Web App', event: 'metadata', id: '2', inserted_at: a_kind_of(String), meta_file_name: 'virus.pdf', meta_type: 'XML'} ]) expect(query('SELECT COUNT(1) FROM metadata').first[:count]).to eq("2") end end describe 'event payload for an existing table with existing data' do let(:config) do <<-CONFIG input { generator { message => '{"app_name": "Web App", "event": "existing", "meta": { "type": "XML" } }' codec => 'json' count => 1 } } output { charrington { connection_string => '#{@url}' driver_jar_path => '#{driver_path}' schema => '' transformer => 'postgres' } } CONFIG end it 'can insert into an existing table' do drop_table('tracks') create_tracks_table # setup pre-existing data drop_table('existing') create_table('CREATE TABLE existing (id SERIAL PRIMARY KEY, inserted_at TIMESTAMP DEFAULT NOW(), app_name VARCHAR(255))') insert("INSERT INTO existing (app_name) VALUES ('Not Agent')") expect(query('SELECT * FROM existing')).to match_array([{id: "1", inserted_at: a_kind_of(String), app_name: "Not Agent"}]) run_pipeline expect(query('SELECT * FROM existing')).to match_array([ {id: '1', app_name: 'Not Agent', event: nil, inserted_at: a_kind_of(String), meta_type: nil}, {id: '2', app_name: 'Web App', event: 'existing', inserted_at: a_kind_of(String), meta_type: 'XML'} ]) end end describe '2 event payloads with different metadata and different schema' do let(:schema) { "dea_test" } let(:config) do <<-CONFIG input { generator { lines => [ '{"app_name": "Web App", "event": "From Agent", "meta": { "type": "XML" } }', '{"app_name": "Web App", "event": "From Agent", "meta": { "type": "XML", "file_name": "virus.pdf" } }' ] codec => 'json' count => 1 } } output { charrington { connection_string => '#{@url}' driver_jar_path => '#{driver_path}' schema => '#{schema}' transformer => 'postgres' } } CONFIG end it 'creates a table and inserts the first record and alters the table for the second record' do drop_table('tracks') create("CREATE SCHEMA IF NOT EXISTS #{schema}") create_tracks_table('dea_test') drop_table("#{schema}.from_agent") run_pipeline expect(query("SELECT * FROM #{schema}.from_agent")).to match_array([ {app_name: 'Web App', event: 'From Agent', id: '1', inserted_at: a_kind_of(String), meta_file_name: nil, meta_type: 'XML'}, {app_name: 'Web App', event: 'From Agent', id: '2', inserted_at: a_kind_of(String), meta_file_name: 'virus.pdf', meta_type: 'XML'} ]) expect(query("SELECT COUNT(1) FROM #{schema}.from_agent").first[:count]).to eq("2") end end describe 'a new payload with one event camelCase' do let(:config) do <<-CONFIG input { generator { message => '{"app_name": "Web App", "event": "camelCase this", "meta": { "type": "XML" } }' codec => 'json' count => 1 } } output { charrington { connection_string => '#{@url}' driver_jar_path => '#{driver_path}' schema => '' transformer => 'postgres' } } CONFIG end it 'creates a table and inserts a record' do drop_table('camel_case_this') run_pipeline expect(query('SELECT * FROM camel_case_this')).to match_array([{id: "1", app_name: "Web App", event: "camelCase this", inserted_at: a_kind_of(String), :meta_type => "XML"}]) expect(query('SELECT COUNT(1) FROM camel_case_this').first[:count]).to eq("1") end end describe 'a new payload with one event transformed to be redshift shape' do let(:config) do <<-CONFIG input { generator { message => '{"app_name": "Web App", "action": "click", "event": "schemaless", "meta": { "type": "XML" }, "published_at": "2019-07-29T20:09:18Z", "user_id": "123", "user_uid": "456" }' codec => 'json' count => 1 } } output { charrington { connection_string => '#{@url}' driver_jar_path => '#{driver_path}' schema => '' transformer => 'redshift' } } CONFIG end it 'creates a table and inserts a record with data the shape of redshift' do drop_table('tracks') create_tracks_table drop_table('schemaless') run_pipeline expect(query('SELECT * FROM schemaless')).to match_array([ {:anonymous_id=>"", :action=>"click", :app_name=>"Web App", :event=>"schemaless", :event_text=>"schemaless", :id=>a_kind_of(String), :original_timestamp=>a_kind_of(String), :received_at=>a_kind_of(String), :sent_at=>"2019-07-29 20:09:18", :timestamp=>a_kind_of(String), :type=>"XML", :user_id=>"123", :user_uid=>"456", :uuid_ts=>a_kind_of(String)} ]) expect(query('SELECT COUNT(1) FROM schemaless').first[:count]).to eq("1") expect(query("SELECT * FROM tracks")).to match_array([ {:id=>a_kind_of(String), :action=>"click", :anonymous_id=>"", :app_name=>"Web App", :context_campaign_content=>nil, :context_campaign_medium=>nil, :context_campaign_name=>nil, :context_campaign_source=>nil, :context_ip=>nil, :context_library_name=>nil, :context_library_version=>nil, :context_page_path=>nil, :context_page_referrer=>nil, :context_page_search=>nil, :context_page_title=>nil, :context_page_url=>nil, :context_user_agent=>nil, :event=>"schemaless", :event_text=>"schemaless", :original_timestamp=>"2019-07-29 20:09:18", :received_at=>"2019-07-29 20:09:18", :segment_dedupe_id=>nil, :sent_at=>"2019-07-29 20:09:18", :timestamp=>"2019-07-29 20:09:18", :user_id=>"123", :user_uid=>"456", :uuid=>nil, :uuid_ts=>a_kind_of(String)} ]) end end describe 'a new payload has timestamp-related keys set to sent_at' do let(:config) do <<-CONFIG input { generator { message => '{"app_name": "Web App", "event": "schemaless", "meta": { "type": "XML" }, "published_at": "2019-07-29T20:09:18Z", "user_id": "123", "user_uid": "456" }' codec => 'json' count => 1 } } output { charrington { connection_string => '#{@url}' driver_jar_path => '#{driver_path}' schema => '' transformer => 'redshift' } } CONFIG end it 'creates a table with timestamp-related keys and inserts a record with timestamp-related keys same as sent_at' do drop_table('tracks') create_tracks_table drop_table('schemaless') run_pipeline expect(query('SELECT * FROM schemaless')).to match_array([ {:anonymous_id=>"", :app_name=>"Web App", :event=>"schemaless", :event_text=>"schemaless", :id=>a_kind_of(String), :original_timestamp=>"2019-07-29 20:09:18", :received_at=>"2019-07-29 20:09:18", :sent_at=>"2019-07-29 20:09:18", :timestamp=>"2019-07-29 20:09:18", :type=>"XML", :user_id=>"123", :user_uid=>"456", :uuid_ts=>a_kind_of(String)} ]) expect(query('SELECT COUNT(1) FROM schemaless').first[:count]).to eq("1") end end end