~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).

"""Tests for Revisions."""

__metaclass__ = type

from datetime import (

    datetime,

    timedelta,

    )

import time

from unittest import TestCase

import psycopg2

import pytz

from storm.store import Store

from zope.component import getUtility

from zope.security.proxy import removeSecurityProxy

from lp.code.enums import BranchLifecycleStatus

from lp.code.interfaces.branchlookup import IBranchLookup

from lp.code.interfaces.revision import IRevisionSet

from lp.code.model.revision import (

    RevisionCache,

    RevisionSet,

    )

from lp.registry.model.karma import Karma

from lp.scripts.garbo import RevisionAuthorEmailLinker

from lp.services.database.lpstorm import IMasterObject

from lp.services.database.sqlbase import cursor

from lp.services.identity.interfaces.account import AccountStatus

from lp.services.log.logger import DevNullLogger

from lp.services.webapp.interfaces import (

    DEFAULT_FLAVOR,

    IStoreSelector,

    MAIN_STORE,

    )

from lp.testing import (

    login,

    logout,

    TestCaseWithFactory,

    time_counter,

    )

from lp.testing.factory import LaunchpadObjectFactory

from lp.testing.layers import DatabaseFunctionalLayer

class TestRevisionCreationDate(TestCaseWithFactory):

    """Test that RevisionSet.new won't create revisions with future dates."""

    layer = DatabaseFunctionalLayer

    def test_new_past_revision_date(self):

        # A revision created with a revision date in the past works fine.

        past_date = datetime(2009, 1, 1, tzinfo=pytz.UTC)

        revision = RevisionSet().new(

            'rev_id', 'log body', past_date, 'author', [], {})

        self.assertEqual(past_date, revision.revision_date)

    def test_new_future_revision_date(self):

        # A revision with a future date gets the revision date set to

        # date_created.

        now = datetime.now(pytz.UTC)

        future_date = now + timedelta(days=1)

        revision = RevisionSet().new(

            'rev_id', 'log body', future_date, 'author', [], {})

        self.assertEqual(revision.date_created, revision.revision_date)

        self.assertTrue(revision.revision_date <= now)

