@Test
public void testDefaultPartitioning()
{

  LogicalPlan dag = new LogicalPlan();
  dag.setAttribute(OperatorContext.STORAGE_AGENT, new StramTestSupport.MemoryStorageAgent());
  GenericTestOperator node1 = dag.addOperator("node1", GenericTestOperator.class);
  GenericTestOperator node2 = dag.addOperator("node2", GenericTestOperator.class);
  dag.addStream("node1.outport1", node1.outport1, node2.inport2, node2.inport1);
  int initialPartitionCount = 5;
  OperatorMeta node2Decl = dag.getMeta(node2);
  node2Decl.getAttributes().put(OperatorContext.PARTITIONER, new StatelessPartitioner<GenericTestOperator>(initialPartitionCount));
  PhysicalPlan plan = new PhysicalPlan(dag, new TestPlanContext());
  List<PTOperator> n2Instances = plan.getOperators(node2Decl);
  Assert.assertEquals("partition instances " + n2Instances, initialPartitionCount, n2Instances.size());
  List<Integer> assignedPartitionKeys = Lists.newArrayList();
  for (int i = 0; i < n2Instances.size(); i++) {
    PTOperator n2Partition = n2Instances.get(i);
    Assert.assertNotNull("partition keys null: " + n2Partition, n2Partition.getPartitionKeys());
    Map<InputPort<?>, PartitionKeys> pkeys = n2Partition.getPartitionKeys();
    Assert.assertEquals("partition keys size: " + pkeys, 1, pkeys.size()); // one port partitioned
    InputPort<?> expectedPort = node2.inport2;
    Assert.assertEquals("partition port: " + pkeys, expectedPort, pkeys.keySet().iterator().next());
    Assert.assertEquals("partition mask: " + pkeys, "111", Integer.toBinaryString(pkeys.get(expectedPort).mask));
    Set<Integer> pks = pkeys.get(expectedPort).partitions;
    Assert.assertTrue("number partition keys: " + pkeys, pks.size() == 1 || pks.size() == 2);
    assignedPartitionKeys.addAll(pks);
  }
  int expectedMask = Integer.parseInt("111", 2);
  Assert.assertEquals("assigned partitions ", expectedMask + 1, assignedPartitionKeys.size());
  for (int i = 0; i <= expectedMask; i++) {
    Assert.assertTrue("" + assignedPartitionKeys, assignedPartitionKeys.contains(i));
  }
}