<html><head><meta name="color-scheme" content="light dark"></head><body><pre style="word-wrap: break-word; white-space: pre-wrap;"># 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 &lt;http://www.gnu.org/licenses/&gt;.

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

from collections import deque
from json import dumps
import tempfile
import unittest

try:
    from json import JSONDecodeError
except ImportError:
    JSONDecodeError = ValueError

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"""&lt;?xml version="1.0"?&gt;
&lt;application xmlns="http://research.sun.com/wadl/2006/10"&gt;
  &lt;resources base="http://www.example.com/"&gt;
    &lt;resource path="" type="#service-root"/&gt;
  &lt;/resources&gt;

  &lt;resource_type id="service-root"&gt;
    &lt;method name="GET" id="service-root-get"&gt;
      &lt;response&gt;
        &lt;representation href="#service-root-json"/&gt;
      &lt;/response&gt;
    &lt;/method&gt;
  &lt;/resource_type&gt;

  &lt;representation id="service-root-json" mediaType="application/json"/&gt;
&lt;/application&gt;
"""

# 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,
        )
</pre></body></html>