<commit_before>import copy
import os
from os import path
import urllib2
from StringIO import StringIO

from nose.tools import assert_raises, assert_equals
from nose import SkipTest

from webassets import Bundle
from webassets.bundle import get_all_bundle_files
from webassets.env import Environment
from webassets.exceptions import BundleError, BuildError
from webassets.filter import Filter
from webassets.updater import TimestampUpdater, BaseUpdater, SKIP_CACHE
from webassets.cache import MemoryCache
from webassets.version import Manifest, Version, VersionIndeterminableError

from helpers import TempEnvironmentHelper, noop, assert_raises_regexp


class TestBundleConfig(TempEnvironmentHelper):

    def test_unknown_init_kwargs(self):
        """We used to silently ignore unsupported kwargs, which can make
        mistakes harder to track down; in particular "filters" vs "filter"
        is confusing. Now we raise an error.
        """
        try:
            Bundle(yaddayada=True)
        except TypeError, e:
            assert "unexpected keyword argument" in ("%s" % e)
        else:
            raise Exception('Expected TypeError not raised')

    def test_init_extra_kwarg(self):
        """Bundles may be given an ``extra`` dictionary."""
        assert Bundle().extra == {}
        assert Bundle(extra={'foo': 'bar'}).extra == {'foo': 'bar'}

        # Nested extra values
        assert Bundle(Bundle(extra={'foo': 'bar'}),
                      Bundle(extra={'baz': 'qux'})).extra == {
            'foo': 'bar', 'baz': 'qux'}

    def test_filter_assign(self):
        """Test the different ways we can assign filters to the bundle.
        """
        class TestFilter(Filter):
            pass

        def _assert(list, length):
            """Confirm that everything in the list is a filter instance,
            and that the list as the required length."""
            assert len(list) == length
            assert bool([f for f in list if isinstance(f, Filter)])

        # Comma-separated string.
        b = self.mkbundle(filters='jsmin,cssutils')
        _assert(b.filters, 2)
        # Whitespace is ignored.
        b = self.mkbundle(filters=' jsmin, cssutils ')
        _assert(b.filters, 2)

        # List of strings.
        b = self.mkbundle(filters=['jsmin', 'cssutils'])
        _assert(b.filters, 2)
        # Strings inside a list may not be further comma separated
        assert_raises(ValueError, self.mkbundle, filters=['jsmin,cssutils'])

        # A single or multiple classes may be given
        b = self.mkbundle(filters=TestFilter)
        _assert(b.filters, 1)
        b = self.mkbundle(filters=[TestFilter, TestFilter, TestFilter])
        _assert(b.filters, 3)

        # A single or multiple instance may be given
        b = self.mkbundle(filters=TestFilter())
        _assert(b.filters, 1)
        b = self.mkbundle(filters=[TestFilter(), TestFilter(), TestFilter()])
        _assert(b.filters, 3)

        # You can mix instances and classes
        b = self.mkbundle(filters=[TestFilter, TestFilter()])
        _assert(b.filters, 2)

        # If something is wrong, an error is raised right away.
        assert_raises(ValueError, self.mkbundle, filters='notreallyafilter')
        assert_raises(ValueError, self.mkbundle, filters=object())

        # [bug] Specifically test that we can assign ``None``.
        self.mkbundle().filters = None

        # Changing filters after bundle creation is no problem, either.
        b = self.mkbundle()
        assert b.filters == ()
        b.filters = TestFilter
        _assert(b.filters, 1)

        # Assigning the own filter list should change nothing.
        old_filters = b.filters
        b.filters = b.filters
        assert b.filters == old_filters

    def test_depends_assign(self):
        """Test the different ways we can assign dependencies.
        """
        # List of strings.
        b = self.mkbundle(depends=['file1', 'file2'])
        assert len(b.depends) == 2

        # Single string
        b = self.mkbundle(depends='*.sass')
        assert len(b.depends) == 1
        assert b.depends == ['*.sass']

    def test_depends_cached(self):
        """Test that the depends property is cached."""
        self.create_files({'file1.sass': ''})
        b = self.mkbundle(depends=['*.sass'])
        assert len(b.resolve_depends(self.env)) == 1
        self.create_files({'file2.sass': ''})
        assert len(b.resolve_depends(self.env)) == 1


