def test_connect_err(self):
     tutils.raises(
         http.HttpError,
         self.pathoc,
         r"get:'http://foo.com/p/202':da",
         connect_to=("localhost", self.d.port)
     )

 def test_absolute_form_in(self):
     s = StringIO("GET oops-no-protocol.com HTTP/1.1")
     tutils.raises("Bad HTTP request line", HTTPRequest.from_stream, s)
     s = StringIO("GET http://address:22/ HTTP/1.1")
     r = HTTPRequest.from_stream(s)
     assert r.assemble(
     ) == "GET http://address:22/ HTTP/1.1\r\nHost: address:22\r\nContent-Length: 0\r\n\r\n"

 def test_parse_err(self):
     tutils.raises(rparse.ParseException, rparse.parse_response, {}, "400:msg,b:")
     try:
         rparse.parse_response({}, "400'msg':b:")
     except rparse.ParseException, v:
         assert v.marked()
         assert str(v)

def test_client_greeting_assert_socks5():
    raw = tutils.treader("\x00\x00")
    msg = socks.ClientGreeting.from_file(raw)
    tutils.raises(socks.SocksError, msg.assert_socks5)

    raw = tutils.treader("HTTP/1.1 200 OK" + " " * 100)
    msg = socks.ClientGreeting.from_file(raw)
    try:
        msg.assert_socks5()
    except socks.SocksError as e:
        assert "Invalid SOCKS version" in str(e)
        assert "HTTP" not in str(e)
    else:
        assert False

    raw = tutils.treader("GET / HTTP/1.1" + " " * 100)
    msg = socks.ClientGreeting.from_file(raw)
    try:
        msg.assert_socks5()
    except socks.SocksError as e:
        assert "Invalid SOCKS version" in str(e)
        assert "HTTP" in str(e)
    else:
        assert False

    raw = tutils.treader("XX")
    tutils.raises(socks.SocksError, socks.ClientGreeting.from_file, raw, fail_early=True)

 def test_logs(self):
     assert self.d.clear_log()
     tutils.raises("no requests logged", self.d.last_log)
     rsp = self.get("202:da")
     assert len(self.d.log()) == 1
     assert self.d.clear_log()
     assert len(self.d.log()) == 0

 def test_simple(self):
     d = test.Daemon()
     rsp = requests.get("http://localhost:%s/p/202" % d.port)
     assert rsp.ok
     assert rsp.status_code == 202
     d.shutdown()
     tutils.raises(requests.ConnectionError, requests.get, "http://localhost:%s/p/202" % d.port)

def test_read_http_body():
    h = odict.ODictCaseless()
    s = cStringIO.StringIO("testing")
    assert http.read_http_body(500, s, h, False, None) == ""

    h["content-length"] = ["foo"]
    s = cStringIO.StringIO("testing")
    tutils.raises(http.HttpError, http.read_http_body, 500, s, h, False, None)

    h["content-length"] = [5]
    s = cStringIO.StringIO("testing")
    assert len(http.read_http_body(500, s, h, False, None)) == 5
    s = cStringIO.StringIO("testing")
    tutils.raises(http.HttpError, http.read_http_body, 500, s, h, False, 4)

    h = odict.ODictCaseless()
    s = cStringIO.StringIO("testing")
    assert len(http.read_http_body(500, s, h, True, 4)) == 4
    s = cStringIO.StringIO("testing")
    assert len(http.read_http_body(500, s, h, True, 100)) == 7

    h = odict.ODictCaseless()
    h["transfer-encoding"] = ["chunked"]
    s = cStringIO.StringIO("5\r\naaaaa\r\n0\r\n\r\n")
    assert http.read_http_body(500, s, h, True, 100) == "aaaaa"

 def test_noweb(self):
     # FIXME: Add float values to spec language, reduce test timeout to
     # increase test performance
     # This is a bodge - we have some platform difference that causes
     # different exceptions to be raised here.
     tutils.raises(Exception, self.pathoc, ["get:/:p1,1"])
     assert self.d.last_log()["type"] == "timeout"

 def test_concurrent_err(self):
     s = flow.State()
     fm = flow.FlowMaster(None, s)
     tutils.raises(
         "decorator not supported for this method",
         script.Script, tutils.test_data.path("scripts/concurrent_decorator_err.py"), fm
     )

def test_intfield():
    class TT(base.IntField):
        preamble = "t"
        names = {
            "one": 1,
            "two": 2,
            "three": 3
        }
        max = 4
    e = TT.expr()

    v = e.parseString("tone")[0]
    assert v.value == 1
    assert v.spec() == "tone"
    assert v.values(language.Settings())

    v = e.parseString("t1")[0]
    assert v.value == 1
    assert v.spec() == "t1"

    v = e.parseString("t4")[0]
    assert v.value == 4
    assert v.spec() == "t4"

    tutils.raises("can't exceed", e.parseString, "t5")

 def test_reader_read_error(self):
     s = cStringIO.StringIO("foobar\nfoobar")
     s = tcp.Reader(s)
     o = mock.MagicMock()
     o.read = mock.MagicMock(side_effect=socket.error)
     s.o = o
     tutils.raises(tcp.NetLibDisconnect, s.read, 10)

