/*
 * Copyright 2018, EnMasse authors.
 * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html).
 */
package io.enmasse.systemtest.common.plans;

import io.enmasse.systemtest.*;
import io.enmasse.systemtest.amqp.AmqpClient;
import io.enmasse.systemtest.bases.TestBase;
import io.enmasse.systemtest.messagingclients.rhea.RheaClientSender;
import io.enmasse.systemtest.resources.*;
import io.enmasse.systemtest.selenium.ISeleniumProviderChrome;
import io.enmasse.systemtest.selenium.page.ConsoleWebPage;
import io.enmasse.systemtest.selenium.resources.AddressWebItem;
import io.enmasse.systemtest.standard.QueueTest;
import io.enmasse.systemtest.standard.TopicTest;
import io.vertx.core.json.JsonObject;
import org.apache.qpid.proton.message.Message;
import org.junit.jupiter.api.*;
import org.slf4j.Logger;

import java.util.*;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

import static io.enmasse.systemtest.TestTag.isolated;
import static io.enmasse.systemtest.TestTag.nonPR;
import static org.hamcrest.CoreMatchers.is;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.junit.jupiter.api.Assertions.*;

@Tag(isolated)
class PlansTest extends TestBase implements ISeleniumProviderChrome {

    private static Logger log = CustomLogger.getLogger();
    private static final PlansProvider plansProvider = new PlansProvider(kubernetes);

    @BeforeEach
    void setUp() throws Exception {
        plansProvider.setUp();
        if (selenium.getDriver() == null)
            selenium.setupDriver(environment, kubernetes, TestUtils.getChromeDriver());
        else
            selenium.clearScreenShots();
    }

    @AfterEach
    void tearDown() throws Exception {
        plansProvider.tearDown();
    }

    @Test
    void testCreateAddressSpacePlan() throws Exception {
        //define and create address plans
        List<AddressResource> addressResourcesQueue = Arrays.asList(new AddressResource("broker", 1.0), new AddressResource("router", 0.0));
        List<AddressResource> addressResourcesTopic = Arrays.asList(
                new AddressResource("broker", 1.0),
                new AddressResource("router", 1.0));
        AddressPlan weakQueuePlan = new AddressPlan("standard-queue-weak", AddressType.QUEUE, addressResourcesQueue);
        AddressPlan weakTopicPlan = new AddressPlan("standard-topic-weak", AddressType.TOPIC, addressResourcesTopic);

        plansProvider.createAddressPlan(weakQueuePlan);
        plansProvider.createAddressPlan(weakTopicPlan);

        //define and create address space plan
        List<AddressSpaceResource> resources = Arrays.asList(
                new AddressSpaceResource("broker", 9.0),
                new AddressSpaceResource("router", 5.0),
                new AddressSpaceResource("aggregate", 10.0));
        List<AddressPlan> addressPlans = Arrays.asList(weakQueuePlan, weakTopicPlan);
        AddressSpacePlan weakSpacePlan = new AddressSpacePlan("weak-plan",
                "default", AddressSpaceType.STANDARD, resources, addressPlans);
        plansProvider.createAddressSpacePlan(weakSpacePlan);

        //create address space plan with new plan
        AddressSpace weakAddressSpace = new AddressSpace("weak-address-space", AddressSpaceType.STANDARD,
                weakSpacePlan.getName(), AuthService.STANDARD);
        createAddressSpace(weakAddressSpace);

        //deploy destinations
        Destination weakQueueDest = Destination.queue("weak-queue", weakQueuePlan.getName());
        Destination weakTopicDest = Destination.topic("weak-topic", weakTopicPlan.getName());
        setAddresses(weakAddressSpace, weakQueueDest, weakTopicDest);

        //get destinations
        Future<List<Address>> getWeakQueue = getAddressesObjects(weakAddressSpace, Optional.of(weakQueueDest.getName()));
        Future<List<Address>> getWeakTopic = getAddressesObjects(weakAddressSpace, Optional.of(weakTopicDest.getName()));

        String assertMessage = "Queue plan wasn't set properly";
        assertAll("Both destination should contain right addressPlan",
                () -> assertEquals(getWeakQueue.get(20, TimeUnit.SECONDS).get(0).getPlan(),
                        weakQueuePlan.getName(), assertMessage),
                () -> assertEquals(getWeakTopic.get(20, TimeUnit.SECONDS).get(0).getPlan(),
                        weakTopicPlan.getName(), assertMessage));

        //simple send/receive
        UserCredentials user = new UserCredentials("test-newplan-name", "test_newplan_password");
        createUser(weakAddressSpace, user);

        AmqpClient queueClient = amqpClientFactory.createQueueClient(weakAddressSpace);
        queueClient.getConnectOptions().setCredentials(user);
        QueueTest.runQueueTest(queueClient, weakQueueDest, 42);

        AmqpClient topicClient = amqpClientFactory.createTopicClient(weakAddressSpace);
        topicClient.getConnectOptions().setCredentials(user);
        TopicTest.runTopicTest(topicClient, weakTopicDest, 42);
    }

    @Test
    void testQuotaLimitsPooled() throws Exception {
        //define and create address plans
        AddressPlan queuePlan = new AddressPlan("queue-pooled-test1", AddressType.QUEUE,
                Arrays.asList(new AddressResource("broker", 0.6), new AddressResource("router", 0.0)));

        AddressPlan queuePlan2 = new AddressPlan("queue-pooled-test2", AddressType.QUEUE,
                Arrays.asList(new AddressResource("broker", 0.1), new AddressResource("router", 0.0)));

        AddressPlan queuePlan3 = new AddressPlan("queue-pooled-test3", AddressType.QUEUE,
                Arrays.asList(new AddressResource("broker", 0.049), new AddressResource("router", 0.0)));

        AddressPlan topicPlan = new AddressPlan("topic-pooled-test1", AddressType.TOPIC,
                Arrays.asList(
                        new AddressResource("broker", 0.4),
                        new AddressResource("router", 0.2)));

        AddressPlan anycastPlan = new AddressPlan("anycast-test1", AddressType.ANYCAST,
                Collections.singletonList(new AddressResource("router", 0.3)));

        plansProvider.createAddressPlan(queuePlan);
        plansProvider.createAddressPlan(queuePlan2);
        plansProvider.createAddressPlan(queuePlan3);
        plansProvider.createAddressPlan(topicPlan);
        plansProvider.createAddressPlan(anycastPlan);

        //define and create address space plan
        List<AddressSpaceResource> resources = Arrays.asList(
                new AddressSpaceResource("broker", 2.0),
                new AddressSpaceResource("router", 1.0),
                new AddressSpaceResource("aggregate", 2.0));
        List<AddressPlan> addressPlans = Arrays.asList(queuePlan, queuePlan2, queuePlan3, topicPlan, anycastPlan);
        AddressSpacePlan addressSpacePlan = new AddressSpacePlan("quota-limits-pooled-plan",
                "default-minimal", AddressSpaceType.STANDARD, resources, addressPlans);
        plansProvider.createAddressSpacePlan(addressSpacePlan);

        //create address space with new plan
        AddressSpace addressSpace = new AddressSpace("test-pooled-space", AddressSpaceType.STANDARD,
                addressSpacePlan.getName(), AuthService.STANDARD);
        createAddressSpace(addressSpace);
        UserCredentials user = new UserCredentials("quota-user", "quotaPa55");
        createUser(addressSpace, user);

        //check router limits
        checkLimits(addressSpace,
                Arrays.asList(
                        Destination.anycast("a1", anycastPlan.getName()),
                        Destination.anycast("a2", anycastPlan.getName()),
                        Destination.anycast("a3", anycastPlan.getName())
                ),
                Collections.singletonList(
                        Destination.anycast("a4", anycastPlan.getName())
                ), user);

        //check broker limits
        checkLimits(addressSpace,
                Arrays.asList(
                        Destination.queue("q1", queuePlan.getName()),
                        Destination.queue("q2", queuePlan.getName())
                ),
                Collections.singletonList(
                        Destination.queue("q3", queuePlan.getName())
                ), user);

        checkLimits(addressSpace,
                Arrays.asList(
                        Destination.queue("q1", queuePlan.getName()), // 0.6
                        Destination.queue("q2", queuePlan.getName()), // 0.6
                        Destination.queue("q3", queuePlan2.getName()), // 0.1
                        Destination.queue("q4", queuePlan2.getName()), // 0.1
                        Destination.queue("q5", queuePlan2.getName()), // 0.1
                        Destination.queue("q6", queuePlan2.getName()), // 0.1
                        Destination.queue("q7", queuePlan3.getName()), // 0.049
                        Destination.queue("q8", queuePlan3.getName()), // 0.049
                        Destination.queue("q9", queuePlan3.getName()), // 0.049
                        Destination.queue("q10", queuePlan3.getName()), // 0.049
                        Destination.queue("q11", queuePlan3.getName()), // 0.049
                        Destination.queue("q12", queuePlan3.getName()) // 0.049
                ), Collections.emptyList(), user);

        //check aggregate limits
        checkLimits(addressSpace,
                Arrays.asList(
                        Destination.topic("t1", topicPlan.getName()),
                        Destination.topic("t2", topicPlan.getName())
                ),
                Collections.singletonList(
                        Destination.topic("t3", topicPlan.getName())
                ), user);
    }

