Mercurial > piecrust2
view tests/test_processing_base.py @ 298:b7ab1b503510
data: Fix incorrect next/previous page URLs in pagination data.
Consolidate splitting an URL between its first URL and its sub page number.
Be careful about the index page's URL not losing its slash.
author | Ludovic Chabant <ludovic@chabant.com> |
---|---|
date | Wed, 11 Mar 2015 23:46:42 -0700 |
parents | e725af1d48fb |
children | e7b865f8f335 |
line wrap: on
line source
import time import os.path import shutil import pytest from piecrust.processing.base import (ProcessorPipeline, SimpleFileProcessor) from piecrust.processing.records import ProcessorPipelineRecord from .mockutil import mock_fs, mock_fs_scope class FooProcessor(SimpleFileProcessor): def __init__(self, exts=None, open_func=None): exts = exts or {'foo', 'foo'} super(FooProcessor, self).__init__({exts[0]: exts[1]}) self.PROCESSOR_NAME = exts[0] self.open_func = open_func or open def _doProcess(self, in_path, out_path): with self.open_func(in_path, 'r') as f: text = f.read() with self.open_func(out_path, 'w') as f: f.write("%s: %s" % (self.PROCESSOR_NAME.upper(), text)) return True class NoopProcessor(SimpleFileProcessor): def __init__(self, exts): super(NoopProcessor, self).__init__({exts[0]: exts[1]}) self.PROCESSOR_NAME = exts[0] self.processed = [] def _doProcess(self, in_path, out_path): self.processed.append(in_path) shutil.copyfile(in_path, out_path) return True def _get_pipeline(fs, app=None): app = app or fs.getApp() app.config.set('baker/num_workers', 1) return ProcessorPipeline(app, fs.path('counter')) def test_empty(): fs = mock_fs() with mock_fs_scope(fs): pp = _get_pipeline(fs) pp.filterProcessors(['copy']) expected = {} assert expected == fs.getStructure('counter') pp.run() expected = {} assert expected == fs.getStructure('counter') def test_one_file(): fs = (mock_fs() .withFile('kitchen/assets/something.html', 'A test file.')) with mock_fs_scope(fs): pp = _get_pipeline(fs) pp.filterProcessors(['copy']) expected = {} assert expected == fs.getStructure('counter') pp.run() expected = {'something.html': 'A test file.'} assert expected == fs.getStructure('counter') def test_one_level_dirtyness(): fs = (mock_fs() .withFile('kitchen/assets/blah.foo', 'A test file.')) with mock_fs_scope(fs): pp = _get_pipeline(fs) pp.filterProcessors(['copy']) pp.run() expected = {'blah.foo': 'A test file.'} assert expected == fs.getStructure('counter') mtime = os.path.getmtime(fs.path('/counter/blah.foo')) assert abs(time.time() - mtime) <= 2 pp.run() assert expected == fs.getStructure('counter') assert mtime == os.path.getmtime(fs.path('/counter/blah.foo')) fs.withFile('kitchen/assets/blah.foo', 'A new test file.') pp.run() expected = {'blah.foo': 'A new test file.'} assert expected == fs.getStructure('counter') assert mtime < os.path.getmtime(fs.path('/counter/blah.foo')) def test_two_levels_dirtyness(): fs = (mock_fs() .withFile('kitchen/assets/blah.foo', 'A test file.')) with mock_fs_scope(fs) as scope: pp = _get_pipeline(fs) pp.processors.append(FooProcessor(('foo', 'bar'), scope._open)) pp.filterProcessors(['foo', 'copy']) pp.run() expected = {'blah.bar': 'FOO: A test file.'} assert expected == fs.getStructure('counter') mtime = os.path.getmtime(fs.path('/counter/blah.bar')) assert abs(time.time() - mtime) <= 2 pp.run() assert expected == fs.getStructure('counter') assert mtime == os.path.getmtime(fs.path('/counter/blah.bar')) fs.withFile('kitchen/assets/blah.foo', 'A new test file.') pp.run() expected = {'blah.bar': 'FOO: A new test file.'} assert expected == fs.getStructure('counter') assert mtime < os.path.getmtime(fs.path('/counter/blah.bar')) def test_removed(): fs = (mock_fs() .withFile('kitchen/assets/blah1.foo', 'A test file.') .withFile('kitchen/assets/blah2.foo', 'Ooops')) with mock_fs_scope(fs): expected = { 'blah1.foo': 'A test file.', 'blah2.foo': 'Ooops'} assert expected == fs.getStructure('kitchen/assets') pp = _get_pipeline(fs) pp.filterProcessors(['copy']) pp.run() assert expected == fs.getStructure('counter') os.remove(fs.path('/kitchen/assets/blah2.foo')) expected = { 'blah1.foo': 'A test file.'} assert expected == fs.getStructure('kitchen/assets') pp.run() assert expected == fs.getStructure('counter') def test_record_version_change(): fs = (mock_fs() .withFile('kitchen/assets/blah.foo', 'A test file.')) with mock_fs_scope(fs): pp = _get_pipeline(fs) noop = NoopProcessor(('foo', 'foo')) pp.processors.append(noop) pp.filterProcessors(['foo', 'copy']) pp.run() assert 1 == len(noop.processed) pp.run() assert 1 == len(noop.processed) ProcessorPipelineRecord.RECORD_VERSION += 1 try: pp.run() assert 2 == len(noop.processed) finally: ProcessorPipelineRecord.RECORD_VERSION -= 1 @pytest.mark.parametrize('patterns, expected', [ (['_'], {'something.html': 'A test file.'}), (['html'], {}), (['/^_/'], {'something.html': 'A test file.', 'foo': {'_important.html': 'Important!'}}) ]) def test_skip_pattern(patterns, expected): fs = (mock_fs() .withFile('kitchen/assets/something.html', 'A test file.') .withFile('kitchen/assets/_hidden.html', 'Shhh') .withFile('kitchen/assets/foo/_important.html', 'Important!')) with mock_fs_scope(fs): pp = _get_pipeline(fs) pp.addSkipPatterns(patterns) pp.filterProcessors(['copy']) assert {} == fs.getStructure('counter') pp.run() assert expected == fs.getStructure('counter') @pytest.mark.parametrize('names, expected', [ ('all', ['copy', 'concat', 'less', 'sass', 'sitemap']), ('all -sitemap', ['copy', 'concat', 'less', 'sass']), ('-sitemap -less -sass all', ['copy', 'concat']), ('copy', ['copy']), ('less sass', ['less', 'sass']) ]) def test_filter_processor(names, expected): fs = mock_fs() with mock_fs_scope(fs): app = fs.getApp() pp = _get_pipeline(fs, app=app) pp.filterProcessors('copy concat less sass sitemap') procs = pp.getFilteredProcessors(names) actual = [p.PROCESSOR_NAME for p in procs] assert sorted(actual) == sorted(expected)