@Test
public void testSingleInputStreamStatusToggling() {

	BufferedValveOutputHandler valveOutput = new BufferedValveOutputHandler();
	StatusWatermarkValve valve = new StatusWatermarkValve(1, valveOutput);
	valve.inputStreamStatus(StreamStatus.ACTIVE, 0);
	// this also implicitly verifies that input channels start as ACTIVE
	assertEquals(null, valveOutput.popLastSeenOutput());
	valve.inputStreamStatus(StreamStatus.IDLE, 0);
	assertEquals(StreamStatus.IDLE, valveOutput.popLastSeenOutput());
	valve.inputStreamStatus(StreamStatus.IDLE, 0);
	assertEquals(null, valveOutput.popLastSeenOutput());
	valve.inputStreamStatus(StreamStatus.ACTIVE, 0);
	assertEquals(StreamStatus.ACTIVE, valveOutput.popLastSeenOutput());
	assertEquals(null, valveOutput.popLastSeenOutput());
}