Commit 67b1296f authored by Christophe de Vienne's avatar Christophe de Vienne
Browse files

Now use webtest based TC (taken from the rememberme cube)

parent f4c4cf9b528d
import urllib import urllib
import urlparse import urlparse
import mock import mock
import webtest
from logilab.common.decorators import monkeypatch
from cubicweb.devtools.httptest import CubicWebServerTC from cubicweb.devtools.httptest import CubicWebServerTC
from cubicweb.devtools.testlib import CubicWebTC from cubicweb.devtools.testlib import CubicWebTC
from cubicweb.predicates import yes from cubicweb.predicates import yes
from cubicweb.wsgi import handler
from cubes.oauth.entities import ServiceAdapter from cubes.oauth.entities import ServiceAdapter
from cubes.oauth import views as oauth_views from cubes.oauth import views as oauth_views
# XXX Taken from rememberme tests
@monkeypatch(handler.CubicWebWSGIApplication)
def __init__(self, config, repo=None):
if repo is None:
repo = config.repository()
self.appli = handler.CubicWebPublisher(repo, config)
self.config = config
self.base_url = self.config['base-url']
self.url_rewriter = self.appli.vreg['components'].select_or_none(
'urlrewriter')
# XXX make a patch
class CubicWebTestTC(CubicWebTC):
@classmethod
def init_config(cls, config):
super(CubicWebTestTC, cls).init_config(config)
config.global_set_option('base-url', 'http://localhost:80/')
def setUp(self):
super(CubicWebTestTC, self).setUp()
webapp = handler.CubicWebWSGIApplication(self.config, self.repo)
self.webapp = webtest.TestApp(webapp)
def tearDown(self):
del self.webapp
super(CubicWebTestTC, self).tearDown()
def login(self, user='admin', password=None, **args):
if password is None:
password = user
args.update({
'__login': user,
'__password': password})
return self.webapp.get('/login', args, headers={'test': 'test'})
def logout(self):
self.webapp.get('/logout')
class FakeRequestRession(object): class FakeRequestRession(object):
return_value = None return_value = None
...@@ -73,7 +117,7 @@ class FakeServiceAdapter(ServiceAdapter): ...@@ -73,7 +117,7 @@ class FakeServiceAdapter(ServiceAdapter):
) )
class OAuthTC(CubicWebServerTC): class OAuthTC(CubicWebTestTC):
anonymous_allowed = True anonymous_allowed = True
application_name = u'oauth_test' application_name = u'oauth_test'
...@@ -81,7 +125,8 @@ class OAuthTC(CubicWebServerTC): ...@@ -81,7 +125,8 @@ class OAuthTC(CubicWebServerTC):
application_secret = u'secret' application_secret = u'secret'
def setUp(self): def setUp(self):
CubicWebTC.setUp(self) super(OAuthTC, self).setUp()
#CubicWebTC.setUp(self)
self.vreg.register(FakeServiceAdapter) self.vreg.register(FakeServiceAdapter)
with self.admin_access.repo_cnx() as cnx: with self.admin_access.repo_cnx() as cnx:
service = cnx.create_entity( service = cnx.create_entity(
...@@ -96,32 +141,31 @@ class OAuthTC(CubicWebServerTC): ...@@ -96,32 +141,31 @@ class OAuthTC(CubicWebServerTC):
service.cw_adapt_to('externalauth.service').provider service.cw_adapt_to('externalauth.service').provider
cnx.commit() cnx.commit()
oauth_views.NEGOSTATE = {} oauth_views.NEGOSTATE = {}
self.start_server()
def tearDown(self): def tearDown(self):
self.vreg.unregister(FakeServiceAdapter) self.vreg.unregister(FakeServiceAdapter)
super(OAuthTC, self).tearDown() super(OAuthTC, self).tearDown()
def login(self): def login(self):
response = self.web_request('?__externalauthprovider=facebook') response = self.webapp.get('?__externalauthprovider=facebook')
location = response.getheader('location') location = response.headers['location']
redirect_uri = urlparse.parse_qs( redirect_uri = urlparse.parse_qs(
location.split('?')[1])['redirect_uri'][0] location.split('?')[1])['redirect_uri'][0]
assert redirect_uri.startswith(self.config['base-url']) assert redirect_uri.startswith(self.config['base-url'])
url = redirect_uri + '&code=goodcode' url = redirect_uri + '&code=goodcode'
url = url[len(self.config['base-url']):] url = url[len(self.config['base-url']) - 1:]
return self.web_request(url) return self.webapp.get(url)
def test_wrong_providername(self): def test_wrong_providername(self):
response = self.web_request('/?__externalauthprovider=wrong') response = self.webapp.get('/?__externalauthprovider=wrong')
self.assertEqual(303, response.status) self.assertEqual(303, response.status_code)
def test_1st_redirect(self): def test_1st_redirect(self):
response = self.web_request('/?__externalauthprovider=facebook') response = self.webapp.get('/?__externalauthprovider=facebook')
location = response.getheader('location') location = response.headers['location']
expected_location = self.provider.authorize_url + '?' \ expected_location = self.provider.authorize_url + '?' \
+ urllib.urlencode({ + urllib.urlencode({
'scope': self.provider.scope, 'scope': self.provider.scope,
...@@ -131,7 +175,7 @@ class OAuthTC(CubicWebServerTC): ...@@ -131,7 +175,7 @@ class OAuthTC(CubicWebServerTC):
+ oauth_views.NEGOSTATE.keys()[0]}) + oauth_views.NEGOSTATE.keys()[0]})
self.assertEqual( self.assertEqual(
expected_location, location) expected_location, location)
self.assertEqual(303, response.status) self.assertEqual(303, response.status_code)
def test_creation(self): def test_creation(self):
with self.admin_access.repo_cnx() as cnx: with self.admin_access.repo_cnx() as cnx:
...@@ -139,9 +183,9 @@ class OAuthTC(CubicWebServerTC): ...@@ -139,9 +183,9 @@ class OAuthTC(CubicWebServerTC):
self.assertEqual(0, rset.rowcount) self.assertEqual(0, rset.rowcount)
response = self.login() response = self.login()
location = response.getheader('location') location = response.headers['location']
expected_location = self.config['base-url'] expected_location = self.config['base-url']
self.assertEqual(303, response.status) self.assertEqual(303, response.status_code)
self.assertEqual( self.assertEqual(
expected_location, location) expected_location, location)
...@@ -243,15 +287,12 @@ class OAuthTC(CubicWebServerTC): ...@@ -243,15 +287,12 @@ class OAuthTC(CubicWebServerTC):
cnx.commit() cnx.commit()
response = self.login() response = self.login()
location = response.getheader('location') location = response.headers['location']
expected_location = self.config['base-url'] expected_location = self.config['base-url']
self.assertEqual(303, response.status) self.assertEqual(303, response.status_code)
self.assertEqual(expected_location, location) self.assertEqual(expected_location, location)
cookie = dict( self.assertEqual('zeuser', self.webapp.cookies['__data_session'][:6])
token.strip().split('=') for token in
response.getheader('Set-Cookie').split(';'))
assert cookie['__data_session'].startswith("zeuser")
with self.admin_access.repo_cnx() as cnx: with self.admin_access.repo_cnx() as cnx:
rset = cnx.execute('Any X WHERE X login "zeuser"') rset = cnx.execute('Any X WHERE X login "zeuser"')
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment