test.py - master - Lexus_666/PIR_LAB_4

test.py
84 строки3 KB
Lexus-666
Lab_4_done added client + server + test run
25 дек 2025, 02:27
25 дек 2025, 02:271e2f432
import json
from client.main import RabbitMQClient

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

    client = RabbitMQClient()

    try:
        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()