/*
 * The contents of this file are subject to the terms of the Common Development and
 * Distribution License (the License). You may not use this file except in compliance with the
 * License.
 *
 * You can obtain a copy of the License at legal/CDDLv1.0.txt. See the License for the
 * specific language governing permission and limitations under the License.
 *
 * When distributing Covered Software, include this CDDL Header Notice in each file and include
 * the License file at legal/CDDLv1.0.txt. If applicable, add the following below the CDDL
 * Header, with the fields enclosed by brackets [] replaced by your own identifying
 * information: "Portions Copyright [year] [name of copyright owner]".
 *
 * Copyright 2014-2015 ForgeRock AS.
 * Portions Copyright 2026 3A Systems, LLC.
 */
package org.opends.server.replication.service;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.*;

import org.opends.server.DirectoryServerTestCase;
import org.opends.server.replication.common.DSInfo;
import org.opends.server.replication.common.RSInfo;
import org.opends.server.replication.common.ServerState;
import org.opends.server.replication.protocol.TopologyMsg;
import org.opends.server.replication.service.ReplicationBroker.RSEvaluations;
import org.opends.server.replication.service.ReplicationBroker.ReplicationServerInfo;
import org.opends.server.replication.service.ReplicationBroker.Topology;
import org.testng.Assert;
import org.testng.SkipException;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;

import static java.util.Collections.*;

import static org.assertj.core.api.Assertions.*;
import static org.opends.server.replication.service.ReplicationBroker.computeBestReplicationServer;
import static org.opends.server.util.CollectionUtils.*;
import static org.testng.Assert.*;

@SuppressWarnings("javadoc")
public class ReplicationBrokerTest extends DirectoryServerTestCase
{

  private static enum TopologyCtorToUse
  {
    BUILD_WITH_TOPOLOGY_MSG, BUILD_WITH_DS_RS_LISTS;
  }

  private final int CURRENT_RS_ID = 91;
  private final int MISSING_RS_ID = 93;
  private final int ANOTHER_RS_ID = 94;
  private final DSInfo CURRENT_DS = dsInfo(11, CURRENT_RS_ID);
  private final DSInfo OTHER_DS = dsInfo(12, CURRENT_RS_ID);
  private final DSInfo MISSING_DS = dsInfo(13, CURRENT_RS_ID);
  private final ReplicationServerInfo CURRENT_RS = rsInfo(CURRENT_RS_ID,
      CURRENT_DS.getDsId(), OTHER_DS.getDsId());
  private final ReplicationServerInfo MISSING_RS = rsInfo(MISSING_RS_ID,
      MISSING_DS.getDsId());
  private final ReplicationServerInfo ANOTHER_RS = rsInfo(ANOTHER_RS_ID);

  @SuppressWarnings("unchecked")
  private DSInfo dsInfo(int dsServerId, int rsServerId)
  {
    byte z = 0;
    return new DSInfo(dsServerId, null, rsServerId, 0, null,
        false, null, z, z, EMPTY_LIST, EMPTY_LIST, EMPTY_LIST, z);
  }

  private ReplicationServerInfo rsInfo(int rsServerId, Integer... dsIds)
  {
    byte z = 0;
    final RSInfo info = new RSInfo(rsServerId, rsServerId + ":1389", 0, z, 0);
    return new ReplicationServerInfo(info, newHashSet(dsIds));
  }

  private Map<Integer, ReplicationServerInfo> newMap(ReplicationServerInfo... infos)
  {
    if (infos.length == 0)
    {
      return Collections.emptyMap();
    }
    final Map<Integer, ReplicationServerInfo> map = new HashMap<>();
    for (ReplicationServerInfo info : infos)
    {
      map.put(info.getServerId(), info);
    }
    return map;
  }

  private void assertInvariants(final Topology topo)
  {
    assertThat(topo.replicaInfos).doesNotContainKey(CURRENT_DS.getDsId());
  }

  private ReplicationServerInfo assertContainsRSWithDSs(
      Map<Integer, ReplicationServerInfo> rsInfos,
      ReplicationServerInfo rsInfo, Integer... connectedDSs)
  {
    return assertContainsRSWithDSs(rsInfos, rsInfo, newHashSet(connectedDSs));
  }

  private ReplicationServerInfo assertContainsRSWithDSs(
      Map<Integer, ReplicationServerInfo> rsInfos,
      ReplicationServerInfo rsInfo, Set<Integer> connectedDSs)
  {
    final ReplicationServerInfo info = find(rsInfos, rsInfo.toRSInfo());
    assertNotNull(info);
    assertThat(info.getConnectedDSs()).containsAll(connectedDSs);
    return info;
  }

