mirror of
https://github.com/vee1e/mitmproxy.git
synced 2026-09-05 04:07:20 +00:00
68 lines
2.3 KiB
Python
68 lines
2.3 KiB
Python
import pytest
|
|
|
|
from mitmproxy import eventsequence
|
|
from mitmproxy.proxy import layers
|
|
from mitmproxy.test import tflow
|
|
|
|
|
|
@pytest.mark.parametrize("resp, err", [
|
|
(False, False),
|
|
(True, False),
|
|
(False, True),
|
|
(True, True),
|
|
])
|
|
def test_http_flow(resp, err):
|
|
f = tflow.tflow(resp=resp, err=err)
|
|
i = eventsequence.iterate(f)
|
|
assert isinstance(next(i), layers.http.HttpRequestHeadersHook)
|
|
assert isinstance(next(i), layers.http.HttpRequestHook)
|
|
if resp:
|
|
assert isinstance(next(i), layers.http.HttpResponseHeadersHook)
|
|
assert isinstance(next(i), layers.http.HttpResponseHook)
|
|
if err:
|
|
assert isinstance(next(i), layers.http.HttpErrorHook)
|
|
|
|
|
|
@pytest.mark.parametrize("err", [False, True])
|
|
def test_websocket_flow(err):
|
|
f = tflow.twebsocketflow(err=err)
|
|
i = eventsequence.iterate(f)
|
|
|
|
assert isinstance(next(i), layers.http.HttpRequestHeadersHook)
|
|
assert isinstance(next(i), layers.http.HttpRequestHook)
|
|
assert isinstance(next(i), layers.http.HttpResponseHeadersHook)
|
|
assert isinstance(next(i), layers.http.HttpResponseHook)
|
|
|
|
assert isinstance(next(i), layers.websocket.WebsocketStartHook)
|
|
assert len(f.websocket.messages) == 0
|
|
assert isinstance(next(i), layers.websocket.WebsocketMessageHook)
|
|
assert len(f.websocket.messages) == 1
|
|
assert isinstance(next(i), layers.websocket.WebsocketMessageHook)
|
|
assert len(f.websocket.messages) == 2
|
|
assert isinstance(next(i), layers.websocket.WebsocketMessageHook)
|
|
assert len(f.websocket.messages) == 3
|
|
if err:
|
|
assert isinstance(next(i), layers.websocket.WebsocketErrorHook)
|
|
else:
|
|
assert isinstance(next(i), layers.websocket.WebsocketEndHook)
|
|
|
|
|
|
@pytest.mark.parametrize("err", [False, True])
|
|
def test_tcp_flow(err):
|
|
f = tflow.ttcpflow(err=err)
|
|
i = eventsequence.iterate(f)
|
|
assert isinstance(next(i), layers.tcp.TcpStartHook)
|
|
assert len(f.messages) == 0
|
|
assert isinstance(next(i), layers.tcp.TcpMessageHook)
|
|
assert len(f.messages) == 1
|
|
assert isinstance(next(i), layers.tcp.TcpMessageHook)
|
|
assert len(f.messages) == 2
|
|
if err:
|
|
assert isinstance(next(i), layers.tcp.TcpErrorHook)
|
|
else:
|
|
assert isinstance(next(i), layers.tcp.TcpEndHook)
|
|
|
|
|
|
def test_invalid():
|
|
with pytest.raises(TypeError):
|
|
next(eventsequence.iterate(42))
|