Logo Search packages:      
Sourcecode: ubuntuone-couch version File versions  Download package

test_couch.py

# Copyright 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 version 3, as published
# by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranties of
# MERCHANTABILITY, SATISFACTORY QUALITY, 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, see <http://www.gnu.org/licenses/>.

"""Tests for u1couch.query."""

from twisted.trial.unittest import TestCase
from mocker import Mocker, ANY, MATCH

import json

from oauth import oauth
from ubuntuone.couch import (
    get_user_info, request, OAuthAuthenticationError, UnknownError)

CONSUMER_KEY = u'this_consumer_key'
CONSUMER_SECRET = u'sssssh!'
TOKEN_KEY = u'tokentokentoken'
TOKEN_SECRET = u'ssssssshhhhhh!'

CONSUMER = oauth.OAuthConsumer(CONSUMER_KEY, CONSUMER_SECRET)
TOKEN = oauth.OAuthToken(TOKEN_KEY, TOKEN_SECRET)

URL = 'https://example.com'


00037 class QueryTestCase(TestCase):
    """Test case for u1couch.query."""

    def setUp(self):
        self.mocker = Mocker()

    def tearDown(self):
        self.mocker.restore()
        self.mocker.verify()

00047     def test_get_user_info(self):
        """Test get_user_info parses json correctly."""
        mock_request = self.mocker.replace("ubuntuone.couch.auth.request")
        mock_request(
            URL, consumer_key=None, consumer_secret=None, access_token=None,
            token_secret=None)
        self.mocker.result((
            {'status': '200'},
            '{"couchdb_root": "https://couchdb.one.ubuntu.com/u/abc/def/1337",'
            ' "id": 1337}'))
        self.mocker.replay()
        user_id, root = get_user_info(URL)
        self.assertEquals(1337, user_id)
        self.assertEquals(
            "https://couchdb.one.ubuntu.com/u/abc/def/1337", root)

00063     def test_get_user_info_404(self):
        """Test get_user_info handles non-success statuses."""
        mock_request = self.mocker.replace("ubuntuone.couch.auth.request")
        mock_request(
            URL, consumer_key=None, consumer_secret=None, access_token=None,
            token_secret=None)
        self.mocker.result(({'status': '404'}, ''))
        self.mocker.replay()
        self.assertRaises(UnknownError, get_user_info, URL)

00073     def test_request(self):
        """Test a full request."""
        fake_result = {
            'committed_update_seq': 654,
            'compact_running': False,
            'db_name': 'u/abc/def/1337/contacts',
            'disk_format_version': 5,
            'disk_size': 929892,
            'doc_count': 626,
            'doc_del_count': 1,
            'instance_start_time': '1297965776474824',
            'purge_seq': 0,
            'update_seq': 654}
        fake_json = json.dumps(fake_result)
        mock_get_oauth_credentials = self.mocker.replace(
            'ubuntuone.couch.auth.get_oauth_credentials')
        mock_get_oauth_credentials()
        self.mocker.result({
            u'token': TOKEN_KEY,
            u'token_secret': TOKEN_SECRET,
            u'consumer_key': CONSUMER_KEY,
            u'consumer_secret': CONSUMER_SECRET})
        mock_get_user_info = self.mocker.replace(
            'ubuntuone.couch.get_user_info')
        mock_get_user_info(
            'https://one.ubuntu.com/api/account/', consumer_key=CONSUMER_KEY,
            consumer_secret=CONSUMER_SECRET, access_token=TOKEN_KEY,
            token_secret=TOKEN_SECRET)
        self.mocker.result((
            1337, "https://couchdb.one.ubuntu.com/u/abc/def/1337"))
        Http = self.mocker.replace("httplib2.Http")  # pylint: disable=C0103
        http = Http()
        http.request(
            'https://couchdb.one.ubuntu.com/u%2Fabc%2Fdef%2F1337%2Fcontacts',
            method='GET', headers=ANY, body=None)
        self.mocker.result(({'status': '200'}, fake_json))
        self.mocker.replay()
        result = request('contacts')
        self.assertEquals(fake_result, result)

00113     def test_request_server_override(self):
        """Test a full request with server_override."""
        fake_result = {
            'committed_update_seq': 654,
            'compact_running': False,
            'db_name': 'u/abc/def/1337/contacts',
            'disk_format_version': 5,
            'disk_size': 929892,
            'doc_count': 626,
            'doc_del_count': 1,
            'instance_start_time': '1297965776474824',
            'purge_seq': 0,
            'update_seq': 654}
        fake_json = json.dumps(fake_result)
        mock_get_oauth_credentials = self.mocker.replace(
            'ubuntuone.couch.auth.get_oauth_credentials')
        mock_get_oauth_credentials()
        self.mocker.result({
            u'token': TOKEN_KEY,
            u'token_secret': TOKEN_SECRET,
            u'consumer_key': CONSUMER_KEY,
            u'consumer_secret': CONSUMER_SECRET})
        mock_get_user_info = self.mocker.replace(
            'ubuntuone.couch.get_user_info')
        mock_get_user_info(
            'https://one.ubuntu.com/api/account/', consumer_key=CONSUMER_KEY,
            consumer_secret=CONSUMER_SECRET, access_token=TOKEN_KEY,
            token_secret=TOKEN_SECRET)
        self.mocker.result((
            1337, "https://couchdb.one.ubuntu.com/u/abc/def/1337"))
        Http = self.mocker.replace("httplib2.Http")  # pylint: disable=C0103
        http = Http()
        http.request(
            'https://example.org/u%2Fabc%2Fdef%2F1337%2Fcontacts',
            method='GET', headers=ANY, body=None)
        self.mocker.result(({'status': '200'}, fake_json))
        self.mocker.replay()
        result = request('contacts', server_override='example.org')
        self.assertEquals(fake_result, result)

