Hopiu/django-cachalot
1
0
Fork 0
mirror of https://github.com/Hopiu/django-cachalot.git synced 2026-05-08 12:44:44 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Adds a test for the management command.

Browse source
This commit is contained in:
Bertrand Bordage 2015-02-22 00:51:47 +01:00
parent 0c8180a6e7
commit b7dbd00d02
5 changed files with 102 additions and 9 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
cachalot/management/commands/invalidate_cachalot.py
View file

@ -21,7 +21,7 @@ class Command(BaseCommand):
def handle(self, *args, **options):
cache_alias = options['cache_alias']
db_alias = options['db_alias']
verbosity = options['verbosity']
verbosity = int(options['verbosity'])
models = []
for arg in args:
@ -38,7 +38,7 @@ class Command(BaseCommand):
db_str = '' if db_alias is None else "for database '%s'" % db_alias
keys_str = 'keys for %s models' % len(models) if args else 'all keys'
if verbosity != '0':
if verbosity > 0:
self.stdout.write(' '.join(filter(bool, ['Invalidating', keys_str,
cache_str, db_str]))
+ '...')
@ -48,5 +48,5 @@ class Command(BaseCommand):
cache_alias=cache_alias, db_alias=db_alias)
else:
invalidate_all(cache_alias=cache_alias, db_alias=db_alias)
if verbosity != '0':
if verbosity > 0:
self.stdout.write('Cache keys successfully invalidated.')

2
cachalot/tests/__init__.py
View file

@ -4,4 +4,4 @@ from .transaction import AtomicTestCase
from .thread_safety import ThreadSafetyTestCase
from .multi_db import MultiDatabaseTestCase
from .settings import SettingsTestCase
from .api import APITestCase
from .api import APITestCase, CommandTestCase

95
cachalot/tests/api.py
View file

@ -1,8 +1,16 @@
# coding: utf-8
from __future__ import unicode_literals
try:
from unittest import skipIf
except ImportError: # For Python 2.6
from unittest2 import skipIf
from django.db import connection, transaction
from django.conf import settings
from django.contrib.auth.models import User
from django.core.cache import DEFAULT_CACHE_ALIAS
from django.core.management import call_command
from django.db import connection, transaction, DEFAULT_DB_ALIAS
from django.test import TransactionTestCase
from ..api import *
@ -82,3 +90,88 @@ class APITestCase(TransactionTestCase):
with self.assertNumQueries(1):
Test.objects.get()
class CommandTestCase(TransactionTestCase):
multi_db = True
def setUp(self):
self.db_alias2 = (alias for alias in settings.DATABASES
if alias != DEFAULT_DB_ALIAS).next()
self.cache_alias2 = (alias for alias in settings.CACHES
if alias != DEFAULT_CACHE_ALIAS).next()
self.t1 = Test.objects.create(name='test1')
self.t2 = Test.objects.using(self.db_alias2).create(name='test2')
self.u = User.objects.create_user('test')
def test_invalidate_cachalot(self):
with self.assertNumQueries(1):
self.assertListEqual(list(Test.objects.all()), [self.t1])
call_command('invalidate_cachalot', verbosity=0)
with self.assertNumQueries(1):
self.assertListEqual(list(Test.objects.all()), [self.t1])
call_command('invalidate_cachalot', 'auth', verbosity=0)
with self.assertNumQueries(0):
self.assertListEqual(list(Test.objects.all()), [self.t1])
call_command('invalidate_cachalot', 'cachalot', verbosity=0)
with self.assertNumQueries(1):
self.assertListEqual(list(Test.objects.all()), [self.t1])
call_command('invalidate_cachalot', 'cachalot.testchild', verbosity=0)
with self.assertNumQueries(0):
self.assertListEqual(list(Test.objects.all()), [self.t1])
call_command('invalidate_cachalot', 'cachalot.test', verbosity=0)
with self.assertNumQueries(1):
self.assertListEqual(list(Test.objects.all()), [self.t1])
with self.assertNumQueries(1):
self.assertListEqual(list(User.objects.all()), [self.u])
call_command('invalidate_cachalot', 'cachalot.test', 'auth.user',
verbosity=0)
with self.assertNumQueries(1):
self.assertListEqual(list(Test.objects.all()), [self.t1])
with self.assertNumQueries(1):
self.assertListEqual(list(User.objects.all()), [self.u])
@skipIf(len(settings.DATABASES) == 1,
'We can’t change the DB used since there’s only one configured')
def test_invalidate_cachalot_multi_db(self):
with self.assertNumQueries(1):
self.assertListEqual(list(Test.objects.all()), [self.t1])
call_command('invalidate_cachalot', verbosity=0,
db_alias=self.db_alias2)
with self.assertNumQueries(0):
self.assertListEqual(list(Test.objects.all()), [self.t1])
with self.assertNumQueries(1, using=self.db_alias2):
self.assertListEqual(list(Test.objects.using(self.db_alias2)),
[self.t2])
call_command('invalidate_cachalot', verbosity=0,
db_alias=self.db_alias2)
with self.assertNumQueries(1, using=self.db_alias2):
self.assertListEqual(list(Test.objects.using(self.db_alias2)),
[self.t2])
@skipIf(len(settings.CACHES) == 1,
'We can’t change the cache used since there’s only one configured')
def test_invalidate_cachalot_multi_cache(self):
with self.assertNumQueries(1):
self.assertListEqual(list(Test.objects.all()), [self.t1])
call_command('invalidate_cachalot', verbosity=0,
cache_alias=self.cache_alias2)
with self.assertNumQueries(0):
self.assertListEqual(list(Test.objects.all()), [self.t1])
with self.assertNumQueries(1):
with self.settings(CACHALOT_CACHE=self.cache_alias2):
self.assertListEqual(list(Test.objects.all()), [self.t1])
call_command('invalidate_cachalot', verbosity=0,
cache_alias=self.cache_alias2)
with self.assertNumQueries(1):
with self.settings(CACHALOT_CACHE=self.cache_alias2):
self.assertListEqual(list(Test.objects.all()), [self.t1])

4
cachalot/tests/multi_db.py
View file

@ -21,8 +21,8 @@ class MultiDatabaseTestCase(TransactionTestCase):
def setUp(self):
self.t1 = Test.objects.create(name='test1')
self.t2 = Test.objects.create(name='test2')
self.db_alias2 = [alias for alias in settings.DATABASES
if alias != DEFAULT_DB_ALIAS][0]
self.db_alias2 = (alias for alias in settings.DATABASES
if alias != DEFAULT_DB_ALIAS).next()
self.is_sqlite2 = connections[self.db_alias2].vendor == 'sqlite'
def test_read(self):

4
cachalot/tests/settings.py
View file

@ -69,8 +69,8 @@ class SettingsTestCase(TransactionTestCase):
with self.assertNumQueries(0):
list(Test.objects.all())
other_cache_alias = [alias for alias in settings.CACHES
if alias != DEFAULT_CACHE_ALIAS][0]
other_cache_alias = (alias for alias in settings.CACHES
if alias != DEFAULT_CACHE_ALIAS).next()
with self.settings(CACHALOT_CACHE=other_cache_alias):
with self.assertNumQueries(1):