<commit_before>#/*******************************************************************************
# * Copyright (c) 2013 GigaSpaces Technologies Ltd. All rights reserved
# *
# * Licensed 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.
# *******************************************************************************/

import os
import urllib2
import tarfile
import zipfile
import shutil
import subprocess
import ast
import _ast
from subprocess import CalledProcessError
from cosmo.celery import celery, get_cosmo_properties
from celery.utils.log import get_task_logger
from os.path import expanduser


TAR_GZ_SUFFIX = "tar.gz"
ZIP_SUFFIX = "zip"
PLUGIN_FILE_PREFIX = "plugin"
REQUIREMENTS_FILE = "requirements.txt"
TASKS_MODULE = "tasks.py"

logger = get_task_logger(__name__)


@celery.task
def install(plugin, __cloudify_id, **kwargs):
    """
    Installs plugin as celery task according to the provided plugins details.
    plugin parameter is expected to be in the following format: { name: "...", url: "..." }
    The plugin's url should be an http url pointing to either a zip or tar.gz file and the compressed file
    should contain a "tasks.py" module with celery tasks.
    """

    logger.debug("installing plugin [%s] in host [%s]", plugin, __cloudify_id)

    name = plugin["name"]
    url = plugin["url"]

    management_ip = get_cosmo_properties()["management_ip"]
    if management_ip != None:
        url = url.replace("#{plugin_repository}", "http://{0}:{1}".format(management_ip, "53229"))

    extract = None
    plugin_file = None

    if url.endswith(TAR_GZ_SUFFIX):
        extract = extract_targz
        plugin_file = "{0}.{1}".format(PLUGIN_FILE_PREFIX, TAR_GZ_SUFFIX)
    elif url.endswith(ZIP_SUFFIX):
        extract = extract_zip
        plugin_file = "{0}.{1}".format(PLUGIN_FILE_PREFIX, ZIP_SUFFIX)
    else:
        raise RuntimeError("archive extension not supported - supported extensions: {0}".format([TAR_GZ_SUFFIX, ZIP_SUFFIX]))

    plugin_path = create_plugin_path(name)
    plugin_file_path = os.path.join(plugin_path, plugin_file)
    # from this stage - on failure remove the created plugin path
    try:
        # TODO: download_plugin can be invoked as a celery subtask (asynchronous)
        download_plugin(url, plugin_file_path)
        extract(plugin_file_path, plugin_path)
        if not os.path.exists(os.path.join(plugin_path, TASKS_MODULE)):
            raise RuntimeError("plugin does not contain a '{0}' module".format(TASKS_MODULE))
        os.remove(plugin_file_path)

        requirements_file = os.path.join(plugin_path, REQUIREMENTS_FILE)
        if os.path.exists(requirements_file):
            logger.info('installing packages from requirements.txt file')
            os.chdir(plugin_path)
            command = ["sudo", "pip", "--default-timeout=120", "install", "-r", requirements_file]
            p = subprocess.Popen(command)
            out, err = p.communicate()
            print("command={0}, output={1}, err={2}".format(command, out, err))
            if p.returncode != 0:
                raise RuntimeError("unable to install requirements.txt [returncode={0}]".format(p.returncode))
    except (CalledProcessError, Exception) as error:
        subprocess.Popen(["sudo", "rm", "-rf", plugin_path])
        raise error


@celery.task
def verify_plugin(worker_id, plugin_name, operation, **kwargs):
    p = subprocess.Popen(["celery", "inspect", "registered", "-d", worker_id, "--no-color"], stdout=subprocess.PIPE)
    out, err = p.communicate()

    if p.returncode != 0:
        raise RuntimeError("unable to get celery worker registered tasks [returncode={0}]".format(p.returncode))

    lines = out.splitlines()
    registered_tasks = list()
    operation_name = operation.split(".")[-1]
    for line in lines:
        processed_line = line.strip()
        if processed_line.startswith("*"):
            task = processed_line[1:].strip()
            if task.startswith(plugin_name) and task.endswith("." + operation_name):
                current_dir = os.path.abspath(os.path.join(__file__, os.pardir))
                plugins_dir = os.path.abspath(os.path.join(current_dir, os.pardir))
                taskspy_path = os.path.join(plugins_dir, plugin_name.split(".")[-2]) + "/tasks.py"

                parsed_tasks_file = ast.parse(open(taskspy_path, 'r').read())
                method_description = filter(lambda item: type(item) == _ast.FunctionDef and item.name == operation_name,
                                            parsed_tasks_file.body)
                if not method_description:
                    raise RuntimeError("unable to locate operation {0} inside plugin file {1} [plugin name={2}]"
                    .format(operation_name, taskspy_path, plugin_name))
                return map(lambda arg: arg.id, method_description[0].args.args)
            else:
                registered_tasks.append(task)
    return False


def download_plugin(url, path):<commit_msg>CLOUDIFY-1999 fixed plugin directory in plugin verification task.<commit_after>#/*******************************************************************************
# * Copyright (c) 2013 GigaSpaces Technologies Ltd. All rights reserved
# *
# * Licensed 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.
# *******************************************************************************/

