| # Copyright 2014 the Melange authors. |
| # |
| # 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. |
| |
| """Unit tests for shipment tracking related views.""" |
| |
| import mock |
| |
| from google.appengine.ext import ndb |
| |
| from seeder import user as user_seeder |
| |
| from summerofcode.models import shipment_tracking as shipment_tracking_model |
| from summerofcode.views import shipment_tracking as shipment_tracking_view |
| |
| from tests import profile_utils |
| from tests import test_utils |
| |
| |
| def _getCreateShipmentTrackingUrl(program): |
| """Returns the shipment info create url for the specified program.""" |
| return '/gsoc/admin/shipment_info/create/%s' % program.key().name() |
| |
| |
| def _getEditShipmentTrackingUrl(program, shipment_info): |
| """Returns the shipment info edit url for the specified program and shipment. |
| """ |
| return '/gsoc/admin/shipment_info/edit/%s/%s' % ( |
| program.key().name(), shipment_info.key.id()) |
| |
| |
| def _getShipmentInfoListUrl(program): |
| """Returns the shipment info list url for the specified program.""" |
| return '/gsoc/admin/shipment_tracking/records/%s' % program.key().name() |
| |
| |
| def _getShipmentSyncUrl(program): |
| """Returns the shipment sync url for the specified program.""" |
| return '/gsoc/admin/shipment_tracking/sync/%s' % program.key().name() |
| |
| |
| def _getShipmentTrackingUrl(program, shipment_info): |
| """Returns the shipment tracking url for the specified program and shipment. |
| """ |
| return '/gsoc/admin/shipment_tracking/sync/%s/%s' % ( |
| program.key().name(), shipment_info.key.id()) |
| |
| ACCESS_TOKEN = 'secret_access_token' |
| APP_ID = 'numeric_app_id' |
| CLIENT_ID = 'public_client_id' |
| |
| |
| class CreateShipmentInfoTest(test_utils.GSoCDjangoTestCase): |
| """Unit test for the CreateShipmentInfo class.""" |
| |
| def setUp(self): |
| self.init() |
| self.site.google_app_id = APP_ID |
| self.site.put() |
| user = user_seeder.seedUser(host_for=[self.program]) |
| profile_utils.loginNDB(user) |
| |
| def testCreateShipment(self): |
| url = _getCreateShipmentTrackingUrl(self.program) |
| |
| # get redirected to the auth flow |
| response = self.get(url) |
| self.assertResponseRedirect(response) |
| self.assertIn("drive.file", response["Location"]) |
| self.assertIn("access_type=offline", response["Location"]) |
| |
| # get the form |
| profile_utils.oauthAuthorizeNDB( |
| access_token=ACCESS_TOKEN, client_id=CLIENT_ID) |
| response = self.get(url) |
| self.assertResponseOkAndGSoCTemplatesUsed(response) |
| self.assertEqual(ACCESS_TOKEN, response.context['access_token']) |
| self.assertEqual(APP_ID, response.context['app_id']) |
| |
| # fill out the form |
| postdata = { |
| 'name': 'super cool shipment', |
| 'spreadsheet_id': 'supersecretid', |
| 'usa_gid': 0, |
| 'intl_gid': 17, |
| } |
| response = self.post(url, postdata=postdata) |
| shipment_info = shipment_tracking_model.ShipmentInfo.query().get() |
| edit_url = _getEditShipmentTrackingUrl(self.program, shipment_info) |
| self.assertResponseRedirect(response, edit_url) |
| self.assertPropertiesEqual(postdata, shipment_info) |
| |
| def testEditShipment(self): |
| shipment_info = shipment_tracking_model.ShipmentInfo( |
| name='shipment', spreadsheet_id='someid', |
| parent=ndb.Key.from_old_key(self.program.key()), |
| intl_gid=0, usa_gid=17) |
| shipment_info.put() |
| |
| url = _getEditShipmentTrackingUrl(self.program, shipment_info) |
| |
| # fill out the form |
| postdata = { |
| 'name': 'super cool shipment', |
| 'spreadsheet_id': 'supersecretid', |
| 'usa_gid': 0, |
| 'intl_gid': 17, |
| } |
| response = self.post(url, postdata=postdata) |
| shipment_info = shipment_tracking_model.ShipmentInfo.query().get() |
| self.assertEqual(1, shipment_tracking_model.ShipmentInfo.query().count(10)) |
| self.assertResponseRedirect(response, url) |
| self.assertPropertiesEqual(postdata, shipment_info) |
| |
| |
| class ShipmentInfoListTest(test_utils.GSoCDjangoTestCase): |
| """Unit test for the ShipmentInfoListPag class.""" |
| |
| def setUp(self): |
| self.init() |
| user = user_seeder.seedUser(host_for=[self.program]) |
| profile_utils.loginNDB(user) |
| |
| shipment_info = shipment_tracking_model.ShipmentInfo( |
| name='shipment', spreadsheet_id='someid', |
| parent=ndb.Key.from_old_key(self.program.key()), |
| intl_gid=0, usa_gid=17) |
| shipment_info.put() |
| |
| shipment_info = shipment_tracking_model.ShipmentInfo( |
| name='other shipment', spreadsheet_id='other_id', |
| parent=ndb.Key.from_old_key(self.program.key()), |
| intl_gid=0, usa_gid=17) |
| shipment_info.put() |
| |
| def testCreateShipment(self): |
| url = _getShipmentInfoListUrl(self.program) |
| |
| response = self.get(url) |
| self.assertGSoCTemplatesUsed(response) |
| |
| idx = 0 |
| list_data = self.getListData(url, idx) |
| self.assertEqual(len(list_data), 2) |
| |
| |
| class ShipmentSyncTest(test_utils.GSoCDjangoTestCase): |
| """Unit test for the SyncData class.""" |
| |
| def setUp(self): |
| self.init() |
| user = user_seeder.seedUser(host_for=[self.program]) |
| profile_utils.loginNDB(user) |
| |
| shipment_info = shipment_tracking_model.ShipmentInfo( |
| name='shipment', spreadsheet_id='someid', |
| parent=ndb.Key.from_old_key(self.program.key()), |
| intl_gid=0, usa_gid=17) |
| shipment_info.put() |
| |
| shipment_info = shipment_tracking_model.ShipmentInfo( |
| name='other shipment', spreadsheet_id='other_id', |
| parent=ndb.Key.from_old_key(self.program.key()), |
| intl_gid=0, usa_gid=17) |
| shipment_info.put() |
| |
| def testSync(self): |
| url = _getShipmentSyncUrl(self.program) |
| |
| response = self.get(url) |
| self.assertResponseOkAndGSoCTemplatesUsed(response) |
| actual = response.context['shipment_infos'] |
| self.assertEqual(2, len(actual)) |
| |
| |
| def successfulFetchSheet(unused_drive_service, unused_url): |
| return "sheet contents", None |
| |
| |
| def failingFetchSheet(unused_drive_service, unused_url): |
| return None, "big bad error" |
| |
| |
| class ShipmentTrackingTest(test_utils.GSoCDjangoTestCase): |
| """Unit test for SyncTracking class.""" |
| |
| def setUp(self): |
| self.init() |
| |
| user = user_seeder.seedUser(host_for=[self.program]) |
| profile_utils.loginNDB(user) |
| |
| self.shipment_info = shipment_tracking_model.ShipmentInfo( |
| name='shipment', spreadsheet_id='someid', |
| parent=ndb.Key.from_old_key(self.program.key()), |
| intl_gid=0, usa_gid=17) |
| self.shipment_info.put() |
| |
| def testSyncOauthFlow(self): |
| url = _getShipmentTrackingUrl(self.program, self.shipment_info) |
| |
| response = self.get(url) |
| self.assertResponseRedirect(response) |
| self.assertIn("drive.file", response["Location"]) |
| self.assertIn("access_type=offline", response["Location"]) |
| |
| @mock.patch.object(shipment_tracking_view, 'fetchSheet', successfulFetchSheet) |
| def testSyncAuthed(self): |
| url = _getShipmentTrackingUrl(self.program, self.shipment_info) |
| profile_utils.oauthAuthorizeNDB() |
| |
| response = self.get(url) |
| sync_url = _getShipmentSyncUrl(self.program) |
| self.assertResponseRedirect(response, sync_url) |
| |
| tasks = self.taskqueue_stub.get_filtered_tasks() |
| self.assertEqual(2, len(tasks)) |
| payload = tasks[0].payload |
| self.assertIn('shipment_info_id', payload) |
| |
| @mock.patch.object(shipment_tracking_view, 'fetchSheet', failingFetchSheet) |
| def testSyncAuthedWithFailure(self): |
| url = _getShipmentTrackingUrl(self.program, self.shipment_info) |
| profile_utils.oauthAuthorizeNDB() |
| |
| response = self.get(url) |
| sync_url = _getShipmentSyncUrl(self.program) |
| self.assertResponseRedirect(response, sync_url) |
| |
| tasks = self.taskqueue_stub.get_filtered_tasks() |
| self.assertEqual(0, len(tasks)) |