/*
 * 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.raft;

import org.apache.kafka.common.message.BeginQuorumEpochRequestData;
import org.apache.kafka.common.message.BeginQuorumEpochResponseData;
import org.apache.kafka.common.message.EndQuorumEpochRequestData;
import org.apache.kafka.common.message.EndQuorumEpochResponseData;
import org.apache.kafka.common.message.FetchResponseData;
import org.apache.kafka.common.message.FetchSnapshotResponseData;
import org.apache.kafka.common.message.VoteResponseData;
import org.apache.kafka.common.message.VotersRecord;
import org.apache.kafka.common.network.ListenerName;
import org.apache.kafka.common.utils.Utils;

import org.junit.jupiter.api.Test;

import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Optional;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotEquals;

final class EndpointsTest {
    private final ListenerName testListener = ListenerName.normalised("listener");
    private final InetSocketAddress testSocketAddress = InetSocketAddress.createUnresolved("localhost", 9092);

    @Test
    void testAddressWithValidEndpoint() {
        Map<ListenerName, InetSocketAddress> endpointMap = Utils.mkMap(
                Utils.mkEntry(testListener, testSocketAddress));
        Endpoints endpoints = Endpoints.fromInetSocketAddresses(endpointMap);

        Optional<InetSocketAddress> address = endpoints.address(testListener);

        assertEquals(Optional.of(testSocketAddress), address);
    }

    @Test
    void testAddressWithEmptyEndpoint() {
        Endpoints endpoints = Endpoints.empty();

        Optional<InetSocketAddress> address = endpoints.address(ListenerName.normalised("nonExistentListener"));

        assertEquals(Optional.empty(), address);
    }

    @Test
    void testVotersRecordEndpointsWithEndpoint() {
        Map<ListenerName, InetSocketAddress> endpointMap = Utils.mkMap(
                Utils.mkEntry(testListener, testSocketAddress));
        Endpoints endpoints = Endpoints.fromInetSocketAddresses(endpointMap);

        VotersRecord.Endpoint endpoint = endpoints.votersRecordEndpoints().next();

        assertEquals(testListener.value(), endpoint.name());
        assertEquals("localhost", endpoint.host());
        assertEquals(9092, endpoint.port());
    }

    @Test
    void testVotersRecordEndpointsWithEmptyEndpoint() {
        assertFalse(Endpoints.empty().votersRecordEndpoints().hasNext());
    }

    @Test
    void testSize() {
        Map<ListenerName, InetSocketAddress> endpointMap = Utils.mkMap(
                Utils.mkEntry(testListener, testSocketAddress));

        assertEquals(1, Endpoints.fromInetSocketAddresses(endpointMap).size());
        assertEquals(0, Endpoints.empty().size());
    }

    @Test
    void testIsEmptyWithEndpoint() {
        Map<ListenerName, InetSocketAddress> endpointMap = Utils.mkMap(
                Utils.mkEntry(testListener, testSocketAddress));

        assertFalse(Endpoints.fromInetSocketAddresses(endpointMap).isEmpty());
    }

    @Test
    void testEqualsAndHashCodeWithSameEndpoint() {
        Map<ListenerName, InetSocketAddress> endpointMap = Utils.mkMap(
                Utils.mkEntry(testListener, testSocketAddress));

        Endpoints endpoints = Endpoints.fromInetSocketAddresses(endpointMap);
        Endpoints sameEndpoints = Endpoints.fromInetSocketAddresses(endpointMap);

        assertEquals(endpoints, sameEndpoints);
        assertEquals(endpoints.hashCode(), sameEndpoints.hashCode());
    }

    @Test
    void testEqualsAndHashCodeWithDifferentEndpoints() {
        Map<ListenerName, InetSocketAddress> endpointMap = Utils.mkMap(
                Utils.mkEntry(testListener, testSocketAddress));
        Endpoints endpoints = Endpoints.fromInetSocketAddresses(endpointMap);

        Map<ListenerName, InetSocketAddress> anotherEndpointMap = Utils.mkMap(
                Utils.mkEntry(
                        ListenerName.normalised("another"),
                        InetSocketAddress.createUnresolved("localhost", 9093)));
        Endpoints differentEndpoints = Endpoints.fromInetSocketAddresses(anotherEndpointMap);

        assertNotEquals(endpoints, differentEndpoints);
        assertNotEquals(endpoints.hashCode(), differentEndpoints.hashCode());
    }

    @Test
    void testToBeginQuorumEpochRequestWithEndpoint() {
        Map<ListenerName, InetSocketAddress> endpointMap = Utils.mkMap(
                Utils.mkEntry(testListener, testSocketAddress));
        Endpoints endpoints = Endpoints.fromInetSocketAddresses(endpointMap);

        BeginQuorumEpochRequestData.LeaderEndpointCollection leaderEndpoints = endpoints.toBeginQuorumEpochRequest();

        assertEquals(1, leaderEndpoints.size());

        BeginQuorumEpochRequestData.LeaderEndpoint leaderEndpoint = leaderEndpoints.iterator().next();
        assertEquals(testListener.value(), leaderEndpoint.name());
        assertEquals("localhost", leaderEndpoint.host());
        assertEquals(9092, leaderEndpoint.port());
    }

    @Test
    void testToBeginQuorumEpochRequestWithEmptyEndpoint() {
        assertEquals(0, Endpoints.empty().toBeginQuorumEpochRequest().size());
    }

    @Test
    void testFromInetSocketAddressesWithEndpoint() {
        Map<ListenerName, InetSocketAddress> endpointMap = Utils.mkMap(
                Utils.mkEntry(testListener, testSocketAddress));

        assertEquals(1, Endpoints.fromInetSocketAddresses(endpointMap).size());
    }

    @Test
    void testFromVotersRecordEndpointsWithEndpoint() {
        Map<ListenerName, InetSocketAddress> endpointMap = Utils.mkMap(
                Utils.mkEntry(testListener, testSocketAddress));
        Endpoints endpoints = Endpoints.fromInetSocketAddresses(endpointMap);

        List<VotersRecord.Endpoint> votersEndpoints = new ArrayList<>();
        votersEndpoints.add(
                new VotersRecord.Endpoint()
                        .setName("listener")
                        .setHost("localhost")
                        .setPort(9092));

        Endpoints createdEndpoints = Endpoints.fromVotersRecordEndpoints(votersEndpoints);

        assertEquals(endpoints, createdEndpoints);
    }

    @Test
    void testFromVotersRecordEndpointsWithEmptyEndpoint() {
        List<VotersRecord.Endpoint> votersEndpoints = List.of();

        assertEquals(Endpoints.empty(), Endpoints.fromVotersRecordEndpoints(votersEndpoints));
    }

    @Test
    void testFromBeginQuorumEpochRequestWithEndpoint() {
        Map<ListenerName, InetSocketAddress> endpointMap = Utils.mkMap(
                Utils.mkEntry(testListener, testSocketAddress));
        Endpoints endpoints = Endpoints.fromInetSocketAddresses(endpointMap);

        BeginQuorumEpochRequestData.LeaderEndpointCollection leaderEndpoints = new BeginQuorumEpochRequestData.LeaderEndpointCollection();
        leaderEndpoints.add(
                new BeginQuorumEpochRequestData.LeaderEndpoint()
                        .setName("listener")
                        .setHost("localhost")
                        .setPort(9092));

        Endpoints createdEndpoints = Endpoints.fromBeginQuorumEpochRequest(leaderEndpoints);

        assertEquals(endpoints, createdEndpoints);
    }

    @Test
    void testFromBeginQuorumEpochRequestWithEmptyEndpoint() {
        BeginQuorumEpochRequestData.LeaderEndpointCollection leaderEndpoints = new BeginQuorumEpochRequestData.LeaderEndpointCollection();

        Endpoints createdEndpoints = Endpoints.fromBeginQuorumEpochRequest(leaderEndpoints);

        assertEquals(Endpoints.empty(), createdEndpoints);
    }

    @Test
    void testFromBeginQuorumEpochResponseWithEndpoint() {
        Map<ListenerName, InetSocketAddress> endpointMap = Utils.mkMap(
                Utils.mkEntry(testListener, testSocketAddress));
        Endpoints endpoints = Endpoints.fromInetSocketAddresses(endpointMap);

        BeginQuorumEpochResponseData.NodeEndpointCollection nodeEndpointCollection = new BeginQuorumEpochResponseData.NodeEndpointCollection();
        nodeEndpointCollection.add(
                new BeginQuorumEpochResponseData.NodeEndpoint()
                        .setNodeId(1)
                        .setHost("localhost")
                        .setPort(9092));

        Endpoints createdEndpoints = Endpoints.fromBeginQuorumEpochResponse(testListener, 1, nodeEndpointCollection);

        assertEquals(endpoints, createdEndpoints);
    }

    @Test
    void testFromBeginQuorumEpochResponseWithEmptyEndpoint() {
        BeginQuorumEpochResponseData.NodeEndpointCollection nodeEndpointCollection = new BeginQuorumEpochResponseData.NodeEndpointCollection();

        Endpoints createdEndpoints = Endpoints.fromBeginQuorumEpochResponse(ListenerName.normalised("nonExistListener"),
                1, nodeEndpointCollection);

        assertEquals(Endpoints.empty(), createdEndpoints);
    }

    @Test
    void testFromEndQuorumEpochRequestWithEndpoint() {
        Map<ListenerName, InetSocketAddress> endpointMap = Utils.mkMap(
                Utils.mkEntry(testListener, testSocketAddress));
        Endpoints endpoints = Endpoints.fromInetSocketAddresses(endpointMap);

        EndQuorumEpochRequestData.LeaderEndpointCollection leaderEndpoints = new EndQuorumEpochRequestData.LeaderEndpointCollection();
        leaderEndpoints.add(
                new EndQuorumEpochRequestData.LeaderEndpoint()
                        .setName("listener")
                        .setHost("localhost")
                        .setPort(9092));

        assertEquals(endpoints, Endpoints.fromEndQuorumEpochRequest(leaderEndpoints));
    }

    @Test
    void testFromEndQuorumEpochRequestWithEmptyEndpoint() {
        EndQuorumEpochRequestData.LeaderEndpointCollection leaderEndpoints = new EndQuorumEpochRequestData.LeaderEndpointCollection();

        assertEquals(Endpoints.empty(), Endpoints.fromEndQuorumEpochRequest(leaderEndpoints));
    }

    @Test
    void testFromEndQuorumEpochResponseWithEndpoint() {
        Map<ListenerName, InetSocketAddress> endpointMap = Utils.mkMap(
                Utils.mkEntry(testListener, testSocketAddress));
        Endpoints endpoints = Endpoints.fromInetSocketAddresses(endpointMap);

        EndQuorumEpochResponseData.NodeEndpointCollection nodeEndpointCollection = new EndQuorumEpochResponseData.NodeEndpointCollection();
        nodeEndpointCollection.add(
                new EndQuorumEpochResponseData.NodeEndpoint()
                        .setNodeId(1)
                        .setHost("localhost")
                        .setPort(9092));

        assertEquals(endpoints, Endpoints.fromEndQuorumEpochResponse(testListener, 1, nodeEndpointCollection));
    }

    @Test
    void testFromEndQuorumEpochResponseWithEmptyEndpoint() {
        EndQuorumEpochResponseData.NodeEndpointCollection nodeEndpointCollection = new EndQuorumEpochResponseData.NodeEndpointCollection();

        Endpoints createdEndpoints = Endpoints.fromEndQuorumEpochResponse(ListenerName.normalised("nonExistListener"),
                1, nodeEndpointCollection);

        assertEquals(Endpoints.empty(), createdEndpoints);
    }

    @Test
    void testFromVoteResponseWithEndpoint() {
        Map<ListenerName, InetSocketAddress> endpointMap = Utils.mkMap(
                Utils.mkEntry(testListener, testSocketAddress));
        Endpoints endpoints = Endpoints.fromInetSocketAddresses(endpointMap);

        VoteResponseData.NodeEndpointCollection nodeEndpointCollection = new VoteResponseData.NodeEndpointCollection();
        nodeEndpointCollection.add(
                new VoteResponseData.NodeEndpoint()
                        .setNodeId(1)
                        .setHost("localhost")
                        .setPort(9092));

        assertEquals(endpoints, Endpoints.fromVoteResponse(testListener, 1, nodeEndpointCollection));
    }

    @Test
    void testFromVoteResponseWithEmptyEndpoint() {
        VoteResponseData.NodeEndpointCollection nodeEndpointCollection = new VoteResponseData.NodeEndpointCollection();

        Endpoints createdEndpoints = Endpoints.fromVoteResponse(ListenerName.normalised("nonExistListener"), 1,
                nodeEndpointCollection);

        assertEquals(Endpoints.empty(), createdEndpoints);
    }

    @Test
    void testFromFetchResponseWithEndpoint() {
        Map<ListenerName, InetSocketAddress> endpointMap = Utils.mkMap(
                Utils.mkEntry(testListener, testSocketAddress));
        Endpoints endpoints = Endpoints.fromInetSocketAddresses(endpointMap);

        FetchResponseData.NodeEndpointCollection nodeEndpointCollection = new FetchResponseData.NodeEndpointCollection();
        nodeEndpointCollection.add(
                new FetchResponseData.NodeEndpoint()
                        .setNodeId(1)
                        .setHost("localhost")
                        .setPort(9092));

        assertEquals(endpoints, Endpoints.fromFetchResponse(testListener, 1, nodeEndpointCollection));
    }

    @Test
    void testFromFetchResponseWithEmptyEndpoint() {
        FetchResponseData.NodeEndpointCollection nodeEndpointCollection = new FetchResponseData.NodeEndpointCollection();

        assertEquals(Endpoints.empty(), Endpoints.fromFetchResponse(ListenerName.normalised("nonExistListener"), 1,
                nodeEndpointCollection));
    }

    @Test
    void testFromFetchSnapshotResponseWithEndpoint() {
        Map<ListenerName, InetSocketAddress> endpointMap = Utils.mkMap(
                Utils.mkEntry(testListener, testSocketAddress));
        Endpoints endpoints = Endpoints.fromInetSocketAddresses(endpointMap);

        FetchSnapshotResponseData.NodeEndpointCollection nodeEndpointCollection = new FetchSnapshotResponseData.NodeEndpointCollection();
        nodeEndpointCollection.add(
                new FetchSnapshotResponseData.NodeEndpoint()
                        .setNodeId(1)
                        .setHost("localhost")
                        .setPort(9092));

        assertEquals(endpoints, Endpoints.fromFetchSnapshotResponse(testListener, 1, nodeEndpointCollection));
    }

    @Test
    void testFromFetchSnapshotResponseWithEmptyEndpoint() {
        FetchSnapshotResponseData.NodeEndpointCollection nodeEndpointCollection = new FetchSnapshotResponseData.NodeEndpointCollection();

        Endpoints createdEndpoints = Endpoints.fromFetchSnapshotResponse(ListenerName.normalised("nonExistListener"), 1,
                nodeEndpointCollection);

        assertEquals(Endpoints.empty(), createdEndpoints);
    }
}
