/brz/remove-bazaar

# Copyright (C) 2011 Canonical Ltd

#

# This program is free software; you can redistribute it and/or modify

# it under the terms of the GNU General Public License as published by

# the Free Software Foundation; either version 2 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU General Public License for more details.

#

# You should have received a copy of the GNU General Public License

# along with this program; if not, write to the Free Software

# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA

"""Tools for dealing with the Launchpad API without using launchpadlib.

"""

import socket

from bzrlib import tests

from bzrlib.plugins import launchpad

from bzrlib.plugins.launchpad import lp_api_lite

class _JSONParserFeature(tests.Feature):

    def _probe(self):

        return lp_api_lite.json is not None

    def feature_name(self):

        return 'simplejson or json'

JSONParserFeature = _JSONParserFeature()

_example_response = r"""

{

    "total_size": 2,

    "start": 0,

    "next_collection_link": "https://api.launchpad.net/1.0/ubuntu/+archive/primary?distro_series=%2Fubuntu%2Flucid&exact_match=true&source_name=%22bzr%22&status=Published&ws.op=getPublishedSources&ws.start=1&ws.size=1",

    "entries": [

        {

            "package_creator_link": "https://api.launchpad.net/1.0/~maxb",

            "package_signer_link": "https://api.launchpad.net/1.0/~jelmer",

            "source_package_name": "bzr",

            "removal_comment": null,

            "display_name": "bzr 2.1.4-0ubuntu1 in lucid",

            "date_made_pending": null,

            "source_package_version": "2.1.4-0ubuntu1",

            "date_superseded": null,

            "http_etag": "\"9ba966152dec474dc0fe1629d0bbce2452efaf3b-5f4c3fbb3eaf26d502db4089777a9b6a0537ffab\"",

            "self_link": "https://api.launchpad.net/1.0/ubuntu/+archive/primary/+sourcepub/1750327",

            "distro_series_link": "https://api.launchpad.net/1.0/ubuntu/lucid",

            "component_name": "main",

            "status": "Published",

            "date_removed": null,

            "pocket": "Updates",

            "date_published": "2011-05-30T06:09:58.653984+00:00",

            "removed_by_link": null,

            "section_name": "devel",

            "resource_type_link": "https://api.launchpad.net/1.0/#source_package_publishing_history",

            "archive_link": "https://api.launchpad.net/1.0/ubuntu/+archive/primary",

            "package_maintainer_link": "https://api.launchpad.net/1.0/~ubuntu-devel-discuss-lists",

            "date_created": "2011-05-30T05:19:12.233621+00:00",

            "scheduled_deletion_date": null

        }

    ]

}"""

_no_versions_response = '{"total_size": 0, "start": 0, "entries": []}'

