script: thecron.app
login: admin_only
App Engine Cron Service では、スケジュールされているタスクが使用する
URL へのアクセスを管理者アカウントのみに制限することが推奨 されています。これは
login: admin_only の設定で実行されます。
お客様は管理者以外のユーザーによる、これらの URL へのアクセスがブロックされるかどうか確認しようとしました。
そこでまず、cron ハンドラ
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)
次にユニット テスト
ut.py が書き込まれました。
(注:下記のテスト コードでは
mock モジュールが使用されています。ローカルの Python 2.7 のインストールで
mock モジュールを使用できるようにするには、
pip install mock を実行します。)
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_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()
最初のテスト
testAdminWorks は問題なく通過したため、2 つめのテストが追加されました。
def testNonAdminFails(self):
response = self._aux(False)
self.assertEqual(response.status_int, 401)
ところが、ステータスは想定された 401(forbidden)ではなく 200(success)となり、このテストは通過しませんでした。
ここでの問題はユニット テストが
app.yaml まで遡っておらず、ユニット テストのスタブではなく、サービスを呼び出すアプリケーションの場合のように、完全なルーティングとログイン制限がテストの対象になっていないことでした。つまり、対象となっていたのはログイン制限が課されていない
thecron.py の二次的なルーティングだったのです。そのため、
app.yaml のルーティングのほか、MIME タイプやログインなどのその他諸々は、ユニット テストのスタブではなくサービスを呼び出すテストだけの対象となっていました。
そこで、管理者の権限をチェックするデコレータ
needs_admin を
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
さらに、これを
CronHandler.get でデコレートしました。
class CronHandler(webapp2.RequestHandler):
@needs_admin
def get(self): # etc, as before
これで cron サービスのスタブを呼び出すローカルのユニット テストは機能しますが、実際の App Engine Cron Service の機能を使ったエンドツーエンドのテストはステータス 401で失敗となります。なぜでしょうか。
簡単に言うと、App Engine Cron Service は指定された URL にアクセスするどのユーザーもログインさせません。そのため、ハンドラでは
users.get_current_user() から
None が返されます。
これを解消するのは、特別なリクエスト ヘッダ
X-AppEngine-Cron: true の設定です。App Engine では外部リクエストで設定されたヘッダが除外されるため、このヘッダはアプリケーション コードが完全に信頼できるヘッダとなります。
つまり、ユニット テスト、エンドツーエンド テスト、ローカル開発のアプリケーション、デプロイされたアプリケーションを機能させるために必要だったのは、デコレータ
needs_admin のわずかな修正だけだったのです。
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
新しい
if の記述はモック化されているかどうかにかかわらず、cron ジョブを処理します。2 つめの
if の記述は、これまで通り管理者以外の(またはログインしていない)ユーザーを確実にブロックします。
コードの品質はユニット テストに時間と注意を注ぐことで継続的に維持できます。こうしたテストの作成と実行は、App Engine の
Local Unit Testing ツール のほか、NoseGAE といったオープンソースのアドオンを活用することによって単純化することが可能です。
- Posted By Alex Martelli, Cloud Technical Support
* この投稿は、米国時間 7 月 6 日、Cloud Technical Support の Alex Martelli によって投稿されたもの の抄訳です。
ソフトウェアの開発ではコードのユニット テストが極めて有効です。モジュール、クラス、機能など、それぞれのコード ユニットを個々に検証する簡単な自動テストを実行することで、開発プロセスの早い段階でエラーの特定とデバッグが可能になります。
Google App Engine は、現在 Python 、Java 、Go に対応している Local Unit Testing ツールによって、ユニット テストを強力にサポートしています。
ローカルでのユニット テストではリモート コンポーネントを呼び出すことなく、開発環境でユニット テストを実行できます。また Local Unit Testing ツールでは、多数の App Engine のサービスをシミュレーションするためのサービス スタブを使用できます。
ローカルのユニット テストでは必要に応じてこのスタブを使用し、アプリケーションのコードを実行することが可能です。また、NoseGAE などのオープンソース パッケージを使用すれば、App Engine でのローカルのユニット テストの作成プロセスをさらに簡略化することができます。
ただし、サービス スタブを使用するローカルのユニット テストの場合、ルーティングとログイン制限はサービスを直接呼び出すコードとは異なって扱われます。テストを作成する際はこの点に留意しなければなりません。
App Engine の cron ハンドラのユニット テストで、あるお客様に問題が発生しました。私はその解決にあたりましたが、ここでは app.yaml に下記を入力した上で cron のハンドラが定義されていました。
- url: /crontask.*
script: thecron.app
login: admin_only
App Engine Cron Service では、スケジュールされているタスクが使用する URL へのアクセスを管理者アカウントのみに制限することが推奨 されています。これは login: admin_only の設定で実行されます。
お客様は管理者以外のユーザーによる、これらの URL へのアクセスがブロックされるかどうか確認しようとしました。
そこでまず、cron ハンドラ 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)
次にユニット テスト ut.py が書き込まれました。
(注:下記のテスト コードでは mock モジュールが使用されています。ローカルの Python 2.7 のインストールで mock モジュールを使用できるようにするには、 pip install mock を実行します。)
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()
最初のテスト testAdminWorks は問題なく通過したため、2 つめのテストが追加されました。
def testNonAdminFails(self):
response = self._aux(False)
self.assertEqual(response.status_int, 401)
ところが、ステータスは想定された 401(forbidden)ではなく 200(success)となり、このテストは通過しませんでした。
ここでの問題はユニット テストが app.yaml まで遡っておらず、ユニット テストのスタブではなく、サービスを呼び出すアプリケーションの場合のように、完全なルーティングとログイン制限がテストの対象になっていないことでした。つまり、対象となっていたのはログイン制限が課されていない thecron.py の二次的なルーティングだったのです。そのため、 app.yaml のルーティングのほか、MIME タイプやログインなどのその他諸々は、ユニット テストのスタブではなくサービスを呼び出すテストだけの対象となっていました。
そこで、管理者の権限をチェックするデコレータ needs_admin を 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
さらに、これを CronHandler.get でデコレートしました。
class CronHandler(webapp2.RequestHandler):
@needs_admin
def get(self): # etc, as before
これで cron サービスのスタブを呼び出すローカルのユニット テストは機能しますが、実際の App Engine Cron Service の機能を使ったエンドツーエンドのテストはステータス 401で失敗となります。なぜでしょうか。
簡単に言うと、App Engine Cron Service は指定された URL にアクセスするどのユーザーもログインさせません。そのため、ハンドラでは users.get_current_user() から None が返されます。
これを解消するのは、特別なリクエスト ヘッダ X-AppEngine-Cron: true の設定です。App Engine では外部リクエストで設定されたヘッダが除外されるため、このヘッダはアプリケーション コードが完全に信頼できるヘッダとなります。
つまり、ユニット テスト、エンドツーエンド テスト、ローカル開発のアプリケーション、デプロイされたアプリケーションを機能させるために必要だったのは、デコレータ needs_admin のわずかな修正だけだったのです。
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
新しい if の記述はモック化されているかどうかにかかわらず、cron ジョブを処理します。2 つめの if の記述は、これまで通り管理者以外の(またはログインしていない)ユーザーを確実にブロックします。
コードの品質はユニット テストに時間と注意を注ぐことで継続的に維持できます。こうしたテストの作成と実行は、App Engine の Local Unit Testing ツール のほか、NoseGAE といったオープンソースのアドオンを活用することによって単純化することが可能です。
- Posted By Alex Martelli, Cloud Technical Support