✅ Added tests

Author: nilskuhn

lib/logstash/outputs/mongodb.rb

@@ -91,7 +91,7 @@ def validate_config
     if @action != "insert" && @action != "update" && @action != "replace"
       raise LogStash::ConfigurationError, "Only insert, update and replace are valid for 'action' setting."
     end
-    if (@action == "update" || @action == "replace") && @query_value.blank?
+    if (@action == "update" || @action == "replace") && (@query_value.nil? || @query_value.empty?)
       raise LogStash::ConfigurationError, "If action is update or replace, query_value must be set."
     end
   end
@@ -160,6 +160,7 @@ def get_write_ops(documents)
     ops = []
     documents.each do |doc|
       replaced_query_value = doc["metadata_mongodb_output_query_value"]
+      doc.delete("metadata_mongodb_output_query_value")
       if @action == "insert"
         ops << {:insert_one => doc}
       elsif @action == "update"

spec/integration/mongodb_spec.rb

@@ -7,10 +7,11 @@
   let(:database)   { 'logstash' }
   let(:collection) { 'logs' }
   let(:uuid)       { SecureRandom.uuid }
+  let(:action) { 'insert' }
 
   let(:config) do
     { "uri" => uri, "database" => database,
-      "collection" => collection, "isodate" => true }
+      "collection" => collection, "isodate" => true, "action" => action }
   end
 
   describe "#send" do

spec/outputs/mongodb_config_validation_spec.rb

@@ -0,0 +1,64 @@
+# encoding: utf-8
+require_relative "../spec_helper"
+require "logstash/plugin"
+
+describe LogStash::Outputs::Mongodb do
+
+  let(:uri) { 'mongodb://localhost:27017' }
+  let(:database) { 'logstash' }
+  let(:collection) { 'logs' }
+
+  describe "validate_config method" do
+
+    subject! { LogStash::Outputs::Mongodb.new(config) }
+
+    [
+        {:action => "not-supported", :query_key => "qk", :query_value => "qv", :upsert => false,
+         :expected_reason => "Only insert, update and replace are valid for 'action' setting."},
+        {:action => "update", :query_key => "qk", :query_value => nil, :upsert => false,
+         :expected_reason => "If action is update or replace, query_value must be set."},
+        {:action => "update", :query_key => "qk", :query_value => "", :upsert => false,
+         :expected_reason => "If action is update or replace, query_value must be set."},
+        {:action => "replace", :query_key => "qk", :query_value => nil, :upsert => false,
+         :expected_reason => "If action is update or replace, query_value must be set."},
+        {:action => "replace", :query_key => "qk", :query_value => "", :upsert => false,
+         :expected_reason => "If action is update or replace, query_value must be set."},
+        {:action => "insert", :bulk_size => 1001,
+         :expected_reason => "Bulk size must be lower than '1000', currently '1001'"},
+    ].each do |test|
+
+      describe "when validating config with action '#{test[:action]}' query_key '#{test[:query_key]}', query_value '#{test[:query_value]}' and upsert '#{test[:upsert]}'" do
+
+        let(:config) {
+          configuration = {
+              "uri" => uri,
+              "database" => database,
+              "collection" => collection
+          }
+          unless test[:action].nil?
+            configuration["action"] = test[:action]
+          end
+          unless test[:query_key].nil?
+            configuration["query_key"] = test[:query_key]
+          end
+          unless test[:query_value].nil?
+            configuration["query_value"] = test[:query_value]
+          end
+          unless test[:upsert].nil?
+            configuration["upsert"] = test[:upsert]
+          end
+          unless test[:bulk_size].nil?
+            configuration["bulk_size"] = test[:bulk_size]
+          end
+          return configuration
+        }
+
+        it "should raise error: #{test[:expected_reason]}" do
+          expect { subject.validate_config }.to raise_error(LogStash::ConfigurationError, test[:expected_reason])
+        end
+      end
+
+    end
+
+  end
+end

spec/outputs/mongodb_spec.rb renamed to spec/outputs/mongodb_insert_spec.rb

@@ -7,11 +7,13 @@
   let(:uri)        { 'mongodb://localhost:27017' }
   let(:database)   { 'logstash' }
   let(:collection) { 'logs' }
+  let(:action) { 'insert' }
 
   let(:config) {{
     "uri" => uri,
     "database" => database,
-    "collection" => collection
+    "collection" => collection,
+    "action" => action
   }}
 
   it "should register and close" do
