Updated testing(unit and integrated) (#372)

Author: deep-splunk

src/main/java/com/splunk/kafka/connect/SplunkSinkConnector.java

@@ -158,7 +158,17 @@ private void validateSplunkConfigurations(final Map<String, String> configs) thr
     }
 
     private void preparePayloadAndExecuteRequest(SplunkSinkConnectorConfig connectorConfig, String index) throws ConfigException {
-        Header[] headers = new Header[]{new BasicHeader("Authorization", String.format("Splunk %s", connectorConfig.splunkToken))};
+        Header[] headers;
+        if (connectorConfig.ack) {
+            headers = new Header[]{
+                new BasicHeader("Authorization", String.format("Splunk %s", connectorConfig.splunkToken)),
+                new BasicHeader("X-Splunk-Request-Channel", java.util.UUID.randomUUID().toString())
+            };
+        } else {
+            headers = new Header[]{
+                new BasicHeader("Authorization", String.format("Splunk %s", connectorConfig.splunkToken)),
+            };
+        }
         String endpoint = "/services/collector";
         String url = connectorConfig.splunkURI + endpoint;
         final HttpPost httpPost = new HttpPost(url);

src/test/java/com/splunk/kafka/connect/SplunkSinkConnecterTest.java

@@ -131,6 +131,44 @@ public void testInvalidKerberosOnlyKeytabSet() {
         assertHasErrorMessage(result, KERBEROS_KEYTAB_PATH_CONF, "must be set");
     }
 
+    @Test
+    public void testInvalidJsonEventEnrichmentConfig1() {
+        final Map<String, String> configs = new HashMap<>();
+        addNecessaryConfigs(configs);
+        SinkConnector connector = new SplunkSinkConnector();
+        configs.put("topics", "b");
+        configs.put("tasks_max", "3");
+        configs.put("splunk.hec.json.event.enrichment", "k1=v1 k2=v2");
+        MockHecClientWrapper clientInstance = new MockHecClientWrapper();
+        ((SplunkSinkConnector) connector).setHecInstance(clientInstance);
+        Assertions.assertThrows(ConfigException.class, ()->connector.validate(configs));
+    }
+
+    @Test
+    public void testInvalidJsonEventEnrichmentConfig2() {
+        final Map<String, String> configs = new HashMap<>();
+        addNecessaryConfigs(configs);
+        SinkConnector connector = new SplunkSinkConnector();
+        configs.put("topics", "b");
+        configs.put("splunk.hec.json.event.enrichment", "testing-testing non KV");
+        MockHecClientWrapper clientInstance = new MockHecClientWrapper();
+        ((SplunkSinkConnector) connector).setHecInstance(clientInstance);
+        Assertions.assertThrows(ConfigException.class, ()->connector.validate(configs));
+    }
+
+    @Test
+    public void testInvalidJsonEventEnrichmentConfig3() {
+        final Map<String, String> configs = new HashMap<>();
+        addNecessaryConfigs(configs);
+        SinkConnector connector = new SplunkSinkConnector();
+        configs.put("topics", "b");
+        configs.put("tasks_max", "3");
+        configs.put("splunk.hec.json.event.enrichment", "k1=v1 k2=v2");
+        MockHecClientWrapper clientInstance = new MockHecClientWrapper();
+        ((SplunkSinkConnector) connector).setHecInstance(clientInstance);
+        Assertions.assertThrows(ConfigException.class, ()->connector.validate(configs));
+    }
+
     @Test
     public void testInvalidToken() {
         final Map<String, String> configs = new HashMap<>();
@@ -144,6 +182,18 @@ public void testInvalidToken() {
         Assertions.assertThrows(ConfigException.class, ()->connector.validate(configs));
     }
 
+    @Test
+    public void testNullHecToken() {
+        final Map<String, String> configs = new HashMap<>();
+        addNecessaryConfigs(configs);
+        SinkConnector connector = new SplunkSinkConnector();
+        configs.put("topics", "b");
+        configs.put("splunk.hec.token", null);
+        MockHecClientWrapper clientInstance = new MockHecClientWrapper();
+        ((SplunkSinkConnector) connector).setHecInstance(clientInstance);
+        Assertions.assertThrows(java.lang.NullPointerException.class, ()->connector.validate(configs));
+    }
+
     @Test
     public void testInvalidIndex() {
         final Map<String, String> configs = new HashMap<>();

test/lib/connect_params.py

@@ -83,10 +83,6 @@
      "tasks_max": "3",
      "splunk_hec_raw": False,
      "splunk_hec_json_event_enrichment": "chars=test_tasks_max_3_hec_raw_false"},
-    {"name": "test_tasks_max_null",
-     "tasks_max": "null",
-     "splunk_hec_raw": False,
-     "splunk_hec_json_event_enrichment": "chars=test_tasks_max_null"},
     {"name": "test_1_source_hec_raw_true",
      "splunk_hec_raw": True,
      "splunk_sources": "test_1_source_hec_raw_true"},
@@ -179,12 +175,6 @@
      "splunk_hec_raw": True,
      "splunk_hec_json_event_formatted": "false",
      "splunk_sourcetypes": "test_splunk_hec_json_event_formatted_false_raw_data"},
-    {"name": "test_empty_hec_token",
-     "splunk_hec_token": None,
-     "splunk_hec_json_event_enrichment": "chars=test_empty_hec_token"},
-    {"name": "test_incorrect_hec_token",
-     "splunk_hec_token": "dummy-tocken",
-     "splunk_hec_json_event_enrichment": "chars=test_incorrect_hec_token"},
     {"name": "test_splunk_hec_empty_event",
      "topics": "test_splunk_hec_malformed_events",
      "splunk_hec_raw": False,
@@ -215,4 +205,4 @@
      "enable_timestamp_extraction" : "true",
      "timestamp_regex": r"\\\"time\\\":\\s*\\\"(?<time>.*?)\"",
      "timestamp_format": "epoch"}
-]
+]