    @Test
    void testQuotaLimitsSharded() throws Exception {
        //define and create address plans
        AddressPlan queuePlan = new AddressPlan("queue-sharded-test1", AddressType.QUEUE,
                Arrays.asList(new AddressResource("broker", 1.0), new AddressResource("router", 0.0)));

        AddressPlan topicPlan = new AddressPlan("topic-sharded-test2", AddressType.TOPIC,
                Arrays.asList(
                        new AddressResource("broker", 1.0),
                        new AddressResource("router", 0.01)));

        plansProvider.createAddressPlan(queuePlan);
        plansProvider.createAddressPlan(topicPlan);

        //define and create address space plan
        List<AddressSpaceResource> resources = Arrays.asList(
                new AddressSpaceResource("broker", 2.0),
                new AddressSpaceResource("router", 2.0),
                new AddressSpaceResource("aggregate", 3.0));
        List<AddressPlan> addressPlans = Arrays.asList(queuePlan, topicPlan);
        AddressSpacePlan addressSpacePlan = new AddressSpacePlan("quota-limits-sharded-plan",
                "default-minimal", AddressSpaceType.STANDARD, resources, addressPlans);
        plansProvider.createAddressSpacePlan(addressSpacePlan);

        //create address space with new plan
        AddressSpace addressSpace = new AddressSpace("test-sharded-space", AddressSpaceType.STANDARD,
                addressSpacePlan.getName(), AuthService.STANDARD);
        createAddressSpace(addressSpace);
        UserCredentials user = new UserCredentials("quota-user", "quotaPa55");
        createUser(addressSpace, user);

        //check broker limits
        checkLimits(addressSpace,
                Arrays.asList(
                        Destination.queue("q1", queuePlan.getName()),
                        Destination.queue("q2", queuePlan.getName())
                ),
                Collections.singletonList(
                        Destination.queue("q3", queuePlan.getName())
                ), user);

        //check aggregate limits
        checkLimits(addressSpace,
                Arrays.asList(
                        Destination.topic("t1", topicPlan.getName()),
                        Destination.topic("t2", topicPlan.getName())
                ),
                Collections.singletonList(
                        Destination.topic("t3", topicPlan.getName())
                ), user);
    }

    @Test
    void testScalePooledBrokers() throws Exception {
        //define and create address plans
        List<AddressResource> addressResourcesQueue = Arrays.asList(new AddressResource("broker", 0.99), new AddressResource("router", 0.0));
        AddressPlan xlQueuePlan = new AddressPlan("pooled-xl-queue", AddressType.QUEUE, addressResourcesQueue);
        plansProvider.createAddressPlan(xlQueuePlan);

        //define and create address space plan
        List<AddressSpaceResource> resources = Arrays.asList(
                new AddressSpaceResource("broker", 10.0),
                new AddressSpaceResource("router", 2.0),
                new AddressSpaceResource("aggregate", 12.0));
        List<AddressPlan> addressPlans = Collections.singletonList(xlQueuePlan);
        AddressSpacePlan manyAddressesPlan = new AddressSpacePlan("many-brokers-plan",
                "default", AddressSpaceType.STANDARD, resources, addressPlans);
        plansProvider.createAddressSpacePlan(manyAddressesPlan);

        //create address space plan with new plan
        AddressSpace manyAddressesSpace = new AddressSpace("many-addresses-standard", AddressSpaceType.STANDARD,
                manyAddressesPlan.getName(), AuthService.STANDARD);
        createAddressSpace(manyAddressesSpace);

        UserCredentials cred = new UserCredentials("testus", "papyrus");
        createUser(manyAddressesSpace, cred);

        ArrayList<Destination> dest = new ArrayList<>();
        int destCount = 4;
        int toDeleteCount = 2;
        for (int i = 0; i < destCount; i++) {
            dest.add(Destination.queue("xl-queue-" + i, xlQueuePlan.getName()));
        }

        setAddresses(manyAddressesSpace, dest.toArray(new Destination[0]));
        for (Destination destination : dest) {
            waitForBrokerReplicas(manyAddressesSpace, destination, 1);
        }

        assertCanConnect(manyAddressesSpace, cred, dest);

        deleteAddresses(manyAddressesSpace, dest.subList(0, toDeleteCount).toArray(new Destination[0]));
        for (Destination destination : dest.subList(toDeleteCount, destCount)) {
            waitForBrokerReplicas(manyAddressesSpace, destination, 1);
        }

        assertCanConnect(manyAddressesSpace, cred, dest.subList(toDeleteCount, destCount));
    }

    @Test
    @Tag(nonPR)
    @Disabled("test disabled as per-address limit enforcement has been removed")
    void testGlobalSizeLimitations() throws Exception {
        UserCredentials user = new UserCredentials("test", "test");
        String messageContent = String.join("", Collections.nCopies(1024, "F"));

        //redefine global max size for queue
        ResourceDefinition limitedResource = new ResourceDefinition(
                "broker",
                "queue-persisted",
                Collections.singletonList(
                        new ResourceParameter("GLOBAL_MAX_SIZE", "1Mb")
                ));
        //plansProvider.replaceResourceDefinitionConfig(limitedResource);

        //define address plans
        AddressPlan queuePlan = new AddressPlan("limited-queue", AddressType.QUEUE,
                Arrays.asList(new AddressResource("broker", 0.1), new AddressResource("router", 0.0))); //should reserve 100Kb

        plansProvider.createAddressPlan(queuePlan);

        //define and create address space plan
        List<AddressSpaceResource> resources = Arrays.asList(
                new AddressSpaceResource("broker", 1.0),
                new AddressSpaceResource("router", 1.0),
                new AddressSpaceResource("aggregate", 2.0));

        AddressSpacePlan addressSpacePlan = new AddressSpacePlan(
                "limited-space",
                "default",
                AddressSpaceType.STANDARD,
                resources,
                Collections.singletonList(queuePlan));
        plansProvider.createAddressSpacePlan(addressSpacePlan);

        //create address space with new plan
        AddressSpace addressSpace = new AddressSpace("global-size-limited-space", AddressSpaceType.STANDARD,
                addressSpacePlan.getName(), AuthService.STANDARD);
        createAddressSpace(addressSpace);
        createUser(addressSpace, user);

        Destination queue = Destination.queue("test-queue", queuePlan.getName());
        Destination queue2 = Destination.queue("test-queue2", queuePlan.getName());
        Destination queue3 = Destination.queue("test-queue3", queuePlan.getName());
        setAddresses(addressSpace, queue, queue2, queue3);

        assertAll(
                () -> assertFalse(sendMessage(addressSpace, new RheaClientSender(), user,
                        queue.getAddress(), messageContent, 100, false),
                        "Client does not fail"),
                () -> assertFalse(sendMessage(addressSpace, new RheaClientSender(), user,
                        queue2.getAddress(), messageContent, 100, false),
                        "Client does not fail"),
                () -> assertTrue(sendMessage(addressSpace, new RheaClientSender(), user,
                        queue3.getAddress(), messageContent, 50, false),
                        "Client fails"));
    }

