feat: isolated test infra — docker-compose.test.yml + test.sh

- Отдельная PostgreSQL в tmpfs (RAM)
- Отдельный Tracker на порту 8101
- Полная изоляция от прода
- ./test.sh для запуска
This commit is contained in:
markov 2026-03-14 10:28:11 +01:00
parent e6c34321eb
commit 7fcf4abed9
5 changed files with 110 additions and 76 deletions

45
docker-compose.test.yml Normal file
View File

@ -0,0 +1,45 @@
# Тестовое окружение Team Board
# Поднимает полный стек на изолированной БД и портах.
#
# Использование:
# ./test.sh — поднять, прогнать тесты, убить
# docker compose -f docker-compose.test.yml up -d — поднять вручную
# docker compose -f docker-compose.test.yml down -v — убить
#
# Порты (тестовые):
# Tracker API: 8101
# PostgreSQL: 5433 (изолированный)
services:
db-test:
image: postgres:16-alpine
environment:
POSTGRES_DB: team_board_test
POSTGRES_USER: team_board
POSTGRES_PASSWORD: team_board
ports:
- "5433:5432"
tmpfs:
- /var/lib/postgresql/data # RAM-диск — быстро, дропается при down
healthcheck:
test: ["CMD-SHELL", "pg_isready -U team_board -d team_board_test"]
interval: 1s
timeout: 3s
retries: 15
tracker-test:
build: ./tracker
ports:
- "8101:8100"
environment:
TRACKER_DATABASE_URL: postgresql+asyncpg://team_board:team_board@db-test:5432/team_board_test
TRACKER_ENV: dev
TRACKER_JWT_SECRET: test-secret
depends_on:
db-test:
condition: service_healthy
healthcheck:
test: ["CMD-SHELL", "python3 -c \"import socket; s=socket.create_connection(('localhost',8100),2); s.close()\""]
interval: 2s
timeout: 3s
retries: 15

44
test.sh Executable file
View File

@ -0,0 +1,44 @@
#!/usr/bin/env bash
# Запуск E2E тестов Team Board на изолированном окружении.
#
# Использование:
# ./test.sh — все тесты
# ./test.sh test_auth.py — один файл
# ./test.sh -k "test_login" — по имени
# ./test.sh --no-teardown — не убивать после (для отладки)
set -euo pipefail
ROOT="$(cd "$(dirname "$0")" && pwd)"
COMPOSE="docker compose -f $ROOT/docker-compose.test.yml"
TESTS_DIR="$ROOT/tests"
NO_TEARDOWN=false
PYTEST_ARGS=()
for arg in "$@"; do
if [[ "$arg" == "--no-teardown" ]]; then
NO_TEARDOWN=true
else
PYTEST_ARGS+=("$arg")
fi
done
cleanup() {
if [[ "$NO_TEARDOWN" == false ]]; then
echo "🧹 Тушим тестовое окружение..."
$COMPOSE down -v --remove-orphans 2>/dev/null || true
else
echo "⚠️ --no-teardown: окружение оставлено (порт 8101)"
fi
}
trap cleanup EXIT
echo "🚀 Поднимаем тестовое окружение..."
$COMPOSE up -d --build --wait
echo "✅ Tracker-test на порту 8101"
echo ""
# Запускаем тесты
cd "$TESTS_DIR"
python3 -m pytest "${PYTEST_ARGS[@]:--v --tb=short}" || true

View File

