public static void adjustImageConfigForTopo(Map<String, Object> conf, Map<String, Object> topoConf, String topoId)
        throws InvalidTopologyException {

        //don't need sanity check here as we assume it's already done during daemon startup
        List<String> allowedImages = getAllowedImages(conf, false);
        String topoImage = (String) topoConf.get(Config.TOPOLOGY_OCI_IMAGE);

        if (allowedImages.isEmpty()) {
            if (topoImage != null) {
                LOG.warn("{} is not configured; this indicates OCI container is not supported; "
                        + "{} config for topology {} will be removed",
                    DaemonConfig.STORM_OCI_ALLOWED_IMAGES, Config.TOPOLOGY_OCI_IMAGE, topoId);
                topoConf.remove(Config.TOPOLOGY_OCI_IMAGE);
            }
        } else {
            if (topoImage == null) {
                //we assume the default image is already validated during daemon startup
                String defaultImage = (String) conf.get(DaemonConfig.STORM_OCI_IMAGE);
                topoImage = defaultImage;
                topoConf.put(Config.TOPOLOGY_OCI_IMAGE, topoImage);
                LOG.info("{} is not set for topology {}; set it to the default image {} configured in {}",
                    Config.TOPOLOGY_OCI_IMAGE, topoId, defaultImage, DaemonConfig.STORM_OCI_IMAGE);
            } else {
                try {
                    validateImage(allowedImages, topoImage, Config.TOPOLOGY_OCI_IMAGE);
                } catch (IllegalArgumentException e) {
                    throw new WrappedInvalidTopologyException(e.getMessage());
                }
            }
        }
    }
@Test
    public void adjustImageConfigForTopoTest() throws InvalidTopologyException {
        Map<String, Object> conf = new HashMap<>();
        conf.put(DaemonConfig.STORM_OCI_ALLOWED_IMAGES, null); //or not set

        Map<String, Object> topoConf = new HashMap<>();
        String topoId = "topo1";

        //case 1: nothing is not; nothing will happen
        OciUtils.adjustImageConfigForTopo(conf, topoConf, topoId);

        String image1 = "storm/rhel7:dev_test";
        String defaultImage = "storm/rhel7:dev_current";

        //case 2: allowed list is not set; topology oci image will be set to null
        topoConf.put(Config.TOPOLOGY_OCI_IMAGE, image1);
        OciUtils.adjustImageConfigForTopo(conf, topoConf, topoId);
        assertNull(topoConf.get(Config.TOPOLOGY_OCI_IMAGE), Config.TOPOLOGY_OCI_IMAGE + " is not removed");

        //set up daemon conf properly
        List<String> allowedImages = new ArrayList<>();
        allowedImages.add(image1);
        allowedImages.add(defaultImage);
        conf.put(DaemonConfig.STORM_OCI_ALLOWED_IMAGES, allowedImages);
        conf.put(DaemonConfig.STORM_OCI_IMAGE, defaultImage);

        //case 3: configs are set properly; nothing will happen
        topoConf.put(Config.TOPOLOGY_OCI_IMAGE, image1);
        OciUtils.adjustImageConfigForTopo(conf, topoConf, topoId);
        assertEquals(image1, topoConf.get(Config.TOPOLOGY_OCI_IMAGE));

        //case 4: topology oci image is not set; will be set to default image
        topoConf.remove(Config.TOPOLOGY_OCI_IMAGE);
        OciUtils.adjustImageConfigForTopo(conf, topoConf, topoId);
        assertEquals(defaultImage, topoConf.get(Config.TOPOLOGY_OCI_IMAGE));

        //case 5: any topology oci image is allowed
        allowedImages.add("*");
        String image2 = "storm/rhel7:dev_wow";
        topoConf.put(Config.TOPOLOGY_OCI_IMAGE, image2);
        OciUtils.adjustImageConfigForTopo(conf, topoConf, topoId);
        assertEquals(image2, topoConf.get(Config.TOPOLOGY_OCI_IMAGE));
    }