~launchpad-pqm/launchpad/devel

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

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

"""Functional tests for publish-distro.py script."""

__metaclass__ = type

from optparse import OptionParser

import os

import shutil

import subprocess

import sys

import unittest

from zope.component import getUtility

from zope.security.proxy import removeSecurityProxy

from canonical.config import config

from lp.archivepublisher.config import getPubConfig

from lp.archivepublisher.interfaces.publisherconfig import IPublisherConfigSet

from lp.registry.interfaces.distribution import IDistributionSet

from lp.registry.interfaces.person import IPersonSet

from lp.services.log.logger import BufferLogger

from lp.services.scripts.base import LaunchpadScriptFailure

from lp.soyuz.enums import (

    ArchivePurpose,

    BinaryPackageFormat,

    PackagePublishingStatus,

    )

from lp.soyuz.interfaces.archive import (

    IArchiveSet,

    )

from lp.soyuz.scripts import publishdistro

from lp.soyuz.tests.test_publishing import TestNativePublishingBase

class TestPublishDistro(TestNativePublishingBase):

    """Test the publish-distro.py script works properly."""

    def runPublishDistro(self, extra_args=None, distribution="ubuntutest"):

        """Run publish-distro without invoking the script.

        This method hooks into the publishdistro module to run the

        publish-distro script without the overhead of using Popen.

        """

        args = ["-d", distribution]

        if extra_args is not None:

            args.extend(extra_args)

        parser = OptionParser()

        publishdistro.add_options(parser)

        options, args = parser.parse_args(args=args)

        self.layer.switchDbUser(config.archivepublisher.dbuser)

        publishdistro.run_publisher(

            options, self.layer.txn, log=BufferLogger())

        self.layer.switchDbUser('launchpad')

    def runPublishDistroScript(self):

        """Run publish-distro.py, returning the result and output."""

        script = os.path.join(config.root, "scripts", "publish-distro.py")

        args = [sys.executable, script, "-v", "-d", "ubuntutest"]

        process = subprocess.Popen(

            args, stdout=subprocess.PIPE, stderr=subprocess.PIPE)

        stdout, stderr = process.communicate()

        return (process.returncode, stdout, stderr)

    def testPublishDistroRun(self):

        """Try a simple publish-distro run.

        Expect database publishing record to be updated to PUBLISHED and

        the file to be written in disk.

        This method also ensures the publish-distro.py script is runnable.

        """

        pub_source = self.getPubSource(filecontent='foo')

        self.layer.txn.commit()

        rc, out, err = self.runPublishDistroScript()

        pub_source.sync()

        self.assertEqual(0, rc, "Publisher failed with:\n%s\n%s" % (out, err))

        self.assertEqual(pub_source.status, PackagePublishingStatus.PUBLISHED)

        foo_path = "%s/main/f/foo/foo_666.dsc" % self.pool_dir

        self.assertEqual(open(foo_path).read().strip(), 'foo')

    def testDirtyPocketProcessing(self):

        """Test dirty pocket processing.

        Make a DELETED source to see if the dirty pocket processing

        works for deletions.

        """

        pub_source = self.getPubSource(filecontent='foo')

        self.layer.txn.commit()

        self.runPublishDistro()

        pub_source.sync()

        random_person = getUtility(IPersonSet).getByName('name16')

        pub_source.requestDeletion(random_person)

        self.layer.txn.commit()

        self.assertTrue(pub_source.scheduleddeletiondate is None,

            "pub_source.scheduleddeletiondate should not be set, and it is.")

        self.runPublishDistro()

        pub_source.sync()

        self.assertTrue(pub_source.scheduleddeletiondate is not None,

            "pub_source.scheduleddeletiondate should be set, and it's not.")

    def assertExists(self, path):

        """Assert if the given path exists."""

        self.assertTrue(os.path.exists(path), "Not Found: '%s'" % path)

    def assertNotExists(self, path):

        """Assert if the given path does not exist."""

        self.assertFalse(os.path.exists(path), "Found: '%s'" % path)

    def testRunWithSuite(self):

        """Try to run publish-distro with restricted suite option.

        Expect only update and disk writing only in the publishing record

        targeted to the specified suite, other records should be untouched

        and not present in disk.

        """

        pub_source = self.getPubSource(filecontent='foo')

        pub_source2 = self.getPubSource(

            sourcename='baz', filecontent='baz',

            distroseries=self.ubuntutest['hoary-test'])

        self.layer.txn.commit()

        self.runPublishDistro(['-s', 'hoary-test'])

        pub_source.sync()

        pub_source2.sync()

        self.assertEqual(pub_source.status, PackagePublishingStatus.PENDING)

        self.assertEqual(

            pub_source2.status, PackagePublishingStatus.PUBLISHED)

        foo_path = "%s/main/f/foo/foo_666.dsc" % self.pool_dir

        self.assertNotExists(foo_path)

        baz_path = "%s/main/b/baz/baz_666.dsc" % self.pool_dir

        self.assertEqual('baz', open(baz_path).read().strip())

    def publishToArchiveWithOverriddenDistsroot(self, archive):

        """Publish a test package to the specified archive.

        Publishes a test package but overrides the distsroot.

        :return: A tuple of the path to the overridden distsroot and the

                 configured distsroot, in that order.

        """

        self.getPubSource(filecontent="flangetrousers", archive=archive)

        self.layer.txn.commit()

        pubconf = getPubConfig(archive)

        tmp_path = os.path.join(pubconf.archiveroot, "tmpdistroot")

        if os.path.exists(tmp_path):

            shutil.rmtree(tmp_path)

        os.makedirs(tmp_path)

        myargs = ['-R', tmp_path]

        if archive.purpose == ArchivePurpose.PARTNER:

            myargs.append('--partner')

        self.runPublishDistro(myargs)

        return tmp_path, pubconf.distsroot

    def testDistsrootOverridePrimaryArchive(self):

        """Test the -R option to publish-distro.

        Make sure that -R works with the primary archive.

        """

        main_archive = getUtility(IDistributionSet)['ubuntutest'].main_archive

        tmp_path, distsroot = self.publishToArchiveWithOverriddenDistsroot(

            main_archive)

        distroseries = 'breezy-autotest'

        self.assertExists(os.path.join(tmp_path, distroseries, 'Release'))

        self.assertNotExists(

            os.path.join("%s" % distsroot, distroseries, 'Release'))

        shutil.rmtree(tmp_path)

    def testDistsrootOverridePartnerArchive(self):

        """Test the -R option to publish-distro.

        Make sure the -R option affects the partner archive.

        """

        ubuntu = getUtility(IDistributionSet)['ubuntutest']

        partner_archive = ubuntu.getArchiveByComponent('partner')

        tmp_path, distsroot = self.publishToArchiveWithOverriddenDistsroot(

            partner_archive)

        distroseries = 'breezy-autotest'

        self.assertExists(os.path.join(tmp_path, distroseries, 'Release'))

        self.assertNotExists(

            os.path.join("%s" % distsroot, distroseries, 'Release'))

        shutil.rmtree(tmp_path)

    def testForPPA(self):

        """Try to run publish-distro in PPA mode.

        It should deal only with PPA publications.

        """

        pub_source = self.getPubSource(filecontent='foo')

        cprov = getUtility(IPersonSet).getByName('cprov')

        pub_source2 = self.getPubSource(

            sourcename='baz', filecontent='baz', archive=cprov.archive)

        ubuntutest = getUtility(IDistributionSet)['ubuntutest']

        name16 = getUtility(IPersonSet).getByName('name16')

        getUtility(IArchiveSet).new(purpose=ArchivePurpose.PPA, owner=name16,

            distribution=ubuntutest)

        pub_source3 = self.getPubSource(

            sourcename='bar', filecontent='bar', archive=name16.archive)

        # Override PPAs distributions

        naked_archive = removeSecurityProxy(cprov.archive)

        naked_archive.distribution = self.ubuntutest

        naked_archive = removeSecurityProxy(name16.archive)

        naked_archive.distribution = self.ubuntutest

        self.layer.txn.commit()

        self.runPublishDistro(['--ppa'])

        pub_source.sync()

        pub_source2.sync()

        pub_source3.sync()

        self.assertEqual(pub_source.status, PackagePublishingStatus.PENDING)

        self.assertEqual(

            pub_source2.status, PackagePublishingStatus.PUBLISHED)

        self.assertEqual(

            pub_source3.status, PackagePublishingStatus.PUBLISHED)

        foo_path = "%s/main/f/foo/foo_666.dsc" % self.pool_dir

        self.assertEqual(False, os.path.exists(foo_path))

        baz_path = os.path.join(

            config.personalpackagearchive.root, cprov.name,

            'ppa/ubuntutest/pool/main/b/baz/baz_666.dsc')

        self.assertEqual('baz', open(baz_path).read().strip())

        bar_path = os.path.join(

            config.personalpackagearchive.root, name16.name,

            'ppa/ubuntutest/pool/main/b/bar/bar_666.dsc')

        self.assertEqual('bar', open(bar_path).read().strip())

    def testForPrivatePPA(self):

        """Run publish-distro in private PPA mode.

        It should only publish private PPAs.

        """

        # First, we'll make a private PPA and populate it with a

        # publishing record.

        ubuntutest = getUtility(IDistributionSet)['ubuntutest']

        private_ppa = self.factory.makeArchive(

            private=True, distribution=ubuntutest)

        # Publish something to the private PPA:

        pub_source =  self.getPubSource(

            sourcename='baz', filecontent='baz', archive=private_ppa)

        self.layer.txn.commit()

        # Try a plain PPA run, to ensure the private one is NOT published.

        self.runPublishDistro(['--ppa'])

        pub_source.sync()

        self.assertEqual(pub_source.status, PackagePublishingStatus.PENDING)

        # Now publish the private PPAs and make sure they are really

        # published.

        self.runPublishDistro(['--private-ppa'])

        pub_source.sync()

        self.assertEqual(pub_source.status, PackagePublishingStatus.PUBLISHED)

    def testPublishPrimaryDebug(self):

        # 'ubuntutest' (default testing distribution) has no DEBUG

        # archive, Thus an error is raised.

        self.assertRaises(

            LaunchpadScriptFailure,

            self.runPublishDistro, ['--primary-debug'])

        # The DEBUG repository path was not created.

        ubuntutest = getUtility(IDistributionSet)['ubuntutest']

        root_dir = getUtility(

            IPublisherConfigSet).getByDistribution(ubuntutest).root_dir

        repo_path = os.path.join(

            root_dir, 'ubuntutest-debug')

        self.assertNotExists(repo_path)

        # We will create the DEBUG archive for ubuntutest, so it can

        # be published.

        debug_archive = getUtility(IArchiveSet).new(

            purpose=ArchivePurpose.DEBUG, owner=ubuntutest.owner,

            distribution=ubuntutest)

        # We will also create a source & binary pair of pending

        # publications. Only the DDEB (binary) will be moved to

        # the DEBUG archive, exactly as it would happen in normal

        # operation, see nascentupload-ddebs.txt.

        self.prepareBreezyAutotest()

        pub_binaries = self.getPubBinaries(format=BinaryPackageFormat.DDEB)

        for binary in pub_binaries:

            binary.archive = debug_archive

        # Commit setup changes, so the script can operate on them.

        self.layer.txn.commit()

        # After publication, the DDEB is published and indexed.

        self.runPublishDistro(['--primary-debug'])

        debug_pool_path = os.path.join(repo_path, 'pool/main/f/foo')

        self.assertEqual(

            os.listdir(debug_pool_path), ['foo-bin_666_all.ddeb'])

        debug_index_path = os.path.join(

            repo_path, 'dists/breezy-autotest/main/binary-i386/Packages')

        self.assertEqual(

            open(debug_index_path).readlines()[0], 'Package: foo-bin\n')

    def testPublishCopyArchive(self):

        """Run publish-distro in copy archive mode.

        It should only publish copy archives.

        """

        ubuntutest = getUtility(IDistributionSet)['ubuntutest']

        cprov = getUtility(IPersonSet).getByName('cprov')

        copy_archive_name = 'test-copy-publish'

        # The COPY repository path is not created yet.

        root_dir = getUtility(

            IPublisherConfigSet).getByDistribution(ubuntutest).root_dir

        repo_path = os.path.join(

            root_dir,

            ubuntutest.name + '-' + copy_archive_name,

            ubuntutest.name)

        self.assertNotExists(repo_path)

        copy_archive = getUtility(IArchiveSet).new(

            distribution=ubuntutest, owner=cprov, name=copy_archive_name,

            purpose=ArchivePurpose.COPY, enabled=True)

        # Save some test CPU cycles by avoiding logging in as the user

        # necessary to alter the publish flag.

        removeSecurityProxy(copy_archive).publish = True

        # Publish something.

        pub_source =  self.getPubSource(

            sourcename='baz', filecontent='baz', archive=copy_archive)

        # Try a plain PPA run, to ensure the copy archive is not published.

        self.runPublishDistro(['--ppa'])

        self.assertEqual(pub_source.status, PackagePublishingStatus.PENDING)

        # Now publish the copy archives and make sure they are really

        # published.

        self.runPublishDistro(['--copy-archive'])

        self.assertEqual(pub_source.status, PackagePublishingStatus.PUBLISHED)

        # Make sure that the files were published in the right place.

        pool_path = os.path.join(repo_path, 'pool/main/b/baz/baz_666.dsc')

        self.assertExists(pool_path)

    def testRunWithEmptySuites(self):

        """Try a publish-distro run on empty suites in careful_apt mode

        Expect it to create all indexes, including current 'Release' file

        for the empty suites specified.

        """

        self.runPublishDistro(

            ['-A', '-s', 'hoary-test-updates', '-s', 'hoary-test-backports'])

        # Check "Release" files

        release_path = "%s/hoary-test-updates/Release" % self.config.distsroot

        self.assertExists(release_path)

        release_path = (

            "%s/hoary-test-backports/Release" % self.config.distsroot)

        self.assertExists(release_path)

        release_path = "%s/hoary-test/Release" % self.config.distsroot

        self.assertNotExists(release_path)

        # Check some index files

        index_path = (

            "%s/hoary-test-updates/main/binary-i386/Packages"

            % self.config.distsroot)

        self.assertExists(index_path)

        index_path = (

            "%s/hoary-test-backports/main/binary-i386/Packages"

            % self.config.distsroot)

        self.assertExists(index_path)

        index_path = (

            "%s/hoary-test/main/binary-i386/Packages" % self.config.distsroot)

        self.assertNotExists(index_path)

    def testExclusiveOptions(self):

        """Test that some command line options are mutually exclusive."""

        self.assertRaises(

            LaunchpadScriptFailure,

            self.runPublishDistro,

            ['--ppa', '--partner', '--primary-debug', '--copy-archive'])

        self.assertRaises(

            LaunchpadScriptFailure,

            self.runPublishDistro, ['--ppa', '--partner'])

        self.assertRaises(

            LaunchpadScriptFailure,

            self.runPublishDistro, ['--ppa', '--private-ppa'])

        self.assertRaises(

            LaunchpadScriptFailure,

            self.runPublishDistro, ['--ppa', '--primary-debug'])

        self.assertRaises(

            LaunchpadScriptFailure,

            self.runPublishDistro, ['--ppa', '--copy-archive'])

        self.assertRaises(

            LaunchpadScriptFailure,

            self.runPublishDistro, ['--partner', '--private-ppa'])

        self.assertRaises(

            LaunchpadScriptFailure,

            self.runPublishDistro, ['--partner', '--primary-debug'])

        self.assertRaises(

            LaunchpadScriptFailure,

            self.runPublishDistro, ['--partner', '--copy-archive'])

        self.assertRaises(

            LaunchpadScriptFailure,

            self.runPublishDistro, ['--primary-debug', '--copy-archive'])

def test_suite():

    return unittest.TestLoader().loadTestsFromName(__name__)