Created
June 10, 2013 15:11
-
-
Save mmulich/5749527 to your computer and use it in GitHub Desktop.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
# ... | |
class CheckTests(unittest.TestCase): | |
# These tests assume a remote (or local) service is communicating with | |
# this service to verify the token it was given is valid. | |
def setUp(self): | |
self.config = testing.setUp() | |
from sqlalchemy import create_engine | |
engine = create_engine('sqlite://') | |
from .models import Base | |
DBSession.configure(bind=engine) | |
Base.metadata.create_all(engine) | |
# Initialize the routes required by the view to generate | |
# followup urls. | |
from . import register_api | |
register_api(self.config) | |
# The token store needs access to our sql-url at runtime. | |
self._tmp_db_file = tempfile.mkstemp('test.db')[1] | |
sql_connect_str = 'sqlite:///{}'.format(self._tmp_db_file) | |
self.config.registry.settings['sqlalchemy.url'] = sql_connect_str | |
def tearDown(self): | |
os.remove(self._tmp_db_file) | |
DBSession.remove() | |
testing.tearDown() | |
def test_request(self): | |
# Check that a remote service (the indented usage) can make | |
# a request to this service and retrieve the correct user id. | |
request = testing.DummyRequest() | |
request.server_name = 'localhost' | |
remote_domain = 'example.com' | |
request.remote_addr = socket.gethostbyname(remote_domain) | |
token = str(uuid.uuid4()) | |
request.params = request.POST = request.GET = {'token': token} | |
from .views import get_token_store | |
token_store = get_token_store() | |
# FIXME The key/value store is nice, but we have a complex value that | |
# requires additional store and retrieval processing. It'd be | |
# nice to abstract this a bit more. | |
# The complex part is the arrangement of the value. Now this | |
# isn't brain surgery, but it is something that takes the | |
# developer's focus for a moment, enough that they might need | |
# to look up the ordering of the value. | |
user_id = '1234-5678-90' | |
value = "{}%{}".format(user_id, remote_domain) | |
token_store.store(token, value) | |
from .views import check | |
resp = check(request) | |
data = json.loads(resp.body) | |
self.assertEqual(data['id'], user_id) | |
url = request.route_url('get-user', user_id=user_id) | |
self.assertEqual(data['url'], url) | |
def test_request_fails_on_domain_mismatch(self): | |
self.fail() | |
def test_fails_on_expired(self): | |
self.fail() | |
# ... |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
# ... | |
@view_config(route_name='server-check', request_method=['GET', 'POST'], | |
renderer='json') | |
def check(request): | |
"""Check the token given to the external service by the user is | |
a valid token.""" | |
# Pull the token out of the request. | |
token = request.params['token'] | |
remote_domain = socket.gethostbyaddr(request.remote_addr)[0] | |
store = get_token_store() | |
try: | |
# FYI expiration of the token/key is checked on retrieval. | |
value = store.retrieve(token) | |
user_id, domain = value.split('%') | |
except: | |
raise httpexceptions.HTTPInternalServerError("Problem connecting to " | |
"the token storage.") | |
try: | |
# Check the token was given is valid for the external service domain. | |
assert remote_domain == domain | |
except AssertionError: | |
raise httpexceptions.HTTPBadRequest("Invalid Token") | |
return {'id': user_id, | |
'url': request.route_url('get-user', user_id=user_id), | |
} | |
# ... |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment