1#@!#!123s

: / proc / self / root / proc / self / root / lib / python3 / dist-packages / launchpadlib / tests /
Filename : test_http.py
back
# Copyright 2010 Canonical Ltd.

# This file is part of launchpadlib.
#
# launchpadlib is free software: you can redistribute it and/or modify it
# under the terms of the GNU Lesser General Public License as published by the
# Free Software Foundation, version 3 of the License.
#
# launchpadlib 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 Lesser General Public License
# for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with launchpadlib. If not, see <http://www.gnu.org/licenses/>.

"""Tests for the LaunchpadOAuthAwareHTTP class."""

from collections import deque
import tempfile
import unittest

try:
    from json import dumps
    JSONDecodeError = ValueError
except ImportError:
    from simplejson import dumps, JSONDecodeError

from launchpadlib.errors import Unauthorized
from launchpadlib.credentials import UnencryptedFileCredentialStore
from launchpadlib.launchpad import (
    Launchpad,
    LaunchpadOAuthAwareHttp,
    )
from launchpadlib.testing.helpers import NoNetworkAuthorizationEngine


# The simplest WADL that looks like a representation of the service root.
SIMPLE_WADL = b'''<?xml version="1.0"?>
<application xmlns="http://research.sun.com/wadl/2006/10">
  <resources base="http://www.example.com/">
    <resource path="" type="#service-root"/>
  </resources>

  <resource_type id="service-root">
    <method name="GET" id="service-root-get">
      <response>
        <representation href="#service-root-json"/>
      </response>
    </method>
  </resource_type>

  <representation id="service-root-json" mediaType="application/json"/>
</application>
'''

# The simplest JSON that looks like a representation of the service root.
SIMPLE_JSON = dumps({}).encode('utf-8')


class Response:
    """A fake HTTP response object."""
    def __init__(self, status, content):
        self.status = status
        self.content = content


class SimulatedResponsesHttp(LaunchpadOAuthAwareHttp):
    """Responds to HTTP requests by shifting responses off a stack."""

    def __init__(self, responses, *args):
        """Constructor.

        :param responses: A list of HttpResponse objects to use
            in response to requests.
        """
        super(SimulatedResponsesHttp, self).__init__(*args)
        self.sent_responses = []
        self.unsent_responses = responses
        self.cache = None

    def _request(self, *args):
        response = self.unsent_responses.popleft()
        self.sent_responses.append(response)
        return self.retry_on_bad_token(response, response.content, *args)


class SimulatedResponsesLaunchpad(Launchpad):

    # Every Http object generated by this class will return these
    # responses, in order.
    responses = []

    def httpFactory(self, *args):
        return SimulatedResponsesHttp(
            deque(self.responses), self, self.authorization_engine, *args)

    @classmethod
    def credential_store_factory(cls, credential_save_failed):
        return UnencryptedFileCredentialStore(
            tempfile.mkstemp()[1], credential_save_failed)


class SimulatedResponsesTestCase(unittest.TestCase):
    """Test cases that give fake responses to launchpad's HTTP requests."""

    def setUp(self):
        """Clear out the list of simulated responses."""
        SimulatedResponsesLaunchpad.responses = []
        self.engine = NoNetworkAuthorizationEngine(
            'http://api.example.com/', 'application name')

    def launchpad_with_responses(self, *responses):
        """Use simulated HTTP responses to get a Launchpad object.

        The given Response objects will be sent, in order, in response
        to launchpadlib's requests.

        :param responses: Some number of Response objects.
        :return: The Launchpad object, assuming that errors in the
            simulated requests didn't prevent one from being created.
        """
        SimulatedResponsesLaunchpad.responses = responses
        return SimulatedResponsesLaunchpad.login_with(
            'application name', authorization_engine=self.engine)


