From ba2beff5aa023b72dedd8e176076ffa6e521ae71 Mon Sep 17 00:00:00 2001 From: Obrams Date: Mon, 22 Dec 2025 00:39:59 +0300 Subject: [PATCH] Enhance database client with upsert functionality for users and posts, improve error handling during table creation, and update timestamp handling to use UTC. Add Allure report paths to .gitignore and refactor test fixtures for better integration with API. --- .gitignore | 4 + run_tests.sh | 48 +++++++++++ src/clients/database_client.py | 114 +++++++++++++++++++------ src/clients/kafka_client.py | 27 ++++-- src/database/models.py | 15 ++-- src/models/kafka.py | 4 +- tests/database/conftest.py | 67 +++------------ tests/database/test_db_verification.py | 17 +++- tests/kafka/conftest.py | 31 +++++++ tests/kafka/test_kafka_consumer.py | 5 +- tests/kafka/test_kafka_integration.py | 5 +- tests/kafka/test_kafka_producer.py | 5 +- tests/rest/test_comments.py | 20 ++++- tests/rest/test_reqres_users.py | 3 +- 14 files changed, 257 insertions(+), 108 deletions(-) create mode 100755 run_tests.sh 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"])