    @Test
    void testMessagePersistenceAfterAutoScale() throws Exception {
        //define and create address plans
        List<AddressResource> addressResourcesQueueAlpha = Arrays.asList(new AddressResource("broker", 0.3), new AddressResource("router", 0));
        List<AddressResource> addressResourcesQueueBeta = Arrays.asList(new AddressResource("broker", 0.6), new AddressResource("router", 0));

        AddressPlan queuePlanAlpha = new AddressPlan("pooled-standard-queue-alpha", AddressType.QUEUE, addressResourcesQueueAlpha);
        plansProvider.createAddressPlan(queuePlanAlpha);
        AddressPlan queuePlanBeta = new AddressPlan("pooled-standard-queue-beta", AddressType.QUEUE, addressResourcesQueueBeta);
        plansProvider.createAddressPlan(queuePlanBeta);


        //define and create address space plan
        List<AddressSpaceResource> resources = Arrays.asList(
                new AddressSpaceResource("broker", 3.0),
                new AddressSpaceResource("router", 5.0),
                new AddressSpaceResource("aggregate", 5.0));
        List<AddressPlan> addressPlans = Arrays.asList(queuePlanAlpha, queuePlanBeta);
        AddressSpacePlan scaleSpacePlan = new AddressSpacePlan("scale-plan",
                "default", AddressSpaceType.STANDARD, resources, addressPlans);
        plansProvider.createAddressSpacePlan(scaleSpacePlan);

        //create address space plan with new plan
        AddressSpace messagePersistAddressSpace = new AddressSpace("persist-messages-space-standard", AddressSpaceType.STANDARD,
                scaleSpacePlan.getName(), AuthService.STANDARD);
        createAddressSpace(messagePersistAddressSpace);

        //deploy destinations
        Destination queue1 = Destination.queue("queue1-beta", queuePlanBeta.getName());
        Destination queue2 = Destination.queue("queue2-beta", queuePlanBeta.getName());
        Destination queue3 = Destination.queue("queue3-alpha", queuePlanAlpha.getName());
        Destination queue4 = Destination.queue("queue4-alpha", queuePlanAlpha.getName());

        setAddresses(messagePersistAddressSpace, queue1, queue2);
        appendAddresses(messagePersistAddressSpace, queue3, queue4);

        //send 1000 messages to each queue
        UserCredentials user = new UserCredentials("test-scale-user-name", "test_scale_user_pswd");
        createUser(messagePersistAddressSpace, user);

        AmqpClient queueClient = amqpClientFactory.createQueueClient(messagePersistAddressSpace);
        queueClient.getConnectOptions().setCredentials(user);

        List<String> msgs = TestUtils.generateMessages(1000);
        Future<Integer> sendResult1 = queueClient.sendMessages(queue1.getAddress(), msgs);
        Future<Integer> sendResult2 = queueClient.sendMessages(queue2.getAddress(), msgs);
        Future<Integer> sendResult3 = queueClient.sendMessages(queue3.getAddress(), msgs);
        Future<Integer> sendResult4 = queueClient.sendMessages(queue4.getAddress(), msgs);
        assertAll("All senders should send all messages",
                () -> assertThat("Incorrect count of messages sent", sendResult1.get(1, TimeUnit.MINUTES), is(msgs.size())),
                () -> assertThat("Incorrect count of messages sent", sendResult2.get(1, TimeUnit.MINUTES), is(msgs.size())),
                () -> assertThat("Incorrect count of messages sent", sendResult3.get(1, TimeUnit.MINUTES), is(msgs.size())),
                () -> assertThat("Incorrect count of messages sent", sendResult4.get(1, TimeUnit.MINUTES), is(msgs.size())));

        //remove addresses from first pod and wait for scale down
        deleteAddresses(messagePersistAddressSpace, queue1, queue2);
        TestUtils.waitForNBrokerReplicas(addressApiClient, kubernetes, messagePersistAddressSpace, 1, queue4, new TimeoutBudget(2, TimeUnit.MINUTES));

        //validate count of addresses
        Future<List<String>> addresses = getAddresses(messagePersistAddressSpace, Optional.empty());
        List<String> addressNames = addresses.get(15, TimeUnit.SECONDS);
        assertThat(String.format("Unexpected count of destinations, got following: %s", addressNames),
                addressNames.size(), is(2));

        //receive messages from remaining addresses
        Future<List<Message>> recvResult3 = queueClient.recvMessages(queue3.getAddress(), msgs.size());
        Future<List<Message>> recvResult4 = queueClient.recvMessages(queue4.getAddress(), msgs.size());
        assertThat("Incorrect count of messages received", recvResult3.get(1, TimeUnit.MINUTES).size(), is(msgs.size()));
        assertThat("Incorrect count of messages received", recvResult4.get(1, TimeUnit.MINUTES).size(), is(msgs.size()));
    }