class TestRevisionKarma(TestCaseWithFactory):

    """Test the allocation of karma for revisions."""

    layer = DatabaseFunctionalLayer

    def setUp(self):

        # Use an administrator to set branch privacy easily.

        TestCaseWithFactory.setUp(self, "admin@canonical.com")

    def test_revisionWithUnknownEmail(self):

        # A revision when created does not have karma allocated.

        rev = self.factory.makeRevision()

        self.failIf(rev.karma_allocated)

        # Even if the revision author is someone we know.

        author = self.factory.makePerson()

        rev = self.factory.makeRevision(

            author=author.preferredemail.email)

        self.failIf(rev.karma_allocated)

    def test_noKarmaForUnknownAuthor(self):

        # If the revision author is unknown, karma isn't allocated.

        rev = self.factory.makeRevision()

        branch = self.factory.makeProductBranch()

        branch.createBranchRevision(1, rev)

        self.failIf(rev.karma_allocated)

    def test_noRevisionsNeedingAllocation(self):

        # There are no outstanding revisions needing karma allocated.

        self.assertEqual(

            [], list(RevisionSet.getRevisionsNeedingKarmaAllocated()))

    def test_karmaAllocatedForKnownAuthor(self):

        # If the revision author is known, allocate karma.

        author = self.factory.makePerson()

        rev = self.factory.makeRevision(

            author=author.preferredemail.email,

            revision_date=datetime.now(pytz.UTC) - timedelta(days=5))

        branch = self.factory.makeProductBranch()

        branch.createBranchRevision(1, rev)

        self.failUnless(rev.karma_allocated)

        # Get the karma event.

        [karma] = list(Store.of(author).find(

            Karma,

            Karma.person == author,

            Karma.product == branch.product))

        self.assertEqual(karma.datecreated, rev.revision_date)

        self.assertEqual(karma.product, branch.product)

        # Since karma has been allocated, the revision isn't in our list.

        self.assertEqual(

            [], list(RevisionSet.getRevisionsNeedingKarmaAllocated()))

    def test_karmaNotAllocatedForKnownAuthorWithInactiveAccount(self):

        # If the revision author is known, but the account is not active,

        # don't allocate karma.

        author = self.factory.makePerson()

        rev = self.factory.makeRevision(

            author=author.preferredemail.email)

        IMasterObject(author.account).status = AccountStatus.SUSPENDED

        branch = self.factory.makeProductBranch()

        branch.createBranchRevision(1, rev)

        self.failIf(rev.karma_allocated)

        # Even though the revision author is connected to the person, since

        # the account status is suspended, the person is not "valid", and so

        # the revisions are not returned as needing karma allocated.

        self.assertEqual(

            [], list(RevisionSet.getRevisionsNeedingKarmaAllocated()))

    def test_noKarmaForJunk(self):

        # Revisions only associated with junk branches don't get karma.

        author = self.factory.makePerson()

        rev = self.factory.makeRevision(

            author=author.preferredemail.email)

        branch = self.factory.makePersonalBranch()

        branch.createBranchRevision(1, rev)

        self.failIf(rev.karma_allocated)

        # Nor is this revision identified as needing karma allocated.

        self.assertEqual(

            [], list(RevisionSet.getRevisionsNeedingKarmaAllocated()))

    def test_junkBranchMovedToProductNeedsKarma(self):

        # A junk branch that moves to a product needs karma allocated.

        author = self.factory.makePerson()

        rev = self.factory.makeRevision(author=author)

        branch = self.factory.makePersonalBranch()

        branch.createBranchRevision(1, rev)

        # Once the branch is connected to the revision, we now specify

        # a product for the branch.

        project = self.factory.makeProduct()

        branch.setTarget(user=branch.owner, project=project)

        # The revision is now identified as needing karma allocated.

        self.assertEqual(

            [rev], list(RevisionSet.getRevisionsNeedingKarmaAllocated()))

    def test_junkBranchMovedToPackageNeedsKarma(self):

        # A junk branch that moves to a package needs karma allocated.

        author = self.factory.makePerson()

        rev = self.factory.makeRevision(author=author)

        branch = self.factory.makePersonalBranch()

        branch.createBranchRevision(1, rev)

        # Once the branch is connected to the revision, we now specify

        # a product for the branch.

        source_package = self.factory.makeSourcePackage()

        branch.setTarget(user=branch.owner, source_package=source_package)

        # The revision is now identified as needing karma allocated.

        self.assertEqual(

            [rev], list(RevisionSet.getRevisionsNeedingKarmaAllocated()))

    def test_newRevisionAuthorLinkNeedsKarma(self):

        # If Launchpad knows of revisions by a particular author, and later

        # that authoer registers with launchpad, the revisions need karma

        # allocated.

        email = self.factory.getUniqueEmailAddress()

        rev = self.factory.makeRevision(author=email)

        branch = self.factory.makeProductBranch()

        branch.createBranchRevision(1, rev)

        self.failIf(rev.karma_allocated)

        # Since the revision author is not known, the revisions do not at this

        # stage need karma allocated.

        self.assertEqual(

            [], list(RevisionSet.getRevisionsNeedingKarmaAllocated()))

        # The person registers with Launchpad.

        self.factory.makePerson(email=email)

        # Garbo runs the RevisionAuthorEmailLinker job.

        RevisionAuthorEmailLinker(log=DevNullLogger()).run()

        # Now the kama needs allocating.

        self.assertEqual(

            [rev], list(RevisionSet.getRevisionsNeedingKarmaAllocated()))

    def test_karmaDateForFutureRevisions(self):

        # If the revision date is some time in the future, then the karma date

        # is set to be the time that the revision was created.

        author = self.factory.makePerson()

        rev = self.factory.makeRevision(

            author=author,

            revision_date=datetime.now(pytz.UTC) + timedelta(days=5))

        branch = self.factory.makeProductBranch()

        karma = rev.allocateKarma(branch)

        self.assertEqual(karma.datecreated, rev.date_created)

    def test_allocateKarma_personal_branch(self):

        # A personal branch gets no karma event.

        author = self.factory.makePerson()

        rev = self.factory.makeRevision(author=author)

        branch = self.factory.makePersonalBranch()

        karma = rev.allocateKarma(branch)

        self.assertIs(None, karma)

    def test_allocateKarma_package_branch(self):

        # A revision on a package branch gets karma.

        author = self.factory.makePerson()

        rev = self.factory.makeRevision(author=author)

        branch = self.factory.makePackageBranch()

        karma = rev.allocateKarma(branch)

        self.assertEqual(author, karma.person)

        self.assertEqual(branch.distribution, karma.distribution)

        self.assertEqual(branch.sourcepackagename, karma.sourcepackagename)

    def test_allocateKarma_product_branch(self):

        # A revision on a product branch gets karma.

        author = self.factory.makePerson()

        rev = self.factory.makeRevision(author=author)

        branch = self.factory.makeProductBranch()

        karma = rev.allocateKarma(branch)

        self.assertEqual(author, karma.person)

        self.assertEqual(branch.product, karma.product)