class TestBuildVarious(TempEnvironmentHelper):
    """Test building various bundle structures, in various different
    circumstances. Generally all things "building" which don't have a
    better place.
    """

    def test_simple_bundle(self):
        """Simple bundle, no child bundles, no filters."""
        self.mkbundle('in1', 'in2', output='out').build()
        assert self.get('out') == 'A\nB'

    def test_nested_bundle(self):
        """A nested bundle."""
        self.mkbundle('in1', self.mkbundle('in3', 'in4'), 'in2', output='out').build()
        assert self.get('out') == 'A\nC\nD\nB'

    def test_container_bundle(self):
        """A container bundle.
        """
        self.mkbundle(
            self.mkbundle('in1', output='out1'),
            self.mkbundle('in2', output='out2')).build()
        assert self.get('out1') == 'A'
        assert self.get('out2') == 'B'

    def test_build_return_value(self):
        """build() method returns list of built hunks.
        """
        # Test a simple bundle (single hunk)
        hunks = self.mkbundle('in1', 'in2', output='out').build()
        assert len(hunks) == 1
        assert hunks[0].data() == 'A\nB'

        # Test container bundle (multiple hunks)
        hunks = self.mkbundle(
            self.mkbundle('in1', output='out1'),
            self.mkbundle('in2', output='out2')).build()
        assert len(hunks) == 2
        assert hunks[0].data() == 'A'
        assert hunks[1].data() == 'B'

    def test_nested_bundle_with_skipped_cache(self):
        """[Regression] There was a bug when doing a build with
        an updater that returned SKIP_CACHE, due to passing arguments
        incorrectly.
        """
        class SkipCacheUpdater(BaseUpdater):
            def needs_rebuild(self, *a, **kw):
                return SKIP_CACHE
        self.env.updater = SkipCacheUpdater()
        self.create_files({'out': ''})  # or updater won't come into play
        self.mkbundle('in1', self.mkbundle('in3', 'in4'), 'in2',
                      output='out').build()
        assert self.get('out') == 'A\nC\nD\nB'

    def test_no_output_error(self):
        """A bundle without an output configured cannot be built.
        """
        assert_raises(BuildError, self.mkbundle('in1', 'in2').build)

    def test_empty_bundle_error(self):
        """An empty bundle cannot be built.
        """
        assert_raises(BuildError, self.mkbundle(output='out').build)
        # That is true even for child bundles
        assert_raises(BuildError, self.mkbundle(self.mkbundle(), 'in1', output='out').build)

    def test_rebuild(self):
        """Regression test for a bug that occurred when a bundle
        was built a second time since Bundle.get_files() did
        not return absolute filenames."""
        self.mkbundle('in1', 'in2', output='out').build()
        assert self.get('out') == 'A\nB'
        self.mkbundle('in1', 'in2', output='out').build()
        assert self.get('out') == 'A\nB'

    def test_deleted_source_files(self):
        """Bundle contents are cached. However, if a file goes missing
        that was included via wildcard, this will not cause any problems
        in subsequent builds.

        If a statically included file goes missing however, the build
        fails.
        """
        self.create_files({'in': 'A', '1.css': '1', '2.css': '2'})
        bundle = self.mkbundle('in', '*.css', output='out')
        bundle.build()
        self.get('out') == 'A12'

        # Delete a wildcard file - we still build fine
        os.unlink(self.path('1.css'))
        bundle.build()
        self.get('out') == 'A1'

        # Delete the static file - now we can't build
        os.unlink(self.path('in'))
        assert_raises(BundleError, bundle.build)

    def test_merge_does_not_apply_filters(self):
        """Test that while we are in merge mode, the filters are not
        applied.
        """
        self.env.debug = 'merge'
        b = self.mkbundle('in1', 'in2', output='out',
                          filters=AppendFilter(':in', ':out'))
        b.build()
        assert self.get('out') == 'A\nB'

        # Check the reverse - filters are applied when not in merge mode
        self.env.debug = False
        b.build(force=True)
        assert self.get('out') == 'A:in\nB:in:out'

    def test_auto_create_target_directory(self):
        """A bundle output's target directory is automatically
        created, if it doesn't exist yet.
        """
        self.mkbundle('in1', 'in2', output='out/nested/x/foo').build()
        assert self.get('out/nested/x/foo') == 'A\nB'

    def test_with_custom_output(self):
        """build() method can write to a custom file object."""
        from StringIO import StringIO
        buffer = StringIO()
        self.mkbundle('in1', 'in2', output='out').build(output=buffer)
        assert buffer.getvalue() == 'A\nB'
        assert not self.exists('out')    # file was not written.


