@Test
public void testMetadata() throws Exception {

  KeyProvider.Metadata mockMeta = Mockito.mock(KeyProvider.Metadata.class);
  KeyProvider mockProv = Mockito.mock(KeyProvider.class);
  Mockito.when(mockProv.getMetadata(Mockito.eq("k1"))).thenReturn(mockMeta);
  Mockito.when(mockProv.getMetadata(Mockito.eq("k2"))).thenReturn(null);
  Mockito.when(mockProv.getConf()).thenReturn(new Configuration());
  KeyProvider cache = new CachingKeyProvider(mockProv, 100, 100);
  // asserting caching
  Assert.assertEquals(mockMeta, cache.getMetadata("k1"));
  Mockito.verify(mockProv, Mockito.times(1)).getMetadata(Mockito.eq("k1"));
  Assert.assertEquals(mockMeta, cache.getMetadata("k1"));
  Mockito.verify(mockProv, Mockito.times(1)).getMetadata(Mockito.eq("k1"));
  Thread.sleep(200);
  Assert.assertEquals(mockMeta, cache.getMetadata("k1"));
  Mockito.verify(mockProv, Mockito.times(2)).getMetadata(Mockito.eq("k1"));
  // asserting no caching when key is not known
  cache = new CachingKeyProvider(mockProv, 100, 100);
  Assert.assertEquals(null, cache.getMetadata("k2"));
  Mockito.verify(mockProv, Mockito.times(1)).getMetadata(Mockito.eq("k2"));
  Assert.assertEquals(null, cache.getMetadata("k2"));
  Mockito.verify(mockProv, Mockito.times(2)).getMetadata(Mockito.eq("k2"));
}