feat: 11. Тесты
This commit is contained in:
@@ -0,0 +1,70 @@
|
|||||||
|
import os
|
||||||
|
|
||||||
|
os.environ["DATABASE_URL"] = "sqlite:///./test_vault.db"
|
||||||
|
os.environ["JWT_SECRET"] = "test-secret"
|
||||||
|
os.environ["PASSWORD_PEPPER"] = "test-pepper"
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
from fastapi.testclient import TestClient
|
||||||
|
|
||||||
|
from app.core.passwords import hash_password
|
||||||
|
from app.db import SessionLocal, engine
|
||||||
|
from app.db.base import Base
|
||||||
|
import app.models # noqa: F401,E402 register models on Base.metadata
|
||||||
|
from app.main import app as fastapi_app
|
||||||
|
from app.models import User
|
||||||
|
|
||||||
|
STRONG_PASSWORD = "Kz8#Fj2!Lm9@Qw4$xR"
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture(autouse=True)
|
||||||
|
def _clean_database():
|
||||||
|
Base.metadata.drop_all(bind=engine)
|
||||||
|
Base.metadata.create_all(bind=engine)
|
||||||
|
yield
|
||||||
|
Base.metadata.drop_all(bind=engine)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def client():
|
||||||
|
with TestClient(fastapi_app) as c:
|
||||||
|
yield c
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def db_session():
|
||||||
|
db = SessionLocal()
|
||||||
|
try:
|
||||||
|
yield db
|
||||||
|
finally:
|
||||||
|
db.close()
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def make_user():
|
||||||
|
def _make_user(email, password=STRONG_PASSWORD, is_admin=False):
|
||||||
|
db = SessionLocal()
|
||||||
|
try:
|
||||||
|
user = User(
|
||||||
|
email=email,
|
||||||
|
hashed_password=hash_password(password),
|
||||||
|
is_admin=is_admin,
|
||||||
|
is_active=True,
|
||||||
|
)
|
||||||
|
db.add(user)
|
||||||
|
db.commit()
|
||||||
|
db.refresh(user)
|
||||||
|
return user
|
||||||
|
finally:
|
||||||
|
db.close()
|
||||||
|
|
||||||
|
return _make_user
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def make_admin(make_user):
|
||||||
|
def _make_admin(email="admin@vault.test", password=STRONG_PASSWORD):
|
||||||
|
return make_user(email=email, password=password, is_admin=True)
|
||||||
|
|
||||||
|
return _make_admin
|
||||||
|
|
||||||
@@ -0,0 +1,73 @@
|
|||||||
|
from sqlalchemy import select
|
||||||
|
|
||||||
|
from app.models import AuditEvent
|
||||||
|
|
||||||
|
STRONG_PASSWORD = "Kz8#Fj2!Lm9@Qw4$xR"
|
||||||
|
CREDENTIALS = {"email": "alice@example.com", "password": STRONG_PASSWORD}
|
||||||
|
|
||||||
|
|
||||||
|
# слабый пароль не регистрируется;
|
||||||
|
def test_register_weak_password_returns_400(client):
|
||||||
|
resp = client.post(
|
||||||
|
"/auth/register", json={"email": "weak@example.com", "password": "password"}
|
||||||
|
)
|
||||||
|
assert resp.status_code == 400
|
||||||
|
|
||||||
|
|
||||||
|
def test_register_strong_password_returns_token(client):
|
||||||
|
resp = client.post("/auth/register", json=CREDENTIALS)
|
||||||
|
assert resp.status_code == 201
|
||||||
|
body = resp.json()
|
||||||
|
assert body["access_token"]
|
||||||
|
assert body["token_type"] == "bearer"
|
||||||
|
|
||||||
|
|
||||||
|
def test_register_then_login(client):
|
||||||
|
client.post("/auth/register", json=CREDENTIALS)
|
||||||
|
resp = client.post("/auth/login", json=CREDENTIALS)
|
||||||
|
assert resp.status_code == 200
|
||||||
|
assert resp.json()["access_token"]
|
||||||
|
|
||||||
|
|
||||||
|
# логин с неверным паролем возвращает 401 и не раскрывает, какое именно поле неверно;
|
||||||
|
def test_login_wrong_password_returns_401(client):
|
||||||
|
client.post("/auth/register", json=CREDENTIALS)
|
||||||
|
resp = client.post(
|
||||||
|
"/auth/login",
|
||||||
|
json={"email": CREDENTIALS["email"], "password": "completely-wrong"},
|
||||||
|
)
|
||||||
|
assert resp.status_code == 401
|
||||||
|
assert resp.json()["detail"] == "Invalid credentials"
|
||||||
|
|
||||||
|
|
||||||
|
def test_register_duplicate_email_returns_409(client):
|
||||||
|
client.post("/auth/register", json=CREDENTIALS)
|
||||||
|
resp = client.post("/auth/register", json=CREDENTIALS)
|
||||||
|
assert resp.status_code == 409
|
||||||
|
|
||||||
|
|
||||||
|
# `/auth/me` без токена → 401;
|
||||||
|
def test_me_requires_token(client):
|
||||||
|
resp = client.get("/auth/me")
|
||||||
|
assert resp.status_code == 401
|
||||||
|
|
||||||
|
|
||||||
|
def test_me_with_token(client):
|
||||||
|
token = client.post("/auth/register", json=CREDENTIALS).json()["access_token"]
|
||||||
|
resp = client.get("/auth/me", headers={"Authorization": f"Bearer {token}"})
|
||||||
|
assert resp.status_code == 200
|
||||||
|
body = resp.json()
|
||||||
|
assert body["email"] == CREDENTIALS["email"]
|
||||||
|
assert body["is_admin"] is False
|
||||||
|
assert isinstance(body["id"], int)
|
||||||
|
|
||||||
|
|
||||||
|
# после успешного логина в таблице `audit_events` появляется `user.login.success`.
|
||||||
|
def test_successful_login_records_audit_event(client, db_session):
|
||||||
|
client.post("/auth/register", json=CREDENTIALS)
|
||||||
|
client.post("/auth/login", json=CREDENTIALS)
|
||||||
|
event = db_session.scalar(
|
||||||
|
select(AuditEvent).where(AuditEvent.event_type == "user.login.success")
|
||||||
|
)
|
||||||
|
assert event is not None
|
||||||
|
|
||||||
@@ -0,0 +1,86 @@
|
|||||||
|
from app.core.jwt import create_access_token
|
||||||
|
|
||||||
|
STRONG_PASSWORD = "Kz8#Fj2!Lm9@Qw4$xR"
|
||||||
|
|
||||||
|
USERS_QUERY = "{ users { email } }"
|
||||||
|
ME_QUERY = "{ me { email } }"
|
||||||
|
|
||||||
|
REGISTER_MUTATION = """
|
||||||
|
mutation Register($email: String!, $password: String!) {
|
||||||
|
registerUser(email: $email, password: $password) {
|
||||||
|
accessToken
|
||||||
|
tokenType
|
||||||
|
}
|
||||||
|
}
|
||||||
|
"""
|
||||||
|
|
||||||
|
|
||||||
|
def _graphql(client, query, variables=None, token=None):
|
||||||
|
headers = {}
|
||||||
|
if token:
|
||||||
|
headers["Authorization"] = f"Bearer {token}"
|
||||||
|
return client.post(
|
||||||
|
"/graphql",
|
||||||
|
json={"query": query, "variables": variables},
|
||||||
|
headers=headers,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def _errors(data):
|
||||||
|
return [e.get("message", "") for e in data.get("errors", [])]
|
||||||
|
|
||||||
|
|
||||||
|
def test_users_requires_authentication(client):
|
||||||
|
data = _graphql(client, USERS_QUERY).json()
|
||||||
|
assert "Authentication required" in _errors(data)
|
||||||
|
|
||||||
|
|
||||||
|
# GraphQL-запрос `users` без админ-прав → ошибка «Admin only»;
|
||||||
|
def test_users_denied_for_non_admin(client, make_user):
|
||||||
|
user = make_user(email="bob@vault.test")
|
||||||
|
token = create_access_token(str(user.id))
|
||||||
|
data = _graphql(client, USERS_QUERY, token=token).json()
|
||||||
|
assert "Admin only" in _errors(data)
|
||||||
|
|
||||||
|
|
||||||
|
def test_users_allowed_for_admin(client, make_admin):
|
||||||
|
admin = make_admin()
|
||||||
|
token = create_access_token(str(admin.id))
|
||||||
|
data = _graphql(client, USERS_QUERY, token=token).json()
|
||||||
|
assert data.get("errors") in (None, [])
|
||||||
|
assert isinstance(data["data"]["users"], list)
|
||||||
|
|
||||||
|
|
||||||
|
def test_register_user_public_mutation(client):
|
||||||
|
data = _graphql(
|
||||||
|
client,
|
||||||
|
REGISTER_MUTATION,
|
||||||
|
{"email": "graph@vault.test", "password": STRONG_PASSWORD},
|
||||||
|
).json()
|
||||||
|
assert data.get("errors") in (None, [])
|
||||||
|
payload = data["data"]["registerUser"]
|
||||||
|
assert payload["accessToken"]
|
||||||
|
assert payload["tokenType"] == "bearer"
|
||||||
|
|
||||||
|
|
||||||
|
def test_register_user_rejects_weak_password(client):
|
||||||
|
data = _graphql(
|
||||||
|
client,
|
||||||
|
REGISTER_MUTATION,
|
||||||
|
{"email": "graph-weak@vault.test", "password": "password"},
|
||||||
|
).json()
|
||||||
|
assert data.get("errors")
|
||||||
|
|
||||||
|
|
||||||
|
def test_me_returns_email_with_token(client, make_user):
|
||||||
|
user = make_user(email="me@vault.test")
|
||||||
|
token = create_access_token(str(user.id))
|
||||||
|
data = _graphql(client, ME_QUERY, token=token).json()
|
||||||
|
assert data.get("errors") in (None, [])
|
||||||
|
assert data["data"]["me"]["email"] == "me@vault.test"
|
||||||
|
|
||||||
|
|
||||||
|
def test_me_requires_authentication(client):
|
||||||
|
data = _graphql(client, ME_QUERY).json()
|
||||||
|
assert "Authentication required" in _errors(data)
|
||||||
|
|
||||||
Reference in New Issue
Block a user