mirror of
https://github.com/Grasscutters/mitmproxy.git
synced 2024-11-23 00:01:36 +00:00
Merge pull request #2679 from cortesi/commander2
commands - further progress
This commit is contained in:
Aldo Cortesi
GitHub
commit
367d3a02e1
@ -181,13 +181,18 @@ class CommandManager:
|
||||
|
||||
parse = [] # type: typing.List[ParseResult]
|
||||
params = [] # type: typing.List[type]
|
||||
typ = None # type: typing.Type
|
||||
for i in range(len(parts)):
|
||||
if i == 0:
|
||||
params[:] = [Cmd]
|
||||
typ = Cmd
|
||||
if parts[i] in self.commands:
|
||||
params.extend(self.commands[parts[i]].paramtypes)
|
||||
if params:
|
||||
elif params:
|
||||
typ = params.pop(0)
|
||||
# FIXME: Do we need to check that Arg is positional?
|
||||
if typ == Cmd and params and params[0] == Arg:
|
||||
if parts[i] in self.commands:
|
||||
params[:] = self.commands[parts[i]].paramtypes
|
||||
else:
|
||||
typ = str
|
||||
parse.append(ParseResult(value=parts[i], type=typ))
|
||||
|
||||
@ -1,13 +1,16 @@
|
||||
import abc
|
||||
import glob
|
||||
import os
|
||||
import typing
|
||||
|
||||
import urwid
|
||||
from urwid.text_layout import calc_coords
|
||||
import typing
|
||||
import abc
|
||||
|
||||
import mitmproxy.master
|
||||
import mitmproxy.command
|
||||
|
||||
|
||||
class Completer:
|
||||
class Completer: # pragma: no cover
|
||||
@abc.abstractmethod
|
||||
def cycle(self) -> str:
|
||||
pass
|
||||
@ -24,6 +27,7 @@ class ListCompleter(Completer):
|
||||
for o in options:
|
||||
if o.startswith(start):
|
||||
self.options.append(o)
|
||||
self.options.sort()
|
||||
self.offset = 0
|
||||
|
||||
def cycle(self) -> str:
|
||||
@ -34,6 +38,30 @@ class ListCompleter(Completer):
|
||||
return ret
|
||||
|
||||
|
||||
# Generates the completion options for a specific starting input
|
||||
def pathOptions(start: str) -> typing.Sequence[str]:
|
||||
if not start:
|
||||
start = "./"
|
||||
path = os.path.expanduser(start)
|
||||
ret = []
|
||||
if os.path.isdir(path):
|
||||
files = glob.glob(os.path.join(path, "*"))
|
||||
prefix = start
|
||||
else:
|
||||
files = glob.glob(path + "*")
|
||||
prefix = os.path.dirname(start)
|
||||
prefix = prefix or "./"
|
||||
for f in files:
|
||||
display = os.path.join(prefix, os.path.normpath(os.path.basename(f)))
|
||||
if os.path.isdir(f):
|
||||
display += "/"
|
||||
ret.append(display)
|
||||
if not ret:
|
||||
ret = [start]
|
||||
ret.sort()
|
||||
return ret
|
||||
|
||||
|
||||
CompletionState = typing.NamedTuple(
|
||||
"CompletionState",
|
||||
[
|
||||
@ -93,6 +121,15 @@ class CommandBuffer():
|
||||
),
|
||||
parse = parts,
|
||||
)
|
||||
elif last.type == mitmproxy.command.Path:
|
||||
self.completion = CompletionState(
|
||||
completer = ListCompleter(
|
||||
"",
|
||||
pathOptions(parts[1].value)
|
||||
),
|
||||
parse = parts,
|
||||
)
|
||||
|
||||
if self.completion:
|
||||
nxt = self.completion.completer.cycle()
|
||||
buf = " ".join([i.value for i in self.completion.parse[:-1]]) + " " + nxt
|
||||
@ -120,9 +157,9 @@ class CommandEdit(urwid.WidgetWrap):
|
||||
leader = ": "
|
||||
|
||||
def __init__(self, master: mitmproxy.master.Master, text: str) -> None:
|
||||
super().__init__(urwid.Text(self.leader))
|
||||
self.master = master
|
||||
self.cbuf = CommandBuffer(master, text)
|
||||
self._w = urwid.Text(self.leader)
|
||||
self.update()
|
||||
|
||||
def keypress(self, size, key):
|
||||
|
||||
@ -513,7 +513,7 @@ class ConsoleAddon:
|
||||
kwidget = self.master.window.current("keybindings")
|
||||
if not kwidget:
|
||||
raise exceptions.CommandError("Not viewing key bindings.")
|
||||
f = kwidget.focus()
|
||||
f = kwidget.get_focused_binding()
|
||||
if not f:
|
||||
raise exceptions.CommandError("No key binding focused")
|
||||
return f
|
||||
|
||||
@ -135,7 +135,7 @@ class KeyBindings(urwid.Pile, layoutwidget.LayoutWidget):
|
||||
)
|
||||
self.master = master
|
||||
|
||||
def focus(self):
|
||||
def get_focused_binding(self):
|
||||
if self.focus_position != 0:
|
||||
return None
|
||||
f = self.widget_list[0]
|
||||
|
||||
@ -1,71 +0,0 @@
|
||||
import glob
|
||||
import os.path
|
||||
|
||||
import urwid
|
||||
|
||||
|
||||
class _PathCompleter:
|
||||
|
||||
def __init__(self, _testing=False):
|
||||
"""
|
||||
_testing: disables reloading of the lookup table to make testing
|
||||
possible.
|
||||
"""
|
||||
self.lookup, self.offset = None, None
|
||||
self.final = None
|
||||
self._testing = _testing
|
||||
|
||||
def reset(self):
|
||||
self.lookup = None
|
||||
self.offset = -1
|
||||
|
||||
def complete(self, txt):
|
||||
"""
|
||||
Returns the next completion for txt, or None if there is no
|
||||
completion.
|
||||
"""
|
||||
path = os.path.expanduser(txt)
|
||||
if not self.lookup:
|
||||
if not self._testing:
|
||||
# Lookup is a set of (display value, actual value) tuples.
|
||||
self.lookup = []
|
||||
if os.path.isdir(path):
|
||||
files = glob.glob(os.path.join(path, "*"))
|
||||
prefix = txt
|
||||
else:
|
||||
files = glob.glob(path + "*")
|
||||
prefix = os.path.dirname(txt)
|
||||
prefix = prefix or "./"
|
||||
for f in files:
|
||||
display = os.path.join(prefix, os.path.basename(f))
|
||||
if os.path.isdir(f):
|
||||
display += "/"
|
||||
self.lookup.append((display, f))
|
||||
if not self.lookup:
|
||||
self.final = path
|
||||
return path
|
||||
self.lookup.sort()
|
||||
self.offset = -1
|
||||
self.lookup.append((txt, txt))
|
||||
self.offset += 1
|
||||
if self.offset >= len(self.lookup):
|
||||
self.offset = 0
|
||||
ret = self.lookup[self.offset]
|
||||
self.final = ret[1]
|
||||
return ret[0]
|
||||
|
||||
|
||||
class PathEdit(urwid.Edit, _PathCompleter):
|
||||
|
||||
def __init__(self, prompt, last_path):
|
||||
urwid.Edit.__init__(self, prompt, last_path)
|
||||
_PathCompleter.__init__(self)
|
||||
|
||||
def keypress(self, size, key):
|
||||
if key == "tab":
|
||||
comp = self.complete(self.get_edit_text())
|
||||
self.set_edit_text(comp)
|
||||
self.set_edit_pos(len(comp))
|
||||
else:
|
||||
self.reset()
|
||||
return urwid.Edit.keypress(self, size, key)
|
||||
@ -24,6 +24,10 @@ class TAddon:
|
||||
def cmd3(self, foo: int) -> int:
|
||||
return foo
|
||||
|
||||
@command.command("subcommand")
|
||||
def subcommand(self, cmd: command.Cmd, *args: command.Arg) -> str:
|
||||
return "ok"
|
||||
|
||||
@command.command("empty")
|
||||
def empty(self) -> None:
|
||||
pass
|
||||
@ -102,6 +106,21 @@ class TestCommand:
|
||||
command.ParseResult(value = "", type = int),
|
||||
]
|
||||
],
|
||||
[
|
||||
"subcommand ",
|
||||
[
|
||||
command.ParseResult(value = "subcommand", type = command.Cmd),
|
||||
command.ParseResult(value = "", type = command.Cmd),
|
||||
]
|
||||
],
|
||||
[
|
||||
"subcommand cmd3 ",
|
||||
[
|
||||
command.ParseResult(value = "subcommand", type = command.Cmd),
|
||||
command.ParseResult(value = "cmd3", type = command.Cmd),
|
||||
command.ParseResult(value = "", type = int),
|
||||
]
|
||||
],
|
||||
]
|
||||
with taddons.context() as tctx:
|
||||
tctx.master.addons.add(TAddon())
|
||||
|
||||
@ -1,5 +1,36 @@
|
||||
import os
|
||||
import contextlib
|
||||
|
||||
from mitmproxy.tools.console.commander import commander
|
||||
from mitmproxy.test import taddons
|
||||
from mitmproxy.test import tutils
|
||||
|
||||
|
||||
@contextlib.contextmanager
|
||||
def chdir(path: str):
|
||||
old_dir = os.getcwd()
|
||||
os.chdir(path)
|
||||
yield
|
||||
os.chdir(old_dir)
|
||||
|
||||
|
||||
def normPathOpts(prefix, match):
|
||||
ret = []
|
||||
for s in commander.pathOptions(match):
|
||||
s = s[len(prefix):]
|
||||
s = s.replace(os.sep, "/")
|
||||
ret.append(s)
|
||||
return ret
|
||||
|
||||
|
||||
def test_pathOptions():
|
||||
cd = os.path.normpath(tutils.test_data.path("mitmproxy/completion"))
|
||||
assert normPathOpts(cd, cd) == ['/aaa', '/aab', '/aac', '/bbb/']
|
||||
assert normPathOpts(cd, os.path.join(cd, "a")) == ['/aaa', '/aab', '/aac']
|
||||
with chdir(cd):
|
||||
assert normPathOpts("", "./") == ['./aaa', './aab', './aac', './bbb/']
|
||||
assert normPathOpts("", "") == ['./aaa', './aab', './aac', './bbb/']
|
||||
assert commander.pathOptions("nonexistent") == ["nonexistent"]
|
||||
|
||||
|
||||
class TestListCompleter:
|
||||
@ -46,6 +77,24 @@ class TestCommandBuffer:
|
||||
assert cb.buf == output[0]
|
||||
assert cb.cursor == output[1]
|
||||
|
||||
def test_left(self):
|
||||
cursors = [3, 2, 1, 0, 0]
|
||||
with taddons.context() as tctx:
|
||||
cb = commander.CommandBuffer(tctx.master)
|
||||
cb.buf, cb.cursor = "abcd", 4
|
||||
for c in cursors:
|
||||
cb.left()
|
||||
assert cb.cursor == c
|
||||
|
||||
def test_right(self):
|
||||
cursors = [1, 2, 3, 4, 4]
|
||||
with taddons.context() as tctx:
|
||||
cb = commander.CommandBuffer(tctx.master)
|
||||
cb.buf, cb.cursor = "abcd", 0
|
||||
for c in cursors:
|
||||
cb.right()
|
||||
assert cb.cursor == c
|
||||
|
||||
def test_insert(self):
|
||||
tests = [
|
||||
[("", 0), ("x", 1)],
|
||||
@ -66,3 +115,9 @@ class TestCommandBuffer:
|
||||
cb.buf = "foo bar"
|
||||
cb.cursor = len(cb.buf)
|
||||
cb.cycle_completion()
|
||||
|
||||
def test_render(self):
|
||||
with taddons.context() as tctx:
|
||||
cb = commander.CommandBuffer(tctx.master)
|
||||
cb.buf = "foo"
|
||||
assert cb.render() == "foo"
|
||||
|
||||
@ -1,72 +0,0 @@
|
||||
import os
|
||||
from os.path import normpath
|
||||
from unittest import mock
|
||||
|
||||
from mitmproxy.tools.console import pathedit
|
||||
from mitmproxy.test import tutils
|
||||
|
||||
|
||||
class TestPathCompleter:
|
||||
|
||||
def test_lookup_construction(self):
|
||||
c = pathedit._PathCompleter()
|
||||
|
||||
cd = os.path.normpath(tutils.test_data.path("mitmproxy/completion"))
|
||||
ca = os.path.join(cd, "a")
|
||||
assert c.complete(ca).endswith(normpath("/completion/aaa"))
|
||||
assert c.complete(ca).endswith(normpath("/completion/aab"))
|
||||
c.reset()
|
||||
ca = os.path.join(cd, "aaa")
|
||||
assert c.complete(ca).endswith(normpath("/completion/aaa"))
|
||||
assert c.complete(ca).endswith(normpath("/completion/aaa"))
|
||||
c.reset()
|
||||
assert c.complete(cd).endswith(normpath("/completion/aaa"))
|
||||
|
||||
def test_completion(self):
|
||||
c = pathedit._PathCompleter(True)
|
||||
c.reset()
|
||||
c.lookup = [
|
||||
("a", "x/a"),
|
||||
("aa", "x/aa"),
|
||||
]
|
||||
assert c.complete("a") == "a"
|
||||
assert c.final == "x/a"
|
||||
assert c.complete("a") == "aa"
|
||||
assert c.complete("a") == "a"
|
||||
|
||||
c = pathedit._PathCompleter(True)
|
||||
r = c.complete("l")
|
||||
assert c.final.endswith(r)
|
||||
|
||||
c.reset()
|
||||
assert c.complete("/nonexistent") == "/nonexistent"
|
||||
assert c.final == "/nonexistent"
|
||||
c.reset()
|
||||
assert c.complete("~") != "~"
|
||||
|
||||
c.reset()
|
||||
s = "thisisatotallynonexistantpathforsure"
|
||||
assert c.complete(s) == s
|
||||
assert c.final == s
|
||||
|
||||
|
||||
class TestPathEdit:
|
||||
|
||||
def test_keypress(self):
|
||||
|
||||
pe = pathedit.PathEdit("", "")
|
||||
|
||||
with mock.patch('urwid.widget.Edit.get_edit_text') as get_text, \
|
||||
mock.patch('urwid.widget.Edit.set_edit_text') as set_text:
|
||||
|
||||
cd = os.path.normpath(tutils.test_data.path("mitmproxy/completion"))
|
||||
get_text.return_value = os.path.join(cd, "a")
|
||||
|
||||
# Pressing tab should set completed path
|
||||
pe.keypress((1,), "tab")
|
||||
set_text_called_with = set_text.call_args[0][0]
|
||||
assert set_text_called_with.endswith(normpath("/completion/aaa"))
|
||||
|
||||
# Pressing any other key should reset
|
||||
pe.keypress((1,), "a")
|
||||
assert pe.lookup is None
|
||||
Loading…
Reference in New Issue
Block a user