From 733e7a8918c360fbb838d196abb2b16cbffa691c Mon Sep 17 00:00:00 2001 From: ribardej Date: Thu, 6 Nov 2025 11:14:57 +0100 Subject: [PATCH] feat(test): added more tests --- 7project/backend/tests/test_e2e.py | 198 ++++++++++++++++++ 7project/backend/tests/test_e2e_auth_flow.py | 98 --------- .../backend/tests/test_integration_app.py | 125 +++++++++++ 3 files changed, 323 insertions(+), 98 deletions(-) create mode 100644 7project/backend/tests/test_e2e.py delete mode 100644 7project/backend/tests/test_e2e_auth_flow.py diff --git a/7project/backend/tests/test_e2e.py b/7project/backend/tests/test_e2e.py new file mode 100644 index 0000000..3171f62 --- /dev/null +++ b/7project/backend/tests/test_e2e.py @@ -0,0 +1,198 @@ +import pytest +import uuid +from httpx import AsyncClient, ASGITransport +from fastapi import status + + +def test_e2e(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) + +@pytest.mark.asyncio +async def test_register_then_login_and_fetch_me(fastapi_app): + transport = ASGITransport(app=fastapi_app, raise_app_exceptions=True) + async with AsyncClient(transport=transport, base_url="http://testserver") as ac: + email = "newuser@example.com" + password = "StrongPassw0rd!" + reg = await ac.post("/auth/register", json={"email": email, "password": password}) + assert reg.status_code in (status.HTTP_201_CREATED, status.HTTP_200_OK) + + login = await ac.post("/auth/jwt/login", data={"username": email, "password": password}) + assert login.status_code == status.HTTP_200_OK + token = login.json()["access_token"] + me = await ac.get("/users/me", headers={"Authorization": f"Bearer {token}"}) + assert me.status_code == status.HTTP_200_OK + assert me.json()["email"] == email + +@pytest.mark.asyncio +async def test_revoked_token_blocked_everywhere(fastapi_app, test_user): + transport = ASGITransport(app=fastapi_app, raise_app_exceptions=True) + async with AsyncClient(transport=transport, base_url="http://testserver") as ac: + login = await ac.post("/auth/jwt/login", data=test_user) + token = login.json()["access_token"] + headers = {"Authorization": f"Bearer {token}"} + + # Sanity check works before logout + ok = await ac.get("/authenticated-route", headers=headers) + assert ok.status_code == status.HTTP_200_OK + + # Logout revokes token + lo = await ac.post("/auth/jwt/logout", headers=headers) + assert lo.status_code in (status.HTTP_200_OK, status.HTTP_204_NO_CONTENT) + + # Token should be rejected on any protected endpoint + blocked = await ac.get("/authenticated-route", headers=headers) + assert blocked.status_code == status.HTTP_401_UNAUTHORIZED + +@pytest.mark.asyncio +async def test_delete_current_user_revokes_access(fastapi_app): + transport = ASGITransport(app=fastapi_app, raise_app_exceptions=True) + async with AsyncClient(transport=transport, base_url="http://testserver") as ac: + email = "todelete@example.com" + password = "Passw0rd!" + reg = await ac.post("/auth/register", json={"email": email, "password": password}) + assert reg.status_code in (status.HTTP_200_OK, status.HTTP_201_CREATED) + + login = await ac.post("/auth/jwt/login", data={"username": email, "password": password}) + token = login.json()["access_token"] + headers = {"Authorization": f"Bearer {token}"} + + # Delete self + d = await ac.delete("/users/me", headers=headers) + assert d.status_code == status.HTTP_204_NO_CONTENT + + # Access should now fail + me = await ac.get("/users/me", headers=headers) + assert me.status_code in (status.HTTP_401_UNAUTHORIZED, status.HTTP_403_FORBIDDEN) + + +@pytest.mark.asyncio +async def test_update_category_conflict_and_404(fastapi_app, test_user): + transport = ASGITransport(app=fastapi_app) + async with AsyncClient(transport=transport, base_url="http://testserver") as ac: + token = (await ac.post("/auth/jwt/login", data=test_user)).json()["access_token"] + h = {"Authorization": f"Bearer {token}"} + + a = (await ac.post("/categories/create", json={"name": "A"}, headers=h)).json() + b = (await ac.post("/categories/create", json={"name": "B"}, headers=h)).json() + + # Attempt to rename A -> B should conflict + conflict = await ac.patch(f"/categories/{a['id']}", json={"name": "B"}, headers=h) + assert conflict.status_code == status.HTTP_409_CONFLICT + + # Update non-existent + missing = await ac.patch("/categories/999999", json={"name": "Z"}, headers=h) + assert missing.status_code == status.HTTP_404_NOT_FOUND + +@pytest.mark.asyncio +async def test_category_cross_user_isolation(fastapi_app): + transport = ASGITransport(app=fastapi_app) + async with AsyncClient(transport=transport, base_url="http://testserver") as ac: + # user1 + u1 = {"email": "u1@example.com", "password": "Aaaaaa1!"} + assert (await ac.post("/auth/register", json=u1)).status_code in (200, 201) + t1 = (await ac.post("/auth/jwt/login", data={"username": u1["email"], "password": u1["password"]})).json()["access_token"] + + # user1 creates a category + c = (await ac.post("/categories/create", json={"name": "Private"}, headers={"Authorization": f"Bearer {t1}"})).json() + + # user2 + u2 = {"email": "u2@example.com", "password": "Aaaaaa1!"} + assert (await ac.post("/auth/register", json=u2)).status_code in (200, 201) + t2 = (await ac.post("/auth/jwt/login", data={"username": u2["email"], "password": u2["password"]})).json()["access_token"] + + # user2 cannot read/delete user1's category + g = await ac.get(f"/categories/{c['id']}", headers={"Authorization": f"Bearer {t2}"}) + assert g.status_code == status.HTTP_404_NOT_FOUND + d = await ac.delete(f"/categories/{c['id']}", headers={"Authorization": f"Bearer {t2}"}) + assert d.status_code == status.HTTP_404_NOT_FOUND + diff --git a/7project/backend/tests/test_e2e_auth_flow.py b/7project/backend/tests/test_e2e_auth_flow.py deleted file mode 100644 index 7ad18d2..0000000 --- a/7project/backend/tests/test_e2e_auth_flow.py +++ /dev/null @@ -1,98 +0,0 @@ -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) \ No newline at end of file diff --git a/7project/backend/tests/test_integration_app.py b/7project/backend/tests/test_integration_app.py index c924e25..32e462f 100644 --- a/7project/backend/tests/test_integration_app.py +++ b/7project/backend/tests/test_integration_app.py @@ -64,3 +64,128 @@ async def test_create_transaction_missing_amount_fails(fastapi_app, test_user): # 4. Assert the expected validation error assert resp.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY + + +@pytest.mark.asyncio +async def test_login_invalid_credentials(fastapi_app, test_user): + transport = ASGITransport(app=fastapi_app) + async with AsyncClient(transport=transport, base_url="http://testserver") as ac: + bad = await ac.post("/auth/jwt/login", data={"username": test_user["username"], "password": "nope"}) + assert bad.status_code in (status.HTTP_401_UNAUTHORIZED, status.HTTP_400_BAD_REQUEST) + unknown = await ac.post("/auth/jwt/login", data={"username": "nouser@example.com", "password": "x"}) + assert unknown.status_code in (status.HTTP_401_UNAUTHORIZED, status.HTTP_400_BAD_REQUEST) + + +@pytest.mark.asyncio +async def test_category_duplicate_name_conflict(fastapi_app, test_user): + transport = ASGITransport(app=fastapi_app) + async with AsyncClient(transport=transport, base_url="http://testserver") as ac: + token = (await ac.post("/auth/jwt/login", data=test_user)).json()["access_token"] + h = {"Authorization": f"Bearer {token}"} + + p = {"name": "Food"} + r1 = await ac.post("/categories/create", json=p, headers=h) + assert r1.status_code == status.HTTP_201_CREATED + r2 = await ac.post("/categories/create", json=p, headers=h) + assert r2.status_code == status.HTTP_409_CONFLICT + +@pytest.mark.asyncio +async def test_create_transaction_invalid_date_format(fastapi_app, test_user): + transport = ASGITransport(app=fastapi_app) + async with AsyncClient(transport=transport, base_url="http://testserver") as ac: + token = (await ac.post("/auth/jwt/login", data=test_user)).json()["access_token"] + h = {"Authorization": f"Bearer {token}"} + bad = await ac.post("/transactions/create", json={"amount": 10, "description": "x", "date": "31-12-2024"}, headers=h) + assert bad.status_code == status.HTTP_400_BAD_REQUEST + +@pytest.mark.asyncio +async def test_update_transaction_rejects_duplicate_category_ids(fastapi_app, test_user): + transport = ASGITransport(app=fastapi_app) + async with AsyncClient(transport=transport, base_url="http://testserver") as ac: + token = (await ac.post("/auth/jwt/login", data=test_user)).json()["access_token"] + h = {"Authorization": f"Bearer {token}"} + tx = (await ac.post("/transactions/create", json={"amount": 5, "description": "x"}, headers=h)).json() + dup = await ac.patch(f"/transactions/{tx['id']}/edit", json={"category_ids": [1, 1]}, headers=h) + assert dup.status_code == status.HTTP_400_BAD_REQUEST + +@pytest.mark.asyncio +async def test_assign_unassign_category_not_found_cases(fastapi_app, test_user): + transport = ASGITransport(app=fastapi_app) + async with AsyncClient(transport=transport, base_url="http://testserver") as ac: + token = (await ac.post("/auth/jwt/login", data=test_user)).json()["access_token"] + h = {"Authorization": f"Bearer {token}"} + + # Create tx and category + tx = (await ac.post("/transactions/create", json={"amount": 1, "description": "a"}, headers=h)).json() + cat = (await ac.post("/categories/create", json={"name": "X"}, headers=h)).json() + + # Missing transaction + r1 = await ac.post(f"/transactions/999999/categories/{cat['id']}", headers=h) + assert r1.status_code == status.HTTP_404_NOT_FOUND + + # Missing category + r2 = await ac.post(f"/transactions/{tx['id']}/categories/999999", headers=h) + assert r2.status_code == status.HTTP_404_NOT_FOUND + +@pytest.mark.asyncio +async def test_transactions_date_filter_and_balance_series(fastapi_app, test_user): + transport = ASGITransport(app=fastapi_app) + async with AsyncClient(transport=transport, base_url="http://testserver") as ac: + token = (await ac.post("/auth/jwt/login", data=test_user)).json()["access_token"] + h = {"Authorization": f"Bearer {token}"} + + # Seed transactions spanning days + data = [ + {"amount": 100, "description": "day1", "date": "2024-01-01"}, + {"amount": -25, "description": "day2", "date": "2024-01-02"}, + {"amount": 50, "description": "day3", "date": "2024-01-03"}, + ] + for p in data: + r = await ac.post("/transactions/create", json=p, headers=h) + assert r.status_code == status.HTTP_201_CREATED + + # Filtered list (2nd and 3rd only) + lst = await ac.get("/transactions/", params={"start_date": "2024-01-02", "end_date": "2024-01-03"}, headers=h) + assert lst.status_code == status.HTTP_200_OK + assert len(lst.json()) == 2 + + # Balance series should be cumulative per date + series = await ac.get("/transactions/balance_series", headers=h) + assert series.status_code == status.HTTP_200_OK + s = series.json() + assert s == [ + {"date": "2024-01-01", "balance": 100.0}, + {"date": "2024-01-02", "balance": 75.0}, + {"date": "2024-01-03", "balance": 125.0}, + ] + +@pytest.mark.asyncio +async def test_delete_transaction_not_found(fastapi_app, test_user): + transport = ASGITransport(app=fastapi_app) + async with AsyncClient(transport=transport, base_url="http://testserver") as ac: + token = (await ac.post("/auth/jwt/login", data=test_user)).json()["access_token"] + h = {"Authorization": f"Bearer {token}"} + r = await ac.delete("/transactions/999999/delete", headers=h) + assert r.status_code == status.HTTP_404_NOT_FOUND + +@pytest.mark.asyncio +async def test_debug_csas_endpoints_require_auth_and_queue(fastapi_app, test_user): + transport = ASGITransport(app=fastapi_app) + async with AsyncClient(transport=transport, base_url="http://testserver") as ac: + # unauthenticated should be blocked + unauth = await ac.get("/debug/scrape/csas/all") + assert unauth.status_code in (status.HTTP_401_UNAUTHORIZED, status.HTTP_403_FORBIDDEN) + + token = (await ac.post("/auth/jwt/login", data=test_user)).json()["access_token"] + h = {"Authorization": f"Bearer {token}"} + + all_resp = await ac.get("/debug/scrape/csas/all", headers=h) + assert all_resp.status_code == status.HTTP_200_OK + assert all_resp.json()["status"] == "queued" + + # Single-user CSAS requires auth and user dep; using current user id via /users/me + me = await ac.get("/users/me", headers=h) + uid = me.json()["id"] + one = await ac.post(f"/debug/scrape/csas/{uid}", headers=h) + assert one.status_code == status.HTTP_200_OK + assert one.json()["status"] == "queued"