@@ -20,7 +22,7 @@
     plugin.close
   end
 
-  describe "receive" do
+  describe "receive method while action is 'insert'" do
     subject! { LogStash::Outputs::Mongodb.new(config) }
 
     let(:event)      { LogStash::Event.new(properties) }
@@ -32,15 +34,15 @@
       allow(Mongo::Client).to receive(:new).and_return(connection)
       allow(connection).to receive(:use).and_return(client)
       allow(client).to receive(:[]).and_return(collection)
-      allow(collection).to receive(:insert_one)
+      allow(collection).to receive(:bulk_write)
       subject.register
     end
 
     after(:each) do
       subject.close
     end
 
-    describe "#send" do
+    describe "when processing an event" do
       let(:properties) {{
         "message" => "This is a message!",
         "uuid" => SecureRandom.uuid,
@@ -49,36 +51,41 @@
       }}
 
       it "should send the event to the database" do
-        expect(collection).to receive(:insert_one)
+        expect(collection).to receive(:bulk_write)
         subject.receive(event)
       end
     end
 
-    describe "no event @timestamp" do
+    describe "when processing an event without @timestamp set" do
       let(:properties) { { "message" => "foo" } }
 
-      it "should not contain a @timestamp field in the mongo document" do
+      it "should send a document without @timestamp field to mongodb" do
         expect(event).to receive(:timestamp).and_return(nil)
         expect(event).to receive(:to_hash).and_return(properties)
-        expect(collection).to receive(:insert_one).with(properties)
+        expect(collection).to receive(:bulk_write).with(
+            [ {:insert_one => properties} ]
+        )
         subject.receive(event)
       end
     end
 
-    describe "generateId" do
+    describe "when generateId is set" do
       let(:properties) { { "message" => "foo" } }
       let(:config) {{
           "uri" => uri,
           "database" => database,
           "collection" => collection,
-          "generateId" => true
+          "generateId" => true,
+          "action" => "insert"
       }}
 
-      it "should contain a BSON::ObjectId as _id" do
+      it "should send a document containing a BSON::ObjectId as _id to mongodb" do
         expect(BSON::ObjectId).to receive(:new).and_return("BSON::ObjectId")
         expect(event).to receive(:timestamp).and_return(nil)
         expect(event).to receive(:to_hash).and_return(properties)
-        expect(collection).to receive(:insert_one).with(properties.merge("_id" => "BSON::ObjectId"))
+        expect(collection).to receive(:bulk_write).with(
+            [ {:insert_one => properties.merge("_id" => "BSON::ObjectId")} ]
+        )
         subject.receive(event)
       end
     end

spec/outputs/mongodb_replace_spec.rb

