import time
import unittest

from mako import lookup
from mako.cache import CacheImpl
from mako.cache import register_plugin
from mako.compat import py27
from mako.ext import beaker_cache
from mako.lookup import TemplateLookup
from mako.template import Template
from test import eq_
from test import module_base
from test import TemplateTest
from test.util import result_lines

if beaker_cache.has_beaker:
    import beaker


class SimpleBackend(object):
    def __init__(self):
        self.cache = {}

    def get(self, key, **kw):
        return self.cache[key]

    def invalidate(self, key, **kw):
        self.cache.pop(key, None)

    def put(self, key, value, **kw):
        self.cache[key] = value

    def get_or_create(self, key, creation_function, **kw):
        if key in self.cache:
            return self.cache[key]
        else:
            self.cache[key] = value = creation_function()
            return value


class MockCacheImpl(CacheImpl):
    realcacheimpl = None

    def __init__(self, cache):
        self.cache = cache

    def set_backend(self, cache, backend):
        if backend == "simple":
            self.realcacheimpl = SimpleBackend()
        else:
            self.realcacheimpl = cache._load_impl(backend)

    def _setup_kwargs(self, kw):
        self.kwargs = kw.copy()
        self.kwargs.pop("regions", None)
        self.kwargs.pop("manager", None)
        if self.kwargs.get("region") != "myregion":
            self.kwargs.pop("region", None)

    def get_or_create(self, key, creation_function, **kw):
        self.key = key
        self._setup_kwargs(kw)
        return self.realcacheimpl.get_or_create(key, creation_function, **kw)

    def put(self, key, value, **kw):
        self.key = key
        self._setup_kwargs(kw)
        self.realcacheimpl.put(key, value, **kw)

    def get(self, key, **kw):
        self.key = key
        self._setup_kwargs(kw)
        return self.realcacheimpl.get(key, **kw)

    def invalidate(self, key, **kw):
        self.key = key
        self._setup_kwargs(kw)
        self.realcacheimpl.invalidate(key, **kw)


register_plugin("mock", __name__, "MockCacheImpl")