class TestLatestPublication(tests.TestCase):

    def make_latest_publication(self, archive='ubuntu', series='natty',

                                project='bzr'):

        return lp_api_lite.LatestPublication(archive, series, project)

    def test_init(self):

        latest_pub = self.make_latest_publication()

        self.assertEqual('ubuntu', latest_pub._archive)

        self.assertEqual('natty', latest_pub._series)

        self.assertEqual('bzr', latest_pub._project)

        self.assertEqual('Release', latest_pub._pocket)

    def test__archive_URL(self):

        latest_pub = self.make_latest_publication()

        self.assertEqual(

            'https://api.launchpad.net/1.0/ubuntu/+archive/primary',

            latest_pub._archive_URL())

    def test__publication_status_for_ubuntu(self):

        latest_pub = self.make_latest_publication()

        self.assertEqual('Published', latest_pub._publication_status())

    def test__publication_status_for_debian(self):

        latest_pub = self.make_latest_publication(archive='debian')

        self.assertEqual('Pending', latest_pub._publication_status())

    def test_pocket(self):

        latest_pub = self.make_latest_publication(series='natty-proposed')

        self.assertEqual('natty', latest_pub._series)

        self.assertEqual('Proposed', latest_pub._pocket)

    def test_series_None(self):

        latest_pub = self.make_latest_publication(series=None)

        self.assertEqual('ubuntu', latest_pub._archive)

        self.assertEqual(None, latest_pub._series)

        self.assertEqual('bzr', latest_pub._project)

        self.assertEqual('Release', latest_pub._pocket)

    def test__query_params(self):

        latest_pub = self.make_latest_publication()

        self.assertEqual({'ws.op': 'getPublishedSources',

                          'exact_match': 'true',

                          'source_name': '"bzr"',

                          'status': 'Published',

                          'ws.size': '1',

                          'distro_series': '/ubuntu/natty',

                          'pocket': 'Release',

                         }, latest_pub._query_params())

    def test__query_params_no_series(self):

        latest_pub = self.make_latest_publication(series=None)

        self.assertEqual({'ws.op': 'getPublishedSources',

                          'exact_match': 'true',

                          'source_name': '"bzr"',

                          'status': 'Published',

                          'ws.size': '1',

                          'pocket': 'Release',

                         }, latest_pub._query_params())

    def test__query_params_pocket(self):

        latest_pub = self.make_latest_publication(series='natty-proposed')

        self.assertEqual({'ws.op': 'getPublishedSources',

                          'exact_match': 'true',

                          'source_name': '"bzr"',

                          'status': 'Published',

                          'ws.size': '1',

                          'distro_series': '/ubuntu/natty',

                          'pocket': 'Proposed',

                         }, latest_pub._query_params())

    def test__query_URL(self):

        latest_pub = self.make_latest_publication()

        # we explicitly sort params, so we can be sure this URL matches exactly

        self.assertEqual(

            'https://api.launchpad.net/1.0/ubuntu/+archive/primary'

            '?distro_series=%2Fubuntu%2Fnatty&exact_match=true'

            '&pocket=Release&source_name=%22bzr%22&status=Published'

            '&ws.op=getPublishedSources&ws.size=1',

            latest_pub._query_URL())

    def DONT_test__gracefully_handle_failed_rpc_connection(self):

        # TODO: This test kind of sucks. We intentionally create an arbitrary

        #       port and don't listen to it, because we want the request to fail.

        #       However, it seems to take 1s for it to timeout. Is there a way

        #       to make it fail faster?

        latest_pub = self.make_latest_publication()

        s = socket.socket()

        s.bind(('127.0.0.1', 0))

        addr, port = s.getsockname()

        latest_pub.LP_API_ROOT = 'http://%s:%s/' % (addr, port)

        s.close()

        self.assertIs(None, latest_pub._get_lp_info())

    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

        #       now, it is commented out for production tests.

        latest_pub = self.make_latest_publication()

        json_txt = latest_pub._get_lp_info()

        self.assertIsNot(None, json_txt)

        if lp_api_lite.json is None:

            # We don't have a way to parse the text

            return

        # The content should be a valid json result

        content = lp_api_lite.json.loads(json_txt)

        entries = content['entries'] # It should have an 'entries' field.

        # ws.size should mean we get 0 or 1, and there should be something

        self.assertEqual(1, len(entries))

        entry = entries[0]

        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):

        self.requireFeature(JSONParserFeature)

        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__parse_json_not_json(self):

        self.requireFeature(JSONParserFeature)

        latest_pub = self.make_latest_publication()

        self.assertIs(None, latest_pub._parse_json_info('Not_valid_json'))

    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_invalid_json(self):

        self.requireFeature(JSONParserFeature)

        latest_pub = self.make_latest_publication()

        latest_pub._get_lp_info = lambda: "not json"

        self.assertEqual(None, latest_pub.get_latest_version())

    def test_get_latest_version_no_versions(self):

        self.requireFeature(JSONParserFeature)

        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_missing_entries(self):

        # Launchpad's no-entries response does have an empty entries value.

        # However, lets test that we handle other failures without tracebacks

        self.requireFeature(JSONParserFeature)

        latest_pub = self.make_latest_publication()

        latest_pub._get_lp_info = lambda: '{}'

        self.assertEqual(None, latest_pub.get_latest_version())

    def test_get_latest_version_invalid_entries(self):

        # Make sure we sanely handle a json response we don't understand

        self.requireFeature(JSONParserFeature)

        latest_pub = self.make_latest_publication()

        latest_pub._get_lp_info = lambda: '{"entries": {"a": 1}}'

        self.assertEqual(None, latest_pub.get_latest_version())

    def test_get_latest_version_example(self):

        self.requireFeature(JSONParserFeature)

        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):

        self.requireFeature(JSONParserFeature)

        latest_pub = self.make_latest_publication()

        self.assertIsNot(None, latest_pub.get_latest_version())

