from __future__ import annotations

import os
import pickle
import time
import types
from decimal import Decimal
from io import StringIO
from typing import Any

import pytest
from django.core.cache import CacheKeyWarning, cache, caches
from django.core.management import CommandError, call_command
from django.db import IntegrityError, connection
from django.db.migrations.state import ProjectState
from django.http import HttpResponse
from django.middleware.cache import FetchFromCacheMiddleware, UpdateCacheMiddleware
from django.test import RequestFactory, TestCase, TransactionTestCase
from django.test.utils import override_settings
from parameterized import parameterized

from django_mysql.cache import BIGINT_SIGNED_MAX, BIGINT_SIGNED_MIN, MySQLCache
from tests.testapp.models import Poll, expensive_calculation


# functions/classes for complex data type tests
def f():  # pragma: no cover
    return 42


class C:
    def m(n):  # pragma: no cover
        return 24


class Unpickable:
    def __getstate__(self):
        raise pickle.PickleError()


def empty_response(request):  # pragma: no cover
    return HttpResponse()


class MyInt(int):
    def times2(self):
        return self * 2


def custom_key_func(key, key_prefix, version):
    "A customized cache key function"
    return "CUSTOM-" + "-".join([key_prefix, str(version), key])


def reverse_custom_key_func(full_key):
    "The reverse of custom_key_func"
    # Remove CUSTOM-
    full_key = full_key[len("CUSTOM-") :]

    first_dash = full_key.find("-")
    key_prefix = full_key[:first_dash]

    second_dash = full_key.find("-", first_dash + 1)
    version = int(full_key[first_dash + 1 : second_dash])

    key = full_key[second_dash + 1 :]

    return key, key_prefix, version


_caches_setting_base = {
    "default": {},
    "prefix": {"KEY_PREFIX": f"cacheprefix{os.getpid()}"},
    "v2": {"VERSION": 2},
    "custom_key": {
        "KEY_FUNCTION": custom_key_func,
        "REVERSE_KEY_FUNCTION": reverse_custom_key_func,
    },
    "custom_key2": {
        "KEY_FUNCTION": __name__ + ".custom_key_func",
        "REVERSE_KEY_FUNCTION": __name__ + ".reverse_custom_key_func",
    },
    "cull": {"OPTIONS": {"CULL_PROBABILITY": 1, "MAX_ENTRIES": 30}},
    "zero_cull": {
        "OPTIONS": {"CULL_FREQUENCY": 0, "CULL_PROBABILITY": 1, "MAX_ENTRIES": 30}
    },
    "no_cull": {
        "OPTIONS": {"CULL_FREQUENCY": 2, "CULL_PROBABILITY": 0, "MAX_ENTRIES": 30}
    },
}


def caches_setting_for_tests(options=None, **params):
    # `params` are test specific overrides and `_caches_settings_base` is the
    # base config for the tests.
    # This results in the following search order:
    # params -> _caches_setting_base -> base
    setting: dict[str, Any] = {k: {} for k in _caches_setting_base}
    for key, cache_params in setting.items():
        cache_params.update(_caches_setting_base[key])
        cache_params.update(params)
        if options is not None:
            cache_params["OPTIONS"] = cache_params.get("OPTIONS", {}).copy()
            cache_params["OPTIONS"].update(**options)
    return setting


# Spaces are used in the table name to ensure quoting/escaping is working
def override_cache_settings(
    BACKEND="django_mysql.cache.MySQLCache", LOCATION="test cache table", **kwargs
):
    return override_settings(
        CACHES=caches_setting_for_tests(BACKEND=BACKEND, LOCATION=LOCATION, **kwargs)
    )


class MySQLCacheTableMixin(TransactionTestCase):
    table_name = "test cache table"

    @classmethod
    def create_table(self):
        sql = MySQLCache.create_table_sql.format(table_name=self.table_name)
        with connection.cursor() as cursor:
            cursor.execute(sql)

    @classmethod
    def drop_table(self):
        with connection.cursor() as cursor:
            cursor.execute(f"DROP TABLE `{self.table_name}`")


