Merge lp:~elachuni/ubuntu-webcatalog/cleanup into lp:ubuntu-webcatalog

[Anthony Lenton]

Added a command to clean up nonces and sessions.

This branch adds a command to clean up the oauth nonces and sessions table.

It's based largely on the script used by SSO to do the same thing, that has proved useful to keep up and cope with large amounts of data that need to be deleted from a high(ish) traffic db.

The replication bits have been stripped out (as webcatalog isn't deployed with slony), and it's been converted into a Django management command, so that it's simpler to run from the appservers, on a regular basis from a cronjob.

[Michael Nelson (michael.nelson)]

On Tue, Apr 17, 2012 at 6:54 AM, Anthony Lenton
<email address hidden> wrote:
> Anthony Lenton has proposed merging lp:~elachuni/ubuntu-webcatalog/cleanup into lp:ubuntu-webcatalog.
>
> Requested reviews:
>  Canonical Consumer Applications Hackers (canonical-ca-hackers)
>
> For more details, see:
> https://code.launchpad.net/~elachuni/ubuntu-webcatalog/cleanup/+merge/102237
>
> This branch adds a command to clean up the oauth nonces and sessions table.
>
> It's based largely on the script used by SSO to do the same thing, that has proved useful to keep up and cope with large amounts of data that need to be deleted from a high(ish) traffic db.

Cool - a few comments below about different assumptions (or things I
haven't understood).

One broader question though, given that, by definition, a nonce should
only be used once, shouldn't we just be deleting them as they are
used... why are they kept around? (I mean, there will be a small
number that are not used which would need to be cleaned, but I don't
understand why we're keeping all the used ones).

>
> The replication bits have been stripped out (as webcatalog isn't deployed with slony), and it's been converted into a Django management command, so that it's simpler to run from the appservers, on a regular basis from a cronjob.
> --
> https://code.launchpad.net/~elachuni/ubuntu-webcatalog/cleanup/+merge/102237
> You are subscribed to branch lp:ubuntu-webcatalog.
>
> === modified file 'setup.py'
> === added file 'src/webcatalog/management/commands/cleanup.py'
> --- src/webcatalog/management/commands/cleanup.py       1970-01-01 00:00:00 +0000
> +++ src/webcatalog/management/commands/cleanup.py       2012-04-17 04:53:18 +0000
> @@ -0,0 +1,165 @@
> +#   -*- coding: utf-8 -*-
> +#   This file is part of the Apps Directory
> +#   Copyright (C) 2011 Canonical Ltd.
> +#
> +#   This program is free software: you can redistribute it and/or modify
> +#   it under the terms of the GNU Affero General Public License as
> +#   published by the Free Software Foundation, either version 3 of the
> +#   License, or (at your option) any later version.
> +#
> +#   This program is distributed in the hope that it will be useful,
> +#   but WITHOUT ANY WARRANTY; without even the implied warranty of
> +#   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
> +#   GNU Affero General Public License for more details.
> +#
> +#   You should have received a copy of the GNU Affero General Public License
> +#   along with this program.  If not, see <http://www.gnu.org/licenses/>.
> +
> +"""Management command to clean up sessions and OAuth nonce tables."""
> +
> +import time
> +from django.core.management.base import (
> +    CommandError,
> +    LabelCommand,
> +    make_option,
> +)
> +from django.db import connection
> +
> +try:
> +    import psycopg2
> +    from psycopg2.extensions import ISOLATION_LEVEL_AUTOCOMMIT
> +    psycopg2_available = True
> +except ImportError:
> +    psycopg2_available = False

I don't really get why this is done - why not just import it and the
command will error if it's not available. As it is, this var seems to
be used just to raise a different error below, in which case we ...

[Anthony Lenton (elachuni)]

> One broader question though, given that, by definition, a nonce should
> only be used once, shouldn't we just be deleting them as they are
> used... why are they kept around? (I mean, there will be a small
> number that are not used which would need to be cleaned, but I don't
> understand why we're keeping all the used ones).

Nonces are generated by the client, and they're stored on the server to avoid replay attacks. The server's plan is to simply refuse any request that includes a nonce it has already seen. As this would mean that the server needs to store all nonces indefinitely, the server *also* sets an interval after which a request with an old timestamp is rejected. So, stored nonces that are older than this interval (5 hours currently) can be deleted.

> > +try:
> > +    import psycopg2
> > +    from psycopg2.extensions import ISOLATION_LEVEL_AUTOCOMMIT
> > +    psycopg2_available = True
> > +except ImportError:
> > +    psycopg2_available = False
>
> I don't really get why this is done - why not just import it and the
> command will error if it's not available. As it is, this var seems to
> be used just to raise a different error below, in which case we could
> do:
>
> try:
>    import psycopg2
>    from psycopg2.extensions import ISOLATION_LEVEL_AUTOCOMMIT
> except ImportError:
> raise CommandError('This command requires psycopg2')
>
> or maybe it's just left over from the SSO version which used the
> variable differently?
>

Right, we could do that import locally within the command's handle method and avoid the variable; I just preferred to keep the import global. The issue with raising an exception outside of the handle() method is that it won't be caught and rendered as a pretty warning message.

> > +class OAuthNonceCleaner(object):
> > +    @staticmethod
> > +    def declare_expired_keys_cursor(cur):
> > +        print "Opening cursor"
>
> Left over from debug? Or should it be a log?

Commands print their output directly, instead of logging. Checking other commands like import_app_install_data, it uses a neat output() method to redirect stdout for testing and only print if the verbosity level is set high enough. I could switch to use something like this?

> > +class CleanupTestCase(TestCaseWithFactory):
> > +    @patch('webcatalog.management.commands.cleanup.connection')
> > +    @patch('sys.stdout')
>
> Ah - if you do switch to logging instead of print, you can use the
> LogLevelTestCase (or whatever the helper is called).

Right, or set the instance's stdout before calling?

> > +    def run_and_check_output_and_sql(self, func, output, sql, mock_stdout,
> > +        mock_connection):
> > +        func()
> > +        actual_output = ''.join(str(call[0][0])
> > +            for call in mock_stdout.write.call_args_list)
> > +        self.assertEqual(output, actual_output.split('\n'))
> > +        call_list =
> mock_connection.cursor.return_value.execute.call_args_list
> > +        for expected, call in zip(sql, call_list):
> > +            actual = (' '.join(call[0][0].split()),) + call[0][1:]
> > +            self.assertEqual(expected, actual)
>
> What's the reason for testiing with a mock here? Why not pop...

[Michael Nelson (michael.nelson)]

Approve

On Tue, Apr 17, 2012 at 2:37 PM, Anthony Lenton
<email address hidden> wrote:
>> One broader question though, given that, by definition, a nonce should
>> only be used once, shouldn't we just be deleting them as they are
>> used... why are they kept around? (I mean, there will be a small
>> number that are not used which would need to be cleaned, but I don't
>> understand why we're keeping all the used ones).
>
> Nonces are generated by the client, and they're stored on the server to avoid replay attacks.  The server's plan is to simply refuse any request that includes a nonce it has already seen.  As this would mean that the server needs to store all nonces indefinitely, the server *also* sets an interval after which a request with an old timestamp is rejected.  So, stored nonces that are older than this interval (5 hours currently) can be deleted.
>

Ah - sorry - I was thinking of something we created and then let
clients use once, of course. *facepalm*.

>> > +try:
>> > +    import psycopg2
>> > +    from psycopg2.extensions import ISOLATION_LEVEL_AUTOCOMMIT
>> > +    psycopg2_available = True
>> > +except ImportError:
>> > +    psycopg2_available = False
>>
>> I don't really get why this is done - why not just import it and the
>> command will error if it's not available. As it is, this var seems to
>> be used just to raise a different error below, in which case we could
>> do:
>>
>> try:
>>     import psycopg2
>>     from psycopg2.extensions import ISOLATION_LEVEL_AUTOCOMMIT
>> except ImportError:
>>     raise CommandError('This command requires psycopg2')
>>
>> or maybe it's just left over from the SSO version which used the
>> variable differently?
>>
>
> Right, we could do that import locally within the command's handle method and avoid the variable; I just preferred to keep the import global.  The issue with raising an exception outside of the handle() method is that it won't be caught and rendered as a pretty warning message.
>

Ah - nice. I'd not read about:

https://docs.djangoproject.com/en/1.3/howto/custom-management-commands/#CommandError

Great.

>> > +class OAuthNonceCleaner(object):
>> > +    @staticmethod
>> > +    def declare_expired_keys_cursor(cur):
>> > +        print "Opening cursor"
>>
>> Left over from debug? Or should it be a log?
>
> Commands print their output directly, instead of logging.  Checking other commands like import_app_install_data, it uses a neat output() method to redirect stdout for testing and only print if the verbosity level is set high enough. I could switch to use something like this?
>

Probably not worth it - I just wasn't sure whether it was intentional.

>> > +class CleanupTestCase(TestCaseWithFactory):
>> > +    @patch('webcatalog.management.commands.cleanup.connection')
>> > +    @patch('sys.stdout')
>>
>> Ah - if you do switch to logging instead of print, you can use the
>> LogLevelTestCase (or whatever the helper is called).
>
> Right, or set the instance's stdout before calling?
>
>> > +    def run_and_check_output_and_sql(self, func, output, sql, mock_stdout,
>> > +        mock_connection):
>> > +        func()
>> > +        actual_output = ''.join(str(call[0][0])
>> >...

[lp:~elachuni/ubuntu-webcatalog/cleanup]

101. By Anthony Lenton on 2012-04-17

Added tests for cleanup commands with data.

=== modified file 'setup.py'
--- setup.py	2012-03-01 14:53:20 +0000
+++ setup.py	2012-04-17 17:00:26 +0000
@@ -35,7 +35,7 @@
     install_requires = [
         'django',
         'setuptools',
-        'south',
+        'south==0.7.3',
         'configglue==0.10',
         'django-configglue==0.4',
         'django-openid-auth==0.2',

=== modified file 'src/webcatalog/fixtures/initial_data.json'
--- src/webcatalog/fixtures/initial_data.json	2012-03-09 15:11:21 +0000
+++ src/webcatalog/fixtures/initial_data.json	2012-04-17 17:00:26 +0000
@@ -90,6 +90,15 @@
         }
     }, 
     {
+        "pk": 13, 
+        "model": "webcatalog.department", 
+        "fields": {
+            "name": "Graphics", 
+            "parent": null, 
+            "slug": "graphics"
+        }
+    }, 
+    {
         "pk": 12, 
         "model": "webcatalog.department", 
         "fields": {
@@ -99,15 +108,6 @@
         }
     }, 
     {
-        "pk": 13, 
-        "model": "webcatalog.department", 
-        "fields": {
-            "name": "Graphics", 
-            "parent": null, 
-            "slug": "graphics"
-        }
-    }, 
-    {
         "pk": 14, 
         "model": "webcatalog.department", 
         "fields": {

=== added file 'src/webcatalog/management/commands/cleanup.py'
--- src/webcatalog/management/commands/cleanup.py	1970-01-01 00:00:00 +0000
+++ src/webcatalog/management/commands/cleanup.py	2012-04-17 17:00:26 +0000
@@ -0,0 +1,163 @@
+#   -*- coding: utf-8 -*-
+#   This file is part of the Apps Directory
+#   Copyright (C) 2011 Canonical Ltd.
+#
+#   This program is free software: you can redistribute it and/or modify
+#   it under the terms of the GNU Affero General Public License as
+#   published by the Free Software Foundation, either version 3 of the
+#   License, or (at your option) any later version.
+#
+#   This program is distributed in the hope that it will be useful,
+#   but WITHOUT ANY WARRANTY; without even the implied warranty of
+#   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+#   GNU Affero General Public License for more details.
+#
+#   You should have received a copy of the GNU Affero General Public License
+#   along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+"""Management command to clean up sessions and OAuth nonce tables."""
+
+import time
+from django.core.management.base import (
+    CommandError,
+    LabelCommand,
+    make_option,
+)
+from django.db import connection
+
+try:
+    import psycopg2
+    from psycopg2.extensions import ISOLATION_LEVEL_AUTOCOMMIT
+    psycopg2_available = True
+except ImportError:
+    psycopg2_available = False
+
+
+class DjangoSessionCleaner(object):
+    @staticmethod
+    def declare_expired_keys_cursor(cur):
+        print "Opening cursor"
+        # We order by expire_date to force the index to be used.
+        cur.execute("CLOSE ALL")
+        cur.execute("""
+            DECLARE _django_session_clean NO SCROLL CURSOR WITH HOLD FOR
+            SELECT session_key FROM django_session
+            WHERE expire_date < CURRENT_TIMESTAMP
+            ORDER BY expire_date
+            """)
+
+    @staticmethod
+    def remove_batch(cur, batch_size):
+        cur.execute("FETCH %s FROM _django_session_clean", [batch_size])
+        session_keys = [session_key for session_key, in cur.fetchall()]
+        if not session_keys:
+            return 0
+        cur.execute("""
+            DELETE FROM django_session
+            WHERE
+                expire_date < CURRENT_TIMESTAMP
+                AND session_key = ANY(%s)
+            """, (session_keys,))
+        # len() is better here than querying the cursor for rows actually
+        # removed, as we don't want to terminate the script early in the
+        # unlikely case all the sessions in out batch were reused between
+        # now and when the expired keys cursor was opened.
+        return len(session_keys)
+
+
+class OAuthNonceCleaner(object):
+    @staticmethod
+    def declare_expired_keys_cursor(cur):
+        print "Opening cursor"
+        # We order by created_at to force the index to be used.
+        # Use a 5 hour clock skew as defined in openid.store.nonce
+        cur.execute("CLOSE ALL")
+        cur.execute("""
+            DECLARE _oauth_nonce_clean NO SCROLL CURSOR WITH HOLD FOR
+            SELECT id FROM webcatalog_nonce
+            WHERE created_at < (CURRENT_TIMESTAMP - INTERVAL '5' HOUR)
+            ORDER BY created_at
+            """)
+
+    @staticmethod
+    def remove_batch(cur, batch_size):
+        cur.execute("FETCH %s FROM _oauth_nonce_clean", [batch_size])
+        nonce_ids = [nonce_id for nonce_id, in cur.fetchall()]
+        if not nonce_ids:
+            return 0
+        cur.execute("""
+            DELETE FROM webcatalog_nonce
+            WHERE id = ANY(%s)
+            """, (nonce_ids,))
+        # len() is better here than querying the cursor for rows actually
+        # removed, as we don't want to terminate the script early in the
+        # unlikely case all the sessions in out batch were reused between
+        # now and when the expired keys cursor was opened.
+        return len(nonce_ids)
+
+
+CLEANERS = {
+    'django_session': DjangoSessionCleaner,
+    'webcatalog_nonce': OAuthNonceCleaner,
+}
+
+
+class Command(LabelCommand):
+    help = "Cleans the Django session and OAuth nonce tables."
+
+    option_list = LabelCommand.option_list + (
+        make_option('--cursor-life', default=3600, dest="cursor_secs",
+            action='store',
+            help="Hold a cursor open a maximum of SECS seconds"),
+        make_option('--batch-time', default=5, dest="batch_secs",
+            action="store",
+            help="Try to keep batch removal transaction time to SECS seconds"),
+        )
+
+    def handle_label(self, table, **options):
+        if not psycopg2_available:
+            raise CommandError('This command requires psycopg2')
+
+        cursor_secs = int(options.get('cursor_secs', 3600))
+        target_batch_time = int(options.get('batch_secs', 5))
+
+        cleaner_cls = CLEANERS.get(table, None)
+        if cleaner_cls is None:
+            msg = ("Invalid cleaner.\nNo cleaner found for table: %s\n"
+                "Supported tables are: %s" % (table, CLEANERS.keys()))
+            raise CommandError(msg)
+        cleaner = cleaner_cls()
+        connection.isolation_level = ISOLATION_LEVEL_AUTOCOMMIT
+        cur = connection.cursor()
+
+        removed = -1
+        total_removed = 0
+        batch_size = 1
+        cursor_opened_time = 0
+
+        while True:
+            # We don't want to hold the cursor open too long to allow
+            # vacuum to reclaim rows. 1 hour default before it is reopened.
+            if time.time() - cursor_opened_time > cursor_secs:
+                cleaner.declare_expired_keys_cursor(cur)
+                cursor_opened_time = time.time()
+
+            # Remove a batch of session rows.
+            batch_start = time.time()
+            removed = cleaner.remove_batch(cur, batch_size)
+            actual_batch_time = time.time() - batch_start
+            total_removed += removed
+            print "Removed %d rows (%d total removed). Batch size %d" % (
+                removed, total_removed, batch_size)
+
+            # Done
+            if removed == 0:
+                print "All done."
+                break
+
+            # Increase or decrease the batch size by 10%, minimum 1.
+            batch_size_wobble = int(batch_size * 0.1 + 1)
+            if actual_batch_time > target_batch_time * 1.1:
+                batch_size -= batch_size_wobble
+            elif actual_batch_time < target_batch_time * 0.9:
+                batch_size += batch_size_wobble

=== modified file 'src/webcatalog/tests/factory.py'
--- src/webcatalog/tests/factory.py	2012-04-13 20:45:49 +0000
+++ src/webcatalog/tests/factory.py	2012-04-17 17:00:26 +0000
@@ -22,9 +22,13 @@
     with_statement,
     )
 import os
+from datetime import (
+    datetime,
+    timedelta,
+)
 from itertools import count
-
 from django.contrib.auth.models import User
+from django.contrib.sessions.models import Session
 from django.test import TestCase
 from django_openid_auth.models import UserOpenID
 
@@ -36,6 +40,7 @@
     DistroSeries,
     Exhibit,
     Machine,
+    Nonce,
     Token,
     )
 from webcatalog.utilities import full_claimed_id
@@ -194,6 +199,27 @@
             token.save()
         return token, consumer
 
+    def make_nonce(self, token=None, consumer=None, created_at=None):
+        if token is None or consumer is None:
+            assert token is None and consumer is None
+            token, consumer = self.make_oauth_token_and_consumer()
+        nonce = Nonce.objects.create(token=token, consumer=consumer,
+            nonce=self.get_unique_string(prefix='nonce-'))
+        if created_at:
+            nonce.created_at = created_at
+            nonce.save()
+        return nonce
+
+    def make_session(self, expire_date=None):
+        if expire_date is None:
+            expire_date = datetime.now() + timedelta(
+                seconds=self.get_unique_integer())
+        return Session.objects.create(
+            session_key=self.get_unique_string(prefix='key-'),
+            session_data=self.get_unique_string(prefix='session-data-'),
+            expire_date=expire_date
+        )
+
 
 class TestCaseWithFactory(TestCase):
 

=== modified file 'src/webcatalog/tests/test_commands.py'
--- src/webcatalog/tests/test_commands.py	2012-04-13 20:45:49 +0000
+++ src/webcatalog/tests/test_commands.py	2012-04-17 17:00:26 +0000
@@ -33,7 +33,12 @@
 from decimal import Decimal
 
 from django.conf import settings
+from django.contrib.sessions.models import Session
 from django.core.management import call_command
+from django.db import connection
+from django.utils.unittest import skipUnless
+from django.test.testcases import TransactionTestCase
+
 from mock import (
     patch,
     MagicMock,
@@ -45,23 +50,27 @@
     Application,
     DistroSeries,
     Exhibit,
+    Nonce,
     ReviewStatsImport,
     )
-from webcatalog.management.commands.import_app_install_data import (
-    Command as ImportAppInstallCommand,
-    )
-from webcatalog.management.commands.import_ratings_stats import (
-    Command as ImportRatingsStatsCommand,
-    )
-from webcatalog.tests.factory import TestCaseWithFactory
+from webcatalog.management.commands import (
+    import_app_install_data,
+    import_ratings_stats,
+)
+from webcatalog.tests.factory import (
+    TestCaseWithFactory,
+    WebCatalogObjectFactory,
+)
 
 __metaclass__ = type
 __all__ = [
+    'CheckAllLatestTestCase',
+    'CleanupTestCase',
+    'CleanupDataTestCase',
     'ImportAppInstallTestCase',
     'ImportExhibitsTestCase',
     'ImportForPurchaseAppsTestCase',
     'ImportRatingsTestCase',
-    'CheckAllLatestTestCase',
     ]
 
 
@@ -188,7 +197,7 @@
         with patch(get_uri_fn) as mock_get_uri:
             mock_get_uri.return_value = 'http://example.com/my.deb'
             with patch('urllib.urlretrieve') as mock_urlretrieve:
-                ImportAppInstallCommand().get_package_data_for_series(
+                import_app_install_data.Command().get_package_data_for_series(
                     'app-install-data', 'natty', tmp_dir)
         shutil.rmtree(tmp_dir)
 
@@ -305,7 +314,7 @@
         self.assertEqual(scribus.screenshots, ["http://example.com/1.png"])
 
     def get_package_uri_for_series(self):
-        uri = ImportAppInstallCommand().get_package_uri_for_series(
+        uri = import_app_install_data.Command().get_package_uri_for_series(
             'app-install-data', 'natty')
 
         self.assertEqual('http://example.com/app-install-1.01.deb', uri)
@@ -397,7 +406,7 @@
         self.assertEqual(0, DistroSeries.objects.filter(
             code_name='natty').count())
 
-        ImportAppInstallCommand().update_from_cache('natty')
+        import_app_install_data.Command().update_from_cache('natty')
 
         self.assertEqual(1, DistroSeries.objects.filter(
             code_name='natty').count())
@@ -405,7 +414,7 @@
     def test_apt_cache_apps_used_also(self):
         self.assertEqual(0, Application.objects.count())
 
-        ImportAppInstallCommand().update_from_cache('natty')
+        import_app_install_data.Command().update_from_cache('natty')
 
         # There are 4 packages in our mock_cache which is patched in the
         # setUp.
@@ -531,7 +540,7 @@
         candidate = Mock()
         candidate.uri = 'http://extras.ubuntu.com/pool/f/foobar.deb'
         candidate.record = {'Icon': 'foo.png'}
-        command = ImportAppInstallCommand()
+        command = import_app_install_data.Command()
 
         command.fetch_icon_from_extras(app, candidate)
 
@@ -546,7 +555,7 @@
         candidate = Mock()
         candidate.uri = 'http://extras.ubuntu.com/pool/f/foobar.deb'
         candidate.record = {}
-        command = ImportAppInstallCommand()
+        command = import_app_install_data.Command()
 
         command.fetch_icon_from_extras(app, candidate)
 
@@ -559,7 +568,7 @@
         candidate = Mock()
         candidate.uri = 'http://archive.ubuntu.com/pool/f/foobar.deb'
         candidate.record = {'Icon': 'foobar.png'}
-        command = ImportAppInstallCommand()
+        command = import_app_install_data.Command()
 
         command.fetch_icon_from_extras(app, candidate)
 
@@ -655,7 +664,7 @@
             package_name=self.factory.get_unique_string(prefix='package-'),
             ratings_average='5.00', ratings_total=1,
             histogram='[0, 0, 0, 0, 1]')) for x in range(3000)]
-        command = ImportRatingsStatsCommand()
+        command = import_ratings_stats.Command()
 
         # update_apps_with_stats returns None on success:
         self.assertIsNone(command.update_apps_with_stats(natty, stats))
@@ -665,7 +674,7 @@
         app = self.factory.make_application()
         stats = [ReviewsStats.from_dict(dict(package_name=app.package_name,
             ratings_average='5.00', ratings_total=1, histogram=None))]
-        command = ImportRatingsStatsCommand()
+        command = import_ratings_stats.Command()
 
         self.assertIsNone(command.update_apps_with_stats(app.distroseries,
             stats))
@@ -788,3 +797,95 @@
         self.assertEqual([True, False], [app.is_latest for app in retrieved])
         self.assertTrue(Application.objects.get(package_name='bar').is_latest)
         self.assertTrue(Application.objects.get(package_name='baz').is_latest)
+
+
+class CleanupTestCase(TestCaseWithFactory):
+    @patch('webcatalog.management.commands.cleanup.connection')
+    @patch('sys.stdout')
+    def run_and_check_output_and_sql(self, func, output, sql, mock_stdout,
+        mock_connection):
+        func()
+        actual_output = ''.join(str(call[0][0])
+            for call in mock_stdout.write.call_args_list)
+        self.assertEqual(output, actual_output.split('\n'))
+        call_list = mock_connection.cursor.return_value.execute.call_args_list
+        for expected, call in zip(sql, call_list):
+            actual = (' '.join(call[0][0].split()),) + call[0][1:]
+            self.assertEqual(expected, actual)
+
+    @skipUnless(connection.vendor == 'postgresql', "Requires postgresql")
+    def test_clean_nonces_no_nonces(self):
+        def cleanup():
+            call_command('cleanup', 'webcatalog_nonce')
+        output = ['Opening cursor',
+            'Removed 0 rows (0 total removed). Batch size 1',
+            'All done.', '']
+        sql = [
+            ('CLOSE ALL',),
+            ('DECLARE _oauth_nonce_clean NO SCROLL CURSOR WITH HOLD FOR '
+            'SELECT id FROM webcatalog_nonce '
+            "WHERE created_at < (CURRENT_TIMESTAMP - INTERVAL '5' HOUR) "
+            'ORDER BY created_at',),
+            ('FETCH %s FROM _oauth_nonce_clean', [1])]
+        self.run_and_check_output_and_sql(cleanup, output, sql)
+
+    @skipUnless(connection.vendor == 'postgresql', "Requires postgresql")
+    def test_clean_session_no_sessions(self):
+        def cleanup():
+            call_command('cleanup', 'django_session')
+        output = ['Opening cursor',
+            'Removed 0 rows (0 total removed). Batch size 1',
+            'All done.', '']
+        sql = [
+            ('CLOSE ALL',),
+            ('DECLARE _django_session_clean NO SCROLL CURSOR WITH HOLD FOR '
+             'SELECT session_key FROM django_session WHERE expire_date < '
+             'CURRENT_TIMESTAMP ORDER BY expire_date',),
+            ('FETCH %s FROM _django_session_clean', [1])]
+        self.run_and_check_output_and_sql(cleanup, output, sql)
+
+
+class CleanupDataTestCase(TransactionTestCase):
+    def setUp(self):
+        self.factory = WebCatalogObjectFactory()
+        self.patch_stdout = patch('sys.stdout')
+        self.mock_stdout = self.patch_stdout.start()
+        self.addCleanup(self.patch_stdout.stop)
+
+    def check_expected_output(self):
+        expected_output = '''Opening cursor
+Removed 1 rows (1 total removed). Batch size 1
+Removed 2 rows (3 total removed). Batch size 2
+Removed 1 rows (4 total removed). Batch size 3
+Removed 0 rows (4 total removed). Batch size 4
+All done.
+'''
+        actual_output = ''.join(str(call[0][0])
+            for call in self.mock_stdout.write.call_args_list)
+        self.assertEqual(expected_output, actual_output)
+
+    @skipUnless(connection.vendor == 'postgresql', "Requires postgresql")
+    def test_cleanup_nonces(self):
+        now = datetime.now()
+        for i in range(8):
+            created_at = now - timedelta(seconds=5000 * i)
+            self.factory.make_nonce(created_at=created_at)
+        self.assertEqual(8, Nonce.objects.count())
+
+        call_command('cleanup', 'webcatalog_nonce')
+
+        self.assertEqual(4, Nonce.objects.count())
+        self.check_expected_output()
+
+    @skipUnless(connection.vendor == 'postgresql', "Requires postgresql")
+    def test_cleanup_sessions(self):
+        now = datetime.now()
+        for i in range(8):
+            expire_date = now + timedelta(seconds=5000 * (4 - i))
+            self.factory.make_session(expire_date=expire_date)
+        self.assertEqual(8, Session.objects.count())
+
+        call_command('cleanup', 'django_session')
+
+        self.assertEqual(4, Session.objects.count())
+        self.check_expected_output()