    @Test
    @Disabled("test disabled due to issue: #1136")
    void testMessagePersistenceAfterChangePlan() throws Exception {
        List<AddressResource> addressResourcesQueueDistributed = Arrays.asList(new AddressResource("broker", 2.0), new AddressResource("router", 0));
        List<AddressResource> addressResourcesSharded = Arrays.asList(new AddressResource("broker", 1.0), new AddressResource("router", 0));

        AddressPlan queuePlanDistributed = new AddressPlan("distributed-standard-queue-alpha", AddressType.QUEUE, addressResourcesQueueDistributed);
        plansProvider.createAddressPlan(queuePlanDistributed);

        AddressPlan queuePlanSharded = new AddressPlan("sharded-standard-queue", AddressType.QUEUE, addressResourcesSharded);
        plansProvider.createAddressPlan(queuePlanSharded);

        //define and create address space plan
        List<AddressSpaceResource> resources = Arrays.asList(
                new AddressSpaceResource("broker", 5.0),
                new AddressSpaceResource("router", 5.0),
                new AddressSpaceResource("aggregate", 5.0));
        List<AddressPlan> addressPlans = Arrays.asList(queuePlanDistributed, queuePlanSharded);
        AddressSpacePlan scaleSpacePlan = new AddressSpacePlan("scale-plan",
                "default", AddressSpaceType.STANDARD, resources, addressPlans);
        plansProvider.createAddressSpacePlan(scaleSpacePlan);

        //create address space plan with new plan
        AddressSpace messagePersistAddressSpace = new AddressSpace("persist-messages-space-standard", AddressSpaceType.STANDARD,
                scaleSpacePlan.getName(), AuthService.STANDARD);
        createAddressSpace(messagePersistAddressSpace);

        //deploy destinations
        Destination queue = Destination.queue("distributed-queue", queuePlanDistributed.getName());
        setAddresses(messagePersistAddressSpace, queue);

        //pod should have 2 replicas
        TestUtils.waitForNBrokerReplicas(addressApiClient, kubernetes, messagePersistAddressSpace, 2, queue, new TimeoutBudget(2, TimeUnit.MINUTES));

        //send 100000 messages to queue
        UserCredentials user = new UserCredentials("test-change-plan-user", "test_change_plan_pswd");
        createUser(messagePersistAddressSpace, user);

        AmqpClient queueClient = amqpClientFactory.createQueueClient(messagePersistAddressSpace);
        queueClient.getConnectOptions().setCredentials(user);

        List<String> msgs = TestUtils.generateMessages(100_000);
        Future<Integer> sendResult1 = queueClient.sendMessages(queue.getAddress(), msgs);
        assertThat("Incorrect count of messages sent", sendResult1.get(1, TimeUnit.MINUTES), is(msgs.size()));

        //replace original plan in address by another
        plansProvider.replaceAddressPlan(queuePlanSharded);

        assertEquals(getAddressesObjects(
                messagePersistAddressSpace,
                Optional.of(queue.getName())).get(10, TimeUnit.SECONDS).get(0).getPlan(),
                queuePlanSharded.getName(),
                "New plan wasn't set correctly");

        //wait until address will be scaled down to 1 pod
        TestUtils.waitForNBrokerReplicas(
                addressApiClient,
                kubernetes,
                messagePersistAddressSpace, 1, queue, new TimeoutBudget(2, TimeUnit.MINUTES));
        //test failed in command above ^, functionality of test code below wasn't verified :) !TODO

        //receive messages
        Future<List<Message>> recvResult = queueClient.recvMessages(queue.getAddress(), msgs.size());
        assertThat("Incorrect count of messages received", recvResult.get(1, TimeUnit.MINUTES).size(), is(msgs.size()));
    }

