feat(auth): one-time login links, JWT + refresh, roles, service token, tests

This commit is contained in:
apuc committed 2026-09-26 19:52:32 +03:00
1 parent 4a610ba94a
commit 049b679654
12 files changed
+553 -1

No files matched your search

+35
View File
@@ -0,0 +1,35 @@
"""Общие фикстуры для тестов."""
import os
os.environ.setdefault("DATABASE_URL", "sqlite:///./_test.db")
os.environ.setdefault("JWT_SECRET", "test-secret")
os.environ.setdefault("BOT_SERVICE_TOKEN", "svc-token")
import pytest # noqa: E402
from fastapi.testclient import TestClient # noqa: E402
import app.models # noqa: E402,F401
from app.db.session import Base, SessionLocal, engine # noqa: E402
from app.main import app # noqa: E402
@pytest.fixture(scope="session", autouse=True)
def _setup_db():
Base.metadata.drop_all(engine)
Base.metadata.create_all(engine)
yield
Base.metadata.drop_all(engine)
@pytest.fixture()
def db():
session = SessionLocal()
try:
yield session
finally:
session.close()
@pytest.fixture()
def client():
return TestClient(app)
+108
View File
@@ -0,0 +1,108 @@
"""Тесты аутентификации."""
from datetime import datetime, timedelta, timezone
from uuid import uuid4
from app.models.enums import MessengerType, RoleCode
from app.models.user import LoginToken, Role, User, UserRole
def _seed_admin(db) -> User:
for code, name in [(RoleCode.ADMIN, "Админ"), (RoleCode.COORDINATOR, "Координатор")]:
if not db.query(Role).filter(Role.code == code).first():
db.add(Role(code=code, name=name))
db.commit()
admin = (
db.query(User)
.filter(User.messenger_type == MessengerType.TELEGRAM, User.messenger_id == "1")
.first()
)
if not admin:
admin = User(messenger_type=MessengerType.TELEGRAM, messenger_id="1", full_name="Админ")
db.add(admin)
db.flush()
role = db.query(Role).filter(Role.code == RoleCode.ADMIN).first()
if not db.query(UserRole).filter(
UserRole.user_id == admin.id, UserRole.role_id == role.id
).first():
db.add(UserRole(user_id=admin.id, role_id=role.id))
db.commit()
return admin
def _make_login_token(db, value: str) -> None:
db.add(
LoginToken(
id=str(uuid4()),
token=value,
expires_at=datetime.now(timezone.utc) + timedelta(minutes=30),
)
)
db.commit()
def test_login_flow(client, db):
_seed_admin(db)
_make_login_token(db, "tok-1")
# вход
r = client.post(
"/api/v1/auth/login",
json={"token": "tok-1", "messenger_type": "telegram", "messenger_id": "1"},
)
assert r.status_code == 200, r.text
data = r.json()
assert data["user"]["roles"][0]["code"] == "admin"
access, refresh = data["access_token"], data["refresh_token"]
# повторное использование ссылки
r2 = client.post(
"/api/v1/auth/login",
json={"token": "tok-1", "messenger_type": "telegram", "messenger_id": "1"},
)
assert r2.status_code == 409
# профиль
r3 = client.get("/api/v1/auth/me", headers={"Authorization": f"Bearer {access}"})
assert r3.status_code == 200
# выдача новой ссылки (как админ)
r4 = client.post(
"/api/v1/auth/login-token",
json={"ttl_minutes": 15},
headers={"Authorization": f"Bearer {access}"},
)
assert r4.status_code == 201, r4.text
assert "token=" in r4.json()["url"]
# refresh
r5 = client.post("/api/v1/auth/refresh", json={"refresh_token": refresh})
assert r5.status_code == 200, r5.text
new_access = r5.json()["access_token"]
# logout
r6 = client.post("/api/v1/auth/logout", headers={"Authorization": f"Bearer {new_access}"})
assert r6.status_code == 204
# после logout
r7 = client.get("/api/v1/auth/me", headers={"Authorization": f"Bearer {new_access}"})
assert r7.status_code == 401
def test_login_token_requires_role(client, db):
_seed_admin(db)
_make_login_token(db, "tok-2")
r = client.post(
"/api/v1/auth/login",
json={"token": "tok-2", "messenger_type": "telegram", "messenger_id": "999"},
)
assert r.status_code == 200
access = r.json()["access_token"]
# обычный пользователь без роли не может выдавать ссылки
r2 = client.post(
"/api/v1/auth/login-token",
json={"ttl_minutes": 15},
headers={"Authorization": f"Bearer {access}"},
)
assert r2.status_code == 403