aboutsummaryrefslogtreecommitdiffstats
path: root/test
diff options
context:
space:
mode:
Diffstat (limited to 'test')
-rw-r--r--test/test_netlib.py93
-rw-r--r--test/test_protocol.py163
2 files changed, 0 insertions, 256 deletions
diff --git a/test/test_netlib.py b/test/test_netlib.py
deleted file mode 100644
index 19902d17..00000000
--- a/test/test_netlib.py
+++ /dev/null
@@ -1,93 +0,0 @@
-import cStringIO, threading, Queue
-from libmproxy import netlib
-import tutils
-
-class ServerThread(threading.Thread):
- def __init__(self, server):
- self.server = server
- threading.Thread.__init__(self)
-
- def run(self):
- self.server.serve_forever()
-
- def shutdown(self):
- self.server.shutdown()
-
-
-class ServerTestBase:
- @classmethod
- def setupAll(cls):
- cls.server = ServerThread(cls.makeserver())
- cls.server.start()
-
- @classmethod
- def teardownAll(cls):
- cls.server.shutdown()
-
-
-class THandler(netlib.BaseHandler):
- def handle(self):
- v = self.rfile.readline()
- if v.startswith("echo"):
- self.wfile.write(v)
- elif v.startswith("error"):
- raise ValueError("Testing an error.")
- self.wfile.flush()
-
-
-class TServer(netlib.TCPServer):
- def __init__(self, addr, q):
- netlib.TCPServer.__init__(self, addr)
- self.q = q
-
- def handle_connection(self, request, client_address):
- THandler(request, client_address, self)
-
- def handle_error(self, request, client_address):
- s = cStringIO.StringIO()
- netlib.TCPServer.handle_error(self, request, client_address, s)
- self.q.put(s.getvalue())
-
-
-class TestServer(ServerTestBase):
- @classmethod
- def makeserver(cls):
- cls.q = Queue.Queue()
- s = TServer(("127.0.0.1", 0), cls.q)
- cls.port = s.port
- return s
-
- def test_echo(self):
- testval = "echo!\n"
- c = netlib.TCPClient(False, "127.0.0.1", self.port, None)
- c.wfile.write(testval)
- c.wfile.flush()
- assert c.rfile.readline() == testval
-
- def test_error(self):
- testval = "error!\n"
- c = netlib.TCPClient(False, "127.0.0.1", self.port, None)
- c.wfile.write(testval)
- c.wfile.flush()
- assert "Testing an error" in self.q.get()
-
-
-class TestTCPClient:
- def test_conerr(self):
- tutils.raises(netlib.NetLibError, netlib.TCPClient, False, "127.0.0.1", 0, None)
-
-
-class TestFileLike:
- def test_wrap(self):
- s = cStringIO.StringIO("foobar\nfoobar")
- s = netlib.FileLike(s)
- s.flush()
- assert s.readline() == "foobar\n"
- assert s.readline() == "foobar"
- # Test __getattr__
- assert s.isatty
-
- def test_limit(self):
- s = cStringIO.StringIO("foobar\nfoobar")
- s = netlib.FileLike(s)
- assert s.readline(3) == "foo"
diff --git a/test/test_protocol.py b/test/test_protocol.py
deleted file mode 100644
index 81b5fefb..00000000
--- a/test/test_protocol.py
+++ /dev/null
@@ -1,163 +0,0 @@
-import cStringIO, textwrap
-from libmproxy import protocol, flow
-import tutils
-
-def test_has_chunked_encoding():
- h = flow.ODictCaseless()
- assert not protocol.has_chunked_encoding(h)
- h["transfer-encoding"] = ["chunked"]
- assert protocol.has_chunked_encoding(h)
-
-
-def test_read_chunked():
- s = cStringIO.StringIO("1\r\na\r\n0\r\n")
- tutils.raises(IOError, protocol.read_chunked, s, None)
-
- s = cStringIO.StringIO("1\r\na\r\n0\r\n\r\n")
- assert protocol.read_chunked(s, None) == "a"
-
- s = cStringIO.StringIO("\r\n")
- tutils.raises(IOError, protocol.read_chunked, s, None)
-
- s = cStringIO.StringIO("1\r\nfoo")
- tutils.raises(IOError, protocol.read_chunked, s, None)
-
- s = cStringIO.StringIO("foo\r\nfoo")
- tutils.raises(protocol.ProtocolError, protocol.read_chunked, s, None)
-
-
-def test_request_connection_close():
- h = flow.ODictCaseless()
- assert protocol.request_connection_close((1, 0), h)
- assert not protocol.request_connection_close((1, 1), h)
-
- h["connection"] = ["keep-alive"]
- assert not protocol.request_connection_close((1, 1), h)
-
-
-def test_read_http_body():
- h = flow.ODict()
- s = cStringIO.StringIO("testing")
- assert protocol.read_http_body(s, h, False, None) == ""
-
- h["content-length"] = ["foo"]
- s = cStringIO.StringIO("testing")
- tutils.raises(protocol.ProtocolError, protocol.read_http_body, s, h, False, None)
-
- h["content-length"] = [5]
- s = cStringIO.StringIO("testing")
- assert len(protocol.read_http_body(s, h, False, None)) == 5
- s = cStringIO.StringIO("testing")
- tutils.raises(protocol.ProtocolError, protocol.read_http_body, s, h, False, 4)
-
- h = flow.ODict()
- s = cStringIO.StringIO("testing")
- assert len(protocol.read_http_body(s, h, True, 4)) == 4
- s = cStringIO.StringIO("testing")
- assert len(protocol.read_http_body(s, h, True, 100)) == 7
-
-def test_parse_http_protocol():
- assert protocol.parse_http_protocol("HTTP/1.1") == (1, 1)
- assert protocol.parse_http_protocol("HTTP/0.0") == (0, 0)
- assert not protocol.parse_http_protocol("foo/0.0")
-
-
-def test_parse_init_connect():
- assert protocol.parse_init_connect("CONNECT host.com:443 HTTP/1.0")
- assert not protocol.parse_init_connect("bogus")
- assert not protocol.parse_init_connect("GET host.com:443 HTTP/1.0")
- assert not protocol.parse_init_connect("CONNECT host.com443 HTTP/1.0")
- assert not protocol.parse_init_connect("CONNECT host.com:443 foo/1.0")
-
-
-def test_prase_init_proxy():
- u = "GET http://foo.com:8888/test HTTP/1.1"
- m, s, h, po, pa, httpversion = protocol.parse_init_proxy(u)
- assert m == "GET"
- assert s == "http"
- assert h == "foo.com"
- assert po == 8888
- assert pa == "/test"
- assert httpversion == (1, 1)
-
- assert not protocol.parse_init_proxy("invalid")
- assert not protocol.parse_init_proxy("GET invalid HTTP/1.1")
- assert not protocol.parse_init_proxy("GET http://foo.com:8888/test foo/1.1")
-
-
-def test_parse_init_http():
- u = "GET /test HTTP/1.1"
- m, u, httpversion= protocol.parse_init_http(u)
- assert m == "GET"
- assert u == "/test"
- assert httpversion == (1, 1)
-
- assert not protocol.parse_init_http("invalid")
- assert not protocol.parse_init_http("GET invalid HTTP/1.1")
- assert not protocol.parse_init_http("GET /test foo/1.1")
-
-
-class TestReadHeaders:
- def test_read_simple(self):
- data = """
- Header: one
- Header2: two
- \r\n
- """
- data = textwrap.dedent(data)
- data = data.strip()
- s = cStringIO.StringIO(data)
- h = protocol.read_headers(s)
- assert h == [["Header", "one"], ["Header2", "two"]]
-
- def test_read_multi(self):
- data = """
- Header: one
- Header: two
- \r\n
- """
- data = textwrap.dedent(data)
- data = data.strip()
- s = cStringIO.StringIO(data)
- h = protocol.read_headers(s)
- assert h == [["Header", "one"], ["Header", "two"]]
-
- def test_read_continued(self):
- data = """
- Header: one
- \ttwo
- Header2: three
- \r\n
- """
- data = textwrap.dedent(data)
- data = data.strip()
- s = cStringIO.StringIO(data)
- h = protocol.read_headers(s)
- assert h == [["Header", "one\r\n two"], ["Header2", "three"]]
-
-
-def test_parse_url():
- assert not protocol.parse_url("")
-
- u = "http://foo.com:8888/test"
- s, h, po, pa = protocol.parse_url(u)
- assert s == "http"
- assert h == "foo.com"
- assert po == 8888
- assert pa == "/test"
-
- s, h, po, pa = protocol.parse_url("http://foo/bar")
- assert s == "http"
- assert h == "foo"
- assert po == 80
- assert pa == "/bar"
-
- s, h, po, pa = protocol.parse_url("http://foo")
- assert pa == "/"
-
- s, h, po, pa = protocol.parse_url("https://foo")
- assert po == 443
-
- assert not protocol.parse_url("https://foo:bar")
- assert not protocol.parse_url("https://foo:")
-
generated by cgit v1.2.3 (git 2.25.1) at 2025-09-09 19:54:13 +0000