Rev 5135: All the bits are now hooked up. The slow tests are disabled, in http://bazaar.launchpad.net/~jameinel/bzr/2.2-is-up-to-date
John Arbash Meinel
john at arbash-meinel.com
Wed Jul 13 12:10:52 UTC 2011
At http://bazaar.launchpad.net/~jameinel/bzr/2.2-is-up-to-date
------------------------------------------------------------
revno: 5135
revision-id: john at arbash-meinel.com-20110713121021-aeuvnkjp5c627lxd
parent: john at arbash-meinel.com-20110713114538-7jvmvne9g12fdgw3
committer: John Arbash Meinel <john at arbash-meinel.com>
branch nick: 2.2-is-up-to-date
timestamp: Wed 2011-07-13 14:10:21 +0200
message:
All the bits are now hooked up. The slow tests are disabled,
but we can be reasonably confident that we can get data out of Launchpad.
-------------- next part --------------
=== modified file 'bzrlib/plugins/launchpad/lp_api_lite.py'
--- a/bzrlib/plugins/launchpad/lp_api_lite.py 2011-07-13 11:45:38 +0000
+++ b/bzrlib/plugins/launchpad/lp_api_lite.py 2011-07-13 12:10:21 +0000
@@ -51,24 +51,42 @@
self._project = project
self._setup_series_and_pocket(series)
+ def _setup_series_and_pocket(self, series):
+ """Parse the 'series' info into a series and a pocket.
+
+ eg::
+ _setup_series_and_pocket('natty-proposed')
+ => _series == 'natty'
+ _pocket == 'Proposed'
+ """
+ self._series = series
+ self._pocket = None
+ if self._series is not None and '-' in self._series:
+ self._series, self._pocket = self._series.split('-', 1)
+ self._pocket = self._pocket.title()
+
def _archive_URL(self):
+ """Return the Launchpad 'Archive' URL that we will query.
+ This is everything in the URL except the query parameters.
+ """
return '%s/%s/+archive/primary' % (self.LP_API_ROOT, self._archive)
def _publication_status(self):
+ """Handle the 'status' field.
+ It seems that Launchpad tracks all 'debian' packages as 'Pending', while
+ for 'ubuntu' we care about the 'Published' packages.
+ """
if self._archive == 'debian':
# Launchpad only tracks debian packages as "Pending", it doesn't mark
# them Published
return 'Pending'
return 'Published'
- def _setup_series_and_pocket(self, series):
- self._series = series
- self._pocket = None
- if self._series is not None and '-' in self._series:
- self._series, self._pocket = self._series.split('-', 1)
- self._pocket = self._pocket.title()
-
def _query_params(self):
+ """Get the parameters defining our query.
+ This defines the actions we are making against the archive.
+ :return: A dict of query parameters.
+ """
params = {'ws.op': 'getPublishedSources',
'exact_match': 'true',
# If we need to use "" shouldn't we quote the project somehow?
@@ -85,22 +103,44 @@
return params
def _query_URL(self):
+ """Create the full URL that we need to query, including parameters."""
params = self._query_params()
# We sort to give deterministic results for testing
encoded = urllib.urlencode(sorted(params.items()))
return '%s?%s' % (self._archive_URL(), encoded)
def _get_lp_info(self):
+ """Place an actual HTTP query against the Launchpad service."""
+ if json is None:
+ return None
query_URL = self._query_URL()
try:
req = urllib2.Request(query_URL)
response = urllib2.urlopen(req)
- json_txt = response.read()
- except (urllib2.URLError,), e:
+ json_info = response.read()
+ # XXX: We haven't tested the HTTPError
+ except (urllib2.URLError, urllib2.HTTPError), e:
trace.mutter('failed to place query to %r' % (query_URL,))
trace.log_exception_quietly()
return None
- return json_txt
+ return json_info
+
+ def _parse_json_info(self, json_info):
+ """Parse the json response from Launchpad into objects."""
+ if json is None:
+ return None
+ return json.loads(json_info)
+
+ def get_latest_version(self):
+ """Get the latest published version for the given package."""
+ json_info = self._get_lp_info()
+ if json_info is None:
+ return None
+ info = self._parse_json_info(json_info)
+ entries = info['entries']
+ if len(entries) == 0:
+ return None
+ return entries[0]['source_package_version']
def get_latest_publication(archive, series, project):
@@ -115,7 +155,7 @@
"""
if json is None:
return None
- archive_url = '%s/%s/+archive/primary?' % (LP_API_ROOT, archive)
+ archive_url = '%s/%s/+archive/primary?' % (LatestPublication.LP_API_ROOT, archive)
pocket = None
# TODO: If series is None, we probably need to hard-code it. I don't have
# proof yet, but otherwise we just get the most-recent version in any
@@ -136,7 +176,7 @@
'exact_match': 'true',
# If we need to use "" shouldn't we quote the project somehow?
'source_name': '"%s"' % (project,),
- 'status': status,
+ 'status': 'Published',
# We only need the latest one, the results seem to be properly
# most-recent-debian-version sorted
'ws.size': '1',
=== modified file 'bzrlib/plugins/launchpad/test_lp_api_lite.py'
--- a/bzrlib/plugins/launchpad/test_lp_api_lite.py 2011-07-13 11:45:38 +0000
+++ b/bzrlib/plugins/launchpad/test_lp_api_lite.py 2011-07-13 12:10:21 +0000
@@ -56,6 +56,9 @@
]
}"""
+_no_versions_response = '{"total_size": 0, "start": 0, "entries": []}'
+
+
class TestLatestPublication(tests.TestCase):
def make_latest_publication(self, archive='ubuntu', series='natty',
@@ -148,7 +151,7 @@
s.close()
self.assertIs(None, latest_pub._get_lp_info())
- def test__query_launchpad(self):
+ def DONT_test__query_launchpad(self):
# TODO: This is a test that we are making a valid request against
# launchpad. This seems important, but it is slow, requires net
# access, and requires launchpad to be up and running. So for
@@ -168,3 +171,52 @@
self.assertEqual('bzr', entry['source_package_name'])
version = entry['source_package_version']
self.assertIsNot(None, version)
+
+ def test__get_lp_info_no_json(self):
+ # If we can't parse the json, we don't make the query.
+ self.overrideAttr(lp_api_lite, 'json', None)
+ latest_pub = self.make_latest_publication()
+ self.assertIs(None, latest_pub._get_lp_info())
+
+ def test__parse_json_info_no_module(self):
+ # If a json parsing module isn't available, we just return None here.
+ self.overrideAttr(lp_api_lite, 'json', None)
+ latest_pub = self.make_latest_publication()
+ self.assertIs(None, latest_pub._parse_json_info(_example_response))
+
+ def test__parse_json_example_response(self):
+ if lp_api_lite.json is None:
+ raise tests.UnavailableFeature('json or simplejson module')
+ latest_pub = self.make_latest_publication()
+ content = latest_pub._parse_json_info(_example_response)
+ self.assertIsNot(None, content)
+ self.assertEqual(2, content['total_size'])
+ entries = content['entries']
+ self.assertEqual(1, len(entries))
+ entry = entries[0]
+ self.assertEqual('bzr', entry['source_package_name'])
+ self.assertEqual("2.1.4-0ubuntu1", entry["source_package_version"])
+
+ def test_get_latest_version_no_response(self):
+ latest_pub = self.make_latest_publication()
+ latest_pub._get_lp_info = lambda: None
+ self.assertEqual(None, latest_pub.get_latest_version())
+
+ def test_get_latest_version_no_json(self):
+ self.overrideAttr(lp_api_lite, 'json', None)
+ latest_pub = self.make_latest_publication()
+ self.assertEqual(None, latest_pub.get_latest_version())
+
+ def test_get_latest_version_no_versions(self):
+ latest_pub = self.make_latest_publication()
+ latest_pub._get_lp_info = lambda: _no_versions_response
+ self.assertEqual(None, latest_pub.get_latest_version())
+
+ def test_get_latest_version_example(self):
+ latest_pub = self.make_latest_publication()
+ latest_pub._get_lp_info = lambda: _example_response
+ self.assertEqual("2.1.4-0ubuntu1", latest_pub.get_latest_version())
+
+ def DONT_test_get_latest_version_from_launchpad(self):
+ latest_pub = self.make_latest_publication()
+ self.assertIsNot(None, latest_pub.get_latest_version())
More information about the bazaar-commits
mailing list