class TestRevisionSet(TestCaseWithFactory):

    """Tests for IRevisionSet."""

    layer = DatabaseFunctionalLayer

    def setUp(self):

        super(TestRevisionSet, self).setUp()

        self.revision_set = getUtility(IRevisionSet)

    def test_getRevisionById_existing(self):

        # IRevisionSet.getByRevisionId returns the revision with that id.

        revision = self.factory.makeRevision()

        found = self.revision_set.getByRevisionId(revision.revision_id)

        self.assertEquals(revision, found)

    def test_getRevisionById_nonexistent(self):

        # IRevisionSet.getByRevisionId returns None if there is no revision

        # with that id.

        found = self.revision_set.getByRevisionId('nonexistent')

        self.assertIs(None, found)

class TestRevisionGetBranch(TestCaseWithFactory):

    """Test the `getBranch` method of the revision."""

    layer = DatabaseFunctionalLayer

    def setUp(self):

        # Use an administrator to set branch privacy easily.

        TestCaseWithFactory.setUp(self, "admin@canonical.com")

        self.author = self.factory.makePerson()

        self.revision = self.factory.makeRevision(

            author=self.author.preferredemail.email)

    def makeBranchWithRevision(self, sequence, **kwargs):

        branch = self.factory.makeAnyBranch(**kwargs)

        branch.createBranchRevision(sequence, self.revision)

        return branch

    def testPreferAuthorBranch(self):

        # If a revision is on the mainline history of two (or more) different

        # branches, then choose one owned by the revision author.

        self.makeBranchWithRevision(1)

        b = self.makeBranchWithRevision(1, owner=self.author)

        self.assertEqual(b, self.revision.getBranch())

    def testPreferMainlineRevisionBranch(self):

        # Choose a branch where the revision is on the mainline history over a

        # branch where the revision is just in the ancestry.

        self.makeBranchWithRevision(None)

        b = self.makeBranchWithRevision(1)

        self.assertEqual(b, self.revision.getBranch())

    def testOwnerTrunksMainline(self):

        # If the revision is mainline on a branch not owned by the revision

        # owner, but in the ancestry of a branch owned by the revision owner,

        # choose the branch owned by the revision author.

        self.makeBranchWithRevision(1)

        b = self.makeBranchWithRevision(None, owner=self.author)

        self.assertEqual(b, self.revision.getBranch())

    def testPublicBranchTrumpsOwner(self):

        # Only public branches are returned.

        b1 = self.makeBranchWithRevision(1)

        b2 = self.makeBranchWithRevision(1, owner=self.author)

        removeSecurityProxy(b2).explicitly_private = True

        self.assertEqual(b1, self.revision.getBranch())

    def testAllowPrivateReturnsPrivateBranch(self):

        # If the allow_private flag is set, then private branches can be

        # returned if they are the best match.

        self.makeBranchWithRevision(1)

        b2 = self.makeBranchWithRevision(1, owner=self.author)

        removeSecurityProxy(b2).explicitly_private = True

        self.assertEqual(b2, self.revision.getBranch(allow_private=True))

    def testAllowPrivateCanReturnPublic(self):

        # Allowing private branches does not change the priority ordering of

        # the branches.

        b1 = self.makeBranchWithRevision(1)

        b2 = self.makeBranchWithRevision(1, owner=self.author)

        removeSecurityProxy(b1).explicitly_private = True

        self.assertEqual(b2, self.revision.getBranch(allow_private=True))

    def testGetBranchNotJunk(self):

        # If allow_junk is set to False, then branches without products are

        # not returned.

        b1 = self.factory.makeProductBranch()

        b1.createBranchRevision(1, self.revision)

        b2 = self.factory.makePersonalBranch(owner=self.author)

        b2.createBranchRevision(1, self.revision)

        self.assertEqual(

            b1, self.revision.getBranch(allow_private=True, allow_junk=False))

    def testGetBranchSourcePackage(self):

        # Branches targetting source packages are not junk.

        b1 = self.factory.makePackageBranch()

        b1.createBranchRevision(1, self.revision)

        b2 = self.factory.makePersonalBranch(owner=self.author)

        b2.createBranchRevision(1, self.revision)

        self.assertEqual(

            b1, self.revision.getBranch(allow_private=True, allow_junk=False))

