@Test
public void testConstructFilters() throws Exception
{

  TestSetup t = new TestSetup();
  //test single Physical Partition subscription
  DatabusSubscription sub1 =
      DatabusSubscription.createPhysicalPartitionReplicationSubscription(new PhysicalPartition(100, "multBufferTest1"));
  DbusFilter filter1 = t._eventBuffer.constructFilters(Arrays.asList(sub1));
  assertNotNull(filter1);
  assertTrue(filter1 instanceof PhysicalPartitionDbusFilter);
  PhysicalPartitionDbusFilter ppfilter1 = (PhysicalPartitionDbusFilter)filter1;
  assertEquals(ppfilter1.getPhysicalPartition(), new PhysicalPartition(100, "multBufferTest1"));
  assertNull(ppfilter1.getNestedFilter());
  DatabusSubscription sub2 =
      DatabusSubscription.createPhysicalPartitionReplicationSubscription(new PhysicalPartition(101, "multBufferTest2"));
  //test two Physical Partition subscriptions
  DbusFilter filter2 = t._eventBuffer.constructFilters(Arrays.asList(sub1, sub2));
  assertNotNull(filter2);
  assertTrue(filter2 instanceof ConjunctionDbusFilter);
  ConjunctionDbusFilter conjFilter2 = (ConjunctionDbusFilter)filter2;
  boolean hasPP100 = false;
  boolean hasPP101 = false;
  assertEquals(conjFilter2.getFilterList().size(), 2);
  for (DbusFilter f: conjFilter2.getFilterList())
  {
    assertTrue(f instanceof PhysicalPartitionDbusFilter);
    PhysicalPartitionDbusFilter ppf = (PhysicalPartitionDbusFilter)f;
    if (ppf.getPhysicalPartition().getId() == 100) hasPP100 = true;
    else if (ppf.getPhysicalPartition().getId() == 101) hasPP101 = true;
    else fail("unknown physical partition filter:" + ppf.getPhysicalPartition());
  }
  assertTrue(hasPP100);
  assertTrue(hasPP101);
  //test a subcription with a logical source
  DatabusSubscription sub3 =
      DatabusSubscription.createSimpleSourceSubscription(new LogicalSource(2, "srcName2"));
  DbusFilter filter3 = t._eventBuffer.constructFilters(Arrays.asList(sub3));
  assertNotNull(filter3);
  assertTrue(filter3 instanceof PhysicalPartitionDbusFilter);
  PhysicalPartitionDbusFilter ppfilter3 = (PhysicalPartitionDbusFilter)filter3;
  assertEquals(ppfilter3.getPhysicalPartition(), PhysicalPartition.ANY_PHYSICAL_PARTITION);
  DbusFilter ppfilter3_child = ppfilter3.getNestedFilter();
  assertNotNull(ppfilter3_child);
  assertTrue(ppfilter3_child instanceof LogicalSourceAndPartitionDbusFilter);
  LogicalSourceAndPartitionDbusFilter lsourceFilter3 = (LogicalSourceAndPartitionDbusFilter)ppfilter3_child;
  LogicalSourceAndPartitionDbusFilter.LogicalPartitionDbusFilter lpartFilter3_1 =
      lsourceFilter3.getSourceFilter(2);
  assertNotNull(lpartFilter3_1);
  assertTrue(lpartFilter3_1.isAllPartitionsWildcard());
  //test a subcription with a physical and logical partition
  DatabusSubscription sub4 =
      new DatabusSubscription(PhysicalSource.MASTER_PHISYCAL_SOURCE,
                              new PhysicalPartition(101, "multBufferTest2"),
                              new LogicalSourceId(new LogicalSource(2, "srcName2"), (short)2)
                              );
  DbusFilter filter4 = t._eventBuffer.constructFilters(Arrays.asList(sub4));
  assertNotNull(filter4);
  assertTrue(filter4 instanceof PhysicalPartitionDbusFilter);
  PhysicalPartitionDbusFilter ppfilter4 = (PhysicalPartitionDbusFilter)filter4;
  assertEquals(ppfilter4.getPhysicalPartition(), new PhysicalPartition(101, "multBufferTest2"));
  DbusFilter ppfilter4_child = ppfilter4.getNestedFilter();
  assertNotNull(ppfilter4_child);
  assertTrue(ppfilter4_child instanceof LogicalSourceAndPartitionDbusFilter);
  LogicalSourceAndPartitionDbusFilter lsourceFilter4 = (LogicalSourceAndPartitionDbusFilter)ppfilter4_child;
  LogicalSourceAndPartitionDbusFilter.LogicalPartitionDbusFilter lpartFilter4_1 =
      lsourceFilter4.getSourceFilter(2);
  assertNotNull(lpartFilter4_1);
  assertTrue(lpartFilter4_1.getPartitionsMask().contains(2));
}