/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.nifi.controller;

import org.apache.nifi.admin.service.AuditService;
import org.apache.nifi.annotation.lifecycle.OnScheduled;
import org.apache.nifi.annotation.lifecycle.OnStopped;
import org.apache.nifi.annotation.lifecycle.OnUnscheduled;
import org.apache.nifi.authorization.Authorizer;
import org.apache.nifi.bundle.Bundle;
import org.apache.nifi.bundle.BundleCoordinate;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.components.PropertyValue;
import org.apache.nifi.components.ValidationContext;
import org.apache.nifi.components.state.StateManagerProvider;
import org.apache.nifi.controller.exception.ControllerServiceInstantiationException;
import org.apache.nifi.controller.exception.ProcessorInstantiationException;
import org.apache.nifi.controller.kerberos.KerberosConfig;
import org.apache.nifi.controller.reporting.ReportingTaskInstantiationException;
import org.apache.nifi.controller.repository.FlowFileEventRepository;
import org.apache.nifi.controller.scheduling.StandardProcessScheduler;
import org.apache.nifi.controller.service.ControllerServiceNode;
import org.apache.nifi.controller.status.history.StatusHistoryRepository;
import org.apache.nifi.engine.FlowEngine;
import org.apache.nifi.events.VolatileBulletinRepository;
import org.apache.nifi.expression.ExpressionLanguageCompiler;
import org.apache.nifi.nar.ExtensionDiscoveringManager;
import org.apache.nifi.nar.InstanceClassLoader;
import org.apache.nifi.nar.NarClassLoader;
import org.apache.nifi.nar.NarCloseable;
import org.apache.nifi.nar.OSUtil;
import org.apache.nifi.nar.StandardExtensionDiscoveringManager;
import org.apache.nifi.nar.SystemBundle;
import org.apache.nifi.parameter.ParameterContext;
import org.apache.nifi.processor.AbstractProcessor;
import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processor.ProcessSession;
import org.apache.nifi.processor.Processor;
import org.apache.nifi.processor.ProcessorInitializationContext;
import org.apache.nifi.processor.StandardProcessContext;
import org.apache.nifi.processor.StandardProcessorInitializationContext;
import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.processor.util.StandardValidators;
import org.apache.nifi.provenance.MockProvenanceRepository;
import org.apache.nifi.registry.VariableDescriptor;
import org.apache.nifi.registry.VariableRegistry;
import org.apache.nifi.registry.flow.FlowRegistryClient;
import org.apache.nifi.registry.variable.FileBasedVariableRegistry;
import org.apache.nifi.registry.variable.StandardComponentVariableRegistry;
import org.apache.nifi.test.processors.DynamicPropertiesTestProcessor;
import org.apache.nifi.test.processors.ModifiesClasspathNoAnnotationProcessor;
import org.apache.nifi.test.processors.ModifiesClasspathProcessor;
import org.apache.nifi.util.MockPropertyValue;
import org.apache.nifi.util.MockVariableRegistry;
import org.apache.nifi.util.NiFiProperties;
import org.apache.nifi.util.SynchronousValidationTrigger;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;

import java.io.File;
import java.net.MalformedURLException;
import java.net.URL;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.Callable;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;

import static org.hamcrest.CoreMatchers.containsString;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import static org.junit.Assume.assumeTrue;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;

public class StandardProcessorNodeIT {

    private MockVariableRegistry variableRegistry;
    private Bundle systemBundle;
    private ExtensionDiscoveringManager extensionManager;
    private NiFiProperties niFiProperties;

    private final AtomicReference<InstanceClassLoader> currentInstanceClassLoaderHolder = new AtomicReference<>();

    @Before
    public void setup() {
        variableRegistry = new MockVariableRegistry();
        niFiProperties = NiFiProperties.createBasicNiFiProperties("src/test/resources/conf/nifi.properties");

        systemBundle = SystemBundle.create(niFiProperties);
        extensionManager = new StandardExtensionDiscoveringManager() {
            @Override
            public InstanceClassLoader createInstanceClassLoader(String classType, String instanceIdentifier, Bundle bundle, Set<URL> additionalUrls) {
                InstanceClassLoader instanceClassLoader = super.createInstanceClassLoader(classType, instanceIdentifier, bundle, additionalUrls);

                currentInstanceClassLoaderHolder.set(instanceClassLoader);

                return instanceClassLoader;

            }
        };
        extensionManager.discoverExtensions(systemBundle, Collections.emptySet());
    }

