mirror of
https://github.com/Grasscutters/mitmproxy.git
synced 2024-11-23 00:01:36 +00:00
[sans-io] add Layer tests
This commit is contained in:
parent
ed753d1a38
commit
64d2ac8ef3
@ -52,7 +52,7 @@ class Layer:
|
||||
log.log_tier(context.options.console_eventlog_verbosity) >= log.log_tier("debug")
|
||||
)
|
||||
)
|
||||
if show_debug_output:
|
||||
if show_debug_output: # pragma: no cover
|
||||
self.debug = " " * len(context.layers)
|
||||
|
||||
def __repr__(self):
|
||||
|
@ -4,13 +4,63 @@ from mitmproxy.proxy2 import commands, events, layer
|
||||
from test.mitmproxy.proxy2 import tutils
|
||||
|
||||
|
||||
class TestLayer:
|
||||
def test_debug_messages(self, tctx):
|
||||
tctx.server.id = "serverid"
|
||||
|
||||
class TLayer(layer.Layer):
|
||||
debug = " "
|
||||
|
||||
def _handle_event(self, event: events.Event) -> layer.CommandGenerator[None]:
|
||||
yield from self.state(event)
|
||||
|
||||
def state_foo(self, event: events.Event) -> layer.CommandGenerator[None]:
|
||||
assert isinstance(event, events.Start)
|
||||
yield commands.OpenConnection(self.context.server)
|
||||
self.state = self.state_bar
|
||||
|
||||
state = state_foo
|
||||
|
||||
def state_bar(self, event: events.Event) -> layer.CommandGenerator[None]:
|
||||
assert isinstance(event, events.DataReceived)
|
||||
yield commands.Log("baz", "info")
|
||||
|
||||
tlayer = TLayer(tctx)
|
||||
assert (
|
||||
tutils.Playbook(tlayer, hooks=True, logs=True)
|
||||
<< commands.Log(" >> Start({})", "debug")
|
||||
<< commands.Log(" << OpenConnection({'connection': Server({'id': 'serverid', 'address': None})})",
|
||||
"debug")
|
||||
<< commands.OpenConnection(tctx.server)
|
||||
>> events.DataReceived(tctx.client, b"foo")
|
||||
<< commands.Log(" >! DataReceived(client, b'foo')", "debug")
|
||||
>> tutils.reply(None, to=-3)
|
||||
<< commands.Log(" >> Reply(OpenConnection({'connection': Server("
|
||||
"{'id': 'serverid', 'address': None, 'state': <ConnectionState.OPEN: 3>})}))", "debug")
|
||||
<< commands.Log(" !> DataReceived(client, b'foo')", "debug")
|
||||
|
||||
<< commands.Log("baz", "info")
|
||||
)
|
||||
assert repr(tlayer) == "TLayer(state: bar)"
|
||||
|
||||
def test_debug_shorten(self, tctx):
|
||||
t = layer.Layer(tctx)
|
||||
t.debug = " "
|
||||
assert t._Layer__debug("x" * 600).message == " " + "x" * 512 + "…"
|
||||
assert t._Layer__debug("x" * 600).message == " " + "x" * 256 + "…"
|
||||
assert t._Layer__debug("foo").message == " foo"
|
||||
|
||||
|
||||
class TestNextLayer:
|
||||
def test_simple(self, tctx):
|
||||
nl = layer.NextLayer(tctx)
|
||||
nl = layer.NextLayer(tctx, ask_on_start=True)
|
||||
nl.debug = " "
|
||||
playbook = tutils.Playbook(nl, hooks=True)
|
||||
|
||||
assert (
|
||||
playbook
|
||||
<< layer.NextLayerHook(nl)
|
||||
>> tutils.reply()
|
||||
>> events.DataReceived(tctx.client, b"foo")
|
||||
<< layer.NextLayerHook(nl)
|
||||
>> tutils.reply()
|
||||
|
Loading…
Reference in New Issue
Block a user