@@ -0,0 +1,98 @@
+# encoding: utf-8
+require_relative "../spec_helper"
+require "logstash/plugin"
+
+describe LogStash::Outputs::Mongodb do
+
+  let(:uri) { 'mongodb://localhost:27017' }
+  let(:database) { 'logstash' }
+  let(:collection) { 'logs' }
+  let(:action) { 'replace' }
+
+  let(:config) { {
+      "uri" => uri,
+      "database" => database,
+      "collection" => collection,
+      "action" => action
+  } }
+
+  describe "receive method while action is 'replace'" do
+    subject! { LogStash::Outputs::Mongodb.new(config) }
+
+    let(:properties) { {
+        "message" => "This is a message!",
+        "uuid" => SecureRandom.uuid,
+        "number" => BigDecimal.new("4321.1234"),
+        "utf8" => "żółć"
+    } }
+    let(:event) { LogStash::Event.new(properties) }
+    let(:connection) { double("connection") }
+    let(:client) { double("client") }
+    let(:collection) { double("collection") }
+
+    before(:each) do
+      allow(Mongo::Client).to receive(:new).and_return(connection)
+      allow(connection).to receive(:use).and_return(client)
+      allow(client).to receive(:[]).and_return(collection)
+      allow(collection).to receive(:bulk_write)
+      subject.register
+    end
+
+    after(:each) do
+      subject.close
+    end
+
+    [
+        {:query_key => nil, :query_value => "qv", :upsert => false,
+         :expected => {:query_key => "_id", :query_value => "qv", :upsert => false}
+        },
+        {:query_key => "qk", :query_value => "qv", :upsert => false,
+         :expected => {:query_key => "qk", :query_value => "qv", :upsert => false}
+        },
+        {:query_key => "qk", :query_value => "qv", :upsert => nil,
+         :expected => {:query_key => "qk", :query_value => "qv", :upsert => false}
+        },
+        {:query_key => nil, :query_value => "qv", :upsert => true,
+         :expected => {:query_key => "_id", :query_value => "qv", :upsert => true}
+        },
+        {:query_key => "qk", :query_value => "qv", :upsert => true,
+         :expected => {:query_key => "qk", :query_value => "qv", :upsert => true}
+        },
+    ].each do |test|
+
+      describe "when processing an event with query_key set to '#{test[:query_key]}', query_value set to '#{test[:query_value]}' and upsert set to '#{test[:upsert]}'" do
+
+        let(:config) {
+          configuration = {
+              "uri" => uri,
+              "database" => database,
+              "collection" => collection,
+              "action" => action
+          }
+          unless test[:query_key].nil?
+            configuration["query_key"] = test[:query_key]
+          end
+          unless test[:query_value].nil?
+            configuration["query_value"] = test[:query_value]
+          end
+          unless test[:upsert].nil?
+            configuration["upsert"] = test[:upsert]
+          end
+          return configuration
+        }
+
+        expected = test[:expected]
+        it "should send that document as a replace to mongodb with query_key '#{expected[:query_key]}', query_value '#{expected[:query_value]}' and upsert '#{expected[:upsert]}'" do
+          expect(event).to receive(:timestamp).and_return(nil)
+          expect(event).to receive(:to_hash).and_return(properties)
+          expect(collection).to receive(:bulk_write).with(
+              [{:replace_one => {:filter => {expected[:query_key] => expected[:query_value]}, :replacement => properties, :upsert => expected[:upsert]}}]
+          )
+          subject.receive(event)
+        end
+      end
+
+    end
+
+  end
+end

spec/outputs/mongodb_update_nested_fields_spec.rb

@@ -0,0 +1,81 @@
+# encoding: utf-8
+require_relative "../spec_helper"
+require "logstash/plugin"
+
+describe LogStash::Outputs::Mongodb do
+
+  let(:uri) { 'mongodb://localhost:27017' }
+  let(:database) { 'logstash' }
+  let(:collection) { 'logs' }
+  let(:action) { 'update' }
+  let(:query_value) { 'qv' }
+
+  let(:config) { {
+      "uri" => uri,
+      "database" => database,
+      "collection" => collection,
+      "action" => action,
+      "query_value" => query_value
+  } }
+
+  describe "receive method while action is 'update'" do
+    subject! { LogStash::Outputs::Mongodb.new(config) }
+
+    let(:properties) { {
+        "message" => "This is a message!",
+        "hashField" => {
+            "numField" => 1,
+            "hashField" => {
+                "numField": 2
+            },
+            "arrayField" => ["one", "two", "three"]
+        },
+        "arrayField": [
+            {"strField" => "four"},
+            {"strField" => "five"},
+            {"strField" => "six"},
+            "numField" => 3
+        ]
+    } }
+    let(:event) { LogStash::Event.new(properties) }
+    let(:connection) { double("connection") }
+    let(:client) { double("client") }
+    let(:collection) { double("collection") }
+
+    before(:each) do
+      allow(Mongo::Client).to receive(:new).and_return(connection)
+      allow(connection).to receive(:use).and_return(client)
+      allow(client).to receive(:[]).and_return(collection)
+      allow(collection).to receive(:bulk_write)
+      subject.register
+    end
+
+    after(:each) do
+      subject.close
+    end
+
+    describe "when processing an event with nested hash" do
+
+      it "should send a document update to mongodb with dotted notation" do
+        expect(event).to receive(:timestamp).and_return(nil)
+        expect(event).to receive(:to_hash).and_return(properties)
+        expect(collection).to receive(:bulk_write).with(
+            [{:update_one => {:filter => {"_id" => query_value}, :update => {"$set" => {
+                "message" => "This is a message!",
+                "hashField.numField" => 1,
+                "hashField.hashField.numField" => 2,
+                "hashField.arrayField.0" => "one",
+                "hashField.arrayField.1" => "two",
+                "hashField.arrayField.2" => "three",
+                "arrayField.0.strField" => "four",
+                "arrayField.1.strField" => "five",
+                "arrayField.2.strField" => "six",
+                "arrayField.3.numField" => 3,
+            }}, :upsert => false}}]
+        )
+        subject.receive(event)
+      end
+    end
+
+  end
+end