mirror of
https://github.com/idrainformatica/PecFlow.git
synced 2026-06-16 12:45:42 +02:00
58a233236c
- docker-compose.yml: PostgreSQL 16, Redis 7, MinIO, Nginx - backend FastAPI: struttura monorepo, config pydantic-settings - modelli SQLAlchemy: tutti i modelli (tenants, users, mailboxes, messages, archival, permissions, labels, audit_log) - migrazione Alembic 0001: schema completo in pure SQL - auth API: login JWT, refresh token rotation, logout, 2FA TOTP (setup/verify/disable) - CRUD utenti: lista, crea, modifica, reset password, soft delete - permessi granulari (Fase 1-A): mailbox_permissions, assegna/revoca/lista - CRUD tenant: gestione super-admin - sicurezza: AES-256-GCM cifratura credenziali IMAP/SMTP, bcrypt password - RLS PostgreSQL: isolamento multi-tenant per request - seed sviluppo: tenant demo + admin + operator - test unit: security (bcrypt, JWT, AES), auth_service - test integration: auth endpoints, users endpoints - CI GitHub Actions: lint (ruff), test (pytest), build Docker, security scan - infra: nginx.conf, redis.conf - Makefile con comandi make dev/test/migrate/seed Definition of Done: ✅ Login, refresh token e TOTP funzionanti ✅ make dev porta in piedi tutto lo stack locale ✅ CI configurata
143 lines
4.6 KiB
Python
143 lines
4.6 KiB
Python
"""
|
|
Test di integrazione per gli endpoint di autenticazione.
|
|
"""
|
|
|
|
import os
|
|
import pytest
|
|
|
|
os.environ.setdefault("ENCRYPTION_KEY", "b" * 64)
|
|
os.environ.setdefault("SECRET_KEY", "integration-test-secret-key-only-for-tests")
|
|
os.environ.setdefault("DATABASE_URL", "sqlite+aiosqlite:///./test_integration.db")
|
|
os.environ.setdefault("DATABASE_URL_SYNC", "sqlite:///./test_integration.db")
|
|
|
|
|
|
class TestLoginEndpoint:
|
|
@pytest.mark.asyncio
|
|
async def test_login_success(self, client, admin_user):
|
|
response = await client.post(
|
|
"/api/v1/auth/login",
|
|
json={
|
|
"email": "admin@test.com",
|
|
"password": "AdminPass1!",
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "access_token" in data
|
|
assert "refresh_token" in data
|
|
assert data["token_type"] == "bearer"
|
|
assert data["expires_in"] > 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_login_wrong_password_returns_401(self, client, admin_user):
|
|
response = await client.post(
|
|
"/api/v1/auth/login",
|
|
json={
|
|
"email": "admin@test.com",
|
|
"password": "WrongPassword1!",
|
|
},
|
|
)
|
|
assert response.status_code == 401
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_login_nonexistent_user_returns_401(self, client):
|
|
response = await client.post(
|
|
"/api/v1/auth/login",
|
|
json={
|
|
"email": "nobody@example.com",
|
|
"password": "Password1!",
|
|
},
|
|
)
|
|
assert response.status_code == 401
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_login_missing_fields_returns_422(self, client):
|
|
response = await client.post(
|
|
"/api/v1/auth/login",
|
|
json={"email": "test@test.com"}, # manca password
|
|
)
|
|
assert response.status_code == 422
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_login_invalid_email_returns_422(self, client):
|
|
response = await client.post(
|
|
"/api/v1/auth/login",
|
|
json={"email": "not-an-email", "password": "Password1!"},
|
|
)
|
|
assert response.status_code == 422
|
|
|
|
|
|
class TestMeEndpoint:
|
|
@pytest.mark.asyncio
|
|
async def test_me_returns_current_user(self, client, admin_token, admin_user):
|
|
response = await client.get(
|
|
"/api/v1/auth/me",
|
|
headers={"Authorization": f"Bearer {admin_token}"},
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["email"] == "admin@test.com"
|
|
assert data["role"] == "admin"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_me_without_token_returns_403(self, client):
|
|
response = await client.get("/api/v1/auth/me")
|
|
assert response.status_code == 403
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_me_with_invalid_token_returns_401(self, client):
|
|
response = await client.get(
|
|
"/api/v1/auth/me",
|
|
headers={"Authorization": "Bearer invalid.token.here"},
|
|
)
|
|
assert response.status_code == 401
|
|
|
|
|
|
class TestHealthEndpoint:
|
|
@pytest.mark.asyncio
|
|
async def test_health_returns_ok(self, client):
|
|
response = await client.get("/health")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["status"] == "ok"
|
|
|
|
|
|
class TestRefreshEndpoint:
|
|
@pytest.mark.asyncio
|
|
async def test_refresh_with_invalid_token_returns_401(self, client):
|
|
response = await client.post(
|
|
"/api/v1/auth/refresh",
|
|
json={"refresh_token": "invalid.token.here"},
|
|
)
|
|
assert response.status_code == 401
|
|
|
|
|
|
class TestTOTPEndpoints:
|
|
@pytest.mark.asyncio
|
|
async def test_totp_setup_returns_qr(self, client, admin_token):
|
|
response = await client.post(
|
|
"/api/v1/auth/totp/setup",
|
|
headers={"Authorization": f"Bearer {admin_token}"},
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "secret" in data
|
|
assert "qr_uri" in data
|
|
assert "qr_image_base64" in data
|
|
assert data["qr_uri"].startswith("otpauth://totp/")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_totp_verify_wrong_code_returns_400(self, client, admin_token):
|
|
# Prima setup
|
|
await client.post(
|
|
"/api/v1/auth/totp/setup",
|
|
headers={"Authorization": f"Bearer {admin_token}"},
|
|
)
|
|
# Poi verify con codice errato
|
|
response = await client.post(
|
|
"/api/v1/auth/totp/verify",
|
|
headers={"Authorization": f"Bearer {admin_token}"},
|
|
json={"totp_code": "000000"},
|
|
)
|
|
assert response.status_code == 400
|