[Pkg-bazaar-commits] ./bzr-builddeb/trunk r360: Complete merging of UpstreamSource code.
James Westby
james.westby at canonical.com
Sun Jul 26 18:21:49 UTC 2009
------------------------------------------------------------
revno: 360
committer: James Westby <james.westby at canonical.com>
branch nick: trunk
timestamp: Sun 2009-07-26 20:21:49 +0200
message:
Complete merging of UpstreamSource code.
modified:
cmds.py
directory.py
errors.py
tests/test_upstream.py
upstream.py
------------------------------------------------------------
revno: 334.4.10
committer: Jelmer Vernooij <jelmer at samba.org>
branch nick: obj
timestamp: Thu 2009-04-16 11:10:06 +0200
message:
Add tests for StackedUpstreamSource.
modified:
tests/test_upstream.py
------------------------------------------------------------
revno: 334.4.11
committer: Jelmer Vernooij <jelmer at samba.org>
branch nick: obj
timestamp: Thu 2009-04-16 11:30:49 +0200
message:
use exception to indicate missing package versions rather than boolean return value
modified:
cmds.py
errors.py
tests/test_upstream.py
upstream.py
------------------------------------------------------------
revno: 334.4.12
committer: Jelmer Vernooij <jelmer at samba.org>
branch nick: obj
timestamp: Thu 2009-04-16 11:42:19 +0200
message:
Add UScanSource tests.
modified:
tests/test_upstream.py
upstream.py
------------------------------------------------------------
revno: 334.4.13
committer: Jelmer Vernooij <jelmer at samba.org>
branch nick: obj
timestamp: Thu 2009-04-16 12:06:01 +0200
message:
Fix apt tests.
modified:
tests/test_upstream.py
upstream.py
------------------------------------------------------------
revno: 334.4.14
committer: Jelmer Vernooij <jelmer at samba.org>
branch nick: obj
timestamp: Thu 2009-04-16 12:42:12 +0200
message:
Test StackedUpstreamSource.__repr__.
modified:
cmds.py
debian/changelog
directory.py
tests/test_upstream.py
-------------- next part --------------
=== modified file 'cmds.py'
--- a/cmds.py 2009-07-26 16:21:15 +0000
+++ b/cmds.py 2009-07-26 18:21:49 +0000
@@ -34,6 +34,7 @@
urlutils,
)
from bzrlib.branch import Branch
+from bzrlib.bzrdir import BzrDir
from bzrlib.commands import Command
from bzrlib.errors import (BzrCommandError,
NoWorkingTree,
@@ -73,7 +74,11 @@
MergeModeDistiller,
NativeSourceDistiller,
)
-from bzrlib.plugins.builddeb.upstream import UpstreamProvider, UpstreamBranchSource
+from bzrlib.plugins.builddeb.upstream import (
+ UpstreamProvider,
+ UpstreamBranchSource,
+ get_upstream_sources,
+ )
from bzrlib.plugins.builddeb.util import (find_changelog,
get_export_upstream_revision,
find_last_distribution,
@@ -211,12 +216,7 @@
is_local = urlparse.urlsplit(location)[0] in ('', 'file')
if is_local:
os.chdir(location)
- try:
- tree, _ = WorkingTree.open_containing(location)
- branch = tree.branch
- except NoWorkingTree:
- tree = None
- branch, _ = Branch.open_containing(location)
+ tree, branch, relpath = BzrDir.open_containing_tree_or_branch(location)
return tree, branch, is_local
def _get_build_tree(self, revision, tree, branch):
@@ -375,10 +375,11 @@
export_upstream_revision, config,
changelog.version)
- upstream_provider = UpstreamProvider(tree, branch,
+ upstream_provider = UpstreamProvider(
changelog.package, changelog.version,
- orig_dir, larstiq=larstiq, upstream_branch=upstream_branch,
- upstream_revision=upstream_revision, allow_split=split)
+ orig_dir, get_upstream_sources(tree, branch,
+ larstiq=larstiq, upstream_branch=upstream_branch,
+ upstream_revision=upstream_revision, allow_split=split))
if merge:
distiller_cls = MergeModeDistiller
@@ -392,8 +393,8 @@
is_working_tree=working_tree)
build_source_dir = os.path.join(build_dir,
- changelog.package + "-"
- + changelog.version.upstream_version)
+ "%s-%s" % (changelog.package,
+ changelog.version.upstream_version))
builder = DebBuild(distiller, build_source_dir, build_cmd,
use_existing=use_existing)
@@ -780,9 +781,9 @@
orig_dir = config.orig_dir
if orig_dir is None:
orig_dir = default_orig_dir
- upstream_provider = UpstreamProvider(t, t.branch,
- changelog.package, changelog.version.upstream_version,
- orig_dir, larstiq=larstiq)
+ upstream_provider = UpstreamProvider(changelog.package,
+ changelog.version.upstream_version, orig_dir,
+ get_upstream_sources(t, t.branch, larstiq=larstiq))
distiller = MergeModeDistiller(t, upstream_provider,
larstiq=larstiq)
=== modified file 'directory.py'
--- a/directory.py 2009-04-15 23:17:50 +0000
+++ b/directory.py 2009-04-16 10:42:12 +0000
@@ -86,5 +86,15 @@
else:
return urls[version]["Git"]
+ if "Hg" in urls[version]:
+ try:
+ import bzrlib.plugins.hg
+ except ImportError:
+ info("This package uses hg. If you would like to "
+ "access it with bzr then please install bzr-hg"
+ "and re-run the command.")
+ else:
+ return urls[version]["Hg"]
+
raise errors.InvalidURL(path=url,
extra='unsupported VCSes %r found' % urls[version].keys())
=== modified file 'errors.py'
--- a/errors.py 2009-03-02 22:36:36 +0000
+++ b/errors.py 2009-04-16 09:30:49 +0000
@@ -153,6 +153,14 @@
_fmt = "You did not specify a package version."
+class PackageVersionNotPresent(BzrError):
+ _fmt = "%(package)s %(version)s was not found in %(upstream)s."
+
+ def __init__(self, package, version, upstream):
+ BzrError.__init__(self, package=package, version=version,
+ upstream=upstream)
+
+
class UnsupportedRepackFormat(BzrError):
_fmt = ('Either the file extension of "%(location)s" indicates that '
'it is a format unsupported for repacking or it is a '
=== modified file 'tests/test_upstream.py'
--- a/tests/test_upstream.py 2009-07-26 08:05:34 +0000
+++ b/tests/test_upstream.py 2009-07-26 18:21:49 +0000
@@ -30,12 +30,15 @@
)
from bzrlib.plugins.builddeb.errors import (
MissingUpstreamTarball,
+ PackageVersionNotPresent,
)
from bzrlib.plugins.builddeb.upstream import (
+ AptSource,
+ PristineTarSource,
+ StackedUpstreamSource,
UpstreamProvider,
UpstreamSource,
- AptSource,
- PristineTarSource,
+ UScanSource,
)
from bzrlib.plugins.builddeb.util import (
get_parent_dir,
@@ -43,151 +46,6 @@
)
-class MockProvider(UpstreamSource):
-
- def create_target(self, path):
- parent_dir = get_parent_dir(path)
- if parent_dir != '' and not os.path.exists(parent_dir):
- os.makedirs(parent_dir)
- f = open(path, "wb")
- try:
- f.write('')
- finally:
- f.close()
-
-
-class MockAptProvider(MockProvider):
-
- def __init__(self, find=False):
- self.find = find
- self.called_times = 0
- self.package = None
- self.upstream_version = None
- self.target_dir = None
-
- def get_specific_version(self, package, upstream_version, target_dir):
- self.called_times += 1
- self.package = package
- self.upstream_version = upstream_version
- self.target_dir = target_dir
- if self.find:
- self.create_target(os.path.join(target_dir,
- tarball_name(package, upstream_version)))
- return self.find
-
-
-class MockUscanProvider(MockProvider):
-
- def __init__(self, find=False):
- self.find = find
- self.called_times = 0
- self.package = None
- self.upstream_version = None
- self.watch_file_contents = None
- self.target_dir = None
-
- def provide(self, package, upstream_version, watch_file, target_dir):
- self.called_times += 1
- self.package = package
- self.upstream_version = upstream_version
- f = open(watch_file, "rb")
- try:
- self.watch_file_contents = f.read()
- finally:
- f.close()
- self.target_dir = target_dir
- if self.find:
- self.create_target(os.path.join(target_dir,
- tarball_name(package, upstream_version)))
- return self.find
-
-
-class MockPristineProvider(MockProvider):
-
- def __init__(self, find=False):
- self.find = find
- self.called_times = 0
- self.tree = None
- self.branch = None
- self.package = None
- self.upstream_version = None
- self.target_filename = None
-
- def provide(self, tree, branch, package, upstream_version, target_filename):
- self.called_times += 1
- self.tree = tree
- self.branch = branch
- self.package = package
- self.upstream_version = upstream_version
- self.target_filename = target_filename
- if self.find:
- self.create_target(target_filename)
- return self.find
-
-
-class MockOrigSourceProvider(MockProvider):
-
- def __init__(self, find=False):
- self.find = find
- self.called_times = 0
- self.source_dir = None
- self.desired_tarball_name = None
- self.target_dir = None
-
- def provide(self, source_dir, desired_tarball_name,
- target_dir):
- self.called_times += 1
- self.source_dir = source_dir
- self.desired_tarball_name = desired_tarball_name
- self.target_dir = target_dir
- if self.find:
- self.create_target(os.path.join(target_dir, desired_tarball_name))
- return self.find
-
-
-class MockOtherBranchProvider(MockProvider):
-
- def __init__(self, find=False):
- self.find = find
- self.called_times = 0
- self.upstream_branch = None
- self.upstream_revision = None
- self.target_filename = None
- self.tarball_base = None
-
- def provide(self, upstream_branch, upstream_revision, target_filename,
- tarball_base):
- self.called_times += 1
- self.upstream_branch = upstream_branch
- self.upstream_revision = upstream_revision
- self.target_filename = target_filename
- self.tarball_base = tarball_base
- if self.find:
- self.create_target(target_filename)
- return self.find
-
-
-class MockSplitProvider(MockProvider):
-
- def __init__(self, find=False):
- self.find = find
- self.called_times = 0
- self.tree = None
- self.package = None
- self.upstream_version = None
- self.target_filename = None
-
- def provide(self, tree, package, upstream_version, target_filename):
- self.called_times += 1
- self.tree = tree
- self.package = package
- self.upstream_version = upstream_version
- self.target_filename = target_filename
- if self.find:
- self.create_target(self.target_filename)
- return self.find
-
-
class MockSources(object):
def __init__(self, versions):
@@ -227,202 +85,6 @@
return self.sources
-class UpstreamProviderTests(TestCaseWithTransport):
-
- def setUp(self):
- super(UpstreamProviderTests, self).setUp()
- self.tree = self.make_branch_and_tree(".")
- self.branch = self.tree.branch
- self.package = "package"
- self.version = Version("0.1-1")
- self.upstream_version = self.version.upstream_version
- self.desired_tarball_name = tarball_name(self.package,
- self.upstream_version)
- self.tarball_base = "%s-%s" % (self.package, self.upstream_version)
- self.store_dir = "store"
- self.provider = UpstreamProvider(self.tree, self.branch,
- self.package, self.version, self.store_dir)
- self.provider._sources = []
- self.providers = {}
- self.providers["apt"] = MockAptProvider()
- self.provider._sources.append(self.providers["apt"])
- self.providers["uscan"] = MockUscanProvider()
- self.provider._sources.append(self.providers["uscan"])
- self.providers["pristine"] = MockPristineProvider()
- self.provider._sources.append(self.providers["pristine"])
- self.providers["orig"] = MockOrigSourceProvider()
- self.provider._sources.append(self.providers["orig"])
- self.providers["upstream"] = MockOtherBranchProvider()
- self.provider._sources.append(self.providers["upstream"])
- self.provider._upstream_branch_provider = \
- self.providers["upstream"].provide
- self.providers["split"] = MockSplitProvider()
- self.provider._sources.append(self.providers["split"])
- self.target_dir = "target"
- self.target_filename = os.path.join(self.target_dir,
- self.desired_tarball_name)
- self.store_filename = os.path.join(self.store_dir,
- tarball_name(self.package, self.version.upstream_version))
-
- def assertProvidersCalled(self, providers):
- for provider_name, provider in self.providers.items():
- if provider_name in providers:
- self.assertCalledCorrectly(provider_name)
- else:
- self.assertEqual(provider.called_times, 0,
- "%s wasn't expected to be called" % provider_name)
-
- def call_provider(self):
- self.assertEqual(self.provider.provide(self.target_dir),
- self.target_filename)
-
- def test_already_in_target(self):
- os.makedirs(self.target_dir)
- f = open(self.target_filename, "wb")
- f.close()
- self.call_provider()
- self.failUnlessExists(self.target_filename)
- # Should this be copied across?
- self.failIfExists(self.store_filename)
- self.assertProvidersCalled({})
-
- def test_already_in_store(self):
- os.makedirs(self.store_dir)
- f = open(self.store_filename, "wb")
- f.close()
- self.call_provider()
- self.failUnlessExists(self.target_filename)
- self.failUnlessExists(self.store_filename)
- self.assertProvidersCalled({})
-
- def assertCalledCorrectly(self, provider_name):
- provider = self.providers[provider_name]
- for attr_name in provider.__dict__:
- if attr_name in ("find", "provide", "source_dir"):
- continue
- if attr_name == "called_times":
- self.assertEqual(provider.called_times, 1,
- "%s was not called" % provider_name)
- continue
- if attr_name == "target_filename":
- self.assertEqual(provider.target_filename,
- self.store_filename)
- continue
- if attr_name == "target_dir":
- self.assertEqual(provider.target_dir,
- self.store_dir)
- continue
- if attr_name == "fetch_dir":
- self.assertEqual(provider.fetch_dir,
- os.path.dirname(provider.source_dir))
- continue
- attr = getattr(provider, attr_name)
- correct_attr = getattr(self, attr_name)
- self.assertEqual(correct_attr, attr,
- "%s doesn't match\nexpected: %s\ngot: %s"
- % (attr_name, correct_attr, attr))
-
- def assertSuccesfulCall(self, provider, other_providers):
- self.providers[provider].find = True
- self.call_provider()
- self.failUnlessExists(self.target_filename)
- self.failUnlessExists(self.store_filename)
- self.assertProvidersCalled([provider] + other_providers)
-
- def test_from_pristine_tar(self):
- self.assertSuccesfulCall("pristine", [])
-
- def test_from_apt(self):
- self.assertSuccesfulCall("apt", ["pristine"])
-
- def test_from_uscan(self):
- self.build_tree(["watch", "debian/", "debian/watch"])
- self.tree.add(["watch", "debian/", "debian/watch"])
- self.watch_file_contents = "contents of debian/watch\n"
- self.assertSuccesfulCall("uscan", ["pristine", "apt"])
-
- def test_uscan_not_called_if_not_watch(self):
- self.build_tree(["watch"])
- self.tree.add(["watch"])
- self.assertRaises(MissingUpstreamTarball, self.provider.provide,
- self.target_dir)
- self.failIfExists(self.target_filename)
- self.failIfExists(self.store_filename)
- self.assertProvidersCalled(["pristine", "apt"])
-
- def test_uscan_in_larstiq(self):
- self.build_tree(["watch", "debian/", "debian/watch"])
- self.tree.add(["watch", "debian/", "debian/watch"])
- self.watch_file_contents = "contents of watch\n"
- self.provider.larstiq = True
- self.assertSuccesfulCall("uscan", ["pristine", "apt"])
-
- def test_from_get_orig_source(self):
- self.build_tree(["rules", "debian/", "debian/rules"])
- self.tree.add(["rules", "debian/", "debian/rules"])
- self.watch_file_contents = "contents of debian/rules\n"
- self.assertSuccesfulCall("orig", ["pristine", "apt"])
-
- def test_get_orig_source_not_called_if_no_rules(self):
- self.build_tree(["rules"])
- self.tree.add(["rules"])
- self.assertRaises(MissingUpstreamTarball, self.provider.provide,
- self.target_dir)
- self.failIfExists(self.target_filename)
- self.failIfExists(self.store_filename)
- self.assertProvidersCalled(["pristine", "apt"])
-
- def test_get_orig_source_in_larstiq(self):
- self.build_tree(["rules", "debian/", "debian/rules"])
- self.tree.add(["rules", "debian/", "debian/rules"])
- self.watch_file_contents = "contents of rules\n"
- self.provider.larstiq = True
- self.assertSuccesfulCall("orig", ["pristine", "apt"])
-
- def test_from_upstream_branch(self):
- upstream_tree = self.make_branch_and_tree("upstream")
- self.build_tree(["upstream/foo"])
- upstream_tree.add(["foo"])
- self.upstream_branch = upstream_tree.branch
- self.upstream_revision = upstream_tree.commit("upstream one")
- self.provider.upstream_revision = self.upstream_revision
- self.provider.upstream_branch = self.upstream_branch
- self.assertSuccesfulCall("upstream", ["pristine", "apt"])
-
- def test_from_split(self):
- self.provider.allow_split = True
- self.assertSuccesfulCall("split", ["pristine", "apt"])
-
- def test_upstream_before_orig_source(self):
- upstream_tree = self.make_branch_and_tree("upstream")
- self.build_tree(["upstream/foo"])
- upstream_tree.add(["foo"])
- self.upstream_branch = upstream_tree.branch
- self.upstream_revision = upstream_tree.commit("upstream one")
- self.provider.upstream_revision = self.upstream_revision
- self.provider.upstream_branch = self.upstream_branch
- self.build_tree(["rules", "debian/", "debian/rules"])
- self.tree.add(["rules", "debian/", "debian/rules"])
- self.watch_file_contents = "contents of debian/rules\n"
- self.assertSuccesfulCall("upstream", ["pristine", "apt"])
-
- def test_get_orig_source_before_uscan(self):
- self.build_tree(["rules", "debian/", "debian/rules"])
- self.tree.add(["rules", "debian/", "debian/rules"])
- self.watch_file_contents = "contents of debian/rules\n"
- self.build_tree(["watch", "debian/watch"])
- self.tree.add(["watch", "debian/watch"])
- self.watch_file_contents = "contents of debian/watch\n"
- self.assertSuccesfulCall("orig", ["pristine", "apt"])
-
- def test_uscan_before_split(self):
- self.build_tree(["watch", "debian/", "debian/watch"])
- self.tree.add(["watch", "debian/", "debian/watch"])
- self.watch_file_contents = "contents of debian/watch\n"
- self.provider.allow_split = True
- self.assertSuccesfulCall("uscan", ["pristine", "apt"])
-
-
class MockAptCaller(object):
def __init__(self, work=False):
@@ -451,8 +113,10 @@
caller = MockAptCaller()
sources = MockSources([])
apt_pkg = MockAptPkg(sources)
- self.assertEqual(False, AptSource().get_specific_version(
- "apackage", "0.2", "target", _apt_pkg=apt_pkg, _apt_caller=caller))
+ src = AptSource()
+ src._run_apt_source = caller.call
+ self.assertRaises(PackageVersionNotPresent, src.get_specific_version,
+ "apackage", "0.2", "target", _apt_pkg=apt_pkg)
self.assertEqual(1, apt_pkg.init_called_times)
self.assertEqual(1, apt_pkg.get_pkg_source_records_called_times)
self.assertEqual(1, sources.restart_called_times)
@@ -464,8 +128,10 @@
caller = MockAptCaller()
sources = MockSources(["0.1-1"])
apt_pkg = MockAptPkg(sources)
- self.assertEqual(False, AptSource().get_specific_version(
- "apackage", "0.2", "target", _apt_pkg=apt_pkg, _apt_caller=caller))
+ src = AptSource()
+ src._run_apt_source = caller.call
+ self.assertRaises(PackageVersionNotPresent, src.get_specific_version,
+ "apackage", "0.2", "target", _apt_pkg=apt_pkg)
self.assertEqual(1, apt_pkg.init_called_times)
self.assertEqual(1, apt_pkg.get_pkg_source_records_called_times)
self.assertEqual(1, sources.restart_called_times)
@@ -477,9 +143,10 @@
caller = MockAptCaller(work=True)
sources = MockSources(["0.1-1", "0.2-1"])
apt_pkg = MockAptPkg(sources)
- self.assertEqual(True, AptSource().get_specific_version(
- "apackage", "0.2", "target",
- _apt_pkg=apt_pkg, _apt_caller=caller.call))
+ src = AptSource()
+ src._run_apt_source = caller.call
+ src.get_specific_version("apackage", "0.2", "target",
+ _apt_pkg=apt_pkg)
self.assertEqual(1, apt_pkg.init_called_times)
self.assertEqual(1, apt_pkg.get_pkg_source_records_called_times)
self.assertEqual(1, sources.restart_called_times)
@@ -495,9 +162,11 @@
caller = MockAptCaller()
sources = MockSources(["0.1-1", "0.2-1"])
apt_pkg = MockAptPkg(sources)
- self.assertEqual(False, AptSource().get_specific_version(
+ src = AptSource()
+ src._run_apt_source = caller.call
+ self.assertRaises(PackageVersionNotPresent, src.get_specific_version,
"apackage", "0.2", "target",
- _apt_pkg=apt_pkg, _apt_caller=caller.call))
+ _apt_pkg=apt_pkg)
self.assertEqual(1, apt_pkg.init_called_times)
self.assertEqual(1, apt_pkg.get_pkg_source_records_called_times)
self.assertEqual(1, sources.restart_called_times)
@@ -508,3 +177,71 @@
self.assertEqual("apackage", caller.package)
self.assertEqual("0.2-1", caller.version_str)
self.assertEqual("target", caller.target_dir)
+
+
+class RecordingSource(object):
+
+ def __init__(self, succeed):
+ self._succeed = succeed
+ self._specific_versions = []
+
+ def get_specific_version(self, package, version, target_dir):
+ self._specific_versions.append((package, version, target_dir))
+ if not self._succeed:
+ raise PackageVersionNotPresent(package, version, self)
+
+ def __repr__(self):
+ return "%s()" % self.__class__.__name__
+
+
+class StackedUpstreamSourceTests(TestCase):
+
+ def test_first_wins(self):
+ a = RecordingSource(False)
+ b = RecordingSource(True)
+ c = RecordingSource(False)
+ stack = StackedUpstreamSource([a, b, c])
+ stack.get_specific_version("mypkg", "1.0", "bla")
+ self.assertEquals([("mypkg", "1.0", "bla")], b._specific_versions)
+ self.assertEquals([("mypkg", "1.0", "bla")], a._specific_versions)
+ self.assertEquals([], c._specific_versions)
+
+ def test_repr(self):
+ self.assertEquals("StackedUpstreamSource([])",
+ repr(StackedUpstreamSource([])))
+ self.assertEquals("StackedUpstreamSource([RecordingSource()])",
+ repr(StackedUpstreamSource([RecordingSource(False)])))
+
+ def test_none(self):
+ a = RecordingSource(False)
+ b = RecordingSource(False)
+ stack = StackedUpstreamSource([a, b])
+ self.assertRaises(PackageVersionNotPresent,
+ stack.get_specific_version, "pkg", "1.0", "bla")
+ self.assertEquals([("pkg", "1.0", "bla")], b._specific_versions)
+ self.assertEquals([("pkg", "1.0", "bla")], a._specific_versions)
+
+
+class UScanSourceTests(TestCaseWithTransport):
+
+ def setUp(self):
+ super(UScanSourceTests, self).setUp()
+ self.tree = self.make_branch_and_tree('.')
+
+ def test_export_watchfile_none(self):
+ src = UScanSource(self.tree, False)
+ self.assertEquals(None, src._export_watchfile())
+
+ def test_export_watchfile_larstiq(self):
+ src = UScanSource(self.tree, True)
+ self.build_tree(['watch'])
+ self.assertEquals(None, src._export_watchfile())
+ self.tree.add(['watch'])
+ self.assertTrue(src._export_watchfile() is not None)
+
+ def test_export_watchfile(self):
+ src = UScanSource(self.tree, False)
+ self.build_tree(['debian/', 'debian/watch'])
+ self.assertEquals(None, src._export_watchfile())
+ self.tree.smart_add(['debian/watch'])
+ self.assertTrue(src._export_watchfile() is not None)
=== modified file 'upstream.py'
--- a/upstream.py 2009-07-26 08:05:34 +0000
+++ b/upstream.py 2009-07-26 18:21:49 +0000
@@ -1,5 +1,6 @@
# upstream.py -- Providers of upstream source
# Copyright (C) 2009 Canonical Ltd.
+# Copyright (C) 2009 Jelmer Vernooij <jelmer at debian.org>
#
# This file is part of bzr-builddeb.
#
@@ -29,7 +30,10 @@
from bzrlib.revisionspec import RevisionSpec
from bzrlib.trace import info
-from bzrlib.plugins.builddeb.errors import MissingUpstreamTarball
+from bzrlib.plugins.builddeb.errors import (
+ MissingUpstreamTarball,
+ PackageVersionNotPresent,
+ )
from bzrlib.plugins.builddeb.import_dsc import DistributionBranch
from bzrlib.plugins.builddeb.repack_tarball import repack_tarball
from bzrlib.plugins.builddeb.util import (
@@ -55,7 +59,6 @@
:param package: Name of the package
:param version: Version string of the version to fetch
:param target_dir: Directory in which to store the tarball
- :return: Boolean indicating whether a tarball was fetched
"""
raise NotImplemented(self.get_specific_version)
@@ -70,32 +73,28 @@
self.branch = branch
self.tree = tree
- def get_specific_version(self, package, upstream_version, target_dir):
- target_filename = self._tarball_path(package, upstream_version,
+ def get_specific_version(self, package, version, target_dir):
+ target_filename = self._tarball_path(package, version,
target_dir)
db = DistributionBranch(self.branch, None, tree=self.tree)
- if not db.has_upstream_version_in_packaging_branch(upstream_version):
- return False
- revid = db._revid_of_upstream_version_from_branch(upstream_version)
+ if not db.has_upstream_version_in_packaging_branch(version):
+ raise PackageVersionNotPresent(package, version, self)
+ revid = db._revid_of_upstream_version_from_branch(version)
if not db.has_pristine_tar_delta(revid):
- return False
+ raise PackageVersionNotPresent(package, version, self)
info("Using pristine-tar to reconstruct the needed tarball.")
- db.reconstruct_pristine_tar(revid, package, upstream_version,
- target_filename)
- return True
+ db.reconstruct_pristine_tar(revid, package, version, target_filename)
class AptSource(UpstreamSource):
"""Upstream source that uses apt-source."""
def get_specific_version(self, package, upstream_version, target_dir,
- _apt_pkg=None, _apt_caller=None):
+ _apt_pkg=None):
if _apt_pkg is None:
import apt_pkg
else:
apt_pkg = _apt_pkg
- if _apt_caller is None:
- _apt_caller = self._run_apt_source
apt_pkg.init()
sources = apt_pkg.GetPkgSrcRecords()
sources.Restart()
@@ -103,11 +102,11 @@
while sources.Lookup(package):
if upstream_version \
== Version(sources.Version).upstream_version:
- if _apt_caller(package, sources.Version, target_dir):
- return True
+ if self._run_apt_source(package, sources.Version, target_dir):
+ return
break
info("apt could not find the needed tarball.")
- return False
+ raise PackageVersionNotPresent(package, upstream_version, self)
def _get_command(self, package, version_str):
return 'apt-get source -y --only-source --tar-only %s=%s' % \
@@ -153,7 +152,6 @@
tarball_base = "%s-%s" % (package, version)
rev_tree = self.upstream_branch.repository.revision_tree(revid)
export(rev_tree, target_filename, 'tgz', tarball_base)
- return True
finally:
self.upstream_branch.unlock()
@@ -198,13 +196,14 @@
os.mkdir(export_dir)
export_dir = os.path.join(export_dir, "debian")
export(self.tree, export_dir, format="dir")
- return self._get_orig_source(base_export_dir,
- desired_tarball_name, target_dir)
+ if not self._get_orig_source(base_export_dir,
+ desired_tarball_name, target_dir):
+ raise PackageVersionNotPresent(package, version, self)
+ return
finally:
shutil.rmtree(tmpdir)
- info("No debian/rules file to try and use for a get-orig-source "
- "rule")
- return False
+ info("No debian/rules file to try and use for a get-orig-source rule")
+ raise PackageVersionNotPresent(package, version, self)
class UScanSource(UpstreamSource):
@@ -226,7 +225,7 @@
return False
return True
- def get_specific_version(self, package, version, target_dir):
+ def _export_watchfile(self):
if self.larstiq:
watchfile = 'watch'
else:
@@ -234,7 +233,7 @@
watch_id = self.tree.path2id(watchfile)
if watch_id is None:
info("No watch file to use to retrieve upstream tarball.")
- return False
+ return None
(tmp, tempfilename) = tempfile.mkstemp()
try:
tmp = os.fdopen(tmp, 'wb')
@@ -242,9 +241,16 @@
tmp.write(watch)
finally:
tmp.close()
+ return tempfilename
+
+ def get_specific_version(self, package, version, target_dir):
+ tempfilename = self._export_watchfile()
+ if tempfilename is None:
+ raise PackageVersionNotPresent(package, version, self)
try:
- return self._uscan(package, version, tempfilename,
- target_dir)
+ if not self._uscan(package, version, tempfilename,
+ target_dir):
+ raise PackageVersionNotPresent(package, version, self)
finally:
os.unlink(tempfilename)
@@ -266,15 +272,14 @@
tar.add(export_dir, "%s-%s" % (package, upstream_version))
finally:
tar.close()
- return True
finally:
shutil.rmtree(tmpdir)
def get_specific_version(self, package, version, target_dir):
info("Using the current branch without the 'debian' directory "
"to create the tarball")
- return self._split(package,
- version, self._tarball_path(package, version, target_dir))
+ self._split(package, version,
+ self._tarball_path(package, version, target_dir))
class StackedUpstreamSource(UpstreamSource):
@@ -286,11 +291,47 @@
def __init__(self, sources):
self._sources = sources
+ def __repr__(self):
+ return "%s(%r)" % (self.__class__.__name__, self._sources)
+
def get_specific_version(self, package, version, target_dir):
for source in self._sources:
- if source.get_specific_version(package, version, target_dir):
- return True
- return False
+ try:
+ return source.get_specific_version(package, version, target_dir)
+ except PackageVersionNotPresent:
+ pass
+ raise PackageVersionNotPresent(package, version, self)
+
+
+def get_upstream_sources(tree, branch, larstiq=False, upstream_branch=None,
+ upstream_revision=None, allow_split=False):
+ """Get the list of upstream sources to retrieve upstream versions from.
+
+ :param tree: The tree that is being built from.
+ :param branch: The branch that is being built from.
+ :param larstiq: Whether the tree versions the root of ./debian.
+ :param upstream_branch: An upstream branch that can be exported
+ if needed.
+ :param upstream_revision: The revision to use of the upstream branch
+ if it is used.
+ :param allow_split: Whether the provider can provide the tarball
+ by exporting the branch and removing the "debian" dir.
+
+ """
+ sources = [
+ PristineTarSource(tree, branch),
+ AptSource(),
+ ]
+ if upstream_branch is not None:
+ sources.append(
+ UpstreamBranchSource(upstream_branch, upstream_revision))
+ sources.extend([
+ GetOrigSourceSource(tree, larstiq),
+ UScanSource(tree, larstiq),
+ ])
+ if allow_split:
+ sources.append(SelfSplitSource(tree))
+ return sources
class UpstreamProvider(object):
@@ -300,40 +341,16 @@
instance using pristine-tar, or using apt.
"""
- def __init__(self, tree, branch, package, version, store_dir,
- larstiq=False, upstream_branch=None, upstream_revision=None,
- allow_split=False):
+ def __init__(self, package, version, store_dir, sources):
"""Create an UpstreamProvider.
- :param tree: The tree that is being built from.
- :param branch: The branch that is being built from.
:param package: the name of the source package that is being built.
:param version: the Version of the package that is being built.
:param store_dir: A directory to cache the tarballs.
- :param larstiq: Whether the tree versions the root of ./debian.
- :param upstream_branch: An upstream branch that can be exported
- if needed.
- :param upstream_revision: The revision to use of the upstream branch
- if it is used.
- :param allow_split: Whether the provider can provide the tarball
- by exporting the branch and removing the "debian" dir.
"""
self.package = package
self.version = Version(version)
self.store_dir = store_dir
- sources = [
- PristineTarSource(tree, branch),
- AptSource(),
- ]
- if upstream_branch is not None:
- sources.append(
- UpstreamBranchSource(upstream_branch, upstream_revision))
- sources.extend([
- GetOrigSourceSource(tree, larstiq),
- UScanSource(tree, larstiq),
- ])
- if allow_split:
- sources.append(SelfSplitSource(tree))
self.source = StackedUpstreamSource(sources)
def provide(self, target_dir):
@@ -367,15 +384,16 @@
if not self.already_exists_in_store():
if not os.path.exists(self.store_dir):
os.makedirs(self.store_dir)
- self.source.get_specific_version(self.package,
- self.version.upstream_version,
- target_dir)
+ try:
+ self.source.get_specific_version(self.package,
+ self.version.upstream_version, target_dir)
+ except PackageVersionNotPresent:
+ raise MissingUpstreamTarball(self._tarball_name())
else:
info("Using the upstream tarball that is present in "
"%s" % self.store_dir)
- if self.provide_from_store_dir(target_dir):
- return os.path.join(target_dir, self._tarball_name())
- raise MissingUpstreamTarball(self._tarball_name())
+ assert self.provide_from_store_dir(target_dir)
+ return os.path.join(target_dir, self._tarball_name())
def already_exists_in_target(self, target_dir):
return os.path.exists(os.path.join(target_dir, self._tarball_name()))
More information about the Pkg-bazaar-commits
mailing list