class GetPublicRevisionsTestCase(TestCaseWithFactory):

    """A base class for the tests for people, products and projects."""

    layer = DatabaseFunctionalLayer

    def setUp(self):

        # Use an administrator to set branch privacy easily.

        TestCaseWithFactory.setUp(self, "admin@canonical.com")

        # Since the tests order by date, but also limit to the last 30

        # days, we want a time counter that starts 10 days ago.

        self.date_generator = time_counter(

            datetime.now(pytz.UTC) - timedelta(days=10),

            delta=timedelta(days=1))

    def _makeRevision(self, revision_date=None):

        """Make a revision using the date generator."""

        if revision_date is None:

            revision_date = self.date_generator.next()

        return self.factory.makeRevision(

            revision_date=revision_date)

    def _addRevisionsToBranch(self, branch, *revs):

        # Add the revisions to the branch.

        for sequence, rev in enumerate(revs):

            branch.createBranchRevision(sequence, rev)

    def _makeBranch(self, product=None):

        # Make a branch.

        if product is None:

            # If the test defines a product, use that, otherwise

            # have the factory generate one.

            product = getattr(self, 'product', None)

        return self.factory.makeProductBranch(product=product)

    def _makeRevisionInBranch(self, product=None):

        # Make a revision, and associate it with a branch.  The branch is made

        # with the product passed in, which means that if there was no product

        # passed in, the factory makes a new one.

        branch = self.factory.makeProductBranch(product=product)

        rev = self._makeRevision()

        branch.createBranchRevision(1, rev)

        return rev

    def _getRevisions(self, day_limit=30):

        raise NotImplementedError('_getRevisions')

class RevisionTestMixin:

    """Common tests for the different GetPublicRevision test cases."""

    def testNewestRevisionFirst(self):

        # The revisions are ordered with the newest first.

        rev1 = self._makeRevision()

        rev2 = self._makeRevision()

        rev3 = self._makeRevision()

        self._addRevisionsToBranch(self._makeBranch(), rev1, rev2, rev3)

        self.assertEqual([rev3, rev2, rev1], self._getRevisions())

    def testRevisionsOnlyReturnedOnce(self):

        # If the revisions appear in multiple branches, they are only returned

        # once.

        rev1 = self._makeRevision()

        rev2 = self._makeRevision()

        rev3 = self._makeRevision()

        self._addRevisionsToBranch(

            self._makeBranch(), rev1, rev2, rev3)

        self._addRevisionsToBranch(

            self._makeBranch(), rev1, rev2, rev3)

        self.assertEqual([rev3, rev2, rev1], self._getRevisions())

    def testRevisionsMustBeInABranch(self):

        # A revision authored by the person must be in a branch to be

        # returned.

        rev1 = self._makeRevision()

        self.assertEqual([], self._getRevisions())

        b = self._makeBranch()

        b.createBranchRevision(1, rev1)

        self.assertEqual([rev1], self._getRevisions())

    def testRevisionsMustBeInAPublicBranch(self):

        # A revision authored by the person must be in a branch to be

        # returned.

        rev1 = self._makeRevision()

        b = self._makeBranch()

        b.createBranchRevision(1, rev1)

        removeSecurityProxy(b).explicitly_private = True

        self.assertEqual([], self._getRevisions())

    def testRevisionDateRange(self):

        # Revisions where the revision_date is older than the day_limit are

        # not returned.

        now = datetime.now(pytz.UTC)

        day_limit = 5

        # Make the first revision earlier than our day limit.

        rev1 = self._makeRevision(

            revision_date=(now - timedelta(days=(day_limit + 2))))

        # The second one is just two days ago.

        rev2 = self._makeRevision(

            revision_date=(now - timedelta(days=2)))

        self._addRevisionsToBranch(self._makeBranch(), rev1, rev2)

        self.assertEqual([rev2], self._getRevisions(day_limit))

class TestGetPublicRevisionsForPerson(GetPublicRevisionsTestCase,

                                      RevisionTestMixin):

    """Test the `getPublicRevisionsForPerson` method of `RevisionSet`."""

    def setUp(self):

        GetPublicRevisionsTestCase.setUp(self)

        self.author = self.factory.makePerson()

        self.revision = self.factory.makeRevision(

            author=self.author.preferredemail.email)

    def _getRevisions(self, day_limit=30):

        # Returns the revisions for the person.

        return list(RevisionSet.getPublicRevisionsForPerson(

                self.author, day_limit))

    def _makeRevision(self, author=None, revision_date=None):

        """Make a revision owned by `author`.

        `author` defaults to self.author if not set."""

        if revision_date is None:

            revision_date = self.date_generator.next()

        if author is None:

            author = self.author

        return self.factory.makeRevision(

            author=author.preferredemail.email,

            revision_date=revision_date)

    def testTeamRevisions(self):

        # Revisions owned by all members of a team are returned.

        team = self.factory.makeTeam(self.author)

        team_member = self.factory.makePerson()

        team.addMember(team_member, self.author)

        rev1 = self._makeRevision()

        rev2 = self._makeRevision(team_member)

        rev3 = self._makeRevision(self.factory.makePerson())

        branch = self.factory.makeAnyBranch()

        self._addRevisionsToBranch(branch, rev1, rev2, rev3)

        self.assertEqual([rev2, rev1],

                         list(RevisionSet.getPublicRevisionsForPerson(team)))

class TestGetPublicRevisionsForProduct(GetPublicRevisionsTestCase,

                                       RevisionTestMixin):

    """Test the `getPublicRevisionsForProduct` method of `RevisionSet`."""

    def setUp(self):

        GetPublicRevisionsTestCase.setUp(self)

        self.product = self.factory.makeProduct()

    def _getRevisions(self, day_limit=30):

        # Returns the revisions for the person.

        return list(RevisionSet.getPublicRevisionsForProduct(

                self.product, day_limit))

    def testRevisionsMustBeInABranchOfProduct(self):

        # The revision must be in a branch for the product.

        # returned.

        rev1 = self._makeRevisionInBranch(product=self.product)

        self._makeRevisionInBranch()

        self.assertEqual([rev1], self._getRevisions())

