<commit_before>from gevent.monkey import patch_all; patch_all()

import os
import time

from io import BytesIO

from pywb.cdx.cdxobject import CDXObject
from pywb.warc.cdxindexer import write_cdx_index
from pywb.utils.bufferedreaders import ChunkedDataReader

from gevent.wsgi import WSGIServer
import gevent

from .testutils import BaseWRTests
from fakeredis import FakeStrictRedis


# ============================================================================
class TestAnonContent(BaseWRTests):
    REDIS_KEYS = [
        'r:{user}:{coll}:{rec}:cdxj',
        'r:{user}:{coll}:{rec}:info',
        'r:{user}:{coll}:{rec}:page',
        'c:{user}:{coll}:warc',
        'c:{user}:{coll}:info',
        'u:{user}',
        'h:roles',
        'h:defaults',
    ]

    @classmethod
    def setup_class(cls):
        agg_port = 30080
        rec_port = 30090

        os.environ['WEBAGG_HOST'] = 'http://localhost:{0}'.format(agg_port)
        os.environ['RECORD_HOST'] = 'http://localhost:{0}'.format(rec_port)

        os.environ['ANON_SLEEP_CHECK'] = '5'

        super(TestAnonContent, cls).setup_class()

        cls.init_webagg(agg_port)
        cls.init_rec(agg_port, rec_port)

    def _get_redis_keys(self, keylist, user, coll, rec):
        keylist = [key.format(user=user, coll=coll, rec=rec) for key in keylist]
        return keylist

    @classmethod
    def init_webagg(cls, port):
        from wrwebagg.wragg import application as webaggapp
        port = cls.make_gevent_server(webaggapp, port)

    @classmethod
    def init_rec(cls, agg_port, rec_port):
        from wrrecorder.main import application as recapp
        port = cls.make_gevent_server(recapp, rec_port)

    @classmethod
    def make_gevent_server(cls, app, port=0):
        server = WSGIServer(('localhost', port), app)

        def run(server):
            print('starting server on ' + str(port))
            server.serve_forever()

        gevent.spawn(run, server)
        return port

    def _assert_rec_keys(self, user, coll, rec_list):
        exp_keys = []

        for rec in rec_list:
            exp_keys.extend(self._get_redis_keys(self.REDIS_KEYS, user, coll, rec))

        res_keys = self.redis.keys()
        res_keys = [k.decode('utf-8') for k in res_keys]

        assert set(exp_keys) == set(res_keys)

    def _assert_size_all_eq(self, user, coll, rec):
        r_info = 'r:{user}:{coll}:{rec}:info'.format(user=user, coll=coll, rec=rec)
        c_info = 'c:{user}:{coll}:info'.format(user=user, coll=coll)
        u_info = 'u:{user}'.format(user=user)

        size = self.redis.hget(r_info, 'size')
        assert size is not None
        assert size == self.redis.hget(c_info, 'size')
        assert size == self.redis.hget(u_info, 'size')

        assert self.redis.hget(r_info, 'updated_at') is not None

    def _get_dechunked(self, stream):
        buff = ChunkedDataReader(BytesIO(stream))

        warcin = BytesIO()
        while True:
            b = buff.read()
            if not b:
                break
            warcin.write(b)

        warcin.seek(0)
        return warcin

    def test_live(self):
        res = self.testapp.get('/live/mp_/http://httpbin.org/get?food=bar')
        res.charset = 'utf-8'

        assert self.testapp.cookies.get('__test_sesh', '') == ''

        assert '"food": "bar"' in res.text, res.text

    def test_live_top_frame(self):
        res = self.testapp.get('/live/http://example.com/')
        res.charset = 'utf-8'
        assert '"http://example.com/"' in res.text
        assert '<iframe' in res.text

    def test_anon_record_redirect(self):
        res = self.testapp.get('/record/mp_/http://example.com/')
        assert res.status_code == 302
        assert res.headers['Location'].endswith('/anonymous/my-recording/record/mp_/http://example.com/')

    def test_anon_replay_redirect(self):
        res = self.testapp.get('/replay/mp_/http://example.com/')
        assert res.status_code == 302
        assert res.headers['Location'].endswith('/anonymous/mp_/http://example.com/')<commit_msg>test: update default rec name<commit_after>from gevent.monkey import patch_all; patch_all()

import os
import time

from io import BytesIO

