Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_typecheck(self):
with taddons.context(loadcore=False) as tctx:
cm = command.CommandManager(tctx.master)
a = TypeErrAddon()
command.Command(cm, "noret", a.noret)
with pytest.raises(exceptions.CommandError):
command.Command(cm, "invalidret", a.invalidret)
with pytest.raises(exceptions.CommandError):
command.Command(cm, "invalidarg", a.invalidarg)
v.request(tft(method="put", start=4))
assert [i.request.timestamp_start for i in v] == [1, 2, 3, 4]
v.set_order("method")
assert v.get_order() == "method"
assert [i.request.method for i in v] == ["GET", "GET", "PUT", "PUT"]
v.set_reversed(True)
assert [i.request.method for i in v] == ["PUT", "PUT", "GET", "GET"]
v.set_order("time")
assert v.get_order() == "time"
assert [i.request.timestamp_start for i in v] == [4, 3, 2, 1]
v.set_reversed(False)
assert [i.request.timestamp_start for i in v] == [1, 2, 3, 4]
with pytest.raises(exceptions.CommandError):
v.set_order("not_an_order")
def test_script_run_nonexistent(self):
sc = script.ScriptLoader()
with taddons.context():
with pytest.raises(exceptions.CommandError):
sc.script_run([tflow.tflow(resp=True)], "/")
def read_file(filename: str, escaped: bool) -> typing.AnyStr:
filename = os.path.expanduser(filename)
try:
with open(filename, "r" if escaped else "rb") as f:
d = f.read()
except IOError as v:
raise exceptions.CommandError(v)
if escaped:
try:
d = strutils.escaped_str_to_bytes(d)
except ValueError:
raise exceptions.CommandError("Invalid Python-style string encoding.")
return d
def flowview_mode_set(self, mode: str) -> None:
"""
Set the display mode for the current flow view.
"""
fv = self.master.window.current_window("flowview")
if not fv:
raise exceptions.CommandError("Not viewing a flow.")
idx = fv.body.tab_offset
if mode not in [i.name.lower() for i in contentviews.views]:
raise exceptions.CommandError("Invalid flowview mode.")
try:
self.master.commands.call_strings(
"view.setval",
["@focus", f"flowview_mode_{idx}", mode]
)
except exceptions.CommandError as e:
signals.status_message.send(message=str(e))
def load_file(self, path: mitmproxy.types.Path) -> None:
"""
Load flows from file, and add them to the replay queue.
"""
try:
flows = io.read_flows_from_paths([path])
except exceptions.FlowReadException as e:
raise exceptions.CommandError(str(e))
self.start_replay(flows)
if spec == "@all":
return [i for i in self._store.values()]
if spec == "@focus":
return [self.focus.flow] if self.focus.flow else []
elif spec == "@shown":
return [i for i in self]
elif spec == "@hidden":
return [i for i in self._store.values() if i not in self._view]
elif spec == "@marked":
return [i for i in self._store.values() if i.marked]
elif spec == "@unmarked":
return [i for i in self._store.values() if not i.marked]
else:
filt = flowfilter.parse(spec)
if not filt:
raise exceptions.CommandError("Invalid flow filter: %s" % spec)
return [i for i in self._store.values() if filt(i)]
def keymap_load_path(self, path: mitmproxy.types.Path) -> None:
try:
self.load_path(ctx.master.keymap, path) # type: ignore
except (OSError, KeyBindingError) as e:
raise exceptions.CommandError(
"Could not load key bindings - %s" % e
) from e
def get_command_by_path(self, path: str) -> Command:
"""
Returns command by its path. May raise CommandError.
"""
if path not in self.commands:
raise exceptions.CommandError(f"Unknown command: {path}")
return self.commands[path]