package com.atlassian.bitbucket.mesh.management;

import com.atlassian.bitbucket.mesh.app.BuildInfo;
import com.atlassian.bitbucket.mesh.auth.AuthenticationManager;
import com.atlassian.bitbucket.mesh.auth.InvalidPublicKeyException;
import com.atlassian.bitbucket.mesh.auth.UpdatePublicKeysParameters;
import com.atlassian.bitbucket.mesh.config.ApplicationConfig;
import com.atlassian.bitbucket.mesh.config.ConfigurationUpdatedEvent;
import com.atlassian.bitbucket.mesh.config.GitConfig;
import com.atlassian.bitbucket.mesh.config.NodeConfig;
import com.atlassian.bitbucket.mesh.event.EventPublisher;
import com.atlassian.bitbucket.mesh.git.binary.GitBinary;
import com.atlassian.bitbucket.mesh.git.builder.GitProcess;
import com.atlassian.bitbucket.mesh.git.builder.GitProcessBuilder;
import com.atlassian.bitbucket.mesh.git.builder.GitProcessBuilderFactory;
import com.atlassian.bitbucket.mesh.git.builder.version.GitVersionBuilder;
import com.atlassian.bitbucket.mesh.git.builder.version.VersionStdioHandler;
import com.atlassian.bitbucket.mesh.io.MoreFiles;
import com.atlassian.bitbucket.mesh.junit.LogListener;
import com.atlassian.bitbucket.mesh.junit.TemporaryDirectory;
import com.atlassian.bitbucket.mesh.layout.StorageLayout;
import com.atlassian.bitbucket.mesh.network.MeshManager;
import com.atlassian.bitbucket.mesh.network.MeshNode;
import com.atlassian.bitbucket.mesh.network.Partition;
import com.atlassian.bitbucket.mesh.network.UpdateTopologyParameters;
import com.atlassian.bitbucket.mesh.repository.MarkInconsistentParameters;
import com.atlassian.bitbucket.mesh.repository.ReplicaStateRegistry;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcTicketInfo;
import com.atlassian.bitbucket.mesh.throttle.ThrottleService;
import com.atlassian.bitbucket.mesh.throttle.ThrottledResource;
import com.atlassian.bitbucket.mesh.throttle.TicketSummary;
import com.atlassian.bitbucket.mesh.util.Version;
import com.google.common.collect.ImmutableMap;
import org.apache.commons.lang3.mutable.MutableInt;
import org.apache.commons.lang3.mutable.MutableObject;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.*;
import org.mockito.junit.MockitoJUnitRunner;
import org.springframework.boot.context.properties.source.ConfigurationPropertySources;
import org.springframework.boot.env.PropertiesPropertySourceLoader;
import org.springframework.boot.env.RandomValuePropertySource;
import org.springframework.core.env.ConfigurableEnvironment;
import org.springframework.core.env.MutablePropertySources;
import org.springframework.core.env.StandardEnvironment;
import org.springframework.core.io.ClassPathResource;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.file.Files;
import java.nio.file.Path;
import java.security.PublicKey;
import java.time.Duration;
import java.time.Instant;
import java.util.*;
import java.util.zip.ZipEntry;
import java.util.zip.ZipFile;

import static com.atlassian.bitbucket.mesh.layout.StorageLayout.MESH_NODE_PROPERTIES;
import static com.atlassian.bitbucket.mesh.layout.StorageLayout.MESH_PROPERTIES;
import static com.atlassian.bitbucket.mesh.management.SupportUtils.APPLICATION_PROPERTIES_JSON;
import static com.atlassian.bitbucket.mesh.rpc.ManagedConfigurationProperties.NODE_ID;
import static com.atlassian.bitbucket.mesh.rpc.ManagedConfigurationProperties.NODE_NAME;
import static java.nio.charset.StandardCharsets.UTF_8;
import static java.util.Objects.requireNonNull;
import static java.util.stream.Collectors.toSet;
import static org.apache.commons.lang3.StringUtils.isBlank;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.hasEntry;
import static org.junit.Assert.*;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.*;

@RunWith(MockitoJUnitRunner.class)
public class DefaultNodeManagerTest {

