// Copyright 2016 Twitter. All rights reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//    http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package com.twitter.heron.spi.packing;

import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.junit.Assert;
import org.junit.Test;

import com.twitter.heron.spi.common.Constants;

public class PackingPlanTest {
  public static PackingPlan generatePacking(Map<String, List<String>> basePacking) {
    Resource resource =
        new Resource(1.0, 1 * Constants.GB, 10 * Constants.GB);

    Map<String, PackingPlan.ContainerPlan> containerPlanMap = new HashMap<>();

    for (Map.Entry<String, List<String>> entry : basePacking.entrySet()) {
      String containerId = entry.getKey();
      List<String> instanceList = entry.getValue();

      Map<String, PackingPlan.InstancePlan> instancePlanMap = new HashMap<>();

      for (String instanceId : instanceList) {
        String componentName = instanceId.split(":")[1];
        PackingPlan.InstancePlan instancePlan =
            new PackingPlan.InstancePlan(instanceId, componentName, resource);
        instancePlanMap.put(instanceId, instancePlan);
      }

      PackingPlan.ContainerPlan containerPlan =
          new PackingPlan.ContainerPlan(containerId, instancePlanMap, resource);

      containerPlanMap.put(containerId, containerPlan);
    }

    return new PackingPlan("", containerPlanMap, resource);
  }

  @Test
  public void testPackingToString() {
    Map<String, List<String>> packing = new HashMap<>();
    packing.put("1", Arrays.asList("1:spout:1:0", "1:bolt:3:0"));
    String packingStr = generatePacking(packing).getInstanceDistribution();
    String expectedStr0 = "1:spout:1:0:bolt:3:0";
    String expectedStr1 = "1:bolt:3:0:spout:1:0";

    Assert.assertTrue(packingStr.equals(expectedStr0) || packingStr.equals(expectedStr1));

    packing.put("2", Arrays.asList("2:spout:2:1"));
    packingStr = generatePacking(packing).getInstanceDistribution();

    for (String str : packingStr.split(",")) {
      if (str.startsWith("1:")) {
        // This is the packing str for container 1
        Assert.assertTrue(str.equals(expectedStr0) || str.equals(expectedStr1));
      } else if (str.startsWith("2:")) {
        // This is the packing str for container 2
        Assert.assertEquals("2:spout:2:1", str);
      } else {
        // Unexpected container string
        throw new RuntimeException("Unexpected results");
      }
    }
  }
}

@@ -867,21 +867,22 @@
 st %7B%0A  p
-ublic
+rivate
  static 
@@ -1474,16 +1474,292 @@
 :%22)%5B1%5D;%0A
+        Resource instanceResource;%0A        if (%22bolt%22.equals(componentName)) %7B%0A          instanceResource = new Resource(1.0, 2 * Constants.GB, 10 * Constants.GB);%0A        %7D else %7B%0A          instanceResource = new Resource(1.0, 3 * Constants.GB, 10 * Constants.GB);%0A        %7D%0A
         
@@ -1866,17 +1866,25 @@
 ntName, 
-r
+instanceR
 esource)
@@ -2380,84 +2380,8 @@
 ));%0A
-    String packingStr = generatePacking(packing).getInstanceDistribution();%0A
     
@@ -2477,16 +2477,136 @@
 :1:0%22;%0A%0A
+    PackingPlan packingPlan = generatePacking(packing);%0A    String packingStr = packingPlan.getInstanceDistribution();%0A%0A
     Asse
@@ -2752,19 +2752,20 @@
  packing
-Str
+Plan
  = gener
@@ -2783,16 +2783,46 @@
 packing)
+;%0A    packingStr = packingPlan
 .getInst
@@ -2858,19 +2858,25 @@
 (String 
-str
+component
  : packi
@@ -2901,27 +2901,33 @@
 %7B%0A      if (
-str
+component
 .startsWith(
@@ -3012,19 +3012,25 @@
 ertTrue(
-str
+component
 .equals(
@@ -3046,19 +3046,25 @@
 tr0) %7C%7C 
-str
+component
 .equals(
@@ -3096,19 +3096,25 @@
 lse if (
-str
+component
 .startsW
@@ -3220,19 +3220,25 @@
 t:2:1%22, 
-str
+component
 );%0A     
@@ -3325,44 +3325,264 @@
 ion(
-%22Unexpected results%22);%0A      %7D%0A    %7D
+String.format(%0A            %22Unexpected component id found in instance distribution: %25s%22, component));%0A      %7D%0A    %7D%0A%0A    String ramDistStr = packingPlan.getComponentRamDistribution();%0A    Assert.assertEquals(%22spout:3221225472,bolt:2147483648%22, ramDistStr);
 %0A  %7D

Improve PackingPlanTest (#1214)