    @Test
    public void testTubeBaseSessionFactory() throws Exception {

        TubeClientConfig config = mock(TubeClientConfig.class);
        when(config.getMasterInfo()).thenReturn(new MasterInfo("192.168.1.1:18080"));

        ClientFactory clientFactory = mock(ClientFactory.class);
        PowerMockito.mockStatic(AddressUtils.class);
        PowerMockito.when(AddressUtils.getLocalAddress()).thenReturn("127.0.0.1:18080");

        TubeBaseSessionFactory factory = new TubeBaseSessionFactory(clientFactory, config);
        assertFalse(factory.isShutdown());

        // Use reflection to change the producer manager to a mock one
        ProducerManager mockPM = mock(ProducerManager.class);
        Field field = factory.getClass().getDeclaredField("producerManager");
        field.setAccessible(true);
        field.set(factory, mockPM);
        MessageProducer producer = factory.createProducer();
        assertEquals(1, factory.getCurrClients().size());

        final PullMessageConsumer pullMessageConsumer = factory.createPullConsumer(
                new ConsumerConfig("192.168.1.1:18080", "test"));
        final PushMessageConsumer pushMessageConsumer = factory.createPushConsumer(
                new ConsumerConfig("192.168.1.1:18080", "test"));

        assertEquals(3, factory.getCurrClients().size());
        factory.removeClient(producer);
        factory.removeClient(pullMessageConsumer);
        factory.removeClient(pushMessageConsumer);
        assertEquals(0, factory.getCurrClients().size());

        factory.shutdown();
        assertTrue(factory.isShutdown());
    }