Marius Mutu
907b7be0fd
- FastAPI app with lifespan, CORS, health endpoint - SQLAlchemy 2.0 async with aiosqlite, Base/UUIDMixin/TenantMixin/TimestampMixin - Tenant and User models with multi-tenant isolation - Auth: register (creates tenant+user), login, /me endpoint - JWT HS256 tokens, bcrypt password hashing - Alembic async setup with initial migration - 6 passing tests (register, login, wrong password, me, no token, health) Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
106 lines
2.9 KiB
Python
106 lines
2.9 KiB
Python
import pytest
|
|
from httpx import ASGITransport, AsyncClient
|
|
|
|
from app.main import app
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_register_creates_tenant():
|
|
async with AsyncClient(
|
|
transport=ASGITransport(app=app), base_url="http://test"
|
|
) as c:
|
|
r = await c.post(
|
|
"/api/auth/register",
|
|
json={
|
|
"email": "owner@service.ro",
|
|
"password": "parola123",
|
|
"tenant_name": "Service Ionescu",
|
|
"telefon": "0722000000",
|
|
},
|
|
)
|
|
assert r.status_code == 200
|
|
data = r.json()
|
|
assert "access_token" in data
|
|
assert data["plan"] == "trial"
|
|
assert data["token_type"] == "bearer"
|
|
assert data["tenant_id"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_login_returns_token():
|
|
async with AsyncClient(
|
|
transport=ASGITransport(app=app), base_url="http://test"
|
|
) as c:
|
|
await c.post(
|
|
"/api/auth/register",
|
|
json={
|
|
"email": "test@s.ro",
|
|
"password": "abc123",
|
|
"tenant_name": "Test",
|
|
"telefon": "0722",
|
|
},
|
|
)
|
|
r = await c.post(
|
|
"/api/auth/login",
|
|
json={"email": "test@s.ro", "password": "abc123"},
|
|
)
|
|
assert r.status_code == 200
|
|
assert "access_token" in r.json()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_login_wrong_password_returns_401():
|
|
async with AsyncClient(
|
|
transport=ASGITransport(app=app), base_url="http://test"
|
|
) as c:
|
|
r = await c.post(
|
|
"/api/auth/login",
|
|
json={"email": "x@x.ro", "password": "wrong"},
|
|
)
|
|
assert r.status_code == 401
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_me_returns_user_info():
|
|
async with AsyncClient(
|
|
transport=ASGITransport(app=app), base_url="http://test"
|
|
) as c:
|
|
reg = await c.post(
|
|
"/api/auth/register",
|
|
json={
|
|
"email": "me@test.ro",
|
|
"password": "pass123",
|
|
"tenant_name": "My Service",
|
|
"telefon": "0733",
|
|
},
|
|
)
|
|
token = reg.json()["access_token"]
|
|
r = await c.get(
|
|
"/api/auth/me",
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
assert r.status_code == 200
|
|
data = r.json()
|
|
assert data["email"] == "me@test.ro"
|
|
assert data["rol"] == "owner"
|
|
assert data["plan"] == "trial"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_me_without_token_returns_403():
|
|
async with AsyncClient(
|
|
transport=ASGITransport(app=app), base_url="http://test"
|
|
) as c:
|
|
r = await c.get("/api/auth/me")
|
|
assert r.status_code == 401
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_health_endpoint():
|
|
async with AsyncClient(
|
|
transport=ASGITransport(app=app), base_url="http://test"
|
|
) as c:
|
|
r = await c.get("/api/health")
|
|
assert r.status_code == 200
|
|
assert r.json() == {"status": "ok"}
|