~launchpad-pqm/launchpad/devel

# Copyright 2009-2011 Canonical Ltd.  This software is licensed under the

# GNU Affero General Public License version 3 (see the file LICENSE).

"""Functional tests for uploadprocessor.py."""

__metaclass__ = type

__all__ = [

    "MockOptions",

    "TestUploadProcessorBase",

    ]

from email import message_from_string

import os

import shutil

from StringIO import StringIO

import tempfile

from fixtures import MonkeyPatch

from storm.locals import Store

import transaction

from zope.component import (

    getGlobalSiteManager,

    getUtility,

    )

from zope.security.proxy import removeSecurityProxy

from canonical.config import config

from canonical.database.constants import UTC_NOW

from canonical.launchpad.interfaces.librarian import ILibraryFileAliasSet

from canonical.launchpad.testing.fakepackager import FakePackager

from canonical.testing.layers import LaunchpadZopelessLayer

from lp.app.errors import NotFoundError

from lp.archiveuploader.nascentupload import NascentUpload

from lp.archiveuploader.nascentuploadfile import DdebBinaryUploadFile

from lp.archiveuploader.tests import (

    datadir,

    getPolicy,

    )

from lp.archiveuploader.uploadpolicy import (

    AbstractUploadPolicy,

    ArchiveUploadType,

    findPolicyByName,

    IArchiveUploadPolicy,

    )

from lp.archiveuploader.uploadprocessor import (

    BuildUploadHandler,

    CannotGetBuild,

    parse_build_upload_leaf_name,

    UploadHandler,

    UploadProcessor,

    UploadStatusEnum,

    )

from lp.buildmaster.enums import (

    BuildFarmJobType,

    BuildStatus,

    )

from lp.registry.interfaces.distribution import IDistributionSet

from lp.registry.interfaces.person import IPersonSet

from lp.registry.interfaces.pocket import PackagePublishingPocket

from lp.registry.interfaces.series import SeriesStatus

from lp.registry.interfaces.sourcepackage import SourcePackageFileType

from lp.registry.interfaces.sourcepackagename import ISourcePackageNameSet

from lp.registry.model.sourcepackagename import SourcePackageName

from lp.services.features.testing import FeatureFixture

from lp.services.log.logger import (

    BufferLogger,

    DevNullLogger,

    )

from lp.services.mail import stub

from lp.soyuz.enums import (

    ArchivePermissionType,

    ArchivePurpose,

    PackageUploadStatus,

    SourcePackageFormat,

    )

from lp.soyuz.interfaces.archive import IArchiveSet

from lp.soyuz.interfaces.archivepermission import IArchivePermissionSet

from lp.soyuz.interfaces.component import IComponentSet

from lp.soyuz.interfaces.packageset import IPackagesetSet

from lp.soyuz.interfaces.publishing import (

    IPublishingSet,

    PackagePublishingStatus,

    )

from lp.soyuz.interfaces.queue import QueueInconsistentStateError

from lp.soyuz.interfaces.sourcepackageformat import (

    ISourcePackageFormatSelectionSet,

    )

from lp.soyuz.model.archivepermission import ArchivePermission

from lp.soyuz.model.binarypackagename import BinaryPackageName

from lp.soyuz.model.binarypackagerelease import BinaryPackageRelease

from lp.soyuz.model.component import Component

from lp.soyuz.model.distroseriesdifferencejob import (

    FEATURE_FLAG_ENABLE_MODULE,

    )

from lp.soyuz.model.publishing import (

    BinaryPackagePublishingHistory,

    SourcePackagePublishingHistory,

    )

from lp.soyuz.model.sourcepackagerelease import SourcePackageRelease

from lp.soyuz.scripts.initialize_distroseries import InitializeDistroSeries

from lp.testing import (

    TestCase,

    TestCaseWithFactory,

    )

from lp.testing.fakemethod import FakeMethod

from lp.testing.gpgkeys import import_public_test_keys

from lp.testing.mail_helpers import pop_notifications

class MockOptions:

    """Use in place of an options object, adding more attributes if needed."""

    keep = False

    dryrun = False

class BrokenUploadPolicy(AbstractUploadPolicy):

    """A broken upload policy, to test error handling."""

    def __init__(self):

        AbstractUploadPolicy.__init__(self)

        self.name = "broken"

        self.unsigned_changes_ok = True

        self.unsigned_dsc_ok = True

    def checkUpload(self, upload):

        """Raise an exception upload processing is not expecting."""

        raise Exception("Exception raised by BrokenUploadPolicy for testing.")

class TestUploadProcessorBase(TestCaseWithFactory):

    """Base class for functional tests over uploadprocessor.py."""

    layer = LaunchpadZopelessLayer

    def switchToUploader(self):

        transaction.commit()

        self.layer.switchDbUser("uploader")

    def switchToAdmin(self):

        transaction.commit()

        self.layer.switchDbUser("launchpad_main")

    def setUp(self):

        super(TestUploadProcessorBase, self).setUp()

        self.queue_folder = tempfile.mkdtemp()

        self.incoming_folder = os.path.join(self.queue_folder, "incoming")

        os.makedirs(self.incoming_folder)

        self.test_files_dir = os.path.join(config.root,

            "lib/lp/archiveuploader/tests/data/suite")

        import_public_test_keys()

        self.options = MockOptions()

        self.options.base_fsroot = self.queue_folder

        self.options.builds = False

        self.options.leafname = None

        self.options.distro = "ubuntu"

        self.options.distroseries = None

        self.options.nomails = False

        self.options.context = 'insecure'

        # common recipients

        self.kinnison_recipient = (

            "Daniel Silverstone <daniel.silverstone@canonical.com>")

        self.name16_recipient = "Foo Bar <foo.bar@canonical.com>"

        self.log = BufferLogger()

        self.switchToAdmin()

        self.useFixture(FeatureFixture({

            FEATURE_FLAG_ENABLE_MODULE: u'on',

        }))

        self.switchToUploader()

    def tearDown(self):

        shutil.rmtree(self.queue_folder)

        self.switchToAdmin()

        super(TestUploadProcessorBase, self).tearDown()

    def getUploadProcessor(self, txn, builds=None):

        self.switchToAdmin()

        if builds is None:

            builds = self.options.builds

        def getUploadPolicy(distro, build):

            self.options.distro = distro.name

            policy = findPolicyByName(self.options.context)

            if builds:

                policy.distroseries = build.distro_series

                policy.pocket = build.pocket

                policy.archive = build.archive

            policy.setOptions(self.options)

            return policy

        upload_processor = UploadProcessor(

            self.options.base_fsroot, self.options.dryrun,

            self.options.nomails, builds, self.options.keep, getUploadPolicy,

            txn, self.log)

        self.switchToUploader()

        return upload_processor

    def publishPackage(self, packagename, version, source=True, archive=None):

        """Publish a single package that is currently NEW in the queue."""

        self.switchToAdmin()

        packagename = unicode(packagename)

        if version is not None:

            version = unicode(version)

        queue_items = self.breezy.getPackageUploads(

            status=PackageUploadStatus.NEW, name=packagename,

            version=version, exact_match=True, archive=archive)

        self.assertEqual(queue_items.count(), 1)

        queue_item = queue_items[0]

        queue_item.setAccepted()

        if source:

            pubrec = queue_item.sources[0].publish(self.log)

        else:

            pubrec = queue_item.builds[0].publish(self.log)

        self.switchToUploader()

        return pubrec

    def assertLogContains(self, line):

        """Assert if a given line is present in the log messages."""

        log_lines = self.log.getLogBuffer()

        self.assertTrue(line in log_lines,

                        "'%s' is not in logged output\n\n%s"

                        % (line, log_lines))

    def assertRaisesAndReturnError(self, excClass, callableObj, *args,

                                   **kwargs):

        """See `TestCase.assertRaises`.

        Unlike `TestCase.assertRaises`, this method returns the exception

        object when it is raised.  Callsites can then easily check the

        exception contents.

        """

        try:

            callableObj(*args, **kwargs)

        except excClass, error:

            return error

        else:

            if getattr(excClass, '__name__', None) is not None:

                excName = excClass.__name__

            else:

                excName = str(excClass)

            raise self.failureException, "%s not raised" % excName

    def setupBreezy(self, name="breezy", permitted_formats=None):

        """Create a fresh distroseries in ubuntu.

        Use *initializeFromParent* procedure to create 'breezy'

        on ubuntu based on the last 'breezy-autotest'.

        Also sets 'changeslist' and 'nominatedarchindep' properly and

        creates a chroot for breezy-autotest/i386 distroarchseries.

        :param name: supply the name of the distroseries if you don't want

            it to be called "breezy"

        :param permitted_formats: list of SourcePackageFormats to allow

            in the new distroseries. Only permits '1.0' by default.

        """

        self.switchToAdmin()

        self.ubuntu = getUtility(IDistributionSet).getByName('ubuntu')

        bat = self.ubuntu['breezy-autotest']

        self.breezy = self.ubuntu.newSeries(

            name, 'Breezy Badger',

            'The Breezy Badger', 'Black and White', 'Someone',

            '5.10', None, bat.owner)

        self.breezy.previous_series = bat

        self.breezy.changeslist = 'breezy-changes@ubuntu.com'

        ids = InitializeDistroSeries(self.breezy, [bat.id])

        ids.initialize()

        fake_chroot = self.addMockFile('fake_chroot.tar.gz')

        self.breezy['i386'].addOrUpdateChroot(fake_chroot)

        if permitted_formats is None:

            permitted_formats = [SourcePackageFormat.FORMAT_1_0]

        for format in permitted_formats:

            if not self.breezy.isSourcePackageFormatPermitted(format):

                getUtility(ISourcePackageFormatSelectionSet).add(

                    self.breezy, format)

        self.switchToUploader()

    def addMockFile(self, filename, content="anything"):

        """Return a librarian file."""

        return getUtility(ILibraryFileAliasSet).create(

            filename, len(content), StringIO(content),

            'application/x-gtar')

    def queueUpload(self, upload_name, relative_path="", test_files_dir=None,

            queue_entry=None):

        """Queue one of our test uploads.

        upload_name is the name of the test upload directory. If there

        is no explicit queue entry name specified, it is also

        the name of the queue entry directory we create.

        relative_path is the path to create inside the upload, eg

        ubuntu/~malcc/default. If not specified, defaults to "".

        Return the path to the upload queue entry directory created.

        """

        if queue_entry is None:

            queue_entry = upload_name

        target_path = os.path.join(

            self.incoming_folder, queue_entry, relative_path)

        if test_files_dir is None:

            test_files_dir = self.test_files_dir

        upload_dir = os.path.join(test_files_dir, upload_name)

        if relative_path:

            os.makedirs(os.path.dirname(target_path))

        shutil.copytree(upload_dir, target_path)

        return os.path.join(self.incoming_folder, queue_entry)

    def processUpload(self, processor, upload_dir, build=None):

        """Process an upload queue entry directory.

        There is some duplication here with logic in UploadProcessor,

        but we need to be able to do this without error handling here,

        so that we can debug failing tests effectively.

        """

        results = []

        self.assertEqual(processor.builds, build is not None)

        handler = UploadHandler.forProcessor(

            processor, '.', upload_dir, build)

        changes_files = handler.locateChangesFiles()

        for changes_file in changes_files:

            result = handler.processChangesFile(changes_file)

            results.append(result)

        return results

    def setupBreezyAndGetUploadProcessor(self, policy=None):

        """Setup Breezy and return an upload processor for it."""

        self.setupBreezy()

        self.switchToAdmin()

        self.layer.txn.commit()

        if policy is not None:

            self.options.context = policy

        upload_processor = self.getUploadProcessor(self.layer.txn)

        self.switchToUploader()

        return upload_processor

    def assertEmail(self, contents=None, recipients=None):

        """Check last email content and recipients.

        :param contents: A list of lines; assert that each is in the email.

        :param recipients: A list of recipients that must be on the email.

                           Supply an empty list if you don't want them

                           checked.  Default action is to check that the

                           recipient is foo.bar@canonical.com, which is the

                           signer on most of the test data uploads.

        """

        if recipients is None:

            recipients = [self.name16_recipient]

        if contents is None:

            contents = []

        self.assertEqual(

            len(stub.test_emails), 1,

            'Unexpected number of emails sent: %s' % len(stub.test_emails))

        from_addr, to_addrs, raw_msg = stub.test_emails.pop()

        msg = message_from_string(raw_msg)

        # This is now a MIMEMultipart message.

        body = msg.get_payload(0)

        body = body.get_payload(decode=True)

        # Only check recipients if callsite didn't provide an empty list.

        if recipients != []:

            clean_recipients = [r.strip() for r in to_addrs]

            for recipient in list(recipients):

                self.assertTrue(

                    recipient in clean_recipients,

                    "%s not found in %s" % (recipients, clean_recipients))

            self.assertEqual(

                len(recipients), len(clean_recipients),

                "Email recipients do not match exactly. Expected %s, got %s" %

                    (recipients, clean_recipients))

        subject = "Subject: %s\n" % msg['Subject']

        body = subject + body

        for content in list(contents):

            self.assertTrue(

                content in body,

                "Expect: '%s'\nGot:\n%s" % (content, body))

    def PGPSignatureNotPreserved(self, archive=None):

        """PGP signatures should be removed from .changes files.

        Email notifications and the librarian file for .changes file should

        both have the PGP signature removed.

        """

        bar = archive.getPublishedSources(

            name='bar', version="1.0-1", exact_match=True)

        changes_lfa = getUtility(IPublishingSet).getChangesFileLFA(

            bar.first().sourcepackagerelease)

        changes_file = changes_lfa.read()

        self.assertTrue(

            "Format: " in changes_file, "Does not look like a changes file")

        self.assertTrue(

            "-----BEGIN PGP SIGNED MESSAGE-----" not in changes_file,

            "Unexpected PGP header found")

        self.assertTrue(

            "-----BEGIN PGP SIGNATURE-----" not in changes_file,

            "Unexpected start of PGP signature found")

        self.assertTrue(

            "-----END PGP SIGNATURE-----" not in changes_file,

            "Unexpected end of PGP signature found")

