mirror of
https://github.com/Grasscutters/mitmproxy.git
synced 2024-11-26 18:18:25 +00:00
commander: highlight commands
Step 1 of the fancification of commander.
This commit is contained in:
parent
d98fc4f03a
commit
3341edc9fa
@ -192,12 +192,9 @@ class CommandManager(mitmproxy.types._CommandBase):
|
|||||||
)
|
)
|
||||||
)
|
)
|
||||||
|
|
||||||
remhelp = []
|
remhelp = [] # type: typing.List[str]
|
||||||
for x in params:
|
for x in params:
|
||||||
remt = mitmproxy.types.CommandTypes.get(x, None)
|
remt = mitmproxy.types.CommandTypes.get(x, None)
|
||||||
if not x:
|
|
||||||
remhelp = []
|
|
||||||
break
|
|
||||||
remhelp.append(remt.display)
|
remhelp.append(remt.display)
|
||||||
|
|
||||||
return parse, remhelp
|
return parse, remhelp
|
||||||
|
@ -50,9 +50,9 @@ CompletionState = typing.NamedTuple(
|
|||||||
class CommandBuffer():
|
class CommandBuffer():
|
||||||
def __init__(self, master: mitmproxy.master.Master, start: str = "") -> None:
|
def __init__(self, master: mitmproxy.master.Master, start: str = "") -> None:
|
||||||
self.master = master
|
self.master = master
|
||||||
self.buf = start
|
self.text = self.flatten(start)
|
||||||
# Cursor is always within the range [0:len(buffer)].
|
# Cursor is always within the range [0:len(buffer)].
|
||||||
self._cursor = len(self.buf)
|
self._cursor = len(self.text)
|
||||||
self.completion = None # type: CompletionState
|
self.completion = None # type: CompletionState
|
||||||
|
|
||||||
@property
|
@property
|
||||||
@ -63,13 +63,25 @@ class CommandBuffer():
|
|||||||
def cursor(self, x) -> None:
|
def cursor(self, x) -> None:
|
||||||
if x < 0:
|
if x < 0:
|
||||||
self._cursor = 0
|
self._cursor = 0
|
||||||
elif x > len(self.buf):
|
elif x > len(self.text):
|
||||||
self._cursor = len(self.buf)
|
self._cursor = len(self.text)
|
||||||
else:
|
else:
|
||||||
self._cursor = x
|
self._cursor = x
|
||||||
|
|
||||||
def render(self):
|
def render(self):
|
||||||
return self.buf
|
parts, _ = self.master.commands.parse_partial(self.text)
|
||||||
|
ret = []
|
||||||
|
for p in parts:
|
||||||
|
if p.type == mitmproxy.types.Cmd and p.valid:
|
||||||
|
ret.append(("title", p.value))
|
||||||
|
else:
|
||||||
|
ret.append(("text", p.value))
|
||||||
|
ret.append(("text", " "))
|
||||||
|
return ret
|
||||||
|
|
||||||
|
def flatten(self, txt):
|
||||||
|
parts, _ = self.master.commands.parse_partial(txt)
|
||||||
|
return " ".join([x.value for x in parts])
|
||||||
|
|
||||||
def left(self) -> None:
|
def left(self) -> None:
|
||||||
self.cursor = self.cursor - 1
|
self.cursor = self.cursor - 1
|
||||||
@ -79,7 +91,7 @@ class CommandBuffer():
|
|||||||
|
|
||||||
def cycle_completion(self) -> None:
|
def cycle_completion(self) -> None:
|
||||||
if not self.completion:
|
if not self.completion:
|
||||||
parts, remainhelp = self.master.commands.parse_partial(self.buf[:self.cursor])
|
parts, remainhelp = self.master.commands.parse_partial(self.text[:self.cursor])
|
||||||
last = parts[-1]
|
last = parts[-1]
|
||||||
ct = mitmproxy.types.CommandTypes.get(last.type, None)
|
ct = mitmproxy.types.CommandTypes.get(last.type, None)
|
||||||
if ct:
|
if ct:
|
||||||
@ -94,13 +106,13 @@ class CommandBuffer():
|
|||||||
nxt = self.completion.completer.cycle()
|
nxt = self.completion.completer.cycle()
|
||||||
buf = " ".join([i.value for i in self.completion.parse[:-1]]) + " " + nxt
|
buf = " ".join([i.value for i in self.completion.parse[:-1]]) + " " + nxt
|
||||||
buf = buf.strip()
|
buf = buf.strip()
|
||||||
self.buf = buf
|
self.text = self.flatten(buf)
|
||||||
self.cursor = len(self.buf)
|
self.cursor = len(self.text)
|
||||||
|
|
||||||
def backspace(self) -> None:
|
def backspace(self) -> None:
|
||||||
if self.cursor == 0:
|
if self.cursor == 0:
|
||||||
return
|
return
|
||||||
self.buf = self.buf[:self.cursor - 1] + self.buf[self.cursor:]
|
self.text = self.flatten(self.text[:self.cursor - 1] + self.text[self.cursor:])
|
||||||
self.cursor = self.cursor - 1
|
self.cursor = self.cursor - 1
|
||||||
self.completion = None
|
self.completion = None
|
||||||
|
|
||||||
@ -108,7 +120,7 @@ class CommandBuffer():
|
|||||||
"""
|
"""
|
||||||
Inserts text at the cursor.
|
Inserts text at the cursor.
|
||||||
"""
|
"""
|
||||||
self.buf = self.buf = self.buf[:self.cursor] + k + self.buf[self.cursor:]
|
self.text = self.flatten(self.text[:self.cursor] + k + self.text[self.cursor:])
|
||||||
self.cursor += 1
|
self.cursor += 1
|
||||||
self.completion = None
|
self.completion = None
|
||||||
|
|
||||||
@ -152,4 +164,4 @@ class CommandEdit(urwid.WidgetWrap):
|
|||||||
return x, y
|
return x, y
|
||||||
|
|
||||||
def get_value(self):
|
def get_value(self):
|
||||||
return self.cbuf.buf
|
return self.cbuf.text
|
||||||
|
@ -318,7 +318,10 @@ class _FlowType(_BaseFlowType):
|
|||||||
display = "flow"
|
display = "flow"
|
||||||
|
|
||||||
def parse(self, manager: _CommandBase, t: type, s: str) -> flow.Flow:
|
def parse(self, manager: _CommandBase, t: type, s: str) -> flow.Flow:
|
||||||
flows = manager.call_args("view.resolve", [s])
|
try:
|
||||||
|
flows = manager.call_args("view.resolve", [s])
|
||||||
|
except exceptions.CommandError as e:
|
||||||
|
raise exceptions.TypeError from e
|
||||||
if len(flows) != 1:
|
if len(flows) != 1:
|
||||||
raise exceptions.TypeError(
|
raise exceptions.TypeError(
|
||||||
"Command requires one flow, specification matched %s." % len(flows)
|
"Command requires one flow, specification matched %s." % len(flows)
|
||||||
@ -334,7 +337,10 @@ class _FlowsType(_BaseFlowType):
|
|||||||
display = "[flow]"
|
display = "[flow]"
|
||||||
|
|
||||||
def parse(self, manager: _CommandBase, t: type, s: str) -> typing.Sequence[flow.Flow]:
|
def parse(self, manager: _CommandBase, t: type, s: str) -> typing.Sequence[flow.Flow]:
|
||||||
return manager.call_args("view.resolve", [s])
|
try:
|
||||||
|
return manager.call_args("view.resolve", [s])
|
||||||
|
except exceptions.CommandError as e:
|
||||||
|
raise exceptions.TypeError from e
|
||||||
|
|
||||||
def is_valid(self, manager: _CommandBase, typ: typing.Any, val: typing.Any) -> bool:
|
def is_valid(self, manager: _CommandBase, typ: typing.Any, val: typing.Any) -> bool:
|
||||||
try:
|
try:
|
||||||
|
@ -136,6 +136,8 @@ def test_strseq():
|
|||||||
class DummyConsole:
|
class DummyConsole:
|
||||||
@command.command("view.resolve")
|
@command.command("view.resolve")
|
||||||
def resolve(self, spec: str) -> typing.Sequence[flow.Flow]:
|
def resolve(self, spec: str) -> typing.Sequence[flow.Flow]:
|
||||||
|
if spec == "err":
|
||||||
|
raise mitmproxy.exceptions.CommandError()
|
||||||
n = int(spec)
|
n = int(spec)
|
||||||
return [tflow.tflow(resp=True)] * n
|
return [tflow.tflow(resp=True)] * n
|
||||||
|
|
||||||
@ -157,9 +159,11 @@ def test_flow():
|
|||||||
assert b.is_valid(tctx.master.commands, flow.Flow, tflow.tflow()) is True
|
assert b.is_valid(tctx.master.commands, flow.Flow, tflow.tflow()) is True
|
||||||
assert b.is_valid(tctx.master.commands, flow.Flow, "xx") is False
|
assert b.is_valid(tctx.master.commands, flow.Flow, "xx") is False
|
||||||
with pytest.raises(mitmproxy.exceptions.TypeError):
|
with pytest.raises(mitmproxy.exceptions.TypeError):
|
||||||
assert b.parse(tctx.master.commands, flow.Flow, "0")
|
b.parse(tctx.master.commands, flow.Flow, "0")
|
||||||
with pytest.raises(mitmproxy.exceptions.TypeError):
|
with pytest.raises(mitmproxy.exceptions.TypeError):
|
||||||
assert b.parse(tctx.master.commands, flow.Flow, "2")
|
b.parse(tctx.master.commands, flow.Flow, "2")
|
||||||
|
with pytest.raises(mitmproxy.exceptions.TypeError):
|
||||||
|
b.parse(tctx.master.commands, flow.Flow, "err")
|
||||||
|
|
||||||
|
|
||||||
def test_flows():
|
def test_flows():
|
||||||
@ -175,6 +179,8 @@ def test_flows():
|
|||||||
assert len(b.parse(tctx.master.commands, typing.Sequence[flow.Flow], "0")) == 0
|
assert len(b.parse(tctx.master.commands, typing.Sequence[flow.Flow], "0")) == 0
|
||||||
assert len(b.parse(tctx.master.commands, typing.Sequence[flow.Flow], "1")) == 1
|
assert len(b.parse(tctx.master.commands, typing.Sequence[flow.Flow], "1")) == 1
|
||||||
assert len(b.parse(tctx.master.commands, typing.Sequence[flow.Flow], "2")) == 2
|
assert len(b.parse(tctx.master.commands, typing.Sequence[flow.Flow], "2")) == 2
|
||||||
|
with pytest.raises(mitmproxy.exceptions.TypeError):
|
||||||
|
b.parse(tctx.master.commands, typing.Sequence[flow.Flow], "err")
|
||||||
|
|
||||||
|
|
||||||
def test_data():
|
def test_data():
|
||||||
|
@ -42,16 +42,16 @@ class TestCommandBuffer:
|
|||||||
with taddons.context() as tctx:
|
with taddons.context() as tctx:
|
||||||
for start, output in tests:
|
for start, output in tests:
|
||||||
cb = commander.CommandBuffer(tctx.master)
|
cb = commander.CommandBuffer(tctx.master)
|
||||||
cb.buf, cb.cursor = start[0], start[1]
|
cb.text, cb.cursor = start[0], start[1]
|
||||||
cb.backspace()
|
cb.backspace()
|
||||||
assert cb.buf == output[0]
|
assert cb.text == output[0]
|
||||||
assert cb.cursor == output[1]
|
assert cb.cursor == output[1]
|
||||||
|
|
||||||
def test_left(self):
|
def test_left(self):
|
||||||
cursors = [3, 2, 1, 0, 0]
|
cursors = [3, 2, 1, 0, 0]
|
||||||
with taddons.context() as tctx:
|
with taddons.context() as tctx:
|
||||||
cb = commander.CommandBuffer(tctx.master)
|
cb = commander.CommandBuffer(tctx.master)
|
||||||
cb.buf, cb.cursor = "abcd", 4
|
cb.text, cb.cursor = "abcd", 4
|
||||||
for c in cursors:
|
for c in cursors:
|
||||||
cb.left()
|
cb.left()
|
||||||
assert cb.cursor == c
|
assert cb.cursor == c
|
||||||
@ -60,7 +60,7 @@ class TestCommandBuffer:
|
|||||||
cursors = [1, 2, 3, 4, 4]
|
cursors = [1, 2, 3, 4, 4]
|
||||||
with taddons.context() as tctx:
|
with taddons.context() as tctx:
|
||||||
cb = commander.CommandBuffer(tctx.master)
|
cb = commander.CommandBuffer(tctx.master)
|
||||||
cb.buf, cb.cursor = "abcd", 0
|
cb.text, cb.cursor = "abcd", 0
|
||||||
for c in cursors:
|
for c in cursors:
|
||||||
cb.right()
|
cb.right()
|
||||||
assert cb.cursor == c
|
assert cb.cursor == c
|
||||||
@ -74,20 +74,25 @@ class TestCommandBuffer:
|
|||||||
with taddons.context() as tctx:
|
with taddons.context() as tctx:
|
||||||
for start, output in tests:
|
for start, output in tests:
|
||||||
cb = commander.CommandBuffer(tctx.master)
|
cb = commander.CommandBuffer(tctx.master)
|
||||||
cb.buf, cb.cursor = start[0], start[1]
|
cb.text, cb.cursor = start[0], start[1]
|
||||||
cb.insert("x")
|
cb.insert("x")
|
||||||
assert cb.buf == output[0]
|
assert cb.text == output[0]
|
||||||
assert cb.cursor == output[1]
|
assert cb.cursor == output[1]
|
||||||
|
|
||||||
def test_cycle_completion(self):
|
def test_cycle_completion(self):
|
||||||
with taddons.context() as tctx:
|
with taddons.context() as tctx:
|
||||||
cb = commander.CommandBuffer(tctx.master)
|
cb = commander.CommandBuffer(tctx.master)
|
||||||
cb.buf = "foo bar"
|
cb.text = "foo bar"
|
||||||
cb.cursor = len(cb.buf)
|
cb.cursor = len(cb.text)
|
||||||
cb.cycle_completion()
|
cb.cycle_completion()
|
||||||
|
|
||||||
def test_render(self):
|
def test_render(self):
|
||||||
with taddons.context() as tctx:
|
with taddons.context() as tctx:
|
||||||
cb = commander.CommandBuffer(tctx.master)
|
cb = commander.CommandBuffer(tctx.master)
|
||||||
cb.buf = "foo"
|
cb.text = "foo"
|
||||||
assert cb.render() == "foo"
|
assert cb.render()
|
||||||
|
|
||||||
|
def test_flatten(self):
|
||||||
|
with taddons.context() as tctx:
|
||||||
|
cb = commander.CommandBuffer(tctx.master)
|
||||||
|
assert cb.flatten("foo bar") == "foo bar"
|
||||||
|
Loading…
Reference in New Issue
Block a user