← Back to branch summary

~launchpad-pqm/launchpad/devel

« back to all changes in this revision

Viewing changes to lib/lp/scripts/tests/test_garbo.py

  • Committer: Ian Booth
  • Date: 2011-04-19 15:10:57 UTC
  • mfrom: (12868 devel)
  • mto: This revision was merged to the branch mainline in revision 12983.
  • Revision ID: ian.booth@canonical.com-20110419151057-he56y6k29c4zeiyk
Merge from trunk

Show diffs side-by-side

added added

removed removed

Lines of Context:
lib/lp/scripts/tests/test_garbo.py
10
10
    datetime,
11
11
    timedelta,
12
12
    )
13
 
from fixtures import TempDir
14
13
import logging
15
 
import os
16
14
from StringIO import StringIO
17
 
import subprocess
18
15
import time
19
16
 
20
17
from pytz import UTC
42
39
from canonical.launchpad.database.oauth import OAuthNonce
43
40
from canonical.launchpad.database.openidconsumer import OpenIDConsumerNonce
44
41
from canonical.launchpad.interfaces.emailaddress import EmailAddressStatus
45
 
from canonical.launchpad.interfaces.librarian import ILibraryFileAliasSet
46
42
from canonical.launchpad.interfaces.lpstorm import IMasterStore
47
43
from canonical.launchpad.scripts.tests import run_script
48
44
from canonical.launchpad.webapp.interfaces import (
56
52
    LaunchpadZopelessLayer,
57
53
    ZopelessDatabaseLayer,
58
54
    )
59
 
from lp.archiveuploader.dscfile import findFile
60
55
from lp.bugs.model.bugnotification import (
61
56
    BugNotification,
62
57
    BugNotificationRecipient,
73
68
    )
74
69
from lp.code.model.codeimportevent import CodeImportEvent
75
70
from lp.code.model.codeimportresult import CodeImportResult
76
 
from lp.registry.interfaces.distribution import IDistributionSet
77
71
from lp.registry.interfaces.person import (
78
72
    IPersonSet,
79
73
    PersonCreationRationale,
93
87
    SessionData,
94
88
    SessionPkgData,
95
89
    )
96
 
from lp.soyuz.enums import PackagePublishingStatus
97
 
from lp.soyuz.model.sourcepackagerelease import SourcePackageRelease
98
90
from lp.testing import (
99
91
    TestCase,
100
92
    TestCaseWithFactory,
114
106
 
115
107
    def test_hourly_script(self):
116
108
        """Ensure garbo-hourly.py actually runs."""
117
 
        # Our sampledata doesn't contain anything that PopulateSPRChangelogs
118
 
        # can process without errors, so it's easier to just set all of the
119
 
        # changelogs to a random LFA. We can't just expire every LFA, since
120
 
        # a bunch of SPRs have no SPRFs at all.
121
 
        IMasterStore(SourcePackageRelease).find(SourcePackageRelease).set(
122
 
            changelogID=1)
123
 
        transaction.commit() # run_script() is a different process.
124
109
        rv, out, err = run_script(
125
110
            "cronscripts/garbo-hourly.py", ["-q"], expect_returncode=0)
126
111
        self.failIf(out.strip(), "Output to stdout: %s" % out)
889
874
            """ % sqlbase.quote(template.id)).get_one()
890
875
 
891
876
        self.assertEqual(1, count)
892
 
 
893
 
    def upload_to_debian(self, restricted=False):
894
 
        sid = getUtility(IDistributionSet)['debian']['sid']
895
 
        spn = self.factory.makeSourcePackageName('9wm')
896
 
        spr = self.factory.makeSourcePackageRelease(
897
 
            sourcepackagename=spn, version='1.2-7', distroseries=sid)
898
 
        archive = sid.main_archive
899
 
        if restricted:
900
 
            archive = self.factory.makeArchive(
901
 
                distribution=sid.distribution, private=True)
902
 
        self.factory.makeSourcePackagePublishingHistory(
903
 
            sourcepackagerelease=spr, archive=archive,
904
 
            status=PackagePublishingStatus.PUBLISHED)
905
 
        for name in (
906
 
            '9wm_1.2-7.diff.gz', '9wm_1.2.orig.tar.gz', '9wm_1.2-7.dsc'):
907
 
            path = os.path.join(
908
 
                'lib/lp/soyuz/scripts/tests/gina_test_archive/pool/main/9',
909
 
                '9wm', name)
910
 
            lfa = getUtility(ILibraryFileAliasSet).create(
911
 
                name, os.stat(path).st_size, open(path, 'r'),
912
 
                'application/octet-stream', restricted=restricted)
913
 
            spr.addFile(lfa)
914
 
        with TempDir() as tmp_dir:
915
 
            fnull = open('/dev/null', 'w')
916
 
            ret = subprocess.call(
917
 
                ['dpkg-source', '-x', path, os.path.join(
918
 
                    tmp_dir.path, 'extracted')],
919
 
                    stdout=fnull, stderr=fnull)
920
 
            fnull.close()
921
 
            self.assertEqual(0, ret)
922
 
            changelog_path = findFile(tmp_dir.path, 'debian/changelog')
923
 
            changelog = open(changelog_path, 'r').read()
924
 
        transaction.commit() # .runHourly() switches dbuser.
925
 
        return (spr, changelog)
926
 
 
927
 
    def test_populateSPRChangelogs(self):
928
 
        # We set SPR.changelog for imported records from Debian.
929
 
        LaunchpadZopelessLayer.switchDbUser('testadmin')
930
 
        spr, changelog = self.upload_to_debian()
931
 
        collector = self.runHourly()
932
 
        log = self.log_buffer.getvalue()
933
 
        self.assertTrue(
934
 
            'SPR %d (9wm 1.2-7) changelog imported.' % spr.id in log)
935
 
        self.assertFalse(spr.changelog == None)
936
 
        self.assertFalse(spr.changelog.restricted)
937
 
        self.assertEqual(changelog, spr.changelog.read())
938
 
 
939
 
    def test_populateSPRChangelogs_restricted_sprf(self):
940
 
        LaunchpadZopelessLayer.switchDbUser('testadmin')
941
 
        spr, changelog = self.upload_to_debian(restricted=True)
942
 
        collector = self.runHourly()
943
 
        log = self.log_buffer.getvalue()
944
 
        self.assertTrue(
945
 
            'SPR %d (9wm 1.2-7) changelog imported.' % spr.id in log)
946
 
        self.assertFalse(spr.changelog == None)
947
 
        self.assertTrue(spr.changelog.restricted)
948
 
        self.assertEqual(changelog, spr.changelog.read())