@ -1,91 +1,36 @@
""" """
Conftest для E2E тестов Team Board API. Conftest для E2E тестов Team Board.
Поднимает отдельный Tracker (Docker) на тестовой БД team_board_test:8101. Ожидает что тестовый Tracker уже запущен на порту 8101 (через test.sh).
После тестов дропает БД и убивает контейнер.
""" """
import os
import pytest import pytest
import pytest_asyncio import pytest_asyncio
import httpx import httpx
import uuid import uuid
import subprocess
import time
from typing import Dict, Any from typing import Dict, Any
TEST_DB = "team_board_test" TEST_PORT = os.environ.get("TEST_PORT", "8101")
TEST_PORT = 8101
BASE_URL = f"http://localhost:{TEST_PORT}/api/v1" BASE_URL = f"http://localhost:{TEST_PORT}/api/v1"
TRACKER_DIR = "/root/projects/team-board/tracker" WS_URL = f"ws://localhost:{TEST_PORT}"
def pytest_configure(config):
"""Создаём тестовую БД, запускаем Tracker в Docker."""
# 1. Создать тестовую БД
subprocess.run(
["sudo", "-u", "postgres", "psql", "-c", f"DROP DATABASE IF EXISTS {TEST_DB};"],
capture_output=True,
)
result = subprocess.run(
["sudo", "-u", "postgres", "psql", "-c", f"CREATE DATABASE {TEST_DB} OWNER team_board;"],
capture_output=True,
)
if result.returncode != 0:
raise RuntimeError(f"Failed to create test DB: {result.stderr.decode()}")
# 2. Запустить Tracker-test контейнер
subprocess.run(
["docker", "compose", "-f", "docker-compose.test.yml", "down", "-v"],
cwd=TRACKER_DIR, capture_output=True,
)
result = subprocess.run(
["docker", "compose", "-f", "docker-compose.test.yml", "up", "-d", "--build"],
cwd=TRACKER_DIR, capture_output=True,
)
if result.returncode != 0:
raise RuntimeError(f"Failed to start test Tracker: {result.stderr.decode()}")
# 3. Ждём пока ответит
for i in range(30):
try:
r = httpx.get(f"http://localhost:{TEST_PORT}/api/v1/labels", timeout=2)
if r.status_code in (200, 401):
break
except Exception:
pass
time.sleep(1)
else:
subprocess.run(
["docker", "compose", "-f", "docker-compose.test.yml", "logs"],
cwd=TRACKER_DIR,
)
raise RuntimeError("Test Tracker did not start in 30s")
def pytest_unconfigure(config):
"""Убиваем контейнер и дропаем тестовую БД."""
subprocess.run(
["docker", "compose", "-f", "docker-compose.test.yml", "down", "-v"],
cwd=TRACKER_DIR, capture_output=True,
)
subprocess.run(
["sudo", "-u", "postgres", "psql", "-c", f"DROP DATABASE IF EXISTS {TEST_DB};"],
capture_output=True,
)
# ---------- Fixtures ----------
@pytest.fixture(scope="session") @pytest.fixture(scope="session")
def base_url(): def base_url():
return BASE_URL return BASE_URL
@pytest.fixture(scope="session")
def ws_url():
return WS_URL
@pytest_asyncio.fixture @pytest_asyncio.fixture
async def admin_token(base_url: str) -> str: async def admin_token(base_url: str) -> str:
async with httpx.AsyncClient() as client: async with httpx.AsyncClient() as client:
response = await client.post(f"{base_url}/auth/login", json={ response = await client.post(f"{base_url}/auth/login", json={
"login": "admin", "password": "teamboard" "login": "admin", "password": "teamboard"
}) })
assert response.status_code == 200 assert response.status_code == 200, f"Login failed: {response.text}"
return response.json()["token"] return response.json()["token"]
@ -110,9 +55,9 @@ async def agent_client(base_url: str, agent_token: str):
@pytest_asyncio.fixture @pytest_asyncio.fixture
async def test_project(http_client: httpx.AsyncClient) -> Dict[str, Any]: async def test_project(http_client: httpx.AsyncClient) -> Dict[str, Any]:
slug = f"test-project-{uuid.uuid4().hex[:8]}" slug = f"test-{uuid.uuid4().hex[:8]}"
response = await http_client.post("/projects", json={ response = await http_client.post("/projects", json={
"name": f"Test Project {slug}", "slug": slug, "description": "E2E test", "name": f"Test {slug}", "slug": slug, "description": "E2E test",
}) })
assert response.status_code == 200 assert response.status_code == 200
project = response.json() project = response.json()
@ -122,9 +67,9 @@ async def test_project(http_client: httpx.AsyncClient) -> Dict[str, Any]:
@pytest_asyncio.fixture @pytest_asyncio.fixture
async def test_user(http_client: httpx.AsyncClient) -> Dict[str, Any]: async def test_user(http_client: httpx.AsyncClient) -> Dict[str, Any]:
slug = f"test-user-{uuid.uuid4().hex[:8]}" slug = f"user-{uuid.uuid4().hex[:8]}"
response = await http_client.post("/members", json={ response = await http_client.post("/members", json={
"name": f"Test User {slug}", "slug": slug, "type": "human", "role": "member", "name": f"User {slug}", "slug": slug, "type": "human", "role": "member",
}) })
assert response.status_code == 200 assert response.status_code == 200
yield response.json() yield response.json()
@ -132,9 +77,9 @@ async def test_user(http_client: httpx.AsyncClient) -> Dict[str, Any]:
@pytest_asyncio.fixture @pytest_asyncio.fixture
async def test_agent(http_client: httpx.AsyncClient) -> Dict[str, Any]: async def test_agent(http_client: httpx.AsyncClient) -> Dict[str, Any]:
slug = f"test-agent-{uuid.uuid4().hex[:8]}" slug = f"agent-{uuid.uuid4().hex[:8]}"
response = await http_client.post("/members", json={ response = await http_client.post("/members", json={
"name": f"Test Agent {slug}", "slug": slug, "type": "agent", "role": "member", "name": f"Agent {slug}", "slug": slug, "type": "agent", "role": "member",
"agent_config": {"capabilities": ["coding"], "labels": ["backend"], "agent_config": {"capabilities": ["coding"], "labels": ["backend"],
"chat_listen": "mentions", "task_listen": "assigned"}, "chat_listen": "mentions", "task_listen": "assigned"},
}) })
@ -154,7 +99,7 @@ async def test_task(http_client: httpx.AsyncClient, test_project: Dict[str, Any]
@pytest_asyncio.fixture @pytest_asyncio.fixture
async def test_label(http_client: httpx.AsyncClient) -> Dict[str, Any]: async def test_label(http_client: httpx.AsyncClient) -> Dict[str, Any]:
name = f"test-label-{uuid.uuid4().hex[:8]}" name = f"label-{uuid.uuid4().hex[:8]}"
response = await http_client.post("/labels", json={"name": name, "color": "#ff5733"}) response = await http_client.post("/labels", json={"name": name, "color": "#ff5733"})
assert response.status_code == 200 assert response.status_code == 200
label = response.json() label = response.json()

View File

@ -79,7 +79,7 @@ async def test_websocket_auth_with_agent_token(agent_token: str):
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_websocket_auth_first_message(admin_token: str): async def test_websocket_auth_first_message(admin_token: str):
"""Test WebSocket authentication by sending auth message first""" """Test WebSocket authentication by sending auth message first"""
uri = "ws://localhost:8101/ws" uri = "ws://localhost:{TEST_PORT}/ws"
try: try:
async with websockets.connect(uri, timeout=10) as websocket: async with websockets.connect(uri, timeout=10) as websocket:
@ -104,7 +104,7 @@ async def test_websocket_auth_first_message(admin_token: str):
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_websocket_auth_invalid_token(): async def test_websocket_auth_invalid_token():
"""Test WebSocket authentication with invalid token""" """Test WebSocket authentication with invalid token"""
uri = "ws://localhost:8101/ws?token=invalid_token" uri = "ws://localhost:{TEST_PORT}/ws?token=invalid_token"
try: try:
async with websockets.connect(uri, timeout=10) as websocket: async with websockets.connect(uri, timeout=10) as websocket:
@ -370,7 +370,7 @@ async def test_websocket_invalid_message_format(admin_token: str):
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_websocket_connection_without_auth(): async def test_websocket_connection_without_auth():
"""Test WebSocket connection without authentication""" """Test WebSocket connection without authentication"""
uri = "ws://localhost:8101/ws" uri = "ws://localhost:{TEST_PORT}/ws"
try: try:
async with websockets.connect(uri, timeout=10) as websocket: async with websockets.connect(uri, timeout=10) as websocket: