Merge pull request #217 from splunk/crud-tests

Crud tests

Author: chaitanyaphalak

test/commonkafka.py

@@ -18,6 +18,7 @@
 import requests
 import sys
 import json
+import time
 
 logger = logging.getLogger(__name__)
 logger.setLevel(logging.INFO)
@@ -31,25 +32,132 @@ def create_kafka_connector(setup, params):
     '''
     Create kafka connect connector using kafka connect REST API
     '''
-    response = requests.post(url=setup["kafka_connect_url"]+"/connectors", data=json.dumps(params),
+    response = requests.post(url=setup["kafka_connect_url"] + "/connectors", data=json.dumps(params),
                       headers={'Accept': 'application/json', 'Content-Type': 'application/json'})
+    if response.status_code != 201:
+        logger.error("Failed to create connector, response code - ", response.status_code)
+        return False
+    else:
+        status = get_kafka_connector_status(setup, params)
+        while status is not None and status["connector"]["state"] != "RUNNING":
+            status = get_kafka_connector_status(setup, params)
+
+        time.sleep(10)
+        status = get_kafka_connector_status(setup, params)
+        if status is not None and len(status["tasks"]) > 0 and status["tasks"][0]["state"] == "RUNNING":
+            logger.info("Created connector successfully - " + json.dumps(params))
+            return True
+
+    logger.error("Failed to create connector, connector and tasks are not in a RUNNING state after 10 seconds")
+    return False
 
-    if response.status_code == 201:
-        logger.info("Created connector successfully - " + json.dumps(params))
+def update_kafka_connector(setup, params):
+    '''
+    Update kafka connect connector using kafka connect REST API
+    '''
+    response = requests.put(url=setup["kafka_connect_url"] + "/connectors/" + params["name"] + "/config", data=json.dumps(params["config"]),
+                      headers={'Accept': 'application/json', 'Content-Type': 'application/json'})
+
+    if response.status_code == 200:
+        status = get_kafka_connector_status(setup, params)
+        while status is not None and status["connector"]["state"] != "RUNNING":
+            status = get_kafka_connector_status(setup, params)
+        logger.info("Updated connector successfully - " + json.dumps(params))
         return True
 
+    logger.error("Failed to update connector, response code - ", response.status_code)
     return False
 
 def delete_kafka_connector(setup, params):
     '''
     Delete kafka connect connector using kafka connect REST API
     '''
-    response = requests.delete(url=setup["kafka_connect_url"]+"/connectors/" + params["name"],
+    response = requests.delete(url=setup["kafka_connect_url"] + "/connectors/" + params["name"],
                         headers={'Accept': 'application/json', 'Content-Type': 'application/json'})
+
     if response.status_code == 204:
         logger.info("Deleted connector successfully - " + json.dumps(params))
         return True
-    else:
-        logger.error("failed to create connector", param)
-        logger.error(response.status_code)
-        return False
+
+    logger.error("Failed to delete connector, response code - ", response.status_code)
+    return False
+
+def get_kafka_connector_tasks(setup, params):
+    '''
+    Get kafka connect connector tasks using kafka connect REST API
+    '''
+
+    t_end = time.time() + 10
+    while time.time() < t_end:
+        response = requests.get(url=setup["kafka_connect_url"] + "/connectors/" + params["name"] + "/tasks",
+                                headers={'Accept': 'application/json', 'Content-Type': 'application/json'})
+        status = response.status_code
+        if status == 200:
+            return len(response.json())
+
+    return 0
+
+def get_kafka_connector_status(setup, params):
+    '''
+    Get kafka connect connector tasks using kafka connect REST API
+    '''
+    t_end = time.time() + 10
+    while time.time() < t_end:
+        response = requests.get(url=setup["kafka_connect_url"] + "/connectors/" + params["name"] + "/status",
+                                headers={'Accept': 'application/json', 'Content-Type': 'application/json'})
+        status = response.status_code
+        if status == 200:
+            return response.json()
+
+    return None
+
+def pause_kafka_connector(setup, params):
+    '''
+    Pause kafka connect connector using kafka connect REST API
+    '''
+    response = requests.put(url=setup["kafka_connect_url"] + "/connectors/" + params["name"] + "/pause",
+                      headers={'Accept': 'application/json', 'Content-Type': 'application/json'})
+
+    if response.status_code == 202:
+        status = get_kafka_connector_status(setup, params)
+        while status is not None and status["connector"]["state"] != "PAUSED":
+            status = get_kafka_connector_status(setup, params)
+        logger.info("Paused connector successfully")
+        return True
+
+    logger.error("Failed to pause connector, response code - ", response.status_code)
+    return False
+
+def resume_kafka_connector(setup, params):
+    '''
+    Resume kafka connect connector using kafka connect REST API
+    '''
+    response = requests.put(url=setup["kafka_connect_url"] + "/connectors/" + params["name"] + "/resume",
+                      headers={'Accept': 'application/json', 'Content-Type': 'application/json'})
+
+    if response.status_code == 202:
+        status = get_kafka_connector_status(setup, params)
+        while status is not None and status["connector"]["state"] != "RUNNING":
+            status = get_kafka_connector_status(setup, params)
+        logger.info("Resumed connector successfully")
+        return True
+
+    logger.error("Failed to resume connector, response code - ", response.status_code)
+    return False
+
+def restart_kafka_connector(setup, params):
+    '''
+    Restart kafka connect connector using kafka connect REST API
+    '''
+    response = requests.post(url=setup["kafka_connect_url"] + "/connectors/" + params["name"] + "/restart",
+                             headers={'Accept': 'application/json', 'Content-Type': 'application/json'})
+
+    if response.status_code == 200 or response.status_code == 204:
+        status = get_kafka_connector_status(setup, params)
+        while status is not None and status["connector"]["state"] != "RUNNING":
+            status = get_kafka_connector_status(setup, params)
+        logger.info("Restarted connector successfully")
+        return True
+
+    logger.error("Failed to restart connector, response code - ", response.status_code)
+    return False

test/kafka_connect_sink_crud_tests/test_crud.py

@@ -1,7 +1,8 @@
 import pytest
 import logging
 import sys
-from ..commonkafka import create_kafka_connector, delete_kafka_connector
+import time
+from ..commonkafka import create_kafka_connector, delete_kafka_connector, update_kafka_connector, get_kafka_connector_tasks, get_kafka_connector_status, pause_kafka_connector, resume_kafka_connector, restart_kafka_connector
 
 logger = logging.getLogger(__name__)
 logger.setLevel(logging.INFO)
@@ -11,13 +12,13 @@
 logger.addHandler(handler)
 
 @pytest.mark.parametrize("test_input,expected", [
-    ("create_and_delete_valid_task", True)
+    ("test_valid_CRUD_tasks", True)
 ])
-def test_create_and_delete_valid_task(setup, test_input, expected):
+def test_valid_CRUD_tasks(setup, test_input, expected):
     '''
-    Test that valid kafka connect task can be created
+    Test that valid kafka connect task can be created, updated, paused, resumed, restarted and deleted
     '''
-    logger.info("testing create_and_delete_valid_task input={0} expected={1} ".format(
+    logger.info("testing test_valid_CRUD_tasks input={0} expected={1} ".format(
         test_input, expected))
 
     # defining a connector definition dict for the parameters to be sent to the API
@@ -26,16 +27,153 @@ def test_create_and_delete_valid_task(setup, test_input, expected):
             "config": {
                "connector.class": "com.splunk.kafka.connect.SplunkSinkConnector",
                "tasks.max": "3",
-               "topics": "test-datagen",  # set kafka topic later
-               "splunk.indexes": setup["kafka_topic"],
+               "topics": setup["kafka_topic"],
+               "splunk.indexes": setup["splunk_index"],
                "splunk.hec.uri": setup["splunkd_url"],
                "splunk.hec.token": setup["splunk_token"],
                "splunk.hec.raw": "false",
                "splunk.hec.ack.enabled": "false",
-               "splunk.hec.ssl.validate.certs": "true"
+               "splunk.hec.ssl.validate.certs": "false"
               }
             }
 
+    #Validate create task
     assert create_kafka_connector(setup, connector_definition) == expected
 
-    assert delete_kafka_connector(setup, connector_definition) == expected
+    # updating the definition to use 5 tasks instead of 3
+    connector_definition = {
+            "name": "kafka-connect-splunk",
+            "config": {
+               "connector.class": "com.splunk.kafka.connect.SplunkSinkConnector",
+               "tasks.max": "5",
+               "topics": setup["kafka_topic"],
+               "splunk.indexes": setup["splunk_index"],
+               "splunk.hec.uri": setup["splunkd_url"],
+               "splunk.hec.token": setup["splunk_token"],
+               "splunk.hec.raw": "false",
+               "splunk.hec.ack.enabled": "false",
+               "splunk.hec.ssl.validate.certs": "false"
+              }
+            }
+
+    # Validate update task
+    assert update_kafka_connector(setup, connector_definition) == expected
+
+    # Validate get tasks
+    tasks = get_kafka_connector_tasks(setup, connector_definition)
+    assert tasks == int(connector_definition["config"]["tasks.max"])
+
+    # Validate pause task
+    assert pause_kafka_connector(setup, connector_definition) == expected
+
+    # Validate resume task
+    assert resume_kafka_connector(setup, connector_definition) == expected
+
+    # Validate restart task
+    assert restart_kafka_connector(setup, connector_definition) == expected
+
+    # Validate delete task
+    assert delete_kafka_connector(setup, connector_definition) == expected
+
+
+@pytest.mark.parametrize("test_input,expected", [
+    ("test_invalid_CRUD_tasks", False)
+])
+def test_invalid_CRUD_tasks(setup, test_input, expected):
+    '''
+    Test that invalid kafka connect task cannot be created
+    '''
+    logger.info("testing test_invalid_CRUD_tasks input={0} expected={1} ".format(
+        test_input, expected))
+
+    # connector definition with tasks.max invalid(not number)
+    connector_definition_invalid_tasks = {
+            "name": "kafka-connect-splunk",
+            "config": {
+               "connector.class": "com.splunk.kafka.connect.SplunkSinkConnector",
+               "tasks.max": "dummy-string",
+               "topics": setup["kafka_topic"],
+               "splunk.indexes": setup["splunk_index"],
+               "splunk.hec.uri": setup["splunkd_url"],
+               "splunk.hec.token": setup["splunk_token"],
+               "splunk.hec.raw": "false",
+               "splunk.hec.ack.enabled": "false",
+               "splunk.hec.ssl.validate.certs": "false"
+              }
+            }
+
+    assert create_kafka_connector(setup, connector_definition_invalid_tasks) == expected
+
+    # connector definition with splunk.hec.raw invalid(not boolean)
+    connector_definition_invalid_tasks = {
+            "name": "kafka-connect-splunk",
+            "config": {
+               "connector.class": "com.splunk.kafka.connect.SplunkSinkConnector",
+               "tasks.max": "3",
+               "topics": setup["kafka_topic"],
+               "splunk.indexes": setup["splunk_index"],
+               "splunk.hec.uri": setup["splunkd_url"],
+               "splunk.hec.token": setup["splunk_token"],
+               "splunk.hec.raw": "disable",
+               "splunk.hec.ack.enabled": "false",
+               "splunk.hec.ssl.validate.certs": "false"
+              }
+            }
+
+    assert create_kafka_connector(setup, connector_definition_invalid_tasks) == expected
+
+    # connector definition with topics invalid(empty string)
+    connector_definition_invalid_tasks = {
+            "name": "kafka-connect-splunk",
+            "config": {
+               "connector.class": "com.splunk.kafka.connect.SplunkSinkConnector",
+               "tasks.max": "3",
+               "topics": "",
+               "splunk.indexes": setup["splunk_index"],
+               "splunk.hec.uri": setup["splunkd_url"],
+               "splunk.hec.token": setup["splunk_token"],
+               "splunk.hec.raw": "false",
+               "splunk.hec.ack.enabled": "false",
+               "splunk.hec.ssl.validate.certs": "false"
+              }
+            }
+
+    assert create_kafka_connector(setup, connector_definition_invalid_tasks) == expected
+
+    # connector definition with splunk.hec.json.event.enrichment invalid(non key value pairs)
+    connector_definition_invalid_tasks = {
+            "name": "kafka-connect-splunk",
+            "config": {
+               "connector.class": "com.splunk.kafka.connect.SplunkSinkConnector",
+               "tasks.max": "3",
+               "topics": setup["kafka_topic"],
+               "splunk.indexes": setup["splunk_index"],
+               "splunk.hec.uri": setup["splunkd_url"],
+               "splunk.hec.token": setup["splunk_token"],
+               "splunk.hec.raw": "false",
+               "splunk.hec.ack.enabled": "false",
+               "splunk.hec.ssl.validate.certs": "false",
+               "splunk.hec.json.event.enrichment": "testing-testing non KV"
+              }
+            }
+
+    assert create_kafka_connector(setup, connector_definition_invalid_tasks) == expected
+
+    # connector definition with splunk.hec.json.event.enrichment invalid(key value pairs not separated by commas)
+    connector_definition_invalid_tasks = {
+            "name": "kafka-connect-splunk",
+            "config": {
+               "connector.class": "com.splunk.kafka.connect.SplunkSinkConnector",
+               "tasks.max": "3",
+               "topics": setup["kafka_topic"],
+               "splunk.indexes": setup["splunk_index"],
+               "splunk.hec.uri": setup["splunkd_url"],
+               "splunk.hec.token": setup["splunk_token"],
+               "splunk.hec.raw": "false",
+               "splunk.hec.ack.enabled": "false",
+               "splunk.hec.ssl.validate.certs": "false",
+                "splunk.hec.json.event.enrichment": "key1=value1 key2=value2"
+              }
+            }
+
+    assert create_kafka_connector(setup, connector_definition_invalid_tasks) == expected