Merge pull request #80 from RedPillAnalytics/develop

Basic Authentication

src/ksqlServerTest/groovy/KsqlRestAuthTest.groovy

@@ -0,0 +1,62 @@
+import com.redpillanalytics.KsqlRest
+import groovy.util.logging.Slf4j
+import spock.lang.Shared
+import spock.lang.Specification
+
+@Slf4j
+class KsqlRestAuthTest extends Specification {
+
+   @Shared
+   String pipelineEndpoint = System.getProperty("pipelineEndpoint") ?: 'http://localhost:8088'
+   String username = System.getProperty("pipelineUsername") ?: 'test'
+   String password = System.getProperty("pipelinePassword") ?: 'test'
+
+   @Shared
+   def ksqlRest = new KsqlRest(restUrl: pipelineEndpoint, username: username, password: password)
+
+   def "KSQL Server properties fetched"() {
+
+      when:
+      def result = ksqlRest.getProperties()
+
+      then:
+      log.warn "result: ${result.toString()}"
+      result
+   }
+
+   def "KSQL extension directory path is returned"() {
+
+      when:
+      def path = ksqlRest.getExtensionPath()
+
+      then:
+      path
+   }
+
+   def "KSQL extension directory file is returned"() {
+
+      when:
+      def dir = ksqlRest.getExtensionDir()
+
+      then:
+      dir
+   }
+
+   def "KSQL REST URL is returned"() {
+
+      when:
+      def url = ksqlRest.getSchemaRegistry()
+
+      then:
+      url
+   }
+
+   def "List of topics returned"() {
+
+      when:
+      def topics = ksqlRest.getTopics()
+
+      then:
+      topics
+   }
+}

src/ksqlServerTest/groovy/KsqlRestTest.groovy

@@ -1,6 +1,5 @@
 import com.redpillanalytics.KsqlRest
 import groovy.util.logging.Slf4j
-import spock.lang.Ignore
 import spock.lang.Shared
 import spock.lang.Specification
 

src/main/groovy/com/redpillanalytics/KsqlRest.groovy

@@ -44,17 +44,17 @@ class KsqlRest {
       def prepared = (ksql + ';').replace('\n', '').replace(';;', ';')
 
       if (['create', 'drop'].contains(getStatementType(ksql))) log.info prepared
+
+      if (username && password) {
+         Unirest.config().setDefaultBasicAuth(username, password)
+      }
+
       HttpResponse<String> response = Unirest.post("${restUrl}/ksql")
               .header("Content-Type", "application/vnd.ksql.v1+json")
               .header("Cache-Control", "no-cache")
-              .header("Postman-Token", "473fbb05-9da1-4020-95c0-f2c60fed8289")
               .body(JsonOutput.toJson([ksql: prepared, streamsProperties: properties]))
               .asString()
 
-      // streamline the addition of basic credentials slightly
-      if (username && password)
-         response.header("Authorization", "Basic " + "${username}:${password}".bytes.encodeBase64().toString())
-
       log.debug "unirest response: ${response.dump()}"
       def body = new JsonSlurper().parseText(response.body)
 
@@ -229,7 +229,7 @@ class KsqlRest {
 
       def result = execKsql(ksql, properties)
 
-         log.debug "result: ${result}"
+      log.debug "result: ${result}"
 
       if (result.status == 400 && result.body.message.contains('Incompatible data source type is STREAM')) {
          log.info "Type is now STREAM. Issuing DROP STREAM..."