00153     def test_request_plaintext(self):
        """Test a full request with PLAINTEXT."""
        fake_result = {
            'committed_update_seq': 654,
            'compact_running': False,
            'db_name': 'u/abc/def/1337/contacts',
            'disk_format_version': 5,
            'disk_size': 929892,
            'doc_count': 626,
            'doc_del_count': 1,
            'instance_start_time': '1297965776474824',
            'purge_seq': 0,
            'update_seq': 654}
        fake_json = json.dumps(fake_result)
        mock_get_oauth_credentials = self.mocker.replace(
            'ubuntuone.couch.auth.get_oauth_credentials')
        mock_get_oauth_credentials()
        self.mocker.result({
            u'token': TOKEN_KEY,
            u'token_secret': TOKEN_SECRET,
            u'consumer_key': CONSUMER_KEY,
            u'consumer_secret': CONSUMER_SECRET})
        mock_get_user_info = self.mocker.replace(
            'ubuntuone.couch.get_user_info')
        mock_get_user_info(
            'https://one.ubuntu.com/api/account/', consumer_key=CONSUMER_KEY,
            consumer_secret=CONSUMER_SECRET, access_token=TOKEN_KEY,
            token_secret=TOKEN_SECRET)
        self.mocker.result((
            1337, "https://couchdb.one.ubuntu.com/u/abc/def/1337"))
        Http = self.mocker.replace("httplib2.Http")  # pylint: disable=C0103
        http = Http()
        http.request(
            'https://couchdb.one.ubuntu.com/u%2Fabc%2Fdef%2F1337%2Fcontacts',
            method='GET', headers=ANY, body=None)
        self.mocker.result(({'status': '200'}, fake_json))
        self.mocker.replay()
        result = request('contacts', sig_meth='PLAINTEXT')
        self.assertEquals(fake_result, result)

00193     def test_request_all_dbs(self):
        """Test a full request to _all_dbs."""
        fake_result = [
            u'u/abc/def/1337/dc_trash', u'u/abc/def/1337/lernid',
            u'u/abc/def/1337/test_suite_db']
        fake_json = json.dumps(fake_result)
        mock_get_oauth_credentials = self.mocker.replace(
            'ubuntuone.couch.auth.get_oauth_credentials')
        mock_get_oauth_credentials()
        self.mocker.result({
            u'token': TOKEN_KEY,
            u'token_secret': TOKEN_SECRET,
            u'consumer_key': CONSUMER_KEY,
            u'consumer_secret': CONSUMER_SECRET})
        mock_get_user_info = self.mocker.replace(
            'ubuntuone.couch.get_user_info')
        mock_get_user_info(
            'https://one.ubuntu.com/api/account/', consumer_key=CONSUMER_KEY,
            consumer_secret=CONSUMER_SECRET, access_token=TOKEN_KEY,
            token_secret=TOKEN_SECRET)
        self.mocker.result((
            1337, "https://couchdb.one.ubuntu.com/u/abc/def/1337"))
        Http = self.mocker.replace("httplib2.Http")  # pylint: disable=C0103
        http = Http()
        http.request(
            'https://couchdb.one.ubuntu.com/_all_dbs?user_id=1337',
            method='GET', headers=ANY, body=None)
        self.mocker.result(({'status': '200'}, fake_json))
        self.mocker.replay()
        result = request('_all_dbs')
        self.assertEquals(fake_result, result)