class TestBuildWithVariousDebugOptions(TempEnvironmentHelper):
    """Test build behavior with respect to the "debug level", and the various
    ways to set and change the debug level within the bundle hierarchy.
    """

    def test_debug_mode_inherited(self):
        """Make sure that if a bundle sets debug=FOO, that value is also used
        for child bundles.
        """
        self.env.debug = True  # To allow "merge" at all
        b = self.mkbundle(
            'in1',
            self.mkbundle(
                'in2', filters=AppendFilter(':childin', ':childout')),
            output='out', debug='merge',
            filters=AppendFilter(':rootin', ':rootout'))
        b.build()
        # Neither the content of in1 or of in2 have filters applied.
        assert self.get('out') == 'A\nB'

    def test_cannot_increase_debug_level(self):
        """Child bundles cannot increase the debug level.

        If they attempt to, these attempts are silently ignored.
        """
        # False to True has no effect
        self.mkbundle('in1', self.mkbundle('in2', debug=True),
                      output='out', debug=False).build()
        assert self.get('out') == 'A\nB'

        # "merge" to True has no effect
        self.mkbundle(
            'in1', self.mkbundle('in2', debug=True),
            output='out', debug='merge', filters=AppendFilter('_')).build()
        assert self.get('out') == 'A\nB'

        # False to "merge" has no effect
        self.mkbundle(
            'in1', self.mkbundle('in2', debug='merge',
                                 filters=AppendFilter('_')),
            output='out', debug=False).build()
        assert self.get('out') == 'A\nB'

    def test_decreasing_debug_level(self):
        """A child bundle may switch to full production mode (turning on the
        filters), while the parent is only in merge mode.

        The other debug level change (from True to "merge" or from True to
        "False" do not concern us here, and are tested in ``TestUrls*``).
        """
        self.env.debug = 'merge'
        b = self.mkbundle(
            'in1',
            self.mkbundle('in2', debug=False,
                          filters=AppendFilter(':childin', ':childout')),
            output='out', debug='merge',
            filters=AppendFilter(':rootin', ':rootout'))
        b.build()
        # Note how the content of "in1" (A) does not have its filters applied.
        assert self.get('out') == 'A\nB:childin:rootin:childout'

    def test_invalid_debug_value(self):
        """Test exception Bundle.build() throws if debug is an invalid value.
        """
        # On the bundle level
        b = self.mkbundle('a', 'b', output='out', debug="invalid")
        assert_raises(BundleError, b.build)

        # On the environment level
        self.env.debug = "invalid"
        b = self.mkbundle('a', 'b', output='out')
        assert_raises(BundleError, b.build)

    def test_building_in_debug_mode(self):
        """When calling build() while we are in debug mode (debug=False),
        the method builds as if in production mode (debug=False).

        This is a question of API design: If the user calls the method, the
        expectation is clearly for something useful to happen.
        """
        self.env.debug = True
        b = self.mkbundle(
            'in1', 'in2', output='out', filters=AppendFilter('foo'))
        b.build()
        assert self.get('out') == 'Afoo\nBfoo'


class ReplaceFilter(Filter):
    """Filter that does a simple string replacement.
    """

    def __init__(self, input=(None, None), output=(None, None)):
        Filter.__init__(self)
        self._input_from, self._input_to = input
        self._output_from, self._output_to = output

    def input(self, in_, out, **kw):
        if self._input_from:
            out.write(in_.read().replace(self._input_from, self._input_to))
        else:
            out.write(in_.read())

    def output(self, in_, out, **kw):
        if self._output_from:
            out.write(in_.read().replace(self._output_from, self._output_to))
        else:
            out.write(in_.read())

    def unique(self):
        # So we can apply this filter multiple times
        return self._input_from, self._output_from


class AppendFilter(Filter):
    """Filter that simply appends stuff.
    """

    def __init__(self, input=None, output=None, unique=True):
        Filter.__init__(self)
        self._input = input
        self._output = output
        self._unique = unique

    def input(self, in_, out, **kw):
        out.write(in_.read())
        if self._input:
            out.write(self._input)

    def output(self, in_, out, **kw):
        out.write(in_.read())
        if self._output:
            out.write(self._output)

    def unique(self):
        if not self._unique:
            return False
        # So we can apply this filter multiple times
        return self._input, self._output


