jdbc_spec.rb in logstash-input-jdbc-4.1.0

- old
+ new

@@ -1,5 +1,6 @@
+# encoding: utf-8
 require "logstash/devutils/rspec/spec_helper"
 require "logstash/inputs/jdbc"
 require "jdbc/derby"
 require "sequel"
 require "sequel/adapters/jdbc"
@@ -27,10 +28,11 @@
   before :each do
     Jdbc::Derby.load_driver
     db.create_table :test_table do
       DateTime :created_at
       Integer  :num
+      String   :string
       DateTime :custom_time
     end
   end
 
   after :each do
@@ -299,33 +301,55 @@
       }
     end
 
     let(:settings) do
       {
-        "statement" => "SELECT * from test_table",
+        "statement" => "SELECT * from test_table WHERE custom_time > :sql_last_value",
+        "use_column_value" => true,
+        "tracking_column" => "custom_time",
+        "last_run_metadata_path" => Stud::Temporary.pathname
       }
     end
 
-    let(:num_rows) { 10 }
+    let(:hour_range) { 10..20 }
 
-    before do
-      num_rows.times do
-        db[:test_table].insert(:num => 1, :custom_time => "2015-01-01 12:00:00", :created_at => Time.now.utc)
+    it "should convert the time to reflect the timezone " do
+      last_run_value = Time.iso8601("2000-01-01T00:00:00.000Z")
+      File.write(settings["last_run_metadata_path"], YAML.dump(last_run_value))
+
+      hour_range.each do |i|
+        db[:test_table].insert(:num => i, :custom_time => "2015-01-01 #{i}:00:00", :created_at => Time.now.utc)
       end
 
       plugin.register
-    end
 
-    after do
+      plugin.run(queue)
+      expected = ["2015-01-01T16:00:00.000Z",
+                  "2015-01-01T17:00:00.000Z",
+                  "2015-01-01T18:00:00.000Z",
+                  "2015-01-01T19:00:00.000Z",
+                  "2015-01-01T20:00:00.000Z",
+                  "2015-01-01T21:00:00.000Z",
+                  "2015-01-01T22:00:00.000Z",
+                  "2015-01-01T23:00:00.000Z",
+                  "2015-01-02T00:00:00.000Z",
+                  "2015-01-02T01:00:00.000Z",
+                  "2015-01-02T02:00:00.000Z"].map { |i| Time.iso8601(i) }
+      actual = queue.size.times.map { queue.pop.get("custom_time").time }
+      expect(actual).to eq(expected)
       plugin.stop
-    end
 
-    it "should convert the time to reflect the timezone " do
       plugin.run(queue)
+      expect(queue.size).to eq(0)
+      db[:test_table].insert(:num => 11, :custom_time => "2015-01-01 11:00:00", :created_at => Time.now.utc)
+      db[:test_table].insert(:num => 12, :custom_time => "2015-01-01 21:00:00", :created_at => Time.now.utc)
+      plugin.run(queue)
+      expect(queue.size).to eq(1)
       event = queue.pop
-      # This reflects a 6 hour time difference between UTC and America/Chicago
-      expect(event.get("custom_time").time).to eq(Time.iso8601("2015-01-01T18:00:00Z"))
+      expect(event.get("num")).to eq(12)
+      expect(event.get("custom_time").time).to eq(Time.iso8601("2015-01-02T03:00:00.000Z"))
+      p settings
     end
   end
 
   context "when fetching time data without jdbc_default_timezone set" do
     let(:mixin_settings) do
@@ -868,8 +892,113 @@
 
     it "should not fail when passed a non-positive value" do
       mixin_settings['connection_retry_attempts'] = -2
       expect { plugin.register }.to_not raise_error
       plugin.stop
+    end
+  end
+
+  context "when encoding of some columns need to be changed" do
+
+    let(:settings) {{ "statement" => "SELECT * from test_table" }}
+    let(:events)   { [] }
+    let(:row) do
+      {
+        "column0" => "foo",
+        "column1" => "bar".force_encoding(Encoding::ISO_8859_1),
+        "column2" => 3
+      }
+    end
+
+    before(:each) do
+      allow_any_instance_of(Sequel::JDBC::Derby::Dataset).to receive(:each).and_yield(row)
+      plugin.register
+    end
+
+    after(:each) do
+      plugin.stop
+    end
+
+    it "should not convert any column by default" do
+      encoded_row = {
+        "column0" => "foo",
+        "column1" => "bar".force_encoding(Encoding::ISO_8859_1),
+        "column2" => 3
+      }
+      expect(LogStash::Event).to receive(:new) do |row|
+        row.each do |k, v|
+          next unless v.is_a?(String)
+          expect(row[k].encoding).to eq(encoded_row[k].encoding)
+        end
+      end
+      plugin.run(events)
+    end
+
+    context "when all string columns should be encoded" do
+
+      let(:settings) do
+        {
+          "statement" => "SELECT * from test_table",
+          "charset" => "ISO-8859-1"
+        }
+      end
+
+      let(:row) do
+        {
+          "column0" => "foo".force_encoding(Encoding::ISO_8859_1),
+          "column1" => "bar".force_encoding(Encoding::ISO_8859_1),
+          "column2" => 3
+        }
+      end
+
+      it "should transform all column string to UTF-8, default encoding" do
+        encoded_row = {
+          "column0" => "foo",
+          "column1" => "bar",
+          "column2" => 3
+        }
+        expect(LogStash::Event).to receive(:new) do |row|
+          row.each do |k, v|
+            next unless v.is_a?(String)
+            expect(row[k].encoding).to eq(encoded_row[k].encoding)
+          end
+        end
+        plugin.run(events)
+      end
+    end
+
+    context "when only an specific column should be converted" do
+
+      let(:settings) do
+        {
+          "statement" => "SELECT * from test_table",
+          "columns_charset" => { "column1" => "ISO-8859-1" }
+        }
+      end
+
+      let(:row) do
+        {
+          "column0" => "foo",
+          "column1" => "bar".force_encoding(Encoding::ISO_8859_1),
+          "column2" => 3,
+          "column3" => "berlin".force_encoding(Encoding::ASCII_8BIT)
+        }
+      end
+
+      it "should only convert the selected column" do
+        encoded_row = {
+          "column0" => "foo",
+          "column1" => "bar",
+          "column2" => 3,
+          "column3" => "berlin".force_encoding(Encoding::ASCII_8BIT)
+        }
+        expect(LogStash::Event).to receive(:new) do |row|
+          row.each do |k, v|
+            next unless v.is_a?(String)
+            expect(row[k].encoding).to eq(encoded_row[k].encoding)
+          end
+        end
+        plugin.run(events)
+      end
     end
   end
 end