    private static final String MIRROR_HOSTING_PROPERTY = "throttle.resource.mirror-hosting";
    private static final String PORT_PROPERTY = "grpc.server.port";
    private static final UpdateTopologyParameters TOPOLOGY = new UpdateTopologyParameters.Builder()
            .nodes(new MeshNode.Builder("local")
                .local(true)
                .name("local-node")
                .rpcUrl("http://localhost:7777")
                .build())
            .build();
    @Rule
    public final LogListener logListener = new LogListener(DefaultNodeManager.class);
    @Rule
    public final TemporaryDirectory temporaryDirectory = new TemporaryDirectory();

    @Spy
    private final PropertiesPropertySourceLoader propertySourceLoader = new PropertiesPropertySourceLoader();
    @Mock
    private BuildInfo buildInfo;
    private GitProcessBuilderFactory builderFactory;
    @Mock
    private ApplicationConfig config;
    private Path configFile;
    private ConfigurableEnvironment environment;
    @Mock
    private EventPublisher eventPublisher;
    @Captor
    private ArgumentCaptor<MarkInconsistentParameters> inconsistentReplicasCaptor;
    private Path logDir;
    @Mock
    private MeshManager meshManager;
    @Mock
    private NodeConfig nodeConfig;
    private DefaultNodeManager nodeManager;
    @Mock
    private ReplicaStateRegistry replicaStateRegistry;
    @Mock
    private StorageLayout storageLayout;
    @Mock
    private SupportInfoRegistry supportInfoRegistry;
    private Path tempDir;
    private ThrottleService throttleService;

    @Before
    public void setup() throws IOException {
        configFile = temporaryDirectory.getRoot().resolve(MESH_PROPERTIES);
        logDir = temporaryDirectory.newFolder("log");
        tempDir = temporaryDirectory.newFolder("temp");

        when(storageLayout.getConfigFile()).thenReturn(configFile);
        when(storageLayout.getLogDir()).thenReturn(logDir);
        when(storageLayout.getTempDir()).thenReturn(tempDir);

        //Prepare the environment property sources similar to how Spring Boot creates it
        environment = new StandardEnvironment();
        MutablePropertySources propertySources = environment.getPropertySources();
        propertySources.addLast(new RandomValuePropertySource("random"));
        propertySourceLoader.load("applicationConfig: [classpath:/application-public.properties]",
                new ClassPathResource("/application-public.properties")).forEach(propertySources::addLast);
        propertySourceLoader.load("applicationConfig: [classpath:/application-private.properties]",
                new ClassPathResource("/application-private.properties")).forEach(propertySources::addLast);
        ConfigurationPropertySources.attach(environment);
        environment = spy(environment); //Wrap the environment for spying (last so we don't record the setup calls)

        when(config.getNode()).thenReturn(nodeConfig);
        when(nodeConfig.getId()).thenReturn(NodeConfig.UNCONFIGURED_NODE_ID);

        throttleService = mock(ThrottleService.class);
        builderFactory = mock(GitProcessBuilderFactory.class);

        when(throttleService.getSummaries()).thenReturn(new TicketSummary[]{});

        nodeManager = create(null);
    }

    @Test
    public void testAwaitConfigurationPropertiesAndTopology() throws InterruptedException {
        assertFalse("Unlatched without configuration", nodeManager.awaitConfiguration(Duration.ofMillis(1L)));

        nodeManager.updateConfiguration(new UpdateConfigurationParameters.Builder()
                .fromControlPlane(true)
                .properties(ImmutableMap.of())
                .topology(TOPOLOGY)
                .build());

        assertTrue("Applying configuration did not unlock", nodeManager.awaitConfiguration(Duration.ofMillis(1L)));
    }

    @Test
    public void testAwaitConfigurationPropertiesThenTopology() throws InterruptedException {
        assertFalse("Unlatched without configuration", nodeManager.awaitConfiguration(Duration.ofMillis(1L)));

        nodeManager.updateConfiguration(new UpdateConfigurationParameters.Builder()
                .fromControlPlane(true)
                .properties(ImmutableMap.of())
                .build());

        assertFalse("Applying configuration without topology unlocked",
                nodeManager.awaitConfiguration(Duration.ofMillis(1L)));

        nodeManager.updateConfiguration(new UpdateConfigurationParameters.Builder()
                .fromControlPlane(true)
                .topology(TOPOLOGY)
                .build());

        assertTrue("Applying configuration did not unlock", nodeManager.awaitConfiguration(Duration.ofMillis(1L)));
    }