    @Test
    void testReplaceAddressSpacePlanStandard() throws Exception {
        //define and create address plans
        AddressPlan beforeQueuePlan = new AddressPlan("before-small-sharded-queue", AddressType.QUEUE,
                Arrays.asList(new AddressResource("broker", 1.0), new AddressResource("router", 0.0)));

        AddressPlan beforeTopicPlan = new AddressPlan("before-small-sharded-topic", AddressType.TOPIC,
                Arrays.asList(
                        new AddressResource("broker", 1.0),
                        new AddressResource("router", 0.01)));

        AddressPlan afterQueuePlan = new AddressPlan("after-large-sharded-queue", AddressType.QUEUE,
                Arrays.asList(new AddressResource("broker", 1.5), new AddressResource("router", 0.0)));

        AddressPlan afterTopicPlan = new AddressPlan("after-large-sharded-topic", AddressType.TOPIC,
                Arrays.asList(
                        new AddressResource("broker", 1.5),
                        new AddressResource("router", 0.01)));

        AddressPlan pooledQueuePlan = new AddressPlan("after-pooled-queue", AddressType.QUEUE,
                Arrays.asList(new AddressResource("broker", 0.44), new AddressResource("router", 0.0)));

        plansProvider.createAddressPlan(beforeQueuePlan);
        plansProvider.createAddressPlan(beforeTopicPlan);
        plansProvider.createAddressPlan(afterQueuePlan);
        plansProvider.createAddressPlan(afterTopicPlan);
        plansProvider.createAddressPlan(pooledQueuePlan);

        //define and create address space plans

        AddressSpacePlan beforeAddressSpacePlan = new AddressSpacePlan("before-update-standard-plan",
                "default-minimal", AddressSpaceType.STANDARD,
                Arrays.asList(
                        new AddressSpaceResource("broker", 5.0),
                        new AddressSpaceResource("router", 5.0),
                        new AddressSpaceResource("aggregate", 10.0)),
                Arrays.asList(beforeQueuePlan, beforeTopicPlan));

        AddressSpacePlan afterAddressSpacePlan = new AddressSpacePlan("after-update-standard-plan",
                "default-minimal", AddressSpaceType.STANDARD,
                Arrays.asList(
                        new AddressSpaceResource("broker", 5.0),
                        new AddressSpaceResource("router", 5.0),
                        new AddressSpaceResource("aggregate", 10.0)),
                Arrays.asList(afterQueuePlan, afterTopicPlan));

        AddressSpacePlan pooledAddressSpacePlan = new AddressSpacePlan("after-update-standard-pooled-plan",
                "default-minimal", AddressSpaceType.STANDARD,
                Arrays.asList(
                        new AddressSpaceResource("broker", 10.0),
                        new AddressSpaceResource("router", 10.0),
                        new AddressSpaceResource("aggregate", 10.0)),
                Collections.singletonList(pooledQueuePlan));


        plansProvider.createAddressSpacePlan(beforeAddressSpacePlan);
        plansProvider.createAddressSpacePlan(afterAddressSpacePlan);
        plansProvider.createAddressSpacePlan(pooledAddressSpacePlan);

        //create address space with new plan
        AddressSpace addressSpace = new AddressSpace("test-sharded-space", AddressSpaceType.STANDARD,
                beforeAddressSpacePlan.getName(), AuthService.STANDARD);
        createAddressSpace(addressSpace);

        UserCredentials user = new UserCredentials("quota-user", "quotaPa55");
        createUser(addressSpace, user);

        Destination queue = Destination.queue("test-queue", beforeQueuePlan.getName());
        Destination topic = Destination.topic("test-topic", beforeTopicPlan.getName());

        setAddresses(addressSpace, queue, topic);

        sendDurableMessages(addressSpace, queue, user, 16);

        addressSpace.setPlan(afterAddressSpacePlan.getName());
        replaceAddressSpace(addressSpace);

        receiveDurableMessages(addressSpace, queue, user, 16);

        Destination afterQueue = Destination.queue("test-queue-2", afterQueuePlan.getName());
        appendAddresses(addressSpace, afterQueue);

        assertCanConnect(addressSpace, user, Arrays.asList(afterQueue, queue, topic));

        addressSpace.setPlan(pooledAddressSpacePlan.getName());
        replaceAddressSpace(addressSpace);

        Destination pooledQueue = Destination.queue("test-queue-3", pooledQueuePlan.getName());
        appendAddresses(addressSpace, pooledQueue);

        assertCanConnect(addressSpace, user, Arrays.asList(queue, topic, afterQueue, pooledQueue));
    }

