mediagoblin/tests/test_http_callback.py

   1 # GNU MediaGoblin -- federated, autonomous media hosting
   2 # Copyright (C) 2011, 2012 MediaGoblin contributors.  See AUTHORS.
   3 #
   4 # This program is free software: you can redistribute it and/or modify
   5 # it under the terms of the GNU Affero General Public License as published by
   6 # the Free Software Foundation, either version 3 of the License, or
   7 # (at your option) any later version.
   8 #
   9 # This program is distributed in the hope that it will be useful,
  10 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  11 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  12 # GNU Affero General Public License for more details.
  13 #
  14 # You should have received a copy of the GNU Affero General Public License
  15 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
  16
  17 import json
  18
  19 from urlparse import urlparse, parse_qs
  20
  21 from mediagoblin import mg_globals
  22 from mediagoblin.tools import processing
  23 from mediagoblin.tests.tools import fixture_add_user
  24 from mediagoblin.tests.test_submission import GOOD_PNG
  25 from mediagoblin.tests import test_oauth as oauth
  26
  27
  28 class TestHTTPCallback(object):
  29     def _setup(self, test_app):
  30         self.db = mg_globals.database
  31
  32         self.user_password = u'secret'
  33         self.user = fixture_add_user(u'call_back', self.user_password)
  34
  35         self.login(test_app)
  36
  37     def login(self, testapp):
  38         testapp.post('/auth/login/', {
  39             'username': self.user.username,
  40             'password': self.user_password})
  41
  42     def get_access_token(self, testapp, client_id, client_secret, code):
  43         response = testapp.get('/oauth/access_token', {
  44                 'code': code,
  45                 'client_id': client_id,
  46                 'client_secret': client_secret})
  47
  48         response_data = json.loads(response.body)
  49
  50         return response_data['access_token']
  51
  52     def test_callback(self, test_app):
  53         ''' Test processing HTTP callback '''
  54         self._setup(test_app)
  55
  56         self.oauth = oauth.TestOAuth()
  57         self.oauth._setup(test_app)
  58
  59         redirect, client_id = self.oauth.test_4_authorize_confidential_client(
  60             test_app)
  61
  62         code = parse_qs(urlparse(redirect.location).query)['code'][0]
  63
  64         client = self.db.OAuthClient.query.filter(
  65                 self.db.OAuthClient.identifier == unicode(client_id)).first()
  66
  67         client_secret = client.secret
  68
  69         access_token = self.get_access_token(test_app, client_id, client_secret, code)
  70
  71         callback_url = 'https://foo.example?secrettestmediagoblinparam'
  72
  73         res = test_app.post('/api/submit?client_id={0}&access_token={1}\
  74 &client_secret={2}'.format(
  75                     client_id,
  76                     access_token,
  77                     client_secret), {
  78             'title': 'Test',
  79             'callback_url': callback_url},
  80             upload_files=[('file', GOOD_PNG)])
  81
  82         assert processing.TESTS_CALLBACKS[callback_url]['state'] == u'processed'