class TestUploadProcessor(TestUploadProcessorBase):

    """Basic tests on uploadprocessor class.

    * Check if the rejection message is send even when an unexpected

      exception occur when processing the upload.

    * Check if known uploads targeted to a FROZEN distroseries

      end up in UNAPPROVED queue.

    This test case is able to setup a fresh distroseries in Ubuntu.

    """

    def _checkPartnerUploadEmailSuccess(self):

        """Ensure partner uploads generate the right email."""

        from_addr, to_addrs, raw_msg = stub.test_emails.pop()

        foo_bar = "Foo Bar <foo.bar@canonical.com>"

        self.assertEqual([e.strip() for e in to_addrs], [foo_bar])

        self.assertTrue(

            "rejected" not in raw_msg,

            "Expected acceptance email not rejection. Actually Got:\n%s"

                % raw_msg)

    def testInstantiate(self):

        """UploadProcessor should instantiate"""

        self.getUploadProcessor(None)

    def testLocateDirectories(self):

        """Return a sorted list of subdirs in a directory.

        We don't test that we block on the lockfile, as this is trivial

        code but tricky to test.

        """

        testdir = tempfile.mkdtemp()

        try:

            os.mkdir("%s/dir3" % testdir)

            os.mkdir("%s/dir1" % testdir)

            os.mkdir("%s/dir2" % testdir)

            up = self.getUploadProcessor(None)

            located_dirs = up.locateDirectories(testdir)

            self.assertEqual(located_dirs, ['dir1', 'dir2', 'dir3'])

        finally:

            shutil.rmtree(testdir)

    def _makeUpload(self, testdir):

        """Create a dummy upload for testing the move/remove methods."""

        upload = tempfile.mkdtemp(dir=testdir)

        distro = upload + ".distro"

        f = open(distro, mode="w")

        f.write("foo")

        f.close()

        return upload, distro

    def testMoveUpload(self):

        """moveUpload should move the upload directory and .distro file."""

        testdir = tempfile.mkdtemp()

        try:

            # Create an upload, a .distro and a target to move it to.

            upload, distro = self._makeUpload(testdir)

            target = tempfile.mkdtemp(dir=testdir)

            target_name = os.path.basename(target)

            upload_name = os.path.basename(upload)

            # Move it

            self.options.base_fsroot = testdir

            up = self.getUploadProcessor(None)

            handler = UploadHandler(up, '.', upload)

            handler.moveUpload(target_name, self.log)

            # Check it moved

            self.assertTrue(os.path.exists(os.path.join(target, upload_name)))

            self.assertTrue(os.path.exists(os.path.join(

                target, upload_name + ".distro")))

            self.assertFalse(os.path.exists(upload))

            self.assertFalse(os.path.exists(distro))

        finally:

            shutil.rmtree(testdir)

    def testMoveProcessUploadAccepted(self):

        testdir = tempfile.mkdtemp()

        try:

            # Create an upload, a .distro and a target to move it to.

            upload, distro = self._makeUpload(testdir)

            upload_name = os.path.basename(upload)

            # Remove it

            self.options.base_fsroot = testdir

            up = self.getUploadProcessor(None)

            handler = UploadHandler(up, '.', upload)

            handler.moveProcessedUpload("accepted", self.log)

            # Check it was removed, not moved

            self.assertFalse(os.path.exists(os.path.join(

                testdir, "accepted")))

            self.assertFalse(os.path.exists(os.path.join(testdir,

                "accepted", upload_name + ".distro")))

            self.assertFalse(os.path.exists(upload))

            self.assertFalse(os.path.exists(distro))

        finally:

            shutil.rmtree(testdir)

    def testMoveProcessUploadFailed(self):

        """moveProcessedUpload moves if the result was not successful."""

        testdir = tempfile.mkdtemp()

        try:

            # Create an upload, a .distro and a target to move it to.

            upload, distro = self._makeUpload(testdir)

            upload_name = os.path.basename(upload)

            # Move it

            self.options.base_fsroot = testdir

            up = self.getUploadProcessor(None)

            handler = UploadHandler(up, '.', upload)

            handler.moveProcessedUpload("rejected", self.log)

            # Check it moved

            self.assertTrue(os.path.exists(os.path.join(testdir,

                "rejected", upload_name)))

            self.assertTrue(os.path.exists(os.path.join(testdir,

                "rejected", upload_name + ".distro")))

            self.assertFalse(os.path.exists(upload))

            self.assertFalse(os.path.exists(distro))

        finally:

            shutil.rmtree(testdir)

    def testRemoveUpload(self):

        """RemoveUpload removes the upload directory and .distro file."""

        testdir = tempfile.mkdtemp()

        try:

            # Create an upload, a .distro and a target to move it to.

            upload, distro = self._makeUpload(testdir)

            os.path.basename(upload)

            # Remove it

            self.options.base_fsroot = testdir

            up = self.getUploadProcessor(None)

            handler = UploadHandler(up, '.', upload)

            handler.removeUpload(self.log)

            # Check it was removed, not moved

            self.assertFalse(os.path.exists(os.path.join(

                testdir, "accepted")))

            self.assertFalse(os.path.exists(upload))

            self.assertFalse(os.path.exists(distro))

        finally:

            shutil.rmtree(testdir)

    def testRejectionEmailForUnhandledException(self):

        """Test there's a rejection email when nascentupload breaks.

        If a developer makes an upload which finds a bug in nascentupload,

        and an unhandled exception occurs, we should try to send a

        rejection email. We'll test that this works, in a case where we

        will have the right information to send the email before the

        error occurs.

        If we haven't extracted enough information to send a rejection

        email when things break, trying to send one will raise a new

        exception, and the upload will fail silently as before. We don't

        test this case.

        See bug 35965.

        """

        self.options.context = u'broken'

        # Register our broken upload policy.

        getGlobalSiteManager().registerUtility(

            component=BrokenUploadPolicy, provided=IArchiveUploadPolicy,

            name=self.options.context)

        uploadprocessor = self.getUploadProcessor(self.layer.txn)

        # Upload a package to Breezy.

        upload_dir = self.queueUpload("baz_1.0-1")

        self.processUpload(uploadprocessor, upload_dir)

        # Check the mailer stub has a rejection email for Daniel

        from_addr, to_addrs, raw_msg = stub.test_emails.pop()

        # This is now a MIMEMultipart message.

        msg = message_from_string(raw_msg)

        body = msg.get_payload(0)

        body = body.get_payload(decode=True)

        daniel = "Daniel Silverstone <daniel.silverstone@canonical.com>"

        self.assertEqual(to_addrs, [daniel])

        self.assertTrue("Unhandled exception processing upload: Exception "

                        "raised by BrokenUploadPolicy for testing."

                        in body)

    def testUploadToFrozenDistro(self):

        """Uploads to a frozen distroseries should work, but be unapproved.

        The rule for a frozen distroseries is that uploads should still

        be permitted, but that the usual rule for auto-accepting uploads

        of existing packages should be suspended. New packages will still

        go into NEW, but new versions will be UNAPPROVED, rather than

        ACCEPTED.

        To test this, we will upload two versions of the same package,

        accepting and publishing the first, and freezing the distroseries

        before the second. If all is well, the second upload should go

        through ok, but end up in status UNAPPROVED, and with the

        appropriate email contents.

        See bug 58187.

        """

        # Set up the uploadprocessor with appropriate options and logger

        uploadprocessor = self.setupBreezyAndGetUploadProcessor()

        # Upload a package for Breezy.

        upload_dir = self.queueUpload("bar_1.0-1")

        self.processUpload(uploadprocessor, upload_dir)

        # Check it went ok to the NEW queue and all is going well so far.

        from_addr, to_addrs, raw_msg = stub.test_emails.pop()

        foo_bar = "Foo Bar <foo.bar@canonical.com>"

        daniel = "Daniel Silverstone <daniel.silverstone@canonical.com>"

        self.assertContentEqual(

            [foo_bar, daniel], [e.strip() for e in to_addrs])

        self.assertTrue(

            "NEW" in raw_msg, "Expected email containing 'NEW', got:\n%s"

            % raw_msg)

        # Accept and publish the upload.

        # This is required so that the next upload of a later version of

        # the same package will work correctly.

        self.switchToAdmin()

        queue_items = self.breezy.getPackageUploads(

            status=PackageUploadStatus.NEW, name=u"bar",

            version=u"1.0-1", exact_match=True)