← Back to branch summary

~zematynnad/ubuntu-webcatalog/fix_no_enlarger_845911

« back to all changes in this revision

Viewing changes to src/webcatalog/tests/test_commands.py

  • Committer: Tarmac
  • Author(s): Anthony Lenton
  • Date: 2012-04-17 17:09:03 UTC
  • mfrom: (99.3.2 flush-nonces)
  • Revision ID: tarmac@199959-20120417170903-f025ppg00zg4j2iu
[r=elachuni],[bug=983106] Added a command to clean up nonces and sessions.

Show diffs side-by-side

added added

removed removed

Lines of Context:
src/webcatalog/tests/test_commands.py
33
33
from decimal import Decimal
34
34
 
35
35
from django.conf import settings
 
36
from django.contrib.sessions.models import Session
36
37
from django.core.management import call_command
 
38
from django.db import connection
 
39
from django.utils.unittest import skipUnless
 
40
from django.test.testcases import TransactionTestCase
 
41
 
37
42
from mock import (
38
43
    patch,
39
44
    MagicMock,
45
50
    Application,
46
51
    DistroSeries,
47
52
    Exhibit,
 
53
    Nonce,
48
54
    ReviewStatsImport,
49
55
    )
50
 
from webcatalog.management.commands.import_app_install_data import (
51
 
    Command as ImportAppInstallCommand,
52
 
    )
53
 
from webcatalog.management.commands.import_ratings_stats import (
54
 
    Command as ImportRatingsStatsCommand,
55
 
    )
56
 
from webcatalog.tests.factory import TestCaseWithFactory
 
56
from webcatalog.management.commands import (
 
57
    import_app_install_data,
 
58
    import_ratings_stats,
 
59
)
 
60
from webcatalog.tests.factory import (
 
61
    TestCaseWithFactory,
 
62
    WebCatalogObjectFactory,
 
63
)
57
64
 
58
65
__metaclass__ = type
59
66
__all__ = [
 
67
    'CheckAllLatestTestCase',
 
68
    'CleanupTestCase',
 
69
    'CleanupDataTestCase',
60
70
    'ImportAppInstallTestCase',
61
71
    'ImportExhibitsTestCase',
62
72
    'ImportForPurchaseAppsTestCase',
63
73
    'ImportRatingsTestCase',
64
 
    'CheckAllLatestTestCase',
65
74
    ]
66
75
 
67
76
 
188
197
        with patch(get_uri_fn) as mock_get_uri:
189
198
            mock_get_uri.return_value = 'http://example.com/my.deb'
190
199
            with patch('urllib.urlretrieve') as mock_urlretrieve:
191
 
                ImportAppInstallCommand().get_package_data_for_series(
 
200
                import_app_install_data.Command().get_package_data_for_series(
192
201
                    'app-install-data', 'natty', tmp_dir)
193
202
        shutil.rmtree(tmp_dir)
194
203
 
305
314
        self.assertEqual(scribus.screenshots, ["http://example.com/1.png"])
306
315
 
307
316
    def get_package_uri_for_series(self):
308
 
        uri = ImportAppInstallCommand().get_package_uri_for_series(
 
317
        uri = import_app_install_data.Command().get_package_uri_for_series(
309
318
            'app-install-data', 'natty')
310
319
 
311
320
        self.assertEqual('http://example.com/app-install-1.01.deb', uri)
397
406
        self.assertEqual(0, DistroSeries.objects.filter(
398
407
            code_name='natty').count())
399
408
 
400
 
        ImportAppInstallCommand().update_from_cache('natty')
 
409
        import_app_install_data.Command().update_from_cache('natty')
401
410
 
402
411
        self.assertEqual(1, DistroSeries.objects.filter(
403
412
            code_name='natty').count())
405
414
    def test_apt_cache_apps_used_also(self):
406
415
        self.assertEqual(0, Application.objects.count())
407
416
 
408
 
        ImportAppInstallCommand().update_from_cache('natty')
 
417
        import_app_install_data.Command().update_from_cache('natty')
409
418
 
410
419
        # There are 4 packages in our mock_cache which is patched in the
411
420
        # setUp.
531
540
        candidate = Mock()
532
541
        candidate.uri = 'http://extras.ubuntu.com/pool/f/foobar.deb'
533
542
        candidate.record = {'Icon': 'foo.png'}
534
 
        command = ImportAppInstallCommand()
 
543
        command = import_app_install_data.Command()
535
544
 
536
545
        command.fetch_icon_from_extras(app, candidate)
537
546
 
546
555
        candidate = Mock()
547
556
        candidate.uri = 'http://extras.ubuntu.com/pool/f/foobar.deb'
548
557
        candidate.record = {}
549
 
        command = ImportAppInstallCommand()
 
558
        command = import_app_install_data.Command()
550
559
 
551
560
        command.fetch_icon_from_extras(app, candidate)
552
561
 
559
568
        candidate = Mock()
560
569
        candidate.uri = 'http://archive.ubuntu.com/pool/f/foobar.deb'
561
570
        candidate.record = {'Icon': 'foobar.png'}
562
 
        command = ImportAppInstallCommand()
 
571
        command = import_app_install_data.Command()
563
572
 
564
573
        command.fetch_icon_from_extras(app, candidate)
565
574
 
655
664
            package_name=self.factory.get_unique_string(prefix='package-'),
656
665
            ratings_average='5.00', ratings_total=1,
657
666
            histogram='[0, 0, 0, 0, 1]')) for x in range(3000)]
658
 
        command = ImportRatingsStatsCommand()
 
667
        command = import_ratings_stats.Command()
659
668
 
660
669
        # update_apps_with_stats returns None on success:
