feat: backend core — models, auth, CRUD, tests
This commit is contained in:
Nox (OpenClaw)
@@ -0,0 +1,77 @@
|
||||
"""
|
||||
Pytest fixtures for integration tests.
|
||||
|
||||
Uses SQLite in-memory (aiosqlite) for speed.
|
||||
Each test function gets a fresh database and HTTP client.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
import pytest_asyncio
|
||||
from httpx import ASGITransport, AsyncClient
|
||||
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
|
||||
|
||||
from app.database import Base, get_session
|
||||
from app.main import app
|
||||
|
||||
TEST_DATABASE_URL = "sqlite+aiosqlite:///:memory:"
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def test_engine():
|
||||
engine = create_async_engine(TEST_DATABASE_URL, echo=False)
|
||||
async with engine.begin() as conn:
|
||||
await conn.run_sync(Base.metadata.create_all)
|
||||
yield engine
|
||||
async with engine.begin() as conn:
|
||||
await conn.run_sync(Base.metadata.drop_all)
|
||||
await engine.dispose()
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def db_session(test_engine):
|
||||
"""Direct database session for use in fixtures and assertions."""
|
||||
session_factory = async_sessionmaker(test_engine, expire_on_commit=False)
|
||||
async with session_factory() as session:
|
||||
yield session
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def client(test_engine) -> AsyncClient:
|
||||
"""HTTP client wired to a fresh in-memory database."""
|
||||
session_factory = async_sessionmaker(test_engine, expire_on_commit=False)
|
||||
|
||||
async def override_get_session():
|
||||
async with session_factory() as session:
|
||||
yield session
|
||||
|
||||
app.dependency_overrides[get_session] = override_get_session
|
||||
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as ac:
|
||||
yield ac
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helper factories
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
async def create_user_and_login(
|
||||
client: AsyncClient,
|
||||
email: str = "user@example.com",
|
||||
password: str = "password123",
|
||||
full_name: str = "Test User",
|
||||
) -> tuple[dict, str]:
|
||||
"""Register a user and return (user_json, access_token)."""
|
||||
reg_resp = await client.post(
|
||||
"/api/v1/auth/register",
|
||||
json={"email": email, "password": password, "full_name": full_name},
|
||||
)
|
||||
assert reg_resp.status_code == 201, reg_resp.text
|
||||
|
||||
login_resp = await client.post(
|
||||
"/api/v1/auth/login",
|
||||
data={"username": email, "password": password},
|
||||
headers={"Content-Type": "application/x-www-form-urlencoded"},
|
||||
)
|
||||
assert login_resp.status_code == 200, login_resp.text
|
||||
return reg_resp.json(), login_resp.json()["access_token"]
|
||||
@@ -0,0 +1,177 @@
|
||||
"""Tests for auth endpoints: register, login, refresh, logout."""
|
||||
|
||||
import pytest
|
||||
from httpx import AsyncClient
|
||||
|
||||
from tests.conftest import create_user_and_login
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_register_success(client: AsyncClient):
|
||||
resp = await client.post(
|
||||
"/api/v1/auth/register",
|
||||
json={"email": "alice@example.com", "password": "s3cr3t", "full_name": "Alice"},
|
||||
)
|
||||
assert resp.status_code == 201
|
||||
data = resp.json()
|
||||
assert data["email"] == "alice@example.com"
|
||||
assert data["full_name"] == "Alice"
|
||||
assert "id" in data
|
||||
assert "hashed_password" not in data
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_register_duplicate_email(client: AsyncClient):
|
||||
payload = {"email": "dup@example.com", "password": "pass", "full_name": "Dup"}
|
||||
await client.post("/api/v1/auth/register", json=payload)
|
||||
resp = await client.post("/api/v1/auth/register", json=payload)
|
||||
assert resp.status_code == 409
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_register_creates_default_categories(client: AsyncClient):
|
||||
_, token = await create_user_and_login(client)
|
||||
resp = await client.get(
|
||||
"/api/v1/categories",
|
||||
headers={"Authorization": f"Bearer {token}"},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
cats = resp.json()
|
||||
assert len(cats) >= 6
|
||||
names = {c["name"] for c in cats}
|
||||
assert "Alimentation" in names
|
||||
assert "Salaire" in names
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_login_success(client: AsyncClient):
|
||||
await client.post(
|
||||
"/api/v1/auth/register",
|
||||
json={"email": "bob@example.com", "password": "mypass", "full_name": "Bob"},
|
||||
)
|
||||
resp = await client.post(
|
||||
"/api/v1/auth/login",
|
||||
data={"username": "bob@example.com", "password": "mypass"},
|
||||
headers={"Content-Type": "application/x-www-form-urlencoded"},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert "access_token" in data
|
||||
assert "refresh_token" in data
|
||||
assert data["token_type"] == "bearer"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_login_wrong_password(client: AsyncClient):
|
||||
await client.post(
|
||||
"/api/v1/auth/register",
|
||||
json={"email": "carol@example.com", "password": "correct", "full_name": "Carol"},
|
||||
)
|
||||
resp = await client.post(
|
||||
"/api/v1/auth/login",
|
||||
data={"username": "carol@example.com", "password": "wrong"},
|
||||
headers={"Content-Type": "application/x-www-form-urlencoded"},
|
||||
)
|
||||
assert resp.status_code == 401
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_login_unknown_email(client: AsyncClient):
|
||||
resp = await client.post(
|
||||
"/api/v1/auth/login",
|
||||
data={"username": "nobody@example.com", "password": "x"},
|
||||
headers={"Content-Type": "application/x-www-form-urlencoded"},
|
||||
)
|
||||
assert resp.status_code == 401
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_access_protected_route_with_valid_token(client: AsyncClient):
|
||||
_, token = await create_user_and_login(client)
|
||||
resp = await client.get(
|
||||
"/api/v1/categories",
|
||||
headers={"Authorization": f"Bearer {token}"},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_access_protected_route_without_token(client: AsyncClient):
|
||||
resp = await client.get("/api/v1/categories")
|
||||
assert resp.status_code == 403 # HTTPBearer returns 403 when no credentials
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_access_protected_route_invalid_token(client: AsyncClient):
|
||||
resp = await client.get(
|
||||
"/api/v1/categories",
|
||||
headers={"Authorization": "Bearer invalidtoken"},
|
||||
)
|
||||
assert resp.status_code == 401
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_refresh_token(client: AsyncClient):
|
||||
await client.post(
|
||||
"/api/v1/auth/register",
|
||||
json={"email": "dave@example.com", "password": "pass", "full_name": "Dave"},
|
||||
)
|
||||
login_resp = await client.post(
|
||||
"/api/v1/auth/login",
|
||||
data={"username": "dave@example.com", "password": "pass"},
|
||||
headers={"Content-Type": "application/x-www-form-urlencoded"},
|
||||
)
|
||||
refresh_token = login_resp.json()["refresh_token"]
|
||||
|
||||
resp = await client.post(
|
||||
"/api/v1/auth/refresh",
|
||||
json={"refresh_token": refresh_token},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
assert "access_token" in resp.json()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_refresh_token_cannot_be_reused(client: AsyncClient):
|
||||
"""Refresh token is revoked after use (rotation)."""
|
||||
await client.post(
|
||||
"/api/v1/auth/register",
|
||||
json={"email": "eve@example.com", "password": "pass", "full_name": "Eve"},
|
||||
)
|
||||
login_resp = await client.post(
|
||||
"/api/v1/auth/login",
|
||||
data={"username": "eve@example.com", "password": "pass"},
|
||||
headers={"Content-Type": "application/x-www-form-urlencoded"},
|
||||
)
|
||||
refresh_token = login_resp.json()["refresh_token"]
|
||||
|
||||
# First use
|
||||
r1 = await client.post("/api/v1/auth/refresh", json={"refresh_token": refresh_token})
|
||||
assert r1.status_code == 200
|
||||
|
||||
# Second use of the same token should fail
|
||||
r2 = await client.post("/api/v1/auth/refresh", json={"refresh_token": refresh_token})
|
||||
assert r2.status_code == 401
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_logout(client: AsyncClient):
|
||||
_, token = await create_user_and_login(client, email="frank@example.com")
|
||||
|
||||
login_resp = await client.post(
|
||||
"/api/v1/auth/login",
|
||||
data={"username": "frank@example.com", "password": "password123"},
|
||||
headers={"Content-Type": "application/x-www-form-urlencoded"},
|
||||
)
|
||||
refresh_token = login_resp.json()["refresh_token"]
|
||||
|
||||
resp = await client.post(
|
||||
"/api/v1/auth/logout",
|
||||
json={"refresh_token": refresh_token},
|
||||
headers={"Authorization": f"Bearer {token}"},
|
||||
)
|
||||
assert resp.status_code == 204
|
||||
|
||||
# Token should be revoked now
|
||||
r = await client.post("/api/v1/auth/refresh", json={"refresh_token": refresh_token})
|
||||
assert r.status_code == 401
|
||||
@@ -0,0 +1,161 @@
|
||||
"""Tests for category CRUD endpoints."""
|
||||
|
||||
import pytest
|
||||
from httpx import AsyncClient
|
||||
|
||||
from tests.conftest import create_user_and_login
|
||||
|
||||
|
||||
async def _get_first_category(client: AsyncClient, token: str) -> dict:
|
||||
resp = await client.get(
|
||||
"/api/v1/categories",
|
||||
headers={"Authorization": f"Bearer {token}"},
|
||||
)
|
||||
return resp.json()[0]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_categories_after_register(client: AsyncClient):
|
||||
_, token = await create_user_and_login(client)
|
||||
resp = await client.get(
|
||||
"/api/v1/categories",
|
||||
headers={"Authorization": f"Bearer {token}"},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
cats = resp.json()
|
||||
assert len(cats) == 9 # 6 expense + 3 income defaults
|
||||
for c in cats:
|
||||
assert c["is_default"] is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_category(client: AsyncClient):
|
||||
_, token = await create_user_and_login(client)
|
||||
resp = await client.post(
|
||||
"/api/v1/categories",
|
||||
json={"name": "Épargne", "type": "expense", "color": "#123456", "icon": "piggy-bank"},
|
||||
headers={"Authorization": f"Bearer {token}"},
|
||||
)
|
||||
assert resp.status_code == 201
|
||||
data = resp.json()
|
||||
assert data["name"] == "Épargne"
|
||||
assert data["color"] == "#123456"
|
||||
assert data["is_default"] is False
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_update_category(client: AsyncClient):
|
||||
_, token = await create_user_and_login(client)
|
||||
create_resp = await client.post(
|
||||
"/api/v1/categories",
|
||||
json={"name": "Old Name", "type": "income"},
|
||||
headers={"Authorization": f"Bearer {token}"},
|
||||
)
|
||||
cat_id = create_resp.json()["id"]
|
||||
|
||||
resp = await client.put(
|
||||
f"/api/v1/categories/{cat_id}",
|
||||
json={"name": "New Name", "color": "#aabbcc"},
|
||||
headers={"Authorization": f"Bearer {token}"},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["name"] == "New Name"
|
||||
assert resp.json()["color"] == "#aabbcc"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_delete_category_no_transactions(client: AsyncClient):
|
||||
_, token = await create_user_and_login(client)
|
||||
create_resp = await client.post(
|
||||
"/api/v1/categories",
|
||||
json={"name": "To Delete", "type": "expense"},
|
||||
headers={"Authorization": f"Bearer {token}"},
|
||||
)
|
||||
cat_id = create_resp.json()["id"]
|
||||
|
||||
resp = await client.delete(
|
||||
f"/api/v1/categories/{cat_id}",
|
||||
headers={"Authorization": f"Bearer {token}"},
|
||||
)
|
||||
assert resp.status_code == 204
|
||||
|
||||
# Category should no longer appear in listing
|
||||
list_resp = await client.get(
|
||||
"/api/v1/categories",
|
||||
headers={"Authorization": f"Bearer {token}"},
|
||||
)
|
||||
ids = [c["id"] for c in list_resp.json()]
|
||||
assert cat_id not in ids
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_delete_category_with_transactions_returns_409(client: AsyncClient):
|
||||
_, token = await create_user_and_login(client)
|
||||
|
||||
# Get an existing default category
|
||||
cat = await _get_first_category(client, token)
|
||||
cat_id = cat["id"]
|
||||
|
||||
# Add a transaction linked to it
|
||||
await client.post(
|
||||
"/api/v1/transactions",
|
||||
json={
|
||||
"amount_cents": 1000,
|
||||
"type": "expense",
|
||||
"category_id": cat_id,
|
||||
"transaction_date": "2026-03-15",
|
||||
},
|
||||
headers={"Authorization": f"Bearer {token}"},
|
||||
)
|
||||
|
||||
resp = await client.delete(
|
||||
f"/api/v1/categories/{cat_id}",
|
||||
headers={"Authorization": f"Bearer {token}"},
|
||||
)
|
||||
assert resp.status_code == 409
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_filter_categories_by_type(client: AsyncClient):
|
||||
_, token = await create_user_and_login(client)
|
||||
|
||||
resp = await client.get(
|
||||
"/api/v1/categories?type=income",
|
||||
headers={"Authorization": f"Bearer {token}"},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
for c in resp.json():
|
||||
assert c["type"] == "income"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_category_isolation_between_users(client: AsyncClient):
|
||||
"""User A cannot see or modify User B's categories."""
|
||||
_, token_a = await create_user_and_login(client, email="a@example.com")
|
||||
_, token_b = await create_user_and_login(client, email="b@example.com")
|
||||
|
||||
# User A creates a category
|
||||
resp = await client.post(
|
||||
"/api/v1/categories",
|
||||
json={"name": "A's private", "type": "expense"},
|
||||
headers={"Authorization": f"Bearer {token_a}"},
|
||||
)
|
||||
cat_id = resp.json()["id"]
|
||||
|
||||
# User B cannot delete it
|
||||
del_resp = await client.delete(
|
||||
f"/api/v1/categories/{cat_id}",
|
||||
headers={"Authorization": f"Bearer {token_b}"},
|
||||
)
|
||||
assert del_resp.status_code == 404
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_nonexistent_category_returns_404(client: AsyncClient):
|
||||
_, token = await create_user_and_login(client)
|
||||
fake_id = "00000000-0000-0000-0000-000000000000"
|
||||
resp = await client.delete(
|
||||
f"/api/v1/categories/{fake_id}",
|
||||
headers={"Authorization": f"Bearer {token}"},
|
||||
)
|
||||
assert resp.status_code == 404
|
||||
@@ -0,0 +1,265 @@
|
||||
"""Tests for transaction CRUD endpoints."""
|
||||
|
||||
import pytest
|
||||
from httpx import AsyncClient
|
||||
|
||||
from tests.conftest import create_user_and_login
|
||||
|
||||
|
||||
async def _get_category_id(client: AsyncClient, token: str, type_: str = "expense") -> str:
|
||||
resp = await client.get(
|
||||
f"/api/v1/categories?type={type_}",
|
||||
headers={"Authorization": f"Bearer {token}"},
|
||||
)
|
||||
return resp.json()[0]["id"]
|
||||
|
||||
|
||||
async def _create_tx(
|
||||
client: AsyncClient,
|
||||
token: str,
|
||||
category_id: str,
|
||||
*,
|
||||
amount_cents: int = 5000,
|
||||
type_: str = "expense",
|
||||
date: str = "2026-03-15",
|
||||
description: str | None = None,
|
||||
) -> dict:
|
||||
payload = {
|
||||
"amount_cents": amount_cents,
|
||||
"type": type_,
|
||||
"category_id": category_id,
|
||||
"transaction_date": date,
|
||||
}
|
||||
if description:
|
||||
payload["description"] = description
|
||||
resp = await client.post(
|
||||
"/api/v1/transactions",
|
||||
json=payload,
|
||||
headers={"Authorization": f"Bearer {token}"},
|
||||
)
|
||||
assert resp.status_code == 201, resp.text
|
||||
return resp.json()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_transaction(client: AsyncClient):
|
||||
_, token = await create_user_and_login(client)
|
||||
cat_id = await _get_category_id(client, token)
|
||||
|
||||
resp = await client.post(
|
||||
"/api/v1/transactions",
|
||||
json={
|
||||
"amount_cents": 4500,
|
||||
"type": "expense",
|
||||
"category_id": cat_id,
|
||||
"description": "Courses",
|
||||
"transaction_date": "2026-03-15",
|
||||
},
|
||||
headers={"Authorization": f"Bearer {token}"},
|
||||
)
|
||||
assert resp.status_code == 201
|
||||
data = resp.json()
|
||||
assert data["amount_cents"] == 4500
|
||||
assert data["type"] == "expense"
|
||||
assert data["description"] == "Courses"
|
||||
assert data["category"]["id"] == cat_id
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_transaction_invalid_amount(client: AsyncClient):
|
||||
_, token = await create_user_and_login(client)
|
||||
cat_id = await _get_category_id(client, token)
|
||||
|
||||
resp = await client.post(
|
||||
"/api/v1/transactions",
|
||||
json={"amount_cents": 0, "type": "expense", "category_id": cat_id, "transaction_date": "2026-03-15"},
|
||||
headers={"Authorization": f"Bearer {token}"},
|
||||
)
|
||||
assert resp.status_code == 422
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_transactions(client: AsyncClient):
|
||||
_, token = await create_user_and_login(client)
|
||||
cat_id = await _get_category_id(client, token)
|
||||
|
||||
await _create_tx(client, token, cat_id, amount_cents=1000)
|
||||
await _create_tx(client, token, cat_id, amount_cents=2000)
|
||||
|
||||
resp = await client.get(
|
||||
"/api/v1/transactions",
|
||||
headers={"Authorization": f"Bearer {token}"},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["total"] == 2
|
||||
assert len(data["items"]) == 2
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_transactions_filter_by_month(client: AsyncClient):
|
||||
_, token = await create_user_and_login(client)
|
||||
cat_id = await _get_category_id(client, token)
|
||||
|
||||
await _create_tx(client, token, cat_id, date="2026-02-10")
|
||||
await _create_tx(client, token, cat_id, date="2026-03-15")
|
||||
await _create_tx(client, token, cat_id, date="2026-03-20")
|
||||
|
||||
resp = await client.get(
|
||||
"/api/v1/transactions?month=2026-03",
|
||||
headers={"Authorization": f"Bearer {token}"},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["total"] == 2
|
||||
for item in data["items"]:
|
||||
assert item["transaction_date"].startswith("2026-03")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_transactions_filter_by_category(client: AsyncClient):
|
||||
_, token = await create_user_and_login(client)
|
||||
cats = (await client.get("/api/v1/categories?type=expense", headers={"Authorization": f"Bearer {token}"})).json()
|
||||
cat_a = cats[0]["id"]
|
||||
cat_b = cats[1]["id"]
|
||||
|
||||
await _create_tx(client, token, cat_a)
|
||||
await _create_tx(client, token, cat_b)
|
||||
|
||||
resp = await client.get(
|
||||
f"/api/v1/transactions?category_id={cat_a}",
|
||||
headers={"Authorization": f"Bearer {token}"},
|
||||
)
|
||||
data = resp.json()
|
||||
assert data["total"] == 1
|
||||
assert data["items"][0]["category"]["id"] == cat_a
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_transactions_filter_by_type(client: AsyncClient):
|
||||
_, token = await create_user_and_login(client)
|
||||
exp_cat = await _get_category_id(client, token, "expense")
|
||||
inc_cat = await _get_category_id(client, token, "income")
|
||||
|
||||
await _create_tx(client, token, exp_cat, type_="expense")
|
||||
await _create_tx(client, token, inc_cat, type_="income")
|
||||
|
||||
resp = await client.get(
|
||||
"/api/v1/transactions?type=income",
|
||||
headers={"Authorization": f"Bearer {token}"},
|
||||
)
|
||||
data = resp.json()
|
||||
assert data["total"] == 1
|
||||
assert data["items"][0]["type"] == "income"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_transaction(client: AsyncClient):
|
||||
_, token = await create_user_and_login(client)
|
||||
cat_id = await _get_category_id(client, token)
|
||||
tx = await _create_tx(client, token, cat_id)
|
||||
|
||||
resp = await client.get(
|
||||
f"/api/v1/transactions/{tx['id']}",
|
||||
headers={"Authorization": f"Bearer {token}"},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["id"] == tx["id"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_update_transaction(client: AsyncClient):
|
||||
_, token = await create_user_and_login(client)
|
||||
cat_id = await _get_category_id(client, token)
|
||||
tx = await _create_tx(client, token, cat_id, amount_cents=1000)
|
||||
|
||||
resp = await client.put(
|
||||
f"/api/v1/transactions/{tx['id']}",
|
||||
json={"amount_cents": 9999, "description": "Updated"},
|
||||
headers={"Authorization": f"Bearer {token}"},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["amount_cents"] == 9999
|
||||
assert resp.json()["description"] == "Updated"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_soft_delete_transaction(client: AsyncClient):
|
||||
_, token = await create_user_and_login(client)
|
||||
cat_id = await _get_category_id(client, token)
|
||||
tx = await _create_tx(client, token, cat_id)
|
||||
|
||||
del_resp = await client.delete(
|
||||
f"/api/v1/transactions/{tx['id']}",
|
||||
headers={"Authorization": f"Bearer {token}"},
|
||||
)
|
||||
assert del_resp.status_code == 204
|
||||
|
||||
# GET should 404
|
||||
get_resp = await client.get(
|
||||
f"/api/v1/transactions/{tx['id']}",
|
||||
headers={"Authorization": f"Bearer {token}"},
|
||||
)
|
||||
assert get_resp.status_code == 404
|
||||
|
||||
# Should not appear in list
|
||||
list_resp = await client.get(
|
||||
"/api/v1/transactions",
|
||||
headers={"Authorization": f"Bearer {token}"},
|
||||
)
|
||||
ids = [t["id"] for t in list_resp.json()["items"]]
|
||||
assert tx["id"] not in ids
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_transaction_isolation_between_users(client: AsyncClient):
|
||||
"""User A cannot see or modify User B's transactions."""
|
||||
_, token_a = await create_user_and_login(client, email="a@example.com")
|
||||
_, token_b = await create_user_and_login(client, email="b@example.com")
|
||||
|
||||
cat_id_a = await _get_category_id(client, token_a)
|
||||
tx = await _create_tx(client, token_a, cat_id_a)
|
||||
|
||||
# User B gets 404 on User A's transaction
|
||||
resp = await client.get(
|
||||
f"/api/v1/transactions/{tx['id']}",
|
||||
headers={"Authorization": f"Bearer {token_b}"},
|
||||
)
|
||||
assert resp.status_code == 404
|
||||
|
||||
# User B's list is empty
|
||||
list_resp = await client.get(
|
||||
"/api/v1/transactions",
|
||||
headers={"Authorization": f"Bearer {token_b}"},
|
||||
)
|
||||
assert list_resp.json()["total"] == 0
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_pagination(client: AsyncClient):
|
||||
_, token = await create_user_and_login(client)
|
||||
cat_id = await _get_category_id(client, token)
|
||||
|
||||
for i in range(5):
|
||||
await _create_tx(client, token, cat_id, amount_cents=100 * (i + 1))
|
||||
|
||||
resp = await client.get(
|
||||
"/api/v1/transactions?page=1&per_page=2",
|
||||
headers={"Authorization": f"Bearer {token}"},
|
||||
)
|
||||
data = resp.json()
|
||||
assert data["total"] == 5
|
||||
assert len(data["items"]) == 2
|
||||
assert data["page"] == 1
|
||||
assert data["per_page"] == 2
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_nonexistent_transaction_returns_404(client: AsyncClient):
|
||||
_, token = await create_user_and_login(client)
|
||||
fake_id = "00000000-0000-0000-0000-000000000000"
|
||||
resp = await client.get(
|
||||
f"/api/v1/transactions/{fake_id}",
|
||||
headers={"Authorization": f"Bearer {token}"},
|
||||
)
|
||||
assert resp.status_code == 404
|
||||
Reference in New Issue
Block a user