def test_simple(self):
        r = replace.ReplaceFile()
        with tutils.tmpdir() as td:
            rp = os.path.join(td, "replacement")
            with open(rp, "w") as f:
                f.write("bar")
            with taddons.context() as tctx:
                tctx.configure(
                    r,
                    replacement_files = [
                        ("~q", "foo", rp),
                        ("~s", "foo", rp),
                        ("~b nonexistent", "nonexistent", "nonexistent"),
                    ]
                )
                f = tflow.tflow()
                f.request.content = b"foo"
                r.request(f)
                assert f.request.content == b"bar"

                f = tflow.tflow(resp=True)
                f.response.content = b"foo"
                r.response(f)
                assert f.response.content == b"bar"

                f = tflow.tflow()
                f.request.content = b"nonexistent"
                assert not tctx.master.event_log
                r.request(f)
                assert tctx.master.event_log

    def test_replay(self):
        o = dump.Options(server_replay=["nonexistent"], replay_kill_extra=True)
        tutils.raises(exceptions.OptionsError, dump.DumpMaster, o, proxy.DummyServer())

        with tutils.tmpdir() as t:
            p = os.path.join(t, "rep")
            self.flowfile(p)

            o = dump.Options(server_replay=[p], replay_kill_extra=True)
            o.verbosity = 0
            o.flow_detail = 0
            m = dump.DumpMaster(o, proxy.DummyServer())

            self.cycle(m, b"content")
            self.cycle(m, b"content")

            o = dump.Options(server_replay=[p], replay_kill_extra=False)
            o.verbosity = 0
            o.flow_detail = 0
            m = dump.DumpMaster(o, proxy.DummyServer())
            self.cycle(m, b"nonexistent")

            o = dump.Options(client_replay=[p], replay_kill_extra=False)
            o.verbosity = 0
            o.flow_detail = 0
            m = dump.DumpMaster(o, proxy.DummyServer())

    def test_no_script_file(self):
        with pytest.raises(Exception, match="not found"):
            script.parse_command("notfound")

        with tutils.tmpdir() as dir:
            with pytest.raises(Exception, match="Not a file"):
                script.parse_command(dir)

    def test_access_control(self):
        v = base.TokValue.parseString("<path")[0]
        with tutils.tmpdir() as t:
            p = os.path.join(t, "path")
            with open(p, "wb") as f:
                f.write(b"x" * 10000)

            assert v.get_generator(language.Settings(staticdir=t))

            v = base.TokValue.parseString("<path2")[0]
            tutils.raises(
                exceptions.FileAccessDenied,
                v.get_generator,
                language.Settings(staticdir=t)
            )
            tutils.raises(
                "access disabled",
                v.get_generator,
                language.Settings()
            )

            v = base.TokValue.parseString("</outside")[0]
            tutils.raises(
                "outside",
                v.get_generator,
                language.Settings(staticdir=t)
            )

    def test_no_script_file(self):
        with tutils.raises("not found"):
            script.parse_command("notfound")

        with tutils.tmpdir() as dir:
            with tutils.raises("not a file"):
                script.parse_command(dir)

 def test_client_certs(self):
     with tutils.tmpdir() as cadir:
         self.assert_noerr("--client-certs", cadir)
         self.assert_noerr(
             "--client-certs",
             os.path.join(tutils.test_data.path("mitmproxy/data/clientcert"), "client.pem"))
         with pytest.raises(Exception, match="path does not exist"):
             self.p("--client-certs", "nonexistent")