class TestAbilityToParseData(SimulatedResponsesTestCase):
    """Test launchpadlib's ability to handle the sample data.

    To create a Launchpad object, two HTTP requests must succeed and
    return usable data: the requests for the WADL and JSON
    representations of the service root. This test shows that the
    minimal data in SIMPLE_WADL and SIMPLE_JSON is good enough to
    create a Launchpad object.
    """

    def test_minimal_data(self):
        """Make sure that launchpadlib can use the minimal data."""
        self.launchpad_with_responses(
            Response(200, SIMPLE_WADL),
            Response(200, SIMPLE_JSON))

    def test_bad_wadl(self):
        """Show that bad WADL causes an exception."""
        self.assertRaises(
            SyntaxError, self.launchpad_with_responses,
            Response(200, b"This is not WADL."),
            Response(200, SIMPLE_JSON))

    def test_bad_json(self):
        """Show that bad JSON causes an exception."""
        self.assertRaises(
            JSONDecodeError, self.launchpad_with_responses,
            Response(200, SIMPLE_WADL),
            Response(200, b"This is not JSON."))


class TestTokenFailureDuringRequest(SimulatedResponsesTestCase):
    """Test access token failures during a request.

    launchpadlib makes two HTTP requests on startup, to get the WADL
    and JSON representations of the service root. If Launchpad
    receives a 401 error during this process, it will acquire a fresh
    access token and try again.
    """

    def test_good_token(self):
        """If our token is good, we never get another one."""
        SimulatedResponsesLaunchpad.responses = [
            Response(200, SIMPLE_WADL),
            Response(200, SIMPLE_JSON)]

        self.assertEqual(self.engine.access_tokens_obtained, 0)
        SimulatedResponsesLaunchpad.login_with(
            'application name', authorization_engine=self.engine)
        self.assertEqual(self.engine.access_tokens_obtained, 1)

    def test_bad_token(self):
        """If our token is bad, we get another one."""
        SimulatedResponsesLaunchpad.responses = [
            Response(401, b"Invalid token."),
            Response(200, SIMPLE_WADL),
            Response(200, SIMPLE_JSON)]

        self.assertEqual(self.engine.access_tokens_obtained, 0)
        SimulatedResponsesLaunchpad.login_with(
            'application name', authorization_engine=self.engine)
        self.assertEqual(self.engine.access_tokens_obtained, 2)

    def test_expired_token(self):
        """If our token is expired, we get another one."""

        SimulatedResponsesLaunchpad.responses = [
            Response(401, b"Expired token."),
            Response(200, SIMPLE_WADL),
            Response(200, SIMPLE_JSON)]

        self.assertEqual(self.engine.access_tokens_obtained, 0)
        SimulatedResponsesLaunchpad.login_with(
            'application name', authorization_engine=self.engine)
        self.assertEqual(self.engine.access_tokens_obtained, 2)

    def test_unknown_token(self):
        """If our token is unknown, we get another one."""

        SimulatedResponsesLaunchpad.responses = [
            Response(401, b"Unknown access token."),
            Response(200, SIMPLE_WADL),
            Response(200, SIMPLE_JSON)]

        self.assertEqual(self.engine.access_tokens_obtained, 0)
        SimulatedResponsesLaunchpad.login_with(
            'application name', authorization_engine=self.engine)
        self.assertEqual(self.engine.access_tokens_obtained, 2)

    def test_delayed_error(self):
        """We get another token no matter when the error happens."""
        SimulatedResponsesLaunchpad.responses = [
            Response(200, SIMPLE_WADL),
            Response(401, b"Expired token."),
            Response(200, SIMPLE_JSON)]

        self.assertEqual(self.engine.access_tokens_obtained, 0)
        SimulatedResponsesLaunchpad.login_with(
            'application name', authorization_engine=self.engine)
        self.assertEqual(self.engine.access_tokens_obtained, 2)

    def test_many_errors(self):
        """We'll keep getting new tokens as long as tokens are the problem."""
        SimulatedResponsesLaunchpad.responses = [
            Response(401, b"Invalid token."),
            Response(200, SIMPLE_WADL),
            Response(401, b"Expired token."),
            Response(401, b"Invalid token."),
            Response(200, SIMPLE_JSON)]
        self.assertEqual(self.engine.access_tokens_obtained, 0)
        SimulatedResponsesLaunchpad.login_with(
            'application name', authorization_engine=self.engine)
        self.assertEqual(self.engine.access_tokens_obtained, 4)

    def test_other_unauthorized(self):
        """If the token is not at fault, a 401 error raises an exception."""

        SimulatedResponsesLaunchpad.responses = [
            Response(401, b"Some other error.")]

        self.assertRaises(
            Unauthorized, SimulatedResponsesLaunchpad.login_with,
            'application name', authorization_engine=self.engine)