@VisibleForTesting
    CompletableFuture<Void> requestDownloadBaseTopologyBlobs(PortAndAssignment pna, BlobChangingCallback cb) {
        final String topologyId = pna.getToplogyId();

        final LocallyCachedBlob topoJar = getTopoJar(topologyId, pna.getAssignment().get_owner());
        topoJar.addReference(pna, cb);

        final LocallyCachedBlob topoCode = getTopoCode(topologyId, pna.getAssignment().get_owner());
        topoCode.addReference(pna, cb);

        final LocallyCachedBlob topoConf = getTopoConf(topologyId, pna.getAssignment().get_owner());
        topoConf.addReference(pna, cb);

        return topologyBasicDownloaded.computeIfAbsent(topologyId,
            (tid) -> downloadOrUpdate(topoJar, topoCode, topoConf));
    }
@Test
    public void testRequestDownloadBaseTopologyBlobs() throws Exception {
        ReflectionUtils mockedReflectionUtils = mock(ReflectionUtils.class);
        ServerUtils mockedServerUtils = mock(ServerUtils.class);

        ReflectionUtils previousReflectionUtils = ReflectionUtils.setInstance(mockedReflectionUtils);
        ServerUtils previousServerUtils = ServerUtils.setInstance(mockedServerUtils);

        // cannot use automatic resource management here in this try because the AsyncLocalizer depends on a config map,
        // which should take the storm local dir, and that storm local dir is declared in the try-with-resources.
        AsyncLocalizer victim = null;

        try (TmpPath stormRoot = new TmpPath(); TmpPath localizerRoot = new TmpPath()) {

            Map<String, Object> conf = new HashMap<>();
            conf.put(DaemonConfig.SUPERVISOR_BLOBSTORE, ClientBlobStore.class.getName());
            conf.put(Config.STORM_PRINCIPAL_TO_LOCAL_PLUGIN, DefaultPrincipalToLocal.class.getName());
            conf.put(Config.STORM_CLUSTER_MODE, "distributed");
            conf.put(Config.STORM_LOCAL_DIR, stormRoot.getPath());

            AdvancedFSOps ops = AdvancedFSOps.make(conf);

            victim = spy(new AsyncLocalizer(conf, ops, localizerRoot.getPath(), new StormMetricsRegistry()));

            final String topoId = "TOPO";

            final LocalAssignment localAssignment = constructLocalAssignment(topoId, "user");
            final int port = 8080;

            ClientBlobStore blobStore = mock(ClientBlobStore.class);
            when(blobStore.getRemoteBlobstoreUpdateTime()).thenReturn(-1L);

            LocallyCachedTopologyBlob jarBlob = mock(LocallyCachedTopologyBlob.class);
            doReturn(jarBlob).when(victim).getTopoJar(topoId, localAssignment.get_owner());
            when(jarBlob.getLocalVersion()).thenReturn(-1L);
            when(jarBlob.getRemoteVersion(any())).thenReturn(100L);
            when(jarBlob.fetchUnzipToTemp(any())).thenReturn(100L);
            when(jarBlob.isUsed()).thenReturn(true);

            LocallyCachedTopologyBlob codeBlob = mock(LocallyCachedTopologyBlob.class);
            doReturn(codeBlob).when(victim).getTopoCode(topoId, localAssignment.get_owner());
            when(codeBlob.getLocalVersion()).thenReturn(-1L);
            when(codeBlob.getRemoteVersion(any())).thenReturn(200L);
            when(codeBlob.fetchUnzipToTemp(any())).thenReturn(200L);
            when(codeBlob.isUsed()).thenReturn(true);

            LocallyCachedTopologyBlob confBlob = mock(LocallyCachedTopologyBlob.class);
            doReturn(confBlob).when(victim).getTopoConf(topoId, localAssignment.get_owner());
            when(confBlob.getLocalVersion()).thenReturn(-1L);
            when(confBlob.getRemoteVersion(any())).thenReturn(300L);
            when(confBlob.fetchUnzipToTemp(any())).thenReturn(300L);
            when(confBlob.isUsed()).thenReturn(true);

            when(mockedReflectionUtils.newInstanceImpl(ClientBlobStore.class)).thenReturn(blobStore);

            PortAndAssignment pna = new PortAndAssignmentImpl(port, localAssignment);
            Future<Void> f = victim.requestDownloadBaseTopologyBlobs(pna, null);
            f.get(20, TimeUnit.SECONDS);

            verify(jarBlob).update(eq(blobStore), eq(-1L));
            verify(codeBlob).update(eq(blobStore), eq(-1L));
            verify(confBlob).update(eq(blobStore), eq(-1L));
        } finally {
            ReflectionUtils.setInstance(previousReflectionUtils);
            ServerUtils.setInstance(previousServerUtils);

            if (victim != null) {
                victim.close();
            }
        }
    }