    @Test
    public void testAwaitConfigurationTopologyThenProperties() throws InterruptedException {
        assertFalse("Unlatched without configuration", nodeManager.awaitConfiguration(Duration.ofMillis(1L)));

        nodeManager.updateConfiguration(new UpdateConfigurationParameters.Builder()
                .fromControlPlane(true)
                .topology(TOPOLOGY)
                .build());

        assertFalse("Applying configuration without config properties unlocked",
                nodeManager.awaitConfiguration(Duration.ofMillis(1L)));

        nodeManager.updateConfiguration(new UpdateConfigurationParameters.Builder()
                .fromControlPlane(true)
                .properties(ImmutableMap.of())
                .build());

        assertTrue("Applying configuration did not unlock", nodeManager.awaitConfiguration(Duration.ofMillis(1L)));
    }

    @Test
    public void testCreateSupportZip() throws Exception {
        Path meshNodeProperties = Files.createFile(tempDir.resolve(MESH_NODE_PROPERTIES));
        MoreFiles.write(meshNodeProperties, "authentication.token = token");
        when(storageLayout.getNodeConfigFile()).thenReturn(meshNodeProperties);

        Path log1 = Files.createTempFile(logDir, "mesh-log-1-", ".log");
        Path log2 = Files.createTempFile(logDir, "mesh-log-2-", ".log");

        Path supportZipPath = nodeManager.createSupportZip(new CreateSupportZipParameters(true, true));

        verify(supportInfoRegistry).writeSupportInfo(any());
        ZipFile supportZip = new ZipFile(supportZipPath.toFile());
        Set<String> zipEntries = supportZip.stream()
                .map(ZipEntry::getName)
                .collect(toSet());

        assertNotNull(zipEntries);
        assertEquals(4, zipEntries.size());
        assertTrue(zipEntries.contains(APPLICATION_PROPERTIES_JSON));
        assertTrue(zipEntries.contains("config/" + meshNodeProperties.getFileName()));
        assertTrue(zipEntries.contains("log/" + log1.getFileName()));
        assertTrue(zipEntries.contains("log/" + log2.getFileName()));
    }

    @Test
    public void testCreateSupportZipWithNoConfigRequest() throws Exception {
        Path log1 = Files.createTempFile(logDir, "mesh-log-1-", ".log");
        Path log2 = Files.createTempFile(logDir, "mesh-log-2-", ".log");

        Path supportZipPath = nodeManager.createSupportZip(new CreateSupportZipParameters(false, true));

        verify(supportInfoRegistry).writeSupportInfo(any());
        ZipFile supportZip = new ZipFile(supportZipPath.toFile());
        Set<String> zipEntries = supportZip.stream()
                .map(ZipEntry::getName)
                .collect(toSet());

        assertNotNull(zipEntries);
        assertEquals(3, zipEntries.size());
        assertTrue(zipEntries.contains(APPLICATION_PROPERTIES_JSON));
        assertTrue(zipEntries.contains("log/" + log1.getFileName()));
        assertTrue(zipEntries.contains("log/" + log2.getFileName()));
    }

    @Test
    public void testGetConfigurationWithLoggingPropertiesReturnsEntry() {
        nodeManager.updateConfiguration(new UpdateConfigurationParameters.Builder()
                .properties(ImmutableMap.of("logging.logger.com.atlassian.bitbucket", "DEBUG"))
                .build());
        Map<String, String> result = nodeManager.getConfigurationProperties();

        assertThat(result, hasEntry("logging.logger.com.atlassian.bitbucket", "DEBUG"));
    }

    @Test
    public void testGetNodeId() {
        when(nodeConfig.getId()).thenReturn("1234");
        assertEquals("1234", nodeManager.getId());
    }