class TestGetPublicRevisionsForProjectGroup(GetPublicRevisionsTestCase,

                                            RevisionTestMixin):

    """Test the `getPublicRevisionsForProjectGroup` method of `RevisionSet`.

    """

    def setUp(self):

        GetPublicRevisionsTestCase.setUp(self)

        self.project = self.factory.makeProject()

        self.product = self.factory.makeProduct(project=self.project)

    def _getRevisions(self, day_limit=30):

        # Returns the revisions for the person.

        return list(RevisionSet.getPublicRevisionsForProjectGroup(

                self.project, day_limit))

    def testRevisionsMustBeInABranchOfProduct(self):

        # The revision must be in a branch for the product.

        # returned.

        rev1 = self._makeRevisionInBranch(product=self.product)

        self._makeRevisionInBranch()

        self.assertEqual([rev1], self._getRevisions())

    def testProjectRevisions(self):

        # Revisions in all products that are part of the project are returned.

        another_product = self.factory.makeProduct(project=self.project)

        rev1 = self._makeRevisionInBranch(product=self.product)

        rev2 = self._makeRevisionInBranch(product=another_product)

        self._makeRevisionInBranch()

        self.assertEqual([rev2, rev1], self._getRevisions())

class TestGetRecentRevisionsForProduct(GetPublicRevisionsTestCase):

    """Test the `getRecentRevisionsForProduct` method of `RevisionSet`."""

    def setUp(self):

        GetPublicRevisionsTestCase.setUp(self)

        self.product = self.factory.makeProduct()

    def _getRecentRevisions(self, day_limit=30):

        # Return a list of the recent revisions and revision authors.

        return list(RevisionSet.getRecentRevisionsForProduct(

                self.product, day_limit))

    def testRevisionAuthorMatchesRevision(self):

        # The revision author returned with the revision is the same as the

        # author for the revision.

        self._makeRevisionInBranch(product=self.product)

        results = self._getRecentRevisions()

        [(revision, revision_author)] = results

        self.assertEqual(revision.revision_author, revision_author)

    def testRevisionsMustBeInABranchOfProduct(self):

        # The revisions returned revision must be in a branch for the product.

        rev1 = self._makeRevisionInBranch(product=self.product)

        self._makeRevisionInBranch()

        self.assertEqual(

            [(rev1, rev1.revision_author)], self._getRecentRevisions())

    def testRevisionsMustBeInActiveBranches(self):

        # The revisions returned revision must be in a branch for the product.

        rev1 = self._makeRevisionInBranch(product=self.product)

        branch = self.factory.makeProductBranch(

            product=self.product,

            lifecycle_status=BranchLifecycleStatus.MERGED)

        branch.createBranchRevision(1, self._makeRevision())

        self.assertEqual([(rev1, rev1.revision_author)],

                         self._getRecentRevisions())

    def testRevisionDateRange(self):

        # Revisions where the revision_date is older than the day_limit are

        # not returned.

        now = datetime.now(pytz.UTC)

        day_limit = 5

        # Make the first revision earlier than our day limit.

        rev1 = self._makeRevision(

            revision_date=(now - timedelta(days=(day_limit + 2))))

        # The second one is just two days ago.

        rev2 = self._makeRevision(revision_date=(now - timedelta(days=2)))

        self._addRevisionsToBranch(self._makeBranch(), rev1, rev2)

        self.assertEqual([(rev2, rev2.revision_author)],

                         self._getRecentRevisions(day_limit))

class TestTipRevisionsForBranches(TestCase):

    """Test that the tip revisions get returned properly."""

    layer = DatabaseFunctionalLayer

    def setUp(self):

        login('test@canonical.com')

        factory = LaunchpadObjectFactory()

        branches = [factory.makeAnyBranch() for count in range(5)]

        branch_ids = [branch.id for branch in branches]

        for branch in branches:

            factory.makeRevisionsForBranch(branch)

        # Retrieve the updated branches (due to transaction boundaries).

        branch_set = getUtility(IBranchLookup)

        self.branches = [branch_set.get(id) for id in branch_ids]

        self.revision_set = getUtility(IRevisionSet)

    def tearDown(self):

        logout()

    def _breakTransaction(self):

        # make sure the current transaction can not be committed by

        # sending a broken SQL statement to the database

        try:

            cursor().execute('break this transaction')

        except psycopg2.DatabaseError:

            pass

    def testNoBranches(self):

        """Assert that when given an empty list, an empty list is returned."""