def test_config():
    s = serverplayback.ServerPlayback()
    with tutils.tmpdir() as p:
        with taddons.context() as tctx:
            fpath = os.path.join(p, "flows")
            tdump(fpath, [tflow.tflow(resp=True)])
            tctx.configure(s, server_replay = [fpath])
            tutils.raises(exceptions.OptionsError, tctx.configure, s, server_replay = [p])

 def test_sans(self):
     with tutils.tmpdir() as d:
         ca = certs.CertStore.from_store(d, "test")
         c1 = ca.get_cert(b"foo.com", [b"*.bar.com"])
         ca.get_cert(b"foo.bar.com", [])
         # assert c1 == c2
         c3 = ca.get_cert(b"bar.com", [])
         assert not c1 == c3

    def test_create_tmp(self):
        with tutils.tmpdir() as d:
            ca = certs.CertStore.from_store(d, "test")
            assert ca.get_cert(b"foo.com", [])
            assert ca.get_cert(b"foo.com", [])
            assert ca.get_cert(b"*.foo.com", [])

            r = ca.get_cert(b"*.foo.com", [])
            assert r[1] == ca.default_privatekey

    def test_create_explicit(self):
        with tutils.tmpdir() as d:
            ca = certs.CertStore.from_store(d, "test")
            assert ca.get_cert(b"foo", [])

            ca2 = certs.CertStore.from_store(d, "test")
            assert ca2.get_cert(b"foo", [])

            assert ca.default_ca.get_serial_number() == ca2.default_ca.get_serial_number()

 def test_client_certs(self):
     with tutils.tmpdir() as cadir:
         self.assert_noerr("--client-certs", cadir)
         self.assert_noerr(
             "--client-certs",
             os.path.join(tutils.test_data.path("mitmproxy/data/clientcert"), "client.pem"))
         self.assert_err(
             "path does not exist",
             "--client-certs",
             "nonexistent")

 def test_configure(self):
     cp = clientplayback.ClientPlayback()
     with taddons.context() as tctx:
         with tutils.tmpdir() as td:
             path = os.path.join(td, "flows")
             tdump(path, [tflow.tflow()])
             tctx.configure(cp, client_replay=[path])
             tctx.configure(cp, client_replay=[])
             tctx.configure(cp)
             with pytest.raises(exceptions.OptionsError):
                 tctx.configure(cp, client_replay=["nonexistent"])

    def test_simple(self):
        with tutils.tmpdir() as tdir:
            path = os.path.join(tdir, "somefile")

            m, sc = tscript("complex/har_dump.py", shlex.quote(path))
            m.addons.invoke(m, "response", self.flow())
            m.addons.remove(sc)

            with open(path, "r") as inp:
                har = json.load(inp)

        assert len(har["log"]["entries"]) == 1

    def test_base64(self):
        with tutils.tmpdir() as tdir:
            path = os.path.join(tdir, "somefile")

            m, sc = tscript("complex/har_dump.py", shlex.quote(path))
            m.addons.invoke(m, "response", self.flow(resp_content=b"foo" + b"\xFF" * 10))
            m.addons.remove(sc)

            with open(path, "r") as inp:
                har = json.load(inp)

        assert har["log"]["entries"][0]["response"]["content"]["encoding"] == "base64"

def test_tcp():
    sa = streamfile.StreamFile()
    with taddons.context() as tctx:
        with tutils.tmpdir() as tdir:
            p = os.path.join(tdir, "foo")
            tctx.configure(sa, streamfile=p)

            tt = tflow.ttcpflow()
            sa.tcp_start(tt)
            sa.tcp_end(tt)
            tctx.configure(sa, streamfile=None)
            assert rd(p)

def test_configure():
    sa = streamfile.StreamFile()
    with taddons.context(options=dump.Options()) as tctx:
        with tutils.tmpdir() as tdir:
            p = os.path.join(tdir, "foo")
            tutils.raises(
                exceptions.OptionsError,
                tctx.configure, sa, streamfile=tdir
            )
            tutils.raises(
                "invalid filter",
                tctx.configure, sa, streamfile=p, filtstr="~~"
            )

def test_configure():
    sa = streamfile.StreamFile()
    with taddons.context(options=dump.Options()) as tctx:
        with tutils.tmpdir() as tdir:
            p = os.path.join(tdir, "foo")
            with pytest.raises(exceptions.OptionsError):
                tctx.configure(sa, streamfile=tdir)
            with pytest.raises(Exception, match="Invalid filter"):
                tctx.configure(sa, streamfile=p, filtstr="~~")
            tctx.configure(sa, filtstr="foo")
            assert sa.filt
            tctx.configure(sa, filtstr=None)
            assert not sa.filt

def test_saving():
    o = TD2()
    o.three = "set"
    with tutils.tmpdir() as tdir:
        dst = os.path.join(tdir, "conf")
        o.save(dst, defaults=True)

        o2 = TD2()
        o2.load_paths(dst)
        o2.three = "foo"
        o2.save(dst, defaults=True)

        o.load_paths(dst)
        assert o.three == "foo"

 def test_reload(self):
     with taddons.context() as tctx:
         with tutils.tmpdir():
             with open("foo.py", "w"):
                 pass
             sc = script.Script("foo.py")
             tctx.configure(sc)
             for _ in range(100):
                 with open("foo.py", "a") as f:
                     f.write(".")
                 sc.tick()
                 time.sleep(0.1)
                 if tctx.master.event_log:
                     return
             raise AssertionError("Change event not detected.")

    def test_overrides(self):
        with tutils.tmpdir() as d:
            ca1 = certs.CertStore.from_store(os.path.join(d, "ca1"), "test")
            ca2 = certs.CertStore.from_store(os.path.join(d, "ca2"), "test")
            assert not ca1.default_ca.get_serial_number(
            ) == ca2.default_ca.get_serial_number()

            dc = ca2.get_cert(b"foo.com", [b"sans.example.com"])
            dcp = os.path.join(d, "dc")
            f = open(dcp, "wb")
            f.write(dc[0].to_pem())
            f.close()
            ca1.add_cert_file(b"foo.com", dcp)

            ret = ca1.get_cert(b"foo.com", [])
            assert ret[0].serial == dc[0].serial