class TestFilterApplication(TempEnvironmentHelper):
    """Test filter application during building.
    """

    default_files = {'1': 'foo', '2': 'foo', '3': 'foo'}

    def test_input_before_output(self):
        """Ensure that input filters are applied, and that they are applied
        before an output filter gets to say something.
        """
        self.mkbundle('1', '2', output='out', filters=ReplaceFilter(
            input=('foo', 'input was here'), output=('foo', 'output was here'))).build()
        assert self.get('out') == 'input was here\ninput was here'

    def test_output_after_input(self):
        """Ensure that output filters are applied, and that they are applied
        after input filters did their job.
        """
        self.mkbundle('1', '2', output='out', filters=ReplaceFilter(
            input=('foo', 'bar'), output=('bar', 'output was here'))).build()
        assert self.get('out') == 'output was here\noutput was here'

    def test_input_before_output_nested(self):
        """Ensure that when nested bundles are used, a parent bundles
        input filters are applied before a child bundles output filter.
        """
        child_bundle_with_output_filter = self.mkbundle('1', '2',
                filters=ReplaceFilter(output=('foo', 'output was here')))
        parent_bundle_with_input_filter = self.mkbundle(child_bundle_with_output_filter,
                output='out',
                filters=ReplaceFilter(input=('foo', 'input was here')))
        parent_bundle_with_input_filter.build()
        assert self.get('out') == 'input was here\ninput was here'

    def test_input_before_output_nested_merged(self):
        """Same thing as above - a parent input filter is passed done -
        but this time, ensure that duplicate filters are not applied twice.
        """
        child_bundle = self.mkbundle('1', '2',
                                     filters=AppendFilter(input='-child', unique=False))
        parent_bundle = self.mkbundle(child_bundle, output='out',
                               filters=AppendFilter(input='-parent', unique=False))
        parent_bundle.build()
        assert self.get('out') == 'foo-child\nfoo-child'

    def test_container_bundle_with_filters(self):
        """If a bundle has no output, but filters, those filters are
        passed down to each sub-bundle.
        """
        self.mkbundle(
            Bundle('1', output='out1', filters=()),
            Bundle('2', output='out2', filters=AppendFilter(':childin', ':childout')),
            Bundle('3', output='out3', filters=AppendFilter(':childin', ':childout', unique=False)),
            filters=AppendFilter(':rootin', ':rootout', unique=False)
        ).urls()
        self.p('out1', 'out2', 'out3')
        assert self.get('out1') == 'foo:rootin:rootout'
        assert self.get('out2') == 'foo:childin:rootin:childout:rootout'
        assert self.get('out3') == 'foo:childin:childout'

    def test_duplicate_open_filters(self):
        """Test that only one open() filter can be used.
        """
        # TODO: For performance reasons, this check could possibly be
        # done earlier, when assigning to the filter property. It wouldn't
        # catch all cases involving bundle nesting though.
        class OpenFilter(Filter):
            def open(self, *a, **kw): pass
            def __init__(self, id): Filter.__init__(self); self.id = id
            def id(self): return self.id
        self.create_files(set('xyz'))
        bundle = self.mkbundle(
            'xyz', filters=(OpenFilter('a'), OpenFilter('b')))
        assert_raises(BuildError, bundle.build)

    def test_concat(self):
        """Test the concat() filter type.
        """
        class ConcatFilter(Filter):
            def concat(self, out, hunks, **kw):
                out.write('%%%'.join([h.data() for h in hunks]))
        self.create_files({'a': '1', 'b': '2'})
        self.mkbundle('a', 'b', filters=ConcatFilter, output='out').build()
        assert self.get('out') == '1%%%2'

    def test_open_before_input(self):
        """[Regression] Test that if an open filter is used, input filters
        still receive the ``source_path`` kwargs.
        """
        captured_kw = {}
        class TestFilter(Filter):
            def open(self, out, *a, **kw): out.write('foo')
            def input(self, *a, **kw):
                assert not captured_kw
                captured_kw.update(kw)
        self.create_files({'a': '1'})
        self.mkbundle('a', filters=TestFilter, output='out').build()
        # TODO: Could be generalized to test all the other values that
        # each filter method expects to receive. This is currently not
        # done anywhere )though it likely still wouldn't have caught this).
        assert 'source_path' in captured_kw


class TestAutoBuild(TempEnvironmentHelper):
    """Test bundle auto rebuild, and generally everything involving
    the updater from the bundle's perspective.
    """

    def setup(self):
        TempEnvironmentHelper.setup(self)

        class CustomUpdater(BaseUpdater):
            allow = True
            def needs_rebuild(self, *a, **kw):<commit_msg>Some more tests of filters with max_debug_level.<commit_after>import copy
import os
from os import path
import urllib2
from StringIO import StringIO

from nose.tools import assert_raises, assert_equals
from nose import SkipTest

from webassets import Bundle
from webassets.bundle import get_all_bundle_files
from webassets.env import Environment
from webassets.exceptions import BundleError, BuildError
from webassets.filter import Filter
from webassets.updater import TimestampUpdater, BaseUpdater, SKIP_CACHE
from webassets.cache import MemoryCache
from webassets.version import Manifest, Version, VersionIndeterminableError

from helpers import TempEnvironmentHelper, noop, assert_raises_regexp


class TestBundleConfig(TempEnvironmentHelper):

    def test_unknown_init_kwargs(self):
        """We used to silently ignore unsupported kwargs, which can make
        mistakes harder to track down; in particular "filters" vs "filter"
        is confusing. Now we raise an error.
        """
        try:
            Bundle(yaddayada=True)
        except TypeError, e:
            assert "unexpected keyword argument" in ("%s" % e)
        else:
            raise Exception('Expected TypeError not raised')

    def test_init_extra_kwarg(self):
        """Bundles may be given an ``extra`` dictionary."""
        assert Bundle().extra == {}
        assert Bundle(extra={'foo': 'bar'}).extra == {'foo': 'bar'}

        # Nested extra values
        assert Bundle(Bundle(extra={'foo': 'bar'}),
                      Bundle(extra={'baz': 'qux'})).extra == {
            'foo': 'bar', 'baz': 'qux'}

    def test_filter_assign(self):
        """Test the different ways we can assign filters to the bundle.
        """
        class TestFilter(Filter):
            pass

        def _assert(list, length):
            """Confirm that everything in the list is a filter instance,
            and that the list as the required length."""
            assert len(list) == length
            assert bool([f for f in list if isinstance(f, Filter)])

        # Comma-separated string.
        b = self.mkbundle(filters='jsmin,cssutils')
        _assert(b.filters, 2)
        # Whitespace is ignored.
        b = self.mkbundle(filters=' jsmin, cssutils ')
        _assert(b.filters, 2)

        # List of strings.
        b = self.mkbundle(filters=['jsmin', 'cssutils'])
        _assert(b.filters, 2)
        # Strings inside a list may not be further comma separated
        assert_raises(ValueError, self.mkbundle, filters=['jsmin,cssutils'])

        # A single or multiple classes may be given
        b = self.mkbundle(filters=TestFilter)
        _assert(b.filters, 1)
        b = self.mkbundle(filters=[TestFilter, TestFilter, TestFilter])
        _assert(b.filters, 3)

        # A single or multiple instance may be given
        b = self.mkbundle(filters=TestFilter())
        _assert(b.filters, 1)
        b = self.mkbundle(filters=[TestFilter(), TestFilter(), TestFilter()])
        _assert(b.filters, 3)

        # You can mix instances and classes
        b = self.mkbundle(filters=[TestFilter, TestFilter()])
        _assert(b.filters, 2)

        # If something is wrong, an error is raised right away.
        assert_raises(ValueError, self.mkbundle, filters='notreallyafilter')
        assert_raises(ValueError, self.mkbundle, filters=object())

        # [bug] Specifically test that we can assign ``None``.
        self.mkbundle().filters = None

        # Changing filters after bundle creation is no problem, either.
        b = self.mkbundle()
        assert b.filters == ()
        b.filters = TestFilter
        _assert(b.filters, 1)

        # Assigning the own filter list should change nothing.
        old_filters = b.filters
        b.filters = b.filters
        assert b.filters == old_filters

    def test_depends_assign(self):
        """Test the different ways we can assign dependencies.
        """
        # List of strings.
        b = self.mkbundle(depends=['file1', 'file2'])
        assert len(b.depends) == 2

        # Single string
        b = self.mkbundle(depends='*.sass')
        assert len(b.depends) == 1
        assert b.depends == ['*.sass']

    def test_depends_cached(self):
        """Test that the depends property is cached."""
        self.create_files({'file1.sass': ''})
        b = self.mkbundle(depends=['*.sass'])
        assert len(b.resolve_depends(self.env)) == 1
        self.create_files({'file2.sass': ''})
        assert len(b.resolve_depends(self.env)) == 1


