#!/usr/bin/env python3 import requests import json import sys import logging from datetime import datetime, date, timedelta from enum import Enum # Настройка логирования logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) # Порты сервисов CALENDAR_SERVICE_PORT = 8004 # Порт основного сервиса SIMPLIFIED_SERVICE_PORT = 8888 # Порт упрощенного сервиса # Мобильные типы записей class MobileEntryType(str, Enum): MENSTRUATION = "MENSTRUATION" OVULATION = "OVULATION" SPOTTING = "SPOTTING" DISCHARGE = "DISCHARGE" PAIN = "PAIN" MOOD = "MOOD" # Мобильные типы настроения class MobileMood(str, Enum): HAPPY = "HAPPY" SAD = "SAD" ANXIOUS = "ANXIOUS" IRRITABLE = "IRRITABLE" ENERGETIC = "ENERGETIC" TIRED = "TIRED" NORMAL = "NORMAL" # Мобильные типы симптомов class MobileSymptom(str, Enum): CRAMPS = "CRAMPS" HEADACHE = "HEADACHE" BLOATING = "BLOATING" FATIGUE = "FATIGUE" NAUSEA = "NAUSEA" BREAST_TENDERNESS = "BREAST_TENDERNESS" ACNE = "ACNE" BACKACHE = "BACKACHE" def test_calendar_apis(): # Для упрощенного сервиса авторизация не требуется token = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIyOSIsImVtYWlsIjoidGVzdDJAZXhhbXBsZS5jb20iLCJleHAiOjE3NTg4NjQwMzV9.Ap4ZD5EtwhLXRtm6KjuFvXMlk6XA-3HtMbaGEu9jX6M" headers = { "Content-Type": "application/json" } # Определяем какой сервис использовать service_port = detect_running_service() if not service_port: logger.error("Ни один из календарных сервисов не запущен!") return 1 # Если используем основной сервис, добавляем токен авторизации if service_port == CALENDAR_SERVICE_PORT: headers["Authorization"] = f"Bearer {token}" base_url = f"http://localhost:{service_port}" logger.info(f"Используем сервис на порту {service_port}") # 1. Тест стандартного формата на /api/v1/calendar/entries standard_entry = { "entry_date": (date.today() + timedelta(days=1)).isoformat(), "entry_type": "period", "flow_intensity": "medium", "notes": f"Стандартный тест {datetime.now().isoformat()}", "period_symptoms": "cramps", "energy_level": 3, "sleep_hours": 7, "medications": "", "symptoms": "headache" } standard_response = test_endpoint( url=f"{base_url}/api/v1/calendar/entries", method="POST", data=standard_entry, headers=headers, description="Стандартный формат - /api/v1/calendar/entries" ) # 2. Тест мобильного формата на /api/v1/calendar/entry mobile_entry = { "date": date.today().isoformat(), "type": "MENSTRUATION", "flow_intensity": 4, "mood": "HAPPY", "symptoms": [MobileSymptom.FATIGUE.value, MobileSymptom.HEADACHE.value], "notes": f"Мобильный тест {datetime.now().isoformat()}" } # Пробуем только для основного сервиса, если он запущен mobile_response = None if service_port == CALENDAR_SERVICE_PORT: mobile_response = test_endpoint( url=f"{base_url}/api/v1/calendar/entry", method="POST", data=mobile_entry, headers=headers, description="Мобильный формат - /api/v1/calendar/entry" ) else: logger.warning("Упрощенный сервис не поддерживает мобильный формат") # 3. Тест стандартного формата на /api/v1/entries legacy_entry = { "entry_date": (date.today() + timedelta(days=2)).isoformat(), "entry_type": "symptoms", "flow_intensity": None, "notes": f"Тест legacy endpoint {datetime.now().isoformat()}", "period_symptoms": "", "energy_level": 4, "sleep_hours": 8, "medications": "vitamin", "symptoms": "fatigue" } legacy_response = test_endpoint( url=f"{base_url}/api/v1/entries", method="POST", data=legacy_entry, headers=headers, description="Стандартный формат - /api/v1/entries (legacy)", expected_status=201 if service_port == CALENDAR_SERVICE_PORT else 404 ) # 4. Тест стандартного формата на /api/v1/entry entry_endpoint_entry = { "entry_date": (date.today() + timedelta(days=3)).isoformat(), "entry_type": "mood", "mood": "happy", "notes": f"Тест /entry endpoint {datetime.now().isoformat()}" } entry_response = test_endpoint( url=f"{base_url}/api/v1/entry", method="POST", data=entry_endpoint_entry, headers=headers, description="Стандартный формат - /api/v1/entry (без префикса)", expected_status=201 if service_port == CALENDAR_SERVICE_PORT else 404 ) # 5. Проверка списка записей get_entries_response = test_endpoint( url=f"{base_url}/api/v1/calendar/entries", method="GET", headers=headers, description="Получение списка записей" ) if get_entries_response and get_entries_response.status_code == 200: entries = get_entries_response.json() logger.info(f"Всего записей в календаре: {len(entries)}") for i, entry in enumerate(entries[-5:]): # Показываем последние 5 записей logger.info(f"Запись {i+1}: ID={entry.get('id')}, Дата={entry.get('entry_date')}, Тип={entry.get('entry_type')}") # Подсчитываем успешные тесты success_count = sum(1 for test in [standard_response, get_entries_response] if test and test.status_code in [200, 201]) # Для основного сервиса нужны все 5 успешных тестов if service_port == CALENDAR_SERVICE_PORT: additional_tests = [mobile_response, legacy_response, entry_response] success_count += sum(1 for test in additional_tests if test and test.status_code in [200, 201]) expected_success = 5 else: # Для упрощенного сервиса достаточно 2 успешных тестов expected_success = 2 if success_count >= expected_success: logger.info(f"✅ Успешно пройдено {success_count}/{expected_success} тестов!") return 0 else: logger.error(f"❌ Пройдено только {success_count}/{expected_success} тестов") return 1 def detect_running_service(): """Определяет, какой сервис запущен""" # Сначала пробуем упрощенный сервис try: response = requests.get(f"http://localhost:{SIMPLIFIED_SERVICE_PORT}", timeout=2) # Упрощенный сервис может не иметь /health эндпоинта, достаточно проверить, что сервер отвечает if response.status_code != 404: # Любой ответ, кроме 404, считаем успехом return SIMPLIFIED_SERVICE_PORT except requests.exceptions.RequestException: pass # Затем пробуем основной сервис try: response = requests.get(f"http://localhost:{CALENDAR_SERVICE_PORT}/health", timeout=2) if response.status_code == 200: return CALENDAR_SERVICE_PORT except requests.exceptions.RequestException: pass return None def test_endpoint(url, method, headers, description, data=None, expected_status=None): """Выполняет тест для конкретного эндпоинта""" logger.info(f"\nТестирование: {description}") logger.info(f"URL: {url}, Метод: {method}") if data: logger.info(f"Данные: {json.dumps(data, indent=2)}") try: if method.upper() == "GET": response = requests.get(url, headers=headers, timeout=5) elif method.upper() == "POST": response = requests.post(url, headers=headers, json=data, timeout=5) else: logger.error(f"Неподдерживаемый метод: {method}") return None logger.info(f"Статус ответа: {response.status_code}") # Проверяем ожидаемый статус, если указан if expected_status and response.status_code != expected_status: logger.warning(f"Получен статус {response.status_code}, ожидался {expected_status}") logger.warning(f"Ответ: {response.text}") return response # Для успешных ответов логируем детали if response.status_code in [200, 201]: logger.info("✅ Тест успешно пройден!") try: response_data = response.json() if isinstance(response_data, dict): logger.debug(f"Ответ: ID={response_data.get('id')}, Тип={response_data.get('entry_type')}") except ValueError: logger.debug(f"Ответ не в формате JSON: {response.text[:100]}...") else: logger.warning(f"❌ Тест не пройден. Статус: {response.status_code}") logger.warning(f"Ответ: {response.text}") return response except requests.exceptions.RequestException as e: logger.error(f"❌ Ошибка при выполнении запроса: {str(e)}") return None if __name__ == "__main__": sys.exit(test_calendar_apis())