00225     def test_request_all_dbs_with_extra_headers(self):
        """Test a full request to _all_dbs with extra headers passed."""
        fake_result = [
            u'u/abc/def/1337/dc_trash', u'u/abc/def/1337/lernid',
            u'u/abc/def/1337/test_suite_db']
        fake_json = json.dumps(fake_result)
        mock_get_oauth_credentials = self.mocker.replace(
            'ubuntuone.couch.auth.get_oauth_credentials')
        mock_get_oauth_credentials()
        self.mocker.result({
            u'token': TOKEN_KEY,
            u'token_secret': TOKEN_SECRET,
            u'consumer_key': CONSUMER_KEY,
            u'consumer_secret': CONSUMER_SECRET})
        mock_get_user_info = self.mocker.replace(
            'ubuntuone.couch.get_user_info')
        mock_get_user_info(
            'https://one.ubuntu.com/api/account/', consumer_key=CONSUMER_KEY,
            consumer_secret=CONSUMER_SECRET, access_token=TOKEN_KEY,
            token_secret=TOKEN_SECRET)
        self.mocker.result((
            1337, "https://couchdb.one.ubuntu.com/u/abc/def/1337"))
        Http = self.mocker.replace("httplib2.Http")  # pylint: disable=C0103
        http = Http()

        def _check_headers(param):
            """Confirm that extra headers really are passed."""
            return (param.get("X-Men") == "unite") and \
                (param.get("X-tasy") == "bad for you")
        http.request(
            'https://couchdb.one.ubuntu.com/_all_dbs?user_id=1337',
            method='GET', headers=MATCH(_check_headers), body=None)
        self.mocker.result(({'status': '200'}, fake_json))
        self.mocker.replay()
        result = request('_all_dbs',
            extra_headers=['X-Men: unite', 'X-tasy: bad for you'])
        self.assertEquals(fake_result, result)

00263     def test_request_400(self):
        """Test a full request on 400."""
        mock_get_oauth_credentials = self.mocker.replace(
            'ubuntuone.couch.auth.get_oauth_credentials')
        mock_get_oauth_credentials()
        self.mocker.result({
            u'token': TOKEN_KEY,
            u'token_secret': TOKEN_SECRET,
            u'consumer_key': CONSUMER_KEY,
            u'consumer_secret': CONSUMER_SECRET})
        mock_get_user_info = self.mocker.replace(
            'ubuntuone.couch.get_user_info')
        mock_get_user_info(
            'https://one.ubuntu.com/api/account/', consumer_key=CONSUMER_KEY,
            consumer_secret=CONSUMER_SECRET, access_token=TOKEN_KEY,
            token_secret=TOKEN_SECRET)
        self.mocker.result((
            1337, "https://couchdb.one.ubuntu.com/u/abc/def/1337"))
        Http = self.mocker.replace("httplib2.Http")  # pylint: disable=C0103
        http = Http()
        http.request(
            'https://couchdb.one.ubuntu.com/u%2Fabc%2Fdef%2F1337%2Fcontacts',
            method='GET', headers=ANY, body=None)
        self.mocker.result(({'status': '400'}, ''))
        self.mocker.replay()
        self.assertRaises(OAuthAuthenticationError, request, 'contacts')

00290     def test_request_401(self):
        """Test a full request on 401."""
        mock_get_oauth_credentials = self.mocker.replace(
            'ubuntuone.couch.auth.get_oauth_credentials')
        mock_get_oauth_credentials()
        self.mocker.result({
            u'token': TOKEN_KEY,
            u'token_secret': TOKEN_SECRET,
            u'consumer_key': CONSUMER_KEY,
            u'consumer_secret': CONSUMER_SECRET})
        mock_get_user_info = self.mocker.replace(
            'ubuntuone.couch.get_user_info')
        mock_get_user_info(
            'https://one.ubuntu.com/api/account/', consumer_key=CONSUMER_KEY,
            consumer_secret=CONSUMER_SECRET, access_token=TOKEN_KEY,
            token_secret=TOKEN_SECRET)
        self.mocker.result((
            1337, "https://couchdb.one.ubuntu.com/u/abc/def/1337"))
        Http = self.mocker.replace("httplib2.Http")  # pylint: disable=C0103
        http = Http()
        http.request(
            'https://couchdb.one.ubuntu.com/u%2Fabc%2Fdef%2F1337%2Fcontacts',
            method='GET', headers=ANY, body=None)
        self.mocker.result(({'status': '401'}, ''))
        self.mocker.replay()
        self.assertRaises(OAuthAuthenticationError, request, 'contacts')

00317     def test_request_911(self):
        """Test a full request on other status codes."""
        mock_get_oauth_credentials = self.mocker.replace(
            'ubuntuone.couch.auth.get_oauth_credentials')
        mock_get_oauth_credentials()
        self.mocker.result({
            u'token': TOKEN_KEY,
            u'token_secret': TOKEN_SECRET,
            u'consumer_key': CONSUMER_KEY,
            u'consumer_secret': CONSUMER_SECRET})
        mock_get_user_info = self.mocker.replace(
            'ubuntuone.couch.get_user_info')
        mock_get_user_info(
            'https://one.ubuntu.com/api/account/', consumer_key=CONSUMER_KEY,
            consumer_secret=CONSUMER_SECRET, access_token=TOKEN_KEY,
            token_secret=TOKEN_SECRET)
        self.mocker.result((
            1337, "https://couchdb.one.ubuntu.com/u/abc/def/1337"))
        Http = self.mocker.replace("httplib2.Http")  # pylint: disable=C0103
        http = Http()
        http.request(
            'https://couchdb.one.ubuntu.com/u%2Fabc%2Fdef%2F1337%2Fcontacts',
            method='GET', headers=ANY, body=None)
        self.mocker.result(({'status': '911'}, ''))
        self.mocker.replay()
        self.assertRaises(UnknownError, request, 'contacts')

Generated by  Doxygen 1.6.0   Back to index