@SuppressWarnings("unchecked")
@Test
public void testOpen() throws Exception {

	// utility mocks
	final StormConfig stormConfig = new StormConfig();
	final Configuration flinkConfig = new Configuration();
	final ExecutionConfig taskConfig = mock(ExecutionConfig.class);
	when(taskConfig.getGlobalJobParameters()).thenReturn(null).thenReturn(stormConfig)
	.thenReturn(flinkConfig);
	final StreamingRuntimeContext taskContext = mock(StreamingRuntimeContext.class);
	when(taskContext.getExecutionConfig()).thenReturn(taskConfig);
	when(taskContext.getTaskName()).thenReturn("name");
	final SetupOutputFieldsDeclarer declarer = new SetupOutputFieldsDeclarer();
	declarer.declare(new Fields("dummy"));
	PowerMockito.whenNew(SetupOutputFieldsDeclarer.class).withNoArguments().thenReturn(declarer);
	// (1) open with no configuration
	{
		ExecutionConfig execConfig = mock(ExecutionConfig.class);
		when(execConfig.getGlobalJobParameters()).thenReturn(null);
		final IRichBolt bolt = mock(IRichBolt.class);
		BoltWrapper<Object, Object> wrapper = new BoltWrapper<Object, Object>(bolt);
		wrapper.setup(createMockStreamTask(execConfig), new StreamConfig(new Configuration()), mock(Output.class));
		wrapper.open();
		verify(bolt).prepare(any(Map.class), any(TopologyContext.class), any(OutputCollector.class));
	}
	// (2) open with a storm specific configuration
	{
		ExecutionConfig execConfig = mock(ExecutionConfig.class);
		when(execConfig.getGlobalJobParameters()).thenReturn(stormConfig);
		final IRichBolt bolt = mock(IRichBolt.class);
		BoltWrapper<Object, Object> wrapper = new BoltWrapper<Object, Object>(bolt);
		wrapper.setup(createMockStreamTask(execConfig), new StreamConfig(new Configuration()), mock(Output.class));
		wrapper.open();
		verify(bolt).prepare(same(stormConfig), any(TopologyContext.class), any(OutputCollector.class));
	}
	// (3) open with a flink config
	{
		final Configuration cfg = new Configuration();
		cfg.setString("foo", "bar");
		cfg.setInteger("the end (the int)", Integer.MAX_VALUE);
		ExecutionConfig execConfig = mock(ExecutionConfig.class);
		when(execConfig.getGlobalJobParameters()).thenReturn(new UnmodifiableConfiguration(cfg));
		TestDummyBolt testBolt = new TestDummyBolt();
		BoltWrapper<Object, Object> wrapper = new BoltWrapper<Object, Object>(testBolt);
		wrapper.setup(createMockStreamTask(execConfig), new StreamConfig(new Configuration()), mock(Output.class));
		wrapper.open();
		for (Entry<String, String> entry : cfg.toMap().entrySet()) {
			Assert.assertEquals(entry.getValue(), testBolt.config.get(entry.getKey()));
		}
	}
}