MCPcopy
hub / github.com/mitmproxy/mitmproxy / Playbook

Class Playbook

test/mitmproxy/proxy/tutils.py:109–304  ·  view source on GitHub ↗

Assert that a layer emits the expected commands in reaction to a given sequence of events. For example, the following code asserts that the TCP layer emits an OpenConnection command immediately after starting and does not yield any further commands as a reaction to successful connec

Source from the content-addressed store, hash-verified

107
108
109class Playbook:
110 """
111 Assert that a layer emits the expected commands in reaction to a given sequence of events.
112 For example, the following code asserts that the TCP layer emits an OpenConnection command
113 immediately after starting and does not yield any further commands as a reaction to successful
114 connection establishment.
115
116 assert playbook(tcp.TCPLayer(tctx)) \
117 << commands.OpenConnection(tctx.server)
118 >> reply(None)
119 << None # this line is optional.
120
121 This is syntactic sugar for the following:
122
123 t = tcp.TCPLayer(tctx)
124 x1 = list(t.handle_event(events.Start()))
125 assert x1 == [commands.OpenConnection(tctx.server)]
126 x2 = list(t.handle_event(events.OpenConnectionReply(x1[-1])))
127 assert x2 == []
128 """
129
130 layer: Layer
131 """The base layer"""
132 expected: PlaybookEntryList
133 """expected command/event sequence"""
134 actual: PlaybookEntryList
135 """actual command/event sequence"""
136 _errored: bool
137 """used to check if playbook as been fully asserted"""
138 logs: bool
139 """If False, the playbook specification doesn't contain log commands."""
140 hooks: bool
141 """If False, the playbook specification doesn't include hooks or hook replies. They are automatically replied to."""
142
143 def __init__(
144 self,
145 layer: Layer,
146 hooks: bool = True,
147 logs: bool = False,
148 expected: PlaybookEntryList | None = None,
149 ):
150 if expected is None:
151 expected = [events.Start()]
152
153 self.layer = layer
154 self.expected = expected
155 self.actual = []
156 self._errored = False
157 self.logs = logs
158 self.hooks = hooks
159
160 def __rshift__(self, e):
161 """Add an event to send"""
162 if isinstance(e, collections.abc.Iterable):
163 for ev in e:
164 self.__rshift__(ev)
165 return self
166 assert isinstance(e, events.Event)

Callers 15

test_upgradeFunction · 0.90
test_upgrade_streamedFunction · 0.90
ws_testdataFunction · 0.90
test_upstream_httpsFunction · 0.90
test_reverse_proxyFunction · 0.90
test_reverse_dnsFunction · 0.90

Calls

no outgoing calls

Tested by 15

test_upgradeFunction · 0.72
test_upgrade_streamedFunction · 0.72
ws_testdataFunction · 0.72
test_upstream_httpsFunction · 0.72
test_reverse_proxyFunction · 0.72
test_reverse_dnsFunction · 0.72

Used in the wild real call sites across dependent graphs

searching dependent graphs…