    @Test(timeout = 10000)
    public void testStart() throws InterruptedException {
        final ProcessorThatThrowsExceptionOnScheduled processor = new ProcessorThatThrowsExceptionOnScheduled();
        final String uuid = UUID.randomUUID().toString();

        ProcessorInitializationContext initContext = new StandardProcessorInitializationContext(uuid, null, null, null, KerberosConfig.NOT_CONFIGURED);
        processor.initialize(initContext);

        final ReloadComponent reloadComponent = mock(ReloadComponent.class);
        final BundleCoordinate coordinate = mock(BundleCoordinate.class);

        final LoggableComponent<Processor> loggableComponent = new LoggableComponent<>(processor, coordinate, null);
        final StandardProcessorNode procNode = new StandardProcessorNode(loggableComponent, uuid, createValidationContextFactory(), null, null,
            new StandardComponentVariableRegistry(VariableRegistry.EMPTY_REGISTRY), reloadComponent, extensionManager, new SynchronousValidationTrigger());
        final ScheduledExecutorService taskScheduler = new FlowEngine(1, "TestClasspathResources", true);

        final StandardProcessContext processContext = new StandardProcessContext(procNode, null, null, null, () -> false, null);
        final SchedulingAgentCallback schedulingAgentCallback = new SchedulingAgentCallback() {
            @Override
            public void onTaskComplete() {
            }

            @Override
            public Future<?> scheduleTask(final Callable<?> task) {
                return taskScheduler.submit(task);
            }

            @Override
            public void trigger() {
                Assert.fail("Should not have completed");
            }
        };

        procNode.performValidation();
        procNode.start(taskScheduler, 20000L, 10000L, () -> processContext, schedulingAgentCallback, true);

        Thread.sleep(1000L);
        assertEquals(1, processor.onScheduledCount);
        assertEquals(1, processor.onUnscheduledCount);
        assertEquals(1, processor.onStoppedCount);
    }


    @Test
    public void testSinglePropertyDynamicallyModifiesClasspath() throws MalformedURLException {
        final MockReloadComponent reloadComponent = new MockReloadComponent();

        final PropertyDescriptor classpathProp = new PropertyDescriptor.Builder().name("Classpath Resources")
            .dynamicallyModifiesClasspath(true).addValidator(StandardValidators.NON_EMPTY_VALIDATOR).build();
        final ModifiesClasspathProcessor processor = new ModifiesClasspathProcessor(Arrays.asList(classpathProp));
        final StandardProcessorNode procNode = createProcessorNode(processor, reloadComponent);

        try (final NarCloseable narCloseable = NarCloseable.withComponentNarLoader(extensionManager, procNode.getProcessor().getClass(), procNode.getIdentifier())){

            // Should not have any of the test resources loaded at this point
            final URL[] testResources = getTestResources();
            for (URL testResource : testResources) {
                if (containsResource(reloadComponent.getAdditionalUrls(), testResource)) {
                    fail("found resource that should not have been loaded");
                }
            }

            // Simulate setting the properties of the processor to point to the test resources directory
            final Map<String, String> properties = new HashMap<>();
            properties.put(classpathProp.getName(), "src/test/resources/TestClasspathResources");
            procNode.setProperties(properties);

            // Should have all of the resources loaded into the InstanceClassLoader now
            for (URL testResource : testResources) {
                assertTrue(containsResource(reloadComponent.getAdditionalUrls(), testResource));
            }

            assertEquals(ModifiesClasspathProcessor.class.getCanonicalName(), reloadComponent.getNewType());

            // Should pass validation
            assertTrue(procNode.computeValidationErrors(procNode.getValidationContext()).isEmpty());
        } finally {
            extensionManager.removeInstanceClassLoader(procNode.getIdentifier());
        }
    }

