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
|
# Copyright 2016 Google LLC All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""App Engine standard application that runs basic system tests for
google.auth.app_engine.
This application has to run tests manually instead of using pytest because
pytest currently doesn't work on App Engine standard.
"""
import contextlib
import json
import sys
from StringIO import StringIO
import traceback
from google.appengine.api import app_identity
import google.auth
from google.auth import _helpers
from google.auth import app_engine
import google.auth.transport.urllib3
import urllib3.contrib.appengine
import webapp2
FAILED_TEST_TMPL = """
Test {} failed: {}
Stacktrace:
{}
Captured output:
{}
"""
TOKEN_INFO_URL = "https://www.googleapis.com/oauth2/v3/tokeninfo"
EMAIL_SCOPE = "https://www.googleapis.com/auth/userinfo.email"
HTTP = urllib3.contrib.appengine.AppEngineManager()
HTTP_REQUEST = google.auth.transport.urllib3.Request(HTTP)
def test_credentials():
credentials = app_engine.Credentials()
scoped_credentials = credentials.with_scopes([EMAIL_SCOPE])
scoped_credentials.refresh(None)
assert scoped_credentials.valid
assert scoped_credentials.token is not None
# Get token info and verify scope
url = _helpers.update_query(
TOKEN_INFO_URL, {"access_token": scoped_credentials.token}
)
response = HTTP_REQUEST(url=url, method="GET")
token_info = json.loads(response.data.decode("utf-8"))
assert token_info["scope"] == EMAIL_SCOPE
def test_default():
credentials, project_id = google.auth.default()
assert isinstance(credentials, app_engine.Credentials)
assert project_id == app_identity.get_application_id()
@contextlib.contextmanager
def capture():
"""Context manager that captures stderr and stdout."""
oldout, olderr = sys.stdout, sys.stderr
try:
out = StringIO()
sys.stdout, sys.stderr = out, out
yield out
finally:
sys.stdout, sys.stderr = oldout, olderr
def run_test_func(func):
with capture() as capsys:
try:
func()
return True, ""
except Exception as exc:
output = FAILED_TEST_TMPL.format(
func.func_name, exc, traceback.format_exc(), capsys.getvalue()
)
return False, output
def run_tests():
"""Runs all tests.
Returns:
Tuple[bool, str]: A tuple containing True if all tests pass, False
otherwise, and any captured output from the tests.
"""
status = True
output = ""
tests = (test_credentials, test_default)
for test in tests:
test_status, test_output = run_test_func(test)
status = status and test_status
output += test_output
return status, output
class MainHandler(webapp2.RequestHandler):
def get(self):
self.response.headers["content-type"] = "text/plain"
status, output = run_tests()
if not status:
self.response.status = 500
self.response.write(output)
app = webapp2.WSGIApplication([("/", MainHandler)], debug=True)
|