def test_read_chunked():

    h = odict.ODictCaseless()
    h["transfer-encoding"] = ["chunked"]
    s = cStringIO.StringIO("1\r\na\r\n0\r\n")

    tutils.raises("malformed chunked body", http.read_http_body, s, h, None, "GET", None, True)

    s = cStringIO.StringIO("1\r\na\r\n0\r\n\r\n")
    assert http.read_http_body(s, h, None, "GET", None, True) == "a"

    s = cStringIO.StringIO("\r\n\r\n1\r\na\r\n0\r\n\r\n")
    assert http.read_http_body(s, h, None, "GET", None, True) == "a"

    s = cStringIO.StringIO("\r\n")
    tutils.raises("closed prematurely", http.read_http_body, s, h, None, "GET", None, True)

    s = cStringIO.StringIO("1\r\nfoo")
    tutils.raises("malformed chunked body", http.read_http_body, s, h, None, "GET", None, True)

    s = cStringIO.StringIO("foo\r\nfoo")
    tutils.raises(http.HttpError, http.read_http_body, s, h, None, "GET", None, True)

    s = cStringIO.StringIO("5\r\naaaaa\r\n0\r\n\r\n")
    tutils.raises("too large", http.read_http_body, s, h, 2, "GET", None, True)

 def test_parse_err(self):
     tutils.raises(language.ParseException, language.parse_pathod, "400:msg,b:")
     try:
         language.parse_pathod("400'msg':b:")
     except language.ParseException as v:
         assert v.marked()
         assert str(v)

 def test_writer_flush_error(self):
     s = cStringIO.StringIO()
     s = tcp.Writer(s)
     o = mock.MagicMock()
     o.flush = mock.MagicMock(side_effect=socket.error)
     s.o = o
     tutils.raises(tcp.NetLibDisconnect, s.flush)

 def test_connect_err(self):
     tutils.raises(
         HttpException,
         self.pathoc,
         [r"get:'http://foo.com/p/202':da"],
         connect_to=("localhost", self.d.port)
     )

 def test_startstop_ssl(self):
     d = test.Daemon(ssl=True)
     rsp = requests.get("https://localhost:%s/p/202"%d.port, verify=False)
     assert rsp.ok
     assert rsp.status_code == 202
     d.shutdown()
     tutils.raises(requests.ConnectionError, requests.get, "http://localhost:%s/p/202"%d.port)

def test_bidi():
    b = utils.BiDi(a=1, b=2)
    assert b.a == 1
    assert b.get_name(1) == "a"
    assert b.get_name(5) is None
    tutils.raises(AttributeError, getattr, b, "c")
    tutils.raises(ValueError, utils.BiDi, one=1, two=1)

def test_err():
    s = flow.State()
    fm = flow.FlowMaster(None, s)
    sc = script.ScriptContext(fm)

    tutils.raises(
        "not found",
        script.Script, "nonexistent", sc
    )

    tutils.raises(
        "not a file",
        script.Script, tutils.test_data.path("scripts"), sc
    )

    tutils.raises(
        script.ScriptException,
        script.Script, tutils.test_data.path("scripts/syntaxerr.py"), sc
    )

    tutils.raises(
        script.ScriptException,
        script.Script, tutils.test_data.path("scripts/loaderr.py"), sc
    )

    scr = script.Script(tutils.test_data.path("scripts/unloaderr.py"), sc)
    tutils.raises(script.ScriptException, scr.unload)

    def test_ignore(self):
        spec = '304:h"Alternate-Protocol"="mitmproxy-will-remove-this"'
        n = self.pathod(spec)
        self._ignore_on()
        i = self.pathod(spec)
        i2 = self.pathod(spec)
        self._ignore_off()

        assert i.status_code == i2.status_code == n.status_code == 304
        assert "Alternate-Protocol" in i.headers
        assert "Alternate-Protocol" in i2.headers
        assert "Alternate-Protocol" not in n.headers

        # Test that we get the original SSL cert
        if self.ssl:
            i_cert = SSLCert(i.sslinfo.certchain[0])
            i2_cert = SSLCert(i2.sslinfo.certchain[0])
            n_cert = SSLCert(n.sslinfo.certchain[0])

            assert i_cert == i2_cert
            assert i_cert != n_cert

        # Test Non-HTTP traffic
        spec = "200:i0,@100:d0"  # this results in just 100 random bytes
        assert self.pathod(spec).status_code == 502  # mitmproxy responds with bad gateway
        self._ignore_on()
        tutils.raises("invalid server response", self.pathod, spec)  # pathoc tries to parse answer as HTTP
        self._ignore_off()

 def test_echo(self):
     c = tcp.TCPClient(("127.0.0.1", self.port))
     c.connect()
     tutils.raises(
         "cipher specification",
         c.convert_to_ssl,
         sni="foo.com",
         cipher_list="bogus")