import os
import urllib2
import tarfile
import zipfile
import shutil
import subprocess
import ast
import _ast
from subprocess import CalledProcessError
from cosmo.celery import celery, get_cosmo_properties
from celery.utils.log import get_task_logger
from os.path import expanduser
import cosmo


TAR_GZ_SUFFIX = "tar.gz"
ZIP_SUFFIX = "zip"
PLUGIN_FILE_PREFIX = "plugin"
REQUIREMENTS_FILE = "requirements.txt"
TASKS_MODULE = "tasks.py"

logger = get_task_logger(__name__)


@celery.task
def install(plugin, __cloudify_id, **kwargs):
    """
    Installs plugin as celery task according to the provided plugins details.
    plugin parameter is expected to be in the following format: { name: "...", url: "..." }
    The plugin's url should be an http url pointing to either a zip or tar.gz file and the compressed file
    should contain a "tasks.py" module with celery tasks.
    """

    logger.debug("installing plugin [%s] in host [%s]", plugin, __cloudify_id)

    name = plugin["name"]
    url = plugin["url"]

    management_ip = get_cosmo_properties()["management_ip"]
    if management_ip != None:
        url = url.replace("#{plugin_repository}", "http://{0}:{1}".format(management_ip, "53229"))

    extract = None
    plugin_file = None

    if url.endswith(TAR_GZ_SUFFIX):
        extract = extract_targz
        plugin_file = "{0}.{1}".format(PLUGIN_FILE_PREFIX, TAR_GZ_SUFFIX)
    elif url.endswith(ZIP_SUFFIX):
        extract = extract_zip
        plugin_file = "{0}.{1}".format(PLUGIN_FILE_PREFIX, ZIP_SUFFIX)
    else:
        raise RuntimeError("archive extension not supported - supported extensions: {0}".format([TAR_GZ_SUFFIX, ZIP_SUFFIX]))

    plugin_path = create_plugin_path(name)
    plugin_file_path = os.path.join(plugin_path, plugin_file)
    # from this stage - on failure remove the created plugin path
    try:
        # TODO: download_plugin can be invoked as a celery subtask (asynchronous)
        download_plugin(url, plugin_file_path)
        extract(plugin_file_path, plugin_path)
        if not os.path.exists(os.path.join(plugin_path, TASKS_MODULE)):
            raise RuntimeError("plugin does not contain a '{0}' module".format(TASKS_MODULE))
        os.remove(plugin_file_path)

        requirements_file = os.path.join(plugin_path, REQUIREMENTS_FILE)
        if os.path.exists(requirements_file):
            logger.info('installing packages from requirements.txt file')
            os.chdir(plugin_path)
            command = ["sudo", "pip", "--default-timeout=120", "install", "-r", requirements_file]
            p = subprocess.Popen(command)
            out, err = p.communicate()
            print("command={0}, output={1}, err={2}".format(command, out, err))
            if p.returncode != 0:
                raise RuntimeError("unable to install requirements.txt [returncode={0}]".format(p.returncode))
    except (CalledProcessError, Exception) as error:
        subprocess.Popen(["sudo", "rm", "-rf", plugin_path])
        raise error


@celery.task
def verify_plugin(worker_id, plugin_name, operation, **kwargs):
    p = subprocess.Popen(["celery", "inspect", "registered", "-d", worker_id, "--no-color"], stdout=subprocess.PIPE)
    out, err = p.communicate()

    if p.returncode != 0:
        raise RuntimeError("unable to get celery worker registered tasks [returncode={0}]".format(p.returncode))

    lines = out.splitlines()
    registered_tasks = list()
    operation_name = operation.split(".")[-1]
    for line in lines:
        processed_line = line.strip()
        if processed_line.startswith("*"):
            task = processed_line[1:].strip()
            if task.startswith(plugin_name) and task.endswith("." + operation_name):
                cosmo_dir = os.path.abspath(os.path.dirname(cosmo.__file__))
                plugin_dir = os.path.join(cosmo_dir, os.sep.join(plugin_name.split(".")[1:-1]))
                taskspy_path = os.path.join(plugin_dir, "tasks.py")
                parsed_tasks_file = ast.parse(open(taskspy_path, 'r').read())
                method_description = filter(lambda item: type(item) == _ast.FunctionDef and item.name == operation_name,
                                            parsed_tasks_file.body)
                if not method_description:
                    raise RuntimeError("unable to locate operation {0} inside plugin file {1} [plugin name={2}]"
                    .format(operation_name, taskspy_path, plugin_name))
                return map(lambda arg: arg.id, method_description[0].args.args)
            else:
                registered_tasks.append(task)
    return False


def download_plugin(url, path):
{'fraction_non_alphanumeric': 0.13761467889908258, 'fraction_numerical': 0.007539286038695613, 'mean_word_length': 3.2674418604651163, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 10, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7666094', 'n_tokens_mistral': 3278, 'n_tokens_neox': 3146, 'n_words': 868}