Enhance database client with upsert functionality for users and posts…

.gitignore

@@ -141,3 +141,7 @@ cython_debug/
 *.pb.go
 *.pb.py
 
+# Allure reports
+reports/allure-results/
+reports/allure-report/
+

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"
+

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."""

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

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"<APILogDB(id={self.id}, {self.method} {self.url} -> {self.status_code})>"
@@ -94,3 +94,4 @@ def __repr__(self) -> str:
 
 
 
+

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)