simple_test.py - master - Frozin/DIS-4

simple_test.py
87 строк4 KB
Frozin-p
Fixed database operations. Added files for testing
16 дек 2025, 16:45
16 дек 2025, 16:45109ec6f
import json
from client.main import RabbitMQClient


def run_simple_test():
    """Простой тест всех операций"""
    print("=" * 60)
    print("Простой тест всех операций RabbitMQ API")
    print("=" * 60)

    client = RabbitMQClient()

    try:
        print("\n1. Проверка здоровья сервиса:")
        response = client.send_request("health_check", {})
        print(f"   Статус: {response['status'] if response else 'Нет ответа'}")

        print("\n2. Получение всех пользователей (до создания):")
        response = client.send_request("get_all_users", {})
        if response and response['status'] == 'ok':
            users = response['data'].get('users', [])
            print(f"   Пользователей в системе: {len(users)}")

        print("\n3. Создание нового пользователя:")
        import random
        user_id = random.randint(1000, 9999)
        response = client.send_request(
            "create_user",
            {
                "username": f"demo_user_{user_id}",
                "email": f"demo{user_id}@example.com",
                "full_name": f"Демо пользователь {user_id}"
            }
        )

        if response and response['status'] == 'ok':
            created_user = response['data']
            print(f"   Пользователь создан: ID={created_user['id']}, Username={created_user['username']}")

            user_id = created_user['id']

            print(f"\n4. Получение созданного пользователя (ID={user_id}):")
            response = client.send_request("get_user", {"user_id": user_id})
            if response and response['status'] == 'ok':
                print(f"   Пользователь найден: {response['data']['email']}")

            print(f"\n5. Обновление пользователя (ID={user_id}):")
            response = client.send_request(
                "update_user",
                {
                    "user_id": user_id,
                    "full_name": "Обновленное имя",
                    "email": f"updated{user_id}@example.com"
                }
            )
            if response and response['status'] == 'ok':
                print(f"   Пользователь обновлен: {response['data']['full_name']}")

            print(f"\n6. Удаление пользователя (ID={user_id}):")
            response = client.send_request("delete_user", {"user_id": user_id})
            if response and response['status'] == 'ok':
                print(f"   Пользователь удален: ID={response['data']['id']}")

            print(f"\n7. Проверка удаления (ID={user_id}):")
            response = client.send_request("get_user", {"user_id": user_id})
            if response and response['status'] == 'error':
                print(f"   Пользователь не найден (как и ожидалось)")

        print("\n8. Получение всех пользователей (после удаления):")
        response = client.send_request("get_all_users", {})
        if response and response['status'] == 'ok':
            users = response['data'].get('users', [])
            print(f"   Пользователей в системе: {len(users)}")

        print("\n" + "=" * 60)
        print("Тест завершен успешно!")
        print("=" * 60)

    except Exception as e:
        print(f"\nОшибка во время теста: {e}")

    finally:
        client.close()


if __name__ == "__main__":
    run_simple_test()