launchpad-reviewers team mailing list archive
-
launchpad-reviewers team
-
Mailing list archive
-
Message #22286
[Merge] lp:~cjwatson/launchpad/remove-branch-distro into lp:launchpad
Colin Watson has proposed merging lp:~cjwatson/launchpad/remove-branch-distro into lp:launchpad.
Commit message:
Remove branch-distro, which we haven't used for some time.
Requested reviews:
Launchpad code reviewers (launchpad-reviewers)
For more details, see:
https://code.launchpad.net/~cjwatson/launchpad/remove-branch-distro/+merge/342065
With git-ubuntu building up steam, it seems unlikely that this will be resurrected, so we probably don't need to keep this around.
--
Your team Launchpad code reviewers is requested to review the proposed merge of lp:~cjwatson/launchpad/remove-branch-distro into lp:launchpad.
=== modified file 'database/schema/security.cfg'
--- database/schema/security.cfg 2017-05-25 17:22:38 +0000
+++ database/schema/security.cfg 2018-03-25 15:48:42 +0000
@@ -715,30 +715,6 @@
public.xref = SELECT, INSERT, DELETE
type=user
-[branch-distro]
-public.accessartifact = SELECT, INSERT, DELETE
-public.accesspolicy = SELECT
-public.accesspolicyartifact = SELECT, INSERT, DELETE
-public.accesspolicygrant = SELECT
-public.accessartifactgrant = SELECT
-public.branch = SELECT, INSERT, UPDATE
-public.branchrevision = SELECT, INSERT
-public.branchsubscription = SELECT, INSERT
-public.distribution = SELECT
-public.distributionsourcepackage = SELECT, INSERT, UPDATE, DELETE
-public.distributionsourcepackagecache = SELECT, INSERT
-public.distroseries = SELECT
-public.karma = SELECT, INSERT
-public.karmaaction = SELECT
-public.person = SELECT
-public.product = SELECT
-public.revision = SELECT
-public.seriessourcepackagebranch = SELECT, INSERT, DELETE
-public.sourcepackagename = SELECT
-public.teamparticipation = SELECT
-public.validpersoncache = SELECT
-type=user
-
[targetnamecacheupdater]
groups=script
public.binarypackagename = SELECT
=== removed file 'lib/lp/codehosting/branchdistro.py'
--- lib/lp/codehosting/branchdistro.py 2013-06-20 05:50:00 +0000
+++ lib/lp/codehosting/branchdistro.py 1970-01-01 00:00:00 +0000
@@ -1,385 +0,0 @@
-# Copyright 2009 Canonical Ltd. This software is licensed under the
-# GNU Affero General Public License version 3 (see the file LICENSE).
-
-"""Opening a new DistroSeries for branch based development.
-
-Intended to be run just after a new distro series has been completed, this
-script will create an official package branch in the new series for every one
-in the old. The old branch will become stacked on the new, to avoid a using
-too much disk space whilst retaining best performance for the new branch.
-"""
-
-__metaclass__ = type
-__all__ = [
- 'DistroBrancher',
- 'switch_branches',
- ]
-
-import os
-
-from bzrlib.branch import Branch
-from bzrlib.bzrdir import BzrDir
-from bzrlib.errors import (
- NotBranchError,
- NotStacked,
- )
-from bzrlib.revision import NULL_REVISION
-import transaction
-from zope.component import getUtility
-
-from lp.code.enums import (
- BranchLifecycleStatus,
- BranchType,
- )
-from lp.code.errors import BranchExists
-from lp.code.interfaces.branchcollection import IAllBranches
-from lp.code.interfaces.branchnamespace import IBranchNamespaceSet
-from lp.code.interfaces.seriessourcepackagebranch import (
- IFindOfficialBranchLinks,
- )
-from lp.code.model.branchrevision import BranchRevision
-from lp.codehosting.vfs import branch_id_to_path
-from lp.registry.interfaces.distribution import IDistributionSet
-from lp.registry.interfaces.pocket import PackagePublishingPocket
-from lp.services.config import config
-from lp.services.database.interfaces import IMasterStore
-
-
-def switch_branches(prefix, scheme, old_db_branch, new_db_branch):
- """Move bzr data from an old to a new branch, leaving old stacked on new.
-
- This function is intended to be used just after Ubuntu is released to
- create (at the bzr level) a new trunk branch for a source package for the
- next release of the distribution. We move the bzr data to the location
- for the new branch and replace the trunk branch for the just released
- version with a stacked branch pointing at the new branch.
-
- The procedure is to complicated to be carried out atomically, so if this
- function is interrupted things may be a little inconsistent (e.g. there
- might be a branch in the old location, but not stacked on the new location
- yet). There should be no data loss though.
-
- :param prefix: The non-branch id dependent part of the physical path to
- the branches on disk.
- :param scheme: The branches should be open-able at a URL of the form
- ``scheme + :/// + unique_name``.
- :param old_db_branch: The branch that currently has the trunk bzr data.
- :param old_db_branch: The new trunk branch. This should not have any
- presence on disk yet.
- """
- # Move .bzr directory from old to new location, crashing through the
- # abstraction we usually hide our branch locations behind.
- old_underlying_path = os.path.join(
- prefix, branch_id_to_path(old_db_branch.id))
- new_underlying_path = os.path.join(
- prefix, branch_id_to_path(new_db_branch.id))
- os.makedirs(new_underlying_path)
- os.rename(
- os.path.join(old_underlying_path, '.bzr'),
- os.path.join(new_underlying_path, '.bzr'))
-
- # Create branch at old location -- we use the "clone('null:')" trick to
- # preserve the format. We have to open at the logical, unique_name-based,
- # location so that it works to set the stacked on url to '/' + a
- # unique_name.
- new_location_bzrdir = BzrDir.open(
- scheme + ':///' + new_db_branch.unique_name)
- old_location_bzrdir = new_location_bzrdir.clone(
- scheme + ':///' + old_db_branch.unique_name, revision_id='null:')
-
- # Set the stacked on url for old location.
- old_location_branch = old_location_bzrdir.open_branch()
- old_location_branch.set_stacked_on_url('/' + new_db_branch.unique_name)
-
- # Pull from new location to old -- this won't actually transfer any
- # revisions, just update the last revision pointer.
- old_location_branch.pull(new_location_bzrdir.open_branch())
-
-
-class DistroBrancher:
- """Open a new distroseries for branch based development.
-
- `makeNewBranches` will create an official package branch in the new series
- for every one in the old. `checkNewBranches` will check that a previous
- run of this script completed successfully -- this is only likely to be
- really useful if a script run died halfway through or had to be killed.
- """
-
- def __init__(self, logger, old_distroseries, new_distroseries):
- """Construct a `DistroBrancher`.
-
- The old and new distroseries must be from the same distribution, but
- not the same distroseries.
-
- :param logger: A Logger. Problems will be logged to this object at
- the WARNING level or higher; progress reports will be logged at
- the DEBUG level.
- :param old_distroseries: The distroseries that will be examined to
- find existing source package branches.
- :param new_distroseries: The distroseries that will have new official
- source branches made for it.
- """
- self.logger = logger
- if old_distroseries.distribution != new_distroseries.distribution:
- raise AssertionError(
- "%s and %s are from different distributions!" %
- (old_distroseries, new_distroseries))
- if old_distroseries == new_distroseries:
- raise AssertionError(
- "New and old distributions must be different!")
- self.old_distroseries = old_distroseries
- self.new_distroseries = new_distroseries
-
- @classmethod
- def fromNames(cls, logger, distribution_name, old_distroseries_name,
- new_distroseries_name):
- """Make a `DistroBrancher` from the names of a distro and two series.
- """
- distribution = getUtility(IDistributionSet).getByName(
- distribution_name)
- new_distroseries = distribution.getSeries(new_distroseries_name)
- old_distroseries = distribution.getSeries(old_distroseries_name)
- return cls(logger, old_distroseries, new_distroseries)
-
- def _existingOfficialBranches(self):
- """Return the collection of official branches in the old distroseries.
- """
- branches = getUtility(IAllBranches)
- distroseries_branches = branches.inDistroSeries(self.old_distroseries)
- return distroseries_branches.officialBranches().getBranches(
- eager_load=False)
-
- def checkConsistentOfficialPackageBranch(self, db_branch):
- """Check that `db_branch` is a consistent official package branch.
-
- 'Consistent official package branch' means:
-
- * It's a package branch (rather than a personal or junk branch).
- * It's official for its SourcePackage and no other.
-
- This function simply returns True or False -- any problems will be
- logged to ``self.logger``.
-
- :param db_branch: The `IBranch` to check.
- :return: ``True`` if the branch is a consistent official package
- branch, ``False`` otherwise.
- """
- if db_branch.product:
- self.logger.warning(
- "Encountered unexpected product branch %r",
- db_branch.unique_name)
- return False
- if not db_branch.distroseries:
- self.logger.warning(
- "Encountered unexpected personal branch %s",
- db_branch.unique_name)
- return False
- find_branch_links = getUtility(IFindOfficialBranchLinks)
- links = list(find_branch_links.findForBranch(db_branch))
- if len(links) == 0:
- self.logger.warning(
- "%s is not an official branch", db_branch.unique_name)
- return False
- elif len(links) > 1:
- series_text = ', '.join([
- link.sourcepackage.path for link in links])
- self.logger.warning(
- "%s is official for multiple series: %s",
- db_branch.unique_name, series_text)
- return False
- elif links[0].sourcepackage != db_branch.sourcepackage:
- self.logger.warning(
- "%s is the official branch for %s but not its "
- "sourcepackage", db_branch.unique_name,
- links[0].sourcepackage.path)
- return False
- return True
-
- def makeNewBranches(self):
- """Make official branches in the new distroseries."""
- for db_branch in self._existingOfficialBranches():
- self.logger.debug("Processing %s" % db_branch.unique_name)
- try:
- self.makeOneNewBranch(db_branch)
- except BranchExists:
- pass
-
- def checkNewBranches(self):
- """Check the branches in the new distroseries are present and correct.
-
- This function checks that every official package branch in the old
- distroseries has a matching branch in the new distroseries and that
- stacking is set up as we expect on disk.
-
- Every branch will be checked, even if some fail.
-
- This function simply returns True or False -- any problems will be
- logged to ``self.logger``.
-
- :return: ``True`` if every branch passes the check, ``False``
- otherwise.
- """
- ok = True
- for db_branch in self._existingOfficialBranches():
- self.logger.debug("Checking %s" % db_branch.unique_name)
- try:
- if not self.checkOneBranch(db_branch):
- ok = False
- except:
- ok = False
- self.logger.exception(
- "Unexpected error checking %s!", db_branch)
- return ok
-
- def checkOneBranch(self, old_db_branch):
- """Check a branch in the old distroseries has been copied to the new.
-
- This function checks that `old_db_branch` has a matching branch in the
- new distroseries and that stacking is set up as we expect on disk.
-
- This function simply returns True or False -- any problems will be
- logged to ``self.logger``.
-
- :param old_db_branch: The branch to check.
- :return: ``True`` if the branch passes the check, ``False`` otherwise.
- """
- ok = self.checkConsistentOfficialPackageBranch(old_db_branch)
- if not ok:
- return ok
- new_sourcepackage = self.new_distroseries.getSourcePackage(
- old_db_branch.sourcepackagename)
- new_db_branch = new_sourcepackage.getBranch(
- PackagePublishingPocket.RELEASE)
- if new_db_branch is None:
- self.logger.warning(
- "No official branch found for %s",
- new_sourcepackage.path)
- return False
- ok = self.checkConsistentOfficialPackageBranch(new_db_branch)
- if not ok:
- return ok
- # the branch in the new distroseries is unstacked
- new_location = 'lp-internal:///' + new_db_branch.unique_name
- try:
- new_bzr_branch = Branch.open(new_location)
- except NotBranchError:
- self.logger.warning(
- "No bzr branch at new location %s", new_location)
- ok = False
- else:
- try:
- new_stacked_on_url = new_bzr_branch.get_stacked_on_url()
- ok = False
- self.logger.warning(
- "New branch at %s is stacked on %s, should be "
- "unstacked.", new_location, new_stacked_on_url)
- except NotStacked:
- pass
- # The branch in the old distroseries is stacked on that in the
- # new.
- old_location = 'lp-internal:///' + old_db_branch.unique_name
- try:
- old_bzr_branch = Branch.open(old_location)
- except NotBranchError:
- self.logger.warning(
- "No bzr branch at old location %s", old_location)
- ok = False
- else:
- try:
- old_stacked_on_url = old_bzr_branch.get_stacked_on_url()
- if old_stacked_on_url != '/' + new_db_branch.unique_name:
- self.logger.warning(
- "Old branch at %s is stacked on %s, should be "
- "stacked on %s", old_location, old_stacked_on_url,
- '/' + new_db_branch.unique_name)
- ok = False
- except NotStacked:
- self.logger.warning(
- "Old branch at %s is not stacked, should be stacked "
- "on %s", old_location,
- '/' + new_db_branch.unique_name)
- ok = False
- # The branch in the old distroseries has no revisions in its
- # repository. We open the repository independently of the
- # branch because the branch's repository has had its fallback
- # location activated. Note that this check might fail if new
- # revisions get pushed to the branch in the old distroseries,
- # which shouldn't happen but isn't totally impossible.
- old_repo = BzrDir.open(old_location).open_repository()
- if len(old_repo.all_revision_ids()) > 0:
- self.logger.warning(
- "Repository at %s has %s revisions.",
- old_location, len(old_repo.all_revision_ids()))
- ok = False
- # The branch in the old distroseries has at least some
- # history. (We can't check that the tips are the same because
- # the branch in the new distroseries might have new revisons).
- if old_bzr_branch.last_revision() == 'null:':
- self.logger.warning(
- "Old branch at %s has null tip revision.",
- old_location)
- ok = False
- return ok
-
- def makeOneNewBranch(self, old_db_branch):
- """Copy a branch to the new distroseries.
-
- This function makes a new database branch for the same source package
- as old_db_branch but in the new distroseries and then uses
- `switch_branches` to move the underlying bzr branch to the new series
- and replace the old branch with a branch stacked on the new series'
- branch.
-
- :param old_db_branch: The branch to copy into the new distroseries.
- :raises BranchExists: This will be raised if old_db_branch has already
- been copied to the new distroseries (in the database, at least).
- """
- if not self.checkConsistentOfficialPackageBranch(old_db_branch):
- self.logger.warning("Skipping branch")
- return
- new_namespace = getUtility(IBranchNamespaceSet).get(
- person=old_db_branch.owner, product=None,
- distroseries=self.new_distroseries,
- sourcepackagename=old_db_branch.sourcepackagename)
- new_db_branch = new_namespace.createBranch(
- BranchType.HOSTED, self.new_distroseries.name,
- old_db_branch.registrant)
- new_db_branch.sourcepackage.setBranch(
- PackagePublishingPocket.RELEASE, new_db_branch,
- new_db_branch.owner)
- old_db_branch.lifecycle_status = BranchLifecycleStatus.MATURE
- # switch_branches *moves* the data to locations dependent on the
- # new_branch's id, so if the transaction was rolled back we wouldn't
- # know the branch id and thus wouldn't be able to find the branch data
- # again. So commit before doing that.
- transaction.commit()
- switch_branches(
- config.codehosting.mirrored_branches_root,
- 'lp-internal', old_db_branch, new_db_branch)
- # Directly copy the branch revisions from the old branch to the new
- # branch.
- store = IMasterStore(BranchRevision)
- store.execute(
- """
- INSERT INTO BranchRevision (branch, revision, sequence)
- SELECT %s, BranchRevision.revision, BranchRevision.sequence
- FROM BranchRevision
- WHERE branch = %s
- """ % (new_db_branch.id, old_db_branch.id))
-
- # Update the scanned details first, that way when hooking into
- # branchChanged, it won't try to create a new scan job.
- tip_revision = old_db_branch.getTipRevision()
- new_db_branch.updateScannedDetails(
- tip_revision, old_db_branch.revision_count)
- tip_revision_id = (
- tip_revision.revision_id if tip_revision is not None else
- NULL_REVISION)
- new_db_branch.branchChanged(
- '', tip_revision_id,
- old_db_branch.control_format,
- old_db_branch.branch_format,
- old_db_branch.repository_format)
- old_db_branch.stacked_on = new_db_branch
- transaction.commit()
- return new_db_branch
=== removed file 'lib/lp/codehosting/tests/test_branchdistro.py'
--- lib/lp/codehosting/tests/test_branchdistro.py 2017-11-06 09:32:45 +0000
+++ lib/lp/codehosting/tests/test_branchdistro.py 1970-01-01 00:00:00 +0000
@@ -1,642 +0,0 @@
-# Copyright 2009-2011 Canonical Ltd. This software is licensed under the
-# GNU Affero General Public License version 3 (see the file LICENSE).
-
-"""Tests for making new source package branches just after a distro release.
-"""
-
-__metaclass__ = type
-
-import os
-import re
-from StringIO import StringIO
-from subprocess import (
- PIPE,
- Popen,
- STDOUT,
- )
-import textwrap
-
-from bzrlib.branch import Branch
-from bzrlib.bzrdir import BzrDir
-from bzrlib.errors import NotStacked
-from bzrlib.tests import TestCaseWithTransport
-from bzrlib.transport import get_transport
-from bzrlib.transport.chroot import ChrootServer
-from lazr.uri import URI
-import transaction
-from zope.component import getUtility
-from zope.security.proxy import removeSecurityProxy
-
-from lp.code.enums import BranchLifecycleStatus
-from lp.code.interfaces.branchjob import IBranchScanJobSource
-from lp.codehosting.branchdistro import (
- DistroBrancher,
- switch_branches,
- )
-from lp.codehosting.vfs import branch_id_to_path
-from lp.registry.interfaces.pocket import PackagePublishingPocket
-from lp.services.config import config
-from lp.services.log.logger import (
- BufferLogger,
- FakeLogger,
- )
-from lp.services.osutils import override_environ
-from lp.testing import TestCaseWithFactory
-from lp.testing.dbuser import switch_dbuser
-from lp.testing.layers import LaunchpadZopelessLayer
-
-# We say "RELEASE" often enough to not want to say "PackagePublishingPocket."
-# each time.
-RELEASE = PackagePublishingPocket.RELEASE
-
-
-class FakeBranch:
- """Just enough of a Branch to pass `test_switch_branches`."""
-
- def __init__(self, id):
- self.id = id
-
- @property
- def unique_name(self):
- return branch_id_to_path(self.id)
-
-
-class TestSwitchBranches(TestCaseWithTransport):
- """Tests for `switch_branches`."""
-
- def test_switch_branches(self):
- # switch_branches moves a branch to the new location and places a
- # branch (with no revisions) stacked on the new branch in the old
- # location.
-
- chroot_server = ChrootServer(self.get_transport())
- chroot_server.start_server()
- self.addCleanup(chroot_server.stop_server)
- scheme = chroot_server.get_url().rstrip('/:')
-
- old_branch = FakeBranch(1)
- self.get_transport(old_branch.unique_name).create_prefix()
- tree = self.make_branch_and_tree(old_branch.unique_name)
- # XXX: AaronBentley 2010-08-06 bug=614404: a bzr username is
- # required to generate the revision-id.
- with override_environ(BZR_EMAIL='me@xxxxxxxxxxx'):
- tree.commit(message='.')
-
- new_branch = FakeBranch(2)
-
- switch_branches('.', scheme, old_branch, new_branch)
-
- # Post conditions:
- # 1. unstacked branch in new_branch's location
- # 2. stacked branch with no revisions in repo at old_branch
- # 3. last_revision() the same for two branches
-
- old_location_bzrdir = BzrDir.open(str(URI(
- scheme=scheme, host='', path='/' + old_branch.unique_name)))
- new_location_bzrdir = BzrDir.open(str(URI(
- scheme=scheme, host='', path='/' + new_branch.unique_name)))
-
- old_location_branch = old_location_bzrdir.open_branch()
- new_location_branch = new_location_bzrdir.open_branch()
-
- # 1. unstacked branch in new_branch's location
- self.assertRaises(NotStacked, new_location_branch.get_stacked_on_url)
-
- # 2. stacked branch with no revisions in repo at old_branch
- self.assertEqual(
- '/' + new_branch.unique_name,
- old_location_branch.get_stacked_on_url())
- self.assertEqual(
- [], old_location_bzrdir.open_repository().all_revision_ids())
-
- # 3. last_revision() the same for two branches
- self.assertEqual(
- old_location_branch.last_revision(),
- new_location_branch.last_revision())
-
-
-class TestDistroBrancher(TestCaseWithFactory):
- """Tests for `DistroBrancher`."""
-
- layer = LaunchpadZopelessLayer
-
- def setUp(self):
- TestCaseWithFactory.setUp(self)
- self.useBzrBranches(direct_database=True)
-
- def makeOfficialPackageBranch(self, distroseries=None,
- make_revisions=True):
- """Make an official package branch with an underlying bzr branch."""
- db_branch = self.factory.makePackageBranch(distroseries=distroseries)
- db_branch.sourcepackage.setBranch(RELEASE, db_branch, db_branch.owner)
- if make_revisions:
- self.factory.makeRevisionsForBranch(db_branch, count=1)
-
- transaction.commit()
-
- _, tree = self.create_branch_and_tree(
- tree_location=self.factory.getUniqueString(), db_branch=db_branch)
- # XXX: AaronBentley 2010-08-06 bug=614404: a bzr username is
- # required to generate the revision-id.
- if make_revisions:
- with override_environ(BZR_EMAIL='me@xxxxxxxxxxx'):
- tree.commit('')
-
- return db_branch
-
- def makeNewSeriesAndBrancher(self, distroseries=None):
- """Make a DistroBrancher.
-
- Any messages logged by this DistroBrancher can be checked by calling
- `assertLogMessages` below.
- """
- if distroseries is None:
- distroseries = self.factory.makeDistroSeries()
- self._log_file = StringIO()
- new_distroseries = self.factory.makeDistroSeries(
- distribution=distroseries.distribution)
- switch_dbuser('branch-distro')
- return DistroBrancher(
- FakeLogger(self._log_file), distroseries, new_distroseries)
-
- def clearLogMessages(self):
- """Forget about all logged messages seen so far."""
- self._log_file.seek(0, 0)
- self._log_file.truncate()
-
- def assertLogMessages(self, patterns):
- """Assert that the messages logged meet expectations.
-
- :param patterns: A list of regular expressions. The length must match
- the number of messages logged, and then each pattern must match
- the messages logged in order.
- """
- log_messages = self._log_file.getvalue().splitlines()
- if len(log_messages) > len(patterns):
- self.fail(
- "More log messages (%s) than expected (%s)" %
- (log_messages, patterns))
- elif len(log_messages) < len(patterns):
- self.fail(
- "Fewer log messages (%s) than expected (%s)" %
- (log_messages, patterns))
- for pattern, message in zip(patterns, log_messages):
- if not re.match(pattern, message):
- self.fail("%r does not match %r" % (pattern, message))
-
- def test_DistroBrancher_same_distro_check(self):
- # DistroBrancher.__init__ raises AssertionError if the two
- # distroseries passed are not from the same distribution.
- self.assertRaises(
- AssertionError, DistroBrancher, None,
- self.factory.makeDistroSeries(),
- self.factory.makeDistroSeries())
-
- def test_DistroBrancher_same_distroseries_check(self):
- # DistroBrancher.__init__ raises AssertionError if passed the same
- # distroseries twice.
- distroseries = self.factory.makeDistroSeries()
- self.assertRaises(
- AssertionError, DistroBrancher, None, distroseries, distroseries)
-
- def test_fromNames(self):
- # DistroBrancher.fromNames constructs a DistroBrancher from the names
- # of a distribution and two distroseries within it.
- distribution = self.factory.makeDistribution()
- distroseries1 = self.factory.makeDistroSeries(
- distribution=distribution)
- distroseries2 = self.factory.makeDistroSeries(
- distribution=distribution)
- brancher = DistroBrancher.fromNames(
- None, distribution.name, distroseries1.name, distroseries2.name)
- self.assertEqual(
- [distroseries1, distroseries2],
- [brancher.old_distroseries, brancher.new_distroseries])
-
- # A word on testing strategy: we don't directly test the post conditions
- # of makeOneNewBranch, but we do test that it satisfies checkOneBranch and
- # the tests for checkOneBranch verify that this function rejects various
- # ways in which makeOneNewBranch could conceivably fail.
-
- def test_makeOneNewBranch(self):
- # makeOneNewBranch creates an official package branch in the new
- # distroseries.
- db_branch = self.makeOfficialPackageBranch()
-
- brancher = self.makeNewSeriesAndBrancher(db_branch.distroseries)
- brancher.makeOneNewBranch(db_branch)
-
- new_branch = brancher.new_distroseries.getSourcePackage(
- db_branch.sourcepackage.name).getBranch(RELEASE)
-
- self.assertIsNot(None, new_branch)
- # The branch owner is the same, the source package name is the same,
- # the distroseries is the new one and the branch name is the name of
- # the new distroseries.
- self.assertEqual(
- [db_branch.owner, db_branch.distribution,
- db_branch.sourcepackagename, brancher.new_distroseries.name],
- [new_branch.owner, new_branch.distribution,
- new_branch.sourcepackagename, new_branch.name])
- # The new branch is set in the development state, and the old one is
- # mature.
- self.assertEqual(
- BranchLifecycleStatus.DEVELOPMENT, new_branch.lifecycle_status)
- self.assertEqual(
- BranchLifecycleStatus.MATURE, db_branch.lifecycle_status)
-
- def test_makeOneNewBranch_avoids_need_for_scan(self):
- # makeOneNewBranch sets the appropriate properties of the new branch
- # so a scan is unnecessary. This can be done because we are making a
- # copy of the source branch.
- db_branch = self.makeOfficialPackageBranch()
- self.factory.makeRevisionsForBranch(db_branch, count=10)
- tip_revision_id = db_branch.last_mirrored_id
- self.assertIsNot(None, tip_revision_id)
- # The makeRevisionsForBranch will create a scan job for the db_branch.
- # We don't really care about that, but what we do care about is that
- # no new jobs are created.
- existing_scan_job_count = len(
- list(getUtility(IBranchScanJobSource).iterReady()))
-
- brancher = self.makeNewSeriesAndBrancher(db_branch.distroseries)
- brancher.makeOneNewBranch(db_branch)
- new_branch = brancher.new_distroseries.getSourcePackage(
- db_branch.sourcepackage.name).getBranch(RELEASE)
-
- self.assertEqual(tip_revision_id, new_branch.last_mirrored_id)
- self.assertEqual(tip_revision_id, new_branch.last_scanned_id)
- # Make sure that the branch revisions have been copied.
- old_ancestry, old_history = removeSecurityProxy(
- db_branch).getScannerData()
- new_ancestry, new_history = removeSecurityProxy(
- new_branch).getScannerData()
- self.assertEqual(old_ancestry, new_ancestry)
- self.assertEqual(old_history, new_history)
- self.assertIs(None, new_branch.stacked_on)
- self.assertEqual(new_branch, db_branch.stacked_on)
- # The script doesn't have permission to create branch jobs, but just
- # to be insanely paranoid.
- switch_dbuser('launchpad')
- self.assertFalse(new_branch.pending_writes)
- scan_jobs = list(getUtility(IBranchScanJobSource).iterReady())
- self.assertEqual(existing_scan_job_count, len(scan_jobs))
-
- def test_makeOneNewBranch_inconsistent_branch(self):
- # makeOneNewBranch skips over an inconsistent official package branch
- # (see `checkConsistentOfficialPackageBranch` for precisely what an
- # "inconsistent official package branch" is).
- unofficial_branch = self.factory.makePackageBranch()
- brancher = self.makeNewSeriesAndBrancher(
- unofficial_branch.distroseries)
- brancher.makeOneNewBranch(unofficial_branch)
-
- new_branch = brancher.new_distroseries.getSourcePackage(
- unofficial_branch.sourcepackage.name).getBranch(RELEASE)
- self.assertIs(None, new_branch)
- self.assertLogMessages(
- ['^WARNING .* is not an official branch$',
- '^WARNING Skipping branch$'])
-
- def test_makeOnewNewBranch_empty_branch(self):
- # Branches with no commits work.
- db_branch = self.makeOfficialPackageBranch(make_revisions=False)
- brancher = self.makeNewSeriesAndBrancher(db_branch.distroseries)
- brancher.makeOneNewBranch(db_branch)
-
- def test_makeNewBranches(self):
- # makeNewBranches calls makeOneNewBranch for each official branch in
- # the old distroseries.
- db_branch = self.makeOfficialPackageBranch()
- db_branch2 = self.makeOfficialPackageBranch(
- distroseries=db_branch.distroseries)
-
- new_distroseries = self.factory.makeDistroSeries(
- distribution=db_branch.distribution)
-
- brancher = DistroBrancher(
- BufferLogger(), db_branch.distroseries, new_distroseries)
-
- brancher.makeNewBranches()
-
- new_sourcepackage = new_distroseries.getSourcePackage(
- db_branch.sourcepackage.name)
- new_branch = new_sourcepackage.getBranch(RELEASE)
- new_sourcepackage2 = new_distroseries.getSourcePackage(
- db_branch2.sourcepackage.name)
- new_branch2 = new_sourcepackage2.getBranch(RELEASE)
-
- self.assertIsNot(None, new_branch)
- self.assertIsNot(None, new_branch2)
-
- def test_makeNewBranches_idempotent(self):
- # makeNewBranches is idempotent in the sense that if a branch in the
- # old distroseries already has a counterpart in the new distroseries,
- # it is silently ignored.
- db_branch = self.makeOfficialPackageBranch()
-
- brancher = self.makeNewSeriesAndBrancher(db_branch.distroseries)
- brancher.makeNewBranches()
- brancher.makeNewBranches()
-
- new_branch = brancher.new_distroseries.getSourcePackage(
- db_branch.sourcepackage.name).getBranch(RELEASE)
-
- self.assertIsNot(new_branch, None)
-
- def test_makeOneNewBranch_checks_ok(self):
- # After calling makeOneNewBranch for a branch, calling checkOneBranch
- # returns True for that branch.
- db_branch = self.makeOfficialPackageBranch()
- brancher = self.makeNewSeriesAndBrancher(db_branch.distroseries)
- brancher.makeOneNewBranch(db_branch)
- self.clearLogMessages()
- ok = brancher.checkOneBranch(db_branch)
- self.assertLogMessages([])
- self.assertTrue(ok)
-
- def test_checkConsistentOfficialPackageBranch_product_branch(self):
- # checkConsistentOfficialPackageBranch returns False when passed a
- # product branch.
- db_branch = self.factory.makeProductBranch()
- brancher = self.makeNewSeriesAndBrancher()
- ok = brancher.checkConsistentOfficialPackageBranch(db_branch)
- self.assertLogMessages([
- '^WARNING Encountered unexpected product branch .*/.*/.*$'])
- self.assertFalse(ok)
-
- def test_checkConsistentOfficialPackageBranch_personal_branch(self):
- # checkConsistentOfficialPackageBranch returns False when passed a
- # personal branch.
- db_branch = self.factory.makePersonalBranch()
- brancher = self.makeNewSeriesAndBrancher()
- ok = brancher.checkConsistentOfficialPackageBranch(db_branch)
- self.assertLogMessages([
- '^WARNING Encountered unexpected personal branch .*/.*/.*$'])
- self.assertFalse(ok)
-
- def test_checkConsistentOfficialPackageBranch_no_official_branch(self):
- # checkConsistentOfficialPackageBranch returns False when passed a
- # branch which is not official for any package.
- db_branch = self.factory.makePackageBranch()
- brancher = self.makeNewSeriesAndBrancher(db_branch.distroseries)
- ok = brancher.checkConsistentOfficialPackageBranch(db_branch)
- self.assertLogMessages(
- ['^WARNING .*/.*/.* is not an official branch$'])
- self.assertFalse(ok)
-
- def test_checkConsistentOfficialPackageBranch_official_elsewhere(self):
- # checkConsistentOfficialPackageBranch returns False when passed a
- # branch which is official for a sourcepackage that it is not a branch
- # for.
- db_branch = self.factory.makePackageBranch()
- self.factory.makeSourcePackage().setBranch(
- RELEASE, db_branch, db_branch.owner)
- brancher = self.makeNewSeriesAndBrancher(db_branch.distroseries)
- ok = brancher.checkConsistentOfficialPackageBranch(db_branch)
- self.assertLogMessages(
- ['^WARNING .*/.*/.* is the official branch for .*/.*/.* but not '
- 'its sourcepackage$'])
- self.assertFalse(ok)
-
- def test_checkConsistentOfficialPackageBranch_official_twice(self):
- # checkConsistentOfficialPackageBranch returns False when passed a
- # branch that is official for two sourcepackages.
- db_branch = self.factory.makePackageBranch()
- db_branch.sourcepackage.setBranch(RELEASE, db_branch, db_branch.owner)
- self.factory.makeSourcePackage().setBranch(
- RELEASE, db_branch, db_branch.owner)
- brancher = self.makeNewSeriesAndBrancher(db_branch.distroseries)
- ok = brancher.checkConsistentOfficialPackageBranch(db_branch)
- self.assertLogMessages([
- '^WARNING .*/.*/.* is official for multiple series: .*/.*/.*, '
- '.*/.*/.*$'])
- self.assertFalse(ok)
-
- def test_checkConsistentOfficialPackageBranch_ok(self):
- # checkConsistentOfficialPackageBranch returns True when passed a
- # branch that is official for its sourcepackage and no other.
- db_branch = self.factory.makePackageBranch()
- brancher = self.makeNewSeriesAndBrancher(db_branch.distroseries)
- db_branch.sourcepackage.setBranch(RELEASE, db_branch, db_branch.owner)
- ok = brancher.checkConsistentOfficialPackageBranch(db_branch)
- self.assertLogMessages([])
- self.assertTrue(ok)
-
- def test_checkOneBranch_inconsistent_old_package_branch(self):
- # checkOneBranch returns False when passed a branch that is not a
- # consistent official package branch.
- db_branch = self.factory.makePackageBranch()
- brancher = self.makeNewSeriesAndBrancher()
- ok = brancher.checkOneBranch(db_branch)
- self.assertFalse(ok)
- self.assertLogMessages(
- ['^WARNING .*/.*/.* is not an official branch$'])
-
- def test_checkOneBranch_no_new_official_branch(self):
- # checkOneBranch returns False when there is no corresponding official
- # package branch in the new distroseries.
- db_branch = self.makeOfficialPackageBranch()
- brancher = self.makeNewSeriesAndBrancher(db_branch.distroseries)
- ok = brancher.checkOneBranch(db_branch)
- self.assertFalse(ok)
- self.assertLogMessages(
- ['^WARNING No official branch found for .*/.*/.*$'])
-
- def test_checkOneBranch_inconsistent_new_package_branch(self):
- # checkOneBranch returns False when the corresponding official package
- # branch in the new distroseries is not consistent.
- db_branch = self.makeOfficialPackageBranch()
- brancher = self.makeNewSeriesAndBrancher(db_branch.distroseries)
- new_db_branch = brancher.makeOneNewBranch(db_branch)
- switch_dbuser('launchpad')
- new_db_branch.setTarget(
- new_db_branch.owner,
- source_package=self.factory.makeSourcePackage())
- switch_dbuser('branch-distro')
- ok = brancher.checkOneBranch(new_db_branch)
- self.assertFalse(ok)
- self.assertLogMessages(
- ['^WARNING .*/.*/.* is the official branch for .*/.*/.* but not '
- 'its sourcepackage$'])
-
- def test_checkOneBranch_new_branch_missing(self):
- # checkOneBranch returns False when there is no bzr branch for the
- # database branch in the new distroseries.
- db_branch = self.makeOfficialPackageBranch()
- brancher = self.makeNewSeriesAndBrancher(db_branch.distroseries)
- new_db_branch = brancher.makeOneNewBranch(db_branch)
- url = 'lp-internal:///' + new_db_branch.unique_name
- get_transport(url).delete_tree('.bzr')
- ok = brancher.checkOneBranch(db_branch)
- self.assertFalse(ok)
- # Deleting the new branch will break the old branch, as that's stacked
- # on the new one.
- self.assertLogMessages([
- '^WARNING No bzr branch at new location '
- 'lp-internal:///.*/.*/.*/.*$',
- '^WARNING No bzr branch at old location '
- 'lp-internal:///.*/.*/.*/.*$',
- ])
-
- def test_checkOneBranch_old_branch_missing(self):
- # checkOneBranch returns False when there is no bzr branchfor the
- # database branch in old distroseries.
- db_branch = self.makeOfficialPackageBranch()
- brancher = self.makeNewSeriesAndBrancher(db_branch.distroseries)
- brancher.makeOneNewBranch(db_branch)
- url = 'lp-internal:///' + db_branch.unique_name
- get_transport(url).delete_tree('.bzr')
- ok = brancher.checkOneBranch(db_branch)
- self.assertFalse(ok)
- self.assertLogMessages([
- '^WARNING No bzr branch at old location '
- 'lp-internal:///.*/.*/.*/.*$'
- ])
-
- def test_checkOneBranch_new_stacked(self):
- # checkOneBranch returns False when the bzr branch for the database
- # branch in new distroseries is stacked.
- db_branch = self.makeOfficialPackageBranch()
- b, _ = self.create_branch_and_tree(self.factory.getUniqueString())
- brancher = self.makeNewSeriesAndBrancher(db_branch.distroseries)
- new_db_branch = brancher.makeOneNewBranch(db_branch)
- url = 'lp-internal:///' + new_db_branch.unique_name
- Branch.open(url).set_stacked_on_url('/' + b.unique_name)
- ok = brancher.checkOneBranch(db_branch)
- self.assertFalse(ok)
- self.assertLogMessages([
- '^WARNING New branch at lp-internal:///.*/.*/.*/.* is stacked on '
- '/.*/.*/.*, should be unstacked.$',
- ])
-
- def test_checkOneBranch_old_unstacked(self):
- # checkOneBranch returns False when the bzr branch for the database
- # branch in old distroseries is not stacked.
- db_branch = self.makeOfficialPackageBranch()
- brancher = self.makeNewSeriesAndBrancher(db_branch.distroseries)
- brancher.makeOneNewBranch(db_branch)
- url = 'lp-internal:///' + db_branch.unique_name
- old_bzr_branch = Branch.open(url)
- old_bzr_branch.set_stacked_on_url(None)
- ok = brancher.checkOneBranch(db_branch)
- self.assertLogMessages([
- '^WARNING Old branch at lp-internal:///.*/.*/.*/.* is not '
- 'stacked, should be stacked on /.*/.*/.*.$',
- '^.*has .* revisions.*$',
- ])
- self.assertFalse(ok)
-
- def test_checkOneBranch_old_misstacked(self):
- # checkOneBranch returns False when the bzr branch for the database
- # branch in old distroseries stacked on some other branch than the
- # branch in the new distroseries.
- db_branch = self.makeOfficialPackageBranch()
- b, _ = self.create_branch_and_tree(self.factory.getUniqueString())
- brancher = self.makeNewSeriesAndBrancher(db_branch.distroseries)
- brancher.makeOneNewBranch(db_branch)
- url = 'lp-internal:///' + db_branch.unique_name
- Branch.open(url).set_stacked_on_url('/' + b.unique_name)
- ok = brancher.checkOneBranch(db_branch)
- self.assertLogMessages([
- '^WARNING Old branch at lp-internal:///.*/.*/.*/.* is stacked on '
- '/.*/.*/.*, should be stacked on /.*/.*/.*.$',
- ])
- self.assertFalse(ok)
-
- def test_checkOneBranch_old_has_revisions(self):
- # checkOneBranch returns False when the bzr branch for the database
- # branch in old distroseries has a repository that contains revisions.
- db_branch = self.makeOfficialPackageBranch()
- brancher = self.makeNewSeriesAndBrancher(db_branch.distroseries)
- brancher.makeOneNewBranch(db_branch)
- url = 'lp-internal:///' + db_branch.unique_name
- old_bzr_branch = Branch.open(url)
- # XXX: AaronBentley 2010-08-06 bug=614404: a bzr username is
- # required to generate the revision-id.
- with override_environ(BZR_EMAIL='me@xxxxxxxxxxx'):
- old_bzr_branch.create_checkout(
- self.factory.getUniqueString()).commit('')
- ok = brancher.checkOneBranch(db_branch)
- self.assertLogMessages([
- '^WARNING Repository at lp-internal:///.*/.*/.*/.* has 1 '
- 'revisions.'
- ])
- self.assertFalse(ok)
-
- def test_checkOneBranch_old_has_null_tip(self):
- # checkOneBranch returns False when the bzr branch for the database
- # branch in old distroseries has tip revision of 'null:'.
- db_branch = self.makeOfficialPackageBranch()
- brancher = self.makeNewSeriesAndBrancher(db_branch.distroseries)
- brancher.makeOneNewBranch(db_branch)
- url = 'lp-internal:///' + db_branch.unique_name
- old_bzr_branch = Branch.open(url)
- old_bzr_branch.set_last_revision_info(0, 'null:')
- ok = brancher.checkOneBranch(db_branch)
- self.assertLogMessages([
- '^WARNING Old branch at lp-internal:///.*/.*/.*/.* has null tip '
- 'revision.'
- ])
- self.assertFalse(ok)
-
- def runBranchDistroScript(self, args):
- """Run the branch-distro.py script with the given arguments.
-
- ;param args: The arguments to pass to the branch-distro.py script.
- :return: A tuple (returncode, output). stderr and stdout are both
- contained in the output.
- """
- script_path = os.path.join(config.root, 'scripts', 'branch-distro.py')
- process = Popen([script_path] + args, stdout=PIPE, stderr=STDOUT)
- output, error = process.communicate()
- return process.returncode, output
-
- def test_makeNewBranches_script(self):
- # Running the script with the arguments 'distro old-series new-series'
- # makes new branches in the new series.
- db_branch = self.makeOfficialPackageBranch()
- brancher = self.makeNewSeriesAndBrancher(db_branch.distroseries)
- returncode, output = self.runBranchDistroScript(
- ['-v', db_branch.distribution.name,
- brancher.old_distroseries.name, brancher.new_distroseries.name])
- self.assertEqual(0, returncode)
- self.assertEqual(
- 'DEBUG Processing ' + db_branch.unique_name + '\n', output)
- brancher.checkOneBranch(db_branch)
-
- def test_checkNewBranches_script_success(self):
- # Running the script with the arguments '--check distro old-series
- # new-series' checks that the branches in the new series are as
- # expected.
- db_branch = self.makeOfficialPackageBranch()
- brancher = self.makeNewSeriesAndBrancher(db_branch.distroseries)
- brancher.makeNewBranches()
- returncode, output = self.runBranchDistroScript(
- ['-v', '--check', db_branch.distribution.name,
- brancher.old_distroseries.name, brancher.new_distroseries.name])
- self.assertEqual(0, returncode)
- self.assertEqual(
- 'DEBUG Checking ' + db_branch.unique_name + '\n', output)
- brancher.checkOneBranch(db_branch)
-
- def test_checkNewBranches_script_failure(self):
- # Running the script with the arguments '--check distro old-series
- # new-series' checks that the branches in the new series are as
- # expected and logs warnings and exits with code 1 is things are not
- # as expected.
- db_branch = self.makeOfficialPackageBranch()
- brancher = self.makeNewSeriesAndBrancher(db_branch.distroseries)
- returncode, output = self.runBranchDistroScript(
- ['-v', '--check', db_branch.distribution.name,
- brancher.old_distroseries.name, brancher.new_distroseries.name])
- sp_path = brancher.new_distroseries.getSourcePackage(
- db_branch.sourcepackagename).path
- expected = '''\
- DEBUG Checking %(branch_name)s
- WARNING No official branch found for %(sp_path)s
- ERROR Check failed
- ''' % {'branch_name': db_branch.unique_name, 'sp_path': sp_path}
- self.assertEqual(
- textwrap.dedent(expected), output)
- self.assertEqual(1, returncode)
=== removed file 'scripts/branch-distro.py'
--- scripts/branch-distro.py 2012-01-01 03:13:08 +0000
+++ scripts/branch-distro.py 1970-01-01 00:00:00 +0000
@@ -1,42 +0,0 @@
-#!/usr/bin/python -S
-#
-# Copyright 2009 Canonical Ltd. This software is licensed under the
-# GNU Affero General Public License version 3 (see the file LICENSE).
-
-import _pythonpath
-
-from lp.codehosting.branchdistro import DistroBrancher
-from lp.codehosting.vfs import get_rw_server
-from lp.services.scripts.base import (
- LaunchpadScript,
- LaunchpadScriptFailure,
- )
-
-
-class BranchDistroScript(LaunchpadScript):
-
- usage = "%prog distro old-series new-series"
-
- def add_my_options(self):
- self.parser.add_option(
- '--check', dest="check", action="store_true", default=False,
- help=("Check that the new distro series has its official "
- "branches set up correctly."))
-
- def main(self):
- if len(self.args) != 3:
- self.parser.error("Wrong number of arguments.")
- brancher = DistroBrancher.fromNames(self.logger, *self.args)
- server = get_rw_server(direct_database=True)
- server.start_server()
- try:
- if self.options.check:
- if not brancher.checkNewBranches():
- raise LaunchpadScriptFailure("Check failed")
- else:
- brancher.makeNewBranches()
- finally:
- server.stop_server()
-
-if __name__ == '__main__':
- BranchDistroScript("branch-distro", dbuser='branch-distro').run()
Follow ups
