path: root/http_prompt/execution.py

import io
import json
import re
import os
import sys
from urllib.parse import urlparse, urljoin

import click

from subprocess import CalledProcessError, Popen, PIPE

from httpie.context import Environment
from httpie.core import main as httpie_main
from parsimonious.exceptions import ParseError, VisitationError
from parsimonious.grammar import Grammar
from parsimonious.nodes import NodeVisitor
from parsimonious.nodes import Node
from pygments.token import String, Name

from .completion import ROOT_COMMANDS, ACTIONS, OPTION_NAMES, HEADER_NAMES
from .context import Context
from .context.transform import (
    extract_args_for_httpie_main,
    format_to_curl,
    format_to_httpie,
    format_to_http_prompt)
from .output import Printer, TextWriter
from .utils import unescape, unquote, colformat


HTTPIE_PROGRAM_NAME = 'http'


grammar = r"""
    command = mutation / immutation

    mutation = concat_mut+ / nonconcat_mut
    immutation = preview / action / ls / env / help / exit / exec / source / clear / _

    concat_mut = option_mut / full_quoted_mut / value_quoted_mut / unquoted_mut
    nonconcat_mut = cd / rm

    preview = _ tool _ (method _)? (urlpath _)? concat_mut* redir_out? _
    action = _ method _ (urlpath _)? concat_mut* redir_out? _
    urlpath = (~r"https?://" unquoted_string) /
              (!concat_mut !redir_out string)

    clear = _ "clear" _
    help = _ "help" _
    exit = _ "exit" _
    ls = _ "ls" _ (urlpath _)? (redir_out)?
    env  = _ "env" _ (redir_out)?
    source = _ "source" _ filepath _
    exec = _ "exec" _ filepath _

    redir_out = redir_append / redir_write / pipe
    redir_append = _ ">>" _ filepath _
    redir_write = _ ">" _ filepath _
    pipe = _ "|" _ (shell_subs / shell_code) _

    unquoted_mut = _ unquoted_mutkey mutop unquoted_mutval _
    full_quoted_mut = full_squoted_mut / full_dquoted_mut
    value_quoted_mut = value_squoted_mut / value_dquoted_mut
    full_squoted_mut = _ "'" squoted_mutkey mutop squoted_mutval "'" _
    full_dquoted_mut = _ '"' dquoted_mutkey mutop dquoted_mutval '"' _
    value_squoted_mut = _ unquoted_mutkey mutop "'" squoted_mutval "'" _
    value_dquoted_mut = _ unquoted_mutkey mutop '"' dquoted_mutval '"' _
    mutop = ":=" / ":" / "==" / "="
    unquoted_mutkey = unquoted_mutkey_item+
    unquoted_mutval = unquoted_stringitem*
    unquoted_mutkey_item = shell_subs / unquoted_mutkey_char / escapeseq
    unquoted_mutkey_char = ~r"[^\s'\"\\=:>]"
    squoted_mutkey = squoted_mutkey_item+
    squoted_mutval = squoted_stringitem*
    squoted_mutkey_item = shell_subs / squoted_mutkey_char / escapeseq
    squoted_mutkey_char = ~r"[^\r\n'\\=:]"
    dquoted_mutkey = dquoted_mutkey_item+
    dquoted_mutval = dquoted_stringitem*
    dquoted_mutkey_item = shell_subs / dquoted_mutkey_char / escapeseq
    dquoted_mutkey_char = ~r'[^\r\n"\\=:]'

    option_mut = flag_option_mut / value_option_mut
    flag_option_mut = _ flag_optname _
    flag_optname = "--json" / "-j" / "--form" / "-f" / "--verbose" / "-v" /
                   "--headers" / "-h" / "--body" / "-b" / "--stream" / "-S" /
                   "--download" / "-d" / "--continue" / "-c" / "--follow" /
                   "--check-status" / "--ignore-stdin" / "--help" /
                   "--version" / "--traceback" / "--debug"
    value_option_mut = _ value_optname ~r"(\s+|=)" string _
    value_optname = "--pretty" / "--style" / "-s" / "--print" / "-p" /
                    "--output" / "-o" / "--session-read-only" / "--session" /
                    "--auth-type" / "--auth" / "-a" / "--proxy" / "--verify" /
                    "--cert" / "--cert-key" / "--timeout" / "--raw"

    cd = _ "cd" _ string? _
    rm = (_ "rm" _ "*" _) / (_ "rm" _ ~r"\-(h|q|b|o)" _ mutkey _)
    tool = "httpie" / "curl"
    method = ~r"get"i / ~r"head"i / ~r"post"i / ~r"put"i / ~r"delete"i /
             ~r"patch"i / ~r"options"i / ~r"connect"i
    mutkey = unquoted_mutkey / ("'" squoted_mutkey "'") /
             ('"' dquoted_mutkey '"') / flag_optname / value_optname

    string = quoted_string / unquoted_string
    quoted_string = ('"' dquoted_stringitem* '"') /
                    ("'" squoted_stringitem* "'")
    unquoted_string = unquoted_stringitem+
    dquoted_stringitem = shell_subs / dquoted_stringchar / escapeseq
    squoted_stringitem = shell_subs / squoted_stringchar / escapeseq
    unquoted_stringitem = shell_subs / unquoted_stringchar / escapeseq
    dquoted_stringchar = ~r'[^\r\n"\\]'
    squoted_stringchar = ~r"[^\r\n'\\]"
    unquoted_stringchar = ~r"[^\s'\\]"
    escapeseq = ~r"\\."
    _ = ~r"\s*"

    shell_subs = "`" shell_code "`"
    shell_code = ~r"[^`]*"
"""