    @Test
    public void testNativeLibLoadedFromDynamicallyModifiesClasspathProperty() throws Exception {
        // GIVEN
        assumeTrue("Test only runs on Mac OS", new OSUtil(){}.isOsMac());

        // Init NiFi
        NarClassLoader narClassLoader = mock(NarClassLoader.class);
        when(narClassLoader.getURLs()).thenReturn(new URL[0]);

        Bundle narBundle = SystemBundle.create(niFiProperties, narClassLoader);

        final NiFiProperties nifiProperties = createBasicNiFiPropertiesWithMocks();

        final FlowController flowController = FlowController.createStandaloneInstance(mock(FlowFileEventRepository.class), nifiProperties,
            mock(Authorizer.class), mock(AuditService.class), null, new VolatileBulletinRepository(),
            new FileBasedVariableRegistry(nifiProperties.getVariableRegistryPropertiesPaths()),
            mock(FlowRegistryClient.class), extensionManager, mock(StatusHistoryRepository.class));

        // Init processor
        final PropertyDescriptor classpathProp = new PropertyDescriptor.Builder().name("Classpath Resources")
            .dynamicallyModifiesClasspath(true).addValidator(StandardValidators.NON_EMPTY_VALIDATOR).build();

        final ModifiesClasspathProcessor processor = new ModifiesClasspathProcessor(Arrays.asList(classpathProp));
        final String uuid = UUID.randomUUID().toString();

        final ValidationContextFactory validationContextFactory = createValidationContextFactory();
        final ProcessScheduler processScheduler = mock(ProcessScheduler.class);
        final TerminationAwareLogger componentLog = mock(TerminationAwareLogger.class);

        final ReloadComponent reloadComponent = new StandardReloadComponent(flowController);
        ProcessorInitializationContext initContext = new StandardProcessorInitializationContext(uuid, componentLog, null, null, KerberosConfig.NOT_CONFIGURED);
        ((Processor) processor).initialize(initContext);

        final LoggableComponent<Processor> loggableComponent = new LoggableComponent<>(processor, narBundle.getBundleDetails().getCoordinate(), componentLog);
        final StandardProcessorNode procNode = new StandardProcessorNode(loggableComponent, uuid, validationContextFactory, processScheduler,
            null, new StandardComponentVariableRegistry(variableRegistry), reloadComponent, extensionManager, new SynchronousValidationTrigger());

        final Map<String, String> properties = new HashMap<>();
        properties.put(classpathProp.getName(), "src/test/resources/native");
        procNode.setProperties(properties);

        try (final NarCloseable narCloseable = NarCloseable.withComponentNarLoader(extensionManager, procNode.getProcessor().getClass(), procNode.getIdentifier())){
            // Should pass validation
            assertTrue(procNode.computeValidationErrors(procNode.getValidationContext()).isEmpty());

            // WHEN
            String actualLibraryLocation = currentInstanceClassLoaderHolder.get().findLibrary("testjni");

            // THEN
            assertThat(actualLibraryLocation, containsString(currentInstanceClassLoaderHolder.get().getIdentifier()));
        } finally {
            extensionManager.removeInstanceClassLoader(procNode.getIdentifier());
        }
    }

