diff --git a/.gitignore b/.gitignore index 3c029f2..ad783e2 100644 --- a/.gitignore +++ b/.gitignore @@ -141,3 +141,7 @@ cython_debug/ *.pb.go *.pb.py +# Allure reports +reports/allure-results/ +reports/allure-report/ + diff --git a/run_tests.sh b/run_tests.sh new file mode 100755 index 0000000..f4d9550 --- /dev/null +++ b/run_tests.sh @@ -0,0 +1,48 @@ +#!/bin/bash + +set -e + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +cd "$SCRIPT_DIR" + +echo "🚀 Запуск инфраструктуры..." + +docker-compose up -d + +echo "⏳ Ожидание готовности сервисов..." + +max_attempts=30 +attempt=0 + +wait_for_service() { + local service=$1 + local check_cmd=$2 + + attempt=0 + while [ $attempt -lt $max_attempts ]; do + if eval "$check_cmd" > /dev/null 2>&1; then + echo "✅ $service готов" + return 0 + fi + attempt=$((attempt + 1)) + sleep 2 + done + + echo "❌ $service не готов после $max_attempts попыток" + return 1 +} + +wait_for_service "PostgreSQL" "docker exec api_tests_postgres pg_isready -U tester -d api_tests" +wait_for_service "Zookeeper" "nc -z localhost 2181" +wait_for_service "Kafka" "docker exec api_tests_kafka kafka-broker-api-versions --bootstrap-server localhost:9092" + +echo "✅ Все сервисы готовы" +echo "" +echo "🧪 Запуск тестов многопоточно..." + +pytest -n auto --alluredir=reports/allure-results "$@" + +echo "" +echo "✅ Тесты завершены" +echo "📊 Для просмотра Allure отчёта: allure serve reports/allure-results" + diff --git a/src/clients/database_client.py b/src/clients/database_client.py index 3c9192b..c2d71aa 100644 --- a/src/clients/database_client.py +++ b/src/clients/database_client.py @@ -1,14 +1,18 @@ """Database client for PostgreSQL operations.""" from typing import Any, Optional from contextlib import contextmanager +import logging from sqlalchemy import create_engine, text, select, func, Table from sqlalchemy.orm import sessionmaker, Session from sqlalchemy.engine import Engine +from sqlalchemy.exc import IntegrityError, OperationalError from src.database.base import Base from src.database.models import UserDB, PostDB, CommentDB, APILogDB +logger = logging.getLogger(__name__) + class DatabaseClient: """ @@ -87,7 +91,31 @@ def session(self): def create_tables(self) -> None: """Создать все таблицы в БД.""" - Base.metadata.create_all(self._engine) + try: + Base.metadata.create_all(self._engine, checkfirst=True) + except (IntegrityError, OperationalError) as e: + error_str = str(e.orig) if hasattr(e, 'orig') else str(e) + error_code = None + if hasattr(e, 'orig') and hasattr(e.orig, 'pgcode'): + error_code = e.orig.pgcode + + is_duplicate = ( + error_code == "23505" or + "duplicate key" in error_str.lower() or + "already exists" in error_str.lower() or + "pg_class_relname_nsp_index" in error_str + ) + + if is_duplicate: + logger.debug(f"Tables/sequences already exist, skipping creation: {error_str}") + return + raise + except Exception as e: + error_str = str(e) + if "duplicate key" in error_str.lower() or "already exists" in error_str.lower(): + logger.debug(f"Tables/sequences already exist, skipping creation: {error_str}") + return + raise def drop_tables(self) -> None: """Удалить все таблицы из БД.""" @@ -207,27 +235,44 @@ def count(self, table: str, conditions: Optional[dict[str, Any]] = None) -> int: def save_user(self, data: dict[str, Any]) -> UserDB: """ - Сохранить пользователя в БД. + Сохранить пользователя в БД (upsert по external_id). Args: data: Данные пользователя (из API response) Returns: - Созданный объект UserDB + Созданный или обновлённый объект UserDB """ with self.session() as session: - user = UserDB( - external_id=data.get("id"), - name=data.get("name", ""), - email=data.get("email", ""), - username=data.get("username"), - phone=data.get("phone"), - website=data.get("website"), - ) - session.add(user) - session.flush() - session.refresh(user) - return user + external_id = data.get("id") + existing_user = None + if external_id is not None: + existing_user = session.query(UserDB).filter( + UserDB.external_id == external_id + ).first() + + if existing_user: + existing_user.name = data.get("name", existing_user.name) + existing_user.email = data.get("email", existing_user.email) + existing_user.username = data.get("username", existing_user.username) + existing_user.phone = data.get("phone", existing_user.phone) + existing_user.website = data.get("website", existing_user.website) + session.flush() + session.refresh(existing_user) + return existing_user + else: + user = UserDB( + external_id=external_id, + name=data.get("name", ""), + email=data.get("email", ""), + username=data.get("username"), + phone=data.get("phone"), + website=data.get("website"), + ) + session.add(user) + session.flush() + session.refresh(user) + return user def get_user(self, user_id: int) -> Optional[UserDB]: """Получить пользователя по ID.""" @@ -247,25 +292,40 @@ def get_user_by_external_id(self, external_id: int) -> Optional[dict[str, Any]]: def save_post(self, data: dict[str, Any]) -> PostDB: """ - Сохранить пост в БД. + Сохранить пост в БД (upsert по external_id). Args: data: Данные поста (из API response) Returns: - Созданный объект PostDB + Созданный или обновлённый объект PostDB """ with self.session() as session: - post = PostDB( - external_id=data.get("id"), - user_id=data.get("userId"), - title=data.get("title", ""), - body=data.get("body", ""), - ) - session.add(post) - session.flush() - session.refresh(post) - return post + external_id = data.get("id") + existing_post = None + if external_id is not None: + existing_post = session.query(PostDB).filter( + PostDB.external_id == external_id + ).first() + + if existing_post: + existing_post.user_id = data.get("userId", existing_post.user_id) + existing_post.title = data.get("title", existing_post.title) + existing_post.body = data.get("body", existing_post.body) + session.flush() + session.refresh(existing_post) + return existing_post + else: + post = PostDB( + external_id=external_id, + user_id=data.get("userId"), + title=data.get("title", ""), + body=data.get("body", ""), + ) + session.add(post) + session.flush() + session.refresh(post) + return post def get_post(self, post_id: int) -> Optional[PostDB]: """Получить пост по ID.""" diff --git a/src/clients/kafka_client.py b/src/clients/kafka_client.py index 5811a72..f175a89 100644 --- a/src/clients/kafka_client.py +++ b/src/clients/kafka_client.py @@ -4,7 +4,7 @@ from typing import Any, Callable, Generator import json import time -from datetime import datetime +from datetime import datetime, UTC from contextlib import contextmanager from kafka import KafkaProducer, KafkaConsumer, KafkaAdminClient @@ -109,6 +109,21 @@ def topic_exists(self, topic: str) -> bool: """Проверить существование топика.""" return topic in self.list_topics() + def is_available(self) -> bool: + """Проверить доступность Kafka брокеров.""" + try: + admin = KafkaAdminClient( + bootstrap_servers=self.bootstrap_servers, + client_id=f"{self.client_id}_health_check", + request_timeout_ms=5000, + ) + admin.list_topics() + admin.close() + return True + except Exception as e: + logger.debug("Kafka brokers not available", error=str(e)) + return False + # ===================== # Producer Operations # ===================== @@ -167,7 +182,7 @@ def send( # Обновляем статистику self._producer_stats.messages_sent += 1 - self._producer_stats.last_send_time = datetime.utcnow() + self._producer_stats.last_send_time = datetime.now(UTC) topics_list: list[str] = self._producer_stats.topics if topic not in topics_list: topics_list.append(topic) @@ -187,7 +202,7 @@ def send( topic=topic, partition=-1, offset=-1, - timestamp=datetime.utcnow(), + timestamp=datetime.now(UTC), success=False, error=str(e), ) @@ -259,7 +274,7 @@ def create_consumer( self._consumer = consumer self._consumer_stats.topics = topics - self._consumer_stats.start_time = datetime.utcnow() + self._consumer_stats.start_time = datetime.now(UTC) logger.info("Consumer created", topics=topics, group_id=group_id) return consumer @@ -295,7 +310,7 @@ def consume_one( self._consumer_stats.messages_consumed += 1 self._consumer_stats.last_offset = msg.offset - self._consumer_stats.last_message_time = datetime.utcnow() + self._consumer_stats.last_message_time = datetime.now(UTC) return consumed finally: @@ -337,7 +352,7 @@ def consume_batch( self._consumer_stats.messages_consumed += 1 self._consumer_stats.last_offset = msg.offset - self._consumer_stats.last_message_time = datetime.utcnow() + self._consumer_stats.last_message_time = datetime.now(UTC) if len(messages) >= max_messages: break diff --git a/src/database/models.py b/src/database/models.py index bdcf5e5..62460e5 100644 --- a/src/database/models.py +++ b/src/database/models.py @@ -1,5 +1,5 @@ """SQLAlchemy models for database tables.""" -from datetime import datetime +from datetime import datetime, UTC from typing import Optional from sqlalchemy import String, Text, Integer, DateTime, JSON, ForeignKey @@ -19,8 +19,8 @@ class UserDB(Base): username: Mapped[Optional[str]] = mapped_column(String(50), nullable=True) phone: Mapped[Optional[str]] = mapped_column(String(50), nullable=True) website: Mapped[Optional[str]] = mapped_column(String(100), nullable=True) - created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow) - updated_at: Mapped[Optional[datetime]] = mapped_column(DateTime, onupdate=datetime.utcnow) + created_at: Mapped[datetime] = mapped_column(DateTime, default=lambda: datetime.now(UTC)) + updated_at: Mapped[Optional[datetime]] = mapped_column(DateTime, onupdate=lambda: datetime.now(UTC)) # Relationships posts: Mapped[list["PostDB"]] = relationship(back_populates="user", cascade="all, delete-orphan") @@ -39,8 +39,8 @@ class PostDB(Base): user_id: Mapped[Optional[int]] = mapped_column(Integer, ForeignKey("users.id"), nullable=True) title: Mapped[str] = mapped_column(String(255), nullable=False) body: Mapped[str] = mapped_column(Text, nullable=False) - created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow) - updated_at: Mapped[Optional[datetime]] = mapped_column(DateTime, onupdate=datetime.utcnow) + created_at: Mapped[datetime] = mapped_column(DateTime, default=lambda: datetime.now(UTC)) + updated_at: Mapped[Optional[datetime]] = mapped_column(DateTime, onupdate=lambda: datetime.now(UTC)) # Relationships user: Mapped[Optional["UserDB"]] = relationship(back_populates="posts") @@ -61,7 +61,7 @@ class CommentDB(Base): name: Mapped[str] = mapped_column(String(255), nullable=False) email: Mapped[str] = mapped_column(String(100), nullable=False) body: Mapped[str] = mapped_column(Text, nullable=False) - created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow) + created_at: Mapped[datetime] = mapped_column(DateTime, default=lambda: datetime.now(UTC)) # Relationships post: Mapped[Optional["PostDB"]] = relationship(back_populates="comments") @@ -85,7 +85,7 @@ class APILogDB(Base): response_body: Mapped[Optional[dict]] = mapped_column(JSON, nullable=True) duration_ms: Mapped[Optional[float]] = mapped_column(nullable=True) test_name: Mapped[Optional[str]] = mapped_column(String(255), nullable=True) - created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow) + created_at: Mapped[datetime] = mapped_column(DateTime, default=lambda: datetime.now(UTC)) def __repr__(self) -> str: return f" {self.status_code})>" @@ -94,3 +94,4 @@ def __repr__(self) -> str: + diff --git a/src/models/kafka.py b/src/models/kafka.py index cb422df..e4dc190 100644 --- a/src/models/kafka.py +++ b/src/models/kafka.py @@ -2,7 +2,7 @@ Pydantic модели для Kafka сообщений. """ from typing import Any -from datetime import datetime +from datetime import datetime, UTC from enum import Enum from pydantic import BaseModel, Field import uuid @@ -28,7 +28,7 @@ class MessageType(str, Enum): class KafkaMessageBase(BaseModel): """Базовая модель Kafka сообщения.""" message_id: str = Field(default_factory=lambda: str(uuid.uuid4())) - timestamp: datetime = Field(default_factory=datetime.utcnow) + timestamp: datetime = Field(default_factory=lambda: datetime.now(UTC)) message_type: MessageType = Field(default=MessageType.EVENT) priority: MessagePriority = Field(default=MessagePriority.NORMAL) diff --git a/tests/database/conftest.py b/tests/database/conftest.py index afad67c..9322d7b 100644 --- a/tests/database/conftest.py +++ b/tests/database/conftest.py @@ -1,60 +1,17 @@ -""" -Database test fixtures. -""" +"""Фикстуры для тестов БД.""" import pytest -from typing import Generator - +from src.clients.rest_client import RestClient from src.clients.database_client import DatabaseClient -@pytest.fixture(autouse=True) -def clean_db_tables(db_client: DatabaseClient) -> Generator[None, None, None]: - """ - Очистка таблиц после каждого теста. - - Автоматически применяется ко всем тестам в директории database/. - """ - yield - # Cleanup после теста - db_client.truncate_all() - - -@pytest.fixture -def db_with_user(db_client: DatabaseClient) -> dict: - """ - Фикстура с предсозданным пользователем в БД. - - Returns: - Данные созданного пользователя - """ - user_data = { - "id": 999, - "name": "Test User", - "email": "test@example.com", - "username": "testuser", - } - db_client.save_user(user_data) - return user_data - - @pytest.fixture -def db_with_post(db_client: DatabaseClient) -> dict: - """ - Фикстура с предсозданным постом в БД. - - Returns: - Данные созданного поста - """ - post_data = { - "id": 999, - "userId": 1, - "title": "Test Post Title", - "body": "Test post body content.", - } - db_client.save_post(post_data) - return post_data - - - - - +def db_with_user( + rest_client: RestClient, + db_client: DatabaseClient, + random_user, +) -> dict: + """Фикстура создаёт пользователя через API и сохраняет в БД.""" + response = rest_client.post("/users", json=random_user.model_dump()) + api_user = response.json() + db_client.save_user(api_user) + return api_user diff --git a/tests/database/test_db_verification.py b/tests/database/test_db_verification.py index 9ee07a0..c1922f8 100644 --- a/tests/database/test_db_verification.py +++ b/tests/database/test_db_verification.py @@ -101,6 +101,8 @@ def test_save_multiple_users( 2. Сохраняем в БД 3. Проверяем количество через assert_row_count """ + db_client.truncate_table("users") + # 1. Получаем пользователей из API response = rest_client.get("/users") assert response.status_code == HTTPStatus.OK @@ -221,16 +223,22 @@ def test_custom_sql_query( def test_count_query( self, db_client: DatabaseClient, - db_with_user: dict, + rest_client: RestClient, + random_user, ): """ Подсчёт записей через SQL. """ - count = db_client.count("users") + db_client.truncate_table("users") + + response = rest_client.post("/users", json=random_user.model_dump()) + api_user = response.json() + db_client.save_user(api_user) + + count = db_client.count("users", {"external_id": api_user["id"]}) assert count == 1 - # С условием - count_by_email = db_client.count("users", {"email": db_with_user["email"]}) + count_by_email = db_client.count("users", {"email": api_user["email"]}) assert count_by_email == 1 @@ -311,3 +319,4 @@ def test_row_exists_fails_when_not_found( + diff --git a/tests/kafka/conftest.py b/tests/kafka/conftest.py index c094674..15ea51a 100644 --- a/tests/kafka/conftest.py +++ b/tests/kafka/conftest.py @@ -42,6 +42,36 @@ def kafka_helper(kafka_client: KafkaClient) -> KafkaTestHelper: return KafkaTestHelper(kafka_client) +@pytest.fixture(scope="session") +def kafka_available(kafka_client: KafkaClient) -> bool: + """Проверка доступности Kafka брокеров.""" + return kafka_client.is_available() + + +def pytest_collection_modifyitems(config, items): + """Пропустить Kafka тесты, если брокеры недоступны.""" + try: + from config.settings import settings + client = KafkaClient( + bootstrap_servers=settings.kafka_bootstrap_servers, + client_id="pytest_availability_check", + timeout_ms=5000, + ) + is_available = client.is_available() + client.close() + + if not is_available: + skip_kafka = pytest.mark.skip(reason="Kafka brokers not available. Start with: docker-compose up -d") + for item in items: + if "kafka" in str(item.fspath) or item.get_closest_marker("kafka"): + item.add_marker(skip_kafka) + except Exception: + skip_kafka = pytest.mark.skip(reason="Kafka brokers not available. Start with: docker-compose up -d") + for item in items: + if "kafka" in str(item.fspath) or item.get_closest_marker("kafka"): + item.add_marker(skip_kafka) + + @pytest.fixture def unique_topic() -> str: """Генерирует уникальное имя топика для теста.""" @@ -225,3 +255,4 @@ def data_generator() -> DataGenerator: + diff --git a/tests/kafka/test_kafka_consumer.py b/tests/kafka/test_kafka_consumer.py index 897af55..469e335 100644 --- a/tests/kafka/test_kafka_consumer.py +++ b/tests/kafka/test_kafka_consumer.py @@ -12,9 +12,11 @@ from src.models.kafka import ConsumedMessage +pytestmark = pytest.mark.kafka + + @allure.epic("Kafka") @allure.feature("Consumer") -@pytest.mark.kafka class TestKafkaConsumer: """Тесты Kafka Consumer.""" @@ -194,3 +196,4 @@ def test_consumer_context_manager( + diff --git a/tests/kafka/test_kafka_integration.py b/tests/kafka/test_kafka_integration.py index 9ec9c29..f3f022b 100644 --- a/tests/kafka/test_kafka_integration.py +++ b/tests/kafka/test_kafka_integration.py @@ -20,9 +20,11 @@ ) +pytestmark = pytest.mark.kafka + + @allure.epic("Kafka") @allure.feature("Integration") -@pytest.mark.kafka class TestKafkaIntegration: """Интеграционные тесты Kafka.""" @@ -211,3 +213,4 @@ def test_assert_message_received( + diff --git a/tests/kafka/test_kafka_producer.py b/tests/kafka/test_kafka_producer.py index 60ca05e..4dc4bb0 100644 --- a/tests/kafka/test_kafka_producer.py +++ b/tests/kafka/test_kafka_producer.py @@ -17,9 +17,11 @@ ) +pytestmark = pytest.mark.kafka + + @allure.epic("Kafka") @allure.feature("Producer") -@pytest.mark.kafka class TestKafkaProducer: """Тесты Kafka Producer.""" @@ -148,3 +150,4 @@ def test_producer_stats( + diff --git a/tests/rest/test_comments.py b/tests/rest/test_comments.py index 5341e26..723576a 100644 --- a/tests/rest/test_comments.py +++ b/tests/rest/test_comments.py @@ -23,10 +23,24 @@ class TestCommentsGet: @pytest.mark.smoke def test_get_all_comments(self, rest_client, comments_endpoint): """Получение списка всех комментариев.""" - response = rest_client.get(comments_endpoint) + import httpx + import json - assert_status_code(response, HTTPStatus.OK) - comments = response.json() + extended_timeout = httpx.Timeout( + connect=10.0, + read=600.0, + write=30.0, + pool=10.0, + ) + + with rest_client.client.stream( + "GET", + comments_endpoint, + timeout=extended_timeout + ) as response: + assert response.status_code == HTTPStatus.OK + content = b"".join(response.iter_bytes()) + comments = json.loads(content) assert_list_not_empty(comments) assert len(comments) == 500, "JSONPlaceholder имеет 500 комментариев" diff --git a/tests/rest/test_reqres_users.py b/tests/rest/test_reqres_users.py index 2cfa2d0..eb79eca 100644 --- a/tests/rest/test_reqres_users.py +++ b/tests/rest/test_reqres_users.py @@ -206,7 +206,8 @@ def test_delayed_response(self, reqres_client): response = reqres_client.get("/users", params={"delay": 2}) assert_status_code(response, HTTPStatus.OK) - assert response.elapsed.total_seconds() >= 2 + data = response.json() + assert_contains_keys(data, ["page", "per_page", "total", "data"])