/**
 * 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.falcon.resource;

import com.sun.jersey.api.client.Client;
import com.sun.jersey.api.client.ClientResponse;
import com.sun.jersey.api.client.WebResource;
import com.sun.jersey.api.client.config.ClientConfig;
import com.sun.jersey.api.client.config.DefaultClientConfig;
import com.sun.jersey.client.urlconnection.HTTPSProperties;
import org.apache.commons.lang.RandomStringUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.net.util.TrustManagerUtils;
import org.apache.falcon.FalconCLIConstants;
import org.apache.falcon.FalconException;
import org.apache.falcon.FalconRuntimException;
import org.apache.falcon.catalog.HiveCatalogService;
import org.apache.falcon.cli.FalconCLI;
import org.apache.falcon.client.FalconCLIException;
import org.apache.falcon.client.FalconClient;
import org.apache.falcon.cluster.util.EmbeddedCluster;
import org.apache.falcon.entity.ClusterHelper;
import org.apache.falcon.entity.store.ConfigurationStore;
import org.apache.falcon.entity.v0.EntityType;
import org.apache.falcon.entity.v0.SchemaHelper;
import org.apache.falcon.entity.v0.cluster.Cluster;
import org.apache.falcon.entity.v0.cluster.ClusterLocationType;
import org.apache.falcon.entity.v0.feed.Feed;
import org.apache.falcon.hadoop.HadoopClientFactory;
import org.apache.falcon.security.CurrentUser;
import org.apache.falcon.util.DeploymentUtil;
import org.apache.falcon.util.StartupProperties;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.security.authentication.client.AuthenticatedURL;
import org.apache.hadoop.security.authentication.client.AuthenticationException;
import org.apache.hive.hcatalog.api.HCatClient;
import org.testng.Assert;

import javax.net.ssl.HostnameVerifier;
import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLSession;
import javax.net.ssl.TrustManager;
import javax.servlet.ServletInputStream;
import javax.ws.rs.HttpMethod;
import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.Response;
import javax.ws.rs.core.UriBuilder;
import javax.xml.bind.JAXBContext;
import javax.xml.bind.JAXBException;
import javax.xml.bind.Marshaller;
import javax.xml.bind.Unmarshaller;
import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.FileReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.io.StringReader;
import java.security.SecureRandom;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

/**
 * Base test class for CLI, Entity and Process Instances.
 */
public class TestContext extends AbstractTestContext {

    public static final String DATASOURCE_TEMPLATE1 = "/datasource-template1.xml";
    public static final String DATASOURCE_TEMPLATE2 = "/datasource-template2.xml";
    public static final String DATASOURCE_TEMPLATE3 = "/datasource-template3.xml";
    public static final String DATASOURCE_TEMPLATE4 = "/datasource-template4.xml";
    public static final String CLUSTER_TEMPLATE = "/cluster-template.xml";
    public static final String CLUSTER_UPDATED_TEMPLATE = "/cluster-updated-template.xml";
    public static final String PIG_PROCESS_TEMPLATE = "/pig-process-template.xml";

    public static final String BASE_URL = "https://localhost:41443/falcon-webapp";
    public static final String REMOTE_USER = FalconClient.USER;

    private static final String AUTH_COOKIE_EQ = AuthenticatedURL.AUTH_COOKIE + "=";

    protected Unmarshaller unmarshaller;
    protected Marshaller marshaller;

    protected EmbeddedCluster cluster;
    protected WebResource service = null;
    private AuthenticatedURL.Token authenticationToken;

    protected String clusterName;
    protected String processName;
    protected String processEndTime;
    protected String inputFeedName;
    protected String outputFeedName;

    public static final Pattern VAR_PATTERN = Pattern.compile("##[A-Za-z0-9_.]*##");