class CacheTest(TemplateTest):

    real_backend = "simple"

    def _install_mock_cache(self, template, implname=None):
        template.cache_impl = "mock"
        impl = template.cache.impl
        impl.set_backend(template.cache, implname or self.real_backend)
        return impl

    def test_def(self):
        t = Template(
            """
        <%!
            callcount = [0]
        %>
        <%def name="foo()" cached="True">
            this is foo
            <%
            callcount[0] += 1
            %>
        </%def>

        ${foo()}
        ${foo()}
        ${foo()}
        callcount: ${callcount}
"""
        )
        m = self._install_mock_cache(t)
        assert result_lines(t.render()) == [
            "this is foo",
            "this is foo",
            "this is foo",
            "callcount: [1]",
        ]
        assert m.kwargs == {}

    def test_cache_enable(self):
        t = Template(
            """
            <%!
                callcount = [0]
            %>
            <%def name="foo()" cached="True">
                <% callcount[0] += 1 %>
            </%def>
            ${foo()}
            ${foo()}
            callcount: ${callcount}
        """,
            cache_enabled=False,
        )
        self._install_mock_cache(t)

        eq_(t.render().strip(), "callcount: [2]")

    def test_nested_def(self):
        t = Template(
            """
        <%!
            callcount = [0]
        %>
        <%def name="foo()">
            <%def name="bar()" cached="True">
                this is foo
                <%
                callcount[0] += 1
                %>
            </%def>
            ${bar()}
        </%def>

        ${foo()}
        ${foo()}
        ${foo()}
        callcount: ${callcount}
"""
        )
        m = self._install_mock_cache(t)
        assert result_lines(t.render()) == [
            "this is foo",
            "this is foo",
            "this is foo",
            "callcount: [1]",
        ]
        assert m.kwargs == {}

    def test_page(self):
        t = Template(
            """
        <%!
            callcount = [0]
        %>
        <%page cached="True"/>
        this is foo
        <%
        callcount[0] += 1
        %>
        callcount: ${callcount}
"""
        )
        m = self._install_mock_cache(t)
        t.render()
        t.render()
        assert result_lines(t.render()) == ["this is foo", "callcount: [1]"]
        assert m.kwargs == {}

    def test_dynamic_key_with_context(self):
        t = Template(
            """
            <%block name="foo" cached="True" cache_key="${mykey}">
                some block
            </%block>
        """
        )
        m = self._install_mock_cache(t)
        t.render(mykey="thekey")
        t.render(mykey="thekey")
        eq_(result_lines(t.render(mykey="thekey")), ["some block"])
        eq_(m.key, "thekey")

        t = Template(
            """
            <%def name="foo()" cached="True" cache_key="${mykey}">
                some def
            </%def>
            ${foo()}
        """
        )
        m = self._install_mock_cache(t)
        t.render(mykey="thekey")
        t.render(mykey="thekey")
        eq_(result_lines(t.render(mykey="thekey")), ["some def"])
        eq_(m.key, "thekey")

    def test_dynamic_key_with_funcargs(self):
        t = Template(
            """
            <%def name="foo(num=5)" cached="True" cache_key="foo_${str(num)}">
             hi
            </%def>

            ${foo()}
        """
        )
        m = self._install_mock_cache(t)
        t.render()
        t.render()
        assert result_lines(t.render()) == ["hi"]
        assert m.key == "foo_5"

        t = Template(
            """
            <%def name="foo(*args, **kwargs)" cached="True"
             cache_key="foo_${kwargs['bar']}">
             hi
            </%def>

            ${foo(1, 2, bar='lala')}
        """
        )
        m = self._install_mock_cache(t)
        t.render()
        assert result_lines(t.render()) == ["hi"]
        assert m.key == "foo_lala"

        t = Template(
            """
        <%page args="bar='hi'" cache_key="foo_${bar}" cached="True"/>
         hi
        """
        )
        m = self._install_mock_cache(t)
        t.render()
        assert result_lines(t.render()) == ["hi"]
        assert m.key == "foo_hi"

    def test_dynamic_key_with_imports(self):
        lookup = TemplateLookup()
        lookup.put_string(
            "foo.html",
            """
        <%!
            callcount = [0]
        %>
        <%namespace file="ns.html" import="*"/>
        <%page cached="True" cache_key="${foo}"/>
        this is foo
        <%
        callcount[0] += 1
        %>
        callcount: ${callcount}
""",
        )
        lookup.put_string("ns.html", """""")
        t = lookup.get_template("foo.html")
        m = self._install_mock_cache(t)
        t.render(foo="somekey")
        t.render(foo="somekey")
        assert result_lines(t.render(foo="somekey")) == [
            "this is foo",
            "callcount: [1]",
        ]
        assert m.kwargs == {}

    def test_fileargs_implicit(self):
        l = lookup.TemplateLookup(module_directory=module_base)
        l.put_string(
            "test",
            """
                <%!
                    callcount = [0]
                %>
                <%def name="foo()" cached="True" cache_type='dbm'>
                    this is foo
                    <%
                    callcount[0] += 1
                    %>
                </%def>

                ${foo()}
                ${foo()}
                ${foo()}
                callcount: ${callcount}
        """,
        )

        m = self._install_mock_cache(l.get_template("test"))
        assert result_lines(l.get_template("test").render()) == [
            "this is foo",
            "this is foo",
            "this is foo",
            "callcount: [1]",
        ]
        eq_(m.kwargs, {"type": "dbm"})

    def test_fileargs_deftag(self):
        t = Template(
            """
        <%%!
            callcount = [0]
        %%>
        <%%def name="foo()" cached="True" cache_type='file' cache_dir='%s'>
            this is foo
            <%%
            callcount[0] += 1
            %%>
        </%%def>

        ${foo()}
        ${foo()}
        ${foo()}
        callcount: ${callcount}
"""
            % module_base
        )
        m = self._install_mock_cache(t)
        assert result_lines(t.render()) == [
            "this is foo",
            "this is foo",
            "this is foo",
            "callcount: [1]",
        ]
        assert m.kwargs == {"type": "file", "dir": module_base}

    def test_fileargs_pagetag(self):
        t = Template(
            """
        <%%page cache_dir='%s' cache_type='dbm'/>
        <%%!
            callcount = [0]
        %%>
        <%%def name="foo()" cached="True">
            this is foo
            <%%
            callcount[0] += 1
            %%>
        </%%def>

        ${foo()}
        ${foo()}
        ${foo()}
        callcount: ${callcount}
"""
            % module_base
        )
        m = self._install_mock_cache(t)
        assert result_lines(t.render()) == [
            "this is foo",
            "this is foo",
            "this is foo",
            "callcount: [1]",
        ]
        eq_(m.kwargs, {"dir": module_base, "type": "dbm"})

    def test_args_complete(self):
        t = Template(
            """
        <%%def name="foo()" cached="True" cache_timeout="30" cache_dir="%s"
         cache_type="file" cache_key='somekey'>
            this is foo
        </%%def>

        ${foo()}
"""
            % module_base
        )
        m = self._install_mock_cache(t)
        t.render()
        eq_(m.kwargs, {"dir": module_base, "type": "file", "timeout": 30})

        t2 = Template(
            """
        <%%page cached="True" cache_timeout="30" cache_dir="%s"
         cache_type="file" cache_key='somekey'/>
        hi
        """
            % module_base
        )
        m = self._install_mock_cache(t2)
        t2.render()
        eq_(m.kwargs, {"dir": module_base, "type": "file", "timeout": 30})

    def test_fileargs_lookup(self):
        l = lookup.TemplateLookup(cache_dir=module_base, cache_type="file")
        l.put_string(
            "test",
            """
                <%!
                    callcount = [0]
                %>
                <%def name="foo()" cached="True">
                    this is foo
                    <%
                    callcount[0] += 1
                    %>
                </%def>

                ${foo()}
                ${foo()}
                ${foo()}
                callcount: ${callcount}
        """,
        )

        t = l.get_template("test")
        m = self._install_mock_cache(t)
        assert result_lines(l.get_template("test").render()) == [
            "this is foo",
            "this is foo",
            "this is foo",
            "callcount: [1]",
        ]
        eq_(m.kwargs, {"dir": module_base, "type": "file"})

    def test_buffered(self):
        t = Template(
            """
        <%!
            def a(text):
                return "this is a " + text.strip()
        %>
        ${foo()}
        ${foo()}
        <%def name="foo()" cached="True" buffered="True">
            this is a test
        </%def>
        """,
            buffer_filters=["a"],
        )
        self._install_mock_cache(t)
        eq_(
            result_lines(t.render()),
            ["this is a this is a test", "this is a this is a test"],
        )

    def test_load_from_expired(self):
        """test that the cache callable can be called safely after the
        originating template has completed rendering.

        """
        t = Template(
            """
        ${foo()}
        <%def name="foo()" cached="True" cache_timeout="1">
            foo
        </%def>
        """
        )
        self._install_mock_cache(t)

        x1 = t.render()
        time.sleep(1.2)
        x2 = t.render()
        assert x1.strip() == x2.strip() == "foo"

    def test_namespace_access(self):
        t = Template(
            """
            <%def name="foo(x)" cached="True">
                foo: ${x}
            </%def>

            <%
                foo(1)
                foo(2)
                local.cache.invalidate_def('foo')
                foo(3)
                foo(4)
            %>
        """
        )
        self._install_mock_cache(t)
        eq_(result_lines(t.render()), ["foo: 1", "foo: 1", "foo: 3", "foo: 3"])

    def test_lookup(self):
        l = TemplateLookup(cache_impl="mock")
        l.put_string(
            "x",
            """
            <%page cached="True" />
            ${y}
        """,
        )
        t = l.get_template("x")
        self._install_mock_cache(t)
        assert result_lines(t.render(y=5)) == ["5"]
        assert result_lines(t.render(y=7)) == ["5"]
        assert isinstance(t.cache.impl, MockCacheImpl)

    def test_invalidate(self):
        t = Template(
            """
            <%%def name="foo()" cached="True">
                foo: ${x}
            </%%def>

            <%%def name="bar()" cached="True" cache_type='dbm' cache_dir='%s'>
                bar: ${x}
            </%%def>
            ${foo()} ${bar()}
        """
            % module_base
        )
        self._install_mock_cache(t)
        assert result_lines(t.render(x=1)) == ["foo: 1", "bar: 1"]
        assert result_lines(t.render(x=2)) == ["foo: 1", "bar: 1"]
        t.cache.invalidate_def("foo")
        assert result_lines(t.render(x=3)) == ["foo: 3", "bar: 1"]
        t.cache.invalidate_def("bar")
        assert result_lines(t.render(x=4)) == ["foo: 3", "bar: 4"]

        t = Template(
            """
            <%%page cached="True" cache_type="dbm" cache_dir="%s"/>

            page: ${x}
        """
            % module_base
        )
        self._install_mock_cache(t)
        assert result_lines(t.render(x=1)) == ["page: 1"]
        assert result_lines(t.render(x=2)) == ["page: 1"]
        t.cache.invalidate_body()
        assert result_lines(t.render(x=3)) == ["page: 3"]
        assert result_lines(t.render(x=4)) == ["page: 3"]

    def test_custom_args_def(self):
        t = Template(
            """
            <%def name="foo()" cached="True" cache_region="myregion"
                    cache_timeout="50" cache_foo="foob">
            </%def>
            ${foo()}
        """
        )
        m = self._install_mock_cache(t, "simple")
        t.render()
        eq_(m.kwargs, {"region": "myregion", "timeout": 50, "foo": "foob"})

    def test_custom_args_block(self):
        t = Template(
            """
            <%block name="foo" cached="True" cache_region="myregion"
                    cache_timeout="50" cache_foo="foob">
            </%block>
        """
        )
        m = self._install_mock_cache(t, "simple")
        t.render()
        eq_(m.kwargs, {"region": "myregion", "timeout": 50, "foo": "foob"})

    def test_custom_args_page(self):
        t = Template(
            """
            <%page cached="True" cache_region="myregion"
                    cache_timeout="50" cache_foo="foob"/>
        """
        )
        m = self._install_mock_cache(t, "simple")
        t.render()
        eq_(m.kwargs, {"region": "myregion", "timeout": 50, "foo": "foob"})

    def test_pass_context(self):
        t = Template(
            """
            <%page cached="True"/>
        """
        )
        m = self._install_mock_cache(t)
        t.render()
        assert "context" not in m.kwargs

        m.pass_context = True
        t.render(x="bar")
        assert "context" in m.kwargs
        assert m.kwargs["context"].get("x") == "bar"


