1038 lines
No EOL
36 KiB
Python
1038 lines
No EOL
36 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Automatisierte Tests für die Tschunk Order API
|
|
Verwendet pytest und FastAPI TestClient
|
|
"""
|
|
|
|
import pytest
|
|
import asyncio
|
|
import json
|
|
from httpx import AsyncClient
|
|
from fastapi.testclient import TestClient
|
|
from main import app
|
|
from models import DrinkType, MateType
|
|
|
|
|
|
class TestTschunkOrderAPI:
|
|
"""Test-Klasse für die Tschunk Order API"""
|
|
|
|
@pytest.fixture
|
|
def client(self):
|
|
"""TestClient für HTTP-Endpunkte"""
|
|
return TestClient(app)
|
|
|
|
@pytest.fixture
|
|
def sample_order(self):
|
|
"""Beispiel-Bestellung für Tests"""
|
|
return {
|
|
"drinks": [
|
|
{
|
|
"drink_type": "Tschunk",
|
|
"mate_type": "Club Mate",
|
|
"quantity": 2,
|
|
"notes": "Test-Bestellung"
|
|
},
|
|
{
|
|
"drink_type": "Virgin Tschunk",
|
|
"mate_type": "Flora Mate",
|
|
"quantity": 1
|
|
}
|
|
]
|
|
}
|
|
|
|
def test_drinks_endpoint(self, client):
|
|
"""Test des Drinks-Endpunkts"""
|
|
response = client.get("/api/drinks")
|
|
assert response.status_code == 200
|
|
|
|
data = response.json()
|
|
assert "drink_types" in data
|
|
assert "mate_types" in data
|
|
|
|
# Prüfe alle Getränketypen
|
|
expected_drinks = [
|
|
"Tschunk",
|
|
"Tschunk Hannover-Mische",
|
|
"Tschunk alkoholfreier Rum",
|
|
"Virgin Tschunk"
|
|
]
|
|
for drink in expected_drinks:
|
|
assert drink in data["drink_types"]
|
|
|
|
# Prüfe Mate-Sorten
|
|
expected_mates = ["Flora Mate", "Club Mate"]
|
|
for mate in expected_mates:
|
|
assert mate in data["mate_types"]
|
|
|
|
def test_create_order_success(self, client, sample_order):
|
|
"""Test erfolgreiche Bestellerstellung"""
|
|
response = client.post("/api/orders", json=sample_order)
|
|
assert response.status_code == 201
|
|
|
|
order = response.json()
|
|
assert "id" in order
|
|
assert "order_date" in order
|
|
assert "drinks" in order
|
|
assert len(order["drinks"]) == 2
|
|
|
|
# Prüfe erste Bestellung
|
|
first_drink = order["drinks"][0]
|
|
assert first_drink["drink_type"] == "Tschunk"
|
|
assert first_drink["mate_type"] == "Club Mate"
|
|
assert first_drink["quantity"] == 2
|
|
assert first_drink["notes"] == "Test-Bestellung"
|
|
|
|
# Prüfe zweite Bestellung (ohne notes)
|
|
second_drink = order["drinks"][1]
|
|
assert second_drink["drink_type"] == "Virgin Tschunk"
|
|
assert second_drink["mate_type"] == "Flora Mate"
|
|
assert second_drink["quantity"] == 1
|
|
assert second_drink["notes"] is None
|
|
|
|
def test_create_order_empty_drinks(self, client):
|
|
"""Test Bestellerstellung ohne Getränke"""
|
|
response = client.post("/api/orders", json={"drinks": []})
|
|
assert response.status_code == 400
|
|
assert "Mindestens ein Getränk muss bestellt werden" in response.json()["detail"]
|
|
|
|
def test_create_order_invalid_drink_type(self, client):
|
|
"""Test Bestellerstellung mit ungültigem Getränketyp"""
|
|
invalid_order = {
|
|
"drinks": [
|
|
{
|
|
"drink_type": "Ungültiges Getränk",
|
|
"mate_type": "Club Mate"
|
|
}
|
|
]
|
|
}
|
|
response = client.post("/api/orders", json=invalid_order)
|
|
assert response.status_code == 422 # Validation Error
|
|
|
|
def test_get_all_orders(self, client, sample_order):
|
|
"""Test Abrufen aller Bestellungen"""
|
|
# Erstelle zuerst eine Bestellung
|
|
create_response = client.post("/api/orders", json=sample_order)
|
|
assert create_response.status_code == 201
|
|
|
|
# Hole alle Bestellungen
|
|
response = client.get("/api/orders")
|
|
assert response.status_code == 200
|
|
|
|
orders = response.json()
|
|
assert isinstance(orders, list)
|
|
assert len(orders) >= 1
|
|
|
|
# Prüfe, dass unsere Bestellung dabei ist
|
|
order_ids = [order["id"] for order in orders]
|
|
assert create_response.json()["id"] in order_ids
|
|
|
|
def test_orders_sorted_by_date(self, client):
|
|
"""Test dass Bestellungen nach Bestelldatum sortiert sind (älteste zuerst)"""
|
|
# Erstelle mehrere Bestellungen mit kurzen Pausen
|
|
import time
|
|
|
|
orders = []
|
|
for i in range(3):
|
|
order_data = {
|
|
"drinks": [
|
|
{
|
|
"drink_type": "Tschunk",
|
|
"mate_type": "Club Mate",
|
|
"quantity": i + 1,
|
|
"notes": f"Test-Bestellung {i + 1}"
|
|
}
|
|
]
|
|
}
|
|
response = client.post("/api/orders", json=order_data)
|
|
assert response.status_code == 201
|
|
orders.append(response.json())
|
|
time.sleep(0.1) # Kurze Pause zwischen Bestellungen
|
|
|
|
# Hole alle Bestellungen
|
|
response = client.get("/api/orders")
|
|
assert response.status_code == 200
|
|
|
|
all_orders = response.json()
|
|
|
|
# Prüfe, dass Bestellungen nach order_date sortiert sind (älteste zuerst)
|
|
order_dates = [order["order_date"] for order in all_orders if order["id"] in [o["id"] for o in orders]]
|
|
assert len(order_dates) == 3
|
|
|
|
# Prüfe Sortierung (order_dates sollten aufsteigend sein)
|
|
for i in range(len(order_dates) - 1):
|
|
assert order_dates[i] <= order_dates[i + 1], f"Bestellungen sind nicht korrekt sortiert: {order_dates[i]} > {order_dates[i + 1]}"
|
|
|
|
def test_delete_order_success(self, client, sample_order):
|
|
"""Test erfolgreiche Bestelllöschung"""
|
|
# Erstelle zuerst eine Bestellung
|
|
create_response = client.post("/api/orders", json=sample_order)
|
|
assert create_response.status_code == 201
|
|
order_id = create_response.json()["id"]
|
|
|
|
# Lösche die Bestellung
|
|
response = client.delete(f"/api/orders/{order_id}")
|
|
assert response.status_code == 200
|
|
assert f"Bestellung {order_id} wurde erfolgreich gelöscht" in response.json()["message"]
|
|
|
|
# Prüfe, dass die Bestellung wirklich gelöscht wurde
|
|
get_response = client.get("/api/orders")
|
|
orders = get_response.json()
|
|
order_ids = [order["id"] for order in orders]
|
|
assert order_id not in order_ids
|
|
|
|
def test_delete_order_not_found(self, client):
|
|
"""Test Löschung einer nicht existierenden Bestellung"""
|
|
response = client.delete("/api/orders/non-existent-id")
|
|
assert response.status_code == 404
|
|
assert "Bestellung nicht gefunden" in response.json()["detail"]
|
|
|
|
def test_order_with_default_quantity(self, client):
|
|
"""Test Bestellung ohne quantity (sollte 1 sein)"""
|
|
order_without_quantity = {
|
|
"drinks": [
|
|
{
|
|
"drink_type": "Tschunk",
|
|
"mate_type": "Club Mate"
|
|
# quantity wird nicht angegeben
|
|
}
|
|
]
|
|
}
|
|
|
|
response = client.post("/api/orders", json=order_without_quantity)
|
|
assert response.status_code == 201
|
|
|
|
order = response.json()
|
|
first_drink = order["drinks"][0]
|
|
assert first_drink["quantity"] == 1 # Default-Wert
|
|
|
|
def test_order_with_notes(self, client):
|
|
"""Test Bestellung mit Anmerkungen"""
|
|
order_with_notes = {
|
|
"drinks": [
|
|
{
|
|
"drink_type": "Virgin Tschunk",
|
|
"mate_type": "Flora Mate",
|
|
"quantity": 1,
|
|
"notes": "Bitte extra kalt servieren"
|
|
}
|
|
]
|
|
}
|
|
|
|
response = client.post("/api/orders", json=order_with_notes)
|
|
assert response.status_code == 201
|
|
|
|
order = response.json()
|
|
first_drink = order["drinks"][0]
|
|
assert first_drink["notes"] == "Bitte extra kalt servieren"
|
|
|
|
|
|
class TestWebSocket:
|
|
"""Test-Klasse für WebSocket-Funktionalität"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_websocket_connection(self):
|
|
"""Test WebSocket-Verbindung und initiale Nachricht"""
|
|
from websocket_manager import websocket_manager
|
|
|
|
# Erstelle einen Test-WebSocket
|
|
class MockWebSocket:
|
|
def __init__(self):
|
|
self.messages = []
|
|
self.closed = False
|
|
|
|
async def accept(self):
|
|
pass
|
|
|
|
async def send_text(self, message):
|
|
self.messages.append(message)
|
|
|
|
async def receive_text(self):
|
|
return "ping"
|
|
|
|
async def close(self):
|
|
self.closed = True
|
|
|
|
# Teste WebSocket-Manager
|
|
mock_ws = MockWebSocket()
|
|
|
|
# Verbinde
|
|
await websocket_manager.connect(mock_ws)
|
|
assert len(websocket_manager.active_connections) == 1
|
|
|
|
# Teste Broadcast
|
|
test_message = {"type": "test", "data": "test"}
|
|
await websocket_manager.broadcast(test_message)
|
|
assert len(mock_ws.messages) == 1
|
|
|
|
# Teste spezifische Broadcasts
|
|
await websocket_manager.broadcast_order_created({"id": "test", "drinks": []})
|
|
assert len(mock_ws.messages) == 2
|
|
|
|
await websocket_manager.broadcast_order_deleted("test-id")
|
|
assert len(mock_ws.messages) == 3
|
|
|
|
# Trenne Verbindung
|
|
websocket_manager.disconnect(mock_ws)
|
|
assert len(websocket_manager.active_connections) == 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_websocket_integration_create_order(self):
|
|
"""Test WebSocket-Broadcast bei Bestellerstellung über API"""
|
|
from websocket_manager import websocket_manager
|
|
from database import OrderDatabase
|
|
from models import Drink, DrinkType, MateType
|
|
|
|
# Erstelle einen Test-WebSocket
|
|
class MockWebSocket:
|
|
def __init__(self):
|
|
self.messages = []
|
|
self.closed = False
|
|
|
|
async def accept(self):
|
|
pass
|
|
|
|
async def send_text(self, message):
|
|
self.messages.append(message)
|
|
|
|
async def receive_text(self):
|
|
return "ping"
|
|
|
|
async def close(self):
|
|
self.closed = True
|
|
|
|
# Verbinde Mock-WebSocket
|
|
mock_ws = MockWebSocket()
|
|
await websocket_manager.connect(mock_ws)
|
|
|
|
# Erstelle Test-Datenbank
|
|
test_db = OrderDatabase()
|
|
|
|
# Erstelle Bestellung über Datenbank (sollte WebSocket-Broadcast auslösen)
|
|
drinks = [
|
|
Drink(
|
|
drink_type=DrinkType.TSCHUNK,
|
|
mate_type=MateType.CLUB_MATE,
|
|
quantity=2,
|
|
notes="Test-Bestellung"
|
|
)
|
|
]
|
|
|
|
order = await test_db.create_order(drinks)
|
|
|
|
# Prüfe, dass WebSocket-Nachricht gesendet wurde
|
|
assert len(mock_ws.messages) >= 1
|
|
|
|
# Finde die "order_created" Nachricht
|
|
order_created_message = None
|
|
for message in mock_ws.messages:
|
|
try:
|
|
data = json.loads(message)
|
|
if data.get("type") == "order_created":
|
|
order_created_message = data
|
|
break
|
|
except json.JSONDecodeError:
|
|
continue
|
|
|
|
assert order_created_message is not None
|
|
assert order_created_message["type"] == "order_created"
|
|
assert "order" in order_created_message
|
|
assert order_created_message["order"]["id"] == order.id
|
|
|
|
# Cleanup
|
|
websocket_manager.disconnect(mock_ws)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_websocket_integration_delete_order(self):
|
|
"""Test WebSocket-Broadcast bei Bestelllöschung über API"""
|
|
from websocket_manager import websocket_manager
|
|
from database import OrderDatabase
|
|
from models import Drink, DrinkType, MateType
|
|
|
|
# Erstelle einen Test-WebSocket
|
|
class MockWebSocket:
|
|
def __init__(self):
|
|
self.messages = []
|
|
self.closed = False
|
|
|
|
async def accept(self):
|
|
pass
|
|
|
|
async def send_text(self, message):
|
|
self.messages.append(message)
|
|
|
|
async def receive_text(self):
|
|
return "ping"
|
|
|
|
async def close(self):
|
|
self.closed = True
|
|
|
|
# Verbinde Mock-WebSocket
|
|
mock_ws = MockWebSocket()
|
|
await websocket_manager.connect(mock_ws)
|
|
|
|
# Erstelle Test-Datenbank
|
|
test_db = OrderDatabase()
|
|
|
|
# Erstelle zuerst eine Bestellung
|
|
drinks = [
|
|
Drink(
|
|
drink_type=DrinkType.VIRGIN_TSCHUNK,
|
|
mate_type=MateType.FLORA_MATE,
|
|
quantity=1
|
|
)
|
|
]
|
|
|
|
order = await test_db.create_order(drinks)
|
|
order_id = order.id
|
|
|
|
# Lösche die Bestellung (sollte WebSocket-Broadcast auslösen)
|
|
success = await test_db.delete_order(order_id)
|
|
assert success is True
|
|
|
|
# Prüfe, dass WebSocket-Nachricht gesendet wurde
|
|
assert len(mock_ws.messages) >= 2 # Mindestens create + delete
|
|
|
|
# Finde die "order_deleted" Nachricht
|
|
order_deleted_message = None
|
|
for message in mock_ws.messages:
|
|
try:
|
|
data = json.loads(message)
|
|
if data.get("type") == "order_deleted":
|
|
order_deleted_message = data
|
|
break
|
|
except json.JSONDecodeError:
|
|
continue
|
|
|
|
assert order_deleted_message is not None
|
|
assert order_deleted_message["type"] == "order_deleted"
|
|
assert "order_id" in order_deleted_message
|
|
assert order_deleted_message["order_id"] == order_id
|
|
|
|
# Cleanup
|
|
websocket_manager.disconnect(mock_ws)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_websocket_multiple_clients(self):
|
|
"""Test WebSocket-Broadcast an mehrere verbundene Clients"""
|
|
from websocket_manager import websocket_manager
|
|
from database import OrderDatabase
|
|
from models import Drink, DrinkType, MateType
|
|
|
|
# Erstelle mehrere Test-WebSockets
|
|
class MockWebSocket:
|
|
def __init__(self, client_id):
|
|
self.client_id = client_id
|
|
self.messages = []
|
|
self.closed = False
|
|
|
|
async def accept(self):
|
|
pass
|
|
|
|
async def send_text(self, message):
|
|
self.messages.append(message)
|
|
|
|
async def receive_text(self):
|
|
return "ping"
|
|
|
|
async def close(self):
|
|
self.closed = True
|
|
|
|
# Verbinde mehrere Mock-WebSockets
|
|
mock_ws1 = MockWebSocket("client1")
|
|
mock_ws2 = MockWebSocket("client2")
|
|
mock_ws3 = MockWebSocket("client3")
|
|
|
|
await websocket_manager.connect(mock_ws1)
|
|
await websocket_manager.connect(mock_ws2)
|
|
await websocket_manager.connect(mock_ws3)
|
|
|
|
assert len(websocket_manager.active_connections) == 3
|
|
|
|
# Erstelle Test-Datenbank
|
|
test_db = OrderDatabase()
|
|
|
|
# Erstelle Bestellung (sollte an alle Clients broadcasten)
|
|
drinks = [
|
|
Drink(
|
|
drink_type=DrinkType.TSCHUNK_HANNOVER_MISCHE,
|
|
mate_type=MateType.CLUB_MATE,
|
|
quantity=3,
|
|
notes="Multi-Client Test"
|
|
)
|
|
]
|
|
|
|
order = await test_db.create_order(drinks)
|
|
|
|
# Prüfe, dass alle Clients die Nachricht erhalten haben
|
|
for mock_ws in [mock_ws1, mock_ws2, mock_ws3]:
|
|
assert len(mock_ws.messages) >= 1
|
|
|
|
# Finde die "order_created" Nachricht
|
|
order_created_found = False
|
|
for message in mock_ws.messages:
|
|
try:
|
|
data = json.loads(message)
|
|
if data.get("type") == "order_created":
|
|
order_created_found = True
|
|
assert data["order"]["id"] == order.id
|
|
break
|
|
except json.JSONDecodeError:
|
|
continue
|
|
|
|
assert order_created_found, f"Client {mock_ws.client_id} hat keine order_created Nachricht erhalten"
|
|
|
|
# Cleanup
|
|
websocket_manager.disconnect(mock_ws1)
|
|
websocket_manager.disconnect(mock_ws2)
|
|
websocket_manager.disconnect(mock_ws3)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_websocket_orders_sorted(self):
|
|
"""Test dass WebSocket-Broadcasts sortierte Bestellungen senden"""
|
|
from websocket_manager import websocket_manager
|
|
from database import db # Verwende globale Datenbank-Instanz
|
|
from models import Drink, DrinkType, MateType
|
|
|
|
# Erstelle einen Test-WebSocket
|
|
class MockWebSocket:
|
|
def __init__(self):
|
|
self.messages = []
|
|
self.closed = False
|
|
|
|
async def accept(self):
|
|
pass
|
|
|
|
async def send_text(self, message):
|
|
self.messages.append(message)
|
|
|
|
async def receive_text(self):
|
|
return "ping"
|
|
|
|
async def close(self):
|
|
self.closed = True
|
|
|
|
# Erstelle mehrere Bestellungen in der globalen Datenbank
|
|
import asyncio
|
|
|
|
orders = []
|
|
for i in range(3):
|
|
drinks = [
|
|
Drink(
|
|
drink_type=DrinkType.TSCHUNK,
|
|
mate_type=MateType.CLUB_MATE,
|
|
quantity=i + 1,
|
|
notes=f"Test-Bestellung {i + 1}"
|
|
)
|
|
]
|
|
order = await db.create_order(drinks)
|
|
orders.append(order)
|
|
await asyncio.sleep(0.1) # Kurze Pause zwischen Bestellungen
|
|
|
|
# Verbinde Mock-WebSocket (sollte all_orders broadcast auslösen)
|
|
mock_ws = MockWebSocket()
|
|
await websocket_manager.connect(mock_ws)
|
|
|
|
# Simuliere den WebSocket-Endpunkt: Sende alle Bestellungen
|
|
all_orders = [order.model_dump() for order in db.get_all_orders()]
|
|
await websocket_manager.broadcast_all_orders(all_orders)
|
|
|
|
# Prüfe, dass die all_orders Nachricht gesendet wurde
|
|
assert len(mock_ws.messages) >= 1
|
|
|
|
# Finde die "all_orders" Nachricht
|
|
all_orders_message = None
|
|
for message in mock_ws.messages:
|
|
try:
|
|
data = json.loads(message)
|
|
if data.get("type") == "all_orders":
|
|
all_orders_message = data
|
|
break
|
|
except json.JSONDecodeError:
|
|
continue
|
|
|
|
assert all_orders_message is not None
|
|
assert "orders" in all_orders_message
|
|
|
|
# Prüfe, dass Bestellungen sortiert sind (älteste zuerst)
|
|
orders_data = all_orders_message["orders"]
|
|
assert len(orders_data) >= 3
|
|
|
|
# Extrahiere unsere Test-Bestellungen
|
|
test_order_ids = [order.id for order in orders]
|
|
test_orders = [order for order in orders_data if order["id"] in test_order_ids]
|
|
|
|
assert len(test_orders) == 3
|
|
|
|
# Prüfe Sortierung nach order_date
|
|
order_dates = [order["order_date"] for order in test_orders]
|
|
for i in range(len(order_dates) - 1):
|
|
assert order_dates[i] <= order_dates[i + 1], f"WebSocket-Bestellungen sind nicht korrekt sortiert: {order_dates[i]} > {order_dates[i + 1]}"
|
|
|
|
# Cleanup: Lösche Test-Bestellungen
|
|
for order in orders:
|
|
await db.delete_order(order.id)
|
|
|
|
websocket_manager.disconnect(mock_ws)
|
|
|
|
|
|
class TestWebSocketMultiClient:
|
|
"""Test-Klasse für Multi-Client WebSocket-Funktionalität"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_multiple_clients_receive_order_created(self):
|
|
"""Test dass alle verbundenen Clients order_created Nachrichten erhalten"""
|
|
from websocket_manager import websocket_manager
|
|
from database import db
|
|
from models import Drink, DrinkType, MateType
|
|
|
|
# Erstelle mehrere Test-WebSockets
|
|
class MockWebSocket:
|
|
def __init__(self, client_id):
|
|
self.client_id = client_id
|
|
self.messages = []
|
|
self.closed = False
|
|
|
|
async def accept(self):
|
|
pass
|
|
|
|
async def send_text(self, message):
|
|
self.messages.append(message)
|
|
|
|
async def receive_text(self):
|
|
return "ping"
|
|
|
|
async def close(self):
|
|
self.closed = True
|
|
|
|
# Verbinde mehrere Mock-WebSockets
|
|
mock_ws1 = MockWebSocket("client1")
|
|
mock_ws2 = MockWebSocket("client2")
|
|
mock_ws3 = MockWebSocket("client3")
|
|
|
|
await websocket_manager.connect(mock_ws1)
|
|
await websocket_manager.connect(mock_ws2)
|
|
await websocket_manager.connect(mock_ws3)
|
|
|
|
assert len(websocket_manager.active_connections) == 3
|
|
|
|
# Erstelle eine neue Bestellung (sollte an alle Clients broadcasten)
|
|
drinks = [
|
|
Drink(
|
|
drink_type=DrinkType.TSCHUNK,
|
|
mate_type=MateType.CLUB_MATE,
|
|
quantity=2,
|
|
notes="Multi-Client Test"
|
|
)
|
|
]
|
|
|
|
order = await db.create_order(drinks)
|
|
|
|
# Prüfe, dass alle Clients die order_created Nachricht erhalten haben
|
|
for mock_ws in [mock_ws1, mock_ws2, mock_ws3]:
|
|
assert len(mock_ws.messages) >= 1
|
|
|
|
# Finde die "order_created" Nachricht
|
|
order_created_found = False
|
|
for message in mock_ws.messages:
|
|
try:
|
|
data = json.loads(message)
|
|
if data.get("type") == "order_created":
|
|
order_created_found = True
|
|
assert data["order"]["id"] == order.id
|
|
assert "timestamp" in data
|
|
break
|
|
except json.JSONDecodeError:
|
|
continue
|
|
|
|
assert order_created_found, f"Client {mock_ws.client_id} hat keine order_created Nachricht erhalten"
|
|
|
|
# Cleanup
|
|
await db.delete_order(order.id)
|
|
websocket_manager.disconnect(mock_ws1)
|
|
websocket_manager.disconnect(mock_ws2)
|
|
websocket_manager.disconnect(mock_ws3)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_multiple_clients_receive_order_deleted(self):
|
|
"""Test dass alle verbundenen Clients order_deleted Nachrichten erhalten"""
|
|
from websocket_manager import websocket_manager
|
|
from database import db
|
|
from models import Drink, DrinkType, MateType
|
|
|
|
# Erstelle mehrere Test-WebSockets
|
|
class MockWebSocket:
|
|
def __init__(self, client_id):
|
|
self.client_id = client_id
|
|
self.messages = []
|
|
self.closed = False
|
|
|
|
async def accept(self):
|
|
pass
|
|
|
|
async def send_text(self, message):
|
|
self.messages.append(message)
|
|
|
|
async def receive_text(self):
|
|
return "ping"
|
|
|
|
async def close(self):
|
|
self.closed = True
|
|
|
|
# Verbinde mehrere Mock-WebSockets
|
|
mock_ws1 = MockWebSocket("client1")
|
|
mock_ws2 = MockWebSocket("client2")
|
|
|
|
await websocket_manager.connect(mock_ws1)
|
|
await websocket_manager.connect(mock_ws2)
|
|
|
|
# Erstelle zuerst eine Bestellung
|
|
drinks = [
|
|
Drink(
|
|
drink_type=DrinkType.VIRGIN_TSCHUNK,
|
|
mate_type=MateType.FLORA_MATE,
|
|
quantity=1,
|
|
notes="Delete Test"
|
|
)
|
|
]
|
|
|
|
order = await db.create_order(drinks)
|
|
order_id = order.id
|
|
|
|
# Lösche die Bestellung (sollte an alle Clients broadcasten)
|
|
success = await db.delete_order(order_id)
|
|
assert success is True
|
|
|
|
# Prüfe, dass alle Clients die order_deleted Nachricht erhalten haben
|
|
for mock_ws in [mock_ws1, mock_ws2]:
|
|
assert len(mock_ws.messages) >= 2 # Mindestens create + delete
|
|
|
|
# Finde die "order_deleted" Nachricht
|
|
order_deleted_found = False
|
|
for message in mock_ws.messages:
|
|
try:
|
|
data = json.loads(message)
|
|
if data.get("type") == "order_deleted":
|
|
order_deleted_found = True
|
|
assert data["order_id"] == order_id
|
|
assert "timestamp" in data
|
|
break
|
|
except json.JSONDecodeError:
|
|
continue
|
|
|
|
assert order_deleted_found, f"Client {mock_ws.client_id} hat keine order_deleted Nachricht erhalten"
|
|
|
|
# Cleanup
|
|
websocket_manager.disconnect(mock_ws1)
|
|
websocket_manager.disconnect(mock_ws2)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_client_disconnection_handling(self):
|
|
"""Test dass getrennte Clients keine weiteren Nachrichten erhalten"""
|
|
from websocket_manager import websocket_manager
|
|
from database import db
|
|
from models import Drink, DrinkType, MateType
|
|
|
|
# Erstelle Test-WebSockets
|
|
class MockWebSocket:
|
|
def __init__(self, client_id):
|
|
self.client_id = client_id
|
|
self.messages = []
|
|
self.closed = False
|
|
|
|
async def accept(self):
|
|
pass
|
|
|
|
async def send_text(self, message):
|
|
if not self.closed:
|
|
self.messages.append(message)
|
|
|
|
async def receive_text(self):
|
|
return "ping"
|
|
|
|
async def close(self):
|
|
self.closed = True
|
|
|
|
# Verbinde WebSockets
|
|
mock_ws1 = MockWebSocket("client1")
|
|
mock_ws2 = MockWebSocket("client2")
|
|
|
|
await websocket_manager.connect(mock_ws1)
|
|
await websocket_manager.connect(mock_ws2)
|
|
|
|
# Trenne ersten Client
|
|
websocket_manager.disconnect(mock_ws1)
|
|
assert len(websocket_manager.active_connections) == 1
|
|
|
|
# Erstelle eine Bestellung (sollte nur an client2 gehen)
|
|
drinks = [
|
|
Drink(
|
|
drink_type=DrinkType.TSCHUNK_HANNOVER_MISCHE,
|
|
mate_type=MateType.CLUB_MATE,
|
|
quantity=1
|
|
)
|
|
]
|
|
|
|
order = await db.create_order(drinks)
|
|
|
|
# Prüfe, dass nur client2 die Nachricht erhalten hat
|
|
assert len(mock_ws1.messages) == 0 # Getrennter Client sollte keine Nachrichten erhalten
|
|
assert len(mock_ws2.messages) >= 1 # Verbundener Client sollte Nachricht erhalten
|
|
|
|
# Prüfe, dass client2 die order_created Nachricht erhalten hat
|
|
order_created_found = False
|
|
for message in mock_ws2.messages:
|
|
try:
|
|
data = json.loads(message)
|
|
if data.get("type") == "order_created":
|
|
order_created_found = True
|
|
assert data["order"]["id"] == order.id
|
|
break
|
|
except json.JSONDecodeError:
|
|
continue
|
|
|
|
assert order_created_found, "Verbundener Client hat keine order_created Nachricht erhalten"
|
|
|
|
# Cleanup
|
|
await db.delete_order(order.id)
|
|
websocket_manager.disconnect(mock_ws2)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_concurrent_client_operations(self):
|
|
"""Test gleichzeitige Operationen mit mehreren Clients"""
|
|
from websocket_manager import websocket_manager
|
|
from database import db
|
|
from models import Drink, DrinkType, MateType
|
|
import asyncio
|
|
|
|
# Erstelle Test-WebSockets
|
|
class MockWebSocket:
|
|
def __init__(self, client_id):
|
|
self.client_id = client_id
|
|
self.messages = []
|
|
self.closed = False
|
|
|
|
async def accept(self):
|
|
pass
|
|
|
|
async def send_text(self, message):
|
|
self.messages.append(message)
|
|
|
|
async def receive_text(self):
|
|
return "ping"
|
|
|
|
async def close(self):
|
|
self.closed = True
|
|
|
|
# Verbinde mehrere WebSockets
|
|
clients = []
|
|
for i in range(5):
|
|
client = MockWebSocket(f"client{i}")
|
|
await websocket_manager.connect(client)
|
|
clients.append(client)
|
|
|
|
assert len(websocket_manager.active_connections) == 5
|
|
|
|
# Erstelle mehrere Bestellungen schnell hintereinander
|
|
orders = []
|
|
for i in range(3):
|
|
drinks = [
|
|
Drink(
|
|
drink_type=DrinkType.TSCHUNK,
|
|
mate_type=MateType.CLUB_MATE,
|
|
quantity=i + 1,
|
|
notes=f"Concurrent Test {i + 1}"
|
|
)
|
|
]
|
|
order = await db.create_order(drinks)
|
|
orders.append(order)
|
|
await asyncio.sleep(0.05) # Kurze Pause
|
|
|
|
# Prüfe, dass alle Clients alle Nachrichten erhalten haben
|
|
for client in clients:
|
|
# Jeder Client sollte mindestens 3 order_created Nachrichten erhalten haben
|
|
order_created_count = 0
|
|
for message in client.messages:
|
|
try:
|
|
data = json.loads(message)
|
|
if data.get("type") == "order_created":
|
|
order_created_count += 1
|
|
except json.JSONDecodeError:
|
|
continue
|
|
|
|
assert order_created_count >= 3, f"Client {client.client_id} hat nicht alle Nachrichten erhalten: {order_created_count}/3"
|
|
|
|
# Lösche alle Bestellungen
|
|
for order in orders:
|
|
await db.delete_order(order.id)
|
|
|
|
# Prüfe, dass alle Clients die delete-Nachrichten erhalten haben
|
|
for client in clients:
|
|
order_deleted_count = 0
|
|
for message in client.messages:
|
|
try:
|
|
data = json.loads(message)
|
|
if data.get("type") == "order_deleted":
|
|
order_deleted_count += 1
|
|
except json.JSONDecodeError:
|
|
continue
|
|
|
|
assert order_deleted_count >= 3, f"Client {client.client_id} hat nicht alle delete-Nachrichten erhalten: {order_deleted_count}/3"
|
|
|
|
# Cleanup
|
|
for client in clients:
|
|
websocket_manager.disconnect(client)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_websocket_message_format_consistency(self):
|
|
"""Test dass alle WebSocket-Nachrichten konsistentes Format haben"""
|
|
from websocket_manager import websocket_manager
|
|
from database import db
|
|
from models import Drink, DrinkType, MateType
|
|
|
|
# Erstelle Test-WebSocket
|
|
class MockWebSocket:
|
|
def __init__(self):
|
|
self.messages = []
|
|
self.closed = False
|
|
|
|
async def accept(self):
|
|
pass
|
|
|
|
async def send_text(self, message):
|
|
self.messages.append(message)
|
|
|
|
async def receive_text(self):
|
|
return "ping"
|
|
|
|
async def close(self):
|
|
self.closed = True
|
|
|
|
mock_ws = MockWebSocket()
|
|
await websocket_manager.connect(mock_ws)
|
|
|
|
# Erstelle und lösche eine Bestellung
|
|
drinks = [
|
|
Drink(
|
|
drink_type=DrinkType.TSCHUNK,
|
|
mate_type=MateType.CLUB_MATE,
|
|
quantity=1
|
|
)
|
|
]
|
|
|
|
order = await db.create_order(drinks)
|
|
order_id = order.id
|
|
|
|
await db.delete_order(order_id)
|
|
|
|
# Prüfe Nachrichtenformat
|
|
for message in mock_ws.messages:
|
|
try:
|
|
data = json.loads(message)
|
|
|
|
# Alle Nachrichten sollten diese Felder haben
|
|
assert "type" in data
|
|
assert "timestamp" in data
|
|
|
|
# Prüfe spezifische Nachrichtentypen
|
|
if data["type"] == "order_created":
|
|
assert "order" in data
|
|
assert "id" in data["order"]
|
|
assert "order_date" in data["order"]
|
|
assert "drinks" in data["order"]
|
|
|
|
elif data["type"] == "order_deleted":
|
|
assert "order_id" in data
|
|
|
|
elif data["type"] == "all_orders":
|
|
assert "orders" in data
|
|
assert isinstance(data["orders"], list)
|
|
|
|
except json.JSONDecodeError:
|
|
assert False, f"Ungültiges JSON-Format: {message}"
|
|
|
|
# Cleanup
|
|
websocket_manager.disconnect(mock_ws)
|
|
|
|
|
|
class TestDatabase:
|
|
"""Test-Klasse für Datenbankfunktionalität"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_database_operations(self):
|
|
"""Test Datenbankoperationen"""
|
|
from database import OrderDatabase
|
|
from models import Drink, DrinkType, MateType
|
|
|
|
# Erstelle neue Datenbank-Instanz für Tests
|
|
test_db = OrderDatabase()
|
|
|
|
# Teste Bestellerstellung
|
|
drinks = [
|
|
Drink(
|
|
drink_type=DrinkType.TSCHUNK,
|
|
mate_type=MateType.CLUB_MATE,
|
|
quantity=2,
|
|
notes="Test"
|
|
)
|
|
]
|
|
|
|
order = await test_db.create_order(drinks)
|
|
assert order.id is not None
|
|
assert len(order.drinks) == 1
|
|
assert order.drinks[0].drink_type == DrinkType.TSCHUNK
|
|
|
|
# Teste Abrufen aller Bestellungen
|
|
all_orders = test_db.get_all_orders()
|
|
assert len(all_orders) == 1
|
|
assert all_orders[0].id == order.id
|
|
|
|
# Teste Abrufen spezifischer Bestellung
|
|
retrieved_order = test_db.get_order_by_id(order.id)
|
|
assert retrieved_order is not None
|
|
assert retrieved_order.id == order.id
|
|
|
|
# Teste Löschung
|
|
success = await test_db.delete_order(order.id)
|
|
assert success is True
|
|
|
|
# Prüfe, dass Bestellung gelöscht wurde
|
|
all_orders_after = test_db.get_all_orders()
|
|
assert len(all_orders_after) == 0
|
|
|
|
# Teste Löschung nicht existierender Bestellung
|
|
success = await test_db.delete_order("non-existent")
|
|
assert success is False
|
|
|
|
|
|
def run_all_tests():
|
|
"""Führe alle Tests aus"""
|
|
import sys
|
|
import subprocess
|
|
|
|
print("🧪 Starte automatisierte Tests...")
|
|
print("=" * 50)
|
|
|
|
# Führe pytest aus
|
|
result = subprocess.run([
|
|
sys.executable, "-m", "pytest",
|
|
"test_automated.py",
|
|
"-v",
|
|
"--tb=short"
|
|
], capture_output=True, text=True)
|
|
|
|
print(result.stdout)
|
|
if result.stderr:
|
|
print("STDERR:", result.stderr)
|
|
|
|
print("=" * 50)
|
|
if result.returncode == 0:
|
|
print("✅ Alle Tests erfolgreich!")
|
|
else:
|
|
print("❌ Einige Tests fehlgeschlagen!")
|
|
|
|
return result.returncode == 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
success = run_all_tests()
|
|
exit(0 if success else 1) |