  private ReplicationServerInfo find(Map<Integer, ReplicationServerInfo> rsInfos, RSInfo rsInfo)
  {
    for (ReplicationServerInfo info : rsInfos.values())
    {
      if (info.getServerId() == rsInfo.getId())
      {
        return info;
      }
    }
    return null;
  }

  private Topology newTopology(TopologyCtorToUse toUse,
      Map<Integer, DSInfo> replicaInfos, List<RSInfo> rsInfos, int dsServerId, int rsServerId,
      Set<String> rsUrls, Map<Integer, ReplicationServerInfo> previousRSs)
  {
    if (TopologyCtorToUse.BUILD_WITH_TOPOLOGY_MSG == toUse)
    {
      final TopologyMsg topologyMsg = new TopologyMsg(replicaInfos.values(), rsInfos);
      return new Topology(topologyMsg, dsServerId, rsServerId, rsUrls, previousRSs);
    }
    else if (TopologyCtorToUse.BUILD_WITH_DS_RS_LISTS == toUse)
    {
      return new Topology(replicaInfos, rsInfos, dsServerId, rsServerId, rsUrls, previousRSs);
    }
    Assert.fail("Do not know which Topology constructor to use: " + toUse);
    return null;
  }

  private Map<Integer, DSInfo> newMap(DSInfo... dsInfos)
  {
    final Map<Integer, DSInfo> results = new HashMap<>();
    for (DSInfo dsInfo : dsInfos)
    {
      results.put(dsInfo.getDsId(), dsInfo);
    }
    return results;
  }

  @DataProvider
  public Object[][] topologyCtorProvider() {
    return new Object[][] { { TopologyCtorToUse.BUILD_WITH_TOPOLOGY_MSG },
      { TopologyCtorToUse.BUILD_WITH_DS_RS_LISTS } };
  }

  @Test(dataProvider = "topologyCtorProvider")
  @SuppressWarnings("unchecked")
  public void topologyShouldContainNothing(TopologyCtorToUse toUse)
      throws Exception
  {
    final Topology topo = newTopology(toUse,
        EMPTY_MAP, EMPTY_LIST,
        CURRENT_DS.getDsId(), CURRENT_RS.getServerId(), EMPTY_SET, EMPTY_MAP);
    assertInvariants(topo);
    assertThat(topo.rsInfos).isEmpty();
  }

  @Test
  @SuppressWarnings("unchecked")
  public void topologyShouldFilterOutCurrentDS()
  {
    final Topology topo = newTopology(TopologyCtorToUse.BUILD_WITH_TOPOLOGY_MSG,
        newMap(OTHER_DS, CURRENT_DS), EMPTY_LIST,
        CURRENT_DS.getDsId(), CURRENT_RS.getServerId(), EMPTY_SET, EMPTY_MAP);
    assertInvariants(topo);
    assertThat(topo.rsInfos).isEmpty();
  }

  @Test(dataProvider = "topologyCtorProvider")
  @SuppressWarnings("unchecked")
  public void topologyShouldContainRSWithoutOtherDS(TopologyCtorToUse toUse)
  {
    final Topology topo = newTopology(toUse,
        newMap(OTHER_DS), newArrayList(CURRENT_RS.toRSInfo()),
        CURRENT_DS.getDsId(), CURRENT_RS.getServerId(), EMPTY_SET, EMPTY_MAP);
    assertInvariants(topo);
    assertThat(topo.rsInfos).hasSize(1);
    assertContainsRSWithDSs(topo.rsInfos, CURRENT_RS, CURRENT_DS.getDsId());
  }

  @Test
  @SuppressWarnings("unchecked")
  public void topologyShouldContainRSWithAllDSs_buildWithTopologyMsg()
  {
    final Topology topo = newTopology(TopologyCtorToUse.BUILD_WITH_TOPOLOGY_MSG,
        newMap(CURRENT_DS, OTHER_DS), newArrayList(CURRENT_RS.toRSInfo()),
        CURRENT_DS.getDsId(), CURRENT_RS.getServerId(), EMPTY_SET, EMPTY_MAP);
    assertInvariants(topo);
    assertThat(topo.rsInfos).hasSize(1);
    assertContainsRSWithDSs(topo.rsInfos, CURRENT_RS, CURRENT_RS.getConnectedDSs());
  }

