/usr/lib/python2.7/dist-packages/launchpadlib/tests/test_http.py is in python-launchpadlib 1.10.2+ds-2.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 | # 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
try:
import unittest2 as unittest
except ImportError:
import unittest
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,
NoNetworkLaunchpad,
)
# The simplest WADL that looks like a representation of the service root.
SIMPLE_WADL = '''<?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({})
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."""
launchpad = 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, "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, "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.assertEquals(self.engine.access_tokens_obtained, 0)
launchpad = SimulatedResponsesLaunchpad.login_with(
'application name', authorization_engine=self.engine)
self.assertEquals(self.engine.access_tokens_obtained, 1)
def test_bad_token(self):
"""If our token is bad, we get another one."""
SimulatedResponsesLaunchpad.responses = [
Response(401, "Invalid token."),
Response(200, SIMPLE_WADL),
Response(200, SIMPLE_JSON)]
self.assertEquals(self.engine.access_tokens_obtained, 0)
launchpad = SimulatedResponsesLaunchpad.login_with(
'application name', authorization_engine=self.engine)
self.assertEquals(self.engine.access_tokens_obtained, 2)
def test_expired_token(self):
"""If our token is expired, we get another one."""
SimulatedResponsesLaunchpad.responses = [
Response(401, "Expired token."),
Response(200, SIMPLE_WADL),
Response(200, SIMPLE_JSON)]
self.assertEquals(self.engine.access_tokens_obtained, 0)
launchpad = SimulatedResponsesLaunchpad.login_with(
'application name', authorization_engine=self.engine)
self.assertEquals(self.engine.access_tokens_obtained, 2)
def test_unknown_token(self):
"""If our token is unknown, we get another one."""
SimulatedResponsesLaunchpad.responses = [
Response(401, "Unknown access token."),
Response(200, SIMPLE_WADL),
Response(200, SIMPLE_JSON)]
self.assertEquals(self.engine.access_tokens_obtained, 0)
launchpad = SimulatedResponsesLaunchpad.login_with(
'application name', authorization_engine=self.engine)
self.assertEquals(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, "Expired token."),
Response(200, SIMPLE_JSON)]
self.assertEquals(self.engine.access_tokens_obtained, 0)
launchpad = SimulatedResponsesLaunchpad.login_with(
'application name', authorization_engine=self.engine)
self.assertEquals(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, "Invalid token."),
Response(200, SIMPLE_WADL),
Response(401, "Expired token."),
Response(401, "Invalid token."),
Response(200, SIMPLE_JSON)]
self.assertEquals(self.engine.access_tokens_obtained, 0)
launchpad = SimulatedResponsesLaunchpad.login_with(
'application name', authorization_engine=self.engine)
self.assertEquals(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, "Some other error.")]
self.assertRaises(
Unauthorized, SimulatedResponsesLaunchpad.login_with,
'application name', authorization_engine=self.engine)
|