kylefang/mitmproxy
1
0
Fork 0
mirror of https://github.com/zhigang1992/mitmproxy.git synced 2026-04-24 20:34:56 +08:00
Code Issues Packages Projects Releases Wiki Activity

simplify ALPN and OpenSSL on macOS

Browse Source
This commit is contained in:
Thomas Kriechbaumer
2016-12-01 10:36:18 +01:00
parent 741c2b7b66
commit 1a36efbb6a
10 changed files with 147 additions and 112 deletions
Download Patch File Download Diff File Expand all files Collapse all files

89
test/pathod/test_pathoc.py
View File

@@ -1,8 +1,8 @@
import io
from mock import Mock
import pytest
from mitmproxy.net import http
from mitmproxy.net import tcp
from mitmproxy.net.http import http1
from mitmproxy import exceptions
@@ -11,6 +11,7 @@ from pathod.protocols.http2 import HTTP2StateProtocol
from mitmproxy.test import tutils
from . import tservers
from ..conftest import requires_alpn
def test_response():
@@ -211,45 +212,57 @@ class TestDaemonHTTP2(PathocTestDaemon):
ssl = True
explain = False
if tcp.HAS_ALPN:
@requires_alpn
def test_http2(self):
c = pathoc.Pathoc(
("127.0.0.1", self.d.port),
fp=None,
ssl=True,
use_http2=True,
)
assert isinstance(c.protocol, HTTP2StateProtocol)
def test_http2(self):
c = pathoc.Pathoc(
("127.0.0.1", self.d.port),
fp=None,
ssl=True,
use_http2=True,
)
assert isinstance(c.protocol, HTTP2StateProtocol)
c = pathoc.Pathoc(
("127.0.0.1", self.d.port),
)
assert c.protocol == http1
c = pathoc.Pathoc(
("127.0.0.1", self.d.port),
)
assert c.protocol == http1
@requires_alpn
def test_http2_alpn(self):
c = pathoc.Pathoc(
("127.0.0.1", self.d.port),
fp=None,
ssl=True,
use_http2=True,
http2_skip_connection_preface=True,
)
def test_http2_alpn(self):
c = pathoc.Pathoc(
("127.0.0.1", self.d.port),
fp=None,
ssl=True,
use_http2=True,
http2_skip_connection_preface=True,
)
tmp_convert_to_ssl = c.convert_to_ssl
c.convert_to_ssl = Mock()
c.convert_to_ssl.side_effect = tmp_convert_to_ssl
with c.connect():
_, kwargs = c.convert_to_ssl.call_args
assert set(kwargs['alpn_protos']) == set([b'http/1.1', b'h2'])
tmp_convert_to_ssl = c.convert_to_ssl
c.convert_to_ssl = Mock()
c.convert_to_ssl.side_effect = tmp_convert_to_ssl
@requires_alpn
def test_request(self):
c = pathoc.Pathoc(
("127.0.0.1", self.d.port),
fp=None,
ssl=True,
use_http2=True,
)
with c.connect():
resp = c.request("get:/p/200")
assert resp.status_code == 200
def test_failing_request(self, disable_alpn):
c = pathoc.Pathoc(
("127.0.0.1", self.d.port),
fp=None,
ssl=True,
use_http2=True,
)
with pytest.raises(NotImplementedError):
with c.connect():
_, kwargs = c.convert_to_ssl.call_args
assert set(kwargs['alpn_protos']) == set([b'http/1.1', b'h2'])
def test_request(self):
c = pathoc.Pathoc(
("127.0.0.1", self.d.port),
fp=None,
ssl=True,
use_http2=True,
)
with c.connect():
resp = c.request("get:/p/200")
assert resp.status_code == 200
c.request("get:/p/200")

12
test/pathod/test_pathod.py
View File

@@ -1,11 +1,14 @@
import io
import pytest
from pathod import pathod
from mitmproxy.net import tcp
from mitmproxy import exceptions
from mitmproxy.test import tutils
from . import tservers
from ..conftest import requires_alpn
class TestPathod:
@@ -257,8 +260,11 @@ class TestHTTP2(tservers.DaemonTests):
ssl = True
nohang = True
if tcp.HAS_ALPN:
@requires_alpn
def test_http2(self):
r, _ = self.pathoc(["GET:/"], ssl=True, use_http2=True)
assert r[0].status_code == 800
def test_http2(self):
def test_no_http2(self, disable_alpn):
with pytest.raises(NotImplementedError):
r, _ = self.pathoc(["GET:/"], ssl=True, use_http2=True)
assert r[0].status_code == 800

34
test/pathod/test_protocols_http2.py
View File

@@ -11,6 +11,8 @@ from ..mitmproxy.net import tservers as net_tservers
from pathod.protocols.http2 import HTTP2StateProtocol, TCPHandler
from ..conftest import requires_alpn
class TestTCPHandlerWrapper:
def test_wrapped(self):
@@ -66,37 +68,35 @@ class TestProtocol:
assert mock_server_method.called
@requires_alpn
class TestCheckALPNMatch(net_tservers.ServerTestBase):
handler = EchoHandler
ssl = dict(
alpn_select=b'h2',
)
if tcp.HAS_ALPN:
def test_check_alpn(self):
c = tcp.TCPClient(("127.0.0.1", self.port))
with c.connect():
c.convert_to_ssl(alpn_protos=[b'h2'])
protocol = HTTP2StateProtocol(c)
assert protocol.check_alpn()
def test_check_alpn(self):
c = tcp.TCPClient(("127.0.0.1", self.port))
with c.connect():
c.convert_to_ssl(alpn_protos=[b'h2'])
protocol = HTTP2StateProtocol(c)
assert protocol.check_alpn()
@requires_alpn
class TestCheckALPNMismatch(net_tservers.ServerTestBase):
handler = EchoHandler
ssl = dict(
alpn_select=None,
)
if tcp.HAS_ALPN:
def test_check_alpn(self):
c = tcp.TCPClient(("127.0.0.1", self.port))
with c.connect():
c.convert_to_ssl(alpn_protos=[b'h2'])
protocol = HTTP2StateProtocol(c)
with raises(NotImplementedError):
protocol.check_alpn()
def test_check_alpn(self):
c = tcp.TCPClient(("127.0.0.1", self.port))
with c.connect():
c.convert_to_ssl(alpn_protos=[b'h2'])
protocol = HTTP2StateProtocol(c)
with raises(NotImplementedError):
protocol.check_alpn()
class TestPerformServerConnectionPreface(net_tservers.ServerTestBase):