  @Test(dataProvider = "topologyCtorProvider")
  @SuppressWarnings("unchecked")
  public void topologyShouldStillContainRS(TopologyCtorToUse toUse) throws Exception
  {
    final Map<Integer, ReplicationServerInfo> previousRSs = newMap(CURRENT_RS);
    final Topology topo = newTopology(toUse,
        newMap(OTHER_DS), newArrayList(CURRENT_RS.toRSInfo()),
        CURRENT_DS.getDsId(), CURRENT_RS.getServerId(), EMPTY_SET, previousRSs);
    assertInvariants(topo);
    assertThat(topo.rsInfos).hasSize(1);
    assertContainsRSWithDSs(topo.rsInfos, CURRENT_RS, CURRENT_RS.getConnectedDSs());
  }

  @Test(dataProvider = "topologyCtorProvider")
  @SuppressWarnings("unchecked")
  public void topologyShouldStillContainRSWithNewlyProvidedDSs(TopologyCtorToUse toUse)
  {
    final ReplicationServerInfo CURRENT_RS_WITHOUT_DS = rsInfo(CURRENT_RS_ID);
    final Map<Integer, ReplicationServerInfo> previousRSs = newMap(CURRENT_RS_WITHOUT_DS);
    final Topology topo = newTopology(toUse,
        newMap(OTHER_DS), newArrayList(CURRENT_RS.toRSInfo()),
        CURRENT_DS.getDsId(), CURRENT_RS.getServerId(), EMPTY_SET, previousRSs);
    assertInvariants(topo);
    assertThat(topo.rsInfos).hasSize(1);
    assertContainsRSWithDSs(topo.rsInfos, CURRENT_RS, CURRENT_DS.getDsId(), OTHER_DS.getDsId());
  }

  @Test(dataProvider = "topologyCtorProvider")
  @SuppressWarnings("unchecked")
  public void topologyShouldHaveRemovedMissingRS(TopologyCtorToUse toUse)
  {
    final Map<Integer, ReplicationServerInfo> previousRSs = newMap(CURRENT_RS, MISSING_RS);
    final Topology topo = newTopology(toUse,
        newMap(OTHER_DS), newArrayList(CURRENT_RS.toRSInfo()),
        CURRENT_DS.getDsId(), CURRENT_RS.getServerId(), EMPTY_SET, previousRSs);
    assertInvariants(topo);
    assertThat(topo.rsInfos).hasSize(1);
    assertContainsRSWithDSs(topo.rsInfos, CURRENT_RS, CURRENT_RS.getConnectedDSs());
  }

  @Test
  @SuppressWarnings("unchecked")
  public void topologyShouldHaveStampedLocallyConfiguredRSs_buildWithDsRsLists()
  {
    final Set<String> locallyConfigured = newHashSet(CURRENT_RS.getServerURL());
    final Topology topo = newTopology(TopologyCtorToUse.BUILD_WITH_DS_RS_LISTS,
        newMap(OTHER_DS), newArrayList(CURRENT_RS.toRSInfo(), ANOTHER_RS.toRSInfo()),
        CURRENT_DS.getDsId(), CURRENT_RS.getServerId(), locallyConfigured, EMPTY_MAP);
    assertInvariants(topo);
    assertThat(topo.rsInfos).hasSize(2);
    ReplicationServerInfo currentRS =
        assertContainsRSWithDSs(topo.rsInfos, CURRENT_RS, CURRENT_RS.getConnectedDSs());
    ReplicationServerInfo anotherRS =
        assertContainsRSWithDSs(topo.rsInfos, ANOTHER_RS);
    assertThat(currentRS.isLocallyConfigured()).isTrue();
    assertThat(anotherRS.isLocallyConfigured()).isFalse();
  }