    @Test
    public void testDynamicPropertiesArePreservedAfterReload() throws ProcessorInstantiationException {
        // Init NiFi
        NarClassLoader narClassLoader = mock(NarClassLoader.class);
        when(narClassLoader.getURLs()).thenReturn(new URL[0]);

        Bundle narBundle = SystemBundle.create(niFiProperties, narClassLoader);

        final NiFiProperties nifiProperties = createBasicNiFiPropertiesWithMocks();

        final FlowController flowController = FlowController.createStandaloneInstance(mock(FlowFileEventRepository.class), nifiProperties,
                mock(Authorizer.class), mock(AuditService.class), null, new VolatileBulletinRepository(),
                new FileBasedVariableRegistry(nifiProperties.getVariableRegistryPropertiesPaths()),
                mock(FlowRegistryClient.class), extensionManager, mock(StatusHistoryRepository.class));

        // Init processor
        final DynamicPropertiesTestProcessor processor = new DynamicPropertiesTestProcessor();
        final String uuid = UUID.randomUUID().toString();

        final ValidationContextFactory validationContextFactory = createValidationContextFactory();
        final TerminationAwareLogger componentLog = mock(TerminationAwareLogger.class);

        final ReloadComponent reloadComponent = new StandardReloadComponent(flowController);

        final StateManagerProvider stateManagerProvider = mock(StateManagerProvider.class);

        final ProcessScheduler processScheduler = new StandardProcessScheduler(null, flowController, null,
                stateManagerProvider, nifiProperties);

        final LoggableComponent<Processor> loggableComponent = new LoggableComponent<>(processor, narBundle.getBundleDetails().getCoordinate(), componentLog);
        final StandardProcessorNode procNode = new StandardProcessorNode(loggableComponent, uuid, validationContextFactory, processScheduler,
                null, new StandardComponentVariableRegistry(variableRegistry), reloadComponent, extensionManager, new SynchronousValidationTrigger());
        processor.setProcessorNode(procNode);
        ProcessorInitializationContext initContext = new StandardProcessorInitializationContext(uuid, componentLog, null, null, KerberosConfig.NOT_CONFIGURED);
        ((Processor) processor).initialize(initContext);
        final BundleCoordinate existingCoordinate = procNode.getBundleCoordinate();

        final Map<String, String> properties = new HashMap<>();
        properties.put(DynamicPropertiesTestProcessor.STATIC_PROPERTY.getName(), "${value}");
        properties.put("dynamic-property", "value");
        procNode.setProperties(properties);
        procNode.performValidation();

        // Start out with one dynamic property name
        assertEquals(1, processor.getDynamicPropertyNames().size());

        // Reload the processor, which happens during a 'terminate' operation
        reloadComponent.reload(procNode, processor.getClass().getName(), existingCoordinate, null);

        // After terminate, we should still have one dynamic property, and the static property should be untouched
        assertEquals(1, ((DynamicPropertiesTestProcessor) procNode.getProcessor()).getDynamicPropertyNames().size());
        assertEquals("${value}", procNode.getProperties().get(DynamicPropertiesTestProcessor.STATIC_PROPERTY).getRawValue());
    }

    private NiFiProperties createBasicNiFiPropertiesWithMocks() {
        HashMap<String, String> additionalProperties = new HashMap<>();
        additionalProperties.put(NiFiProperties.ADMINISTRATIVE_YIELD_DURATION, "1 sec");
        additionalProperties.put(NiFiProperties.STATE_MANAGEMENT_CONFIG_FILE, "target/test-classes/state-management.xml");
        additionalProperties.put(NiFiProperties.STATE_MANAGEMENT_LOCAL_PROVIDER_ID, "local-provider");
        additionalProperties.put(NiFiProperties.PROVENANCE_REPO_IMPLEMENTATION_CLASS, MockProvenanceRepository.class.getName());
        additionalProperties.put("nifi.remote.input.socket.port", "");
        additionalProperties.put("nifi.remote.input.secure", "");

        return NiFiProperties.createBasicNiFiProperties("src/test/resources/conf/nifi.properties", additionalProperties);
    }

    @Test
    public void testUpdateOtherPropertyDoesNotImpactClasspath() throws MalformedURLException {
        final MockReloadComponent reloadComponent = new MockReloadComponent();

        final PropertyDescriptor classpathProp = new PropertyDescriptor.Builder().name("Classpath Resources")
            .dynamicallyModifiesClasspath(true).addValidator(StandardValidators.NON_EMPTY_VALIDATOR).build();

        final PropertyDescriptor otherProp = new PropertyDescriptor.Builder().name("My Property")
            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR).build();

        final ModifiesClasspathProcessor processor = new ModifiesClasspathProcessor(Arrays.asList(classpathProp, otherProp));
        final StandardProcessorNode procNode = createProcessorNode(processor, reloadComponent);