    @Test
    public void testGetNodeIdUnconfigured() {
        assertEquals(NodeConfig.UNCONFIGURED_NODE_ID, nodeManager.getId());
    }

    @Test
    public void testGetNodeInfo() {
        when(buildInfo.getBuildVersion()).thenReturn("test-build-version");

        Map<String, String> result = nodeManager.getNodeInfo();
        assertNotNull(result);
        assertEquals("test-build-version", result.get("buildVersion"));
    }

    @Test
    public void testGetSupportInfo() {
        GitProcessBuilder builder = mock(GitProcessBuilder.class);
        when(builderFactory.builder()).thenReturn(builder);

        Version current = new Version(2, 31, 1);
        Version startup = new Version(2, 31);
        mockVersionInfo(builder, current, startup);

        Instant latestRejection = Instant.now();
        Instant oldestQueued = Instant.now();
        MutableInt idx = new MutableInt(0);
        when(throttleService.getSummaries())
                .thenReturn(Arrays.stream(ThrottledResource.values())
                        .map(r -> idx.intValue() == 0
                                ? new TicketSummary(r, idx.intValue(), latestRejection, oldestQueued, idx.intValue(),
                                idx.intValue(), idx.getAndIncrement())
                                : new TicketSummary(r, idx.intValue(), null, null, idx.intValue(),
                                idx.intValue(), idx.getAndIncrement()))
                        .toArray(TicketSummary[]::new));

        SupportInfo info = nodeManager.getSupportInfo();

        assertEquals(current.toString(), info.toSupportInfoResponse().getGitInfo().getCurrentVersion());
        assertEquals(startup.toString(), info.toSupportInfoResponse().getGitInfo().getStartupVersion());

        List<RpcTicketInfo> ticketInfos = info.toSupportInfoResponse().getTicketInfoList();
        for (int i = 0; i < ticketInfos.size(); i++) {
            RpcTicketInfo ticketInfo = ticketInfos.get(i);

            if (i == 0) {
                assertEquals(latestRejection.toEpochMilli(), ticketInfo.getLastRejection());
                assertEquals(oldestQueued.toEpochMilli(), ticketInfo.getOldestQueuedRequest());
            } else {
                assertFalse(ticketInfo.hasLastRejection());
                assertFalse(ticketInfo.hasOldestQueuedRequest());
            }

            assertEquals(ThrottledResource.values()[i].getId().toLowerCase(Locale.ROOT), ticketInfo.getResource());
            assertEquals(i, ticketInfo.getAvailable());
            assertEquals(i, ticketInfo.getQueuedRequests());
            assertEquals(i, ticketInfo.getTotal());
            assertEquals(i, ticketInfo.getUsed());
        }
    }

    @Test
    public void testIsConfigured() {
        assertFalse("The node should not be configured yet", nodeManager.isConfigured());

        nodeManager.updateConfiguration(new UpdateConfigurationParameters.Builder()
                .fromControlPlane(true)
                .properties(ImmutableMap.of())
                .topology(TOPOLOGY)
                .build());
        assertTrue("Applying configuration did not mark configured", nodeManager.isConfigured());
    }

    @Test
    public void testProvision() {
        PublicKey control = mock(PublicKey.class, "control");
        PublicKey node = mock(PublicKey.class, "node");

        AuthenticationManager authenticationManager = mock(AuthenticationManager.class);
        when(authenticationManager.provision(same(control))).thenReturn(node);

        PublicKey actual = create(authenticationManager)
                .provision(new ProvisionParameters.Builder(control).build());
        assertSame(node, actual);
    }

    @Test
    public void testProvisionReturnsNullWhenAuthenticationIsDisabled() {
        PublicKey control = mock(PublicKey.class);

        PublicKey node = nodeManager.provision(new ProvisionParameters.Builder(control).build());
        assertNull(node);

        logListener.assertInfo("authentication is disabled");
    }

