t_50a4f551: Fix auth.py bugs + add 27 tests
- Fix _get_anti_forgery_state: parse JSON response (not raw text) to get Token - Fix _do_login: pass __RequestVerificationToken header, add FormsAuthReturnUrl - Fix login(): pass anti_forgery token through to _do_login - Fix AuthSession: store password for refresh, handle NotVerifiedUserEmail - Add 27 unit tests with respx mocking (all pass) - Add respx+pytest to requirements.txt
This commit is contained in:
@@ -0,0 +1,358 @@
|
||||
"""Tests for auth module — unit tests with mocked httpx via respx."""
|
||||
import os
|
||||
from unittest.mock import patch
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
import respx
|
||||
|
||||
from src.auth import (
|
||||
ANTI_FORGERY_API,
|
||||
AUTH_COOKIE_NAMES,
|
||||
SIGNIN_API,
|
||||
AuthSession,
|
||||
AuthenticationError,
|
||||
_do_login,
|
||||
_get_anti_forgery_state,
|
||||
_get_credentials,
|
||||
get_session,
|
||||
login,
|
||||
)
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────────
|
||||
# _get_credentials
|
||||
# ─────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_get_credentials_from_env():
|
||||
with patch.dict(os.environ, {"CANTALOUPE_EMAIL": "u@test.com", "CANTALOUPE_PASSWORD": "pw"}):
|
||||
email, password = _get_credentials()
|
||||
assert email == "u@test.com"
|
||||
assert password == "pw"
|
||||
|
||||
|
||||
def test_get_credentials_missing_email():
|
||||
with patch.dict(os.environ, {"CANTALOUPE_PASSWORD": "pw"}, clear=True):
|
||||
with pytest.raises(AuthenticationError, match="Credentials not found"):
|
||||
_get_credentials()
|
||||
|
||||
|
||||
def test_get_credentials_missing_password():
|
||||
with patch.dict(os.environ, {"CANTALOUPE_EMAIL": "u@test.com"}, clear=True):
|
||||
with pytest.raises(AuthenticationError, match="Credentials not found"):
|
||||
_get_credentials()
|
||||
|
||||
|
||||
def test_get_credentials_none_set():
|
||||
with patch.dict(os.environ, {}, clear=True):
|
||||
with pytest.raises(AuthenticationError, match="Credentials not found"):
|
||||
_get_credentials()
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────────
|
||||
# _get_anti_forgery_state
|
||||
# ─────────────────────────────────────────────────
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_anti_forgery_success():
|
||||
token = "CfDJ8ExampleToken12345"
|
||||
respx.get(ANTI_FORGERY_API).respond(
|
||||
json={"HeaderName": "__RequestVerificationToken", "Token": token}
|
||||
)
|
||||
client = httpx.Client(base_url="https://mycantaloupe.com")
|
||||
result = _get_anti_forgery_state(client)
|
||||
assert result == token
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_anti_forgery_empty_token():
|
||||
respx.get(ANTI_FORGERY_API).respond(
|
||||
json={"HeaderName": "__RequestVerificationToken", "Token": ""}
|
||||
)
|
||||
client = httpx.Client(base_url="https://mycantaloupe.com")
|
||||
with pytest.raises(AuthenticationError, match="Failed to obtain anti-forgery"):
|
||||
_get_anti_forgery_state(client)
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_anti_forgery_missing_token_key():
|
||||
respx.get(ANTI_FORGERY_API).respond(json={"other": "data"})
|
||||
client = httpx.Client(base_url="https://mycantaloupe.com")
|
||||
with pytest.raises(AuthenticationError, match="Failed to obtain anti-forgery"):
|
||||
_get_anti_forgery_state(client)
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_anti_forgery_non_200():
|
||||
respx.get(ANTI_FORGERY_API).respond(status_code=500)
|
||||
client = httpx.Client(base_url="https://mycantaloupe.com")
|
||||
with pytest.raises(httpx.HTTPStatusError):
|
||||
_get_anti_forgery_state(client)
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────────
|
||||
# _do_login
|
||||
# ─────────────────────────────────────────────────
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_do_login_success_with_result():
|
||||
respx.post(SIGNIN_API).respond(json={"SuccessfulResult": {"UserId": 42}})
|
||||
client = httpx.Client(base_url="https://mycantaloupe.com")
|
||||
result = _do_login(client, "u@test.com", "pw", "csrf-token")
|
||||
assert result["SuccessfulResult"]["UserId"] == 42
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_do_login_success_with_redirect():
|
||||
respx.post(SIGNIN_API).respond(json={"RedirectUrl": "/cs4/VueMachineList"})
|
||||
client = httpx.Client(base_url="https://mycantaloupe.com")
|
||||
result = _do_login(client, "u@test.com", "pw", "csrf-token")
|
||||
assert result["RedirectUrl"] == "/cs4/VueMachineList"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_do_login_error_message():
|
||||
respx.post(SIGNIN_API).respond(
|
||||
json={"ErrorMessage": "Incorrect username or password"}
|
||||
)
|
||||
client = httpx.Client(base_url="https://mycantaloupe.com")
|
||||
with pytest.raises(AuthenticationError, match="Login rejected: Incorrect"):
|
||||
_do_login(client, "bad@test.com", "wrong", "csrf-token")
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_do_login_not_verified():
|
||||
respx.post(SIGNIN_API).respond(
|
||||
json={"NotVerifiedUserEmail": "u@test.com"}
|
||||
)
|
||||
client = httpx.Client(base_url="https://mycantaloupe.com")
|
||||
with pytest.raises(AuthenticationError, match="Email not verified"):
|
||||
_do_login(client, "u@test.com", "pw", "csrf-token")
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_do_login_non_200():
|
||||
respx.post(SIGNIN_API).respond(status_code=500, text="Internal Server Error")
|
||||
client = httpx.Client(base_url="https://mycantaloupe.com")
|
||||
with pytest.raises(AuthenticationError, match="failed with status 500"):
|
||||
_do_login(client, "u@test.com", "pw", "csrf-token")
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_do_login_sends_correct_headers():
|
||||
route = respx.post(SIGNIN_API).respond(json={"SuccessfulResult": {}})
|
||||
client = httpx.Client(base_url="https://mycantaloupe.com")
|
||||
_do_login(client, "u@test.com", "pw", "csrf-token-abc")
|
||||
req = route.calls.last.request
|
||||
assert req.headers["__RequestVerificationToken"] == "csrf-token-abc"
|
||||
assert req.headers["Content-Type"] == "application/json"
|
||||
body = req.content.decode()
|
||||
assert "u@test.com" in body
|
||||
assert '"pw"' in body
|
||||
assert "FormsAuthReturnUrl" in body
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_do_login_unclear_response_still_returns():
|
||||
respx.post(SIGNIN_API).respond(json={"SomeField": "value"})
|
||||
client = httpx.Client(base_url="https://mycantaloupe.com")
|
||||
result = _do_login(client, "u@test.com", "pw", "csrf-token")
|
||||
assert result == {"SomeField": "value"}
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────────
|
||||
# login() — full flow
|
||||
# ─────────────────────────────────────────────────
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_login_full_flow():
|
||||
csrf_token = "CsrfToken123"
|
||||
session_cookie = "abc123session"
|
||||
auth_cookie = "authcookie456"
|
||||
|
||||
respx.get(ANTI_FORGERY_API).respond(
|
||||
json={"HeaderName": "__RequestVerificationToken", "Token": csrf_token}
|
||||
)
|
||||
login_route = respx.post(SIGNIN_API).respond(
|
||||
json={"SuccessfulResult": {"UserId": 1}},
|
||||
headers=[
|
||||
("Set-Cookie", f"ASP.NET_SessionId={session_cookie}; Path=/; HttpOnly; SameSite=Lax"),
|
||||
("Set-Cookie", f"ApplicationGatewayAffinity={auth_cookie}; Path=/"),
|
||||
],
|
||||
)
|
||||
|
||||
with patch.dict(
|
||||
os.environ,
|
||||
{"CANTALOUPE_EMAIL": "u@test.com", "CANTALOUPE_PASSWORD": "pw"},
|
||||
):
|
||||
auth = login()
|
||||
|
||||
assert auth.email == "u@test.com"
|
||||
assert auth.password == "pw"
|
||||
assert "ASP.NET_SessionId" in auth.cookies
|
||||
assert "ApplicationGatewayAffinity" in auth.cookies
|
||||
assert auth.client.cookies["ApplicationGatewayAffinity"] == auth_cookie
|
||||
|
||||
csrf_call = respx.get(ANTI_FORGERY_API).calls.last
|
||||
assert csrf_call is not None
|
||||
login_call = login_route.calls.last
|
||||
assert login_call.request.headers["__RequestVerificationToken"] == csrf_token
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_login_with_explicit_credentials():
|
||||
csrf_token = "token"
|
||||
respx.get(ANTI_FORGERY_API).respond(
|
||||
json={"HeaderName": "__RequestVerificationToken", "Token": csrf_token}
|
||||
)
|
||||
respx.post(SIGNIN_API).respond(json={"SuccessfulResult": {}})
|
||||
|
||||
auth = login(email="explicit@test.com", password="explicitpw")
|
||||
assert auth.email == "explicit@test.com"
|
||||
assert auth.password == "explicitpw"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_login_with_custom_client():
|
||||
respx.get(ANTI_FORGERY_API).respond(
|
||||
json={"HeaderName": "__RequestVerificationToken", "Token": "t"}
|
||||
)
|
||||
respx.post(SIGNIN_API).respond(json={"SuccessfulResult": {}})
|
||||
|
||||
custom_client = httpx.Client(base_url="https://mycantaloupe.com", timeout=30.0)
|
||||
auth = login(email="u@test.com", password="pw", client=custom_client)
|
||||
assert auth.client is custom_client
|
||||
|
||||
|
||||
def test_login_missing_credentials():
|
||||
with patch.dict(os.environ, {}, clear=True):
|
||||
with pytest.raises(AuthenticationError, match="Credentials not found"):
|
||||
login()
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────────
|
||||
# get_session() — convenience factory
|
||||
# ─────────────────────────────────────────────────
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_get_session_force_login():
|
||||
respx.get(ANTI_FORGERY_API).respond(
|
||||
json={"HeaderName": "__RequestVerificationToken", "Token": "t"}
|
||||
)
|
||||
respx.post(SIGNIN_API).respond(json={"SuccessfulResult": {}})
|
||||
|
||||
with patch.dict(
|
||||
os.environ,
|
||||
{"CANTALOUPE_EMAIL": "u@test.com", "CANTALOUPE_PASSWORD": "pw"},
|
||||
):
|
||||
client = get_session(force_login=True)
|
||||
|
||||
assert isinstance(client, httpx.Client)
|
||||
assert respx.get(ANTI_FORGERY_API).called
|
||||
assert respx.post(SIGNIN_API).called
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_get_session_cached():
|
||||
respx.get("https://mycantaloupe.com/cs4/VueMachineList").respond(
|
||||
status_code=200,
|
||||
headers={
|
||||
"Set-Cookie": "ApplicationGatewayAffinity=cached-cookie; Path=/"
|
||||
},
|
||||
)
|
||||
|
||||
with patch.dict(
|
||||
os.environ,
|
||||
{"CANTALOUPE_EMAIL": "u@test.com", "CANTALOUPE_PASSWORD": "pw"},
|
||||
):
|
||||
client = get_session()
|
||||
|
||||
assert isinstance(client, httpx.Client)
|
||||
assert client.cookies.get("ApplicationGatewayAffinity") == "cached-cookie"
|
||||
assert not respx.get(ANTI_FORGERY_API).called
|
||||
assert not respx.post(SIGNIN_API).called
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_get_session_probe_fails_triggers_login():
|
||||
respx.get("https://mycantaloupe.com/cs4/VueMachineList").respond(status_code=200)
|
||||
respx.get(ANTI_FORGERY_API).respond(
|
||||
json={"HeaderName": "__RequestVerificationToken", "Token": "t"}
|
||||
)
|
||||
respx.post(SIGNIN_API).respond(json={"SuccessfulResult": {}})
|
||||
|
||||
with patch.dict(
|
||||
os.environ,
|
||||
{"CANTALOUPE_EMAIL": "u@test.com", "CANTALOUPE_PASSWORD": "pw"},
|
||||
):
|
||||
client = get_session()
|
||||
|
||||
assert isinstance(client, httpx.Client)
|
||||
assert respx.post(SIGNIN_API).called
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────────
|
||||
# AuthSession.refresh()
|
||||
# ─────────────────────────────────────────────────
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_auth_session_refresh_cookies_present():
|
||||
client = httpx.Client(base_url="https://mycantaloupe.com")
|
||||
client.cookies.set("ApplicationGatewayAffinity", "still-here", domain="mycantaloupe.com")
|
||||
session = AuthSession(client=client, email="u@test.com", password="pw")
|
||||
assert session.refresh() is True
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_auth_session_refresh_relogin_success():
|
||||
client = httpx.Client(base_url="https://mycantaloupe.com")
|
||||
session = AuthSession(client=client, email="u@test.com", password="pw")
|
||||
|
||||
respx.get(ANTI_FORGERY_API).respond(
|
||||
json={"HeaderName": "__RequestVerificationToken", "Token": "new-token"}
|
||||
)
|
||||
respx.post(SIGNIN_API).respond(
|
||||
json={"SuccessfulResult": {}},
|
||||
headers={
|
||||
"Set-Cookie": "ApplicationGatewayAffinity=new-auth; Path=/"
|
||||
},
|
||||
)
|
||||
|
||||
assert session.refresh() is True
|
||||
assert session.client.cookies.get("ApplicationGatewayAffinity") == "new-auth"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_auth_session_refresh_relogin_fails():
|
||||
client = httpx.Client(base_url="https://mycantaloupe.com")
|
||||
session = AuthSession(client=client, email="u@test.com", password="bad")
|
||||
|
||||
respx.get(ANTI_FORGERY_API).respond(
|
||||
json={"HeaderName": "__RequestVerificationToken", "Token": "t"}
|
||||
)
|
||||
respx.post(SIGNIN_API).respond(
|
||||
json={"ErrorMessage": "Invalid credentials"}
|
||||
)
|
||||
|
||||
assert session.refresh() is False
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────────
|
||||
# Constants
|
||||
# ─────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_auth_cookie_names():
|
||||
assert "ApplicationGatewayAffinity" in AUTH_COOKIE_NAMES
|
||||
assert "AeonPrincipalCacheVersion" in AUTH_COOKIE_NAMES
|
||||
|
||||
|
||||
def test_url_constants():
|
||||
assert SIGNIN_API == "https://mycantaloupe.com/login/api/SignIn/SignIn/en-US"
|
||||
assert ANTI_FORGERY_API == "https://mycantaloupe.com/login/api/DataContext/GenerateAntiForgeryState"
|
||||
Reference in New Issue
Block a user