class TestBuildVarious(TempEnvironmentHelper):
    """Test building various bundle structures, in various different
    circumstances. Generally all things "building" which don't have a
    better place.
    """

    def test_simple_bundle(self):
        """Simple bundle, no child bundles, no filters."""
        self.mkbundle('in1', 'in2', output='out').build()
        assert self.get('out') == 'A\nB'

    def test_nested_bundle(self):
        """A nested bundle."""
        self.mkbundle('in1', self.mkbundle('in3', 'in4'), 'in2', output='out').build()
        assert self.get('out') == 'A\nC\nD\nB'

    def test_container_bundle(self):
        """A container bundle.
        """
        self.mkbundle(
            self.mkbundle('in1', output='out1'),
            self.mkbundle('in2', output='out2')).build()
        assert self.get('out1') == 'A'
        assert self.get('out2') == 'B'

    def test_build_return_value(self):
        """build() method returns list of built hunks.
        """
        # Test a simple bundle (single hunk)
        hunks = self.mkbundle('in1', 'in2', output='out').build()
        assert len(hunks) == 1
        assert hunks[0].data() == 'A\nB'

        # Test container bundle (multiple hunks)
        hunks = self.mkbundle(
            self.mkbundle('in1', output='out1'),
            self.mkbundle('in2', output='out2')).build()
        assert len(hunks) == 2
        assert hunks[0].data() == 'A'
        assert hunks[1].data() == 'B'

    def test_nested_bundle_with_skipped_cache(self):
        """[Regression] There was a bug when doing a build with
        an updater that returned SKIP_CACHE, due to passing arguments
        incorrectly.
        """
        class SkipCacheUpdater(BaseUpdater):
            def needs_rebuild(self, *a, **kw):
                return SKIP_CACHE
        self.env.updater = SkipCacheUpdater()
        self.create_files({'out': ''})  # or updater won't come into play
        self.mkbundle('in1', self.mkbundle('in3', 'in4'), 'in2',
                      output='out').build()
        assert self.get('out') == 'A\nC\nD\nB'

    def test_no_output_error(self):
        """A bundle without an output configured cannot be built.
        """
        assert_raises(BuildError, self.mkbundle('in1', 'in2').build)

    def test_empty_bundle_error(self):
        """An empty bundle cannot be built.
        """
        assert_raises(BuildError, self.mkbundle(output='out').build)
        # That is true even for child bundles
        assert_raises(BuildError, self.mkbundle(self.mkbundle(), 'in1', output='out').build)

    def test_rebuild(self):
        """Regression test for a bug that occurred when a bundle
        was built a second time since Bundle.get_files() did
        not return absolute filenames."""
        self.mkbundle('in1', 'in2', output='out').build()
        assert self.get('out') == 'A\nB'
        self.mkbundle('in1', 'in2', output='out').build()
        assert self.get('out') == 'A\nB'

    def test_deleted_source_files(self):
        """Bundle contents are cached. However, if a file goes missing
        that was included via wildcard, this will not cause any problems
        in subsequent builds.

        If a statically included file goes missing however, the build
        fails.
        """
        self.create_files({'in': 'A', '1.css': '1', '2.css': '2'})
        bundle = self.mkbundle('in', '*.css', output='out')
        bundle.build()
        self.get('out') == 'A12'

        # Delete a wildcard file - we still build fine
        os.unlink(self.path('1.css'))
        bundle.build()
        self.get('out') == 'A1'

        # Delete the static file - now we can't build
        os.unlink(self.path('in'))
        assert_raises(BundleError, bundle.build)

    def test_merge_does_not_apply_filters(self):
        """Test that while we are in merge mode, the filters are not
        applied.
        """
        self.env.debug = 'merge'
        b = self.mkbundle('in1', 'in2', output='out',
                          filters=AppendFilter(':in', ':out'))
        b.build()
        assert self.get('out') == 'A\nB'

        # Check the reverse - filters are applied when not in merge mode
        self.env.debug = False
        b.build(force=True)
        assert self.get('out') == 'A:in\nB:in:out'

    def test_auto_create_target_directory(self):
        """A bundle output's target directory is automatically
        created, if it doesn't exist yet.
        """
        self.mkbundle('in1', 'in2', output='out/nested/x/foo').build()
        assert self.get('out/nested/x/foo') == 'A\nB'

    def test_with_custom_output(self):
        """build() method can write to a custom file object."""
        from StringIO import StringIO
        buffer = StringIO()
        self.mkbundle('in1', 'in2', output='out').build(output=buffer)
        assert buffer.getvalue() == 'A\nB'
        assert not self.exists('out')    # file was not written.


class TestBuildWithVariousDebugOptions(TempEnvironmentHelper):
    """Test build behavior with respect to the "debug level", and the various
    ways to set and change the debug level within the bundle hierarchy.
    """

    def test_debug_mode_inherited(self):
        """Make sure that if a bundle sets debug=FOO, that value is also used
        for child bundles.
        """
        self.env.debug = True  # To allow "merge" at all
        b = self.mkbundle(
            'in1',
            self.mkbundle(
                'in2', filters=AppendFilter(':childin', ':childout')),
            output='out', debug='merge',
            filters=AppendFilter(':rootin', ':rootout'))
        b.build()
        # Neither the content of in1 or of in2 have filters applied.
        assert self.get('out') == 'A\nB'

    def test_cannot_increase_debug_level(self):
        """Child bundles cannot increase the debug level.

        If they attempt to, these attempts are silently ignored.
        """
        # False to True has no effect
        self.mkbundle('in1', self.mkbundle('in2', debug=True),
                      output='out', debug=False).build()
        assert self.get('out') == 'A\nB'

        # "merge" to True has no effect
        self.mkbundle(
            'in1', self.mkbundle('in2', debug=True),
            output='out', debug='merge', filters=AppendFilter('_')).build()
        assert self.get('out') == 'A\nB'

        # False to "merge" has no effect
        self.mkbundle(
            'in1', self.mkbundle('in2', debug='merge',
                                 filters=AppendFilter('_')),
            output='out', debug=False).build()
        assert self.get('out') == 'A\nB'

    def test_decreasing_debug_level(self):
        """A child bundle may switch to full production mode (turning on the
        filters), while the parent is only in merge mode.

        The other debug level change (from True to "merge" or from True to
        "False" do not concern us here, and are tested in ``TestUrls*``).
        """
        self.env.debug = 'merge'
        b = self.mkbundle(
            'in1',
            self.mkbundle('in2', debug=False,
                          filters=AppendFilter(':childin', ':childout')),
            output='out', debug='merge',
            filters=AppendFilter(':rootin', ':rootout'))
        b.build()
        # Note how the content of "in1" (A) does not have its filters applied.
        assert self.get('out') == 'A\nB:childin:rootin:childout'

    def test_invalid_debug_value(self):
        """Test exception Bundle.build() throws if debug is an invalid value.
        """
        # On the bundle level
        b = self.mkbundle('a', 'b', output='out', debug="invalid")
        assert_raises(BundleError, b.build)

        # On the environment level
        self.env.debug = "invalid"
        b = self.mkbundle('a', 'b', output='out')
        assert_raises(BundleError, b.build)

    def test_building_in_debug_mode(self):
        """When calling build() while we are in debug mode (debug=False),
        the method builds as if in production mode (debug=False).

        This is a question of API design: If the user calls the method, the
        expectation is clearly for something useful to happen.
        """
        self.env.debug = True
        b = self.mkbundle(
            'in1', 'in2', output='out', filters=AppendFilter('foo'))
        b.build()
        assert self.get('out') == 'Afoo\nBfoo'