  /**
   * A topology names a replication server before the configuration of the domain lists it: the next
   * topology, built once the configuration lists it, must stamp it as locally configured.
   */
  @Test(dataProvider = "topologyCtorProvider")
  @SuppressWarnings("unchecked")
  public void topologyShouldStampRSAddedToTheConfigurationAfterThePreviousTopology(TopologyCtorToUse toUse)
  {
    final RSInfo announcedRS = new RSInfo(ANOTHER_RS_ID, "127.0.0.1:1389", 0, (byte) 0, 0);
    final Topology before = newTopology(TopologyCtorToUse.BUILD_WITH_TOPOLOGY_MSG,
        newMap(OTHER_DS), newArrayList(CURRENT_RS.toRSInfo(), announcedRS),
        CURRENT_DS.getDsId(), CURRENT_RS.getServerId(), newHashSet(CURRENT_RS.getServerURL()), EMPTY_MAP);
    assertThat(find(before.rsInfos, announcedRS).isLocallyConfigured()).isFalse();

    final RSInfo configuredRS = new RSInfo(ANOTHER_RS_ID, "localhost:1389", 0, (byte) 0, 0);
    final Topology after = newTopology(toUse,
        newMap(OTHER_DS), newArrayList(CURRENT_RS.toRSInfo(), configuredRS),
        CURRENT_DS.getDsId(), CURRENT_RS.getServerId(),
        newHashSet(CURRENT_RS.getServerURL(), "localhost:1389"), before.rsInfos);
    assertInvariants(after);
    final ReplicationServerInfo anotherRS = find(after.rsInfos, configuredRS);
    assertThat(anotherRS.isLocallyConfigured()).isTrue();
    assertThat(anotherRS.getServerURL()).isEqualTo("localhost:1389");
    assertThat(find(after.rsInfos, CURRENT_RS.toRSInfo()).isLocallyConfigured()).isTrue();
  }

  /**
   * A configured replication server announces in the topology messages a URL other than the
   * configured one (a port mapped by NAT or by Docker), and the broker hears of it first that way.
   * Once a phase-one handshake has reached it at the configured URL, it is locally configured, and
   * stays so when the next topology message names it with the announced URL again.
   */
  @Test
  @SuppressWarnings("unchecked")
  public void topologyShouldStampRSReachedAtAConfiguredURLItDoesNotAnnounce()
  {
    final Set<String> configured = newHashSet(CURRENT_RS.getServerURL(), "localhost:28989");
    final RSInfo announcedRS = new RSInfo(ANOTHER_RS_ID, "localhost:8989", 0, (byte) 0, 0);
    final Topology heard = newTopology(TopologyCtorToUse.BUILD_WITH_TOPOLOGY_MSG,
        newMap(OTHER_DS), newArrayList(CURRENT_RS.toRSInfo(), announcedRS),
        CURRENT_DS.getDsId(), CURRENT_RS.getServerId(), configured, EMPTY_MAP);
    assertThat(find(heard.rsInfos, announcedRS).isLocallyConfigured()).isFalse();

    final RSInfo reachedRS = new RSInfo(ANOTHER_RS_ID, "localhost:28989", 0, (byte) 0, 0);
    final Topology connected = newTopology(TopologyCtorToUse.BUILD_WITH_DS_RS_LISTS,
        newMap(OTHER_DS), newArrayList(CURRENT_RS.toRSInfo(), reachedRS),
        CURRENT_DS.getDsId(), ANOTHER_RS_ID, configured, heard.rsInfos);
    ReplicationServerInfo anotherRS = find(connected.rsInfos, reachedRS);
    assertThat(anotherRS.isLocallyConfigured()).isTrue();
    assertThat(anotherRS.getServerURL()).isEqualTo("localhost:28989");

    final Topology next = newTopology(TopologyCtorToUse.BUILD_WITH_TOPOLOGY_MSG,
        newMap(OTHER_DS), newArrayList(CURRENT_RS.toRSInfo(), announcedRS),
        CURRENT_DS.getDsId(), ANOTHER_RS_ID, configured, connected.rsInfos);
    anotherRS = find(next.rsInfos, announcedRS);
    assertThat(anotherRS.isLocallyConfigured()).isTrue();
    assertThat(anotherRS.getServerURL()).isEqualTo("localhost:28989");
  }

