/*
 * search-doc-service
 * Copyright (C) 2020 Memoriav
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Affero General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Affero General Public License for more details.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program.  If not, see <https://www.gnu.org/licenses/>.
 */
package ch.memobase

import ch.memobase.helpers.ElasticSearchWrapper
import ch.memobase.model.FacetContainer
import ch.memobase.model.LanguageContainer
import ch.memobase.params.TestParam
import io.mockk.every
import io.mockk.mockk
import java.io.File
import java.nio.charset.Charset
import java.util.stream.Stream
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

@TestInstance(TestInstance.Lifecycle.PER_CLASS)
class IntegrationTest {
    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 elasticSearchWrapperMocked = run {
        val internal = mockk<ElasticSearchWrapper>()
        every { internal.getRecordSetName("soz-004") } returns LanguageContainer.placeholder("soz-004")
        every { internal.getExtraInstitutionsFromRecordSet("soz-004", "accessInstitution") } returns emptyList()
        every { internal.getExtraInstitutionsFromRecordSet("soz-004", "originalInstitution") } returns emptyList()
        every { internal.getExtraInstitutionsFromRecordSet("soz-004", "masterInstitution") } returns emptyList()
        every { internal.getInstitutionNamesFromRecordSet("soz-004") } returns listOf(FacetContainer.placeholder("soz"))
        internal
    }

    @ParameterizedTest
    @MethodSource("testParams")
    fun `integration tests`(params: TestParam) {
        val settings = App.createSettings("kafkaTest1.yml")

        every { elasticSearchWrapperMocked.countTotalNumberOfDocuments("") } returns 100
        every { elasticSearchWrapperMocked.countNumberOfDocumentsPublished("") } returns 90
        every { elasticSearchWrapperMocked.getDocumentTypesFromRecords("", "") } returns listOf(
            FacetContainer(
                LanguageContainer(listOf("Foto"), listOf("Foto"), listOf("Foto"), emptyList()), "Foto", emptyList()
            )
        )


        val topology =
            KafkaTopology(settings, TestUtilities.translationMappers, elasticSearchWrapperMocked)
        val testDriver = TopologyTestDriver(topology.build(), settings.kafkaStreamsSettings)
        val factory = ConsumerRecordFactory(
            StringSerializer(), StringSerializer()
        )
        testDriver.pipeInput(
            factory.create(
                settings.inputTopic, params.inputKey, readFile("${params.count}/input.json")
            )
        )

        val record = testDriver.readOutput(
            settings.outputTopic,
            StringDeserializer(),
            StringDeserializer()
        )

        assertAll("",
            {
                assertThat(record.value())
                    .isEqualTo(readFile("${params.count}/output.json"))
            },
            { assertThat(record.key()).isEqualTo(params.outputKey) }
        )
    }

    private fun testParams() = Stream.of(
        TestParam(
            "test-1",
            1,
            "kafkaTest1.yml",
            "https://memobase.ch/record/fad-001-DON3196",
            "fad-001-DON3196"
        ),
        TestParam(
            "test-2",
            2,
            "kafkaTest1.yml",
            "Burgerbib-Krebser-208576",
            "Burgerbib-Krebser-208576"
        ),
        TestParam(
            "test-3",
            3,
            "kafkaTest1.yml",
            "LS-film-162354",
            "LS-film-162354"
        )
    )
}