class TestIsUpToDate(tests.TestCase):

    def assertPackageBranchRe(self, url, user, archive, series, project):

        m = launchpad._package_branch.search(url)

        if m is None:

            self.fail('package_branch regex did not match url: %s' % (url,))

        self.assertEqual(

            (user, archive, series, project),

            m.group('user', 'archive', 'series', 'project'))

    def assertNotPackageBranch(self, url):

        self.assertIs(None, launchpad._get_package_branch_info(url))

    def assertBranchInfo(self, url, archive, series, project):

        self.assertEqual((archive, series, project),

            launchpad._get_package_branch_info(url))

    def test_package_branch_regex(self):

        self.assertPackageBranchRe(

            'http://bazaar.launchpad.net/+branch/ubuntu/foo',

            None, 'ubuntu', None, 'foo')

        self.assertPackageBranchRe(

            'bzr+ssh://bazaar.launchpad.net/+branch/ubuntu/natty/foo',

            None, 'ubuntu', 'natty/', 'foo')

        self.assertPackageBranchRe(

            'sftp://bazaar.launchpad.net/+branch/debian/foo',

            None, 'debian', None, 'foo')

        self.assertPackageBranchRe(

            'http://bazaar.launchpad.net/+branch/debian/sid/foo',

            None, 'debian', 'sid/', 'foo')

        self.assertPackageBranchRe(

            'http://bazaar.launchpad.net/+branch'

            '/~ubuntu-branches/ubuntu/natty/foo/natty',

            '~ubuntu-branches/', 'ubuntu', 'natty/', 'foo')

        self.assertPackageBranchRe(

            'http://bazaar.launchpad.net/+branch'

            '/~user/ubuntu/natty/foo/test',

            '~user/', 'ubuntu', 'natty/', 'foo')

    def test_package_branch_doesnt_match(self):

        self.assertNotPackageBranch('http://example.com/ubuntu/foo')

        self.assertNotPackageBranch(

            'http://bazaar.launchpad.net/+branch/bzr')

        self.assertNotPackageBranch(

            'http://bazaar.launchpad.net/+branch/~bzr-pqm/bzr/bzr.dev')

        # Not a packaging branch because ~user isn't ~ubuntu-branches

        self.assertNotPackageBranch(

            'http://bazaar.launchpad.net/+branch'

            '/~user/ubuntu/natty/foo/natty')

    def test__get_package_branch_info(self):

        self.assertBranchInfo(

            'bzr+ssh://bazaar.launchpad.net/+branch/ubuntu/natty/foo',

            'ubuntu', 'natty', 'foo')

        self.assertBranchInfo(

            'bzr+ssh://bazaar.launchpad.net/+branch'

            '/~ubuntu-branches/ubuntu/natty/foo/natty',

            'ubuntu', 'natty', 'foo')

        self.assertBranchInfo(

            'http://bazaar.launchpad.net/+branch'

            '/~ubuntu-branches/debian/sid/foo/sid',

            'debian', 'sid', 'foo')

class TestGetMostRecentTag(tests.TestCaseWithMemoryTransport):

    def make_simple_builder(self):

        builder = self.make_branch_builder('tip')

        builder.build_snapshot('A', [], [

            ('add', ('', 'root-id', 'directory', None))])

        b = builder.get_branch()

        b.tags.set_tag('tip-1.0', 'A')

        return builder, b, b.tags.get_tag_dict()

    def test_get_most_recent_tag_tip(self):

        builder, b, tag_dict = self.make_simple_builder()

        self.assertEqual('tip-1.0',

                         lp_api_lite.get_most_recent_tag(tag_dict, b))

    def test_get_most_recent_tag_older(self):

        builder, b, tag_dict = self.make_simple_builder()

        builder.build_snapshot('B', ['A'], [])

        self.assertEqual('B', b.last_revision())

        self.assertEqual('tip-1.0',

                         lp_api_lite.get_most_recent_tag(tag_dict, b))