gallery-dl/test/test_cache.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

# Copyright 2020 Mike Fährmann
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 2 as
# published by the Free Software Foundation.

import unittest
import tempfile
import time

from gallery_dl import config, util
dbpath = tempfile.mkstemp()[1]
config.set(("cache",), "file", dbpath)
from gallery_dl import cache  # noqa


def tearDownModule():
    util.remove_file(dbpath)


class TestCache(unittest.TestCase):

    def test_decorator(self):

        @cache.memcache()
        def mc1():
            pass

        @cache.memcache(maxage=10)
        def mc2():
            pass

        @cache.cache()
        def dbc():
            pass

        self.assertIsInstance(mc1, cache.CacheDecorator)
        self.assertIsInstance(mc2, cache.MemoryCacheDecorator)
        self.assertIsInstance(dbc, cache.DatabaseCacheDecorator)

    def test_keyarg_mem_simple(self):
        @cache.memcache(keyarg=2)
        def ka(a, b, c):
            return a+b+c

        self.assertEqual(ka(1, 1, 1), 3)
        self.assertEqual(ka(2, 2, 2), 6)

        self.assertEqual(ka(0, 0, 1), 3)
        self.assertEqual(ka(9, 9, 1), 3)
        self.assertEqual(ka(0, 0, 2), 6)
        self.assertEqual(ka(9, 9, 2), 6)

    def test_keyarg_mem(self):
        @cache.memcache(keyarg=2, maxage=10)
        def ka(a, b, c):
            return a+b+c

        self.assertEqual(ka(1, 1, 1), 3)
        self.assertEqual(ka(2, 2, 2), 6)

        self.assertEqual(ka(0, 0, 1), 3)
        self.assertEqual(ka(9, 9, 1), 3)
        self.assertEqual(ka(0, 0, 2), 6)
        self.assertEqual(ka(9, 9, 2), 6)

    def test_keyarg_db(self):
        @cache.cache(keyarg=2, maxage=10)
        def ka(a, b, c):
            return a+b+c

        self.assertEqual(ka(1, 1, 1), 3)
        self.assertEqual(ka(2, 2, 2), 6)

        self.assertEqual(ka(0, 0, 1), 3)
        self.assertEqual(ka(9, 9, 1), 3)
        self.assertEqual(ka(0, 0, 2), 6)
        self.assertEqual(ka(9, 9, 2), 6)

    def test_expires_mem(self):
        @cache.memcache(maxage=1)
        def ex(a, b, c):
            return a+b+c

        self.assertEqual(ex(1, 1, 1), 3)
        self.assertEqual(ex(2, 2, 2), 3)
        self.assertEqual(ex(3, 3, 3), 3)

        time.sleep(2)
        self.assertEqual(ex(3, 3, 3), 9)
        self.assertEqual(ex(2, 2, 2), 9)
        self.assertEqual(ex(1, 1, 1), 9)

    def test_expires_db(self):
        @cache.cache(maxage=1)
        def ex(a, b, c):
            return a+b+c

        self.assertEqual(ex(1, 1, 1), 3)
        self.assertEqual(ex(2, 2, 2), 3)
        self.assertEqual(ex(3, 3, 3), 3)

        time.sleep(2)
        self.assertEqual(ex(3, 3, 3), 9)
        self.assertEqual(ex(2, 2, 2), 9)
        self.assertEqual(ex(1, 1, 1), 9)

    def test_update_mem_simple(self):
        @cache.memcache(keyarg=0)
        def up(a, b, c):
            return a+b+c

        self.assertEqual(up(1, 1, 1), 3)
        up.update(1, 0)
        up.update(2, 9)
        self.assertEqual(up(1, 0, 0), 0)
        self.assertEqual(up(2, 0, 0), 9)

    def test_update_mem(self):
        @cache.memcache(keyarg=0, maxage=10)
        def up(a, b, c):
            return a+b+c

        self.assertEqual(up(1, 1, 1), 3)
        up.update(1, 0)
        up.update(2, 9)
        self.assertEqual(up(1, 0, 0), 0)
        self.assertEqual(up(2, 0, 0), 9)

    def test_update_db(self):
        @cache.cache(keyarg=0, maxage=10)
        def up(a, b, c):
            return a+b+c

        self.assertEqual(up(1, 1, 1), 3)
        up.update(1, 0)
        up.update(2, 9)
        self.assertEqual(up(1, 0, 0), 0)
        self.assertEqual(up(2, 0, 0), 9)

    def test_invalidate_mem_simple(self):
        @cache.memcache(keyarg=0)
        def inv(a, b, c):
            return a+b+c

        self.assertEqual(inv(1, 1, 1), 3)
        inv.invalidate(1)
        inv.invalidate(2)
        self.assertEqual(inv(1, 0, 0), 1)
        self.assertEqual(inv(2, 0, 0), 2)

    def test_invalidate_mem(self):
        @cache.memcache(keyarg=0, maxage=10)
        def inv(a, b, c):
            return a+b+c

        self.assertEqual(inv(1, 1, 1), 3)
        inv.invalidate(1)
        inv.invalidate(2)
        self.assertEqual(inv(1, 0, 0), 1)
        self.assertEqual(inv(2, 0, 0), 2)

    def test_invalidate_db(self):
        @cache.cache(keyarg=0, maxage=10)
        def inv(a, b, c):
            return a+b+c

        self.assertEqual(inv(1, 1, 1), 3)
        inv.invalidate(1)
        inv.invalidate(2)
        self.assertEqual(inv(1, 0, 0), 1)
        self.assertEqual(inv(2, 0, 0), 2)

    def test_database_read(self):
        @cache.cache(keyarg=0, maxage=10)
        def db(a, b, c):
            return a+b+c

        # initialize cache
        self.assertEqual(db(1, 1, 1), 3)
        db.update(2, 6)

        # check and clear the in-memory portion of said cache
        self.assertEqual(db.cache[1][0], 3)
        self.assertEqual(db.cache[2][0], 6)
        db.cache.clear()
        self.assertEqual(db.cache, {})

        # fetch results from database
        self.assertEqual(db(1, 0, 0), 3)
        self.assertEqual(db(2, 0, 0), 6)

        # check in-memory cache updates
        self.assertEqual(db.cache[1][0], 3)
        self.assertEqual(db.cache[2][0], 6)


if __name__ == '__main__':
    unittest.main()