from .base_config_test import BaseConfigTest, CollsDirMixin, fmod import os import tempfile import shutil import sys import webtest import time import gevent from six import StringIO import webtest from pytest import raises from mock import patch from pywb import get_test_dir from pywb.warcserver.test.testutils import BaseTestClass from pywb.manager.autoindex import AutoIndexer from pywb.manager.manager import main from pywb.indexer.cdxindexer import main as cdxindexer_main from pywb.warcserver.index.cdxobject import CDXObject from pywb.apps.frontendapp import FrontEndApp #============================================================================= ARCHIVE_DIR = 'archive' INDEX_DIR = 'indexes' COLLECTIONS = '_test_colls' INDEX_FILE = 'index.cdxj' AUTOINDEX_FILE = 'autoindex.cdxj' #============================================================================= class TestManagedColls(CollsDirMixin, BaseConfigTest): @classmethod def setup_class(cls): super(TestManagedColls, cls).setup_class('config_test.yaml') def _check_dirs(self, base, dirlist): for dir_ in dirlist: assert os.path.isdir(os.path.join(base, dir_)) def _get_sample_warc(self, name): return os.path.join(get_test_dir(), 'warcs', name) @patch('pywb.apps.cli.BaseCli.run_gevent', lambda *args, **kwargs: None) def test_run_cli(self): """ test new wayback cli interface test autoindex error before collections inited """ from pywb.apps.cli import wayback wayback(['-p', '0']) # Nothing to auto-index.. yet with raises(SystemExit): wayback(['-a', '-p', '0']) colls = os.path.join(self.root_dir, COLLECTIONS) os.mkdir(colls) wayback(['-a', '-p', '0', '--auto-interval', '0']) def test_create_first_coll(self): """ Test first collection creation, with all required dirs """ main(['init', 'test']) colls = os.path.join(self.root_dir, COLLECTIONS) assert os.path.isdir(colls) test = os.path.join(colls, 'test') assert os.path.isdir(test) self._check_dirs(test, [INDEX_DIR, ARCHIVE_DIR, 'static', 'templates']) def test_add_warcs(self): """ Test adding warc to new coll, check replay """ warc1 = self._get_sample_warc('example.warc.gz') main(['add', 'test', warc1]) def test_add_warcs_replay(self, fmod): resp = self.get('/test/20140103030321{0}/http://example.com/?example=1', fmod) assert resp.status_int == 200 def test_another_coll(self): """ Test adding warc to a new coll, check replay """ warc1 = self._get_sample_warc('example.warc.gz') main(['init', 'foo']) main(['add', 'foo', warc1]) def test_another_coll_replay(self, fmod): resp = self.get('/foo/20140103030321{0}/http://example.com/?example=1', fmod) assert resp.status_int == 200 def test_add_more_warcs(self): """ Test adding additional warcs, check replay of added content """ warc1 = self._get_sample_warc('iana.warc.gz') warc2 = self._get_sample_warc('example-extra.warc') main(['add', 'test', warc1, warc2]) # Spurrious file in collections with open(os.path.join(self.root_dir, COLLECTIONS, 'blah'), 'w+b') as fh: fh.write(b'foo\n') with raises(IOError): main(['add', 'test', 'non-existent-file.warc.gz']) def test_add_more_warcs_replay(self, fmod): # check new cdx resp = self.get('/test/20140126200624{0}/http://www.iana.org/', fmod) assert resp.status_int == 200 def test_add_custom_nested_warcs(self): """ Test recursive indexing of custom created WARC hierarchy, warcs/A/..., warcs/B/sub/... Ensure CDX is relative to root archive dir, test replay """ main(['init', 'nested']) nested_root = os.path.join(self.root_dir, COLLECTIONS, 'nested', ARCHIVE_DIR) nested_a = os.path.join(nested_root, 'A') nested_b = os.path.join(nested_root, 'B', 'sub') os.makedirs(nested_a) os.makedirs(nested_b) warc1 = self._get_sample_warc('iana.warc.gz') warc2 = self._get_sample_warc('example.warc.gz') shutil.copy2(warc1, nested_a) shutil.copy2(warc2, nested_b) main(['index', 'nested', os.path.join(nested_a, 'iana.warc.gz'), os.path.join(nested_b, 'example.warc.gz') ]) nested_cdx = os.path.join(self.root_dir, COLLECTIONS, 'nested', INDEX_DIR, INDEX_FILE) with open(nested_cdx) as fh: nested_cdx_index = fh.read() assert '1043' in nested_cdx_index assert '333' in nested_cdx_index assert 'B/sub/example.warc.gz' in nested_cdx_index assert '2258' in nested_cdx_index assert '334' in nested_cdx_index assert 'A/iana.warc.gz' in nested_cdx_index def test_nested_replay(self, fmod): resp = self.get('/nested/20140126200624{0}/http://www.iana.org/', fmod) assert resp.status_int == 200 resp = self.get('/nested/20140103030321{0}/http://example.com/?example=1', fmod) assert resp.status_int == 200 def test_merge_vs_reindex_equality(self): """ Test full reindex vs merged update when adding warcs to ensure equality of indexes """ # ensure merged index is same as full reindex coll_dir = os.path.join(self.root_dir, COLLECTIONS, 'test', INDEX_DIR) orig = os.path.join(coll_dir, INDEX_FILE) bak = os.path.join(coll_dir, 'index.bak') shutil.copy(orig, bak) main(['reindex', 'test']) with open(orig) as orig_fh: merged_cdx = orig_fh.read() with open(bak) as bak_fh: reindex_cdx = bak_fh.read() assert len(reindex_cdx.splitlines()) == len(merged_cdx.splitlines()) assert merged_cdx == reindex_cdx def test_add_static(self): """ Test adding static file to collection, check access """ a_static = os.path.join(self.root_dir, COLLECTIONS, 'test', 'static', 'abc.js') with open(a_static, 'w+b') as fh: fh.write(b'/* Some JS File */') resp = self.testapp.get('/static/_/test/abc.js') assert resp.status_int == 200 assert resp.content_type == 'application/javascript' resp.charset = 'utf-8' assert '/* Some JS File */' in resp.text def test_add_shared_static(self): """ Test adding shared static file to root static/ dir, check access """ a_static = os.path.join(self.root_dir, 'static', 'foo.css') with open(a_static, 'w+b') as fh: fh.write(b'/* Some CSS File */') resp = self.testapp.get('/static/foo.css') assert resp.status_int == 200 assert resp.content_type == 'text/css' resp.charset = 'utf-8' assert '/* Some CSS File */' in resp.text def test_add_title_metadata_index_page(self): """ Test adding title metadata to a collection, test retrieval on default index page """ main(['metadata', 'foo', '--set', 'title=Collection Title']) resp = self.testapp.get('/') assert resp.status_int == 200 assert resp.content_type == 'text/html' resp.charset = 'utf-8' assert '(Collection Title)' in resp.text # test cache resp = self.testapp.get('/') resp.charset = 'utf-8' assert '(Collection Title)' in resp.text def test_other_metadata_search_page(self): main(['metadata', 'foo', '--set', 'desc=Some Description Text', 'other=custom value']) with raises(ValueError): main(['metadata', 'foo', '--set', 'name_only']) resp = self.testapp.get('/foo/') resp.charset = 'utf-8' assert resp.status_int == 200 assert resp.content_type == 'text/html' assert 'Collection Title' in resp.text assert 'desc' in resp.text assert 'Some Description Text' in resp.text assert 'other' in resp.text assert 'custom value' in resp.text def test_custom_template_search(self): """ Test manually added custom search template search.html """ custom_search = os.path.join(self.root_dir, COLLECTIONS, 'test', 'templates', 'search.html') with open(custom_search, 'w+b') as fh: fh.write(b'pywb custom search page') resp = self.testapp.get('/test/') resp.charset = 'utf-8' assert resp.status_int == 200 assert resp.content_type == 'text/html' assert 'pywb custom search page' in resp.text def test_add_custom_banner(self): """ Test adding custom banner.html per-collection template """ banner_file = os.path.join(self.root_dir, COLLECTIONS, 'test', 'templates', 'banner.html') with open(banner_file, 'w+b') as fh: fh.write(b'