<commit_before>/**
 * Copyright (c) 2012-2012 Malhar, Inc.
 * All rights reserved.
 */
package com.malhartech.stram;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.junit.Assert;
import org.junit.Test;

import com.google.common.collect.Sets;
import com.malhartech.annotation.InputPortFieldAnnotation;
import com.malhartech.annotation.ModuleAnnotation;
import com.malhartech.annotation.OutputPortFieldAnnotation;
import com.malhartech.annotation.PortAnnotation;
import com.malhartech.annotation.PortAnnotation.PortType;
import com.malhartech.api.BaseOperator;
import com.malhartech.api.DAG;
import com.malhartech.api.DefaultInputPort;
import com.malhartech.api.DefaultOutputPort;
import com.malhartech.api.DAG.OperatorWrapper;
import com.malhartech.api.DAG.StreamDecl;
import com.malhartech.api.Operator;
import com.malhartech.api.Operator.InputPort;
import com.malhartech.api.Sink;
import com.malhartech.dag.DefaultSerDe;
import com.malhartech.dag.GenericTestModule;
import com.malhartech.stram.cli.StramClientUtils;

public class DAGBuilderTest {

  public static OperatorWrapper assertNode(DAG tplg, String id) {
      OperatorWrapper n = tplg.getOperatorWrapper(id);
      assertNotNull("operator exists id=" + id, n);
      return n;
  }

  /**
   * Test read from stram-site.xml in Hadoop configuration format.
   */
  @Test
  public void testLoadFromConfigXml() {
    Configuration conf = StramClientUtils.addStramResources(new Configuration());
    //Configuration.dumpConfiguration(conf, new PrintWriter(System.out));

    DAGPropertiesBuilder tb = new DAGPropertiesBuilder();
    tb.addFromConfiguration(conf);

    DAG dag = tb.getApplication(new Configuration(false));
    dag.validate();

//    Map<String, NodeConf> moduleConfs = tb.getAllOperators();
    assertEquals("number of module confs", 6, dag.getAllOperators().size());

    OperatorWrapper module1 = assertNode(dag, "module1");
    OperatorWrapper module2 = assertNode(dag, "module2");
    OperatorWrapper module3 = assertNode(dag, "module3");
    OperatorWrapper module4 = assertNode(dag, "module4");

    assertNotNull("moduleConf for root", module1);
    assertEquals("moduleId set", "module1", module1.getId());

    // verify module instantiation
    assertEquals(module1.getModule().getClass(), GenericTestModule.class);
    GenericTestModule GenericTestNode = (GenericTestModule)module1.getModule();
    assertEquals("myStringPropertyValue", GenericTestNode.getMyStringProperty());

    // check links
    assertEquals("module1 inputs", 0, module1.getInputStreams().size());
    assertEquals("module1 outputs", 1, module1.getOutputStreams().size());
    StreamDecl n1n2 = module2.getInputStreams().get(module2.getInputPortMeta(((GenericTestModule)module2.getModule()).inport1));
    assertNotNull("n1n2", n1n2);

    // output/input stream object same
    assertEquals("rootNode out is module2 in", n1n2, module1.getOutputStreams().get(module1.getOutputPortMeta(((GenericTestModule)module1.getModule()).outport1)));
    assertEquals("n1n2 source", module1, n1n2.getSource().getOperator());
    Assert.assertEquals("n1n2 targets", 1, n1n2.getSinks().size());
    Assert.assertEquals("n1n2 target", module2, n1n2.getSinks().get(0).getOperator());

    assertEquals("stream name", "n1n2", n1n2.getId());
    Assert.assertFalse("n1n2 not inline (default)", n1n2.isInline());

    // module 2 streams to module 3 and module 4
    assertEquals("module 2 number of outputs", 1, module2.getOutputStreams().size());
    StreamDecl fromNode2 = module2.getOutputStreams().values().iterator().next();

    Set<OperatorWrapper> targetNodes = new HashSet<OperatorWrapper>();
    for (DAG.InputPortMeta ip : fromNode2.getSinks()) {
      targetNodes.add(ip.getOperator());
    }
    Assert.assertEquals("outputs " + fromNode2, Sets.newHashSet(module3, module4), targetNodes);

    OperatorWrapper module6 = assertNode(dag, "module6");

    List<OperatorWrapper> rootNodes = dag.getRootOperators();
    assertEquals("number root modules", 2, rootNodes.size());
    assertTrue("root module2", rootNodes.contains(module1));
    assertTrue("root module6", rootNodes.contains(module6));

    for (OperatorWrapper n : rootNodes) {
      printTopology(n, dag, 0);
    }

  }

  public void printTopology(OperatorWrapper module, DAG tplg, int level) {
      String prefix = "";
      if (level > 0) {
        prefix = StringUtils.repeat(" ", 20*(level-1)) + "   |" + StringUtils.repeat("-", 17);
      }
      System.out.println(prefix + module.getId());
      for (StreamDecl downStream : module.getOutputStreams().values()) {
          if (!downStream.getSinks().isEmpty()) {
            for (DAG.InputPortMeta targetNode : downStream.getSinks()) {
              printTopology(targetNode.getOperator(), tplg, level+1);
            }
          }
      }
  }