        try (final NarCloseable narCloseable = NarCloseable.withComponentNarLoader(extensionManager, procNode.getProcessor().getClass(), procNode.getIdentifier())){
            // Should not have any of the test resources loaded at this point
            final URL[] testResources = getTestResources();
            for (URL testResource : testResources) {
                if (containsResource(reloadComponent.getAdditionalUrls(), testResource)) {
                    fail("found resource that should not have been loaded");
                }
            }

            // Simulate setting the properties of the processor to point to the test resources directory
            final Map<String, String> properties = new HashMap<>();
            properties.put(classpathProp.getName(), "src/test/resources/TestClasspathResources");
            procNode.setProperties(properties);

            // Should have all of the resources loaded into the InstanceClassLoader now
            for (URL testResource : testResources) {
                assertTrue(containsResource(reloadComponent.getAdditionalUrls(), testResource));
            }

            // Should pass validation
            assertTrue(procNode.computeValidationErrors(procNode.getValidationContext()).isEmpty());

            // Simulate setting updating the other property which should not change the classpath
            final Map<String, String> otherProperties = new HashMap<>();
            otherProperties.put(otherProp.getName(), "foo");
            procNode.setProperties(otherProperties);

            // Should STILL have all of the resources loaded into the InstanceClassLoader now
            for (URL testResource : testResources) {
                assertTrue(containsResource(reloadComponent.getAdditionalUrls(), testResource));
            }

            // Should STILL pass validation
            assertTrue(procNode.computeValidationErrors(procNode.getValidationContext()).isEmpty());

            // Lets update the classpath property and make sure the resources get updated
            final Map<String, String> newClasspathProperties = new HashMap<>();
            newClasspathProperties.put(classpathProp.getName(), "src/test/resources/TestClasspathResources/resource1.txt");
            procNode.setProperties(newClasspathProperties);

            // Should only have resource1 loaded now
            assertTrue(containsResource(reloadComponent.getAdditionalUrls(), testResources[0]));
            assertFalse(containsResource(reloadComponent.getAdditionalUrls(), testResources[1]));
            assertFalse(containsResource(reloadComponent.getAdditionalUrls(), testResources[2]));

            assertEquals(ModifiesClasspathProcessor.class.getCanonicalName(), reloadComponent.getNewType());

            // Should STILL pass validation
            assertTrue(procNode.computeValidationErrors(procNode.getValidationContext()).isEmpty());
        } finally {
            extensionManager.removeInstanceClassLoader(procNode.getIdentifier());
        }
    }

    @Test
    public void testMultiplePropertiesDynamicallyModifyClasspathWithExpressionLanguage() throws MalformedURLException {
        final MockReloadComponent reloadComponent = new MockReloadComponent();

        final PropertyDescriptor classpathProp1 = new PropertyDescriptor.Builder().name("Classpath Resource 1")
            .dynamicallyModifiesClasspath(true).addValidator(StandardValidators.NON_EMPTY_VALIDATOR).build();
        final PropertyDescriptor classpathProp2 = new PropertyDescriptor.Builder().name("Classpath Resource 2")
            .dynamicallyModifiesClasspath(true).addValidator(StandardValidators.NON_EMPTY_VALIDATOR).build();

        final ModifiesClasspathProcessor processor = new ModifiesClasspathProcessor(Arrays.asList(classpathProp1, classpathProp2));
        final StandardProcessorNode procNode = createProcessorNode(processor, reloadComponent);

        try (final NarCloseable narCloseable = NarCloseable.withComponentNarLoader(extensionManager, procNode.getProcessor().getClass(), procNode.getIdentifier())){

            // Should not have any of the test resources loaded at this point
            final URL[] testResources = getTestResources();
            for (URL testResource : testResources) {
                if (containsResource(reloadComponent.getAdditionalUrls(), testResource)) {
                    fail("found resource that should not have been loaded");
                }
            }

            // Simulate setting the properties pointing to two of the resources
            final Map<String, String> properties = new HashMap<>();
            properties.put(classpathProp1.getName(), "src/test/resources/TestClasspathResources/resource1.txt");
            properties.put(classpathProp2.getName(), "src/test/resources/TestClasspathResources/${myResource}");

            variableRegistry.setVariable(new VariableDescriptor("myResource"), "resource3.txt");

            procNode.setProperties(properties);

            // Should have resources 1 and 3 loaded into the InstanceClassLoader now
            assertTrue(containsResource(reloadComponent.getAdditionalUrls(), testResources[0]));
            assertTrue(containsResource(reloadComponent.getAdditionalUrls(), testResources[2]));
            assertFalse(containsResource(reloadComponent.getAdditionalUrls(), testResources[1]));

            assertEquals(ModifiesClasspathProcessor.class.getCanonicalName(), reloadComponent.getNewType());

            // Should pass validation
            assertTrue(procNode.computeValidationErrors(procNode.getValidationContext()).isEmpty());
        } finally {
            extensionManager.removeInstanceClassLoader(procNode.getIdentifier());
        }
    }

    @Test
    public void testSomeNonExistentPropertiesDynamicallyModifyClasspath() throws MalformedURLException {
        final MockReloadComponent reloadComponent = new MockReloadComponent();

        final PropertyDescriptor classpathProp1 = new PropertyDescriptor.Builder().name("Classpath Resource 1")
            .dynamicallyModifiesClasspath(true).addValidator(StandardValidators.NON_EMPTY_VALIDATOR).build();
        final PropertyDescriptor classpathProp2 = new PropertyDescriptor.Builder().name("Classpath Resource 2")
            .dynamicallyModifiesClasspath(true).addValidator(StandardValidators.NON_EMPTY_VALIDATOR).build();

        final ModifiesClasspathProcessor processor = new ModifiesClasspathProcessor(Arrays.asList(classpathProp1, classpathProp2));
        final StandardProcessorNode procNode = createProcessorNode(processor, reloadComponent);

        try (final NarCloseable narCloseable = NarCloseable.withComponentNarLoader(extensionManager, procNode.getProcessor().getClass(), procNode.getIdentifier())){

            // Should not have any of the test resources loaded at this point
            final URL[] testResources = getTestResources();
            for (URL testResource : testResources) {
                if (containsResource(reloadComponent.getAdditionalUrls(), testResource)) {
                    fail("found resource that should not have been loaded");
                }
            }

            // Simulate setting the properties pointing to two of the resources
            final Map<String, String> properties = new HashMap<>();
            properties.put(classpathProp1.getName(), "src/test/resources/TestClasspathResources/resource1.txt");
            properties.put(classpathProp2.getName(), "src/test/resources/TestClasspathResources/DoesNotExist.txt");
            procNode.setProperties(properties);

            // Should have resources 1 and 3 loaded into the InstanceClassLoader now
            assertTrue(containsResource(reloadComponent.getAdditionalUrls(), testResources[0]));
            assertFalse(containsResource(reloadComponent.getAdditionalUrls(), testResources[1]));
            assertFalse(containsResource(reloadComponent.getAdditionalUrls(), testResources[2]));

            assertEquals(ModifiesClasspathProcessor.class.getCanonicalName(), reloadComponent.getNewType());

            // Should pass validation
            assertTrue(procNode.computeValidationErrors(procNode.getValidationContext()).isEmpty());
        } finally {
            extensionManager.removeInstanceClassLoader(procNode.getIdentifier());
        }
    }

    @Test
    public void testPropertyModifiesClasspathWhenProcessorMissingAnnotation() throws MalformedURLException {
        final MockReloadComponent reloadComponent = new MockReloadComponent();
        final ModifiesClasspathNoAnnotationProcessor processor = new ModifiesClasspathNoAnnotationProcessor();
        final StandardProcessorNode procNode = createProcessorNode(processor, reloadComponent);

        try (final NarCloseable narCloseable = NarCloseable.withComponentNarLoader(extensionManager, procNode.getProcessor().getClass(), procNode.getIdentifier())){

            final Map<String, String> properties = new HashMap<>();
            properties.put(ModifiesClasspathNoAnnotationProcessor.CLASSPATH_RESOURCE.getName(),
                "src/test/resources/TestClasspathResources/resource1.txt");
            procNode.setProperties(properties);

            final URL[] testResources = getTestResources();
            assertTrue(containsResource(reloadComponent.getAdditionalUrls(), testResources[0]));
            assertFalse(containsResource(reloadComponent.getAdditionalUrls(), testResources[1]));
            assertFalse(containsResource(reloadComponent.getAdditionalUrls(), testResources[2]));

            assertEquals(ModifiesClasspathNoAnnotationProcessor.class.getCanonicalName(), reloadComponent.getNewType());

            // Should pass validation
            assertTrue(procNode.computeValidationErrors(procNode.getValidationContext()).isEmpty());
        } finally {
            extensionManager.removeInstanceClassLoader(procNode.getIdentifier());
        }
    }

    @Test
    public void testVerifyCanUpdateBundle() {
        final ReloadComponent reloadComponent = new MockReloadComponent();
        final ModifiesClasspathNoAnnotationProcessor processor = new ModifiesClasspathNoAnnotationProcessor();
        final StandardProcessorNode procNode = createProcessorNode(processor, reloadComponent);
        final BundleCoordinate existingCoordinate = procNode.getBundleCoordinate();

        // should be allowed to update when the bundle is the same
        procNode.verifyCanUpdateBundle(existingCoordinate);

        // should be allowed to update when the group and id are the same but version is different
        final BundleCoordinate diffVersion = new BundleCoordinate(existingCoordinate.getGroup(), existingCoordinate.getId(), "v2");
        assertTrue(!existingCoordinate.getVersion().equals(diffVersion.getVersion()));
        procNode.verifyCanUpdateBundle(diffVersion);

        // should not be allowed to update when the bundle id is different
        final BundleCoordinate diffId = new BundleCoordinate(existingCoordinate.getGroup(), "different-id", existingCoordinate.getVersion());
        assertTrue(!existingCoordinate.getId().equals(diffId.getId()));
        try {
            procNode.verifyCanUpdateBundle(diffId);
            Assert.fail("Should have thrown exception");
        } catch (Exception e) {

        }

        // should not be allowed to update when the bundle group is different
        final BundleCoordinate diffGroup = new BundleCoordinate("different-group", existingCoordinate.getId(), existingCoordinate.getVersion());
        assertTrue(!existingCoordinate.getGroup().equals(diffGroup.getGroup()));
        try {
            procNode.verifyCanUpdateBundle(diffGroup);
            Assert.fail("Should have thrown exception");
        } catch (Exception e) {

        }
    }

    private StandardProcessorNode createProcessorNode(final Processor processor, final ReloadComponent reloadComponent) {
        final String uuid = UUID.randomUUID().toString();
        final ValidationContextFactory validationContextFactory = createValidationContextFactory();
        final ProcessScheduler processScheduler = mock(ProcessScheduler.class);
        final TerminationAwareLogger componentLog = mock(TerminationAwareLogger.class);

        extensionManager.createInstanceClassLoader(processor.getClass().getName(), uuid, systemBundle, null);

        ProcessorInitializationContext initContext = new StandardProcessorInitializationContext(uuid, componentLog, null, null, KerberosConfig.NOT_CONFIGURED);
        processor.initialize(initContext);

        final LoggableComponent<Processor> loggableComponent = new LoggableComponent<>(processor, systemBundle.getBundleDetails().getCoordinate(), componentLog);
        return new StandardProcessorNode(loggableComponent, uuid, validationContextFactory, processScheduler,
            null, new StandardComponentVariableRegistry(variableRegistry), reloadComponent, extensionManager, new SynchronousValidationTrigger());
    }

    private static class MockReloadComponent implements ReloadComponent {

        private String newType;
        private final Set<URL> additionalUrls = new LinkedHashSet<>();

        public Set<URL> getAdditionalUrls() {
            return this.additionalUrls;
        }

        public String getNewType() {
            return newType;
        }

        @Override
        public void reload(ProcessorNode existingNode, String newType, BundleCoordinate bundleCoordinate, Set<URL> additionalUrls) throws ProcessorInstantiationException {
            reload(newType, additionalUrls);
        }

        @Override
        public void reload(ControllerServiceNode existingNode, String newType, BundleCoordinate bundleCoordinate, Set<URL> additionalUrls) throws ControllerServiceInstantiationException {
            reload(newType, additionalUrls);
        }

        @Override
        public void reload(ReportingTaskNode existingNode, String newType, BundleCoordinate bundleCoordinate, Set<URL> additionalUrls) throws ReportingTaskInstantiationException {
            reload(newType, additionalUrls);
        }

        private void reload(String newType, Set<URL> additionalUrls) {
            this.newType = newType;
            this.additionalUrls.clear();
            if (additionalUrls != null) {
                this.additionalUrls.addAll(additionalUrls);
            }
        }
    }

    private boolean containsResource(Set<URL> resources, URL resourceToFind) {
        for (URL resource : resources) {
            if (resourceToFind.getPath().equals(resource.getPath())) {
                return true;
            }
        }
        return false;
    }

    private URL[] getTestResources() throws MalformedURLException {
        URL resource1 = new File("src/test/resources/TestClasspathResources/resource1.txt").toURI().toURL();
        URL resource2 = new File("src/test/resources/TestClasspathResources/resource2.txt").toURI().toURL();
        URL resource3 = new File("src/test/resources/TestClasspathResources/resource3.txt").toURI().toURL();
        return new URL[] { resource1, resource2, resource3 };
    }


    private ValidationContextFactory createValidationContextFactory() {
        return new ValidationContextFactory() {
            @Override
            public ValidationContext newValidationContext(Map<PropertyDescriptor, PropertyConfiguration> properties, String annotationData, String groupId, String componentId,
                                                          ParameterContext context, boolean validateConnections) {
                return new ValidationContext() {

                    @Override
                    public ControllerServiceLookup getControllerServiceLookup() {
                        return null;
                    }

                    @Override
                    public ValidationContext getControllerServiceValidationContext(ControllerService controllerService) {
                        return null;
                    }

                    @Override
                    public ExpressionLanguageCompiler newExpressionLanguageCompiler() {
                        return null;
                    }

                    @Override
                    public PropertyValue getProperty(PropertyDescriptor property) {
                        final PropertyConfiguration configuration = properties.get(property);
                        return newPropertyValue(configuration == null ? null : configuration.getRawValue());
                    }

                    @Override
                    public PropertyValue newPropertyValue(String value) {
                        return new MockPropertyValue(value);
                    }

                    @Override
                    public Map<PropertyDescriptor, String> getProperties() {
                        final Map<PropertyDescriptor, String> propertyMap = new HashMap<>();
                        properties.forEach((k, v) -> propertyMap.put(k, v == null ? null : v.getRawValue()));
                        return propertyMap;
                    }

                    @Override
                    public Map<String, String> getAllProperties() {
                        final Map<String,String> propValueMap = new LinkedHashMap<>();
                        for (final Map.Entry<PropertyDescriptor, String> entry : getProperties().entrySet()) {
                            propValueMap.put(entry.getKey().getName(), entry.getValue());
                        }
                        return propValueMap;
                    }

                    @Override
                    public String getAnnotationData() {
                        return null;
                    }

                    @Override
                    public boolean isValidationRequired(ControllerService service) {
                        return false;
                    }

                    @Override
                    public boolean isExpressionLanguagePresent(String value) {
                        return false;
                    }

                    @Override
                    public boolean isExpressionLanguageSupported(String propertyName) {
                        return false;
                    }

                    @Override
                    public String getProcessGroupIdentifier() {
                        return groupId;
                    }

                    @Override
                    public Collection<String> getReferencedParameters(final String propertyName) {
                        return Collections.emptyList();
                    }

                    @Override
                    public boolean isParameterDefined(final String parameterName) {
                        return false;
                    }

                    @Override
                    public boolean isParameterSet(final String parameterName) {
                        return false;
                    }

                    @Override
                    public boolean isDependencySatisfied(final PropertyDescriptor propertyDescriptor, final Function<String, PropertyDescriptor> propertyDescriptorLookup) {
                        return false;
                    }
                };
            }

        };

    }


    public static class ProcessorThatThrowsExceptionOnScheduled extends AbstractProcessor {
        private int onScheduledCount = 0;
        private int onUnscheduledCount = 0;
        private int onStoppedCount = 0;

        @Override
        public void onTrigger(ProcessContext context, ProcessSession session) throws ProcessException {
        }

        @OnScheduled
        public void onScheduled() {
            onScheduledCount++;
            throw new ProcessException("OnScheduled called - Unit Test throws Exception intentionally");
        }

        @OnUnscheduled
        public void onUnscheduled() {
            onUnscheduledCount++;
        }

        @OnStopped
        public void onStopped() {
            onStoppedCount++;
        }
    }

}