if sys.platform == 'win32':
    # XXX: Windows use backslashes as separators in its filesystem path, so we
    # have to avoid using backslashes to escape chars here.
    grammar += r"""
        filepath = quoted_filepath / unquoted_filepath
        quoted_filepath = ('"' dquoted_filepath_char+ '"') /
                          ("'" squoted_filepath_char+ "'")
        dquoted_filepath_char = ~r'[^\r\n"]'
        squoted_filepath_char = ~r"[^\r\n']"
        unquoted_filepath = unquoted_filepath_char+
        unquoted_filepath_char = ~r"[^\s\"]"
    """
else:
    grammar += r"""
        filepath = string
    """

grammar = Grammar(grammar)


if Environment.colors == 256:
    from pygments.formatters.terminal256 import (
        Terminal256Formatter as TerminalFormatter)
else:
    from pygments.formatters.terminal import TerminalFormatter


def urljoin2(base, path, **kwargs):
    if not base.endswith('/'):
        base += '/'
    url = urljoin(base, path, **kwargs)
    if url.endswith('/') and not path.endswith('/'):
        url = url[:-1]
    return url


def generate_help_text():
    """Return a formatted string listing commands, HTTPie options, and HTTP
    actions.
    """
    def generate_cmds_with_explanations(summary, cmds):
        text = '{0}:\n'.format(summary)
        for cmd, explanation in cmds:
            text += '\t{0:<10}\t{1:<20}\n'.format(cmd, explanation)
        return text + '\n'

    text = generate_cmds_with_explanations('Commands', ROOT_COMMANDS.items())
    text += generate_cmds_with_explanations('Options', OPTION_NAMES.items())
    text += generate_cmds_with_explanations('Actions', ACTIONS.items())
    text += generate_cmds_with_explanations('Headers', HEADER_NAMES.items())
    return text


if sys.platform == 'win32':  # nocover
    def normalize_filepath(path):
        return unquote(path)
else:
    def normalize_filepath(path):
        return unescape(unquote(path))


class DummyExecutionListener(object):

    def context_changed(self, context):
        pass

    def response_returned(self, context, response):
        pass


class ExecutionVisitor(NodeVisitor):

    unwrapped_exceptions = (CalledProcessError,)

    def __init__(self, context, listener=None, style=None):
        super(ExecutionVisitor, self).__init__()
        self.context = context

        self.context_override = Context(context.url)
        self.method = None
        self.tool = None
        self._output = Printer()

        # If there's a pipe, as in "httpie post | sed s/POST/GET/", this
        # variable points to the "sed" Popen object. The variable is necessary
        # because the we need to redirect Popen.stdout to Printer, which does
        # output pagination.
        self.pipe_proc = None

        self.listener = listener or DummyExecutionListener()

        # Last response object returned by HTTPie
        self.last_response = None

        # Pygments formatter, used to render output with colors in some cases
        if style:
            self.formatter = TerminalFormatter(style=style)
        else:
            self.formatter = None

    @property
    def output(self):
        return self._output

    @output.setter
    def output(self, new_output):
        if self._output:
            self._output.close()
        self._output = new_output

    def visit_method(self, node, children):
        self.method = node.text
        return node

    def visit_urlpath(self, node, children):
        path = node.text
        self.context_override.url = urljoin2(self.context_override.url, path)
        return node

    def visit_cd(self, node, children):
        _, _, _, path, _ = children

        if isinstance(path, Node):
            seg = urlparse(self.context_override.url)
            self.context_override.url = seg.scheme + '://' + seg.netloc
        else:
            self.context_override.url = urljoin2(
                self.context_override.url, path)

        return node

    def visit_rm(self, node, children):
        children = children[0]
        kind = children[3].text

        if kind == '*':
            # Clear context
            for target in [self.context.headers,
                           self.context.querystring_params,
                           self.context.body_params,
                           self.context.body_json_params,
                           self.context.options]:
                target.clear()
            return node

        name = children[5]
        if kind == '-h':
            target = self.context.headers
        elif kind == '-q':
            target = self.context.querystring_params
        elif kind == '-o':
            target = self.context.options
        else:
            assert kind == '-b'
            # TODO: This is kind of ugly, will fix it
            if name == '*':
                self.context.body_params.clear()
                self.context.body_json_params.clear()
            else:
                try:
                    del self.context.body_params[name]
                except KeyError:
                    del self.context.body_json_params[name]
            return node

        if name == '*':
            target.clear()
        else:
            del target[name]

        return node

    def visit_help(self, node, children):
        self.output.write(generate_help_text())
        return node

    def _redirect_output(self, filepath, mode):
        filepath = normalize_filepath(filepath)
        self.output = TextWriter(open(os.path.expandvars(filepath), mode))

    def visit_redir_append(self, node, children):
        self._redirect_output(children[3], 'ab')
        return node

    def visit_redir_write(self, node, children):
        self._redirect_output(children[3], 'wb')
        return node

    def visit_pipe(self, node, children):
        cmd = children[3]
        self.pipe_proc = Popen(cmd, shell=True, stdin=PIPE, stdout=PIPE)
        self.output = TextWriter(self.pipe_proc.stdin)
        return node

    def visit_exec(self, node, children):
        path = normalize_filepath(children[3])
        with open(path, encoding='utf-8') as f:
            # Wipe out context first
            execute(