chat/tests/test_standalone.py

#!/usr/bin/env python

import json
import sys
import asyncio
import uvicorn
from fastapi import FastAPI, Depends, HTTPException
from datetime import date
from typing import Dict, Optional, List
from pydantic import BaseModel
from sqlalchemy.ext.asyncio import AsyncSession

# Импортируем необходимые модули из календарного сервиса
from services.calendar_service.schemas import (
    EntryType, FlowIntensity, MoodType, CalendarEntryCreate, CalendarEntryResponse
)
from services.calendar_service.mobile_endpoint import MobileCalendarEntryCreate
from shared.database import get_db

# Создаем тестовое приложение
app = FastAPI(title="Test Mobile Endpoint")

@app.get("/health")
async def health_check():
    """Health check endpoint"""
    return {"status": "healthy", "service": "test_mobile_endpoint"}

@app.post("/debug/mobile-entry", status_code=201)
async def test_mobile_endpoint(
    mobile_entry: MobileCalendarEntryCreate,
    db: AsyncSession = Depends(get_db),
):
    """Test mobile endpoint without authentication"""
    
    # Преобразуем симптомы из списка в строку
    symptoms_str = ""
    if mobile_entry.symptoms:
        symptoms_str = ", ".join(mobile_entry.symptoms)
    
    # Преобразуем данные из мобильного формата в формат сервера
    entry_type = EntryType.PERIOD
    if mobile_entry.type == "MENSTRUATION":
        entry_type = EntryType.PERIOD
    elif mobile_entry.type == "OVULATION":
        entry_type = EntryType.OVULATION
    else:
        entry_type = EntryType.SYMPTOMS
        
    # Преобразуем интенсивность потока
    flow_intensity = None
    if mobile_entry.flow_intensity is not None:
        if mobile_entry.flow_intensity <= 1:
            flow_intensity = FlowIntensity.SPOTTING
        elif mobile_entry.flow_intensity <= 2:
            flow_intensity = FlowIntensity.LIGHT
        elif mobile_entry.flow_intensity <= 4:
            flow_intensity = FlowIntensity.MEDIUM
        else:
            flow_intensity = FlowIntensity.HEAVY
    
    # Преобразуем настроение
    mood = None
    if mobile_entry.mood:
        if mobile_entry.mood == "HAPPY":
            mood = MoodType.HAPPY
        elif mobile_entry.mood == "SAD":
            mood = MoodType.SAD
        elif mobile_entry.mood == "NORMAL":
            mood = MoodType.HAPPY  # NORMAL мапится на HAPPY
        elif mobile_entry.mood == "ANXIOUS":
            mood = MoodType.ANXIOUS
        elif mobile_entry.mood == "IRRITATED":
            mood = MoodType.IRRITATED
        elif mobile_entry.mood == "ENERGETIC":
            mood = MoodType.ENERGETIC
        elif mobile_entry.mood == "TIRED":
            mood = MoodType.TIRED
    
    # Создаем объект CalendarEntryResponse для возврата
    response = {
        "id": 1,
        "user_id": 1,
        "entry_date": mobile_entry.date.isoformat(),
        "entry_type": entry_type.value,
        "flow_intensity": flow_intensity.value if flow_intensity else None,
        "period_symptoms": "",
        "mood": mood.value if mood else None,
        "energy_level": 1,  # Минимальное значение должно быть 1
        "sleep_hours": 0,
        "symptoms": symptoms_str,
        "medications": "",
        "notes": mobile_entry.notes or "",
        "created_at": date.today().isoformat(),
        "updated_at": date.today().isoformat(),
    }
    
    return response

# Функция для отправки тестового запроса
async def send_test_request():
    """Отправка тестового запроса к эндпоинту"""
    import httpx
    
    # Данные в формате мобильного приложения
    mobile_data = {
        "date": date.today().isoformat(),
        "type": "MENSTRUATION",
        "flow_intensity": 3,  # средний (1-5)
        "symptoms": ["CRAMPS", "HEADACHE"],  # массив строк
        "mood": "NORMAL",
        "notes": "Тестовая запись из мобильного приложения"
    }
    
    print(f"Отправляемые данные: {json.dumps(mobile_data, indent=2)}")
    
    # Подождем, пока сервис запустится
    await asyncio.sleep(2)
    
    async with httpx.AsyncClient() as client:
        try:
            # Проверяем работоспособность сервиса
            health_response = await client.get("http://localhost:8080/health")
            print(f"Проверка работоспособности: {health_response.status_code}")
            print(f"Ответ сервиса: {health_response.text}")
            
            # Отправляем запрос к тестовому эндпоинту
            response = await client.post(
                "http://localhost:8080/debug/mobile-entry", 
                json=mobile_data
            )
            
            print(f"Статус ответа: {response.status_code}")
            
            if response.status_code >= 200 and response.status_code < 300:
                print(f"Тело успешного ответа: {json.dumps(response.json(), indent=2)}")
                return True
            else:
                print("Ошибка при создании записи")
                try:
                    print(f"Тело ответа с ошибкой: {json.dumps(response.json(), indent=2)}")
                except:
                    print(f"Тело ответа не является JSON: {response.text}")
                return False
                
        except Exception as e:
            print(f"Ошибка при выполнении запроса: {e}")
            return False
            
# Основная функция
async def main():
    # Запускаем сервер в отдельном процессе
    config = uvicorn.Config(app, host="0.0.0.0", port=8080, log_level="info")
    server = uvicorn.Server(config)
    
    # Создаем задачу для запуска сервера
    server_task = asyncio.create_task(server.serve())
    
    # Дадим серверу немного времени на запуск
    await asyncio.sleep(1)
    
    try:
        # Отправляем тестовый запрос
        result = await send_test_request()
        
        if result:
            print("Тест успешно завершен!")
            return 0
        else:
            print("Тест завершился с ошибкой.")
            return 1
    finally:
        # Останавливаем сервер
        server.should_exit = True
        await server_task
        
if __name__ == "__main__":
    # Запускаем асинхронную функцию main()
    import asyncio
    try:
        exit_code = asyncio.run(main())
        sys.exit(exit_code)
    except KeyboardInterrupt:
        print("Выполнение прервано")
        sys.exit(1)