diff options
| author | Aldo Cortesi <aldo@nullcube.com> | 2015-05-05 10:47:02 +1200 |
|---|---|---|
| committer | Aldo Cortesi <aldo@nullcube.com> | 2015-05-05 10:47:02 +1200 |
| commit | f2bc58cdd2f2b9b0025a88c0faccf55e10b29353 (patch) | |
| tree | 9353f4b78f4bf8ec1e6b4169155da084a2c05ea9 /test/test_socks.py | |
| parent | 08b2e2a6a98fd175e1b49d62dffde34e91c77b1c (diff) | |
| download | mitmproxy-f2bc58cdd2f2b9b0025a88c0faccf55e10b29353.tar.gz mitmproxy-f2bc58cdd2f2b9b0025a88c0faccf55e10b29353.tar.bz2 mitmproxy-f2bc58cdd2f2b9b0025a88c0faccf55e10b29353.zip | |
Add tcp.Reader.safe_read, use it in socks and websockets
safe_read is guaranteed to raise or return a byte string of the
requested length. It's particularly useful for implementing binary
protocols.
Diffstat (limited to 'test/test_socks.py')
| -rw-r--r-- | test/test_socks.py | 16 |
1 files changed, 8 insertions, 8 deletions
diff --git a/test/test_socks.py b/test/test_socks.py index aa4f9c11..6e522826 100644 --- a/test/test_socks.py +++ b/test/test_socks.py @@ -7,7 +7,7 @@ import tutils def test_client_greeting(): - raw = StringIO("\x05\x02\x00\xBE\xEF") + raw = tutils.treader("\x05\x02\x00\xBE\xEF") out = StringIO() msg = socks.ClientGreeting.from_file(raw) msg.to_file(out) @@ -20,7 +20,7 @@ def test_client_greeting(): def test_server_greeting(): - raw = StringIO("\x05\x02") + raw = tutils.treader("\x05\x02") out = StringIO() msg = socks.ServerGreeting.from_file(raw) msg.to_file(out) @@ -31,7 +31,7 @@ def test_server_greeting(): def test_message(): - raw = StringIO("\x05\x01\x00\x03\x0bexample.com\xDE\xAD\xBE\xEF") + raw = tutils.treader("\x05\x01\x00\x03\x0bexample.com\xDE\xAD\xBE\xEF") out = StringIO() msg = socks.Message.from_file(raw) assert raw.read(2) == "\xBE\xEF" @@ -46,7 +46,7 @@ def test_message(): def test_message_ipv4(): # Test ATYP=0x01 (IPV4) - raw = StringIO("\x05\x01\x00\x01\x7f\x00\x00\x01\xDE\xAD\xBE\xEF") + raw = tutils.treader("\x05\x01\x00\x01\x7f\x00\x00\x01\xDE\xAD\xBE\xEF") out = StringIO() msg = socks.Message.from_file(raw) assert raw.read(2) == "\xBE\xEF" @@ -62,7 +62,7 @@ def test_message_ipv6(): # Test ATYP=0x04 (IPV6) ipv6_addr = "2001:db8:85a3:8d3:1319:8a2e:370:7344" - raw = StringIO("\x05\x01\x00\x04" + socket.inet_pton(socket.AF_INET6, ipv6_addr) + "\xDE\xAD\xBE\xEF") + raw = tutils.treader("\x05\x01\x00\x04" + socket.inet_pton(socket.AF_INET6, ipv6_addr) + "\xDE\xAD\xBE\xEF") out = StringIO() msg = socks.Message.from_file(raw) assert raw.read(2) == "\xBE\xEF" @@ -73,12 +73,12 @@ def test_message_ipv6(): def test_message_invalid_rsv(): - raw = StringIO("\x05\x01\xFF\x01\x7f\x00\x00\x01\xDE\xAD\xBE\xEF") + raw = tutils.treader("\x05\x01\xFF\x01\x7f\x00\x00\x01\xDE\xAD\xBE\xEF") tutils.raises(socks.SocksError, socks.Message.from_file, raw) def test_message_unknown_atyp(): - raw = StringIO("\x05\x02\x00\x02\x7f\x00\x00\x01\xDE\xAD\xBE\xEF") + raw = tutils.treader("\x05\x02\x00\x02\x7f\x00\x00\x01\xDE\xAD\xBE\xEF") tutils.raises(socks.SocksError, socks.Message.from_file, raw) m = socks.Message(5, 1, 0x02, tcp.Address(("example.com", 5050))) @@ -93,4 +93,4 @@ def test_read(): cs = mock.Mock() cs.read = mock.Mock(side_effect=socket.error) - tutils.raises(socks.SocksError, socks._read, cs, 4)
\ No newline at end of file + tutils.raises(socks.SocksError, socks._read, cs, 4) |