aboutsummaryrefslogtreecommitdiffstats
path: root/test/netlib/websockets
diff options
context:
space:
mode:
authorAldo Cortesi <aldo@nullcube.com>2016-10-20 11:56:38 +1300
committerAldo Cortesi <aldo@nullcube.com>2016-10-20 11:56:38 +1300
commit8430f857b504a3e7406dc36e54dc32783569d0dd (patch)
treed3116cd540faf01f272a0892fc6a9b83b4f6de8a /test/netlib/websockets
parent853e03a5e753354fad3a3fa5384ef3a09384ef43 (diff)
downloadmitmproxy-8430f857b504a3e7406dc36e54dc32783569d0dd.tar.gz
mitmproxy-8430f857b504a3e7406dc36e54dc32783569d0dd.tar.bz2
mitmproxy-8430f857b504a3e7406dc36e54dc32783569d0dd.zip
The final piece: netlib -> mitproxy.net
Diffstat (limited to 'test/netlib/websockets')
-rw-r--r--test/netlib/websockets/__init__.py0
-rw-r--r--test/netlib/websockets/test_frame.py164
-rw-r--r--test/netlib/websockets/test_masker.py23
-rw-r--r--test/netlib/websockets/test_utils.py105
4 files changed, 0 insertions, 292 deletions
diff --git a/test/netlib/websockets/__init__.py b/test/netlib/websockets/__init__.py
deleted file mode 100644
index e69de29b..00000000
--- a/test/netlib/websockets/__init__.py
+++ /dev/null
diff --git a/test/netlib/websockets/test_frame.py b/test/netlib/websockets/test_frame.py
deleted file mode 100644
index 3b7c9ed4..00000000
--- a/test/netlib/websockets/test_frame.py
+++ /dev/null
@@ -1,164 +0,0 @@
-import os
-import codecs
-import pytest
-
-from netlib import websockets
-from mitmproxy.test import tutils
-
-
-class TestFrameHeader:
-
- @pytest.mark.parametrize("input,expected", [
- (0, '0100'),
- (125, '017D'),
- (126, '017E007E'),
- (127, '017E007F'),
- (142, '017E008E'),
- (65534, '017EFFFE'),
- (65535, '017EFFFF'),
- (65536, '017F0000000000010000'),
- (8589934591, '017F00000001FFFFFFFF'),
- (2 ** 64 - 1, '017FFFFFFFFFFFFFFFFF'),
- ])
- def test_serialization_length(self, input, expected):
- h = websockets.FrameHeader(
- opcode=websockets.OPCODE.TEXT,
- payload_length=input,
- )
- assert bytes(h) == codecs.decode(expected, 'hex')
-
- def test_serialization_too_large(self):
- h = websockets.FrameHeader(
- payload_length=2 ** 64 + 1,
- )
- with pytest.raises(ValueError):
- bytes(h)
-
- @pytest.mark.parametrize("input,expected", [
- ('0100', 0),
- ('017D', 125),
- ('017E007E', 126),
- ('017E007F', 127),
- ('017E008E', 142),
- ('017EFFFE', 65534),
- ('017EFFFF', 65535),
- ('017F0000000000010000', 65536),
- ('017F00000001FFFFFFFF', 8589934591),
- ('017FFFFFFFFFFFFFFFFF', 2 ** 64 - 1),
- ])
- def test_deserialization_length(self, input, expected):
- h = websockets.FrameHeader.from_file(tutils.treader(codecs.decode(input, 'hex')))
- assert h.payload_length == expected
-
- @pytest.mark.parametrize("input,expected", [
- ('0100', (False, None)),
- ('018000000000', (True, '00000000')),
- ('018012345678', (True, '12345678')),
- ])
- def test_deserialization_masking(self, input, expected):
- h = websockets.FrameHeader.from_file(tutils.treader(codecs.decode(input, 'hex')))
- assert h.mask == expected[0]
- if h.mask:
- assert h.masking_key == codecs.decode(expected[1], 'hex')
-
- def test_equality(self):
- h = websockets.FrameHeader(mask=True, masking_key=b'1234')
- h2 = websockets.FrameHeader(mask=True, masking_key=b'1234')
- assert h == h2
-
- h = websockets.FrameHeader(fin=True)
- h2 = websockets.FrameHeader(fin=False)
- assert h != h2
-
- assert h != 'foobar'
-
- def test_roundtrip(self):
- def round(*args, **kwargs):
- h = websockets.FrameHeader(*args, **kwargs)
- h2 = websockets.FrameHeader.from_file(tutils.treader(bytes(h)))
- assert h == h2
-
- round()
- round(fin=True)
- round(rsv1=True)
- round(rsv2=True)
- round(rsv3=True)
- round(payload_length=1)
- round(payload_length=100)
- round(payload_length=1000)
- round(payload_length=10000)
- round(opcode=websockets.OPCODE.PING)
- round(masking_key=b"test")
-
- def test_human_readable(self):
- f = websockets.FrameHeader(
- masking_key=b"test",
- fin=True,
- payload_length=10
- )
- assert repr(f)
-
- f = websockets.FrameHeader()
- assert repr(f)
-
- def test_funky(self):
- f = websockets.FrameHeader(masking_key=b"test", mask=False)
- raw = bytes(f)
- f2 = websockets.FrameHeader.from_file(tutils.treader(raw))
- assert not f2.mask
-
- def test_violations(self):
- tutils.raises("opcode", websockets.FrameHeader, opcode=17)
- tutils.raises("masking key", websockets.FrameHeader, masking_key=b"x")
-
- def test_automask(self):
- f = websockets.FrameHeader(mask=True)
- assert f.masking_key
-
- f = websockets.FrameHeader(masking_key=b"foob")
- assert f.mask
-
- f = websockets.FrameHeader(masking_key=b"foob", mask=0)
- assert not f.mask
- assert f.masking_key
-
-
-class TestFrame:
- def test_equality(self):
- f = websockets.Frame(payload=b'1234')
- f2 = websockets.Frame(payload=b'1234')
- assert f == f2
-
- assert f != b'1234'
-
- def test_roundtrip(self):
- def round(*args, **kwargs):
- f = websockets.Frame(*args, **kwargs)
- raw = bytes(f)
- f2 = websockets.Frame.from_file(tutils.treader(raw))
- assert f == f2
- round(b"test")
- round(b"test", fin=1)
- round(b"test", rsv1=1)
- round(b"test", opcode=websockets.OPCODE.PING)
- round(b"test", masking_key=b"test")
-
- def test_human_readable(self):
- f = websockets.Frame()
- assert repr(f)
-
- f = websockets.Frame(b"foobar")
- assert "foobar" in repr(f)
-
- @pytest.mark.parametrize("masked", [True, False])
- @pytest.mark.parametrize("length", [100, 50000, 150000])
- def test_serialization_bijection(self, masked, length):
- frame = websockets.Frame(
- os.urandom(length),
- fin=True,
- opcode=websockets.OPCODE.TEXT,
- mask=int(masked),
- masking_key=(os.urandom(4) if masked else None)
- )
- serialized = bytes(frame)
- assert frame == websockets.Frame.from_bytes(serialized)
diff --git a/test/netlib/websockets/test_masker.py b/test/netlib/websockets/test_masker.py
deleted file mode 100644
index 23a9b876..00000000
--- a/test/netlib/websockets/test_masker.py
+++ /dev/null
@@ -1,23 +0,0 @@
-import codecs
-import pytest
-
-from netlib import websockets
-
-
-class TestMasker:
-
- @pytest.mark.parametrize("input,expected", [
- ([b"a"], '00'),
- ([b"four"], '070d1616'),
- ([b"fourf"], '070d161607'),
- ([b"fourfive"], '070d1616070b1501'),
- ([b"a", b"aasdfasdfa", b"asdf"], '000302170504021705040205120605'),
- ([b"a" * 50, b"aasdfasdfa", b"asdf"], '00030205000302050003020500030205000302050003020500030205000302050003020500030205000302050003020500030205120605051206050500110702'), # noqa
- ])
- def test_masker(self, input, expected):
- m = websockets.Masker(b"abcd")
- data = b"".join([m(t) for t in input])
- assert data == codecs.decode(expected, 'hex')
-
- data = websockets.Masker(b"abcd")(data)
- assert data == b"".join(input)
diff --git a/test/netlib/websockets/test_utils.py b/test/netlib/websockets/test_utils.py
deleted file mode 100644
index f6f1e40a..00000000
--- a/test/netlib/websockets/test_utils.py
+++ /dev/null
@@ -1,105 +0,0 @@
-import pytest
-
-from netlib import http
-from netlib import websockets
-
-
-class TestUtils:
-
- def test_client_handshake_headers(self):
- h = websockets.client_handshake_headers(version='42')
- assert h['sec-websocket-version'] == '42'
-
- h = websockets.client_handshake_headers(key='some-key')
- assert h['sec-websocket-key'] == 'some-key'
-
- h = websockets.client_handshake_headers(protocol='foobar')
- assert h['sec-websocket-protocol'] == 'foobar'
-
- h = websockets.client_handshake_headers(extensions='foo; bar')
- assert h['sec-websocket-extensions'] == 'foo; bar'
-
- def test_server_handshake_headers(self):
- h = websockets.server_handshake_headers('some-key')
- assert h['sec-websocket-accept'] == '8iILEZtcVdtFD7MDlPKip9ec9nw='
- assert 'sec-websocket-protocol' not in h
- assert 'sec-websocket-extensions' not in h
-
- h = websockets.server_handshake_headers('some-key', 'foobar', 'foo; bar')
- assert h['sec-websocket-accept'] == '8iILEZtcVdtFD7MDlPKip9ec9nw='
- assert h['sec-websocket-protocol'] == 'foobar'
- assert h['sec-websocket-extensions'] == 'foo; bar'
-
- @pytest.mark.parametrize("input,expected", [
- ([(b'connection', b'upgrade'), (b'upgrade', b'websocket'), (b'sec-websocket-key', b'foobar')], True),
- ([(b'connection', b'upgrade'), (b'upgrade', b'websocket'), (b'sec-websocket-accept', b'foobar')], True),
- ([(b'Connection', b'UpgRaDe'), (b'Upgrade', b'WebSocKeT'), (b'Sec-WebSockeT-KeY', b'foobar')], True),
- ([(b'Connection', b'UpgRaDe'), (b'Upgrade', b'WebSocKeT'), (b'Sec-WebSockeT-AccePt', b'foobar')], True),
- ([(b'connection', b'foo'), (b'upgrade', b'bar'), (b'sec-websocket-key', b'foobar')], False),
- ([(b'connection', b'upgrade'), (b'upgrade', b'websocket')], False),
- ([(b'connection', b'upgrade'), (b'sec-websocket-key', b'foobar')], False),
- ([(b'upgrade', b'websocket'), (b'sec-websocket-key', b'foobar')], False),
- ([], False),
- ])
- def test_check_handshake(self, input, expected):
- h = http.Headers(input)
- assert websockets.check_handshake(h) == expected
-
- @pytest.mark.parametrize("input,expected", [
- ([(b'sec-websocket-version', b'13')], True),
- ([(b'Sec-WebSockeT-VerSion', b'13')], True),
- ([(b'sec-websocket-version', b'9')], False),
- ([(b'sec-websocket-version', b'42')], False),
- ([(b'sec-websocket-version', b'')], False),
- ([], False),
- ])
- def test_check_client_version(self, input, expected):
- h = http.Headers(input)
- assert websockets.check_client_version(h) == expected
-
- @pytest.mark.parametrize("input,expected", [
- ('foobar', b'AzhRPA4TNwR6I/riJheN0TfR7+I='),
- (b'foobar', b'AzhRPA4TNwR6I/riJheN0TfR7+I='),
- ])
- def test_create_server_nonce(self, input, expected):
- assert websockets.create_server_nonce(input) == expected
-
- @pytest.mark.parametrize("input,expected", [
- ([(b'sec-websocket-extensions', b'foo; bar')], 'foo; bar'),
- ([(b'Sec-WebSockeT-ExteNsionS', b'foo; bar')], 'foo; bar'),
- ([(b'sec-websocket-extensions', b'')], ''),
- ([], None),
- ])
- def test_get_extensions(self, input, expected):
- h = http.Headers(input)
- assert websockets.get_extensions(h) == expected
-
- @pytest.mark.parametrize("input,expected", [
- ([(b'sec-websocket-protocol', b'foobar')], 'foobar'),
- ([(b'Sec-WebSockeT-ProTocoL', b'foobar')], 'foobar'),
- ([(b'sec-websocket-protocol', b'')], ''),
- ([], None),
- ])
- def test_get_protocol(self, input, expected):
- h = http.Headers(input)
- assert websockets.get_protocol(h) == expected
-
- @pytest.mark.parametrize("input,expected", [
- ([(b'sec-websocket-key', b'foobar')], 'foobar'),
- ([(b'Sec-WebSockeT-KeY', b'foobar')], 'foobar'),
- ([(b'sec-websocket-key', b'')], ''),
- ([], None),
- ])
- def test_get_client_key(self, input, expected):
- h = http.Headers(input)
- assert websockets.get_client_key(h) == expected
-
- @pytest.mark.parametrize("input,expected", [
- ([(b'sec-websocket-accept', b'foobar')], 'foobar'),
- ([(b'Sec-WebSockeT-AccepT', b'foobar')], 'foobar'),
- ([(b'sec-websocket-accept', b'')], ''),
- ([], None),
- ])
- def test_get_server_accept(self, input, expected):
- h = http.Headers(input)
- assert websockets.get_server_accept(h) == expected