  /**
   * A replication server the previous topology stamped as locally configured stays so as long as the
   * configuration lists the URL it is known with, even when the topology names it with another URL (a port
   * mapped by NAT or by Docker) and its host name cannot be resolved at that moment: losing the flag
   * would make the broker leave the server it is connected to.
   * <p>
   * The case shows that the host is not resolved only where resolving it fails: a resolver which
   * answers for any name (the fake addresses of a proxy or VPN DNS) skips it.
   */
  @Test(dataProvider = "topologyCtorProvider")
  @SuppressWarnings("unchecked")
  public void topologyShouldKeepStampedAConfiguredRSWithoutResolvingItsHost(TopologyCtorToUse toUse)
  {
    try
    {
      final InetAddress[] answer = InetAddress.getAllByName("rs.invalid");
      throw new SkipException("The resolver answers for rs.invalid with " + Arrays.toString(answer)
          + ": this case could not tell a resolved host from one that is not");
    }
    catch (UnknownHostException expected)
    {
      // The premise of this case: resolving the host fails
    }
    final String unresolvableURL = "rs.invalid:1389";
    final RSInfo connectedRS = new RSInfo(ANOTHER_RS_ID, unresolvableURL, 0, (byte) 0, 0);
    final Map<Integer, ReplicationServerInfo> previousRSs =
        newMap(new ReplicationServerInfo(connectedRS, newHashSet(CURRENT_DS.getDsId())));

    final RSInfo announcedRS = new RSInfo(ANOTHER_RS_ID, "rs.invalid:8989", 0, (byte) 0, 0);
    final Topology topo = newTopology(toUse,
        newMap(OTHER_DS), newArrayList(announcedRS),
        CURRENT_DS.getDsId(), ANOTHER_RS_ID, newHashSet(unresolvableURL), previousRSs);
    assertInvariants(topo);
    final ReplicationServerInfo anotherRS = find(topo.rsInfos, announcedRS);
    assertThat(anotherRS.isLocallyConfigured()).isTrue();
    assertThat(anotherRS.getServerURL()).isEqualTo(unresolvableURL);
  }

  /**
   * A replication server is removed from the configuration of the domain: the next topology must no
   * longer stamp it as locally configured.
   */
  @Test(dataProvider = "topologyCtorProvider")
  @SuppressWarnings("unchecked")
  public void topologyShouldNotStampRSRemovedFromTheConfigurationAfterThePreviousTopology(TopologyCtorToUse toUse)
  {
    final RSInfo announcedRS = new RSInfo(ANOTHER_RS_ID, "localhost:1389", 0, (byte) 0, 0);
    final Topology before = newTopology(TopologyCtorToUse.BUILD_WITH_TOPOLOGY_MSG,
        newMap(OTHER_DS), newArrayList(CURRENT_RS.toRSInfo(), announcedRS),
        CURRENT_DS.getDsId(), CURRENT_RS.getServerId(),
        newHashSet(CURRENT_RS.getServerURL(), "localhost:1389"), EMPTY_MAP);
    assertThat(find(before.rsInfos, announcedRS).isLocallyConfigured()).isTrue();

    final Topology after = newTopology(toUse,
        newMap(OTHER_DS), newArrayList(CURRENT_RS.toRSInfo(), announcedRS),
        CURRENT_DS.getDsId(), CURRENT_RS.getServerId(), newHashSet(CURRENT_RS.getServerURL()), before.rsInfos);
    assertInvariants(after);
    assertThat(find(after.rsInfos, announcedRS).isLocallyConfigured()).isFalse();
  }

  /**
   * A topology names a replication server before the configuration of the domain lists it, with the URL
   * the configuration then lists it with: the next topology must stamp it as locally configured.
   */
  @Test(dataProvider = "topologyCtorProvider")
  @SuppressWarnings("unchecked")
  public void topologyShouldStampRSAddedToTheConfigurationWithTheURLItAnnounces(TopologyCtorToUse toUse)
  {
    final RSInfo announcedRS = new RSInfo(ANOTHER_RS_ID, "localhost:1389", 0, (byte) 0, 0);
    final Topology before = newTopology(TopologyCtorToUse.BUILD_WITH_TOPOLOGY_MSG,
        newMap(OTHER_DS), newArrayList(CURRENT_RS.toRSInfo(), announcedRS),
        CURRENT_DS.getDsId(), CURRENT_RS.getServerId(), newHashSet(CURRENT_RS.getServerURL()), EMPTY_MAP);
    assertThat(find(before.rsInfos, announcedRS).isLocallyConfigured()).isFalse();

    final Topology after = newTopology(toUse,
        newMap(OTHER_DS), newArrayList(CURRENT_RS.toRSInfo(), announcedRS),
        CURRENT_DS.getDsId(), CURRENT_RS.getServerId(),
        newHashSet(CURRENT_RS.getServerURL(), "localhost:1389"), before.rsInfos);
    assertInvariants(after);
    assertThat(find(after.rsInfos, announcedRS).isLocallyConfigured()).isTrue();
  }

