@Test
public void testLoadFromConfigXml()
{

  Configuration conf = new Configuration(false);
  conf.addResource(StramClientUtils.DT_SITE_XML_FILE);
  LogicalPlanConfiguration builder = new LogicalPlanConfiguration(conf);
  LogicalPlan dag = new LogicalPlan();
  builder.populateDAG(dag);
  dag.validate();
  assertEquals("number of operator confs", 6, dag.getAllOperators().size());
  OperatorMeta operator1 = assertNode(dag, "operator1");
  OperatorMeta operator2 = assertNode(dag, "operator2");
  OperatorMeta operator3 = assertNode(dag, "operator3");
  OperatorMeta operator4 = assertNode(dag, "operator4");
  assertNotNull("operatorConf for root", operator1);
  assertEquals("operatorId set", "operator1", operator1.getName());
  // verify operator instantiation
  assertEquals(operator1.getOperator().getClass(), TestGeneratorInputOperator.class);
  TestGeneratorInputOperator GenericTestNode = (TestGeneratorInputOperator)operator1.getOperator();
  assertEquals("myStringPropertyValue", GenericTestNode.getMyStringProperty());
  // check links
  assertEquals("operator1 inputs", 0, operator1.getInputStreams().size());
  assertEquals("operator1 outputs", 1, operator1.getOutputStreams().size());
  StreamMeta n1n2 = operator2.getInputStreams().get(operator2.getMeta(((GenericTestOperator)operator2.getOperator()).inport1));
  assertNotNull("n1n2", n1n2);
  // output/input stream object same
  assertEquals("rootNode out is operator2 in", n1n2, operator1.getOutputStreams().get(operator1.getMeta(((TestGeneratorInputOperator)operator1.getOperator()).outport)));
  assertEquals("n1n2 source", operator1, n1n2.getSource().getOperatorMeta());
  Assert.assertEquals("n1n2 targets", 1, n1n2.getSinks().size());
  Assert.assertEquals("n1n2 target", operator2, n1n2.getSinks().iterator().next().getOperatorMeta());
  assertEquals("stream name", "n1n2", n1n2.getName());
  Assert.assertEquals("n1n2 not inline (default)", null, n1n2.getLocality());
  // operator 2 streams to operator 3 and operator 4
  assertEquals("operator 2 number of outputs", 1, operator2.getOutputStreams().size());
  StreamMeta fromNode2 = operator2.getOutputStreams().values().iterator().next();
  Set<OperatorMeta> targetNodes = Sets.newHashSet();
  for (LogicalPlan.InputPortMeta ip : fromNode2.getSinks()) {
    targetNodes.add(ip.getOperatorMeta());
  }
  Assert.assertEquals("outputs " + fromNode2, Sets.newHashSet(operator3, operator4), targetNodes);
  OperatorMeta operator6 = assertNode(dag, "operator6");
  List<OperatorMeta> rootNodes = dag.getRootOperators();
  assertEquals("number root operators", 2, rootNodes.size());
  assertTrue("root operator2", rootNodes.contains(operator1));
  assertTrue("root operator6", rootNodes.contains(operator6));
  for (OperatorMeta n : rootNodes) {
    printTopology(n, dag, 0);
  }
}