import unittest
try:
import mock
except ImportError:
from unittest import mock
import json
import urllib.parse
import datetime
from requests import Session, Response
from app import create_app
headers = {
'X-CLIENT-ID': 'test_client_id',
'X-CLIENT-SECRET': 'test_client_secret',
'X-REDIRECT-URI': 'https://127.0.0.123:7654/callback',
}
class MockRequest:
def __init__(self):
self.headers = []
self.body = ''
class MockResponse:
def __init__(self, status_code, json_data=None):
self.request = MockRequest()
self.ok = status_code == 200
self.status_code = status_code
self.json = lambda: json_data
self.headers = []
self.text = json.dumps(json_data)
def mock_request(self, method, url, **kwargs):
if method == 'POST' and url == 'https://127.0.0.123:4567/token':
return MockResponse(200, {'access_token': '2YotnFZFEjr1zCsicMWpAA',
'token_type': 'Bearer',
'expires_in': 3600,
'refresh_token': 'tGzv3JOkF0XG5Qx2TlKWIA'})
if method == 'GET' and url == 'https://127.0.0.123:4567/userinfo':
if kwargs['headers']['Authorization'] != 'Bearer 2YotnFZFEjr1zCsicMWpAA':
raise Exception()
return MockResponse(200, {'nickname': 'testuser'})
print(repr(method), repr(url), repr(kwargs))
raise Exception()
@mock.patch.object(Session, 'request', new=mock_request)
class TestCases(unittest.TestCase):
def setUp(self):
config = {
'TESTING': True,
'DEBUG': True,
'SECRET_KEY': 'DEBUGKEY',
'OAUTH2_AUTH_URL': 'https://127.0.0.123:4567/authorize',
'OAUTH2_TOKEN_URL': 'https://127.0.0.123:4567/token',
'OAUTH2_USERINFO_URL': 'https://127.0.0.123:4567/userinfo',
}
self.app = create_app(config)
self.client = self.app.test_client()
self.client.__enter__()
# Just do some request so that we can use url_for
self.client.get(path='/')
def tearDown(self):
self.client.__exit__(None, None, None)
def test_status(self):
r = self.client.get(path='/status', headers=headers)
self.assertEqual(r.status_code, 200)
self.assertIn('test_client_id', r.data.decode())
self.assertNotIn('test_client_secret', r.data.decode())
def test_auth_no_session(self):
r = self.client.get(path='/auth', headers=headers)
self.assertEqual(r.status_code, 401)
def test_login(self):
r = self.client.get(path='/login', query_string={'url': 'https://127.0.0.123:7654/app'}, headers=headers, follow_redirects=False)
self.assertEqual(r.status_code, 302)
url = urllib.parse.urlparse(r.location)
qs = urllib.parse.parse_qs(url.query)
self.assertEqual(url.scheme, 'https')
self.assertEqual(url.netloc, '127.0.0.123:4567')
self.assertEqual(url.path, '/authorize')
self.assertEqual(qs['response_type'], ['code'])
self.assertEqual(qs['client_id'], ['test_client_id'])
self.assertEqual(qs['redirect_uri'], ['https://127.0.0.123:7654/callback'])
self.assertGreater(len(qs['state'][0]), 8)
with self.client.session_transaction() as session:
self.assertEqual(session['state'], qs['state'][0])
self.assertEqual(session['url'], 'https://127.0.0.123:7654/app')
def test_callback(self):
code = 'testcode'
state = 'teststate'
with self.client.session_transaction() as session:
session['state'] = state
session['url'] = 'https://127.0.0.123:7654/app'
r = self.client.get(path='/callback', headers=headers, query_string={'code': code, 'state': state}, follow_redirects=False)
self.assertEqual(r.status_code, 302)
self.assertEqual(r.location, 'https://127.0.0.123:7654/app')
with self.client.session_transaction() as session:
self.assertGreaterEqual(session['timestamp'], (datetime.datetime.now() - datetime.timedelta(seconds=60)).timestamp())
self.assertEqual(session['client_id'], 'test_client_id')
self.assertEqual(session['user_nickname'], 'testuser')
self.assertNotIn('state', session)
self.assertNotIn('url', session)
def test_callback_no_session(self):
code = 'testcode'
state = 'teststate'
r = self.client.get(path='/callback', headers=headers, query_string={'code': code, 'state': state}, follow_redirects=False)
self.assertEqual(r.status_code, 302)
url = urllib.parse.urlparse(r.location)
self.assertEqual(url.path, '/cookiecheck')
with self.client.session_transaction() as session:
self.assertEqual(session['cookies_enabled'], True)
def test_cookiecheck(self):
with self.client.session_transaction() as session:
session['cookies_enabled'] = True
r = self.client.get(path='/cookiecheck', headers=headers, follow_redirects=False)
self.assertEqual(r.status_code, 302)
url = urllib.parse.urlparse(r.location)
self.assertEqual(url.path, '/login')
def test_cookiecheck_no_session(self):
r = self.client.get(path='/cookiecheck', headers=headers, follow_redirects=False)
self.assertEqual(r.status_code, 400)
with self.client.session_transaction() as session:
self.assertEqual(session['cookies_enabled'], True)
def test_auth_session(self):
with self.client.session_transaction() as session:
session['timestamp'] = datetime.datetime.now().timestamp()
session['client_id'] = 'test_client_id'
session['user_nickname'] = 'testuser'
r = self.client.get(path='/auth', headers=headers)
self.assertEqual(r.status_code, 200)
self.assertEqual(r.headers['OAUTH-USER-NICKNAME'], 'testuser')
def test_auth_session_timeout(self):
with self.client.session_transaction() as session:
session['timestamp'] = (datetime.datetime.now() - datetime.timedelta(days=3)).timestamp()
session['client_id'] = 'test_client_id'
session['user_nickname'] = 'testuser'
r = self.client.get(path='/auth', headers=headers)
self.assertEqual(r.status_code, 401)
def test_auth_session_wrong_client(self):
with self.client.session_transaction() as session:
session['timestamp'] = (datetime.datetime.now() - datetime.timedelta(days=3)).timestamp()
session['client_id'] = 'other_client_id'
session['user_nickname'] = 'testuser'
r = self.client.get(path='/auth', headers=headers)
self.assertEqual(r.status_code, 401)
def test_logout(self):
with self.client.session_transaction() as session:
session['timestamp'] = datetime.datetime.now().timestamp()
session['client_id'] = 'test_client_id'
session['user_nickname'] = 'testuser'
r = self.client.get(path='/logout', headers=headers)
self.assertEqual(r.status_code, 200)
with self.client.session_transaction() as session:
self.assertEqual(list(session.keys()), [])
def test_logout_no_session(self):
r = self.client.get(path='/logout', headers=headers)
self.assertEqual(r.status_code, 200)
with self.client.session_transaction() as session:
self.assertEqual(list(session.keys()), [])
def test_logout_redirect(self):
with self.client.session_transaction() as session:
session['timestamp'] = datetime.datetime.now().timestamp()
session['client_id'] = 'test_client_id'
session['user_nickname'] = 'testuser'
r = self.client.get(path='/logout', headers=headers, query_string={'redirect_url': 'https://127.0.0.123:7654/app/logout'})
self.assertEqual(r.status_code, 302)
self.assertEqual(r.location, 'https://127.0.0.123:7654/app/logout')
with self.client.session_transaction() as session:
self.assertEqual(list(session.keys()), [])