<commit_before>import os
from cStringIO import StringIO
import libpry
from libmproxy import dump, flow, proxy
import tutils

def test_strfuncs():
    t = tutils.tresp()
    t._set_replay()
    dump.str_response(t)

    t = tutils.treq()
    t.client_conn = None
    t.stickycookie = True
    assert "stickycookie" in dump.str_request(t)
    assert "replay" in dump.str_request(t)


class TestDumpMaster:
    def _cycle(self, m, content):
        req = tutils.treq()
        req.content = content
        l = proxy.Log("connect")
        m.handle_log(l)
        cc = req.client_conn
        cc.connection_error = "error"
        resp = tutils.tresp(req)
        resp.content = content
        m.handle_clientconnect(cc)
        m.handle_request(req)
        f = m.handle_response(resp)
        m.handle_clientdisconnect(flow.ClientDisconnect(cc))
        return f

    def _dummy_cycle(self, n, filt, content, **options):
        cs = StringIO()
        o = dump.Options(**options)
        m = dump.DumpMaster(None, o, filt, outfile=cs)
        for i in range(n):
            self._cycle(m, content)
        return cs.getvalue()

    def _flowfile(self, path):
        f = open(path, "w")
        fw = flow.FlowWriter(f)
        t = tutils.tflow_full()
        t.response = tutils.tresp(t.request)
        fw.add(t)
        f.close()

    def test_replay(self):
        cs = StringIO()

        o = dump.Options(server_replay="nonexistent", kill=True)
        libpry.raises(dump.DumpError, dump.DumpMaster, None, o, None, outfile=cs)

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

            o = dump.Options(server_replay=p, kill=True)
            m = dump.DumpMaster(None, o, None, outfile=cs)

            self._cycle(m, "content")
            self._cycle(m, "content")

            o = dump.Options(server_replay=p, kill=False)
            m = dump.DumpMaster(None, o, None, outfile=cs)
            self._cycle(m, "nonexistent")

            o = dump.Options(client_replay=p, kill=False)
            m = dump.DumpMaster(None, o, None, outfile=cs)

    def test_read(self):
        with tutils.tmpdir() as t:
            p = os.path.join(t, "read")
            self._flowfile(p)
            assert "GET" in self._dummy_cycle(0, None, "", verbosity=1, rfile=p)

            libpry.raises(
                dump.DumpError, self._dummy_cycle,
                0, None, "", verbosity=1, rfile="/nonexistent"
            )

            libpry.raises(
                dump.DumpError, self._dummy_cycle,
                0, None, "", verbosity=1, rfile=tutils.test_data.path("test_dump.py")
            )

    def test_options(self):
        o = dump.Options(verbosity = 2)
        assert o.verbosity == 2
        libpry.raises(AttributeError, dump.Options, nonexistent = 2)

    def test_filter(self):
        assert not "GET" in self._dummy_cycle(1, "~u foo", "", verbosity=1)

    def test_replacements(self):
        o = dump.Options(replacements=[(".*", "content", "foo")])
        m = dump.DumpMaster(None, o, None)
        f = self._cycle(m, "content")
        assert f.request.content == "foo"

    def test_basic(self):
        for i in (1, 2, 3):
            assert "GET" in self._dummy_cycle(1, "~s", "", verbosity=i, eventlog=True)
            assert "GET" in self._dummy_cycle(1, "~s", "\x00\x00\x00", verbosity=i)
            assert "GET" in self._dummy_cycle(1, "~s", "ascii", verbosity=i)

    def test_write(self):<commit_msg>Fix flow read unit test to accomodate more tolerant dumpfile parsing.<commit_after>import os
from cStringIO import StringIO
import libpry
from libmproxy import dump, flow, proxy
import tutils

def test_strfuncs():
    t = tutils.tresp()
    t._set_replay()
    dump.str_response(t)

    t = tutils.treq()
    t.client_conn = None
    t.stickycookie = True
    assert "stickycookie" in dump.str_request(t)
    assert "replay" in dump.str_request(t)


class TestDumpMaster:
    def _cycle(self, m, content):
        req = tutils.treq()
        req.content = content
        l = proxy.Log("connect")
        m.handle_log(l)
        cc = req.client_conn
        cc.connection_error = "error"
        resp = tutils.tresp(req)
        resp.content = content
        m.handle_clientconnect(cc)
        m.handle_request(req)
        f = m.handle_response(resp)
        m.handle_clientdisconnect(flow.ClientDisconnect(cc))
        return f

    def _dummy_cycle(self, n, filt, content, **options):
        cs = StringIO()
        o = dump.Options(**options)
        m = dump.DumpMaster(None, o, filt, outfile=cs)
        for i in range(n):
            self._cycle(m, content)
        return cs.getvalue()

    def _flowfile(self, path):
        f = open(path, "w")
        fw = flow.FlowWriter(f)
        t = tutils.tflow_full()
        t.response = tutils.tresp(t.request)
        fw.add(t)
        f.close()

    def test_replay(self):
        cs = StringIO()

        o = dump.Options(server_replay="nonexistent", kill=True)
        libpry.raises(dump.DumpError, dump.DumpMaster, None, o, None, outfile=cs)

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

            o = dump.Options(server_replay=p, kill=True)
            m = dump.DumpMaster(None, o, None, outfile=cs)

            self._cycle(m, "content")
            self._cycle(m, "content")

            o = dump.Options(server_replay=p, kill=False)
            m = dump.DumpMaster(None, o, None, outfile=cs)
            self._cycle(m, "nonexistent")

            o = dump.Options(client_replay=p, kill=False)
            m = dump.DumpMaster(None, o, None, outfile=cs)

    def test_read(self):
        with tutils.tmpdir() as t:
            p = os.path.join(t, "read")
            self._flowfile(p)
            assert "GET" in self._dummy_cycle(0, None, "", verbosity=1, rfile=p)

            libpry.raises(
                dump.DumpError, self._dummy_cycle,
                0, None, "", verbosity=1, rfile="/nonexistent"
            )

            # We now just ignore errors
            self._dummy_cycle(0, None, "", verbosity=1, rfile=tutils.test_data.path("test_dump.py"))

    def test_options(self):
        o = dump.Options(verbosity = 2)
        assert o.verbosity == 2
        libpry.raises(AttributeError, dump.Options, nonexistent = 2)

    def test_filter(self):
        assert not "GET" in self._dummy_cycle(1, "~u foo", "", verbosity=1)

    def test_replacements(self):
        o = dump.Options(replacements=[(".*", "content", "foo")])
        m = dump.DumpMaster(None, o, None)
        f = self._cycle(m, "content")
        assert f.request.content == "foo"

    def test_basic(self):
        for i in (1, 2, 3):
            assert "GET" in self._dummy_cycle(1, "~s", "", verbosity=i, eventlog=True)
            assert "GET" in self._dummy_cycle(1, "~s", "\x00\x00\x00", verbosity=i)
            assert "GET" in self._dummy_cycle(1, "~s", "ascii", verbosity=i)

    def test_write(self):
{'fraction_non_alphanumeric': 0.13097446140676272, 'fraction_numerical': 0.006562990440861749, 'mean_word_length': 2.212648945921173, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5241476', 'n_tokens_mistral': 2405, 'n_tokens_neox': 2327, 'n_words': 515}