mirror of
https://github.com/webrecorder/pywb.git
synced 2025-03-24 06:59:52 +01:00
5b72d96871
It seems like JSON in responses is now minimized without spaces? I'm not sure what changed in the stack to do that.
57 lines
2.0 KiB
Python
57 lines
2.0 KiB
Python
from .base_config_test import BaseConfigTest, CollsDirMixin, BaseTestClass
|
|
from pywb.manager.manager import main as manager
|
|
from pywb.warcserver.test.testutils import to_path, HttpBinLiveTests, FakeRedisTests
|
|
|
|
from fakeredis import FakeStrictRedis
|
|
|
|
from warcio import ArchiveIterator
|
|
|
|
import os
|
|
import time
|
|
import json
|
|
|
|
import pytest
|
|
|
|
|
|
# ============================================================================
|
|
class TestRecordDedup(HttpBinLiveTests, CollsDirMixin, BaseConfigTest, FakeRedisTests, BaseTestClass):
|
|
@classmethod
|
|
def setup_class(cls):
|
|
super(TestRecordDedup, cls).setup_class('config_test_record_dedup.yaml', custom_config={'recorder': 'live'})
|
|
cls.redis = FakeStrictRedis.from_url("redis://localhost/0")
|
|
|
|
def test_init_coll(self):
|
|
manager(['init', 'test-dedup'])
|
|
assert os.path.isdir(os.path.join(self.root_dir, '_test_colls', 'test-dedup', 'archive'))
|
|
|
|
def test_record_1(self):
|
|
res = self.testapp.get('/test-dedup/record/mp_/http://httpbin.org/get?A=B', headers={"Referer": "http://httpbin.org/"})
|
|
assert '"A":"B"' in res.text
|
|
|
|
time.sleep(1.2)
|
|
|
|
res = self.testapp.get('/test-dedup/record/mp_/http://httpbin.org/get?A=B', headers={"Referer": "http://httpbin.org/"})
|
|
assert '"A":"B"' in res.text
|
|
|
|
def test_single_redis_entry(self):
|
|
res = self.redis.zrange("pywb:test-dedup:cdxj", 0, -1)
|
|
assert len(res) == 1
|
|
|
|
def test_single_warc_record(self):
|
|
dir_name = os.path.join(self.root_dir, '_test_colls', 'test-dedup', 'archive')
|
|
files = os.listdir(dir_name)
|
|
assert len(files) == 1
|
|
|
|
records = []
|
|
|
|
with open(os.path.join(dir_name, files[0]), 'rb') as fh:
|
|
for record in ArchiveIterator(fh):
|
|
records.append(record.rec_type)
|
|
|
|
# ensure only one response/request pair written
|
|
assert records == ['response', 'request']
|
|
|
|
def test_redis_pending_count(self):
|
|
res = self.redis.get("pywb:test-dedup:pending")
|
|
assert res == b'0'
|