Skip to content

Commit

Permalink
Add full end-to-end login test
Browse files Browse the repository at this point in the history
  • Loading branch information
leplatrem committed Nov 14, 2024
1 parent 4e9e470 commit 8b06fee
Show file tree
Hide file tree
Showing 2 changed files with 104 additions and 4 deletions.
6 changes: 2 additions & 4 deletions src/kinto_http/login.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@ def __init__(self, *args, set_jwt_token_callback=None, **kwargs):

def do_GET(self):
# Ignore non-auth requests (eg. favicon.ico).
if "/auth" not in self.path:
if "/auth" not in self.path: # pragma: no cover
self.send_response(404)
self.end_headers()
return
Expand Down Expand Up @@ -63,9 +63,7 @@ def __call__(self, r):
if self.provider is None:
provider_info = openid_info["providers"][0]
else:
provider_info = [p for p in provider_info["providers"] if p["name"] == self.provider][
0
]
provider_info = [p for p in openid_info["providers"] if p["name"] == self.provider][0]

# Spawn a local server on a random port, in order to receive the OAuth dance
# redirection and JWT token content.
Expand Down
102 changes: 102 additions & 0 deletions tests/test_login.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,102 @@
import base64
import json
import threading
from http.server import BaseHTTPRequestHandler, HTTPServer
from unittest import mock
from urllib.parse import parse_qs, quote, urlparse

import pytest
import requests

from kinto_http.login import BrowserOAuth


class RequestHandler(BaseHTTPRequestHandler):
def __init__(self, body, *args, **kwargs):
self.body = body
super().__init__(*args, **kwargs)

def do_GET(self):
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.end_headers()
self.wfile.write(json.dumps(self.body).encode("utf-8"))


@pytest.fixture
def http_server():
rs_server = HTTPServer(
("", 0),
lambda *args, **kwargs: RequestHandler(
{
"capabilities": {
"openid": {
"providers": [
{
"name": "other",
"auth_path": "/openid/ldap/login",
},
{
"name": "ldap",
"auth_path": "/openid/ldap/login",
},
]
}
}
},
*args,
**kwargs,
),
)
rs_server.port = rs_server.server_address[1]
threading.Thread(target=rs_server.serve_forever).start()

yield rs_server

rs_server.shutdown()


@pytest.fixture
def mock_oauth_dance():
def simulate_navigate(url):
"""
Behave as the user going through the OAuth dance in the browser.
"""
parsed = urlparse(url)
qs = parse_qs(parsed.query)
callback_url = qs["callback"][0]

token = {
"token_type": "Bearer",
"access_token": "fake-token",
}
json_token = json.dumps(token).encode("utf-8")
json_base64 = base64.urlsafe_b64encode(json_token)
encoded_token = quote(json_base64)
# This will open the local server started in `login.py`.
threading.Thread(target=lambda: requests.get(callback_url + encoded_token)).start()

with mock.patch("kinto_http.login.webbrowser") as mocked:
mocked.open.side_effect = simulate_navigate
yield


def test_uses_first_openid_provider(mock_oauth_dance, http_server):
auth = BrowserOAuth()
auth.server_url = f"http://localhost:{http_server.port}/v1"

req = requests.Request()
auth(req)
assert "Bearer fake-token" in req.headers["Authorization"]

# Can be called infinitely
auth(req)


def test_uses_specified_openid_provider(mock_oauth_dance, http_server):
auth = BrowserOAuth(provider="ldap")
auth.server_url = f"http://localhost:{http_server.port}/v1"

req = requests.Request()
auth(req)
assert "Bearer fake-token" in req.headers["Authorization"]

0 comments on commit 8b06fee

Please sign in to comment.