Google Cloud Platform Blog
Product updates, customer stories, and tips and tricks on Google Cloud Platform
Unit-Testing cron handlers in Google App Engine
2015年7月6日星期一
Unit testing your code is a best practice in software development. By running small automated tests to individually verify each unit of code, such as a module, class or function, you can catch and debug errors early in your development process.
Google App Engine
provides strong support for unit testing with Local Unit Testing tools, currently available for
Python
,
Java
, and
Go
. With local unit testing, you run the unit tests within your development environment, without calling any remote components. The Local Unit Testing tools offer service stubs to simulate many App Engine services. You can use stubs as needed to exercise your application code in local unit tests. You can also use open source packages like
NoseGAE
to further simplify the process of writing local App Engine unit tests.
Local unit tests using service stubs, however, handle routing and login constraints differently than code that calls the services directly. You have to keep these differences in mind when you design your tests.
When a customer had problems unit testing an App Engine
cron
handler, I was, of course, eager to help. The cron handler was defined by the following entry in
app.yaml
:
- url: /crontask.*
script: thecron.app
login: admin_only
The App Engine Cron Service
recommends
that you limit access to URLs used by scheduled tasks to
administrator accounts. That’s what the
login: admin_only
setting does.
The customer wanted to check that non-admin users would indeed be blocked from those URLs.
The customer started with a placeholder implementation for the cron handler,
thecron.py
:
import webapp2
import time
class TestCronHandler(webapp2.RequestHandler):
def get(self):
self.response.headers['Content-Type'] = 'text/plain'
self.response.write('Cron: {}'.format(time.time()))
app = webapp2.WSGIApplication([
('/crontask/test', TestCronHandler),
], debug=True)
Then they wrote a unit test ,
ut.py
.
Note
: the following test code uses the
mock
module. Run
pip install mock
to make the
mock
module available to your local Python 2.7 installation.
import sys
# configure unit testing for the case
# where the App Engine SDK is installed in
# /usr/local/google_appengine
sdk_path = '/usr/local/google_appengine'
sys.path.insert(0, sdk_path)
import dev_appserver
dev_appserver.fix_sys_path()
import mock
import unittest
import webapp2
from google.appengine.ext import testbed
import thecron
class CronTestCase(unittest.TestCase):
def setUp(self):
self.testbed = testbed.Testbed()
self.testbed.activate()
self.testbed.init_user_stub()
def _aux(self, is_admin):
self.testbed.setup_env(
USER_EMAIL = 'test@example.com',
USER_ID = '123',
USER_IS_ADMIN = str(int(bool(is_admin))),
overwrite = True)
request = webapp2.Request.blank('/crontask/test')
with mock.patch.object(thecron.time,
'time', return_value=12345678):
response = request.get_response(thecron.app)
return response
def testAdminWorks(self):
response = self._aux(True)
self.assertEqual(response.status_int, 200)
self.assertEqual(response.body, 'Cron: 12345678')
if __name__ == '__main__':
unittest.main()
This first test,
testAdminWorks
, passed with flying colors, so the user added a second one:
def testNonAdminFails(self):
response = self._aux(False)
self.assertEqual(response.status_int, 401)
But the new test failed--the status was 200 (success),
not
401 (forbidden) as expected.
The problem was that unit tests do
not
go all the way back to
app.yaml
to get the complete routing and login constraints as would an application calling the services, not the unit-testing stubs. This test reached into the secondary routing in
thecron.py
, which doesn’t impose login constraints. The
app.yaml
routing and everything else, such as mime-types, login constraints, etc., get tested only by tests calling the services instead of the unit-testing stubs.
So the customer added an admin-checking decorator,
needs_admin
, to
thecron.py
:
def needs_admin(func):
def inner(self, *args, **kwargs):
if users.get_current_user():
if not users.is_current_user_admin():
webapp2.abort(401)
return func(self, *args, **kwargs)
return self.redirect(users.create_login_url(request.url))
return inner
Then they decorated
CronHandler.get
with it:
class CronHandler(webapp2.RequestHandler):
@needs_admin
def get(self): # etc, as before
Now the local unit tests calling the stub version of the cron service work, but an end-to-end test using the actual App Engine Cron Service functionality fails with status 401. What’s going on?
Long story short -- the App Engine Cron Service doesn’t log-in
any
user as it visits the appointed URLs -- therefore, in the handler,
users.get_current_user()
returns
None
. Instead, the Cron Service sets a special request header --
X-AppEngine-Cron: true.
This is a header that application code can fully trust, since App Engine removes such headers if they’re set in an external request.
All that the customer needed, to get their unit tests, end-to-end tests, local development application, and deployed application, working, was a slight modification to their
needs_admin
decorator:
def needs_admin(func):
def inner(self, *args, **kwargs):
if self.request.headers.get('X-AppEngine-Cron') == 'true':
return func(self, *args, **kwargs)
if users.get_current_user():
if not users.is_current_user_admin():
webapp2.abort(401)
return func(self, *args, **kwargs)
return self.redirect(users.create_login_url(request.url))
return inner
The new
if
statement handles the cron job case, mocked or not. The second
if
, as before, ensures that non-admin (or non-logged-in) users are blocked.
The moral is,
do
invest time and care in unit-testing. It ensures the present and ongoing quality of your code. The App Engine
Local Unit Testing
tools and open-source add-ons like NoseGAE simplify the process of writing and running unit tests.
- Posted By
Alex Martelli, Cloud Technical Support
Free Trial
GCP Blogs
Big Data & Machine Learning
Kubernetes
GCP Japan Blog
Firebase Blog
Apigee Blog
Popular Posts
Understanding Cloud Pricing
World's largest event dataset now publicly available in BigQuery
A look inside Google’s Data Center Networks
New in Google Cloud Storage: auto-delete, regional buckets and faster uploads
Enter the Andromeda zone - Google Cloud Platform’s latest networking stack
标签
Announcements
193
Big Data & Machine Learning
134
Compute
271
Containers & Kubernetes
92
CRE
27
Customers
107
Developer Tools & Insights
151
Events
38
Infrastructure
44
Management Tools
87
Networking
43
Open
1
Open Source
135
Partners
102
Pricing
28
Security & Identity
85
Solutions
24
Stackdriver
24
Storage & Databases
164
Weekly Roundups
20
Feed
Subscribe by email
Demonstrate your proficiency to design, build and manage solutions on Google Cloud Platform.
Learn More
Technical questions? Check us out on
Stack Overflow
.
Subscribe to
our monthly newsletter
.
Google
on
Follow @googlecloud
Follow
Follow