@override_cache_settings()
class MySQLCacheTests(MySQLCacheTableMixin, TestCase):
    factory = RequestFactory()

    @classmethod
    def setUpClass(cls):
        cls.create_table()
        super().setUpClass()

    @classmethod
    def tearDownClass(cls):
        super().tearDownClass()
        cls.drop_table()

    def table_count(self):
        with connection.cursor() as cursor:
            cursor.execute(f"SELECT COUNT(*) FROM `{self.table_name}`")
            return cursor.fetchone()[0]

    # These tests were copied from django's tests/cache/tests.py file

    def test_simple(self):
        # Simple cache set/get works
        cache.set("key", "value")
        assert cache.get("key") == "value"

    def test_add(self):
        # A key can be added to a cache
        result = cache.add("addkey1", "value")
        assert result
        result = cache.add("addkey1", "newvalue")
        assert not result
        assert cache.get("addkey1") == "value"

    def test_prefix(self):
        # Test for same cache key conflicts between shared backend
        cache.set("somekey", "value")

        # should not be set in the prefixed cache
        assert not caches["prefix"].has_key("somekey")  # noqa

        caches["prefix"].set("somekey", "value2")

        assert cache.get("somekey") == "value"
        assert caches["prefix"].get("somekey") == "value2"

    def test_non_existent(self):
        # Non-existent cache keys return as None/default
        # get with non-existent keys
        assert cache.get("does_not_exist") is None
        assert cache.get("does_not_exist", "bang!") == "bang!"

    def test_delete(self):
        # Cache keys can be deleted
        cache.set("key1", "spam")
        cache.set("key2", "eggs")
        assert cache.get("key1") == "spam"
        cache.delete("key1")
        assert cache.get("key1") is None
        assert cache.get("key2") == "eggs"

    def test_has_key(self):
        # The cache can be inspected for cache keys
        cache.set("hello1", "goodbye1")
        assert cache.has_key("hello1")  # noqa
        assert not cache.has_key("goodbye1")  # noqa
        cache.set("no_expiry", "here", None)
        assert cache.has_key("no_expiry")  # noqa

    def test_in(self):
        # The in operator can be used to inspect cache contents
        cache.set("hello2", "goodbye2")
        assert "hello2" in cache
        assert "goodbye2" not in cache

    def test_incr(self):
        # Cache values can be incremented
        cache.set("answer", 41)
        assert cache.incr("answer") == 42
        assert cache.get("answer") == 42
        assert cache.incr("answer", 10) == 52
        assert cache.get("answer") == 52
        assert cache.incr("answer", -10) == 42
        with pytest.raises(ValueError):
            cache.incr("does_not_exist")

    def test_decr(self):
        # Cache values can be decremented
        cache.set("answer", 43)
        assert cache.decr("answer") == 42
        assert cache.get("answer") == 42
        assert cache.decr("answer", 10) == 32
        assert cache.get("answer") == 32
        assert cache.decr("answer", -10) == 42
        with pytest.raises(ValueError):
            cache.decr("does_not_exist")

    def test_close(self):
        assert hasattr(cache, "close")
        cache.close()

    def test_data_types(self):
        # Many different data types can be cached
        stuff = {
            "string": "this is a string",
            "int": 42,
            "list": [1, 2, 3, 4],
            "tuple": (1, 2, 3, 4),
            "dict": {"A": 1, "B": 2},
            "function": f,
            "class": C,
        }
        cache.set("stuff", stuff)
        assert cache.get("stuff") == stuff

    def test_cache_read_for_model_instance(self):
        # Don't want fields with callable as default to be called on cache read
        expensive_calculation.reset()
        Poll.objects.all().delete()
        my_poll = Poll.objects.create(question="Well?")
        assert Poll.objects.count() == 1
        pub_date = my_poll.pub_date
        cache.set("question", my_poll)
        cached_poll = cache.get("question")
        assert cached_poll.pub_date == pub_date
        # We only want the default expensive calculation run once
        assert expensive_calculation.call_count() == 1

    def test_cache_write_for_model_instance_with_deferred(self):
        # Don't want fields with callable as default to be called on cache
        # write
        expensive_calculation.reset()
        Poll.objects.all().delete()
        Poll.objects.create(question="What?")
        assert expensive_calculation.call_count() == 1
        defer_qs = Poll.objects.all().defer("question")
        assert defer_qs.count() == 1
        assert expensive_calculation.call_count() == 1
        cache.set("deferred_queryset", defer_qs)
        # cache set should not re-evaluate default functions
        assert expensive_calculation.call_count() == 1

    def test_cache_read_for_model_instance_with_deferred(self):
        # Don't want fields with callable as default to be called on cache read
        expensive_calculation.reset()
        Poll.objects.all().delete()
        Poll.objects.create(question="What?")
        assert expensive_calculation.call_count() == 1
        defer_qs = Poll.objects.all().defer("question")
        assert defer_qs.count() == 1
        cache.set("deferred_queryset", defer_qs)
        assert expensive_calculation.call_count() == 1
        runs_before_cache_read = expensive_calculation.call_count()
        cache.get("deferred_queryset")
        # We only want the default expensive calculation run on creation and
        # set
        assert expensive_calculation.call_count() == runs_before_cache_read

    def test_unicode(self):
        # Unicode values can be cached
        stuff = {
            "ascii": "ascii_value",
            "unicode_ascii": "Iñtërnâtiônàlizætiøn1",
            "Iñtërnâtiônàlizætiøn": "Iñtërnâtiônàlizætiøn2",
            "ascii2": {"x": 1},
        }
        # Test `set`
        for key, value in stuff.items():
            cache.set(key, value)
            assert cache.get(key) == value

        # Test `add`
        for key, value in stuff.items():
            cache.delete(key)
            cache.add(key, value)
            assert cache.get(key) == value

        # Test `set_many`
        for key, _value in stuff.items():
            cache.delete(key)
        cache.set_many(stuff)
        for key, value in stuff.items():
            assert cache.get(key) == value

    def test_binary_string(self):
        # Binary strings should be cacheable
        from zlib import compress, decompress

        value = "value_to_be_compressed"
        compressed_value = compress(value.encode())

        # Test set
        cache.set("binary1", compressed_value)
        compressed_result = cache.get("binary1")
        assert compressed_value == compressed_result
        assert value == decompress(compressed_result).decode()

        # Test add
        cache.add("binary1-add", compressed_value)
        compressed_result = cache.get("binary1-add")
        assert compressed_value == compressed_result
        assert value == decompress(compressed_result).decode()

        # Test set_many
        cache.set_many({"binary1-set_many": compressed_value})
        compressed_result = cache.get("binary1-set_many")
        assert compressed_value == compressed_result
        assert value == decompress(compressed_result).decode()

    def test_clear(self):
        # The cache can be emptied using clear
        cache.set("key1", "spam")
        cache.set("key2", "eggs")
        cache.clear()
        assert cache.get("key1") is None
        assert cache.get("key2") is None

    def test_touch_without_timeout(self):
        cache.set("key1", "spam", timeout=0.1)
        result = cache.touch("key1", timeout=0.4)
        assert result is True
        time.sleep(0.2)
        assert "key1" in cache

    def test_touch_with_timeout(self):
        cache.set("key1", "spam", timeout=0.1)
        result = cache.touch("key1")
        assert result is True
        time.sleep(0.2)
        assert "key1" in cache

    def test_touch_already_expired(self):
        cache.set("key1", "spam", timeout=0.1)
        time.sleep(0.2)
        result = cache.touch("key1", timeout=0.4)
        assert result is False
        assert "key1" not in cache

    def test_long_timeout(self):
        """
        Using a timeout greater than 30 days makes memcached think
        it is an absolute expiration timestamp instead of a relative
        offset. Test that we honour this convention. Refs #12399.
        """
        cache.set("key1", "eggs", 60 * 60 * 24 * 30 + 1)  # 30 days + 1 second
        assert cache.get("key1") == "eggs"

        cache.add("key2", "ham", 60 * 60 * 24 * 30 + 1)
        assert cache.get("key2") == "ham"

        cache.set_many(
            {"key3": "sausage", "key4": "lobster bisque"}, 60 * 60 * 24 * 30 + 1
        )
        assert cache.get("key3") == "sausage"
        assert cache.get("key4") == "lobster bisque"

    def test_forever_timeout(self):
        """
        Passing in None into timeout results in a value that is cached forever
        """
        cache.set("key1", "eggs", None)
        assert cache.get("key1") == "eggs"

        cache.add("key2", "ham", None)
        assert cache.get("key2") == "ham"
        added = cache.add("key1", "new eggs", None)
        assert not added
        assert cache.get("key1") == "eggs"

        cache.set_many({"key3": "sausage", "key4": "lobster bisque"}, None)
        assert cache.get("key3") == "sausage"
        assert cache.get("key4") == "lobster bisque"

    def test_zero_timeout(self):
        """
        Passing in zero into timeout results in a value that is not cached
        """
        cache.set("key1", "eggs", 0)
        assert cache.get("key1") is None

        cache.add("key2", "ham", 0)
        assert cache.get("key2") is None

        cache.set_many({"key3": "sausage", "key4": "lobster bisque"}, 0)
        assert cache.get("key3") is None
        assert cache.get("key4") is None

    def test_float_timeout(self):
        # Make sure a timeout given as a float doesn't crash anything.
        cache.set("key1", "spam", 100.2)
        assert cache.get("key1") == "spam"

    def test_cache_versioning_get_set(self):
        # set, using default version = 1
        cache.set("answer1", 42)
        assert cache.get("answer1") == 42
        assert cache.get("answer1", version=1) == 42
        assert cache.get("answer1", version=2) is None

        assert caches["v2"].get("answer1") is None
        assert caches["v2"].get("answer1", version=1) == 42
        assert caches["v2"].get("answer1", version=2) is None

        # set, default version = 1, but manually override version = 2
        cache.set("answer2", 42, version=2)
        assert cache.get("answer2") is None
        assert cache.get("answer2", version=1) is None
        assert cache.get("answer2", version=2) == 42

        assert caches["v2"].get("answer2") == 42
        assert caches["v2"].get("answer2", version=1) is None
        assert caches["v2"].get("answer2", version=2) == 42

        # v2 set, using default version = 2
        caches["v2"].set("answer3", 42)
        assert cache.get("answer3") is None
        assert cache.get("answer3", version=1) is None
        assert cache.get("answer3", version=2) == 42

        assert caches["v2"].get("answer3") == 42
        assert caches["v2"].get("answer3", version=1) is None
        assert caches["v2"].get("answer3", version=2) == 42

        # v2 set, default version = 2, but manually override version = 1
        caches["v2"].set("answer4", 42, version=1)
        assert cache.get("answer4") == 42
        assert cache.get("answer4", version=1) == 42
        assert cache.get("answer4", version=2) is None

        assert caches["v2"].get("answer4") is None
        assert caches["v2"].get("answer4", version=1) == 42
        assert caches["v2"].get("answer4", version=2) is None

    def test_cache_versioning_add(self):
        # add, default version = 1, but manually override version = 2
        cache.add("answer1", 42, version=2)
        assert cache.get("answer1", version=1) is None
        assert cache.get("answer1", version=2) == 42

        cache.add("answer1", 37, version=2)
        assert cache.get("answer1", version=1) is None
        assert cache.get("answer1", version=2) == 42

        cache.add("answer1", 37, version=1)
        assert cache.get("answer1", version=1) == 37
        assert cache.get("answer1", version=2) == 42

        # v2 add, using default version = 2
        caches["v2"].add("answer2", 42)
        assert cache.get("answer2", version=1) is None
        assert cache.get("answer2", version=2) == 42

        caches["v2"].add("answer2", 37)
        assert cache.get("answer2", version=1) is None
        assert cache.get("answer2", version=2) == 42

        caches["v2"].add("answer2", 37, version=1)
        assert cache.get("answer2", version=1) == 37
        assert cache.get("answer2", version=2) == 42

        # v2 add, default version = 2, but manually override version = 1
        caches["v2"].add("answer3", 42, version=1)
        assert cache.get("answer3", version=1) == 42
        assert cache.get("answer3", version=2) is None

        caches["v2"].add("answer3", 37, version=1)
        assert cache.get("answer3", version=1) == 42
        assert cache.get("answer3", version=2) is None

        caches["v2"].add("answer3", 37)
        assert cache.get("answer3", version=1) == 42
        assert cache.get("answer3", version=2) == 37

    def test_cache_versioning_has_key(self):
        cache.set("answer1", 42)

        # has_key
        assert cache.has_key("answer1")  # noqa
        assert cache.has_key("answer1", version=1)  # noqa
        assert not cache.has_key("answer1", version=2)  # noqa

        assert not caches["v2"].has_key("answer1")  # noqa
        assert caches["v2"].has_key("answer1", version=1)  # noqa
        assert not caches["v2"].has_key("answer1", version=2)  # noqa

    def test_cache_versioning_delete(self):
        cache.set("answer1", 37, version=1)
        cache.set("answer1", 42, version=2)
        cache.delete("answer1")
        assert cache.get("answer1", version=1) is None
        assert cache.get("answer1", version=2) == 42

        cache.set("answer2", 37, version=1)
        cache.set("answer2", 42, version=2)
        cache.delete("answer2", version=2)
        assert cache.get("answer2", version=1) == 37
        assert cache.get("answer2", version=2) is None

        cache.set("answer3", 37, version=1)
        cache.set("answer3", 42, version=2)
        caches["v2"].delete("answer3")
        assert cache.get("answer3", version=1) == 37
        assert cache.get("answer3", version=2) is None

        cache.set("answer4", 37, version=1)
        cache.set("answer4", 42, version=2)
        caches["v2"].delete("answer4", version=1)
        assert cache.get("answer4", version=1) is None
        assert cache.get("answer4", version=2) == 42

    def test_cache_versioning_incr_decr(self):
        cache.set("answer1", 37, version=1)
        cache.set("answer1", 42, version=2)
        cache.incr("answer1")
        assert cache.get("answer1", version=1) == 38
        assert cache.get("answer1", version=2) == 42
        cache.decr("answer1")
        assert cache.get("answer1", version=1) == 37
        assert cache.get("answer1", version=2) == 42

        cache.set("answer2", 37, version=1)
        cache.set("answer2", 42, version=2)
        cache.incr("answer2", version=2)
        assert cache.get("answer2", version=1) == 37
        assert cache.get("answer2", version=2) == 43
        cache.decr("answer2", version=2)
        assert cache.get("answer2", version=1) == 37
        assert cache.get("answer2", version=2) == 42

        cache.set("answer3", 37, version=1)
        cache.set("answer3", 42, version=2)
        caches["v2"].incr("answer3")
        assert cache.get("answer3", version=1) == 37
        assert cache.get("answer3", version=2) == 43
        caches["v2"].decr("answer3")
        assert cache.get("answer3", version=1) == 37
        assert cache.get("answer3", version=2) == 42

        cache.set("answer4", 37, version=1)
        cache.set("answer4", 42, version=2)
        caches["v2"].incr("answer4", version=1)
        assert cache.get("answer4", version=1) == 38
        assert cache.get("answer4", version=2) == 42
        caches["v2"].decr("answer4", version=1)
        assert cache.get("answer4", version=1) == 37
        assert cache.get("answer4", version=2) == 42

    def test_cache_versioning_get_set_many(self):
        # set, using default version = 1
        cache.set_many({"ford1": 37, "arthur1": 42})
        assert cache.get_many(["ford1", "arthur1"]) == {"ford1": 37, "arthur1": 42}
        assert cache.get_many(["ford1", "arthur1"], version=1) == {
            "ford1": 37,
            "arthur1": 42,
        }
        assert cache.get_many(["ford1", "arthur1"], version=2) == {}

        assert caches["v2"].get_many(["ford1", "arthur1"]) == {}
        assert caches["v2"].get_many(["ford1", "arthur1"], version=1) == {
            "ford1": 37,
            "arthur1": 42,
        }
        assert caches["v2"].get_many(["ford1", "arthur1"], version=2) == {}

        # set, default version = 1, but manually override version = 2
        cache.set_many({"ford2": 37, "arthur2": 42}, version=2)
        assert cache.get_many(["ford2", "arthur2"]) == {}
        assert cache.get_many(["ford2", "arthur2"], version=1) == {}
        assert cache.get_many(["ford2", "arthur2"], version=2) == {
            "ford2": 37,
            "arthur2": 42,
        }

        assert caches["v2"].get_many(["ford2", "arthur2"]) == {
            "ford2": 37,
            "arthur2": 42,
        }
        assert caches["v2"].get_many(["ford2", "arthur2"], version=1) == {}
        assert caches["v2"].get_many(["ford2", "arthur2"], version=2) == {
            "ford2": 37,
            "arthur2": 42,
        }

        # v2 set, using default version = 2
        caches["v2"].set_many({"ford3": 37, "arthur3": 42})
        assert cache.get_many(["ford3", "arthur3"]) == {}
        assert cache.get_many(["ford3", "arthur3"], version=1) == {}
        assert cache.get_many(["ford3", "arthur3"], version=2) == {
            "ford3": 37,
            "arthur3": 42,
        }

        assert caches["v2"].get_many(["ford3", "arthur3"]) == {
            "ford3": 37,
            "arthur3": 42,
        }
        assert caches["v2"].get_many(["ford3", "arthur3"], version=1) == {}
        assert caches["v2"].get_many(["ford3", "arthur3"], version=2) == {
            "ford3": 37,
            "arthur3": 42,
        }

        # v2 set, default version = 2, but manually override version = 1
        caches["v2"].set_many({"ford4": 37, "arthur4": 42}, version=1)
        assert cache.get_many(["ford4", "arthur4"]) == {"ford4": 37, "arthur4": 42}
        assert cache.get_many(["ford4", "arthur4"], version=1) == {
            "ford4": 37,
            "arthur4": 42,
        }
        assert cache.get_many(["ford4", "arthur4"], version=2) == {}

        assert caches["v2"].get_many(["ford4", "arthur4"]) == {}
        assert caches["v2"].get_many(["ford4", "arthur4"], version=1) == {
            "ford4": 37,
            "arthur4": 42,
        }
        assert caches["v2"].get_many(["ford4", "arthur4"], version=2) == {}

    def test_incr_version(self):
        cache.set("answer", 42, version=2)
        assert cache.get("answer") is None
        assert cache.get("answer", version=1) is None
        assert cache.get("answer", version=2) == 42
        assert cache.get("answer", version=3) is None

        assert cache.incr_version("answer", version=2) == 3
        assert cache.get("answer") is None
        assert cache.get("answer", version=1) is None
        assert cache.get("answer", version=2) is None
        assert cache.get("answer", version=3) == 42

        caches["v2"].set("answer2", 42)
        assert caches["v2"].get("answer2") == 42
        assert caches["v2"].get("answer2", version=1) is None
        assert caches["v2"].get("answer2", version=2) == 42
        assert caches["v2"].get("answer2", version=3) is None

        assert caches["v2"].incr_version("answer2") == 3
        assert caches["v2"].get("answer2") is None
        assert caches["v2"].get("answer2", version=1) is None
        assert caches["v2"].get("answer2", version=2) is None
        assert caches["v2"].get("answer2", version=3) == 42

        with pytest.raises(ValueError):
            cache.incr_version("does_not_exist")

    def test_decr_version(self):
        cache.set("answer", 42, version=2)
        assert cache.get("answer") is None
        assert cache.get("answer", version=1) is None
        assert cache.get("answer", version=2) == 42

        assert cache.decr_version("answer", version=2) == 1
        assert cache.get("answer") == 42
        assert cache.get("answer", version=1) == 42
        assert cache.get("answer", version=2) is None

        caches["v2"].set("answer2", 42)
        assert caches["v2"].get("answer2") == 42
        assert caches["v2"].get("answer2", version=1) is None
        assert caches["v2"].get("answer2", version=2) == 42

        assert caches["v2"].decr_version("answer2") == 1
        assert caches["v2"].get("answer2") is None
        assert caches["v2"].get("answer2", version=1) == 42
        assert caches["v2"].get("answer2", version=2) is None

        with pytest.raises(ValueError):
            cache.decr_version("does_not_exist", version=2)

    def test_custom_key_func(self):
        # Two caches with different key functions aren't visible to each other
        cache.set("answer1", 42)
        assert cache.get("answer1") == 42
        assert caches["custom_key"].get("answer1") is None
        assert caches["custom_key2"].get("answer1") is None

        caches["custom_key"].set("answer2", 42)
        assert cache.get("answer2") is None
        assert caches["custom_key"].get("answer2") == 42
        assert caches["custom_key2"].get("answer2") == 42

    def test_cache_write_unpicklable_object(self):
        fetch_middleware = FetchFromCacheMiddleware(empty_response)

        request = self.factory.get("/cache/test")
        request._cache_update_cache = True
        get_cache_data = FetchFromCacheMiddleware(empty_response).process_request(
            request
        )
        assert get_cache_data is None

        content = "Testing cookie serialization."

        def get_response(req):
            response = HttpResponse(content)
            response.set_cookie("foo", "bar")
            return response

        update_middleware = UpdateCacheMiddleware(get_response)
        response = update_middleware(request)

        get_cache_data = fetch_middleware.process_request(request)
        assert get_cache_data is not None
        assert get_cache_data.content == content.encode()
        assert get_cache_data.cookies == response.cookies

        UpdateCacheMiddleware(lambda req: get_cache_data)(request)
        get_cache_data = fetch_middleware.process_request(request)
        assert get_cache_data is not None
        assert get_cache_data.content == content.encode()
        assert get_cache_data.cookies == response.cookies

    def test_add_fail_on_pickleerror(self):
        "See https://code.djangoproject.com/ticket/21200"
        with pytest.raises(pickle.PickleError):
            cache.add("unpickable", Unpickable())

    def test_set_fail_on_pickleerror(self):
        "See https://code.djangoproject.com/ticket/21200"
        with pytest.raises(pickle.PickleError):
            cache.set("unpickable", Unpickable())

    def test_get_or_set(self):
        assert cache.get("projector") is None
        assert cache.get_or_set("projector", 42) == 42
        assert cache.get("projector") == 42

    def test_get_or_set_callable(self):
        def my_callable():
            return "value"

        assert cache.get_or_set("mykey", my_callable) == "value"

    def test_get_or_set_version(self):
        msg_re = r"get_or_set\(\) missing 1 required positional argument: 'default'"
        cache.get_or_set("brian", 1979, version=2)

        with pytest.raises(TypeError, match=msg_re):
            cache.get_or_set("brian")

        with pytest.raises(TypeError, match=msg_re):
            cache.get_or_set("brian", version=1)

        assert cache.get("brian", version=1) is None
        assert cache.get_or_set("brian", 42, version=1) == 42
        assert cache.get_or_set("brian", 1979, version=2) == 1979
        assert cache.get("brian", version=3) is None

    # Modified Django tests

    def test_expiration(self):
        # Cache values can be set to expire
        cache.set("expire1", "very quickly", 0.1)
        cache.set("expire2", "very quickly", 0.1)
        cache.set("expire3", "very quickly", 0.1)

        time.sleep(0.2)
        assert cache.get("expire1") is None

        cache.add("expire2", "newvalue")
        assert cache.get("expire2") == "newvalue"
        assert not cache.has_key("expire3")  # noqa

    def test_get_many(self):
        # Multiple cache keys can be returned using get_many
        cache.set("a", "a")
        cache.set("b", "b")
        cache.set("c", "c")
        cache.set("d", "d")

        with self.assertNumQueries(1):
            value = cache.get_many(["a", "c", "d"])
        assert value == {"a": "a", "c": "c", "d": "d"}

        with self.assertNumQueries(1):
            value = cache.get_many(["a", "b", "e"])

        assert value == {"a": "a", "b": "b"}

    def test_get_many_with_one_expired(self):
        # Multiple cache keys can be returned using get_many
        the_cache = caches["no_cull"]
        the_cache.set("a", "a", 0.1)
        time.sleep(0.2)

        the_cache.set("b", "b")
        the_cache.set("c", "c")
        the_cache.set("d", "d")

        with self.assertNumQueries(1):
            value = the_cache.get_many(["a", "c", "d"])
        assert value == {"c": "c", "d": "d"}

        with self.assertNumQueries(1):
            value = the_cache.get_many(["a", "b", "e"])

        assert value == {"b": "b"}

    def test_set_many(self):
        # Single keys can be set using set_many
        # Perform a single query first to avoid spurious on-connect queries
        caches["no_cull"].get("nonexistent")

        with self.assertNumQueries(1):
            result = caches["no_cull"].set_many({"key1": "spam"})
        assert result == []

        # Multiple keys can be set using set_many
        with self.assertNumQueries(1):
            result = caches["no_cull"].set_many({"key1": "spam", "key2": "eggs"})
        assert result == []
        assert cache.get("key1") == "spam"
        assert cache.get("key2") == "eggs"

    def test_set_many_expiration(self):
        # set_many takes a second ``timeout`` parameter
        # Perform a single query first to avoid spurious on-connect queries
        caches["no_cull"].get("nonexistent")
        with self.assertNumQueries(1):
            caches["no_cull"].set_many({"key1": "spam", "key2": "eggs"}, 0.1)
        cache.set("key3", "ham")
        time.sleep(0.2)

        assert cache.get("key1") is None
        assert cache.get("key2") is None
        assert cache.get("key3") == "ham"

        # set_many expired values can be replaced
        with self.assertNumQueries(1):
            caches["no_cull"].set_many(
                {"key1": "spam", "key2": "egg", "key3": "spam", "key4": "ham"}, 1
            )
        v = cache.get("key1")
        assert v == "spam"
        assert cache.get("key2") == "egg"
        assert cache.get("key3") == "spam"
        assert cache.get("key4") == "ham"

    def test_delete_many(self):
        # Multiple keys can be deleted using delete_many
        cache.set("key1", "spam")
        cache.set("key2", "eggs")
        cache.set("key3", "ham")
        with self.assertNumQueries(1):
            cache.delete_many(["key1", "key2"])
        assert cache.get("key1") is None
        assert cache.get("key2") is None
        assert cache.get("key3") == "ham"

    def test_invalid_keys(self):
        # mimic custom ``make_key`` method being defined since the default will
        # never show the below warnings
        def func(key, *args):
            return key

        old_func = cache.key_func
        cache.key_func = func

        try:
            with pytest.warns(CacheKeyWarning):
                # memcached does not allow whitespace or control characters in
                # keys
                cache.set("space key", "value")

            with pytest.raises(ValueError):
                # memcached limits key length to 250
                # We have a 250 character max length on our table
                cache.set("a" * 251, "value")
        finally:
            cache.key_func = old_func

    # Original tests

    def test_base_set_bad_value(self):
        with pytest.raises(ValueError) as excinfo:
            cache._base_set("foo", "key", "value")
        assert "'mode' should be" in str(excinfo.value)

    def test_add_with_expired(self):
        cache.add("mykey", "value", 0.3)
        assert cache.get("mykey") == "value"

        result = cache.add("mykey", "newvalue", 0.3)
        assert not result
        assert cache.get("mykey") == "value"

        time.sleep(0.4)

        result = cache.add("mykey", "newvalue", 60)
        assert result
        assert cache.get("mykey") == "newvalue"

    @override_cache_settings(options={"COMPRESS_MIN_LENGTH": 10})
    def test_compressed(self):
        cache.set("key", "a" * 11)
        assert cache.get("key") == "a" * 11

    @override_cache_settings(options={"COMPRESS_MIN_LENGTH": 10, "COMPRESS_LEVEL": 9})
    def test_compress_level(self):
        cache.set("key", "a" * 11)
        assert cache.get("key") == "a" * 11

        # Check a bad compression level = zlib error
        with override_cache_settings(
            options={"COMPRESS_MIN_LENGTH": 10, "COMPRESS_LEVEL": 123}
        ):
            with pytest.raises(Exception) as excinfo:
                cache.set("key", "a" * 11)
            assert "Bad compression level" in str(excinfo.value)

    @override_cache_settings(options={"COMPRESS_MIN_LENGTH": 10})
    def test_changing_compressed_option_leaves_compressed_data_readable(self):
        a11 = "a" * 11
        cache.set("key", a11)

        # Turn it off - remains readable and settable
        with override_cache_settings(options={"COMPRESS_MIN_LENGTH": 0}):
            assert cache.get("key") == a11
            cache.set("key", a11)
            assert cache.get("key") == a11

        # Back on, still readable
        assert cache.get("key") == a11
        cache.set("key", a11)
        assert cache.get("key") == a11

    def test_our_options_quacks_like_djangos(self):
        from django.core.cache.backends.db import Options

        from django_mysql.cache import Options as OurOptions

        theirs = Options("something")
        ours = OurOptions("something")
        assert set(ours.__dict__.keys()) == set(theirs.__dict__.keys())

    def test_cull(self):
        self._perform_cull_test(caches["cull"], 50, 30)

    def test_zero_cull(self):
        self._perform_cull_test(caches["zero_cull"], 50, 20)

    def test_no_cull_only_deletes_when_told(self):
        self._perform_cull_test(caches["no_cull"], 50, 50)
        caches["no_cull"].cull()
        assert self.table_count() == 25

    def test_cull_deletes_expired_first(self):
        cull_cache = caches["cull"]
        cull_cache.set("key", "value", 0.3)
        time.sleep(0.4)

        # Add 30 more entries. The expired key should get deleted, leaving the
        # 30 new keys
        self._perform_cull_test(cull_cache, 30, 30)
        assert cull_cache.get("key") is None

    def _perform_cull_test(self, cull_cache, initial_count, final_count):
        # Create initial cache key entries. This will overflow the cache,
        # causing a cull.
        for i in range(1, initial_count + 1):
            cull_cache.set(f"cull{i}", "value", 1000)
        count = 0
        # Count how many keys are left in the cache.
        for i in range(1, initial_count + 1):
            if cull_cache.has_key("cull%d" % i):  # noqa
                count = count + 1
        assert count == final_count

    def test_incr_range(self):
        cache.set("overwhelm", BIGINT_SIGNED_MAX - 1)
        cache.incr("overwhelm")
        with pytest.raises(IntegrityError):
            cache.incr("overwhelm")

    def test_decr_range(self):
        cache.set("underwhelm", BIGINT_SIGNED_MIN + 1)
        cache.decr("underwhelm")
        with pytest.raises(IntegrityError):
            cache.decr("underwhelm")

    def test_cant_incr_decimals(self):
        # Cached values that aren't ints can't be incremented
        cache.set("answer", Decimal("1.1"))
        with pytest.raises(ValueError):
            cache.incr("answer")

    def test_cant_decr_decimals(self):
        # Cached values that aren't ints can't be decremented
        cache.set("answer", Decimal("9.9"))
        with pytest.raises(ValueError):
            cache.decr("answer")

    def test_set_int_subclass(self):
        # Storing an int subclass should return that int subclass
        cache.set("myint", MyInt(2))
        val = cache.get("myint")
        assert val.times2() == 4

        # Can't increment it since it's a pickle object on the table, not an
        # integer
        with pytest.raises(ValueError):
            cache.incr("myint")

    def test_unknown_value_type_errors(self):
        # Unknown value_type values should be errors, since we don't know how
        # to deserialize them. New value_types will probably be introduced by
        # later versions or subclasses of MySQLCache

        cache.set("mykey", 123)
        with connection.cursor() as cursor:
            cursor.execute(f"UPDATE `{self.table_name}` SET value_type = '?'")

        with pytest.raises(ValueError):
            cache.get("mykey")

    def test_key_case_sensitivity(self):
        """
        Check that we can store both upper and lowercase keys separately

        At first MySQLCache did not use a binary collation for cache_key, which
        meant it was not case sensitive.
        """
        cache.set("akey", 123)
        cache.set("Akey", 456)
        assert cache.get("akey") == 123
        assert cache.get("Akey") == 456

    def test_value_type_case_sensitivity(self):
        cache.set("akey", 123)
        with connection.cursor() as cursor:
            # Check that value_type is 'i' for integer
            cursor.execute(f"SELECT value_type FROM `{self.table_name}`")
            t = cursor.fetchone()[0]
            assert t == "i"

            # Should be case-sensitive, so i != I
            cursor.execute(
                f"""SELECT COUNT(*) FROM `{self.table_name}`
                   WHERE value_type = 'I'"""
            )
            n = cursor.fetchone()[0]
            assert n == 0

    def test_bad_key_prefix_for_reverse_function(self):
        with pytest.raises(ValueError) as excinfo:
            MySQLCache(
                table="test cache table",
                params={"KEY_PREFIX": "a:bad:prefix"},
            )
        assert str(excinfo.value).startswith("Cannot use the default KEY_FUNCTION")

    @parameterized.expand(["default", "prefix", "custom_key", "custom_key2"])
    def test_keys_with_prefix(self, cache_name):
        cache = caches[cache_name]
        assert cache.keys_with_prefix("") == set()
        assert cache.keys_with_prefix("K") == set()

        cache.set("A2", True)
        cache.set("K1", True)
        cache.set("K23", True, 1000)
        cache.set("K99", True, 0.1)
        time.sleep(0.2)
        assert cache.keys_with_prefix("") == {"A2", "K1", "K23"}
        assert cache.keys_with_prefix("K") == {"K1", "K23"}

        cache.delete("K1")
        assert cache.keys_with_prefix("K") == {"K23"}

        cache.set("Y3", True, version=1)
        cache.set("Y4", True, version=2)
        assert cache.keys_with_prefix("Y", version=1) == {
            "Y3",
        }
        assert cache.keys_with_prefix("Y", version=2) == {
            "Y4",
        }

        cache.clear()
        assert cache.keys_with_prefix("") == set()
        assert cache.keys_with_prefix("K") == set()

    @parameterized.expand(["default", "prefix", "custom_key", "custom_key2"])
    def test_keys_with_prefix_version(self, cache_name):
        cache = caches[cache_name]

        cache.set("V12", True, version=1)
        cache.set("V12", True, version=2)
        cache.set("V2", True, version=2)
        cache.set("V3", True, version=3)
        assert cache.keys_with_prefix("V", version=1) == {"V12"}
        assert cache.keys_with_prefix("V", version=2) == {"V12", "V2"}
        assert cache.keys_with_prefix("V", version=3) == {"V3"}

    @override_cache_settings(KEY_FUNCTION=custom_key_func)
    def test_keys_with_prefix_with_bad_cache(self):
        with pytest.raises(ValueError) as excinfo:
            cache.keys_with_prefix("")
        assert str(excinfo.value).startswith("To use the _with_prefix commands")

    @parameterized.expand(["default", "prefix", "custom_key", "custom_key2"])
    def test_get_with_prefix(self, cache_name):
        cache = caches[cache_name]
        assert cache.get_with_prefix("") == {}
        assert cache.get_with_prefix("K") == {}

        cache.set("A2", [True])
        cache.set("K1", "Value1")
        cache.set("K23", 2, 1000)
        cache.set("K99", ["Value", 99], 0.1)
        time.sleep(0.2)
        assert cache.get_with_prefix("") == {"A2": [True], "K1": "Value1", "K23": 2}
        assert cache.get_with_prefix("K") == {"K1": "Value1", "K23": 2}

        cache.delete("K1")
        assert cache.get_with_prefix("K") == {"K23": 2}

        cache.clear()
        assert cache.get_with_prefix("") == {}
        assert cache.get_with_prefix("K") == {}

    @parameterized.expand(["default", "prefix", "custom_key", "custom_key2"])
    def test_get_with_prefix_version(self, cache_name):
        cache = caches[cache_name]

        cache.set("V12", ("version1",), version=1)
        cache.set("V12", "str", version=2)
        cache.set("V2", 2, version=2)
        cache.set("V3", object, version=3)
        assert cache.get_with_prefix("V", version=1) == {"V12": ("version1",)}
        assert cache.get_with_prefix("V", version=2) == {"V12": "str", "V2": 2}
        assert cache.get_with_prefix("V", version=3) == {"V3": object}

    @override_cache_settings(KEY_FUNCTION=custom_key_func)
    def test_get_with_prefix_with_bad_cache(self):
        with pytest.raises(ValueError) as excinfo:
            cache.get_with_prefix("")
        assert str(excinfo.value).startswith("To use the _with_prefix commands")

    @parameterized.expand(["default", "prefix", "custom_key", "custom_key2"])
    def test_delete_with_prefix(self, cache_name):
        cache = caches[cache_name]

        # Check it runs on an empty cache
        assert cache.delete_with_prefix("") == 0
        assert cache.delete_with_prefix("K") == 0

        cache.set("A1", True)
        cache.set("A2", True)
        cache.set("K2", True)
        cache.set("K44", True)

        assert cache.keys_with_prefix("") == {"A1", "A2", "K2", "K44"}
        assert cache.delete_with_prefix("A") == 2
        assert cache.keys_with_prefix("") == {"K2", "K44"}
        assert cache.delete_with_prefix("A") == 0
        assert cache.keys_with_prefix("") == {"K2", "K44"}
        assert cache.delete_with_prefix("K") == 2
        assert cache.keys_with_prefix("K") == set()
        assert cache.keys_with_prefix("") == set()

    @parameterized.expand(["default", "prefix", "custom_key", "custom_key2"])
    def test_delete_with_prefix_version(self, cache_name):
        cache = caches[cache_name]

        cache.set("V12", True, version=1)
        cache.set("V12", True, version=2)
        cache.set("V2", True, version=2)
        cache.set("V3", True, version=3)

        has_key = cache.has_key  # avoid lint error

        assert cache.delete_with_prefix("V", version=1) == 1
        assert not has_key("V12", version=1)
        assert has_key("V12", version=2)
        assert has_key("V2", version=2)
        assert has_key("V3", version=3)

        assert cache.delete_with_prefix("V", version=2) == 2
        assert not has_key("V12", version=1)
        assert not has_key("V12", version=2)
        assert not has_key("V2", version=2)
        assert has_key("V3", version=3)

        assert cache.delete_with_prefix("V", version=3) == 1
        assert not has_key("V12", version=1)
        assert not has_key("V12", version=2)
        assert not has_key("V2", version=2)
        assert not has_key("V3", version=3)

    @override_cache_settings(KEY_FUNCTION=custom_key_func)
    def test_delete_with_prefix_with_no_reverse_works(self):
        cache.set_many({"K1": "value", "K2": "value2", "B2": "Anothervalue"})
        assert cache.delete_with_prefix("K") == 2
        assert cache.get_many(["K1", "K2", "B2"]) == {"B2": "Anothervalue"}

    def test_mysql_cache_migration_alias(self):
        out = StringIO()
        call_command("mysql_cache_migration", "default", stdout=out)
        output = out.getvalue()

        num_run_sqls = len(output.split("RunSQL")) - 1
        assert num_run_sqls == 1

    def test_mysql_cache_migration_non_existent(self):
        out = StringIO()
        with pytest.raises(CommandError):
            call_command("mysql_cache_migration", "nonexistent", stdout=out)

    @override_cache_settings(BACKEND="django.core.cache.backends.dummy.DummyCache")
    def test_mysql_cache_migration_no_mysql_caches(self):
        err = StringIO()
        call_command("mysql_cache_migration", stderr=err)
        assert "No MySQLCache instances in CACHES" in err.getvalue()

    # cull_mysql_caches tests

    @override_cache_settings(options={"MAX_ENTRIES": -1})
    def test_cull_max_entries_minus_one(self):
        # cull with MAX_ENTRIES = -1 should never clear anything that is not
        # expired

        # one expired key
        cache.set("key", "value", 0.1)
        time.sleep(0.2)

        # 90 non-expired keys
        for n in range(9):
            cache.set_many({str(n * 10 + i): True for i in range(10)})

        cache.cull()
        assert self.table_count() == 90

    def test_cull_mysql_caches_basic(self):
        cache.set("key", "value", 0.1)
        time.sleep(0.2)
        assert self.table_count() == 1
        call_command("cull_mysql_caches", verbosity=0)
        assert self.table_count() == 0

    def test_cull_mysql_caches_named_cache(self):
        cache.set("key", "value", 0.1)
        time.sleep(0.2)
        assert self.table_count() == 1

        out = StringIO()
        call_command("cull_mysql_caches", "default", verbosity=1, stdout=out)
        output = out.getvalue()
        assert output.strip() == "Deleting from cache 'default'... 1 entries deleted."
        assert self.table_count() == 0

    def test_cull_mysql_caches_bad_cache_name(self):
        with pytest.raises(CommandError) as excinfo:
            call_command("cull_mysql_caches", "NOTACACHE", verbosity=0)
        assert str(excinfo.value) == "Cache 'NOTACACHE' does not exist"


@override_cache_settings()
class MySQLCacheMigrationTests(MySQLCacheTableMixin, TransactionTestCase):
    @pytest.fixture(autouse=True)
    def flake8_path(self, flake8_path):
        self.flake8_path = flake8_path

    def test_mysql_cache_migration(self):
        out = StringIO()
        call_command("mysql_cache_migration", stdout=out)
        output = out.getvalue()

        # Lint it
        (self.flake8_path / "example.py").write_text(output)
        result = self.flake8_path.run_flake8()
        assert result.out_lines == []

        # Dynamic import and check
        migration_mod = types.ModuleType("0001_add_cache_tables")
        exec(output, migration_mod.__dict__)
        assert hasattr(migration_mod, "Migration")
        migration = migration_mod.Migration
        assert hasattr(migration, "dependencies")
        assert hasattr(migration, "operations")

        # Since they all have the same table name, there should only be one
        # operation
        assert len(migration.operations) == 1

        # Now run the migration forwards and backwards to check it works
        operation = migration.operations[0]
        assert not self.table_exists(self.table_name)

        state = ProjectState()
        new_state = state.clone()
        with connection.schema_editor() as editor:
            operation.database_forwards("testapp", editor, state, new_state)
        assert self.table_exists(self.table_name)

        new_state = state.clone()
        with connection.schema_editor() as editor:
            operation.database_backwards("testapp", editor, new_state, state)
        assert not self.table_exists(self.table_name)

    def table_exists(self, table_name):
        with connection.cursor() as cursor:
            cursor.execute(
                """SELECT COUNT(*) FROM INFORMATION_SCHEMA.TABLES
                   WHERE TABLE_SCHEMA = DATABASE() AND
                         TABLE_NAME = %s""",
                (table_name,),
            )
            return bool(cursor.fetchone()[0])
