mirror of
https://github.com/Grasscutters/mitmproxy.git
synced 2024-11-26 10:16:27 +00:00
Merge pull request #2288 from cortesi/replayfiles
commands: "replay.client.file" and "replay.server.file"
This commit is contained in:
commit
a32f87fef9
@ -35,6 +35,14 @@ class ClientPlayback:
|
||||
self.flows = flows
|
||||
ctx.master.addons.trigger("update", [])
|
||||
|
||||
@command.command("replay.client.file")
|
||||
def load_file(self, path: str) -> None:
|
||||
try:
|
||||
flows = io.read_flows_from_paths([path])
|
||||
except exceptions.FlowReadException as e:
|
||||
raise exceptions.CommandError(str(e))
|
||||
self.flows = flows
|
||||
|
||||
def configure(self, updated):
|
||||
if not self.configured and ctx.options.client_replay:
|
||||
self.configured = True
|
||||
|
@ -30,6 +30,14 @@ class ServerPlayback:
|
||||
l.append(i)
|
||||
ctx.master.addons.trigger("update", [])
|
||||
|
||||
@command.command("replay.server.file")
|
||||
def load_file(self, path: str) -> None:
|
||||
try:
|
||||
flows = io.read_flows_from_paths([path])
|
||||
except exceptions.FlowReadException as e:
|
||||
raise exceptions.CommandError(str(e))
|
||||
self.load_flows(flows)
|
||||
|
||||
@command.command("replay.server.stop")
|
||||
def clear(self) -> None:
|
||||
"""
|
||||
|
@ -48,6 +48,16 @@ class TestClientPlayback:
|
||||
cp.stop_replay()
|
||||
assert not cp.flows
|
||||
|
||||
def test_load_file(self, tmpdir):
|
||||
cp = clientplayback.ClientPlayback()
|
||||
with taddons.context():
|
||||
fpath = str(tmpdir.join("flows"))
|
||||
tdump(fpath, [tflow.tflow(resp=True)])
|
||||
cp.load_file(fpath)
|
||||
assert cp.flows
|
||||
with pytest.raises(exceptions.CommandError):
|
||||
cp.load_file("/nonexistent")
|
||||
|
||||
def test_configure(self, tmpdir):
|
||||
cp = clientplayback.ClientPlayback()
|
||||
with taddons.context() as tctx:
|
||||
|
@ -16,6 +16,17 @@ def tdump(path, flows):
|
||||
w.add(i)
|
||||
|
||||
|
||||
def test_load_file(tmpdir):
|
||||
s = serverplayback.ServerPlayback()
|
||||
with taddons.context():
|
||||
fpath = str(tmpdir.join("flows"))
|
||||
tdump(fpath, [tflow.tflow(resp=True)])
|
||||
s.load_file(fpath)
|
||||
assert s.flowmap
|
||||
with pytest.raises(exceptions.CommandError):
|
||||
s.load_file("/nonexistent")
|
||||
|
||||
|
||||
def test_config(tmpdir):
|
||||
s = serverplayback.ServerPlayback()
|
||||
with taddons.context() as tctx:
|
||||
|
Loading…
Reference in New Issue
Block a user