test/testcases/test_configurations.py

@@ -25,7 +25,7 @@ def setup_class(self, setup):
         ("test_tasks_max_1_hec_raw_false", "chars::test_tasks_max_1_hec_raw_false", 3),
         # ("test_tasks_max_3_hec_raw_true", "sourcetype::raw_data-tasks_max_3", 1),
         # ("test_tasks_max_3_hec_raw_false", "chars::test_tasks_max_3_hec_raw_false", 3),
-        ("test_tasks_max_null", "chars::test_tasks_max_null", 0)
+        # ("test_tasks_max_null", "chars::test_tasks_max_null", 0)
     ])
     def test_tasks_max(self, setup, test_scenario, test_input, expected):
         logger.info(f"testing {test_scenario} input={test_input} expected={expected} event(s)")
@@ -141,21 +141,6 @@ def test_header_support_true_event_data(self, setup, test_scenario, test_input,
         logger.info("Splunk received %s events in the last 15m", len(events))
         assert len(events) == expected
 
-    @pytest.mark.parametrize("test_case, test_input, expected", [
-        ("test_incorrect_hec_token", "chars::test_incorrect_hec_token", 0),
-        ("test_empty_hec_token", "chars::test_empty_hec_token", 0)
-    ])
-    def test_create_connector_with_incorrect_hec_token(self, setup, test_case, test_input, expected):
-        search_query = f"index={setup['kafka_header_index']} | search timestamp=\"{setup['timestamp']}\" {test_input}"
-        logger.info(search_query)
-        events = check_events_from_splunk(start_time="-15m@m",
-                                          url=setup["splunkd_url"],
-                                          user=setup["splunk_user"],
-                                          query=[f"search {search_query}"],
-                                          password=setup["splunk_password"])
-        logger.info("Splunk received %s events in the last 15m", len(events))
-        assert len(events) == expected
-
     @pytest.mark.parametrize("test_scenario, test_input, expected", [
         ("test_splunk_hec_json_event_formatted_true_event_data",
          "chars::test_splunk_hec_json_event_formatted_true_event_data", 3),

test/testcases/test_crud.py

@@ -36,7 +36,7 @@ def test_valid_crud_tasks(self, setup, test_input, expected):
                 "tasks.max": "3",
                 "topics": setup["kafka_topic"],
                 "splunk.indexes": setup["splunk_index"],
-                "splunk.hec.uri": setup["splunkd_url"],
+                "splunk.hec.uri": setup["splunk_hec_url"],
                 "splunk.hec.token": setup["splunk_token"],
                 "splunk.hec.raw": "false",
                 "splunk.hec.ack.enabled": "false",
@@ -56,7 +56,7 @@ def test_valid_crud_tasks(self, setup, test_input, expected):
                 "tasks.max": "5",
                 "topics": setup["kafka_topic"],
                 "splunk.indexes": setup["splunk_index"],
-                "splunk.hec.uri": setup["splunkd_url"],
+                "splunk.hec.uri": setup["splunk_hec_url"],
                 "splunk.hec.token": setup["splunk_token"],
                 "splunk.hec.raw": "false",
                 "splunk.hec.ack.enabled": "false",
@@ -101,30 +101,25 @@ def test_invalid_crud_tasks(self, setup, test_case, config_input, expected):
 
     @pytest.mark.parametrize("test_case, config_input, expected", [
         ("event_enrichment_non_key_value", {"name": "event_enrichment_non_key_value",
-                                            "splunk_hec_json_event_enrichment": "testing-testing non KV"},
-         ["FAILED"]),
+                                            "splunk_hec_json_event_enrichment": "testing-testing non KV"}, False),
         ("event_enrichment_non_key_value_3_tasks", {"name": "event_enrichment_non_key_value_3_tasks",
                                                     "tasks_max": "3",
-                                                    "splunk_hec_json_event_enrichment": "testing-testing non KV"},
-         ["FAILED", "FAILED", "FAILED"]),
+                                                    "splunk_hec_json_event_enrichment": "testing-testing non KV"}, False),
         ("event_enrichment_not_separated_by_commas", {"name": "event_enrichment_not_separated_by_commas",
-                                                      "splunk_hec_json_event_enrichment": "key1=value1 key2=value2"},
-         ["FAILED"]),
+                                                      "splunk_hec_json_event_enrichment": "key1=value1 key2=value2"}, False),
         ("event_enrichment_not_separated_by_commas_3_tasks", {"name": "event_enrichment_not_separated_by_commas_3_tasks",
                                                               "tasks_max": "3",
-                                                              "splunk_hec_json_event_enrichment": "key1=value1 key2=value2"},
-         ["FAILED", "FAILED", "FAILED"])
+                                                              "splunk_hec_json_event_enrichment": "key1=value1 key2=value2"}, False)
 
     ])
     def test_invalid_crud_event_enrichment_tasks(self, setup, test_case, config_input, expected):
         '''
         Test that invalid event_enrichment kafka connect task can be created but task status should be FAILED
         and no data should enter splunk
         '''
-        logger.info(f"testing {test_case} input={config_input} expected={expected} ")
+        logger.info(f"testing {test_case} input={config_input}")
 
         connector_definition_invalid_tasks = generate_connector_content(config_input)
         setup['connectors'].append(test_case)
 
-        assert create_kafka_connector(setup, connector_definition_invalid_tasks) is True
-        assert get_running_kafka_connector_task_status(setup, connector_definition_invalid_tasks) == expected
+        assert create_kafka_connector(setup, connector_definition_invalid_tasks) == expected