class ReplaceFilter(Filter):
    """Filter that does a simple string replacement.
    """

    def __init__(self, input=(None, None), output=(None, None)):
        Filter.__init__(self)
        self._input_from, self._input_to = input
        self._output_from, self._output_to = output

    def input(self, in_, out, **kw):
        if self._input_from:
            out.write(in_.read().replace(self._input_from, self._input_to))
        else:
            out.write(in_.read())

    def output(self, in_, out, **kw):
        if self._output_from:
            out.write(in_.read().replace(self._output_from, self._output_to))
        else:
            out.write(in_.read())

    def unique(self):
        # So we can apply this filter multiple times
        return self._input_from, self._output_from


class AppendFilter(Filter):
    """Filter that simply appends stuff.
    """

    def __init__(self, input=None, output=None, unique=True):
        Filter.__init__(self)
        self._input = input
        self._output = output
        self._unique = unique

    def input(self, in_, out, **kw):
        out.write(in_.read())
        if self._input:
            out.write(self._input)

    def output(self, in_, out, **kw):
        out.write(in_.read())
        if self._output:
            out.write(self._output)

    def unique(self):
        if not self._unique:
            return False
        # So we can apply this filter multiple times
        return self._input, self._output


class TestFilterApplication(TempEnvironmentHelper):
    """Test filter application during building - order, passing down to child
    bundles, that kind of thing.
    """

    default_files = {'1': 'foo', '2': 'foo', '3': 'foo',
                     'a': 'bar', 'b': 'qux'}

    def test_input_before_output(self):
        """Ensure that input filters are applied, and that they are applied
        before an output filter gets to say something.
        """
        self.mkbundle('1', '2', output='out', filters=ReplaceFilter(
            input=('foo', 'input was here'), output=('foo', 'output was here'))).build()
        assert self.get('out') == 'input was here\ninput was here'

    def test_output_after_input(self):
        """Ensure that output filters are applied, and that they are applied
        after input filters did their job.
        """
        self.mkbundle('1', '2', output='out', filters=ReplaceFilter(
            input=('foo', 'bar'), output=('bar', 'output was here'))).build()
        assert self.get('out') == 'output was here\noutput was here'

    def test_input_before_output_nested(self):
        """Ensure that when nested bundles are used, a parent bundles
        input filters are applied before a child bundles output filter.
        """
        child_bundle_with_output_filter = self.mkbundle('1', '2',
                filters=ReplaceFilter(output=('foo', 'output was here')))
        parent_bundle_with_input_filter = self.mkbundle(child_bundle_with_output_filter,
                output='out',
                filters=ReplaceFilter(input=('foo', 'input was here')))
        parent_bundle_with_input_filter.build()
        assert self.get('out') == 'input was here\ninput was here'

    def test_input_before_output_nested_unique(self):
        """Same thing as above - a parent input filter is passed done -
        but this time, ensure that duplicate filters are not applied twice.
        """
        child_bundle = self.mkbundle('1', '2',
                                     filters=AppendFilter(input='-child', unique=False))
        parent_bundle = self.mkbundle(child_bundle, output='out',
                               filters=AppendFilter(input='-parent', unique=False))
        parent_bundle.build()
        assert self.get('out') == 'foo-child\nfoo-child'

    def test_input_with_nested_in_merge_mode(self):
        """[Regression] In merge mode, the input filters are not applied for
        child bundles.
        """
        self.env.debug = True  # To allow "merge" at all
        b = self.mkbundle(
            'a',
            self.mkbundle('b',
                          filters=AppendFilter(':childin', ':childout')),
            output='out', filters=AppendFilter(':rootin', ':rootout'),
            debug='merge')
        b.build()
        # Neither the content of in1 or of in2 have filters applied.
        assert self.get('out') == 'bar\nqux'

    def test_input_with_nested_switch_from_merge_to_full_mode(self):
        """A child bundle switches to full production mode (turning on the
        filters), while the parent is only in merge mode. Ensure that in such
        a case input filters declared in the parent are applied in the child
        (and only there).
        """
        self.env.debug = 'merge'
        child_filters = AppendFilter(':childin', ':childout')
        parent_filters = AppendFilter(':rootin', ':rootout')
        b = self.mkbundle(
            'a', self.mkbundle('b', filters=child_filters, debug=False),
            output='out', filters=parent_filters, debug='merge')
        b.build()
        # Note how the content of "in1" (A) does not have its filters applied.
        assert self.get('out') == 'bar\nqux:childin:rootin:childout'

    def test_open_before_input(self):
        """[Regression] Test that if an open filter is used, input filters
        still receive the ``source_path`` kwargs.
        """
        captured_kw = {}
        class TestFilter(Filter):
            def open(self, out, *a, **kw): out.write('foo')
            def input(self, *a, **kw):
                assert not captured_kw
                captured_kw.update(kw)
        self.create_files({'a': '1'})
        self.mkbundle('a', filters=TestFilter, output='out').build()
        # TODO: Could be generalized to test all the other values that
        # each filter method expects to receive. This is currently not
        # done anywhere )though it likely still wouldn't have caught this).
        assert 'source_path' in captured_kw

    def test_duplicate_open_filters(self):
        """Test that only one open() filter can be used.
        """
        # TODO: For performance reasons, this check could possibly be
        # done earlier, when assigning to the filter property. It wouldn't
        # catch all cases involving bundle nesting though.
        class OpenFilter(Filter):
            def open(self, *a, **kw): pass
            def __init__(self, id): Filter.__init__(self); self.id = id
            def id(self): return self.id
        self.create_files(set('xyz'))
        bundle = self.mkbundle(
            'xyz', filters=(OpenFilter('a'), OpenFilter('b')))
        assert_raises(BuildError, bundle.build)

    def test_concat(self):
        """Test the concat() filter type.
        """
        class ConcatFilter(Filter):
            def concat(self, out, hunks, **kw):
                out.write('%%%'.join([h.data() for h in hunks]))
        self.create_files({'a': '1', 'b': '2'})
        self.mkbundle('a', 'b', filters=ConcatFilter, output='out').build()
        assert self.get('out') == '1%%%2'

    # TODO: concat filter child bundle behavior: This should probably be
    # "special", i.e.  the parent concat filter is used, unless overridden
    # in the child.

    def test_container_bundle_with_filters(self):
        """If a bundle has no output, but filters, those filters are
        passed down to each sub-bundle.
        """
        self.mkbundle(
            Bundle('1', output='out1', filters=()),
            Bundle('2', output='out2', filters=AppendFilter(':childin', ':childout')),
            Bundle('3', output='out3', filters=AppendFilter(':childin', ':childout', unique=False)),
            filters=AppendFilter(':rootin', ':rootout', unique=False)
        ).urls()
        assert self.get('out1') == 'foo:rootin:rootout'
        assert self.get('out2') == 'foo:childin:rootin:childout:rootout'
        assert self.get('out3') == 'foo:childin:childout'

    def test_max_debug_level_filters(self):
        """Test how filters are applied when they define a non-default
        ``max_debug_level`` value.
        """

        class MaxDebugLevelFilters(TempEnvironmentHelper):
            default_files = {'1': 'foo'}

            def test_with_level(self, level):
                self.env.debug = True  # allows all bundle debug levels
                f = AppendFilter(':in', ':out'); f.max_debug_level = level
                self.mkbundle('1', output='out', filters=f, debug=level).build()
                assert self.get('out') == 'foo:in:out'

            def test_upgrading_affect_on_normal_filters(self):
                """max_debug_level 'merge' upgrade does not cause filters with
                a 'normal' max_debug_value to run. Note: A nested bundle is
                used here, as otherwise the bundle's debug=True would also
                override any upgrades through filter `max_debug_value``
                attributes."""
                self.env.debug = True  # allows all bundle debug levels
                f = AppendFilter(':in_upgr', ':out_upgr')
                f.max_debug_level = None
                g = AppendFilter(':in_def', ':out_def')
                self.mkbundle(Bundle('1', filters=(f, g), debug=True),
                              output='out', debug='merge').build()
                assert self.get('out') == 'foo:in_upgr:out_upgr'

        yield MaxDebugLevelFilters.test_with_level, 'merge'
        # With max_debug_level=True causes merge mode
        yield MaxDebugLevelFilters.test_with_level, True
        # With max_debug_level=None is same as =True
        yield MaxDebugLevelFilters.test_with_level, None

        yield MaxDebugLevelFilters.test_upgrading_affect_on_normal_filters


class TestAutoBuild(TempEnvironmentHelper):
    """Test bundle auto rebuild, and generally everything involving
    the updater from the bundle's perspective.
    """

    def setup(self):
        TempEnvironmentHelper.setup(self)

        class CustomUpdater(BaseUpdater):
            allow = True
            def needs_rebuild(self, *a, **kw):
{'fraction_non_alphanumeric': 0.12468289016254816, 'fraction_numerical': 0.005191205487174669, 'mean_word_length': 2.45223807979241, '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': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '879427', 'n_tokens_mistral': 12269, 'n_tokens_neox': 11969, 'n_words': 3511}