Exception nazi and other suppression removals. Added tests on auth flows that were simplified to linter requirements.
This commit is contained in:
@@ -0,0 +1,220 @@
|
||||
from http.cookies import SimpleCookie
|
||||
from pathlib import Path
|
||||
from uuid import uuid4
|
||||
|
||||
import pytest
|
||||
import pytest_asyncio
|
||||
from sanic import Sanic
|
||||
|
||||
from cista import auth, config
|
||||
from cista.app import use_session
|
||||
from cista.auth import bp as auth_bp
|
||||
|
||||
|
||||
def _set_cookie_headers(response) -> list[str]:
|
||||
return list(response.headers.get_list("set-cookie"))
|
||||
|
||||
|
||||
def _cookie_header(response, name: str = "cista") -> dict[str, str]:
|
||||
for header in _set_cookie_headers(response):
|
||||
cookie = SimpleCookie()
|
||||
cookie.load(header)
|
||||
morsel = cookie.get(name)
|
||||
if morsel is not None and morsel.value:
|
||||
return {"Cookie": f"{name}={morsel.value}"}
|
||||
raise AssertionError(f"response did not set cookie {name!r}")
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def setup_auth_config(tmp_path: Path):
|
||||
alice = config.User()
|
||||
auth.set_password(alice, "secret")
|
||||
admin = config.User(privileged=True)
|
||||
auth.set_password(admin, "admin-secret")
|
||||
config.config = config.Config(
|
||||
path=tmp_path,
|
||||
listen=":0",
|
||||
public=False,
|
||||
users={"alice": alice, "admin": admin},
|
||||
)
|
||||
return tmp_path
|
||||
|
||||
|
||||
@pytest_asyncio.fixture()
|
||||
async def client(setup_auth_config: Path):
|
||||
app = Sanic(f"auth-builtins-test-{uuid4().hex}", strict_slashes=True)
|
||||
|
||||
@app.on_request
|
||||
async def load_auth_context(request):
|
||||
await use_session(request)
|
||||
|
||||
app.blueprint(auth_bp)
|
||||
yield app.asgi_client
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_restricted_page_renders_login_form_when_logged_out(client):
|
||||
_, res = await client.get("/auth/restricted/")
|
||||
|
||||
assert res.status_code == 200
|
||||
assert "Authentication Required" in res.text
|
||||
assert "Username:" in res.text
|
||||
assert "Password:" in res.text
|
||||
assert "/auth/login" in res.text
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_restricted_page_with_invalid_session_clears_cookie(client):
|
||||
_, res = await client.get(
|
||||
"/auth/restricted/",
|
||||
headers={"Cookie": "cista=missing-session"},
|
||||
)
|
||||
|
||||
assert res.status_code == 200
|
||||
assert "Authentication Required" in res.text
|
||||
assert any("cista=" in header.lower() for header in _set_cookie_headers(res))
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_json_login_sets_session_cookie_and_allows_session_authenticated_api_access(
|
||||
client,
|
||||
):
|
||||
_, res = await client.post(
|
||||
"/auth/login",
|
||||
json={"username": "alice", "password": "secret"},
|
||||
)
|
||||
|
||||
assert res.status_code == 200
|
||||
assert res.json == {"data": {"username": "alice", "privileged": False}}
|
||||
|
||||
session_cookie = _cookie_header(res)
|
||||
|
||||
_, tokens_res = await client.get("/auth/tokens", headers=session_cookie)
|
||||
assert tokens_res.status_code == 200
|
||||
assert tokens_res.json == {"tokens": []}
|
||||
|
||||
_, restricted_res = await client.get("/auth/restricted/", headers=session_cookie)
|
||||
assert restricted_res.status_code == 200
|
||||
assert "auth-success" in restricted_res.text
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_json_login_rejects_missing_fields(client):
|
||||
_, res = await client.post(
|
||||
"/auth/login",
|
||||
json={"username": "alice"},
|
||||
)
|
||||
|
||||
assert res.status_code == 400
|
||||
assert "Missing username or password" in res.json["message"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_json_login_rejects_invalid_password(client):
|
||||
_, res = await client.post(
|
||||
"/auth/login",
|
||||
json={"username": "alice", "password": "wrong"},
|
||||
)
|
||||
|
||||
assert res.status_code == 403
|
||||
assert "Invalid password" in res.json["message"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_html_login_redirects_and_sets_flash_and_session_cookies(client):
|
||||
_, res = await client.post(
|
||||
"/auth/login",
|
||||
data={"username": "alice", "password": "secret"},
|
||||
headers={"Accept": "text/html"},
|
||||
follow_redirects=False,
|
||||
)
|
||||
|
||||
assert res.status_code == 302
|
||||
assert res.headers["location"] == "/"
|
||||
headers = _set_cookie_headers(res)
|
||||
assert any("cista=" in header.lower() for header in headers)
|
||||
assert any("message=" in header.lower() for header in headers)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_logout_json_revokes_the_existing_session(client):
|
||||
_, login_res = await client.post(
|
||||
"/auth/login",
|
||||
json={"username": "alice", "password": "secret"},
|
||||
)
|
||||
session_cookie = _cookie_header(login_res)
|
||||
|
||||
_, logout_res = await client.post("/auth/api/logout", headers=session_cookie)
|
||||
|
||||
assert logout_res.status_code == 200
|
||||
assert logout_res.json == {"message": "Logged out"}
|
||||
assert any("cista=" in header.lower() for header in _set_cookie_headers(logout_res))
|
||||
|
||||
_, retry_res = await client.get("/auth/tokens", headers=session_cookie)
|
||||
assert retry_res.status_code == 401
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_logout_without_session_reports_not_logged_in(client):
|
||||
_, res = await client.post("/auth/api/logout")
|
||||
|
||||
assert res.status_code == 200
|
||||
assert res.json == {"message": "Not logged in"}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_password_change_updates_credentials_and_reissues_session(client):
|
||||
_, change_res = await client.post(
|
||||
"/auth/password-change",
|
||||
json={
|
||||
"username": "alice",
|
||||
"password": "secret",
|
||||
"passwordChange": "fresh-secret",
|
||||
},
|
||||
)
|
||||
|
||||
assert change_res.status_code == 200
|
||||
assert change_res.json == {"message": "Password updated"}
|
||||
|
||||
session_cookie = _cookie_header(change_res)
|
||||
_, tokens_res = await client.get("/auth/tokens", headers=session_cookie)
|
||||
assert tokens_res.status_code == 200
|
||||
|
||||
_, old_login_res = await client.post(
|
||||
"/auth/login",
|
||||
json={"username": "alice", "password": "secret"},
|
||||
)
|
||||
assert old_login_res.status_code == 403
|
||||
|
||||
_, new_login_res = await client.post(
|
||||
"/auth/login",
|
||||
json={"username": "alice", "password": "fresh-secret"},
|
||||
)
|
||||
assert new_login_res.status_code == 200
|
||||
assert new_login_res.json == {"data": {"username": "alice", "privileged": False}}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_password_change_rejects_wrong_current_password(client):
|
||||
_, res = await client.post(
|
||||
"/auth/password-change",
|
||||
json={
|
||||
"username": "alice",
|
||||
"password": "wrong",
|
||||
"passwordChange": "fresh-secret",
|
||||
},
|
||||
)
|
||||
|
||||
assert res.status_code == 403
|
||||
assert "Invalid password" in res.json["message"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_password_change_rejects_missing_fields(client):
|
||||
_, res = await client.post(
|
||||
"/auth/password-change",
|
||||
json={"username": "alice", "password": "secret"},
|
||||
)
|
||||
|
||||
assert res.status_code == 400
|
||||
assert "Missing username, passwordChange or password" in res.json["message"]
|
||||
Reference in New Issue
Block a user