<commit_before>from flow.commands.base import CommandBase
from flow.petri.safenet import Token, SetTokenMessage
from tempfile import NamedTemporaryFile
import flow.redisom as rom
import json
import subprocess
import os

import logging

LOG = logging.getLogger(__name__)


class WrapperCommand(CommandBase):
    default_logging_mode = 'debug'

    def __init__(self, broker=None, storage=None, routing_key=None, exchange=None):
        self.broker = broker
        self.storage = storage
        self.routing_key = routing_key
        self.exchange = exchange


    @staticmethod
    def annotate_parser(parser):
        parser.add_argument('--net-key', '-n')
        parser.add_argument('--running-place-id', '-r', type=int)
        parser.add_argument('--success-place-id', '-s', type=int)
        parser.add_argument('--failure-place-id', '-f', type=int)
        parser.add_argument('--with-inputs', default=None, type=str)
        parser.add_argument('--with-outputs', default=False,
                action="store_true")

        parser.add_argument('command_line', nargs='+')

    def __call__(self, parsed_arguments):
        self.send_token(net_key=parsed_arguments.net_key,
                place_idx=parsed_arguments.running_place_id,
                data={"pid": str(os.getpid())})

        rv = 1
        cmdline = parsed_arguments.command_line

        with NamedTemporaryFile() as inputs_file:
            with NamedTemporaryFile() as outputs_file:
                if parsed_arguments.with_inputs:
                    inputs_hash = rom.Hash(
                            connection=self.storage,
                            key=parsed_arguments.with_inputs,
                            value_decoder=rom.json_dec,
                            value_encoder=rom.json_enc)

                    inputs = inputs_hash.value

                    LOG.debug("Fetched inputs from key %s" %
                            parsed_arguments.with_inputs)
                    LOG.debug("Input values: %r" % inputs)

                    json.dump(inputs, inputs_file)
                    inputs_file.flush()
                    cmdline += ["--inputs-file", inputs_file.name]
                else:
                    LOG.debug("No inputs for command")

                if parsed_arguments.with_outputs:
                    cmdline += ["--outputs-file", outputs_file.name]

                try:
                    subprocess.check_call(cmdline)

                    outputs = None
                    if parsed_arguments.with_outputs:
                        outputs = json.load(outputs_file)

                    self.send_token(net_key=parsed_arguments.net_key,
                            place_idx=parsed_arguments.success_place_id,
                            data={"exit_code": 0, "outputs": outputs})

                    rv = 0
                except CalledProcessError as e:
                    LOG.error("Failed to execute command '%s': %s",
                            " ".join(cmdline), str(e))
                    self.send_token(net_key=parsed_arguments.net_key,
                            place_idx=parsed_arguments.failure_place_id,
                            data={"exit_code": e.returncode})

                    rv = e.returncode

        return rv
<commit_msg>qualify subprocess.CalledProcessError<commit_after>from flow.commands.base import CommandBase
from flow.petri.safenet import Token, SetTokenMessage
from tempfile import NamedTemporaryFile
import flow.redisom as rom
import json
import subprocess
import os

import logging

LOG = logging.getLogger(__name__)


class WrapperCommand(CommandBase):
    default_logging_mode = 'debug'

    def __init__(self, broker=None, storage=None, routing_key=None, exchange=None):
        self.broker = broker
        self.storage = storage
        self.routing_key = routing_key
        self.exchange = exchange


    @staticmethod
    def annotate_parser(parser):
        parser.add_argument('--net-key', '-n')
        parser.add_argument('--running-place-id', '-r', type=int)
        parser.add_argument('--success-place-id', '-s', type=int)
        parser.add_argument('--failure-place-id', '-f', type=int)
        parser.add_argument('--with-inputs', default=None, type=str)
        parser.add_argument('--with-outputs', default=False,
                action="store_true")

        parser.add_argument('command_line', nargs='+')

    def __call__(self, parsed_arguments):
        self.send_token(net_key=parsed_arguments.net_key,
                place_idx=parsed_arguments.running_place_id,
                data={"pid": str(os.getpid())})

        rv = 1
        cmdline = parsed_arguments.command_line

        with NamedTemporaryFile() as inputs_file:
            with NamedTemporaryFile() as outputs_file:
                if parsed_arguments.with_inputs:
                    inputs_hash = rom.Hash(
                            connection=self.storage,
                            key=parsed_arguments.with_inputs,
                            value_decoder=rom.json_dec,
                            value_encoder=rom.json_enc)

                    inputs = inputs_hash.value

                    LOG.debug("Fetched inputs from key %s" %
                            parsed_arguments.with_inputs)
                    LOG.debug("Input values: %r" % inputs)

                    json.dump(inputs, inputs_file)
                    inputs_file.flush()
                    cmdline += ["--inputs-file", inputs_file.name]
                else:
                    LOG.debug("No inputs for command")

                if parsed_arguments.with_outputs:
                    cmdline += ["--outputs-file", outputs_file.name]

                try:
                    subprocess.check_call(cmdline)

                    outputs = None
                    if parsed_arguments.with_outputs:
                        outputs = json.load(outputs_file)

                    self.send_token(net_key=parsed_arguments.net_key,
                            place_idx=parsed_arguments.success_place_id,
                            data={"exit_code": 0, "outputs": outputs})

                    rv = 0
                except subprocess.CalledProcessError as e:
                    LOG.error("Failed to execute command '%s': %s",
                            " ".join(cmdline), str(e))
                    self.send_token(net_key=parsed_arguments.net_key,
                            place_idx=parsed_arguments.failure_place_id,
                            data={"exit_code": e.returncode})

                    rv = e.returncode

        return rv

{'fraction_non_alphanumeric': 0.09421060629083726, 'fraction_numerical': 0.000911715544750038, 'mean_word_length': 1.7108731466227347, 'pattern_counts': {'":': 8, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2192982', 'n_tokens_mistral': 1782, 'n_tokens_neox': 1694, 'n_words': 317}