from pywb.cdx.cdxobject import CDXObject
from pywb.warc.cdxindexer import write_cdx_index
from pywb.utils.bufferedreaders import ChunkedDataReader

from gevent.wsgi import WSGIServer
import gevent

from .testutils import BaseWRTests
from fakeredis import FakeStrictRedis


# ============================================================================
class TestAnonContent(BaseWRTests):
    REDIS_KEYS = [
        'r:{user}:{coll}:{rec}:cdxj',
        'r:{user}:{coll}:{rec}:info',
        'r:{user}:{coll}:{rec}:page',
        'c:{user}:{coll}:warc',
        'c:{user}:{coll}:info',
        'u:{user}',
        'h:roles',
        'h:defaults',
    ]

    @classmethod
    def setup_class(cls):
        agg_port = 30080
        rec_port = 30090

        os.environ['WEBAGG_HOST'] = 'http://localhost:{0}'.format(agg_port)
        os.environ['RECORD_HOST'] = 'http://localhost:{0}'.format(rec_port)

        os.environ['ANON_SLEEP_CHECK'] = '5'

        super(TestAnonContent, cls).setup_class()

        cls.init_webagg(agg_port)
        cls.init_rec(agg_port, rec_port)

    def _get_redis_keys(self, keylist, user, coll, rec):
        keylist = [key.format(user=user, coll=coll, rec=rec) for key in keylist]
        return keylist

    @classmethod
    def init_webagg(cls, port):
        from wrwebagg.wragg import application as webaggapp
        port = cls.make_gevent_server(webaggapp, port)

    @classmethod
    def init_rec(cls, agg_port, rec_port):
        from wrrecorder.main import application as recapp
        port = cls.make_gevent_server(recapp, rec_port)

    @classmethod
    def make_gevent_server(cls, app, port=0):
        server = WSGIServer(('localhost', port), app)

        def run(server):
            print('starting server on ' + str(port))
            server.serve_forever()

        gevent.spawn(run, server)
        return port

    def _assert_rec_keys(self, user, coll, rec_list):
        exp_keys = []

        for rec in rec_list:
            exp_keys.extend(self._get_redis_keys(self.REDIS_KEYS, user, coll, rec))

        res_keys = self.redis.keys()
        res_keys = [k.decode('utf-8') for k in res_keys]

        assert set(exp_keys) == set(res_keys)

    def _assert_size_all_eq(self, user, coll, rec):
        r_info = 'r:{user}:{coll}:{rec}:info'.format(user=user, coll=coll, rec=rec)
        c_info = 'c:{user}:{coll}:info'.format(user=user, coll=coll)
        u_info = 'u:{user}'.format(user=user)

        size = self.redis.hget(r_info, 'size')
        assert size is not None
        assert size == self.redis.hget(c_info, 'size')
        assert size == self.redis.hget(u_info, 'size')

        assert self.redis.hget(r_info, 'updated_at') is not None

    def _get_dechunked(self, stream):
        buff = ChunkedDataReader(BytesIO(stream))

        warcin = BytesIO()
        while True:
            b = buff.read()
            if not b:
                break
            warcin.write(b)

        warcin.seek(0)
        return warcin

    def test_live(self):
        res = self.testapp.get('/live/mp_/http://httpbin.org/get?food=bar')
        res.charset = 'utf-8'

        assert self.testapp.cookies.get('__test_sesh', '') == ''

        assert '"food": "bar"' in res.text, res.text

    def test_live_top_frame(self):
        res = self.testapp.get('/live/http://example.com/')
        res.charset = 'utf-8'
        assert '"http://example.com/"' in res.text
        assert '<iframe' in res.text

    def test_anon_record_redirect(self):
        res = self.testapp.get('/record/mp_/http://example.com/')
        assert res.status_code == 302
        assert res.headers['Location'].endswith('/anonymous/My First Recording/record/mp_/http://example.com/')

    def test_anon_replay_redirect(self):
        res = self.testapp.get('/replay/mp_/http://example.com/')
        assert res.status_code == 302
        assert res.headers['Location'].endswith('/anonymous/mp_/http://example.com/')
{'fraction_non_alphanumeric': 0.152863111329565, 'fraction_numerical': 0.005898255099533055, 'mean_word_length': 2.7680555555555557, 'pattern_counts': {'":': 2, '<': 5, '<?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': '7304805', 'n_tokens_mistral': 2806, 'n_tokens_neox': 2701, 'n_words': 519}