pi3c/blog.pyaqa.ru
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: RBAC E2E тесты и фикс admin-прав для редактирования постов

Browse Source 
Основные изменения:
- Добавлены E2E тесты для проверки ownership (TC-E2E-102/103):
  * test_admin_can_edit_any_post — admin может редактировать любой пост
  * test_user_cannot_edit_other_users_post — user не может редактировать чужой пост
- Исправлены use cases (UpdatePost, DeletePost, PublishPost) — добавлена проверка роли admin
- Обновлены web routes и API routes для передачи роли в use cases
- Добавлены unit тесты для admin-сценариев

Реструктуризация тестов:
- Удалены старые API тесты (tests/api/) — требуют переработки
- Удалены старые integration тесты (tests/integration/)
- Переработаны E2E тесты: удалены старые, добавлены новые с POM
- Добавлена документация тестов: FEATURE_*.md, TEST_MODEL.md, AGENTS.md

Инфраструктура:
- Добавлен MockKeycloakClient для dev-режима
- Добавлены статические файлы: EasyMDE, Highlight.js, стили markdown
- Обновлены шаблоны: base.html, post_form.html, post_detail.html
- Обновлена DI конфигурация и провайдеры

Документация:
- tests/FEATURE_RBAC.md — матрица тестов RBAC
- tests/FEATURE_POST_LIFECYCLE.md — тесты жизненного цикла поста
- tests/FEATURE_DOMAIN_FOUNDATION.md — тесты доменного слоя
- tests/FEATURE_INFRASTRUCTURE.md — тесты инфраструктуры
- tests/TEST_MODEL.md — глобальная матрица покрытия
- app/presentation/web/AGENTS.md — гайд по Web UI
- tests/AGENTS.md — гайд по тестированию
This commit is contained in:
Сергей Ванюшкин
2026-05-07 19:55:15 +03:00
parent 41f2a3d98e
commit 46cc06b596
58 changed files with 4234 additions and 4014 deletions
Download Patch File Download Diff File Expand all files Collapse all files

651
tests/e2e/conftest.py
View File

@@ -1,476 +1,299 @@
"""E2E test fixtures with isolated test server.
"""E2E test configuration for blog application.
Provides fixtures for running E2E tests with:
- Isolated SQLite database per test session
- In-memory fake Keycloak (no external server needed)
- Test server on random port
- Automatic test user creation and authentication
Provides DevAuthProvider for cookie-based dev authentication
and role-specific browser context fixtures.
"""
import asyncio
import contextlib
import os
import socket
import tempfile
from typing import Any
from __future__ import annotations
from collections.abc import Generator
from typing import TYPE_CHECKING, Any
import httpx
import pytest
from dishka import Provider, Scope, make_async_container, provide
from dishka.integrations.fastapi import setup_dishka
from fastapi import FastAPI
from playwright.sync_api import Browser, BrowserContext
from sqlalchemy.ext.asyncio import AsyncEngine, create_async_engine
from pytfm.auth import AuthProvider, TestUser
from app.infrastructure.auth import KeycloakAuthClient
from app.infrastructure.config.settings import Settings
from tests.e2e.fake_keycloak import FakeKeycloakClient
if TYPE_CHECKING:
from playwright.sync_api import Browser, BrowserContext, Page
def pytest_configure(config):
"""Disable pytest-asyncio for E2E tests.
class DevAuthProvider(AuthProvider):
"""Authentication provider for blog dev mode.
pytest-playwright manages its own event loop and conflicts
with pytest-asyncio. We disable asyncio_mode for E2E tests.
Bypasses real Keycloak by generating dev-specific tokens
recognized by MockKeycloakClient.
Attributes:
_users: Mapping of usernames to test users.
"""
if hasattr(config, "option") and hasattr(config.option, "asyncio_mode"):
config.option.asyncio_mode = None
def __init__(self) -> None:
"""Initialize the dev auth provider."""
self._users: dict[str, TestUser] = {}
def _get_free_port() -> int:
"""Get a free TCP port from the OS."""
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
sock.bind(("127.0.0.1", 0))
return sock.getsockname()[1]
def create_user(
self,
username: str,
password: str,
email: str,
roles: list[str] | None = None,
) -> TestUser:
"""Create a test user mapped to a dev token role.
Args:
username: Login name (used as display name).
password: Password (ignored in dev mode).
email: Email address.
roles: List of roles. First role determines dev token.
@pytest.fixture(scope="session")
def browser_type_launch_args() -> dict:
"""Return launch args for browser - ensure headless mode."""
return {"headless": True}
@pytest.fixture(scope="session")
def test_db_path() -> str:
"""Create temporary database file for test session."""
fd, path = tempfile.mkstemp(suffix=".db", prefix="blog_e2e_")
os.close(fd)
yield path
with contextlib.suppress(FileNotFoundError):
os.unlink(path)
@pytest.fixture(scope="session")
def test_database_url(test_db_path: str) -> str:
"""Build database URL for test database."""
return f"sqlite+aiosqlite:///{test_db_path}"
@pytest.fixture(scope="session")
def test_settings(test_database_url: str) -> Settings:
"""Create test settings with isolated database."""
return Settings(
environment="dev",
app={"name": "Blog E2E Test", "debug": True, "host": "127.0.0.1", "port": 0},
db={"url": test_database_url, "echo": False},
kc={"server_url": "http://fake-keycloak:8080", "realm": "test", "client_id": "test"},
security={"secret_key": "test-secret-key-not-for-production"},
)
@pytest.fixture(scope="session")
def test_engine(test_database_url: str):
"""Create database engine for test session."""
import asyncio
engine = create_async_engine(
test_database_url,
echo=False,
future=True,
)
yield engine
# Cleanup
asyncio.run(engine.dispose())
@pytest.fixture(scope="session")
def fake_keycloak():
"""Create fake Keycloak client for testing."""
client = FakeKeycloakClient(token_ttl=3600)
yield client
client.clear()
class FakeKeycloakProvider(Provider):
"""Provider that supplies fake Keycloak client."""
def __init__(self, fake_client: FakeKeycloakClient) -> None:
"""Initialize with fake client."""
self._fake_client = fake_client
super().__init__()
@provide(scope=Scope.APP)
def get_keycloak_client(self) -> KeycloakAuthClient:
"""Provide fake Keycloak client."""
return self._fake_client
@pytest.fixture(scope="session")
def test_server(
test_settings: Settings,
test_engine: AsyncEngine,
fake_keycloak: FakeKeycloakClient,
):
"""Start test server on random port using threading for sync compatibility."""
import threading
import time
from app.infrastructure.database.models import Base
from app.presentation import router
from app.presentation.web import router as web_router
from app.presentation.web.error_handlers import register_error_handlers
from app.presentation.web.flash import setup_flash_manager
port = _get_free_port()
base_url = f"http://127.0.0.1:{port}"
print(f"\n[TestServer] Starting server on port {port}")
# Initialize database using asyncio.run
async def init_db():
async with test_engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
try:
asyncio.run(init_db())
print("[TestServer] Database initialized")
except Exception as e:
print(f"[TestServer] Database init failed: {e}")
raise
from collections.abc import Awaitable, Callable
from fastapi import Request, Response
from fastapi.middleware.cors import CORSMiddleware
from fastapi.staticfiles import StaticFiles
from app.infrastructure.di.providers import (
DatabaseProvider,
RepositoryProvider,
TransactionManagerProvider,
UseCaseProvider,
)
app = FastAPI(
title=test_settings.app.name,
debug=test_settings.app.debug,
docs_url="/docs",
redoc_url="/redoc",
)
container = make_async_container(
DatabaseProvider(),
RepositoryProvider(),
TransactionManagerProvider(),
UseCaseProvider(),
FakeKeycloakProvider(fake_keycloak),
)
setup_dishka(container, app)
from app.infrastructure import register_exception_handlers
register_exception_handlers(app)
register_error_handlers(app)
@app.middleware("http")
async def flash_middleware(
request: Request, call_next: Callable[[Request], Awaitable[Response]]
) -> Response:
await setup_flash_manager(request)
response = await call_next(request)
return response
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
@app.get("/")
async def root_redirect() -> Response:
from fastapi.responses import HTMLResponse
return HTMLResponse(
content='<meta http-equiv="refresh" content="0;url=/web/">', status_code=200
Returns:
Created TestUser instance.
"""
role = (roles or ["user"])[0]
user = TestUser(
id=f"dev-{role}",
username=username,
email=email,
password=password,
roles=roles or ["user"],
)
self._users[username] = user
return user
@app.get("/health")
async def health_check() -> dict[str, str]:
return {"status": "ok", "env": "e2e-test"}
def login(self, username: str, password: str) -> str:
"""Return dev token for the user's role.
# Include main routers
app.include_router(router, prefix="/api")
app.include_router(web_router)
Args:
username: User login name.
password: User password (ignored).
# Add fake auth routes instead of including auth_router
async def fake_login(request: Request, redirect: str = "/web/") -> Response:
from fastapi.responses import HTMLResponse
Returns:
Dev authentication token string.
html = f'''
<!DOCTYPE html>
<html>
<head><title>Test Login</title></head>
<body>
<h1>Test Login Page</h1>
<form method="post" action="/auth/callback">
<input type="hidden" name="redirect" value="{redirect}">
<select name="username">
<option value="testuser">Test User</option>
<option value="admin">Admin User</option>
</select>
<button type="submit">Login</button>
</form>
</body>
</html>
'''
return HTMLResponse(content=html)
Raises:
ValueError: If user does not exist.
"""
user = self._users.get(username)
if not user:
raise ValueError("User not found")
role = user.roles[0] if user.roles else "user"
return f"dev-token-{role}"
@app.post("/auth/callback")
async def fake_callback(request: Request) -> Response:
from fastapi.responses import RedirectResponse
def build_auth_cookie(self, token: str, domain: str) -> dict[str, Any]:
"""Build access_token cookie for blog dev auth.
form = await request.form()
username = form.get("username", "testuser")
redirect = form.get("redirect", "/web/")
Args:
token: Dev authentication token.
domain: Cookie domain.
try:
user = fake_keycloak.create_user(
username=username,
password="test",
email=f"{username}@test.com",
roles=["admin" if username == "admin" else "user"],
)
except ValueError:
_ = fake_keycloak._users.get(username)
token = fake_keycloak.login(username, "test")
response = RedirectResponse(url=redirect, status_code=302)
response.set_cookie(
key="access_token",
value=token,
httponly=True,
secure=False,
samesite="lax",
max_age=3600,
)
return response
@app.get("/auth/logout")
async def fake_logout(request: Request) -> Response:
from fastapi.responses import RedirectResponse
response = RedirectResponse(url="/web/", status_code=302)
response.delete_cookie(key="access_token")
return response
app.mount("/static", StaticFiles(directory="static"), name="static")
import uvicorn
from uvicorn import Config
config = Config(app=app, host="127.0.0.1", port=port, log_level="warning")
server = uvicorn.Server(config)
# Run server in a separate thread with error handling
server_exception = None
def run_server():
nonlocal server_exception
try:
asyncio.run(server.serve())
except Exception as e:
server_exception = e
print(f"[TestServer] Server error: {e}")
import traceback
traceback.print_exc()
server_thread = threading.Thread(target=run_server, daemon=True)
server_thread.start()
print("[TestServer] Server thread started")
# Wait for server to be ready using sync httpx
server_started = False
last_error = None
for attempt in range(50):
try:
response = httpx.get(f"{base_url}/health", timeout=0.5)
if response.status_code == 200:
server_started = True
print(f"[TestServer] Server ready after {attempt + 1} attempts")
break
except (httpx.ConnectError, httpx.TimeoutException) as e:
last_error = e
time.sleep(0.1)
if not server_started:
print(f"[TestServer] Server failed to start after 50 attempts. Last error: {last_error}")
server.should_exit = True
raise RuntimeError(f"Test server failed to start after 50 attempts on port {port}")
# Test that web routes work before yielding
try:
test_response = httpx.get(f"{base_url}/web/", timeout=2.0, follow_redirects=True)
print(f"[TestServer] Test /web/: {test_response.status_code}")
except Exception as e:
print(f"[TestServer] Test /web/ failed: {e}")
# Test auth redirect
try:
test_response = httpx.get(f"{base_url}/web/posts/new", timeout=2.0, follow_redirects=True)
print(f"[TestServer] Test /web/posts/new: {test_response.status_code}")
print(f"[TestServer] Final URL: {test_response.url}")
except Exception as e:
print(f"[TestServer] Test /web/posts/new failed: {e}")
print(f"[TestServer] Yielding server info: {base_url}")
yield {"base_url": base_url, "port": port, "fake_keycloak": fake_keycloak}
# Cleanup
print("[TestServer] Shutting down server...")
server.should_exit = True
server_thread.join(timeout=5.0)
print("[TestServer] Server shutdown complete")
Returns:
Cookie dict compatible with Playwright.
"""
return {
"name": "access_token",
"value": token,
"domain": domain,
"path": "/",
"httpOnly": True,
"secure": False,
"sameSite": "Lax",
}
@pytest.fixture(scope="session")
def base_url(test_server: dict[str, Any]) -> str:
"""Get base URL of running test server."""
return test_server["base_url"]
def base_url() -> str:
"""Return the base URL for the blog application.
Returns:
Application base URL.
"""
return "http://127.0.0.1:8000"
@pytest.fixture(scope="session")
def keycloak_client(test_server: dict[str, Any]) -> FakeKeycloakClient:
"""Get fake Keycloak client from test server."""
return test_server["fake_keycloak"]
def pytfm_auth_provider() -> DevAuthProvider:
"""Return DevAuthProvider for blog dev mode.
Returns:
DevAuthProvider instance.
"""
return DevAuthProvider()
@pytest.fixture
def test_user_data() -> dict[str, str]:
"""Generate test user data."""
import uuid
unique_id = uuid.uuid4().hex[:8]
return {
"username": f"testuser_{unique_id}",
"email": f"test_{unique_id}@example.com",
"password": "TestPass123!",
}
@pytest.fixture
def authenticated_context(
def _create_authenticated_context(
browser: Browser,
keycloak_client: FakeKeycloakClient,
test_user_data: dict[str, str],
base_url: str,
pytfm_auth_provider: DevAuthProvider,
role: str,
) -> BrowserContext:
"""Create authenticated browser context with logged-in user."""
user = keycloak_client.create_user(
username=test_user_data["username"],
password=test_user_data["password"],
email=test_user_data["email"],
roles=["user"],
)
"""Create a browser context authenticated with a dev token role.
token = keycloak_client.login(user.username, user.password)
Args:
browser: Playwright Browser instance.
base_url: Application base URL.
pytfm_auth_provider: Dev auth provider.
role: Dev role (user, user2, admin, guest).
Returns:
Authenticated BrowserContext.
"""
user = pytfm_auth_provider.create_user(
username=f"e2e_{role}",
password="pass",
email=f"{role}@example.com",
roles=[role],
)
token = pytfm_auth_provider.login(user.username, user.password)
context = browser.new_context(
viewport={"width": 1280, "height": 720},
)
cookie_domain = base_url.replace("http://", "").replace("https://", "").split(":")[0]
cookie = pytfm_auth_provider.build_auth_cookie(token, cookie_domain)
context.add_cookies([cookie])
context.add_cookies(
[
{
"name": "access_token",
"value": token,
"domain": cookie_domain,
"path": "/",
"httpOnly": True,
"secure": False,
}
]
)
return context
@pytest.fixture
def user_context(
browser: Browser,
base_url: str,
pytfm_auth_provider: DevAuthProvider,
) -> Generator[BrowserContext, None, None]:
"""Create a browser context authenticated as a regular user.
Args:
browser: Playwright Browser instance.
base_url: Application base URL.
pytfm_auth_provider: Dev auth provider.
Yields:
Authenticated BrowserContext for user role.
"""
context = _create_authenticated_context(browser, base_url, pytfm_auth_provider, "user")
yield context
context.close()
@pytest.fixture
def authenticated_page(authenticated_context: BrowserContext):
"""Create authenticated page for testing."""
page = authenticated_context.new_page()
def user_page(user_context: BrowserContext) -> Generator[Page, None, None]:
"""Create a page authenticated as a regular user.
Args:
user_context: Authenticated browser context.
Yields:
Authenticated Playwright Page.
"""
page = user_context.new_page()
yield page
page.close()
@pytest.fixture
def admin_user(keycloak_client: FakeKeycloakClient) -> dict[str, str]:
"""Create admin user and return credentials with token."""
import uuid
def admin_context(
browser: Browser,
base_url: str,
pytfm_auth_provider: DevAuthProvider,
) -> Generator[BrowserContext, None, None]:
"""Create a browser context authenticated as admin.
unique_id = uuid.uuid4().hex[:8]
username = f"admin_{unique_id}"
password = "AdminPass123!"
Args:
browser: Playwright Browser instance.
base_url: Application base URL.
pytfm_auth_provider: Dev auth provider.
user = keycloak_client.create_user(
username=username,
password=password,
email=f"admin_{unique_id}@example.com",
roles=["user", "admin"],
)
token = keycloak_client.login(username, password)
return {
"id": user.id,
"username": username,
"password": password,
"email": user.email,
"token": token,
"roles": user.roles,
}
Yields:
Authenticated BrowserContext for admin role.
"""
context = _create_authenticated_context(browser, base_url, pytfm_auth_provider, "admin")
yield context
context.close()
@pytest.fixture
def regular_user(keycloak_client: FakeKeycloakClient) -> dict[str, str]:
"""Create regular user and return credentials with token."""
import uuid
def admin_page(admin_context: BrowserContext) -> Generator[Page, None, None]:
"""Create a page authenticated as admin.
unique_id = uuid.uuid4().hex[:8]
username = f"user_{unique_id}"
password = "UserPass123!"
Args:
admin_context: Authenticated browser context.
user = keycloak_client.create_user(
username=username,
password=password,
email=f"user_{unique_id}@example.com",
roles=["user"],
Yields:
Authenticated Playwright Page.
"""
page = admin_context.new_page()
yield page
page.close()
@pytest.fixture
def user2_context(
browser: Browser,
base_url: str,
pytfm_auth_provider: DevAuthProvider,
) -> Generator[BrowserContext, None, None]:
"""Create a browser context authenticated as a second regular user.
Args:
browser: Playwright Browser instance.
base_url: Application base URL.
pytfm_auth_provider: Dev auth provider.
Yields:
Authenticated BrowserContext for user2 role.
"""
context = _create_authenticated_context(browser, base_url, pytfm_auth_provider, "user2")
yield context
context.close()
@pytest.fixture
def user2_page(user2_context: BrowserContext) -> Generator[Page, None, None]:
"""Create a page authenticated as a second regular user.
Args:
user2_context: Authenticated browser context.
Yields:
Authenticated Playwright Page.
"""
page = user2_context.new_page()
yield page
page.close()
@pytest.fixture
def guest_context(
browser: Browser,
base_url: str,
) -> Generator[BrowserContext, None, None]:
"""Create an unauthenticated browser context.
Args:
browser: Playwright Browser instance.
base_url: Application base URL.
Yields:
Unauthenticated BrowserContext.
"""
context = browser.new_context(
viewport={"width": 1280, "height": 720},
)
yield context
context.close()
token = keycloak_client.login(username, password)
return {
"id": user.id,
"username": username,
"password": password,
"email": user.email,
"token": token,
"roles": user.roles,
}
@pytest.fixture
def guest_page(guest_context: BrowserContext) -> Generator[Page, None, None]:
"""Create an unauthenticated page.
Args:
guest_context: Unauthenticated browser context.
Yields:
Unauthenticated Playwright Page.
"""
page = guest_context.new_page()
yield page
page.close()