This commit is contained in:
184
test_standalone.py
Normal file
184
test_standalone.py
Normal file
@@ -0,0 +1,184 @@
|
||||
#!/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)
|
||||
Reference in New Issue
Block a user