    @Test
    public void testUpdateConfigurationDoesNotUnlatchOnFailure() throws IOException {
        //It's not straightforward to try and make writing the properties file fail, so instead we train the loader
        //to throw when the code tries to update the PropertySources
        doThrow(IOException.class).when(propertySourceLoader).load(anyString(), any());

        ConfigurationFailedException thrown = assertThrows(ConfigurationFailedException.class,
                () -> nodeManager.updateConfiguration(new UpdateConfigurationParameters.Builder()
                        .fromControlPlane(true)
                        .property(PORT_PROPERTY, "8888")
                        .topology(TOPOLOGY)
                        .build()));
        assertEquals("Configuration was not applied", thrown.getMessage());
        assertFalse(nodeManager.isConfigured());
    }

    @Test
    public void testUpdateConfigurationDoesNotUnlatchOnInternalConfigUpdates() throws InterruptedException {
        nodeManager.updateConfiguration(new UpdateConfigurationParameters.Builder()
                .fromControlPlane(false)
                .property(PORT_PROPERTY, "8888")
                .topology(TOPOLOGY)
                .build());
        assertFalse(nodeManager.isConfigured());
        assertFalse(nodeManager.awaitConfiguration(Duration.ZERO));
    }

    @Test
    public void testUpdateConfigurationDoesNotUnlatchTopologyOnFailure() {
        doThrow(RuntimeException.class).when(meshManager).updateTopology(any());

        ConfigurationFailedException thrown = assertThrows(ConfigurationFailedException.class,
                () -> nodeManager.updateConfiguration(new UpdateConfigurationParameters.Builder()
                        .fromControlPlane(true)
                        .property(PORT_PROPERTY, "8888") //Ensure the configLatch is released
                        .topology(TOPOLOGY)
                        .build()));
        assertEquals("Topology was not applied", thrown.getMessage());
        assertFalse(nodeManager.isConfigured()); //Configuration is unlatched, but topology isn't
    }

    @Test
    public void testUpdateConfigurationFirstTimeDefaultsOnly() throws IOException {
        //This tests whether the updateConfiguration updates the config correctly when no config file is present, and
        //the node configuration is updated with values that all match the application defaults. This should result in
        //an empty config file.
        String defaultPortValue = requireNonNull(environment.getProperty(PORT_PROPERTY), "port is not defined");
        //Also verify that a property with placeholders is recognized as being equal to the default value
        String mirrorHostingValue = "2*${scaling.concurrency}";
        //Catch any changes in the default configuration
        assertEquals(environment.resolvePlaceholders(mirrorHostingValue), environment.getProperty(MIRROR_HOSTING_PROPERTY));
        nodeManager.updateConfiguration(new UpdateConfigurationParameters.Builder()
                .fromControlPlane(true)
                .property(MIRROR_HOSTING_PROPERTY, mirrorHostingValue)
                .property(PORT_PROPERTY, defaultPortValue)
                .build());
        assertTrue(Files.isRegularFile(configFile));
        assertEquals(DefaultNodeManager.CONFIG_BANNER, MoreFiles.toString(configFile));
        assertEquals(defaultPortValue, environment.getProperty(PORT_PROPERTY));
        assertEquals(environment.resolvePlaceholders(mirrorHostingValue), environment.getProperty(MIRROR_HOSTING_PROPERTY));

        verify(config, never()).reload();
        verify(eventPublisher, never()).publish(any(ConfigurationUpdatedEvent.class));
        verify(replicaStateRegistry).markInconsistent(inconsistentReplicasCaptor.capture());

        MarkInconsistentParameters inconsistentParameters = inconsistentReplicasCaptor.getValue();
        assertEquals(Collections.emptyMap(), inconsistentParameters.getInconsistentReplicas());
        assertTrue(inconsistentParameters.isFromControlPlane());
        assertTrue(inconsistentParameters.isComplete());
    }

