wip: sso flow test (error), existing
This commit is contained in:
parent
888e398496
commit
e16cc575aa
|
|
@ -1,37 +0,0 @@
|
||||||
import uuid
|
|
||||||
from unittest.mock import patch
|
|
||||||
|
|
||||||
from django.test import TestCase
|
|
||||||
from django.urls import reverse
|
|
||||||
|
|
||||||
from vbv_lernwelt.core.models import User
|
|
||||||
|
|
||||||
|
|
||||||
class TestSSO(TestCase):
|
|
||||||
def test_walking_skeleton(self):
|
|
||||||
self.assertTrue(True)
|
|
||||||
|
|
||||||
@patch("vbv_lernwelt.sso.views.oauth")
|
|
||||||
@patch("vbv_lernwelt.sso.views.decode_jwt")
|
|
||||||
def test_authorize_redirects_on_success(self, mock_decode_jwt, mock_oauth_service):
|
|
||||||
# GIVEN
|
|
||||||
email = "bobby@drop.table"
|
|
||||||
|
|
||||||
mock_oauth_service.authorize_access_token.return_value = {
|
|
||||||
"id_token": "test_token"
|
|
||||||
}
|
|
||||||
|
|
||||||
mock_decode_jwt.return_value = {
|
|
||||||
"emails": [email],
|
|
||||||
"oid": uuid.uuid4(),
|
|
||||||
"given_name": "Bobby",
|
|
||||||
"family_name": "Drop-Table",
|
|
||||||
}
|
|
||||||
|
|
||||||
# WHEN
|
|
||||||
response = self.client.get(reverse("sso:authorize"))
|
|
||||||
|
|
||||||
# THEN
|
|
||||||
self.assertTrue(User.objects.filter(email=email).exists())
|
|
||||||
self.assertEqual(response.status_code, 302)
|
|
||||||
self.assertEqual(response.url, "/")
|
|
||||||
|
|
@ -0,0 +1,53 @@
|
||||||
|
import uuid
|
||||||
|
from unittest.mock import patch, Mock
|
||||||
|
|
||||||
|
from authlib.integrations.base_client import OAuthError
|
||||||
|
from django.conf import settings
|
||||||
|
from django.test import TestCase
|
||||||
|
from django.urls import reverse
|
||||||
|
|
||||||
|
from vbv_lernwelt.core.models import User
|
||||||
|
|
||||||
|
|
||||||
|
def decoded_token(email, oid=None, given_name="Bobby", family_name="Table"):
|
||||||
|
return {
|
||||||
|
"emails": [email],
|
||||||
|
"oid": oid or uuid.uuid4(),
|
||||||
|
"given_name": given_name,
|
||||||
|
"family_name": family_name,
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
class TestSSOFlow(TestCase):
|
||||||
|
@patch("vbv_lernwelt.sso.views.oauth")
|
||||||
|
@patch("vbv_lernwelt.sso.views.decode_jwt")
|
||||||
|
def test_authorize_redirects_on_success(self, mock_decode_jwt, _):
|
||||||
|
# GIVEN
|
||||||
|
email = "bobby@drop.table"
|
||||||
|
mock_decode_jwt.return_value = decoded_token(email)
|
||||||
|
|
||||||
|
# WHEN
|
||||||
|
response = self.client.get(reverse("sso:authorize"))
|
||||||
|
|
||||||
|
# THEN
|
||||||
|
self.assertTrue(User.objects.filter(email=email).exists())
|
||||||
|
self.assertEqual(response.status_code, 302)
|
||||||
|
self.assertEqual(response.url, "/") # noqa
|
||||||
|
|
||||||
|
@patch("vbv_lernwelt.sso.views.oauth")
|
||||||
|
def test_authorize_on_tampered_token(self, mock_oauth_service):
|
||||||
|
# GIVEN
|
||||||
|
client_name = settings.OAUTH["client_name"]
|
||||||
|
client_mock = Mock()
|
||||||
|
client_mock.authorize_access_token.side_effect = OAuthError()
|
||||||
|
setattr(mock_oauth_service, client_name, client_mock)
|
||||||
|
|
||||||
|
# WHEN
|
||||||
|
response = self.client.get(reverse("sso:authorize"))
|
||||||
|
|
||||||
|
# THEN
|
||||||
|
# sanity check that the mock was called (-> setup is correct)
|
||||||
|
self.assertEqual(client_mock.authorize_access_token.call_count, 1)
|
||||||
|
|
||||||
|
self.assertEqual(response.status_code, 302)
|
||||||
|
self.assertEqual(response.url, "/login-error?state=someerror") # noqa
|
||||||
|
|
@ -22,10 +22,6 @@ def login(request):
|
||||||
|
|
||||||
|
|
||||||
def authorize(request):
|
def authorize(request):
|
||||||
print("authorize")
|
|
||||||
print(oauth)
|
|
||||||
print(decode_jwt)
|
|
||||||
|
|
||||||
try:
|
try:
|
||||||
logger.debug(request, label="sso")
|
logger.debug(request, label="sso")
|
||||||
token = getattr(oauth, settings.OAUTH["client_name"]).authorize_access_token(
|
token = getattr(oauth, settings.OAUTH["client_name"]).authorize_access_token(
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue