Compare commits

..
8 Commits
Author SHA1 Message Date
Agzamus 4f2e66e811 Create README.md 2026-10-12 19:48:57 +03:00
Agzamus 8f98adc7cd feat: seed 2026-10-12 19:15:50 +03:00
Agzamus 0ac5efdc91 Update .gitignore 2026-10-12 19:15:12 +03:00
Agzamus 444c24f297 ui: rename header 2026-10-12 18:11:46 +03:00
Agzamus f5aa99d147 feat: 11. Тесты 2026-10-12 18:11:04 +03:00
Agzamus 6ecffc2934 Create run.py 2026-10-12 17:16:33 +03:00
Agzamus ba9e09f342 feat: 10. Мини-админка (vanilla JS + GraphQL) 2026-10-12 17:15:57 +03:00
Agzamus 9d9e190328 feat: 9.5. Сборка и монтирование 2026-10-12 17:12:27 +03:00
13 changed files with 912 additions and 1 deletions
+1
View File
@@ -7,3 +7,4 @@ venv/
.pytest_cache/
*.egg-info/
*.docx
.~lock.*#
+269
View File
@@ -0,0 +1,269 @@
# Лабараторная 2
Минимальный костяк сервиса управления пользователями и OAuth-клиентами. Предоставляет REST API по пути `/auth/*`, GraphQL-эндпоинт по адресу `/graphql` и небольшое админ-SPA на чистом JS, отдаваемое по `/admin`.
Основные технологии:
- Python 3.12
- **FastAPI** - REST API (`/auth/register`, `/auth/login`, `/auth/me`)
- **Ariadne** - GraphQL-сервер
- **SQLAlchemy 2.x** - ORM
- **Alembic** - миграции
- **SQLite** - база данных
- **argon2-cffi** - хеширование паролей Argon2id + стойкость через `zxcvbn`
- **python-jose** - JWT токены
## Установка
```sh
git clone https://vantablacktea.ddns.net/Agzamus/scripts-2.git
cd scripts-2
```
### С помощью uv
[Скачать uv](https://docs.astral.sh/uv/)
```sh
uv sync
uv run python run.py
```
Тесты:
```sh
uv run pytest -q
```
### Альтернатива с pip
Linux:
```sh
python -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
```
Windows:
```sh
python -m venv .venv
source .venv\Scripts\activate
pip install -r requirements.txt
```
## Настройка
Необходимо создать `.env` файл, например скопировав пример:
```sh
cp .env.example .env
```
> Записать для `JWT_SECRET` и `PASSWORD_PEPPER` длинные случайные строки
> `JWT_SECRET` подписывает каждый access-токен, а `PASSWORD_PEPPER` добавляется к паролю перед хешированием
Переменные из `.env.example`:
| Переменная | Пример | Назначение |
| ---------------------- | ----------------------------------------- | ------------------------------------------------- |
| `DATABASE_URL` | `sqlite:///./vault.db` | Строка подключения SQLAlchemy |
| `JWT_SECRET` | `replace_with_a_long_random_string` | Секрет для подписи/проверки JWT |
| `JWT_ALGORITHM` | `HS256` | Алгоритм подписи JWT |
| `ACCESS_TOKEN_TTL_MIN` | `60` | Время жизни access-токена в минутах |
| `PASSWORD_PEPPER` | `replace_with_another_long_random_string` | Глобальный секрет, добавляемый перед хешированием |
| `PASSWORD_MIN_LENGTH` | `12` | Минимальная длина пароля |
| `ENVIRONMENT` | `development` | Метка окружения выполнения |
| `APP_NAME` | `Vault` | Заголовок приложения |
## Запуск
Примените миграции базы данных, затем запустите сервер:
```bash
uv run alembic upgrade head # создать/обновить схему базы данных
uv run python run.py # uvicorn на 0.0.0.0:8000 с reload
```
Или напрямую:
```bash
uv run uvicorn app.main:app --reload
```
После запуска:
- GraphQL Playground: <http://localhost:8000/graphql>
- Admin UI: <http://localhost:8000/admin>
### Заполнение демо-данными
##### Администратор
Может открыть админ-панель
Почта:
```
admin@example.com
```
Пароль:
```
Kz8#Fj2!Lm9@Qw4$xR
```
##### Пользователь
```
user@example.com
```
```
DemoUser-Pass#2024
```
##### OAuth-клиент
`demo-client` - "Demo App"
```
redirect_uris = ["https://example.com/callback"]
```
#### Заполнение:
Убедитесь, что схема создана (запустите один раз или при каждом изменении миграций):
```bash
uv run alembic upgrade head
```
Запустите скрипт заполнения:
```bash
uv run python seed.py
```
Запустите сервер:
```bash
uv run python run.py
```
Админ панель: <http://localhost:8000/admin>
#### Свои учётные данные
Переопределите значения по умолчанию через переменные окружения, например:
```bash
SEED_ADMIN_EMAIL=root@corp.io \
SEED_ADMIN_PASSWORD='S0me-very-Long-Random-Passphrase!42' \
uv run python seed.py
```
| Переменная | Значение по умолчанию |
| --------------------- | --------------------- |
| `SEED_ADMIN_EMAIL` | `admin@example.com` |
| `SEED_ADMIN_PASSWORD` | `Kz8#Fj2!Lm9@Qw4$xR` |
| `SEED_USER_EMAIL` | `user@example.com` |
| `SEED_USER_PASSWORD` | `DemoUser-Pass#2024` |
> Пароли из seed-скрипта обходят политику стойкости при регистрации
#### Сброс
Чтобы сбросить базу данных:
```sh
rm -f vault.db
uv run alembic upgrade head
uv run python seed.py
```
## Структура проекта
```
.
├── alembic/ # окружение миграций Alembic
├── alembic.ini # конфигурация Alembic
├── app/
│ ├── main.py # FastAPI приложение
│ ├── config.py # настройки из .env
│ ├── db/ # движок, SessionLocal, get_db() и декларативная Base
│ ├── models/ # модели SQLAlchemy: User, OAuthClient, AuditEvent
│ ├── core/ # JWT, хеширование паролей, зависимости аутентификации
│ ├── auth/ # REST и схемы Pydantic
│ ├── audit/ # запись событий в базу данных
│ ├── graphql/ # схема Ariadne, резолверы, контекст и директивы
│ │ └── schema.graphql # SDL GraphQL
│ └── static/ # Фронт
├── tests/ # тесты Pytest
├── run.py # точка входа Uvicorn
├── seed.py # заполнение демо-данными
├── pyproject.toml # метаданные проекта и зависимости (uv)
└── requirements.txt # зафиксированные зависимости для установки через pip
```
<!-- ## Обзор API -->
<!---->
<!-- ### REST (`/auth`) -->
<!---->
<!-- | Метод | Путь | Аутентификация | Описание | -->
<!-- | --- | --- | --- | --- | -->
<!-- | `POST` | `/auth/register` | нет | Зарегистрировать нового пользователя; возвращает access-токен | -->
<!-- | `POST` | `/auth/login` | нет | Аутентифицировать и вернуть access-токен | -->
<!-- | `GET` | `/auth/me` | bearer-токен | Вернуть текущего пользователя (`id`, `email`, `is_admin`) | -->
<!---->
<!-- ### GraphQL (`/graphql`) -->
<!---->
<!-- Запросы (queries): -->
<!---->
<!-- | Query | Аутентификация | Описание | -->
<!-- | --- | --- | --- | -->
<!-- | `me` | аутентифицирован | Текущий пользователь | -->
<!-- | `users(search, limit, offset)` | администратор | Список пользователей с опциональной фильтрацией по email | -->
<!-- | `clients` | администратор | Список зарегистрированных OAuth-клиентов | -->
<!-- | `auditEvents(eventType, limit, offset)` | администратор | Список событий аудита | -->
<!---->
<!-- Мутации (mutations): -->
<!---->
<!-- | Mutation | Аутентификация | Описание | -->
<!-- | --- | --- | --- | -->
<!-- | `registerUser(email, password)` | нет | Создать пользователя; возвращает access-токен | -->
<!-- | `createClient(name, isConfidential)` | администратор | Создать OAuth-клиент | -->
<!-- | `disableUser(userId)` | администратор | Деактивировать пользователя | -->
## Пример GraphQL
Передайте JWT как bearer-токен:
```sh
curl http://localhost:8000/graphql/ \
-H 'Content-Type: application/json' \
-H 'Authorization: Bearer <token>' \
-d '{"query": "{ me { email } }"}'
```
Тело запроса:
```graphql
{ me { email } }
```
Пример ответа:
```json
{
"data": {
"me": {
"email": "user@example.com"
}
}
}
```
## Тестирование
```bash
uv run pytest -q
```
## Примечания
`.env` и `*.db` не коммитятся - они перечислены в `.gitignore` вместе с `__pycache__/`, `.venv/` и другими локальными артефактами
Держите секреты и локальные базы данных вне системы контроля версий
+11 -1
View File
@@ -1,7 +1,8 @@
import json
from typing import Any
from uuid import uuid4
from ariadne import MutationType, QueryType, convert_kwargs_to_snake_case
from ariadne import MutationType, ObjectType, QueryType, convert_kwargs_to_snake_case
from graphql import GraphQLError
from sqlalchemy import select
@@ -19,6 +20,8 @@ from app.models import AuditEvent, OAuthClient, User
query = QueryType()
mutation = MutationType()
audit_event = ObjectType("AuditEvent")
# me – возвращает `info.context["user"]`;
@query.field("me")
@@ -128,3 +131,10 @@ def resolve_disable_user(_, info: Any, user_id: int) -> User:
actor = info.context.get("user")
write_event(db, "user.disabled", actor_id=actor.id if actor else None)
return user
@audit_event.field("metadata")
def resolve_metadata(obj: AuditEvent, info: Any) -> str | None:
if obj.metadata_json is None:
return None
return json.dumps(obj.metadata_json, ensure_ascii=False)
+44
View File
@@ -0,0 +1,44 @@
from datetime import datetime
from pathlib import Path
from typing import Any
from ariadne import ScalarType, load_schema_from_path, make_executable_schema
from ariadne import snake_case_fallback_resolvers
from ariadne.asgi import GraphQL
from app.graphql.context import get_context_value
from app.graphql.directives import IsAdminDirective, IsAuthenticatedDirective
from app.graphql.resolvers import audit_event, mutation, query
datetime_scalar = ScalarType("DateTime")
@datetime_scalar.serializer
def _serialize_datetime(value: Any) -> str:
if isinstance(value, datetime):
return value.isoformat()
return str(value)
# Загрузить SDL-схему из файла (в Ariadne есть функция load_schema_from_path)
type_defs = load_schema_from_path(str(Path(__file__).parent / "schema.graphql"))
# Собрать исполняемую схему через `make_executable_schema`, передав:
schema = make_executable_schema(
# типы;
type_defs,
# объекты `query` и `mutation` из `resolvers.py`;
query,
mutation,
audit_event,
datetime_scalar,
snake_case_fallback_resolvers,
# словарь директив `{"isAuthenticated": ..., "isAdmin": ...}`.
directives={
"isAuthenticated": IsAuthenticatedDirective,
"isAdmin": IsAdminDirective,
},
)
# Обернуть схему в ASGI-приложение `GraphQL(schema, context_value=..., debug=True)`.
graphql_app = GraphQL(schema, context_value=get_context_value, debug=True)
+10
View File
@@ -1,8 +1,18 @@
from pathlib import Path
from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles
from app.auth.router import router as auth_router
from app.config import settings
from app.graphql.schema import graphql_app
app = FastAPI(title=settings.app_name)
app.include_router(auth_router)
app.mount("/graphql", graphql_app)
app.mount(
"/admin",
StaticFiles(directory=str(Path(__file__).parent / "static"), html=True),
name="admin",
)
+133
View File
@@ -0,0 +1,133 @@
async function api(path, body) {
const res = await fetch(path, {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify(body),
});
const json = await res.json().catch(() => ({}));
if (!res.ok) {
throw new Error(json.detail || res.statusText);
}
return json;
}
async function gql(query, variables) {
const headers = { "Content-Type": "application/json" };
const token = localStorage.getItem("access_token");
if (token) {
headers.Authorization = `Bearer ${token}`;
}
const res = await fetch("/graphql", {
method: "POST",
headers,
body: JSON.stringify({ query, variables }),
});
const json = await res.json();
if (json.errors && json.errors.length) {
throw new Error(json.errors[0].message);
}
return json.data;
}
const CLIENTS_QUERY = `
query { clients { id clientId name isConfidential redirectUris createdAt } }
`;
document.addEventListener("DOMContentLoaded", () => {
const loginForm = document.getElementById("login-form");
const panel = document.getElementById("panel");
const statusEl = document.getElementById("status");
const clientsBody = document.getElementById("clients-body");
const createBtn = document.getElementById("create-client");
const logoutBtn = document.getElementById("logout");
function setStatus(message, ok = false) {
statusEl.textContent = message || "";
statusEl.classList.toggle("ok", ok);
}
function setLoggedIn(isLoggedIn) {
loginForm.classList.toggle("hidden", isLoggedIn);
panel.classList.toggle("hidden", !isLoggedIn);
}
function renderClients(clients) {
clientsBody.textContent = "";
for (const client of clients) {
const row = document.createElement("tr");
const cells = [
client.id,
client.clientId,
client.name,
client.isConfidential ? "yes" : "no",
(client.redirectUris || []).join(", "),
client.createdAt,
];
for (const value of cells) {
const td = document.createElement("td");
td.textContent = value;
row.appendChild(td);
}
clientsBody.appendChild(row);
}
}
async function loadClients() {
try {
const data = await gql(CLIENTS_QUERY);
renderClients(data.clients);
setStatus("");
} catch (err) {
renderClients([]);
setStatus(err.message);
}
}
async function handleLogin(event) {
event.preventDefault();
setStatus("");
const email = document.getElementById("email").value;
const password = document.getElementById("password").value;
try {
const { access_token } = await api("/auth/login", { email, password });
localStorage.setItem("access_token", access_token);
setLoggedIn(true);
await loadClients();
} catch (err) {
setStatus(err.message);
}
}
async function handleCreateClient() {
const name = prompt("Название клиента");
if (!name) {
return;
}
try {
await gql(
"mutation ($name: String!) { createClient(name: $name) { id clientId name } }",
{ name }
);
setStatus("");
await loadClients();
} catch (err) {
setStatus(err.message);
}
}
function handleLogout() {
localStorage.removeItem("access_token");
setLoggedIn(false);
setStatus("");
loginForm.reset();
}
loginForm.addEventListener("submit", handleLogin);
createBtn.addEventListener("click", handleCreateClient);
logoutBtn.addEventListener("click", handleLogout);
if (localStorage.getItem("access_token")) {
setLoggedIn(true);
loadClients();
}
});
+142
View File
@@ -0,0 +1,142 @@
<!DOCTYPE html>
<html lang="ru">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>Vault Admin</title>
<style>
:root {
--bg: #f4f6f8;
--card: #ffffff;
--border: #d7dce1;
--text: #1f2933;
--muted: #6b7280;
--accent: #2563eb;
--accent-hover: #1d4ed8;
--danger: #b91c1c;
}
* { box-sizing: border-box; }
body {
margin: 0;
font-family: system-ui, -apple-system, "Segoe UI", Roboto, sans-serif;
background: var(--bg);
color: var(--text);
}
.wrap {
max-width: 880px;
margin: 40px auto;
padding: 0 16px;
}
.card {
background: var(--card);
border: 1px solid var(--border);
border-radius: 8px;
padding: 20px;
}
h1 { font-size: 20px; margin: 0 0 16px; }
label {
display: block;
font-size: 13px;
color: var(--muted);
margin: 12px 0 4px;
}
input[type="email"],
input[type="password"] {
width: 100%;
padding: 8px 10px;
border: 1px solid var(--border);
border-radius: 6px;
font-size: 14px;
}
button {
font: inherit;
padding: 8px 14px;
border: 1px solid transparent;
border-radius: 6px;
background: var(--accent);
color: #fff;
cursor: pointer;
}
button:hover { background: var(--accent-hover); }
button.secondary {
background: #fff;
color: var(--text);
border-color: var(--border);
}
button.secondary:hover { background: #f0f2f4; }
.toolbar {
display: flex;
justify-content: space-between;
align-items: center;
margin-bottom: 16px;
}
.actions { display: flex; gap: 8px; }
table {
width: 100%;
border-collapse: collapse;
font-size: 14px;
}
th, td {
text-align: left;
padding: 8px 10px;
border-bottom: 1px solid var(--border);
}
th {
font-size: 12px;
text-transform: uppercase;
letter-spacing: 0.03em;
color: var(--muted);
}
#status {
min-height: 20px;
margin-top: 14px;
font-size: 14px;
color: var(--danger);
white-space: pre-wrap;
}
#status.ok { color: #15803d; }
.hidden { display: none; }
</style>
</head>
<body>
<div class="wrap">
<form id="login-form" class="card">
<h1>Вход</h1>
<label for="email">Email</label>
<input id="email" type="email" autocomplete="username" required>
<label for="password">Пароль</label>
<input id="password" type="password" autocomplete="current-password" required>
<div style="margin-top:16px">
<button type="submit">Войти</button>
</div>
</form>
<div id="panel" class="card hidden">
<div class="toolbar">
<h1 style="margin:0">Клиенты</h1>
<div class="actions">
<button id="create-client" type="button">Создать клиента</button>
<button id="logout" type="button" class="secondary">Выйти</button>
</div>
</div>
<table>
<thead>
<tr>
<th>ID</th>
<th>Client ID</th>
<th>Название</th>
<th>Confidential</th>
<th>Redirect URIs</th>
<th>Создан</th>
</tr>
</thead>
<tbody id="clients-body"></tbody>
</table>
</div>
<div id="status" role="status" aria-live="polite"></div>
</div>
<script src="app.js"></script>
</body>
</html>
+4
View File
@@ -0,0 +1,4 @@
import uvicorn
if __name__ == "__main__":
uvicorn.run("app.main:app", host="0.0.0.0", port=8000, reload=True)
+69
View File
@@ -0,0 +1,69 @@
import os
from sqlalchemy import select
from sqlalchemy.orm import Session
from app.core.passwords import hash_password
from app.db import SessionLocal
from app.models import OAuthClient, User
ADMIN_EMAIL = os.getenv("SEED_ADMIN_EMAIL", "admin@example.com")
ADMIN_PASSWORD = os.getenv("SEED_ADMIN_PASSWORD", "Kz8#Fj2!Lm9@Qw4$xR")
DEMO_EMAIL = os.getenv("SEED_USER_EMAIL", "user@example.com")
DEMO_PASSWORD = os.getenv("SEED_USER_PASSWORD", "DemoUser-Pass#2024")
DEMO_CLIENT_ID = "demo-client"
DEMO_CLIENT_URIS = ["https://example.com/callback"]
def upsert_user(db: Session, email: str, password: str, is_admin: bool) -> User:
user = db.scalar(select(User).where(User.email == email))
if user is None:
user = User(email=email, hashed_password=hash_password(password))
db.add(user)
user.is_admin = is_admin
user.is_active = True
db.commit()
db.refresh(user)
return user
def upsert_client(db: Session) -> OAuthClient:
client = db.scalar(
select(OAuthClient).where(OAuthClient.client_id == DEMO_CLIENT_ID)
)
if client is None:
client = OAuthClient(client_id=DEMO_CLIENT_ID, name="Demo App")
db.add(client)
client.name = "Demo App"
client.is_confidential = True
client.redirect_uris = DEMO_CLIENT_URIS
db.commit()
db.refresh(client)
return client
def main() -> None:
db = SessionLocal()
try:
admin = upsert_user(db, ADMIN_EMAIL, ADMIN_PASSWORD, is_admin=True)
demo = upsert_user(db, DEMO_EMAIL, DEMO_PASSWORD, is_admin=False)
client = upsert_client(db)
summary = [
(admin.email, admin.id),
(demo.email, demo.id),
(client.client_id, client.name),
]
finally:
db.close()
admin_email, admin_id = summary[0]
demo_email, demo_id = summary[1]
client_id, client_name = summary[2]
print("Seed complete:")
print(f" admin {admin_email} / {ADMIN_PASSWORD} (id={admin_id})")
print(f" user {demo_email} / {DEMO_PASSWORD} (id={demo_id})")
print(f" client {client_id} - {client_name}")
if __name__ == "__main__":
main()
View File
+70
View File
@@ -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
+73
View File
@@ -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
+86
View File
@@ -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)