    @Test
    public void testUpdateConfigurationFirstTimeNonDefaults() throws IOException {
        Partition partition2 = mock(Partition.class);
        Partition partition3 = mock(Partition.class);
        when(meshManager.getPartition("0002")).thenReturn(partition2);
        when(meshManager.getPartition("0003")).thenReturn(partition3);
        //This tests whether the updateConfiguration updates the config correctly when no config file is present, and
        //the node configuration is updated with values that deviate from match the application defaults.
        //This should result in a non-empty config file.
        nodeManager.updateConfiguration(new UpdateConfigurationParameters.Builder()
                .property(PORT_PROPERTY, "98765")
                .inconsistentReplica("p/0002/h/01234567890123456789/r/repository-1", 7L)
                .inconsistentReplica("p/0003/h/12345678901234567890/r/repository-2", 9L)
                .fromControlPlane(true)
                .build());
        assertTrue(Files.isRegularFile(configFile));
        assertFalse(isBlank(MoreFiles.toString(configFile)));
        assertEquals("98765", environment.getProperty(PORT_PROPERTY));

        verify(config).reload();
        verify(eventPublisher).publish(any(ConfigurationUpdatedEvent.class));
        verify(replicaStateRegistry).markInconsistent(inconsistentReplicasCaptor.capture());

        MarkInconsistentParameters inconsistentParameters = inconsistentReplicasCaptor.getValue();
        assertEquals(ImmutableMap.of("p/0002/h/01234567890123456789/r/repository-1", 7L,
                        "p/0003/h/12345678901234567890/r/repository-2", 9L),
                inconsistentParameters.getInconsistentReplicas());
        assertTrue(inconsistentParameters.isFromControlPlane());
        assertTrue(inconsistentParameters.isComplete());
    }

    @Test
    public void testUpdateConfigurationFiltersOutInconsistentReplicasNotHostedOnThisNode() throws IOException {
        Partition partition2 = mock(Partition.class);
        Partition partition3 = mock(Partition.class);
        when(meshManager.getPartition("0002")).thenReturn(partition2);
        when(meshManager.getPartition("0003")).thenReturn(partition3);
        //This includes an inconsistent replica on partition 0004, which is not hosted on this node according to
        //MeshManager, and should be filtered out.
        nodeManager.updateConfiguration(new UpdateConfigurationParameters.Builder()
                .inconsistentReplica("p/0002/h/01234567890123456789/r/repository-1", 7L)
                .inconsistentReplica("p/0003/h/12345678901234567890/r/repository-2", 9L)
                .inconsistentReplica("p/0004/h/23456789012345678901/r/repository-3", 12L)
                .fromControlPlane(true)
                .build());

        verify(replicaStateRegistry).markInconsistent(inconsistentReplicasCaptor.capture());
        MarkInconsistentParameters inconsistentParameters = inconsistentReplicasCaptor.getValue();
        assertEquals(ImmutableMap.of("p/0002/h/01234567890123456789/r/repository-1", 7L,
                        "p/0003/h/12345678901234567890/r/repository-2", 9L),
                inconsistentParameters.getInconsistentReplicas());
        assertTrue(inconsistentParameters.isFromControlPlane());
        assertTrue(inconsistentParameters.isComplete());
    }

    @Test
    public void testUpdateConfigurationIgnoresPublicKeysWhenAuthenticationIsDisabled() {
        nodeManager.updateConfiguration(new UpdateConfigurationParameters.Builder()
                .fromControlPlane(true)
                .publicKeys(new UpdatePublicKeysParameters.Builder().build())
                .build());

        logListener.assertDebug("authentication is disabled");
    }

    @Test
    public void testUpdateConfigurationResultsInEmptyFile() throws IOException {
        Files.write(configFile, Collections.singletonList(PORT_PROPERTY + "=12334"));
        String defaultValue = requireNonNull(environment.getProperty(PORT_PROPERTY), "property is not defined");

        nodeManager.updateConfiguration(new UpdateConfigurationParameters.Builder()
                .property(PORT_PROPERTY, defaultValue)
                .build());
        //Setting the default value should clear the property from the config file and leave an empty file behind
        assertTrue(Files.isRegularFile(configFile));
        assertTrue(isBlank(MoreFiles.toString(configFile)));
        assertEquals(defaultValue, environment.getProperty(PORT_PROPERTY));

        verify(config).reload();
        verify(eventPublisher).publish(any(ConfigurationUpdatedEvent.class));
    }