661
670
        self.assertIsNone(command.update_apps_with_stats(natty, stats))
665
674
        app = self.factory.make_application()
666
675
        stats = [ReviewsStats.from_dict(dict(package_name=app.package_name,
667
676
            ratings_average='5.00', ratings_total=1, histogram=None))]
668
 
        command = ImportRatingsStatsCommand()
 
677
        command = import_ratings_stats.Command()
669
678
 
670
679
        self.assertIsNone(command.update_apps_with_stats(app.distroseries,
671
680
            stats))
788
797
        self.assertEqual([True, False], [app.is_latest for app in retrieved])
789
798
        self.assertTrue(Application.objects.get(package_name='bar').is_latest)
790
799
        self.assertTrue(Application.objects.get(package_name='baz').is_latest)
 
800
 
 
801
 
 
802
class CleanupTestCase(TestCaseWithFactory):
 
803
    @patch('webcatalog.management.commands.cleanup.connection')
 
804
    @patch('sys.stdout')
 
805
    def run_and_check_output_and_sql(self, func, output, sql, mock_stdout,
 
806
        mock_connection):
 
807
        func()
 
808
        actual_output = ''.join(str(call[0][0])
 
809
            for call in mock_stdout.write.call_args_list)
 
810
        self.assertEqual(output, actual_output.split('\n'))
 
811
        call_list = mock_connection.cursor.return_value.execute.call_args_list
 
812
        for expected, call in zip(sql, call_list):
 
813
            actual = (' '.join(call[0][0].split()),) + call[0][1:]
 
814
            self.assertEqual(expected, actual)
 
815
 
 
816
    @skipUnless(connection.vendor == 'postgresql', "Requires postgresql")
 
817
    def test_clean_nonces_no_nonces(self):
 
818
        def cleanup():
 
819
            call_command('cleanup', 'webcatalog_nonce')
 
820
        output = ['Opening cursor',
 
821
            'Removed 0 rows (0 total removed). Batch size 1',
 
822
            'All done.', '']
 
823
        sql = [
 
824
            ('CLOSE ALL',),
 
825
            ('DECLARE _oauth_nonce_clean NO SCROLL CURSOR WITH HOLD FOR '
 
826
            'SELECT id FROM webcatalog_nonce '
 
827
            "WHERE created_at < (CURRENT_TIMESTAMP - INTERVAL '5' HOUR) "
 
828
            'ORDER BY created_at',),
 
829
            ('FETCH %s FROM _oauth_nonce_clean', [1])]
 
830
        self.run_and_check_output_and_sql(cleanup, output, sql)
 
831
 
 
832
    @skipUnless(connection.vendor == 'postgresql', "Requires postgresql")
 
833
    def test_clean_session_no_sessions(self):
 
834
        def cleanup():
 
835
            call_command('cleanup', 'django_session')
 
836
        output = ['Opening cursor',
 
837
            'Removed 0 rows (0 total removed). Batch size 1',
 
838
            'All done.', '']
 
839
        sql = [
 
840
            ('CLOSE ALL',),
 
841
            ('DECLARE _django_session_clean NO SCROLL CURSOR WITH HOLD FOR '
 
842
             'SELECT session_key FROM django_session WHERE expire_date < '
 
843
             'CURRENT_TIMESTAMP ORDER BY expire_date',),
 
844
            ('FETCH %s FROM _django_session_clean', [1])]
 
845
        self.run_and_check_output_and_sql(cleanup, output, sql)
 
846
 
 
847
 
 
848
class CleanupDataTestCase(TransactionTestCase):
 
849
    def setUp(self):
 
850
        self.factory = WebCatalogObjectFactory()
 
851
        self.patch_stdout = patch('sys.stdout')
 
852
        self.mock_stdout = self.patch_stdout.start()
 
853
        self.addCleanup(self.patch_stdout.stop)
 
854
 
 
855
    def check_expected_output(self):
 
856
        expected_output = '''Opening cursor
 
857
Removed 1 rows (1 total removed). Batch size 1
 
858
Removed 2 rows (3 total removed). Batch size 2
 
859
Removed 1 rows (4 total removed). Batch size 3
 
860
Removed 0 rows (4 total removed). Batch size 4
 
861
All done.
 
862
'''
 
863
        actual_output = ''.join(str(call[0][0])
 
864
            for call in self.mock_stdout.write.call_args_list)
 
865
        self.assertEqual(expected_output, actual_output)
 
866
 
 
867
    @skipUnless(connection.vendor == 'postgresql', "Requires postgresql")
 
868
    def test_cleanup_nonces(self):
 
869
        now = datetime.now()
 
870
        for i in range(8):
 
871
            created_at = now - timedelta(seconds=5000 * i)
 
872
            self.factory.make_nonce(created_at=created_at)
 
873
        self.assertEqual(8, Nonce.objects.count())
 
874
 
 
875
        call_command('cleanup', 'webcatalog_nonce')
 
876
 
 
877
        self.assertEqual(4, Nonce.objects.count())
 
878
        self.check_expected_output()
 
879
 
 
880
    @skipUnless(connection.vendor == 'postgresql', "Requires postgresql")
 
881
    def test_cleanup_sessions(self):
 
882
        now = datetime.now()
 
883
        for i in range(8):
 
884
            expire_date = now + timedelta(seconds=5000 * (4 - i))
 
885
            self.factory.make_session(expire_date=expire_date)
 
886
        self.assertEqual(8, Session.objects.count())
 
887
 
 
888
        call_command('cleanup', 'django_session')
 
889
 
 
890
        self.assertEqual(4, Session.objects.count())
 
891
        self.check_expected_output()