  @Test
  public void testLoadFromPropertiesFile() throws IOException {
      Properties props = new Properties();
      String resourcePath = "/testTopology.properties";
      InputStream is = this.getClass().getResourceAsStream(resourcePath);
      if (is == null) {
        fail("Could not load " + resourcePath);
      }
      props.load(is);
      DAGPropertiesBuilder pb = new DAGPropertiesBuilder()
        .addFromProperties(props);

      DAG dag = pb.getApplication(new Configuration(false));
      dag.validate();

      assertEquals("number of module confs", 5, dag.getAllOperators().size());
      assertEquals("number of root modules", 1, dag.getRootOperators().size());

      StreamDecl s1 = dag.getStream("n1n2");
      assertNotNull(s1);
      assertTrue("n1n2 inline", s1.isInline());

      OperatorWrapper module3 = dag.getOperatorWrapper("module3");
      assertEquals("module3.classname", GenericTestModule.class.getName(), module3.getModule().getClass());

      GenericTestModule dmodule3 = (GenericTestModule)module3.getModule();
      assertEquals("module3.myStringProperty", "myStringPropertyValueFromTemplate", dmodule3.getMyStringProperty());
      assertFalse("module3.booleanProperty", dmodule3.booleanProperty);

      OperatorWrapper module4 = dag.getOperatorWrapper("module4");
      GenericTestModule dmodule4 = (GenericTestModule)module4.getModule();
      assertEquals("module4.myStringProperty", "overrideModule4", dmodule4.getMyStringProperty());
      assertTrue("module4.booleanProperty", dmodule4.booleanProperty);

      StreamDecl input1 = dag.getStream("inputStream");
      assertNotNull(input1);
      Assert.assertEquals("input1 source", dag.getOperatorWrapper("inputModule"), input1.getSource().getOperator());
      Set<OperatorWrapper> targetNodes = new HashSet<OperatorWrapper>();
      for (DAG.InputPortMeta targetPort : input1.getSinks()) {
        targetNodes.add(targetPort.getOperator());
      }<commit_msg>fixed newly introduced errors in DAGBuilderTest due to parameterized StreamDecl.<commit_after>/**
 * Copyright (c) 2012-2012 Malhar, Inc.
 * All rights reserved.
 */
package com.malhartech.stram;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.junit.Assert;
import org.junit.Test;

import com.google.common.collect.Sets;
import com.malhartech.annotation.InputPortFieldAnnotation;
import com.malhartech.annotation.ModuleAnnotation;
import com.malhartech.annotation.OutputPortFieldAnnotation;
import com.malhartech.annotation.PortAnnotation;
import com.malhartech.annotation.PortAnnotation.PortType;
import com.malhartech.api.BaseOperator;
import com.malhartech.api.DAG;
import com.malhartech.api.DefaultInputPort;
import com.malhartech.api.DefaultOutputPort;
import com.malhartech.api.DAG.OperatorWrapper;
import com.malhartech.api.DAG.StreamDecl;
import com.malhartech.api.Operator;
import com.malhartech.api.Operator.InputPort;
import com.malhartech.api.Sink;
import com.malhartech.dag.DefaultSerDe;
import com.malhartech.dag.GenericTestModule;
import com.malhartech.stram.cli.StramClientUtils;

public class DAGBuilderTest {

  public static OperatorWrapper assertNode(DAG tplg, String id) {
      OperatorWrapper n = tplg.getOperatorWrapper(id);
      assertNotNull("operator exists id=" + id, n);
      return n;
  }

  /**
   * Test read from stram-site.xml in Hadoop configuration format.
   */
  @Test
  public void testLoadFromConfigXml() {
    Configuration conf = StramClientUtils.addStramResources(new Configuration());
    //Configuration.dumpConfiguration(conf, new PrintWriter(System.out));

    DAGPropertiesBuilder tb = new DAGPropertiesBuilder();
    tb.addFromConfiguration(conf);

    DAG dag = tb.getApplication(new Configuration(false));
    dag.validate();

//    Map<String, NodeConf> moduleConfs = tb.getAllOperators();
    assertEquals("number of module confs", 6, dag.getAllOperators().size());

    OperatorWrapper module1 = assertNode(dag, "module1");
    OperatorWrapper module2 = assertNode(dag, "module2");
    OperatorWrapper module3 = assertNode(dag, "module3");
    OperatorWrapper module4 = assertNode(dag, "module4");

    assertNotNull("moduleConf for root", module1);
    assertEquals("moduleId set", "module1", module1.getId());

    // verify module instantiation
    assertEquals(module1.getModule().getClass(), GenericTestModule.class);
    GenericTestModule GenericTestNode = (GenericTestModule)module1.getModule();
    assertEquals("myStringPropertyValue", GenericTestNode.getMyStringProperty());

    // check links
    assertEquals("module1 inputs", 0, module1.getInputStreams().size());
    assertEquals("module1 outputs", 1, module1.getOutputStreams().size());
    StreamDecl<?> n1n2 = module2.getInputStreams().get(module2.getInputPortMeta(((GenericTestModule)module2.getModule()).inport1));
    assertNotNull("n1n2", n1n2);

    // output/input stream object same
    assertEquals("rootNode out is module2 in", n1n2, module1.getOutputStreams().get(module1.getOutputPortMeta(((GenericTestModule)module1.getModule()).outport1)));
    assertEquals("n1n2 source", module1, n1n2.getSource().getOperator());
    Assert.assertEquals("n1n2 targets", 1, n1n2.getSinks().size());
    Assert.assertEquals("n1n2 target", module2, n1n2.getSinks().get(0).getOperator());

    assertEquals("stream name", "n1n2", n1n2.getId());
    Assert.assertFalse("n1n2 not inline (default)", n1n2.isInline());

    // module 2 streams to module 3 and module 4
    assertEquals("module 2 number of outputs", 1, module2.getOutputStreams().size());
    StreamDecl<?> fromNode2 = module2.getOutputStreams().values().iterator().next();

    Set<OperatorWrapper> targetNodes = new HashSet<OperatorWrapper>();
    for (DAG.InputPortMeta ip : fromNode2.getSinks()) {
      targetNodes.add(ip.getOperator());
    }
    Assert.assertEquals("outputs " + fromNode2, Sets.newHashSet(module3, module4), targetNodes);

    OperatorWrapper module6 = assertNode(dag, "module6");

    List<OperatorWrapper> rootNodes = dag.getRootOperators();
    assertEquals("number root modules", 2, rootNodes.size());
    assertTrue("root module2", rootNodes.contains(module1));
    assertTrue("root module6", rootNodes.contains(module6));

    for (OperatorWrapper n : rootNodes) {
      printTopology(n, dag, 0);
    }

  }

  public void printTopology(OperatorWrapper module, DAG tplg, int level) {
      String prefix = "";
      if (level > 0) {
        prefix = StringUtils.repeat(" ", 20*(level-1)) + "   |" + StringUtils.repeat("-", 17);
      }
      System.out.println(prefix + module.getId());
      for (StreamDecl<?> downStream : module.getOutputStreams().values()) {
          if (!downStream.getSinks().isEmpty()) {
            for (DAG.InputPortMeta targetNode : downStream.getSinks()) {
              printTopology(targetNode.getOperator(), tplg, level+1);
            }
          }
      }
  }

  @Test
  public void testLoadFromPropertiesFile() throws IOException {
      Properties props = new Properties();
      String resourcePath = "/testTopology.properties";
      InputStream is = this.getClass().getResourceAsStream(resourcePath);
      if (is == null) {
        fail("Could not load " + resourcePath);
      }
      props.load(is);
      DAGPropertiesBuilder pb = new DAGPropertiesBuilder()
        .addFromProperties(props);

      DAG dag = pb.getApplication(new Configuration(false));
      dag.validate();

      assertEquals("number of module confs", 5, dag.getAllOperators().size());
      assertEquals("number of root modules", 1, dag.getRootOperators().size());

      StreamDecl s1 = dag.getStream("n1n2");
      assertNotNull(s1);
      assertTrue("n1n2 inline", s1.isInline());

      OperatorWrapper module3 = dag.getOperatorWrapper("module3");
      assertEquals("module3.classname", GenericTestModule.class.getName(), module3.getModule().getClass());

      GenericTestModule dmodule3 = (GenericTestModule)module3.getModule();
      assertEquals("module3.myStringProperty", "myStringPropertyValueFromTemplate", dmodule3.getMyStringProperty());
      assertFalse("module3.booleanProperty", dmodule3.booleanProperty);

      OperatorWrapper module4 = dag.getOperatorWrapper("module4");
      GenericTestModule dmodule4 = (GenericTestModule)module4.getModule();
      assertEquals("module4.myStringProperty", "overrideModule4", dmodule4.getMyStringProperty());
      assertTrue("module4.booleanProperty", dmodule4.booleanProperty);

      StreamDecl<?> input1 = dag.getStream("inputStream");
      assertNotNull(input1);
      Assert.assertEquals("input1 source", dag.getOperatorWrapper("inputModule"), input1.getSource().getOperator());
      Set<OperatorWrapper> targetNodes = new HashSet<OperatorWrapper>();
      for (DAG.InputPortMeta targetPort : input1.getSinks()) {
        targetNodes.add(targetPort.getOperator());
      }
{'fraction_non_alphanumeric': 0.12658485321038715, 'fraction_numerical': 0.01749271137026239, 'mean_word_length': 4.610498288322556, 'pattern_counts': {'":': 0, '<': 19, '<?xml version=': 0, '>': 21, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 2}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6344462', 'n_tokens_mistral': 4262, 'n_tokens_neox': 4207, 'n_words': 841}