/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements. See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License. You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.kafka.coordinator.transaction;

import org.apache.kafka.common.config.AbstractConfig;

import org.junit.jupiter.api.Test;

import java.util.Arrays;
import java.util.Set;
import java.util.stream.Collectors;

import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.verifyNoMoreInteractions;

class TransactionLogConfigTest {
    @Test
    void ShouldDefineAllConfigInConfigDef() {
        Set<String> declaredConfigs = Arrays.stream(TransactionLogConfig.class.getDeclaredFields())
                .filter(field -> field.getName().endsWith("_CONFIG"))
                .peek(field -> field.setAccessible(true))
                .map(field -> assertDoesNotThrow(() -> (String) field.get(null)))
                .collect(Collectors.toSet());
        assertEquals(declaredConfigs,  TransactionLogConfig.CONFIG_DEF.names());
    }

    @SuppressWarnings("ResultOfMethodCallIgnored")
    @Test
    void ShouldGetStaticValueFromClassAttribute() {
        AbstractConfig config = mock(AbstractConfig.class);
        doReturn(1).when(config).getInt(TransactionLogConfig.TRANSACTIONS_TOPIC_MIN_ISR_CONFIG);
        doReturn(2).when(config).getInt(TransactionLogConfig.TRANSACTIONS_LOAD_BUFFER_SIZE_CONFIG);
        doReturn((short) 3).when(config).getShort(TransactionLogConfig.TRANSACTIONS_TOPIC_REPLICATION_FACTOR_CONFIG);
        doReturn(4).when(config).getInt(TransactionLogConfig.TRANSACTIONS_TOPIC_PARTITIONS_CONFIG);
        doReturn(5).when(config).getInt(TransactionLogConfig.TRANSACTIONS_TOPIC_SEGMENT_BYTES_CONFIG);
        doReturn(6).when(config).getInt(TransactionLogConfig.PRODUCER_ID_EXPIRATION_CHECK_INTERVAL_MS_CONFIG);
        doReturn(false).when(config).getBoolean(TransactionLogConfig.TRANSACTION_PARTITION_VERIFICATION_ENABLE_CONFIG);
        doReturn(88).when(config).getInt(TransactionLogConfig.PRODUCER_ID_EXPIRATION_MS_CONFIG);
        
        TransactionLogConfig transactionLogConfig = new TransactionLogConfig(config);

        assertEquals(1, transactionLogConfig.transactionTopicMinISR());
        assertEquals(2, transactionLogConfig.transactionLoadBufferSize());
        assertEquals((short) 3, transactionLogConfig.transactionTopicReplicationFactor());
        assertEquals(4, transactionLogConfig.transactionTopicPartitions());
        assertEquals(5, transactionLogConfig.transactionTopicSegmentBytes());
        assertEquals(6, transactionLogConfig.producerIdExpirationCheckIntervalMs());
        assertFalse(transactionLogConfig.transactionPartitionVerificationEnable());
        assertEquals(88, transactionLogConfig.producerIdExpirationMs());

        // This TransactionLogConfig instance is expected to be replaced entirely when dynamic config updates occur.
        // Calling the getters a second time ensures they return the values established at construction,
        // rather than invoking AbstractConfig again on each call.
        transactionLogConfig.transactionTopicMinISR();
        transactionLogConfig.transactionLoadBufferSize();
        transactionLogConfig.transactionTopicReplicationFactor();
        transactionLogConfig.transactionTopicPartitions();
        transactionLogConfig.transactionTopicSegmentBytes();
        transactionLogConfig.producerIdExpirationCheckIntervalMs();
        transactionLogConfig.transactionPartitionVerificationEnable();
        transactionLogConfig.producerIdExpirationMs();

        verify(config, times(1)).getInt(TransactionLogConfig.TRANSACTIONS_TOPIC_MIN_ISR_CONFIG);
        verify(config, times(1)).getInt(TransactionLogConfig.TRANSACTIONS_LOAD_BUFFER_SIZE_CONFIG);
        verify(config, times(1)).getShort(TransactionLogConfig.TRANSACTIONS_TOPIC_REPLICATION_FACTOR_CONFIG);
        verify(config, times(1)).getInt(TransactionLogConfig.TRANSACTIONS_TOPIC_PARTITIONS_CONFIG);
        verify(config, times(1)).getInt(TransactionLogConfig.TRANSACTIONS_TOPIC_SEGMENT_BYTES_CONFIG);
        verify(config, times(1)).getInt(TransactionLogConfig.PRODUCER_ID_EXPIRATION_CHECK_INTERVAL_MS_CONFIG);
        verify(config, times(1)).getBoolean(TransactionLogConfig.TRANSACTION_PARTITION_VERIFICATION_ENABLE_CONFIG);
        verify(config, times(1)).getInt(TransactionLogConfig.PRODUCER_ID_EXPIRATION_MS_CONFIG);
        verifyNoMoreInteractions(config);
    }
}