mirror of
https://github.com/dat515-2025/Group-8.git
synced 2026-03-22 15:12:08 +01:00
98 lines
4.5 KiB
Python
98 lines
4.5 KiB
Python
import pytest
|
|
import uuid
|
|
from httpx import AsyncClient, ASGITransport
|
|
from fastapi import status
|
|
|
|
|
|
def test_e2e_minimal_auth_flow(client):
|
|
# 1) Service is alive
|
|
alive = client.get("/")
|
|
assert alive.status_code == status.HTTP_200_OK
|
|
|
|
# 2) Attempt to login without payload should fail fast (validation error)
|
|
login = client.post("/auth/jwt/login")
|
|
assert login.status_code in (status.HTTP_400_BAD_REQUEST, status.HTTP_422_UNPROCESSABLE_CONTENT)
|
|
|
|
# 3) Protected endpoint should not be accessible without token
|
|
me = client.get("/users/me")
|
|
assert me.status_code in (status.HTTP_401_UNAUTHORIZED, status.HTTP_403_FORBIDDEN)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_e2e_full_user_lifecycle(fastapi_app, test_user):
|
|
# Use an AsyncClient with ASGITransport for async tests
|
|
transport = ASGITransport(app=fastapi_app, raise_app_exceptions=True)
|
|
async with AsyncClient(transport=transport, base_url="http://testserver") as ac:
|
|
login_payload = test_user
|
|
|
|
# 1. Log in with the new credentials
|
|
login_resp = await ac.post("/auth/jwt/login", data=login_payload)
|
|
assert login_resp.status_code == status.HTTP_200_OK
|
|
token = login_resp.json()["access_token"]
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
|
|
# 2. Access a protected endpoint
|
|
me_resp = await ac.get("/users/me", headers=headers)
|
|
assert me_resp.status_code == status.HTTP_200_OK
|
|
assert me_resp.json()["email"] == test_user["username"]
|
|
|
|
# 3. Update the user's profile
|
|
update_payload = {"first_name": "Test"}
|
|
patch_resp = await ac.patch("/users/me", json=update_payload, headers=headers)
|
|
assert patch_resp.status_code == status.HTTP_200_OK
|
|
assert patch_resp.json()["first_name"] == "Test"
|
|
|
|
# 4. Log out
|
|
logout_resp = await ac.post("/auth/jwt/logout", headers=headers)
|
|
assert logout_resp.status_code in (status.HTTP_200_OK, status.HTTP_204_NO_CONTENT)
|
|
|
|
# 5. Verify token is invalid
|
|
me_again_resp = await ac.get("/users/me", headers=headers)
|
|
assert me_again_resp.status_code == status.HTTP_401_UNAUTHORIZED
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_e2e_transaction_workflow(fastapi_app, test_user):
|
|
transport = ASGITransport(app=fastapi_app, raise_app_exceptions=True)
|
|
async with AsyncClient(transport=transport, base_url="http://testserver") as ac:
|
|
# 1. Log in to get the token
|
|
login_resp = await ac.post("/auth/jwt/login", data=test_user)
|
|
token = login_resp.json()["access_token"]
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
|
|
# NEW STEP: Create a category first to get a valid ID
|
|
category_payload = {"name": "Test Category for E2E"}
|
|
create_category_resp = await ac.post("/categories/create", json=category_payload, headers=headers)
|
|
assert create_category_resp.status_code == status.HTTP_201_CREATED
|
|
category_id = create_category_resp.json()["id"]
|
|
|
|
# 2. Create a new transaction
|
|
tx_payload = {"amount": -55.40, "description": "Milk and eggs"}
|
|
tx_resp = await ac.post("/transactions/create", json=tx_payload, headers=headers)
|
|
assert tx_resp.status_code == status.HTTP_201_CREATED
|
|
tx_id = tx_resp.json()["id"]
|
|
|
|
# 3. Assign the category
|
|
assign_resp = await ac.post(f"/transactions/{tx_id}/categories/{category_id}", headers=headers)
|
|
assert assign_resp.status_code == status.HTTP_200_OK
|
|
|
|
# 4. Verify assignment
|
|
get_tx_resp = await ac.get(f"/transactions/{tx_id}", headers=headers)
|
|
assert category_id in get_tx_resp.json()["category_ids"]
|
|
|
|
# 5. Unassign the category
|
|
unassign_resp = await ac.delete(f"/transactions/{tx_id}/categories/{category_id}", headers=headers)
|
|
assert unassign_resp.status_code == status.HTTP_200_OK
|
|
|
|
# 6. Get the transaction again and verify the category is gone
|
|
get_tx_again_resp = await ac.get(f"/transactions/{tx_id}", headers=headers)
|
|
final_tx_data = get_tx_again_resp.json()
|
|
assert category_id not in final_tx_data["category_ids"]
|
|
|
|
# 7. Delete the transaction for cleanup
|
|
delete_resp = await ac.delete(f"/transactions/{tx_id}/delete", headers=headers)
|
|
assert delete_resp.status_code in (status.HTTP_200_OK, status.HTTP_204_NO_CONTENT)
|
|
|
|
# NEW STEP: Clean up the created category
|
|
delete_category_resp = await ac.delete(f"/categories/{category_id}", headers=headers)
|
|
assert delete_category_resp.status_code in (status.HTTP_200_OK, status.HTTP_204_NO_CONTENT) |