  /**
   * The broker has reconnected to a replication server a topology named before the configuration of
   * the domain listed it, and the topology of the new session has the infos of the phase-one handshakes:
   * the periodic check of the best replication server must not reject it as a server which is not
   * locally configured.
   * <p>
   * The other server is not on this host, so the same-host filter keeps only this one, and the check
   * returns it before it reads which server the broker is connected to: the case pins the filter of the
   * locally configured servers only.
   */
  @Test
  @SuppressWarnings("unchecked")
  public void bestServerCheckShouldNotRejectTheRSAddedToTheConfiguration()
  {
    final RSInfo announcedRS = new RSInfo(ANOTHER_RS_ID, "localhost:1389", 0, (byte) 0, 0);
    final Topology before = newTopology(TopologyCtorToUse.BUILD_WITH_TOPOLOGY_MSG,
        newMap(OTHER_DS), newArrayList(CURRENT_RS.toRSInfo(), announcedRS),
        CURRENT_DS.getDsId(), CURRENT_RS.getServerId(), newHashSet(CURRENT_RS.getServerURL()), EMPTY_MAP);

    final Topology reconnected = newTopology(TopologyCtorToUse.BUILD_WITH_DS_RS_LISTS,
        newMap(OTHER_DS), newArrayList(CURRENT_RS.toRSInfo(), announcedRS),
        CURRENT_DS.getDsId(), ANOTHER_RS_ID,
        newHashSet(CURRENT_RS.getServerURL(), "localhost:1389"), before.rsInfos);
    final RSEvaluations evals = computeBestReplicationServer(false, ANOTHER_RS_ID, new ServerState(),
        reconnected.rsInfos, CURRENT_DS.getDsId(), (byte) 0, 0);
    assertThat(evals.getBestRS()).isNotNull();
    assertThat(evals.getBestRS().getServerId()).as(evals.toString()).isEqualTo(ANOTHER_RS_ID);
  }

  /**
   * The entry of the configuration which lists a replication server stamped as locally configured is
   * written another way, with an equivalent URL: the server must keep the flag, with the configured URL,
   * even when the restart of the session this asks for is refused (while a total update runs).
   */
  @Test(dataProvider = "topologyCtorProvider")
  @SuppressWarnings("unchecked")
  public void topologyShouldKeepStampedRSWhoseConfiguredURLIsWrittenAnotherWay(TopologyCtorToUse toUse)
  {
    final RSInfo announcedRS = new RSInfo(ANOTHER_RS_ID, "localhost:8989", 0, (byte) 0, 0);
    final Topology before = newTopology(TopologyCtorToUse.BUILD_WITH_TOPOLOGY_MSG,
        newMap(OTHER_DS), newArrayList(CURRENT_RS.toRSInfo(), announcedRS),
        CURRENT_DS.getDsId(), CURRENT_RS.getServerId(),
        newHashSet(CURRENT_RS.getServerURL(), "localhost:8989"), EMPTY_MAP);
    assertThat(find(before.rsInfos, announcedRS).isLocallyConfigured()).isTrue();

    final Topology after = newTopology(toUse,
        newMap(OTHER_DS), newArrayList(CURRENT_RS.toRSInfo(), announcedRS),
        CURRENT_DS.getDsId(), CURRENT_RS.getServerId(),
        newHashSet(CURRENT_RS.getServerURL(), "127.0.0.1:8989"), before.rsInfos);
    assertInvariants(after);
    final ReplicationServerInfo anotherRS = find(after.rsInfos, announcedRS);
    assertThat(anotherRS.isLocallyConfigured()).isTrue();
    assertThat(anotherRS.getServerURL()).isEqualTo("127.0.0.1:8989");
  }

  /**
   * A known replication server which is not locally configured comes with a URL which is only an
   * equivalent of a configured one: the topology message must not resolve it, so the server gets the flag
   * at the next connection only.
   */
  @Test(dataProvider = "topologyCtorProvider")
  @SuppressWarnings("unchecked")
  public void topologyShouldNotResolveTheURLAKnownRSComesWith(TopologyCtorToUse toUse)
  {
    final Set<String> configured = newHashSet(CURRENT_RS.getServerURL(), "localhost:28989");
    final RSInfo announcedRS = new RSInfo(ANOTHER_RS_ID, "localhost:8989", 0, (byte) 0, 0);
    final Topology before = newTopology(TopologyCtorToUse.BUILD_WITH_TOPOLOGY_MSG,
        newMap(OTHER_DS), newArrayList(CURRENT_RS.toRSInfo(), announcedRS),
        CURRENT_DS.getDsId(), CURRENT_RS.getServerId(), configured, EMPTY_MAP);
    assertThat(find(before.rsInfos, announcedRS).isLocallyConfigured()).isFalse();

    final RSInfo equivalentRS = new RSInfo(ANOTHER_RS_ID, "127.0.0.1:28989", 0, (byte) 0, 0);
    final Topology after = newTopology(toUse,
        newMap(OTHER_DS), newArrayList(CURRENT_RS.toRSInfo(), equivalentRS),
        CURRENT_DS.getDsId(), CURRENT_RS.getServerId(), configured, before.rsInfos);
    assertInvariants(after);
    assertThat(find(after.rsInfos, equivalentRS).isLocallyConfigured()).isFalse();
  }

  /**
   * A known replication server which is not locally configured gets an equivalent of its URL in the
   * configuration of the domain, written another way: the next topology must not resolve it either, so
   * the server gets the flag at the next connection only.
   */
  @Test(dataProvider = "topologyCtorProvider")
  @SuppressWarnings("unchecked")
  public void topologyShouldNotResolveTheURLAKnownRSIsKnownWith(TopologyCtorToUse toUse)
  {
    final RSInfo announcedRS = new RSInfo(ANOTHER_RS_ID, "127.0.0.1:28989", 0, (byte) 0, 0);
    final Topology before = newTopology(TopologyCtorToUse.BUILD_WITH_TOPOLOGY_MSG,
        newMap(OTHER_DS), newArrayList(CURRENT_RS.toRSInfo(), announcedRS),
        CURRENT_DS.getDsId(), CURRENT_RS.getServerId(), newHashSet(CURRENT_RS.getServerURL()), EMPTY_MAP);
    assertThat(find(before.rsInfos, announcedRS).isLocallyConfigured()).isFalse();

    final Topology after = newTopology(toUse,
        newMap(OTHER_DS), newArrayList(CURRENT_RS.toRSInfo(), announcedRS),
        CURRENT_DS.getDsId(), CURRENT_RS.getServerId(),
        newHashSet(CURRENT_RS.getServerURL(), "localhost:28989"), before.rsInfos);
    assertInvariants(after);
    assertThat(find(after.rsInfos, announcedRS).isLocallyConfigured()).isFalse();
  }

  /**
   * A known replication server comes with another URL in the next topology (a mapped port) while the
   * configuration of the domain now lists the URL it is known with: it is locally configured, without
   * resolving any host name.
   */
  @Test(dataProvider = "topologyCtorProvider")
  @SuppressWarnings("unchecked")
  public void topologyShouldStampRSAddedToTheConfigurationWithTheURLItIsKnownWith(TopologyCtorToUse toUse)
  {
    final RSInfo knownRS = new RSInfo(ANOTHER_RS_ID, "127.0.0.2:28989", 0, (byte) 0, 0);
    final Topology before = newTopology(TopologyCtorToUse.BUILD_WITH_TOPOLOGY_MSG,
        newMap(OTHER_DS), newArrayList(CURRENT_RS.toRSInfo(), knownRS),
        CURRENT_DS.getDsId(), CURRENT_RS.getServerId(), newHashSet(CURRENT_RS.getServerURL()), EMPTY_MAP);
    assertThat(find(before.rsInfos, knownRS).isLocallyConfigured()).isFalse();

    final RSInfo mappedRS = new RSInfo(ANOTHER_RS_ID, "127.0.0.3:28989", 0, (byte) 0, 0);
    final Topology after = newTopology(toUse,
        newMap(OTHER_DS), newArrayList(CURRENT_RS.toRSInfo(), mappedRS),
        CURRENT_DS.getDsId(), CURRENT_RS.getServerId(),
        newHashSet(CURRENT_RS.getServerURL(), "127.0.0.2:28989"), before.rsInfos);
    assertInvariants(after);
    final ReplicationServerInfo rs = find(after.rsInfos, mappedRS);
    assertThat(rs.isLocallyConfigured()).isTrue();
    assertThat(rs.getServerURL()).isEqualTo("127.0.0.2:28989");
  }

  /**
   * A known replication server with no URL (protocol version below 4) is not locally configured, and the
   * check must not ask the configured URLs for a null one: they are a sorted set, as in the configuration,
   * whose contains(null) throws.
   */
  @Test(dataProvider = "topologyCtorProvider")
  @SuppressWarnings("unchecked")
  public void topologyShouldNotStampKnownRSWithoutURL(TopologyCtorToUse toUse)
  {
    // sorted, as the configuration is: contains(null) throws
    final Set<String> configured = new TreeSet<>(Arrays.asList(CURRENT_RS.getServerURL(), "localhost:8989"));
    final RSInfo noUrlRS = new RSInfo(ANOTHER_RS_ID, null, 0, (byte) 0, 0);
    final Topology before = newTopology(TopologyCtorToUse.BUILD_WITH_TOPOLOGY_MSG,
        newMap(OTHER_DS), newArrayList(CURRENT_RS.toRSInfo(), noUrlRS),
        CURRENT_DS.getDsId(), CURRENT_RS.getServerId(), configured, EMPTY_MAP);
    final Topology after = newTopology(toUse,
        newMap(OTHER_DS), newArrayList(CURRENT_RS.toRSInfo(), noUrlRS),
        CURRENT_DS.getDsId(), CURRENT_RS.getServerId(), configured, before.rsInfos);
    assertThat(find(after.rsInfos, noUrlRS).isLocallyConfigured()).isFalse();
  }

  /**
   * A locally configured replication server, known with the configured URL a phase-one handshake reached
   * it at, comes with another URL (a port mapped by NAT or by Docker), and the entry of the configuration
   * is written another way, as an equivalent of the URL it comes with: the server must keep the flag,
   * with the configured URL.
   */
  @Test(dataProvider = "topologyCtorProvider")
  @SuppressWarnings("unchecked")
  public void topologyShouldKeepStampedRSWhoseConfiguredURLIsWrittenAsTheURLItComesWith(TopologyCtorToUse toUse)
  {
    final RSInfo reachedRS = new RSInfo(ANOTHER_RS_ID, "localhost:28989", 0, (byte) 0, 0);
    final Topology connected = newTopology(TopologyCtorToUse.BUILD_WITH_DS_RS_LISTS,
        newMap(OTHER_DS), newArrayList(CURRENT_RS.toRSInfo(), reachedRS),
        CURRENT_DS.getDsId(), ANOTHER_RS_ID, newHashSet(CURRENT_RS.getServerURL(), "localhost:28989"), EMPTY_MAP);
    assertThat(find(connected.rsInfos, reachedRS).isLocallyConfigured()).isTrue();

    final RSInfo announcedRS = new RSInfo(ANOTHER_RS_ID, "localhost:8989", 0, (byte) 0, 0);
    final Topology after = newTopology(toUse,
        newMap(OTHER_DS), newArrayList(CURRENT_RS.toRSInfo(), announcedRS),
        CURRENT_DS.getDsId(), ANOTHER_RS_ID,
        newHashSet(CURRENT_RS.getServerURL(), "127.0.0.1:8989"), connected.rsInfos);
    assertInvariants(after);
    final ReplicationServerInfo rs = find(after.rsInfos, announcedRS);
    assertThat(rs.isLocallyConfigured()).isTrue();
    assertThat(rs.getServerURL()).isEqualTo("127.0.0.1:8989");
  }

  /**
   * A locally configured replication server, known with the configured URL a phase-one handshake reached
   * it at, comes with another URL, and the entry of the configuration is replaced with a URL equivalent to
   * neither: the server loses the flag, and keeps the URL it is known with.
   */
  @Test(dataProvider = "topologyCtorProvider")
  @SuppressWarnings("unchecked")
  public void topologyShouldNotStampRSWhoseConfiguredURLNamesNeitherOfItsURLs(TopologyCtorToUse toUse)
  {
    final RSInfo reachedRS = new RSInfo(ANOTHER_RS_ID, "localhost:28989", 0, (byte) 0, 0);
    final Topology connected = newTopology(TopologyCtorToUse.BUILD_WITH_DS_RS_LISTS,
        newMap(OTHER_DS), newArrayList(CURRENT_RS.toRSInfo(), reachedRS),
        CURRENT_DS.getDsId(), ANOTHER_RS_ID, newHashSet(CURRENT_RS.getServerURL(), "localhost:28989"), EMPTY_MAP);
    assertThat(find(connected.rsInfos, reachedRS).isLocallyConfigured()).isTrue();

    final RSInfo announcedRS = new RSInfo(ANOTHER_RS_ID, "localhost:8989", 0, (byte) 0, 0);
    final Topology after = newTopology(toUse,
        newMap(OTHER_DS), newArrayList(CURRENT_RS.toRSInfo(), announcedRS),
        CURRENT_DS.getDsId(), ANOTHER_RS_ID,
        newHashSet(CURRENT_RS.getServerURL(), "127.0.0.1:18989"), connected.rsInfos);
    assertInvariants(after);
    final ReplicationServerInfo rs = find(after.rsInfos, announcedRS);
    assertThat(rs.isLocallyConfigured()).isFalse();
    assertThat(rs.getServerURL()).isEqualTo("localhost:28989");
  }
}