Commit 53016921 authored by Guido Günther's avatar Guido Günther

Move rollback code out of import_orig

so it can be reused in gbp import-ref
parent 23874c2c
......@@ -364,4 +364,12 @@ class DebianGitRepository(PkgGitRepository):
raise GitRepositoryError("Error creating %s: %s" % (output, e))
return True
def vcs_tag_parent(self, vcs_tag_format, version):
"""If linking to the upstream VCS get the commit id"""
if vcs_tag_format:
return [self.rev_parse("%s^{}" % self.version_to_tag(vcs_tag_format, version))],
else:
return None
# vim:et:ts=4:sw=4:et:sts=4:ai:set list listchars=tab\:»·,trail\:·:
# vim: set fileencoding=utf-8 :
#
# (C) 2018 Guido Günther <agx@sigxcpu.org>
"""A git repository for Debian packages that can roll back operations"""
from .. import log
from .. git import GitRepositoryError
from . git import DebianGitRepository
class RollbackError(GitRepositoryError):
"""
Error raised if the rollback failed
"""
def __init__(self, errors):
self.msg = "Automatic rollback failed"
super(RollbackError, self).__init__(self.msg)
self.errors = errors
def __str__(self):
return "%s %s" % (self.msg, self.errors)
class RollbackDebianGitRepository(DebianGitRepository):
"""
Like a DebianGitRepository but can also perform rollbacks and knows
about some of the inner workings upstream vcs_tag, …
"""
def __init__(self, *args, **kwargs):
self.rollbacks = []
self.rollback_errors = []
DebianGitRepository.__init__(self, *args, **kwargs)
def has_rollbacks(self):
return len(self.rollbacks) > 0
def rrr(self, refname, action, reftype):
"""
Remember ref for rollback
@param refname: ref to roll back
@param action: the rollback action (delete, reset, ...)
@param reftype: the reference type (tag, branch, ...)
"""
sha = None
if action == 'reset':
try:
sha = self.rev_parse(refname)
except GitRepositoryError as err:
log.warn("Failed to rev-parse %s: %s" % (refname, err))
elif action == 'delete':
pass
elif action == 'abortmerge':
pass
else:
raise GitRepositoryError("Unknown action %s for %s %s" % (action, reftype, refname))
self.rollbacks.append((refname, reftype, action, sha))
def rrr_branch(self, branchname, action='reset-or-delete'):
if action == 'reset-or-delete':
if self.has_branch(branchname):
return self.rrr(branchname, 'reset', 'branch')
else:
return self.rrr(branchname, 'delete', 'branch')
else:
return self.rrr(branchname, action, 'branch')
def rrr_tag(self, tagname, action='delete'):
return self.rrr(tagname, action, 'tag')
def rrr_merge(self, commit, action='abortmerge'):
return self.rrr(commit, action, 'commit')
def rollback(self):
"""
Perform a complete rollback
Try to roll back as much as possible and remember what failed.
"""
for (name, reftype, action, sha) in self.rollbacks:
try:
if action == 'delete':
log.info('Rolling back %s %s by deleting it' % (reftype, name))
if reftype == 'tag':
self.delete_tag(name)
elif reftype == 'branch':
self.delete_branch(name)
else:
raise GitRepositoryError("Don't know how to delete %s %s" % (reftype, name))
elif action == 'reset' and reftype == 'branch':
log.info('Rolling back branch %s by resetting it to %s' % (name, sha))
self.update_ref("refs/heads/%s" % name, sha, msg="gbp import-orig: failure rollback of %s" % name)
elif action == 'abortmerge':
if self.is_in_merge():
log.info('Rolling back failed merge of %s' % name)
self.abort_merge()
else:
log.info("Nothing to rollback for merge of '%s'" % name)
else:
raise GitRepositoryError("Don't know how to %s %s %s" % (action, reftype, name))
except GitRepositoryError as e:
self.rollback_errors.append((name, reftype, action, sha, e))
if self.rollback_errors:
raise RollbackError(self.rollback_errors)
# Wrapped methods for rollbacks
def create_tag(self, *args, **kwargs):
name = kwargs['name']
ret = super(RollbackDebianGitRepository, self).create_tag(*args, **kwargs)
self.rrr_tag(name)
return ret
def commit_dir(self, *args, **kwargs):
import_branch = kwargs['branch']
self.rrr_branch(import_branch)
return super(RollbackDebianGitRepository, self).commit_dir(*args, **kwargs)
def create_branch(self, *args, **kwargs):
branch = kwargs['branch']
ret = super(RollbackDebianGitRepository, self).create_branch(*args, **kwargs)
self.rrr_branch(branch, 'delete')
return ret
def merge(self, *args, **kwargs):
commit = args[0] if args else kwargs['commit']
try:
return super(RollbackDebianGitRepository, self).merge(*args, **kwargs)
except GitRepositoryError:
# Only cleanup in the error case to undo working copy
# changes. Resetting the refs handles the other cases.
self.rrr_merge(commit)
raise
......@@ -28,7 +28,7 @@ from gbp.deb.format import DebianSourceFormat
from gbp.deb.upstreamsource import DebianUpstreamSource, unpack_component_tarball
from gbp.deb.uscan import (Uscan, UscanError)
from gbp.deb.changelog import ChangeLog, NoChangeLogError
from gbp.deb.git import (GitRepositoryError, DebianGitRepository)
from gbp.deb.git import GitRepositoryError
from gbp.config import GbpOptionParserDebian, GbpOptionGroup, no_upstream_branch_msg
from gbp.errors import GbpError
from gbp.format import format_str
......@@ -39,138 +39,7 @@ from gbp.scripts.common.import_orig import (orig_needs_repack, cleanup_tmp_tree,
ask_package_name, ask_package_version,
repack_upstream, is_link_target, download_orig)
from gbp.scripts.common.hook import Hook
class RollbackError(GitRepositoryError):
"""
Error raised if the rollback failed
"""
def __init__(self, errors):
self.msg = "Automatic rollback failed"
super(RollbackError, self).__init__(self.msg)
self.errors = errors
def __str__(self):
return "%s %s" % (self.msg, self.errors)
class ImportOrigDebianGitRepository(DebianGitRepository):
"""
Like a DebianGitRepository but can also perform rollbacks and knows
about some of the inner workings upstream vcs_tag, …
"""
def __init__(self, *args, **kwargs):
self.rollbacks = []
self.rollback_errors = []
DebianGitRepository.__init__(self, *args, **kwargs)
def has_rollbacks(self):
return len(self.rollbacks) > 0
def vcs_tag_parent(self, vcs_tag_format, version):
"""If linking to the upstream VCS get the commit id"""
if vcs_tag_format:
return [self.rev_parse("%s^{}" % self.version_to_tag(vcs_tag_format, version))],
else:
return None
def rrr(self, refname, action, reftype):
"""
Remember ref for rollback
@param refname: ref to roll back
@param action: the rollback action (delete, reset, ...)
@param reftype: the reference type (tag, branch, ...)
"""
sha = None
if action == 'reset':
try:
sha = self.rev_parse(refname)
except GitRepositoryError as err:
gbp.log.warn("Failed to rev-parse %s: %s" % (refname, err))
elif action == 'delete':
pass
elif action == 'abortmerge':
pass
else:
raise GbpError("Unknown action %s for %s %s" % (action, reftype, refname))
self.rollbacks.append((refname, reftype, action, sha))
def rrr_branch(self, branchname, action='reset-or-delete'):
if action == 'reset-or-delete':
if self.has_branch(branchname):
return self.rrr(branchname, 'reset', 'branch')
else:
return self.rrr(branchname, 'delete', 'branch')
else:
return self.rrr(branchname, action, 'branch')
def rrr_tag(self, tagname, action='delete'):
return self.rrr(tagname, action, 'tag')
def rrr_merge(self, commit, action='abortmerge'):
return self.rrr(commit, action, 'commit')
def rollback(self):
"""
Perform a complete rollback
Try to roll back as much as possible and remember what failed.
"""
for (name, reftype, action, sha) in self.rollbacks:
try:
if action == 'delete':
gbp.log.info('Rolling back %s %s by deleting it' % (reftype, name))
if reftype == 'tag':
self.delete_tag(name)
elif reftype == 'branch':
self.delete_branch(name)
else:
raise GitRepositoryError("Don't know how to delete %s %s" % (reftype, name))
elif action == 'reset' and reftype == 'branch':
gbp.log.info('Rolling back branch %s by resetting it to %s' % (name, sha))
self.update_ref("refs/heads/%s" % name, sha, msg="gbp import-orig: failure rollback of %s" % name)
elif action == 'abortmerge':
if self.is_in_merge():
gbp.log.info('Rolling back failed merge of %s' % name)
self.abort_merge()
else:
gbp.log.info("Nothing to rollback for merge of '%s'" % name)
else:
raise GitRepositoryError("Don't know how to %s %s %s" % (action, reftype, name))
except GitRepositoryError as e:
self.rollback_errors.append((name, reftype, action, sha, e))
if self.rollback_errors:
raise RollbackError(self.rollback_errors)
# Wrapped methods for rollbacks
def create_tag(self, *args, **kwargs):
name = kwargs['name']
ret = super(ImportOrigDebianGitRepository, self).create_tag(*args, **kwargs)
self.rrr_tag(name)
return ret
def commit_dir(self, *args, **kwargs):
import_branch = kwargs['branch']
self.rrr_branch(import_branch)
return super(ImportOrigDebianGitRepository, self).commit_dir(*args, **kwargs)
def create_branch(self, *args, **kwargs):
branch = kwargs['branch']
ret = super(ImportOrigDebianGitRepository, self).create_branch(*args, **kwargs)
self.rrr_branch(branch, 'delete')
return ret
def merge(self, *args, **kwargs):
commit = args[0] if args else kwargs['commit']
try:
return super(ImportOrigDebianGitRepository, self).merge(*args, **kwargs)
except GitRepositoryError:
# Only cleanup in the error case to undo working copy
# changes. Resetting the refs handles the other cases.
self.rrr_merge(commit)
raise
from gbp.deb.rollbackgit import RollbackDebianGitRepository
def prepare_pristine_tar(archive, pkg, version):
......@@ -535,7 +404,7 @@ def main(argv):
try:
try:
repo = ImportOrigDebianGitRepository('.')
repo = RollbackDebianGitRepository('.')
except GitRepositoryError:
raise GbpError("%s is not a git repository" % (os.path.abspath('.')))
......
# vim: set fileencoding=utf-8 :
"""Test L{gbp.command_wrappers.Command}'s tarball unpack"""
"""Test L{gbp.scripts.import_orig}"""
import os
import unittest
......@@ -8,62 +8,18 @@ from collections import namedtuple
from gbp.scripts.import_orig import (debian_branch_merge_by_replace,
GbpError,
ImportOrigDebianGitRepository,
is_30_quilt)
from gbp.scripts.common.import_orig import download_orig
from . testutils import DebianGitTestRepo
class TestImportOrigGitRepository(DebianGitTestRepo):
def setUp(self):
DebianGitTestRepo.setUp(self, ImportOrigDebianGitRepository)
def test_empty_rollback(self):
self.repo.rollback()
self.assertEquals(self.repo.rollback_errors, [])
def test_rrr_tag(self):
self.repo.rrr_tag('doesnotexist')
self.assertEquals(self.repo.rollbacks, [('doesnotexist', 'tag', 'delete', None)])
self.repo.rollback()
self.assertEquals(self.repo.rollback_errors, [])
def test_rrr_branch(self):
self.repo.rrr_branch('doesnotexist', 'delete')
self.assertEquals(self.repo.rollbacks, [('doesnotexist', 'branch', 'delete', None)])
self.repo.rollback()
self.assertEquals(self.repo.rollback_errors, [])
def test_rrr_merge(self):
self.repo.rrr_merge('HEAD')
self.assertEquals(self.repo.rollbacks, [('HEAD', 'commit', 'abortmerge', None)])
self.repo.rollback()
self.assertEquals(self.repo.rollback_errors, [])
def test_rrr_merge_abort(self):
self.repo.rrr_merge('HEAD')
self.assertEquals(self.repo.rollbacks, [('HEAD', 'commit', 'abortmerge', None)])
# Test that we abort the merge in case MERGE_HEAD exists
with open(os.path.join(self.repo.git_dir, 'MERGE_HEAD'), 'w'):
pass
self.assertTrue(self.repo.is_in_merge())
self.repo.rollback()
self.assertFalse(self.repo.is_in_merge())
self.assertEquals(self.repo.rollback_errors, [])
def test_rrr_unknown_action(self):
with self.assertRaisesRegexp(GbpError, "Unknown action unknown for tag doesnotmatter"):
self.repo.rrr('doesnotmatter', 'unknown', 'tag')
@unittest.skipUnless(os.getenv("GBP_NETWORK_TESTS"), "network tests disabled")
class TestImportOrigDownload(DebianGitTestRepo):
HOST = 'git.sigxcpu.org'
def setUp(self):
DebianGitTestRepo.setUp(self, ImportOrigDebianGitRepository)
os.chdir(self.repodir)
DebianGitTestRepo.setUp(self)
os.chdir(self.repo.path)
def test_404_download(self):
with self.assertRaisesRegexp(GbpError, "404 Client Error: Not Found for url"):
......
# vim: set fileencoding=utf-8 :
"""Test L{gbp.deb.rollbackgit}"""
import os
from . testutils import DebianGitTestRepo
from gbp.deb.rollbackgit import RollbackDebianGitRepository
from gbp.git.repository import GitRepositoryError
class TestRollbackGitRepository(DebianGitTestRepo):
def setUp(self):
DebianGitTestRepo.setUp(self, RollbackDebianGitRepository)
def test_empty_rollback(self):
self.repo.rollback()
self.assertEquals(self.repo.rollback_errors, [])
def test_rrr_tag(self):
self.repo.rrr_tag('doesnotexist')
self.assertEquals(self.repo.rollbacks, [('doesnotexist', 'tag', 'delete', None)])
self.repo.rollback()
self.assertEquals(self.repo.rollback_errors, [])
def test_rrr_branch(self):
self.repo.rrr_branch('doesnotexist', 'delete')
self.assertEquals(self.repo.rollbacks, [('doesnotexist', 'branch', 'delete', None)])
self.repo.rollback()
self.assertEquals(self.repo.rollback_errors, [])
def test_rrr_merge(self):
self.repo.rrr_merge('HEAD')
self.assertEquals(self.repo.rollbacks, [('HEAD', 'commit', 'abortmerge', None)])
self.repo.rollback()
self.assertEquals(self.repo.rollback_errors, [])
def test_rrr_merge_abort(self):
self.repo.rrr_merge('HEAD')
self.assertEquals(self.repo.rollbacks, [('HEAD', 'commit', 'abortmerge', None)])
# Test that we abort the merge in case MERGE_HEAD exists
with open(os.path.join(self.repo.git_dir, 'MERGE_HEAD'), 'w'):
pass
self.assertTrue(self.repo.is_in_merge())
self.repo.rollback()
self.assertFalse(self.repo.is_in_merge())
self.assertEquals(self.repo.rollback_errors, [])
def test_rrr_unknown_action(self):
with self.assertRaisesRegexp(GitRepositoryError, "Unknown action unknown for tag doesnotmatter"):
self.repo.rrr('doesnotmatter', 'unknown', 'tag')
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment