Andrew K. Choi
ddce9f5125
All checks were successful
continuous-integration/drone/push Build is passing
427 lines
18 KiB
Python
Executable File
427 lines
18 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
|
||
"""
|
||
Скрипт для тестирования API приложения для безопасности женщин с использованием Python
|
||
"""
|
||
|
||
import json
|
||
import requests
|
||
import time
|
||
import sys
|
||
from datetime import datetime, timedelta
|
||
from termcolor import colored
|
||
|
||
BASE_URLS = {
|
||
"gateway": "http://localhost:8000",
|
||
"user": "http://localhost:8001",
|
||
"emergency": "http://localhost:8002",
|
||
"location": "http://localhost:8003",
|
||
"calendar": "http://localhost:8004",
|
||
"notification": "http://localhost:8005"
|
||
}
|
||
|
||
# Класс для тестирования API
|
||
class APITester:
|
||
def __init__(self):
|
||
self.token = None
|
||
self.user_id = None
|
||
self.emergency_contact_id = None
|
||
self.alert_id = None
|
||
self.calendar_entry_id = None
|
||
self.device_id = None
|
||
|
||
def print_section(self, title):
|
||
"""Печать заголовка секции"""
|
||
print("\n" + colored("="*50, "yellow"))
|
||
print(colored(f" {title}", "green", attrs=["bold"]))
|
||
print(colored("="*50, "yellow"))
|
||
|
||
def print_step(self, title):
|
||
"""Печать шага тестирования"""
|
||
print(colored(f"\n▶ {title}", "cyan"))
|
||
|
||
def print_response(self, response, title=None):
|
||
"""Печать ответа API"""
|
||
if title:
|
||
print(colored(f"● {title}:", "magenta"))
|
||
|
||
try:
|
||
if response.status_code >= 400:
|
||
print(colored(f"HTTP Status: {response.status_code}", "red"))
|
||
else:
|
||
print(colored(f"HTTP Status: {response.status_code}", "green"))
|
||
|
||
json_response = response.json()
|
||
print(json.dumps(json_response, indent=2, ensure_ascii=False))
|
||
return json_response
|
||
except:
|
||
print(colored("Не удалось распарсить JSON ответ", "red"))
|
||
print(response.text)
|
||
return {}
|
||
|
||
def check_services(self):
|
||
"""Проверка доступности всех сервисов"""
|
||
self.print_section("Проверка доступности сервисов")
|
||
|
||
all_up = True
|
||
|
||
for service_name, url in BASE_URLS.items():
|
||
print(colored(f"Проверка {service_name.upper()} ({url})...", "cyan"), end=" ")
|
||
|
||
try:
|
||
# Пробуем разные варианты эндпоинтов health
|
||
health_endpoints = ["/health", "/api/v1/health"]
|
||
service_up = False
|
||
|
||
for endpoint in health_endpoints:
|
||
try:
|
||
response = requests.get(f"{url}{endpoint}", timeout=5)
|
||
if response.status_code == 200:
|
||
if "status" in response.json() and "healthy" in response.json()["status"]:
|
||
service_up = True
|
||
break
|
||
except:
|
||
pass
|
||
|
||
if service_up:
|
||
print(colored("OK", "green"))
|
||
else:
|
||
print(colored("НЕДОСТУПЕН", "red"))
|
||
all_up = False
|
||
except Exception as e:
|
||
print(colored(f"ОШИБКА: {str(e)}", "red"))
|
||
all_up = False
|
||
|
||
if not all_up:
|
||
choice = input(colored("Некоторые сервисы недоступны. Продолжить тестирование? (y/n): ", "yellow"))
|
||
if choice.lower() != "y":
|
||
sys.exit(1)
|
||
|
||
def register_user(self):
|
||
"""Регистрация нового пользователя"""
|
||
self.print_step("Регистрация нового пользователя")
|
||
|
||
data = {
|
||
"username": f"testuser_{int(time.time())}", # Уникальное имя для каждого запуска
|
||
"email": f"test{int(time.time())}@example.com",
|
||
"password": "Test@1234",
|
||
"full_name": "Тестовый Пользователь",
|
||
"phone_number": "+79001234567"
|
||
}
|
||
|
||
response = requests.post(f"{BASE_URLS['user']}/api/v1/users/register", json=data)
|
||
user_data = self.print_response(response, "Ответ сервера")
|
||
|
||
if response.status_code == 201 or response.status_code == 200:
|
||
print(colored("✓ Пользователь успешно зарегистрирован", "green"))
|
||
self.user_id = user_data.get("id")
|
||
return data
|
||
else:
|
||
print(colored("✗ Ошибка при регистрации пользователя", "red"))
|
||
return None
|
||
|
||
def login_user(self, credentials):
|
||
"""Авторизация пользователя"""
|
||
self.print_step("Авторизация пользователя")
|
||
|
||
data = {
|
||
"username": credentials["username"],
|
||
"password": credentials["password"]
|
||
}
|
||
|
||
response = requests.post(f"{BASE_URLS['user']}/api/v1/auth/login", json=data)
|
||
login_data = self.print_response(response, "Ответ сервера")
|
||
|
||
if response.status_code == 200 and "access_token" in login_data:
|
||
self.token = login_data["access_token"]
|
||
print(colored("✓ Успешная авторизация", "green"))
|
||
return True
|
||
else:
|
||
print(colored("✗ Ошибка авторизации", "red"))
|
||
return False
|
||
|
||
def update_profile(self):
|
||
"""Обновление профиля пользователя"""
|
||
self.print_step("Обновление профиля пользователя")
|
||
|
||
data = {
|
||
"full_name": "Обновленное Имя",
|
||
"phone_number": "+79009876543"
|
||
}
|
||
|
||
headers = {"Authorization": f"Bearer {self.token}"}
|
||
response = requests.patch(f"{BASE_URLS['user']}/api/v1/users/me", json=data, headers=headers)
|
||
self.print_response(response, "Ответ сервера")
|
||
|
||
if response.status_code == 200:
|
||
print(colored("✓ Профиль успешно обновлен", "green"))
|
||
return True
|
||
else:
|
||
print(colored("✗ Ошибка при обновлении профиля", "red"))
|
||
return False
|
||
|
||
def add_emergency_contact(self):
|
||
"""Добавление контакта для экстренной связи"""
|
||
self.print_step("Добавление экстренного контакта")
|
||
|
||
data = {
|
||
"name": "Экстренный Контакт",
|
||
"phone_number": "+79991112233",
|
||
"relationship": "Родственник"
|
||
}
|
||
|
||
headers = {"Authorization": f"Bearer {self.token}"}
|
||
response = requests.post(f"{BASE_URLS['user']}/api/v1/users/me/emergency-contacts",
|
||
json=data, headers=headers)
|
||
contact_data = self.print_response(response, "Ответ сервера")
|
||
|
||
if response.status_code == 201 or response.status_code == 200:
|
||
self.emergency_contact_id = contact_data.get("id")
|
||
print(colored("✓ Экстренный контакт добавлен", "green"))
|
||
return True
|
||
else:
|
||
print(colored("✗ Ошибка при добавлении контакта", "red"))
|
||
return False
|
||
|
||
def update_location(self):
|
||
"""Обновление местоположения пользователя"""
|
||
self.print_step("Обновление местоположения")
|
||
|
||
data = {
|
||
"latitude": 55.7558,
|
||
"longitude": 37.6173,
|
||
"accuracy": 10.0
|
||
}
|
||
|
||
headers = {"Authorization": f"Bearer {self.token}"}
|
||
response = requests.post(f"{BASE_URLS['location']}/api/v1/locations/update",
|
||
json=data, headers=headers)
|
||
self.print_response(response, "Ответ сервера")
|
||
|
||
if response.status_code == 200 or response.status_code == 201:
|
||
print(colored("✓ Местоположение обновлено", "green"))
|
||
return True
|
||
else:
|
||
print(colored("✗ Ошибка при обновлении местоположения", "red"))
|
||
return False
|
||
|
||
def get_location(self):
|
||
"""Получение текущего местоположения"""
|
||
self.print_step("Получение местоположения")
|
||
|
||
headers = {"Authorization": f"Bearer {self.token}"}
|
||
response = requests.get(f"{BASE_URLS['location']}/api/v1/locations/last", headers=headers)
|
||
self.print_response(response, "Ответ сервера")
|
||
|
||
if response.status_code == 200:
|
||
print(colored("✓ Местоположение получено", "green"))
|
||
return True
|
||
else:
|
||
print(colored("✗ Ошибка при получении местоположения", "red"))
|
||
return False
|
||
|
||
def create_emergency_alert(self):
|
||
"""Создание экстренного оповещения"""
|
||
self.print_step("Создание экстренного оповещения")
|
||
|
||
data = {
|
||
"latitude": 55.7558,
|
||
"longitude": 37.6173,
|
||
"alert_type": "SOS",
|
||
"message": "Тестовое экстренное оповещение"
|
||
}
|
||
|
||
headers = {"Authorization": f"Bearer {self.token}"}
|
||
response = requests.post(f"{BASE_URLS['emergency']}/api/v1/emergency/alerts",
|
||
json=data, headers=headers)
|
||
alert_data = self.print_response(response, "Ответ сервера")
|
||
|
||
if response.status_code == 201 or response.status_code == 200:
|
||
self.alert_id = alert_data.get("id")
|
||
print(colored("✓ Экстренное оповещение создано", "green"))
|
||
return True
|
||
else:
|
||
print(colored("✗ Ошибка при создании оповещения", "red"))
|
||
return False
|
||
|
||
def get_alerts(self):
|
||
"""Получение активных оповещений"""
|
||
self.print_step("Получение активных оповещений")
|
||
|
||
headers = {"Authorization": f"Bearer {self.token}"}
|
||
response = requests.get(f"{BASE_URLS['emergency']}/api/v1/emergency/alerts/my", headers=headers)
|
||
self.print_response(response, "Ответ сервера")
|
||
|
||
if response.status_code == 200:
|
||
print(colored("✓ Список оповещений получен", "green"))
|
||
return True
|
||
else:
|
||
print(colored("✗ Ошибка при получении оповещений", "red"))
|
||
return False
|
||
|
||
def cancel_alert(self):
|
||
"""Отмена экстренного оповещения"""
|
||
if not self.alert_id:
|
||
print(colored("⚠ Нет активного оповещения для отмены", "yellow"))
|
||
return False
|
||
|
||
self.print_step("Отмена экстренного оповещения")
|
||
|
||
headers = {"Authorization": f"Bearer {self.token}"}
|
||
response = requests.patch(f"{BASE_URLS['emergency']}/api/v1/emergency/alerts/{self.alert_id}/cancel",
|
||
headers=headers)
|
||
self.print_response(response, "Ответ сервера")
|
||
|
||
if response.status_code == 200:
|
||
print(colored("✓ Оповещение отменено", "green"))
|
||
return True
|
||
else:
|
||
print(colored("✗ Ошибка при отмене оповещения", "red"))
|
||
return False
|
||
|
||
def create_calendar_entry(self):
|
||
"""Создание записи в календаре"""
|
||
self.print_step("Создание записи в календаре")
|
||
|
||
data = {
|
||
"entry_date": datetime.now().strftime("%Y-%m-%d"),
|
||
"cycle_day": 1,
|
||
"symptoms": ["HEADACHE", "FATIGUE"],
|
||
"mood": "NORMAL",
|
||
"notes": "Тестовая запись"
|
||
}
|
||
|
||
headers = {"Authorization": f"Bearer {self.token}"}
|
||
response = requests.post(f"{BASE_URLS['calendar']}/api/v1/calendar/entries",
|
||
json=data, headers=headers)
|
||
entry_data = self.print_response(response, "Ответ сервера")
|
||
|
||
if response.status_code == 201 or response.status_code == 200:
|
||
self.calendar_entry_id = entry_data.get("id")
|
||
print(colored("✓ Запись в календаре создана", "green"))
|
||
return True
|
||
else:
|
||
print(colored("✗ Ошибка при создании записи в календаре", "red"))
|
||
return False
|
||
|
||
def get_calendar_entries(self):
|
||
"""Получение записей календаря"""
|
||
self.print_step("Получение записей календаря")
|
||
|
||
start_date = (datetime.now() - timedelta(days=30)).strftime("%Y-%m-%d")
|
||
end_date = (datetime.now() + timedelta(days=30)).strftime("%Y-%m-%d")
|
||
|
||
headers = {"Authorization": f"Bearer {self.token}"}
|
||
response = requests.get(
|
||
f"{BASE_URLS['calendar']}/api/v1/calendar/entries?start_date={start_date}&end_date={end_date}",
|
||
headers=headers
|
||
)
|
||
self.print_response(response, "Ответ сервера")
|
||
|
||
if response.status_code == 200:
|
||
print(colored("✓ Записи календаря получены", "green"))
|
||
return True
|
||
else:
|
||
print(colored("✗ Ошибка при получении записей календаря", "red"))
|
||
return False
|
||
|
||
def register_device(self):
|
||
"""Регистрация устройства для уведомлений"""
|
||
self.print_step("Регистрация устройства для уведомлений")
|
||
|
||
data = {
|
||
"device_token": f"test-token-{int(time.time())}",
|
||
"device_type": "ANDROID",
|
||
"app_version": "1.0.0"
|
||
}
|
||
|
||
headers = {"Authorization": f"Bearer {self.token}"}
|
||
response = requests.post(f"{BASE_URLS['notification']}/api/v1/notifications/devices",
|
||
json=data, headers=headers)
|
||
device_data = self.print_response(response, "Ответ сервера")
|
||
|
||
if response.status_code == 201 or response.status_code == 200:
|
||
self.device_id = device_data.get("id")
|
||
print(colored("✓ Устройство зарегистрировано", "green"))
|
||
return True
|
||
else:
|
||
print(colored("✗ Ошибка при регистрации устройства", "red"))
|
||
return False
|
||
|
||
def set_notification_preferences(self):
|
||
"""Настройка предпочтений уведомлений"""
|
||
self.print_step("Настройка предпочтений уведомлений")
|
||
|
||
data = {
|
||
"emergency_alerts": True,
|
||
"nearby_incidents": True,
|
||
"calendar_reminders": True,
|
||
"system_notifications": True
|
||
}
|
||
|
||
headers = {"Authorization": f"Bearer {self.token}"}
|
||
response = requests.post(f"{BASE_URLS['notification']}/api/v1/notifications/preferences",
|
||
json=data, headers=headers)
|
||
self.print_response(response, "Ответ сервера")
|
||
|
||
if response.status_code == 200 or response.status_code == 201:
|
||
print(colored("✓ Предпочтения уведомлений настроены", "green"))
|
||
return True
|
||
else:
|
||
print(colored("✗ Ошибка при настройке предпочтений", "red"))
|
||
return False
|
||
|
||
def get_gateway_dashboard(self):
|
||
"""Получение данных через API Gateway"""
|
||
self.print_step("Получение данных пользователя через Gateway")
|
||
|
||
headers = {"Authorization": f"Bearer {self.token}"}
|
||
response = requests.get(f"{BASE_URLS['gateway']}/api/v1/users/dashboard", headers=headers)
|
||
self.print_response(response, "Ответ сервера")
|
||
|
||
if response.status_code == 200:
|
||
print(colored("✓ Данные получены через Gateway", "green"))
|
||
return True
|
||
else:
|
||
print(colored("✗ Ошибка при получении данных через Gateway", "red"))
|
||
return False
|
||
|
||
def run_tests(self):
|
||
"""Запуск полной серии тестов"""
|
||
self.print_section("НАЧАЛО ТЕСТИРОВАНИЯ API")
|
||
|
||
# Проверка доступности сервисов
|
||
self.check_services()
|
||
|
||
# Регистрация и авторизация
|
||
user_data = self.register_user()
|
||
if not user_data:
|
||
print(colored("Критическая ошибка: Не удалось зарегистрировать пользователя", "red"))
|
||
return False
|
||
|
||
if not self.login_user(user_data):
|
||
print(colored("Критическая ошибка: Не удалось авторизоваться", "red"))
|
||
return False
|
||
|
||
# Основные тесты
|
||
self.update_profile()
|
||
self.add_emergency_contact()
|
||
self.update_location()
|
||
self.get_location()
|
||
self.create_emergency_alert()
|
||
self.get_alerts()
|
||
self.cancel_alert()
|
||
self.create_calendar_entry()
|
||
self.get_calendar_entries()
|
||
self.register_device()
|
||
self.set_notification_preferences()
|
||
self.get_gateway_dashboard()
|
||
|
||
self.print_section("ТЕСТИРОВАНИЕ ЗАВЕРШЕНО")
|
||
print(colored("✓ Тесты API выполнены успешно!", "green", attrs=["bold"]))
|
||
return True
|
||
|
||
|
||
if __name__ == "__main__":
|
||
tester = APITester()
|
||
tester.run_tests() |