class RealBackendTest(object):
    def test_cache_uses_current_context(self):
        t = Template(
            """
        ${foo()}
        <%def name="foo()" cached="True" cache_timeout="1">
            foo: ${x}
        </%def>
        """
        )
        self._install_mock_cache(t)

        x1 = t.render(x=1)
        time.sleep(1.2)
        x2 = t.render(x=2)
        eq_(x1.strip(), "foo: 1")
        eq_(x2.strip(), "foo: 2")

    def test_region(self):
        t = Template(
            """
            <%block name="foo" cached="True" cache_region="short">
                short term ${x}
            </%block>
            <%block name="bar" cached="True" cache_region="long">
                long term ${x}
            </%block>
            <%block name="lala">
                none ${x}
            </%block>
        """
        )

        self._install_mock_cache(t)
        r1 = result_lines(t.render(x=5))
        time.sleep(1.2)
        r2 = result_lines(t.render(x=6))
        r3 = result_lines(t.render(x=7))
        eq_(r1, ["short term 5", "long term 5", "none 5"])
        eq_(r2, ["short term 6", "long term 5", "none 6"])
        eq_(r3, ["short term 6", "long term 5", "none 7"])


class BeakerCacheTest(RealBackendTest, CacheTest):
    real_backend = "beaker"

    def setUp(self):
        if not beaker_cache.has_beaker:
            raise unittest.SkipTest("Beaker is required for these tests.")
        if not py27:
            raise unittest.SkipTest("newer beakers not working w/ py26")

    def _install_mock_cache(self, template, implname=None):
        template.cache_args["manager"] = self._regions()
        impl = super(BeakerCacheTest, self)._install_mock_cache(
            template, implname
        )
        return impl

    def _regions(self):
        return beaker.cache.CacheManager(
            cache_regions={
                "short": {"expire": 1, "type": "memory"},
                "long": {"expire": 60, "type": "memory"},
            }
        )


class DogpileCacheTest(RealBackendTest, CacheTest):
    real_backend = "dogpile.cache"

    def setUp(self):
        try:
            import dogpile.cache  # noqa
        except ImportError:
            raise unittest.SkipTest(
                "dogpile.cache is required to run these tests"
            )

    def _install_mock_cache(self, template, implname=None):
        template.cache_args["regions"] = self._regions()
        template.cache_args.setdefault("region", "short")
        impl = super(DogpileCacheTest, self)._install_mock_cache(
            template, implname
        )
        return impl

    def _regions(self):
        from dogpile.cache import make_region

        my_regions = {
            "short": make_region().configure(
                "dogpile.cache.memory", expiration_time=1
            ),
            "long": make_region().configure(
                "dogpile.cache.memory", expiration_time=60
            ),
            "myregion": make_region().configure(
                "dogpile.cache.memory", expiration_time=60
            ),
        }

        return my_regions