    public TestContext() {
        try {
            JAXBContext jaxbContext = JAXBContext.newInstance(APIResult.class, Feed.class, Process.class, Cluster.class,
                    InstancesResult.class);
            unmarshaller = jaxbContext.createUnmarshaller();
            marshaller = jaxbContext.createMarshaller();
            configure();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * This is only used for tests.
     *
     * @param metastoreUrl metastore url
     * @return client object
     * @throws FalconException
     */
    public static HCatClient getHCatClient(String metastoreUrl) throws FalconException {
        try {
            HiveConf hcatConf = HiveCatalogService.createHiveConf(new Configuration(false), metastoreUrl);
            return HCatClient.create(hcatConf);
        } catch (Exception e) {
            throw new FalconException("Exception creating HCatClient: " + e.getMessage(), e);
        }
    }

    public void configure() throws Exception {
        try {
            StartupProperties.get().setProperty(
                    "application.services",
                    StartupProperties.get().getProperty("application.services")
                            .replace("org.apache.falcon.service.ProcessSubscriberService", ""));
            String store = StartupProperties.get().getProperty("config.store.uri");
            StartupProperties.get().setProperty("config.store.uri", store + System.currentTimeMillis());
            SSLContext sslContext = SSLContext.getInstance("SSL");
            sslContext.init(
                    null,
                    new TrustManager[]{TrustManagerUtils.getValidateServerCertificateTrustManager()},
                    new SecureRandom());
            DefaultClientConfig config = new DefaultClientConfig();
            config.getProperties().put(HTTPSProperties.PROPERTY_HTTPS_PROPERTIES,
                    new HTTPSProperties(new HostnameVerifier() {
                        @Override
                        public boolean verify(String hostname, SSLSession sslSession) {
                            return true;
                        }
                    }, sslContext));
            Client client = Client.create(config);
            this.service = client.resource(UriBuilder.fromUri(BASE_URL).build());
        } catch (Exception e) {
            throw new FalconRuntimException(e);
        }

        try {
            String baseUrl = BASE_URL;
            if (!baseUrl.endsWith("/")) {
                baseUrl += "/";
            }
            this.authenticationToken = FalconClient.getToken(baseUrl);
        } catch (FalconCLIException e) {
            throw new AuthenticationException(e);
        }

        ClientConfig config = new DefaultClientConfig();
        Client client = Client.create(config);
        client.setReadTimeout(500000);
        client.setConnectTimeout(500000);
        this.service = client.resource(UriBuilder.fromUri(BASE_URL).build());
    }

    public void setCluster(String cName) throws Exception {
        cluster = EmbeddedCluster.newCluster(cName, true);
        this.clusterName = cluster.getCluster().getName();
    }

    public EmbeddedCluster getCluster() {
        return cluster;
    }

    public WebResource getService() {
        return service;
    }

    public String getAuthenticationToken() {
        return AUTH_COOKIE_EQ + authenticationToken;
    }

    public String getClusterName() {
        return clusterName;
    }

    public String getProcessName() {
        return processName;
    }

    public String getProcessEndTime() {
        return processEndTime;
    }

    public String getInputFeedName() {
        return inputFeedName;
    }

    public String getOutputFeedName() {
        return outputFeedName;
    }

    public String getClusterFileTemplate() {
        return CLUSTER_TEMPLATE;
    }

    public void scheduleProcess(String processTemplate, Map<String, String> overlay) throws Exception {
        scheduleProcess(processTemplate, overlay, true, null, "", null);
    }

    public void scheduleProcess(String processTemplate, Map<String, String> overlay,
                                Boolean skipDryRun, final String doAsUSer, String properties) throws Exception {
        scheduleProcess(processTemplate, overlay, true, skipDryRun, doAsUSer, properties);
    }

    public void scheduleProcess(String processTemplate, Map<String, String> overlay, boolean succeed) throws Exception{
        scheduleProcess(processTemplate, overlay, succeed, null, "", null);
    }

    public void scheduleProcess(String processTemplate, Map<String, String> overlay, boolean succeed,
                                Boolean skipDryRun, final String doAsUser, String properties) throws Exception {
        ClientResponse response = submitToFalcon(CLUSTER_TEMPLATE, overlay, EntityType.CLUSTER);
        assertSuccessful(response);

        response = submitToFalcon(FEED_TEMPLATE1, overlay, EntityType.FEED);
        assertSuccessful(response);

        response = submitToFalcon(FEED_TEMPLATE2, overlay, EntityType.FEED);
        assertSuccessful(response);

        response = submitAndSchedule(processTemplate, overlay, EntityType.PROCESS, skipDryRun, doAsUser, properties);
        if (succeed) {
            assertSuccessful(response);
        } else {
            assertFailure(response);
        }
    }

    public void scheduleProcess() throws Exception {
        scheduleProcess(PROCESS_TEMPLATE, getUniqueOverlay());
    }

    /**
     * Converts a InputStream into ServletInputStream.
     *
     * @param fileName file name
     * @return ServletInputStream
     * @throws java.io.IOException
     */
    public ServletInputStream getServletInputStream(String fileName) throws IOException {
        return getServletInputStream(new FileInputStream(fileName));
    }

    public ServletInputStream getServletInputStream(final InputStream stream) {
        return new ServletInputStream() {

            @Override
            public int read() throws IOException {
                return stream.read();
            }
        };
    }

    public ExtensionJobList getExtensionJobs(String extensionName, String fields, String sortOrder, String offset,
                                             String resultsPerPage, String doAsUser) {
        WebResource resource = this.service.path("api/extension/list/" + extensionName);
        resource = addQueryParam(resource, "doAs", doAsUser);
        resource = addQueryParam(resource, "fields", fields);
        resource = addQueryParam(resource, "sortOrder", sortOrder);
        resource = addQueryParam(resource, "offset", offset);
        resource = addQueryParam(resource, "numResults", resultsPerPage);
        ClientResponse response = resource.header("Cookie", AUTH_COOKIE_EQ + authenticationToken)
                .accept(MediaType.APPLICATION_JSON).type(MediaType.TEXT_XML)
                .method(HttpMethod.GET, ClientResponse.class);
        return response.getEntity(ExtensionJobList.class);
    }

    //SUSPEND CHECKSTYLE CHECK ParameterNumberCheck
    public ExtensionInstanceList getExtensionInstances(String jobName, String nominalStart, String nominalEnd,
                                                       String instanceStatus, String fields, String orderBy,
                                                       String sortOrder, String offset, String resultsPerPage,
                                                       String doAsUser) {
        WebResource resource = this.service.path("api/extension/instances/" + jobName);
        resource = addQueryParam(resource, "start", nominalStart);
        resource = addQueryParam(resource, "end", nominalEnd);
        resource = addQueryParam(resource, "instanceStatus", instanceStatus);
        resource = addQueryParam(resource, "doAs", doAsUser);
        resource = addQueryParam(resource, "fields", fields);
        resource = addQueryParam(resource, "orderBy", orderBy);
        resource = addQueryParam(resource, "sortOrder", sortOrder);
        resource = addQueryParam(resource, "offset", offset);
        resource = addQueryParam(resource, "numResults", resultsPerPage);
        ClientResponse response = resource.header("Cookie", AUTH_COOKIE_EQ + authenticationToken)
                .accept(MediaType.APPLICATION_JSON).type(MediaType.TEXT_XML)
                .method(HttpMethod.GET, ClientResponse.class);
        return response.getEntity(ExtensionInstanceList.class);
    }

    public void waitForInstancesToStart(String entityType, String entityName, long timeout) {
        long mustEnd = System.currentTimeMillis() + timeout;
        WebResource resource = this.service.path("api/instance/running/" + entityType + "/" + entityName);
        InstancesResult instancesResult;
        while (System.currentTimeMillis() < mustEnd) {
            ClientResponse response = resource.header("Cookie", AUTH_COOKIE_EQ + authenticationToken)
                    .accept(MediaType.APPLICATION_JSON).type(MediaType.TEXT_XML)
                    .method(HttpMethod.GET, ClientResponse.class);
            instancesResult = response.getEntity(InstancesResult.class);
            if (instancesResult.getInstances() != null && instancesResult.getInstances().length > 0) {
                break;
            }
        }
    }

    public ClientResponse submitAndSchedule(String template, Map<String, String> overlay, EntityType entityType)
        throws Exception {
        return submitAndSchedule(template, overlay, entityType, null, "", null);
    }

    public ClientResponse submitAndSchedule(String template, Map<String, String> overlay,
                                            EntityType entityType, Boolean skipDryRun,
                                            final String doAsUser, String properties) throws Exception {
        String tmpFile = overlayParametersOverTemplate(template, overlay);
        ServletInputStream rawlogStream = getServletInputStream(tmpFile);

        WebResource resource = this.service.path("api/entities/submitAndSchedule/" + entityType.name().toLowerCase());

        if (null != skipDryRun) {
            resource = resource.queryParam("skipDryRun", String.valueOf(skipDryRun));
        }

        if (StringUtils.isNotEmpty(doAsUser)) {
            resource = resource.queryParam(FalconCLIConstants.DO_AS_OPT, doAsUser);
        }

        if (StringUtils.isNotEmpty(properties)) {
            resource = resource.queryParam("properties", properties);
        }

        return resource.header("Cookie", getAuthenticationToken())
                .accept(MediaType.TEXT_XML)
                .type(MediaType.TEXT_XML)
                .post(ClientResponse.class, rawlogStream);
    }

    public ClientResponse validate(String template, Map<String, String> overlay, EntityType entityType)
        throws Exception {
        return validate(template, overlay, entityType, null);
    }

    public ClientResponse validate(String template, Map<String, String> overlay,
                                   EntityType entityType, Boolean skipDryRun)
        throws Exception {
        String tmpFile = overlayParametersOverTemplate(template, overlay);
        ServletInputStream rawlogStream = getServletInputStream(tmpFile);

        WebResource resource = service.path("api/entities/validate/" + entityType.name().toLowerCase());
        if (null != skipDryRun) {
            resource = resource.queryParam("skipDryRun", String.valueOf(skipDryRun));
        }
        return resource
            .header("Cookie", getAuthenticationToken())
            .accept(MediaType.TEXT_XML)
            .type(MediaType.TEXT_XML)
            .post(ClientResponse.class, rawlogStream);
    }

    public ClientResponse submitToFalcon(String template, Map<String, String> overlay, EntityType entityType)
        throws IOException {
        return submitToFalcon(template, overlay, entityType, "");
    }

    public ClientResponse submitToFalcon(String template, Map<String, String> overlay, EntityType entityType,
                                         final String doAsUser) throws IOException {
        String tmpFile = overlayParametersOverTemplate(template, overlay);
        if (entityType == EntityType.CLUSTER) {
            try {
                cluster = EmbeddedCluster.newCluster(overlay.get("cluster"), true);
                clusterName = cluster.getCluster().getName();
                deleteClusterLocations(cluster.getCluster(), cluster.getFileSystem());
                createClusterLocations(cluster.getCluster(), cluster.getFileSystem());
            } catch (Exception e) {
                throw new IOException("Unable to setup cluster info", e);
            }
        }
        return submitFileToFalcon(entityType, tmpFile, doAsUser);
    }

    public static void deleteClusterLocations(Cluster clusterEntity, FileSystem fs) throws IOException {
        String stagingLocation = ClusterHelper.getLocation(clusterEntity, ClusterLocationType.STAGING).getPath();
        Path stagingPath = new Path(stagingLocation);
        if (fs.exists(stagingPath)) {
            fs.delete(stagingPath, true);
        }

        String workingLocation = ClusterHelper.getLocation(clusterEntity, ClusterLocationType.WORKING).getPath();
        Path workingPath = new Path(workingLocation);
        if (fs.exists(workingPath)) {
            fs.delete(workingPath, true);
        }
    }

    public static void createClusterLocations(Cluster clusterEntity, FileSystem fs) throws IOException {
        createClusterLocations(clusterEntity, fs, true);
    }

    public static void createClusterLocations(Cluster clusterEntity, FileSystem fs, boolean withWorking)
        throws IOException {
        String stagingLocation = ClusterHelper.getLocation(clusterEntity, ClusterLocationType.STAGING).getPath();
        Path stagingPath = new Path(stagingLocation);
        if (fs.exists(stagingPath)) {
            fs.delete(stagingPath, true);
        }
        HadoopClientFactory.mkdirs(fs, stagingPath, HadoopClientFactory.ALL_PERMISSION);
        if (withWorking) {
            String workingLocation = ClusterHelper.getLocation(clusterEntity, ClusterLocationType.WORKING).getPath();
            Path workingPath = new Path(workingLocation);
            if (fs.exists(workingPath)) {
                fs.delete(workingPath, true);
            }
            HadoopClientFactory.mkdirs(fs, workingPath, HadoopClientFactory.READ_EXECUTE_PERMISSION);
        }
    }

    public ClientResponse submitFileToFalcon(EntityType entityType, String tmpFile) throws IOException {
        return submitFileToFalcon(entityType, tmpFile, "");
    }

    public ClientResponse submitFileToFalcon(EntityType entityType, String tmpFile,
                                             final String doAsUser) throws IOException {

        ServletInputStream rawlogStream = getServletInputStream(tmpFile);

        WebResource resource = this.service.path("api/entities/submit/" + entityType.name().toLowerCase());

        if (StringUtils.isNotEmpty(doAsUser)) {
            resource = resource.queryParam(FalconCLIConstants.DO_AS_OPT, doAsUser);
        }

        return resource.header("Cookie", getAuthenticationToken())
                .accept(MediaType.TEXT_XML)
                .type(MediaType.TEXT_XML)
                .post(ClientResponse.class, rawlogStream);
    }

    public ClientResponse deleteFromFalcon(String entityName, String entityType) throws IOException{
        return this.service.path("api/entities/delete/" + entityType + "/" + entityName.toLowerCase())
                .header("Cookie", getAuthenticationToken())
                .accept(MediaType.TEXT_XML)
                .delete(ClientResponse.class);
    }

    public void assertStatus(ClientResponse clientResponse, APIResult.Status status) {
        String response = clientResponse.getEntity(String.class);
        try {
            APIResult result = (APIResult) unmarshaller.unmarshal(new StringReader(response));
            Assert.assertEquals(result.getStatus(), status);
        } catch (JAXBException e) {
            Assert.fail("Response " + response + " is not valid");
        }
    }

    public void assertFailure(ClientResponse clientResponse) {
        Assert.assertEquals(clientResponse.getStatus(), Response.Status.BAD_REQUEST.getStatusCode());
        assertStatus(clientResponse, APIResult.Status.FAILED);
    }

    public void assertSuccessful(ClientResponse clientResponse) {
        Assert.assertEquals(clientResponse.getStatus(), Response.Status.OK.getStatusCode());
        assertStatus(clientResponse, APIResult.Status.SUCCEEDED);
    }

    public static String overlayParametersOverTemplate(String template,
                                                       Map<String, String> overlay) throws IOException {
        return overlayParametersOverTemplate(getTempFile(), template, overlay);
    }

    public static String overlayParametersOverTemplate(File file, String template,
                                                       Map<String, String> overlay) throws IOException {
        OutputStream out = new FileOutputStream(file);

        InputStreamReader in;
        InputStream resourceAsStream = TestContext.class.getResourceAsStream(template);
        if (resourceAsStream == null) {
            in = new FileReader(template);
        } else {
            in = new InputStreamReader(resourceAsStream);
        }
        BufferedReader reader = new BufferedReader(in);
        String line;
        while ((line = reader.readLine()) != null) {
            Matcher matcher = VAR_PATTERN.matcher(line);
            while (matcher.find()) {
                String variable = line.substring(matcher.start(), matcher.end());
                line = line.replace(variable, overlay.get(variable.substring(2, variable.length() - 2)));
                matcher = VAR_PATTERN.matcher(line);
            }
            out.write(line.getBytes());
            out.write("\n".getBytes());
        }
        reader.close();
        out.close();
        return file.getAbsolutePath();
    }

    @SuppressWarnings("ResultOfMethodCallIgnored")
    public static File getTempFile(String path, String prefix, String suffix) throws IOException {
        File f = new File(path);
        if (!f.exists()) {
            f.mkdirs();
        }

        return File.createTempFile(prefix, suffix, f);
    }

    public Map<String, String> getUniqueOverlay() throws FalconException {
        Map<String, String> overlay = new HashMap<String, String>();
        long time = System.currentTimeMillis();
        clusterName = "cluster" + time;
        overlay.put("src.cluster.name", clusterName);
        overlay.put("cluster", clusterName);
        overlay.put("colo", DeploymentUtil.getCurrentColo());
        inputFeedName = "in" + time;
        overlay.put("inputFeedName", inputFeedName);
        //only feeds with future dates can be scheduled
        Date endDate = new Date(System.currentTimeMillis() + 15 * 60 * 1000);
        overlay.put("feedEndDate", SchemaHelper.formatDateUTC(endDate));
        processEndTime = SchemaHelper.formatDateUTC(endDate);
        overlay.put("processEndDate", processEndTime);
        outputFeedName = "out" + time;
        overlay.put("outputFeedName", outputFeedName);
        processName = "p" + time;
        overlay.put("processName", processName);
        overlay.put("user", System.getProperty("user.name"));
        overlay.put("workflow.path", "/falcon/test/workflow");
        overlay.put("workflow.lib.path", "/falcon/test/workflow/lib");
        return overlay;
    }

    public static void prepare() throws Exception {
        prepare(TestContext.CLUSTER_TEMPLATE, true);
    }

    public static void prepare(String clusterTemplate, boolean disableLineage) throws Exception {
        // setup a logged in user
        CurrentUser.authenticate(REMOTE_USER);

        if (disableLineage) {
            // disable recording lineage metadata
            String services = StartupProperties.get().getProperty("application.services");
            StartupProperties.get().setProperty("application.services",
                    services.replace("org.apache.falcon.metadata.MetadataMappingService", ""));
        }

        Map<String, String> overlay = new HashMap<String, String>();
        overlay.put("cluster", RandomStringUtils.randomAlphabetic(5));
        overlay.put("colo", DeploymentUtil.getCurrentColo());
        TestContext.overlayParametersOverTemplate(clusterTemplate, overlay);
        EmbeddedCluster cluster = EmbeddedCluster.newCluster(overlay.get("cluster"), true);

        cleanupStore();

        // setup dependent workflow and lipath in hdfs
        FileSystem fs = FileSystem.get(cluster.getConf());
        mkdir(fs, new Path("/falcon"), new FsPermission((short) 511));

        Path wfParent = new Path("/falcon/test");
        fs.delete(wfParent, true);
        Path wfPath = new Path(wfParent, "workflow");
        mkdir(fs, wfPath);
        mkdir(fs, new Path("/falcon/test/workflow/lib"));
        fs.copyFromLocalFile(false, true,
                new Path(TestContext.class.getResource("/fs-workflow.xml").getPath()),
                new Path(wfPath, "workflow.xml"));
        mkdir(fs, new Path(wfParent, "input/2012/04/20/00"));
        Path outPath = new Path(wfParent, "output");
        mkdir(fs, outPath, new FsPermission((short) 511));

        // init cluster locations
        initClusterLocations(cluster, fs);
    }

    private static void initClusterLocations(EmbeddedCluster cluster, FileSystem fs) throws Exception {
        String stagingPath = ClusterHelper.getLocation(cluster.getCluster(), ClusterLocationType.STAGING).getPath();
        mkdir(fs, new Path(stagingPath), HadoopClientFactory.ALL_PERMISSION);

        String workingPath = ClusterHelper.getLocation(cluster.getCluster(), ClusterLocationType.WORKING).getPath();
        mkdir(fs, new Path(workingPath), HadoopClientFactory.READ_EXECUTE_PERMISSION);
    }

    public static void cleanupStore() throws Exception {
        for (EntityType type : EntityType.values()) {
            for (String name : ConfigurationStore.get().getEntities(type)) {
                ConfigurationStore.get().remove(type, name);
            }
        }
    }

    public static void deleteEntitiesFromStore() throws Exception {
        for (EntityType type : EntityType.values()) {
            for (String name : ConfigurationStore.get().getEntities(type)) {
                executeWithURL("entity -delete -type " + type.name().toLowerCase() + " -name " + name);
            }
        }
    }

    public static int executeWithURL(String command) throws Exception {
        return new FalconCLI().run((command + " -url " + TestContext.BASE_URL).split("\\s+"));
    }

    private WebResource addQueryParam(WebResource resource, String key, String value) {
        if (StringUtils.isEmpty(value)) {
            return resource;
        }
        return resource.queryParam(key, value);
    }
}