Skip to content

Instantly share code, notes, and snippets.

@mmulich
Created June 10, 2013 15:11
Show Gist options
  • Save mmulich/5749527 to your computer and use it in GitHub Desktop.
Save mmulich/5749527 to your computer and use it in GitHub Desktop.
# ...
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()
# ...
# ...
@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