Initial commit: backend scaffold, auth, frontend login
- FastAPI + SQLAlchemy async + Alembic + Postgres backend - Auth: JWT access + rotating refresh tokens, argon2, roles, audit log - React 19 + Vite frontend: login page, protected route, auth context - Docker Compose: postgres, redis, migrate, api, worker (placeholder), frontend/nginx Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Claude Sonnet 5
@@ -0,0 +1,18 @@
|
||||
"""Общая настройка тестов.
|
||||
|
||||
Переменные окружения проставляются до импорта `app.core.config`, иначе
|
||||
Settings прочитает рабочий .env (или упадёт на его отсутствии).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
|
||||
from cryptography.fernet import Fernet
|
||||
|
||||
os.environ.setdefault("ENVIRONMENT", "local")
|
||||
os.environ.setdefault("SECRET_KEY", "test-secret-key-not-for-production")
|
||||
os.environ.setdefault("ENCRYPTION_KEY", Fernet.generate_key().decode())
|
||||
os.environ.setdefault("POSTGRES_HOST", "localhost")
|
||||
os.environ.setdefault("POSTGRES_PASSWORD", "test")
|
||||
os.environ.setdefault("CORS_ORIGINS", "")
|
||||
@@ -0,0 +1,57 @@
|
||||
"""Сборка DSN из настроек.
|
||||
|
||||
Регрессия: `PostgresDsn.build` не экранирует username/password сам —
|
||||
спецсимвол в пароле (запятая, `@`, `:`, `/`) ломает разбор URL. Баг
|
||||
воспроизводится только с «настоящим» паролем и незаметен на тестовых
|
||||
значениях вроде `test` или `postgres`.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
from app.core.config import Settings
|
||||
|
||||
|
||||
def _settings(**overrides: str) -> Settings:
|
||||
defaults = {
|
||||
"secret_key": "test-secret-key",
|
||||
"encryption_key": "dGVzdC1lbmNyeXB0aW9uLWtleS0zMi1ieXRlcyEh",
|
||||
"postgres_host": "postgres",
|
||||
"postgres_port": 5432,
|
||||
"postgres_db": "lux_fiscal",
|
||||
"postgres_user": "lux_fiscal",
|
||||
"postgres_password": "postgres",
|
||||
}
|
||||
return Settings(**{**defaults, **overrides}) # type: ignore[arg-type]
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"password",
|
||||
[
|
||||
"Jvth,fqkjn2020", # запятая — реальный пароль из этого проекта
|
||||
"p@ss:word/with#special?chars",
|
||||
"простой-пароль-с-кириллицей",
|
||||
],
|
||||
)
|
||||
def test_database_url_survives_special_characters_in_password(password: str) -> None:
|
||||
settings = _settings(postgres_password=password)
|
||||
|
||||
url = settings.database_url
|
||||
|
||||
assert url.startswith("postgresql+asyncpg://")
|
||||
# Порт обязан остаться числом 5432 — при поломанном парсинге он
|
||||
# «съезжает» вместе с частью пароля или пропадает совсем.
|
||||
assert "@postgres:5432/lux_fiscal" in url
|
||||
|
||||
|
||||
def test_database_url_roundtrips_via_sqlalchemy_make_url() -> None:
|
||||
"""URL должен оставаться валидным для драйвера, а не только для pydantic."""
|
||||
from sqlalchemy.engine import make_url
|
||||
|
||||
settings = _settings(postgres_password="Jvth,fqkjn2020")
|
||||
parsed = make_url(settings.database_url)
|
||||
|
||||
assert parsed.password == "Jvth,fqkjn2020"
|
||||
assert parsed.port == 5432
|
||||
assert parsed.database == "lux_fiscal"
|
||||
@@ -0,0 +1,83 @@
|
||||
"""Миграция 0001 обязана точно соответствовать моделям.
|
||||
|
||||
Расхождение между `Base.metadata` и миграцией обнаруживается только на живой
|
||||
БД и обычно уже в проде. Проверка статическая: сравниваем таблицы, колонки,
|
||||
nullability и индексы, которые создаёт миграция, с тем, что описано моделями.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from app.db.models import Base
|
||||
|
||||
MIGRATION = Path(__file__).resolve().parents[1] / "alembic" / "versions" / "0001_users_and_audit.py"
|
||||
SOURCE = MIGRATION.read_text(encoding="utf-8")
|
||||
|
||||
# Миграция 0001 создаёт схему с нуля, поэтому в ней обязаны быть все таблицы.
|
||||
MODEL_TABLES = set(Base.metadata.tables)
|
||||
|
||||
|
||||
def _migration_block(table: str) -> str:
|
||||
"""Тело вызова op.create_table для указанной таблицы.
|
||||
|
||||
Границей служит следующий op.create_table; для последней таблицы — конец файла.
|
||||
"""
|
||||
start = SOURCE.index(f'op.create_table(\n "{table}",')
|
||||
next_table = SOURCE.find("op.create_table(", start + 1)
|
||||
end = next_table if next_table != -1 else len(SOURCE)
|
||||
return SOURCE[start:end]
|
||||
|
||||
|
||||
def test_migration_creates_every_model_table() -> None:
|
||||
created = set(re.findall(r'op\.create_table\(\n\s+"(\w+)"', SOURCE))
|
||||
assert created == MODEL_TABLES, (
|
||||
f"Миграция и модели разошлись. Только в моделях: {MODEL_TABLES - created}; "
|
||||
f"только в миграции: {created - MODEL_TABLES}"
|
||||
)
|
||||
|
||||
|
||||
def test_downgrade_drops_every_created_table() -> None:
|
||||
downgrade = SOURCE[SOURCE.index("def downgrade()") :]
|
||||
dropped = set(re.findall(r'op\.drop_table\("(\w+)"\)', downgrade))
|
||||
assert dropped == MODEL_TABLES, f"downgrade не удаляет: {MODEL_TABLES - dropped}"
|
||||
|
||||
|
||||
@pytest.mark.parametrize("table_name", sorted(MODEL_TABLES))
|
||||
def test_columns_match(table_name: str) -> None:
|
||||
block = _migration_block(table_name)
|
||||
in_migration = set(re.findall(r'sa\.Column\(\n?\s*"(\w+)"', block))
|
||||
in_model = {column.name for column in Base.metadata.tables[table_name].columns}
|
||||
|
||||
assert in_migration == in_model, (
|
||||
f"Таблица {table_name}: только в модели {in_model - in_migration}, "
|
||||
f"только в миграции {in_migration - in_model}"
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("table_name", sorted(MODEL_TABLES))
|
||||
def test_nullability_matches(table_name: str) -> None:
|
||||
block = _migration_block(table_name)
|
||||
model_table = Base.metadata.tables[table_name]
|
||||
|
||||
for column in model_table.columns:
|
||||
match = re.search(rf'sa\.Column\(\n?\s*"{column.name}".*?\n', block)
|
||||
assert match, f"{table_name}.{column.name} отсутствует в миграции"
|
||||
|
||||
migration_nullable = "nullable=True" in match.group(0)
|
||||
assert migration_nullable == column.nullable, (
|
||||
f"{table_name}.{column.name}: в модели nullable={column.nullable}, "
|
||||
f"в миграции nullable={migration_nullable}"
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("table_name", sorted(MODEL_TABLES))
|
||||
def test_indexes_match(table_name: str) -> None:
|
||||
model_indexes = {index.name for index in Base.metadata.tables[table_name].indexes}
|
||||
migration_indexes = set(re.findall(r'op\.create_index\(\s*(?:op\.f\()?"(\w+)"', SOURCE))
|
||||
|
||||
missing = model_indexes - migration_indexes
|
||||
assert not missing, f"Таблица {table_name}: миграция не создаёт индексы {missing}"
|
||||
@@ -0,0 +1,124 @@
|
||||
"""Тесты паролей, JWT и шифрования секретов. Без БД и без сети."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import uuid
|
||||
from datetime import UTC, datetime, timedelta
|
||||
|
||||
import jwt
|
||||
import pytest
|
||||
|
||||
from app.core import crypto
|
||||
from app.core.config import settings
|
||||
from app.core.security import (
|
||||
ALGORITHM,
|
||||
TokenError,
|
||||
create_access_token,
|
||||
decode_access_token,
|
||||
generate_refresh_token,
|
||||
hash_password,
|
||||
hash_refresh_token,
|
||||
verify_password,
|
||||
)
|
||||
|
||||
|
||||
class TestPasswords:
|
||||
def test_verifies_correct_password(self) -> None:
|
||||
assert verify_password("correct horse battery", hash_password("correct horse battery"))
|
||||
|
||||
def test_rejects_wrong_password(self) -> None:
|
||||
assert not verify_password("wrong", hash_password("correct horse battery"))
|
||||
|
||||
def test_hash_is_salted(self) -> None:
|
||||
"""Одинаковые пароли обязаны давать разные хеши."""
|
||||
assert hash_password("same") != hash_password("same")
|
||||
|
||||
def test_rejects_garbage_hash_without_raising(self) -> None:
|
||||
"""Повреждённый хеш в БД не должен ронять вход с 500."""
|
||||
assert not verify_password("anything", "not-a-valid-argon2-hash")
|
||||
|
||||
|
||||
class TestAccessToken:
|
||||
def test_roundtrip_carries_identity(self) -> None:
|
||||
user_id = uuid.uuid4()
|
||||
payload = decode_access_token(create_access_token(user_id, "cashier"))
|
||||
assert payload["sub"] == str(user_id)
|
||||
assert payload["role"] == "cashier"
|
||||
|
||||
def test_rejects_expired_token(self) -> None:
|
||||
expired = jwt.encode(
|
||||
{
|
||||
"sub": str(uuid.uuid4()),
|
||||
"role": "admin",
|
||||
"type": "access",
|
||||
"exp": datetime.now(UTC) - timedelta(minutes=1),
|
||||
},
|
||||
settings.secret_key,
|
||||
algorithm=ALGORITHM,
|
||||
)
|
||||
with pytest.raises(TokenError):
|
||||
decode_access_token(expired)
|
||||
|
||||
def test_rejects_token_signed_with_other_key(self) -> None:
|
||||
forged = jwt.encode(
|
||||
{
|
||||
"sub": str(uuid.uuid4()),
|
||||
"role": "admin",
|
||||
"type": "access",
|
||||
"exp": datetime.now(UTC) + timedelta(hours=1),
|
||||
},
|
||||
"attacker-key",
|
||||
algorithm=ALGORITHM,
|
||||
)
|
||||
with pytest.raises(TokenError):
|
||||
decode_access_token(forged)
|
||||
|
||||
def test_refresh_token_is_not_accepted_as_access(self) -> None:
|
||||
"""Ключевая проверка: подмена типа токена не должна давать доступ."""
|
||||
refresh_shaped = jwt.encode(
|
||||
{
|
||||
"sub": str(uuid.uuid4()),
|
||||
"role": "admin",
|
||||
"type": "refresh",
|
||||
"exp": datetime.now(UTC) + timedelta(days=7),
|
||||
},
|
||||
settings.secret_key,
|
||||
algorithm=ALGORITHM,
|
||||
)
|
||||
with pytest.raises(TokenError, match="access"):
|
||||
decode_access_token(refresh_shaped)
|
||||
|
||||
|
||||
class TestRefreshToken:
|
||||
def test_tokens_are_unique(self) -> None:
|
||||
assert generate_refresh_token()[0] != generate_refresh_token()[0]
|
||||
|
||||
def test_hash_matches_raw_value(self) -> None:
|
||||
raw, stored = generate_refresh_token()
|
||||
assert hash_refresh_token(raw) == stored
|
||||
|
||||
def test_raw_token_is_not_recoverable_from_hash(self) -> None:
|
||||
raw, stored = generate_refresh_token()
|
||||
assert raw not in stored
|
||||
|
||||
|
||||
class TestCrypto:
|
||||
def test_roundtrip(self) -> None:
|
||||
secret = "license-key-abc-123"
|
||||
assert crypto.decrypt(crypto.encrypt(secret)) == secret
|
||||
|
||||
def test_ciphertext_hides_plaintext(self) -> None:
|
||||
assert "license-key" not in crypto.encrypt("license-key-abc-123")
|
||||
|
||||
def test_tampered_ciphertext_is_rejected(self) -> None:
|
||||
token = crypto.encrypt("license-key-abc-123")
|
||||
tampered = token[:-4] + ("AAAA" if not token.endswith("AAAA") else "BBBB")
|
||||
with pytest.raises(crypto.DecryptionError):
|
||||
crypto.decrypt(tampered)
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("value", "expected"),
|
||||
[("", ""), ("abc", "***"), ("abcd", "****"), ("abcdefgh", "****efgh")],
|
||||
)
|
||||
def test_mask(self, value: str, expected: str) -> None:
|
||||
assert crypto.mask(value) == expected
|
||||
Reference in New Issue
Block a user