    @Test
    public void testUpdateConfigurationSettingCredentials() throws IOException {
        nodeManager.updateConfiguration(new UpdateConfigurationParameters.Builder()
                .property(NODE_ID, "node-id")
                .property(NODE_NAME, "node-name")
                .build());
        assertTrue(Files.isRegularFile(configFile));
        assertFalse(isBlank(MoreFiles.toString(configFile)));
        assertEquals("node-id", environment.getProperty("node.id"));
        assertEquals("node-name", environment.getProperty("node.name"));

        verify(config).reload();
        verify(eventPublisher).publish(any(ConfigurationUpdatedEvent.class));
    }

    @Test
    public void testUpdateConfigurationThrowsIfUpdatingInconsistentReplicasFails() {
        doThrow(RuntimeException.class).when(replicaStateRegistry).markInconsistent(any());

        ConfigurationFailedException thrown = assertThrows(ConfigurationFailedException.class,
                () -> nodeManager.updateConfiguration(new UpdateConfigurationParameters.Builder()
                        .inconsistentReplica("id", 42L)
                        .build()));
        assertEquals("Inconsistent replica registrations were not applied", thrown.getMessage());
    }

    @Test
    public void testUpdateConfigurationThrowsIfUpdatingPublicKeysFails() {
        AuthenticationManager authenticationManager = mock(AuthenticationManager.class);
        doThrow(InvalidPublicKeyException.class).when(authenticationManager).updatePublicKeys(any());

        ConfigurationFailedException thrown = assertThrows(ConfigurationFailedException.class,
                () -> create(authenticationManager).updateConfiguration(new UpdateConfigurationParameters.Builder()
                        .fromControlPlane(true)
                        .publicKeys(new UpdatePublicKeysParameters.Builder().build())
                        .build()));
        assertEquals("Public keys were not applied", thrown.getMessage());
    }

    @Test
    public void testUpdateConfigurationUpdatesPublicKeysBeforeTopology() {
        AuthenticationManager authenticationManager = mock(AuthenticationManager.class);
        UpdatePublicKeysParameters publicKeys = new UpdatePublicKeysParameters.Builder().build();

        create(authenticationManager).updateConfiguration(new UpdateConfigurationParameters.Builder()
                .fromControlPlane(true)
                .publicKeys(publicKeys)
                .topology(TOPOLOGY)
                .build());

        InOrder ordered = inOrder(authenticationManager, meshManager);
        ordered.verify(authenticationManager).updatePublicKeys(same(publicKeys));
        ordered.verify(meshManager).updateTopology(same(TOPOLOGY));
    }

    @Test
    public void testUpdateConfigurationWithEmptyMap() {
        //Ignore any environment calls made while constructing DefaultNodeManager so we can focus on
        //verifying updateConfiguration doesn't make any calls of its own
        reset(environment);

        nodeManager.updateConfiguration(new UpdateConfigurationParameters.Builder().build());

        //If there's no configuration to apply, the config should not be reloaded and the environment
        //should not be updated
        verifyNoInteractions(config, environment);
    }

    private DefaultNodeManager create(AuthenticationManager authenticationManager) {
        return new DefaultNodeManager(authenticationManager, buildInfo, builderFactory, config, environment,
                eventPublisher, new FileSanitizer(storageLayout.getTempDir()), meshManager, propertySourceLoader,
                replicaStateRegistry, storageLayout, supportInfoRegistry, throttleService, true);
    }

    private void mockVersionInfo(GitProcessBuilder builder, Version current, Version startup) {
        GitBinary binary = mock(GitBinary.class);
        when(binary.getVersion()).thenReturn(startup);

        GitConfig gitConfig = mock(GitConfig.class);
        when(gitConfig.getBinary()).thenReturn(binary);
        when(config.getGit()).thenReturn(gitConfig);

        GitVersionBuilder version = mock(GitVersionBuilder.class);
        when(builder.version()).thenReturn(version);

        MutableObject<VersionStdioHandler> handler = new MutableObject<>();

        @SuppressWarnings("unchecked")
        GitProcess<Version> process = mock(GitProcess.class);
        when(version.build(any(VersionStdioHandler.class))).thenAnswer(i -> {
            handler.setValue(i.getArgument(0));
            return process;
        });
        when(process.run()).thenAnswer(i -> {
            handler.getValue().onStdout(ByteBuffer.wrap(("git version " + current).getBytes(UTF_8)), true);
            return handler.getValue().getResult();
        });
    }
}