    @Test
    void testReplaceAddressSpacePlanBrokered() throws Exception {
        //define and create address plans
        AddressPlan beforeQueuePlan = new AddressPlan("small-queue", AddressType.QUEUE,
                Collections.singletonList(new AddressResource("broker", 0.4)));

        AddressPlan afterQueuePlan = new AddressPlan("bigger-queue", AddressType.QUEUE,
                Collections.singletonList(new AddressResource("broker", 0.7)));

        plansProvider.createAddressPlan(beforeQueuePlan);
        plansProvider.createAddressPlan(afterQueuePlan);

        //define and create address space plans

        AddressSpacePlan beforeAddressSpacePlan = new AddressSpacePlan("before-update-brokered-plan",
                "default", AddressSpaceType.BROKERED,
                Collections.singletonList(new AddressSpaceResource("broker", 5.0)),
                Collections.singletonList(beforeQueuePlan));

        AddressSpacePlan afterAddressSpacePlan = new AddressSpacePlan("after-update-standard-plan",
                "default", AddressSpaceType.BROKERED,
                Collections.singletonList(new AddressSpaceResource("broker", 5.0)),
                Collections.singletonList(afterQueuePlan));

        plansProvider.createAddressSpacePlan(beforeAddressSpacePlan);
        plansProvider.createAddressSpacePlan(afterAddressSpacePlan);

        //create address space with new plan
        AddressSpace addressSpace = new AddressSpace("test-sharded-space", AddressSpaceType.BROKERED,
                beforeAddressSpacePlan.getName(), AuthService.STANDARD);
        createAddressSpace(addressSpace);

        UserCredentials user = new UserCredentials("quota-user", "quotaPa55");
        createUser(addressSpace, user);

        Destination queue = Destination.queue("test-queue", beforeQueuePlan.getName());

        setAddresses(addressSpace, queue);

        sendDurableMessages(addressSpace, queue, user, 16);

        addressSpace.setPlan(afterAddressSpacePlan.getName());
        replaceAddressSpace(addressSpace);

        receiveDurableMessages(addressSpace, queue, user, 16);

        Destination afterQueue = Destination.queue("test-queue-2", afterQueuePlan.getName());
        appendAddresses(addressSpace, afterQueue);

        assertCanConnect(addressSpace, user, Arrays.asList(afterQueue, queue));
    }

    @Test
    void testCannotReplaceAddressSpacePlanStandard() throws Exception {
        //define and create address plans
        AddressPlan afterQueuePlan = new AddressPlan("after-small-sharded-queue", AddressType.QUEUE,
                Arrays.asList(new AddressResource("broker", 1.0), new AddressResource("router", 0)));

        AddressPlan beforeQueuePlan = new AddressPlan("before-large-sharded-queue", AddressType.QUEUE,
                Arrays.asList(new AddressResource("broker", 2.0), new AddressResource("router", 0)));

        plansProvider.createAddressPlan(beforeQueuePlan);
        plansProvider.createAddressPlan(afterQueuePlan);

        //define and create address space plans

        AddressSpacePlan beforeAddressSpacePlan = new AddressSpacePlan("before-update-standard-plan",
                "default-minimal", AddressSpaceType.STANDARD,
                Arrays.asList(
                        new AddressSpaceResource("broker", 5.0),
                        new AddressSpaceResource("router", 2.0),
                        new AddressSpaceResource("aggregate", 7.0)),
                Collections.singletonList(beforeQueuePlan));

        AddressSpacePlan afterAddressSpacePlan = new AddressSpacePlan("after-update-standard-plan",
                "default-minimal", AddressSpaceType.STANDARD,
                Arrays.asList(
                        new AddressSpaceResource("broker", 2.0),
                        new AddressSpaceResource("router", 2.0),
                        new AddressSpaceResource("aggregate", 4.0)),
                Collections.singletonList(afterQueuePlan));


        plansProvider.createAddressSpacePlan(beforeAddressSpacePlan);
        plansProvider.createAddressSpacePlan(afterAddressSpacePlan);

        //create address space with new plan
        AddressSpace addressSpace = new AddressSpace("test-sharded-space", AddressSpaceType.STANDARD,
                beforeAddressSpacePlan.getName(), AuthService.STANDARD);
        createAddressSpace(addressSpace);

        UserCredentials user = new UserCredentials("quota-user", "quotaPa55");
        createUser(addressSpace, user);

        List<Destination> queues = Arrays.asList(
                Destination.queue("test-queue-1", beforeQueuePlan.getName()),
                Destination.queue("test-queue-2", beforeQueuePlan.getName())
        );


        setAddresses(addressSpace, queues.toArray(new Destination[0]));
        assertCanConnect(addressSpace, user, queues);

        addressSpace.setPlan(afterAddressSpacePlan.getName());
        replaceAddressSpace(addressSpace, false);

        JsonObject data = addressApiClient.getAddressSpace(addressSpace.getName());
        assertEquals(beforeAddressSpacePlan.getName(),
                data.getJsonObject("metadata").getJsonObject("annotations").getString("enmasse.io/applied-plan"));
        assertEquals(String.format("Unable to apply plan [%s] to address space %s:%s: quota exceeded for resource broker",
                afterQueuePlan.getName(), environment.namespace(), addressSpace.getName()),
                data.getJsonObject("status").getJsonArray("messages").getString(0));
    }

