Add test and fix some minor issues

src/main/kotlin/KafkaTopology.kt

@@ -104,7 +104,7 @@ class KafkaTopology(private val settings: SettingsLoader) {
                 val report = value.getReport()
                 if (report.status == ReportStatus.success) {
                     ProcessReport(
-                        report.id,
+                        "xml-data-transform",
                         ReportStatus.success,
                         1,
                         1,
@@ -112,7 +112,7 @@ class KafkaTopology(private val settings: SettingsLoader) {
                     )
                 } else {
                     ProcessReport(
-                        report.id,
+                        "xml-data-transform",
                         ReportStatus.failure,
                         1,
                         0,

src/main/kotlin/SAXContentHandler.kt

@@ -128,28 +128,32 @@ class SAXContentHandler(key: String, private val identifierFieldName: String, pr
         if (recordTag == localName)
             return
 
-        if (currentElementTag == localName) {
-            if (currentElementTag == identifierFieldName) {
-                identifier = currentElementContent
+        when {
+            currentElementTag == localName -> {
+                if (currentElementTag == identifierFieldName) {
+                    identifier = currentElementContent
+                }
+                if (innerElements.isEmpty()) {
+                    if (currentElementContent.isNotEmpty()) {
+                        jsonResult[currentElementTag] = currentElementContent
+                    }
+                } else {
+                    jsonResult[currentElementTag] = innerElements.toMap()
+                    innerElements.clear()
+                }
+                currentElementTag = ""
+                currentElementContent = ""
             }
-            if (innerElements.isEmpty()) {
-                if (currentElementContent.isNotEmpty()) {
-                    jsonResult[currentElementTag] = currentElementContent
+            currentInnerElementTag == localName -> {
+                if (currentInnerElementContent.isNotEmpty()) {
+                    innerElements.add(Pair(currentInnerElementTag, currentInnerElementContent))
                 }
-            } else {
-                jsonResult[currentElementTag] = innerElements.toMap()
-                innerElements.clear()
+                currentInnerElementTag = ""
+                currentInnerElementContent = ""
             }
-            currentElementTag = ""
-            currentElementContent = ""
-        } else if (currentInnerElementTag == localName) {
-            if (currentInnerElementContent.isNotEmpty()) {
-                innerElements.add(Pair(currentInnerElementTag, currentInnerElementContent))
+            else -> {
+                reportText += "Unmatched end element: $localName.\n"
             }
-            currentInnerElementTag = ""
-            currentInnerElementContent = ""
-        } else {
-            reportText += "Unmatched end element: $localName.\n"
         }
     }
 

src/test/kotlin/TestIntegration.kt

@@ -18,11 +18,107 @@
 
 package org.memobase
 
+import com.beust.klaxon.Klaxon
+import org.apache.kafka.common.serialization.StringDeserializer
+import org.apache.kafka.common.serialization.StringSerializer
+import org.apache.kafka.streams.TopologyTestDriver
+import org.apache.kafka.streams.test.ConsumerRecordFactory
+import org.apache.logging.log4j.LogManager
+import org.assertj.core.api.Assertions.assertThat
 import org.junit.jupiter.api.TestInstance
+import org.junit.jupiter.api.assertAll
+import org.junit.jupiter.params.ParameterizedTest
+import org.junit.jupiter.params.provider.MethodSource
+import org.memobase.testing.EmbeddedSftpServer
+import java.io.File
+import java.io.FileInputStream
+import java.nio.charset.Charset
+import java.nio.file.Paths
+import java.util.stream.Stream
 
 @TestInstance(TestInstance.Lifecycle.PER_CLASS)
 class TestIntegration {
+    private val log = LogManager.getLogger("TestLogger")
 
+    private val resourcePath = "src/test/resources/data"
+    private fun readFile(fileName: String): String {
+        return File("$resourcePath/$fileName").readText(Charset.defaultCharset())
+    }
 
+    private val sftpServer = EmbeddedSftpServer(22000, "user", "password")
+
+    @ParameterizedTest
+    @MethodSource("integrationTestParams")
+    fun `test kafka integration single record`(params: TestParams) {
+        sftpServer.putFile(
+            "/memobase/test${params.count}/data.xml",
+            FileInputStream("$resourcePath/${params.count}/data.xml")
+        )
+
+
+        val service = Service("test${params.count}.yml")
+        val testDriver = TopologyTestDriver(service.topology, service.settings.kafkaStreamsSettings)
+        val factory = ConsumerRecordFactory(
+            StringSerializer(), StringSerializer()
+        )
+        testDriver.pipeInput(
+            factory.create(
+                service.settings.inputTopic, params.inputKey, readFile("${params.count}/input.json")
+            )
+        )
+
+        val record = testDriver.readOutput(
+            service.settings.outputTopic,
+            StringDeserializer(),
+            StringDeserializer()
+        )
+
+        val reportedRecord = testDriver.readOutput(
+            "${service.settings.outputTopic}-reporting",
+            StringDeserializer(),
+            StringDeserializer()
+        )
+        val data = reportedRecord.value()
+        val report = Klaxon().parse<Report>(data)
+        val processReportRecord = testDriver.readOutput(
+            service.settings.processReportTopic,
+            StringDeserializer(),
+            StringDeserializer()
+        )
+        val processReport = Klaxon().parse<ProcessReport>(processReportRecord.value())
+
+        assertAll("check single record transform",
+            {
+                assertThat(record)
+                    .isNotNull
+                    .hasFieldOrPropertyWithValue("key", params.outputKey)
+                    .hasFieldOrPropertyWithValue("value", readFile("${params.count}/output.json"))
+            },
+            {
+                assertThat(report)
+                    .isNotNull
+                    .isEqualTo(params.expectedOutputReport)
+            },
+            {
+                assertThat(processReport)
+                    .isNotNull
+                    .isEqualTo(params.expectedOutputProcessReport)
+            }
+
+        )
+
+    }
+
+
+    fun integrationTestParams() = Stream.of(
+        TestParams(
+            1,
+            "simple test",
+            "data.xml",
+            "ADG-102683",
+            Report("ADG-102683", ReportStatus.success, "Successfully transformed xml to json!"),
+            ProcessReport("xml-data-transform", ReportStatus.success, 1, 1, 0)
+        )
+    )
 
 }
\ No newline at end of file

src/test/resources/data/1/input.json

 {
   "path": "/memobase/test1/data.xml",
-  "format": "ERROR"
+  "format": "XML"
 }
\ No newline at end of file

src/test/resources/data/1/output.json

-ERROR
\ No newline at end of file
+{"title":"Ausgrabung Crestaulta","SerieTitle":"Grabung Walo Burkart und Karl Keller-Tarnuzzer, Lumbrein-Surin (Crestaulta)","creatorPerson":{"name":"Karl Keller-Tarnuzzer"},"Keywords":"Ausgrabung, Archäologie","Abstract":"Herdstelle in Fläche D, 2. Situation (mittlere Schicht)","Context":"Verwandte Dokumente: RM_1_1_7","Claim":"Dieses Dokument wurde Dank der Unterstützung von Memoriav erhalten.","RecordingLocation":"Lumbrein-Surin, Crestaulta","contributorCorporateBody":{"name":"ADG"},"genre":"Grabungsfotografie / Arbeitsfotografie / Sachfotografie","objectType":"photograph","medium":"[keine Information vorhanden]","imageFormatColorMode":"sw","imageFormatRemarks":"Album-Nr.: A4","identifierOriginal":"102683","identifierCallNumber":"RM_1_3_10","identifierMain":"ADG-102683","location":"Schweiz, Graubünden, Lumbrein-Surin, Crestaulta","rightsHolder":"Archäologischer Dienst Graubünden","accessPhsyical":"onsite","accessDigital":"public"}
\ No newline at end of file

src/test/resources/test1.yml

@@ -5,8 +5,8 @@ sftp:
   password: password
 app:
   xsltFilePath: src/test/resources/data/1/config/transform.xslt
-  identifierFieldName: id
-  recordTag: root
+  identifierFieldName: identifierMain
+  recordTag: record
 kafka:
   streams:
     bootstrap.servers: localhost:12345