    @Test
    void testSwitchQueuePlan() throws Exception {
        AddressPlan beforeQueuePlan = new AddressPlan("small-queue", AddressType.QUEUE,
                Arrays.asList(new AddressResource("broker", 0.2), new AddressResource("router", 0.0)));

        AddressPlan afterQueuePlan = new AddressPlan("bigger-queue", AddressType.QUEUE,
                Arrays.asList(new AddressResource("broker", 0.8), new AddressResource("router", 0.0)));

        plansProvider.createAddressPlan(beforeQueuePlan);
        plansProvider.createAddressPlan(afterQueuePlan);

        AddressSpacePlan addressPlan = new AddressSpacePlan("address-switch-address-plan",
                "default-minimal", AddressSpaceType.STANDARD,
                Arrays.asList(
                        new AddressSpaceResource("broker", 5.0),
                        new AddressSpaceResource("router", 5.0),
                        new AddressSpaceResource("aggregate", 10.0)),
                Arrays.asList(beforeQueuePlan, afterQueuePlan));

        plansProvider.createAddressSpacePlan(addressPlan);

        AddressSpace addressSpace = new AddressSpace("test-pooled-space", AddressSpaceType.STANDARD,
                addressPlan.getName(), AuthService.STANDARD);
        createAddressSpace(addressSpace);
        UserCredentials cred = new UserCredentials("test-user", "test-password");
        createUser(addressSpace, cred);

        List<Destination> queues = IntStream.range(0, 8).boxed().map(i ->
                Destination.queue("queue-" + i, beforeQueuePlan.getName()))
                .collect(Collectors.toList());
        setAddresses(addressSpace, queues.toArray(new Destination[0]));

        assertThat("Failed there are no 2 broker pods", TestUtils.listBrokerPods(kubernetes, addressSpace).size(), is(2));

        for (Destination queue : queues) {
            sendDurableMessages(addressSpace, queue, cred, 400);
        }

        Destination queueAfter = Destination.queue("queue-1", afterQueuePlan.getName());
        replaceAddress(addressSpace, queueAfter);

        assertThat("Failed there are no 3 broker pods", TestUtils.listBrokerPods(kubernetes, addressSpace).size(), is(3));

        for (Destination queue : queues) {
            receiveDurableMessages(addressSpace, queue, cred, 400);
        }
    }

    //------------------------------------------------------------------------------------------------
    // Help methods
    //------------------------------------------------------------------------------------------------

    private void checkLimits(AddressSpace addressSpace, List<Destination> allowedDest, List<Destination> notAllowedDest, UserCredentials credentials)
            throws Exception {

        log.info("Try to create {} addresses, and make sure that {} addresses will be not created",
                Arrays.toString(allowedDest.stream().map(Destination::getName).toArray(String[]::new)),
                Arrays.toString(notAllowedDest.stream().map(Destination::getName).toArray(String[]::new)));

        setAddresses(addressSpace, new TimeoutBudget(10, TimeUnit.MINUTES), allowedDest.toArray(new Destination[0]));
        List<Future<List<Address>>> getAddresses = new ArrayList<>();
        for (Destination dest : allowedDest) {
            getAddresses.add(getAddressesObjects(addressSpace, Optional.of(dest.getName())));
        }

        for (Future<List<Address>> getAddress : getAddresses) {
            Address address = getAddress.get(20, TimeUnit.SECONDS).get(0);
            log.info("Address {} with plan {} is in phase {}", address.getName(), address.getPlan(), address.getPhase());
            String assertMessage = String.format("Address from allowed %s is not ready", address.getName());
            assertEquals("Active", address.getPhase(), assertMessage);
        }

        assertCanConnect(addressSpace, credentials, allowedDest);

        getAddresses.clear();
        if (notAllowedDest.size() > 0) {
            try {
                appendAddresses(addressSpace, new TimeoutBudget(30, TimeUnit.SECONDS), notAllowedDest.toArray(new Destination[0]));
            } catch (IllegalStateException ex) {
                if (!ex.getMessage().contains("addresses are not matched")) {
                    throw ex;
                }
            }

            for (Destination dest : notAllowedDest) {
                getAddresses.add(getAddressesObjects(addressSpace, Optional.of(dest.getName())));
            }

            for (Future<List<Address>> getAddress : getAddresses) {
                Address address = getAddress.get(20, TimeUnit.SECONDS).get(0);
                log.info("Address {} with plan {} is in phase {}", address.getName(), address.getPlan(), address.getPhase());
                String assertMessage = String.format("Address from notAllowed %s is ready", address.getName());
                assertEquals("Pending", address.getPhase(), assertMessage);
                assertTrue(address.getStatusMessages().contains("Quota exceeded"), "No status message is present");
            }
        }

        ConsoleWebPage page = new ConsoleWebPage(selenium, getConsoleRoute(addressSpace), addressApiClient, addressSpace, credentials);
        page.openWebConsolePage();
        page.openAddressesPageWebConsole();

        for (Destination dest : allowedDest) {
            AddressWebItem item = (AddressWebItem) selenium.waitUntilItemPresent(25, () -> page.getAddressItem(dest));
            assertNotNull(item, String.format("Address '%s' is not visible in console", dest));
            assertThat("Item is not in state Ready", item.getStatus(), is(AddressStatus.READY));
        }

        for (Destination dest : notAllowedDest) {
            AddressWebItem item = (AddressWebItem) selenium.waitUntilItemPresent(25, () -> page.getAddressItem(dest));
            assertNotNull(item, String.format("Address '%s' is not visible in console", dest));
            assertThat("Item is not in state Pending", item.getStatus(), is(AddressStatus.PENDING));
        }

        setAddresses(addressSpace);
    }
}
