File size: 5,830 Bytes
aefe381 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 | """Queue management service tests."""
import sys
import os
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from services.queue_service import QueueService
from models.queue import QueueStation, QueueCategory
def _make_service():
"""Create a queue service with test stations."""
qs = QueueService()
stations = [
QueueStation(id="f1", name="Pizza Stall", category="food", zone_id="z1",
avg_service_time_sec=60, current_length=5),
QueueStation(id="f2", name="Burger Joint", category="food", zone_id="z1",
avg_service_time_sec=90, current_length=10),
QueueStation(id="r1", name="Restroom A", category="restroom", zone_id="z2",
avg_service_time_sec=120, current_length=3),
QueueStation(id="m1", name="Merch Shop", category="merch", zone_id="z3",
avg_service_time_sec=120, current_length=0, is_open=False),
]
qs.register_stations(stations)
return qs
class TestQueueStation:
"""Test queue station model."""
def test_wait_time_calculation(self):
station = QueueStation(id="t", name="T", category="food", zone_id="z",
avg_service_time_sec=60, current_length=5)
assert station.estimated_wait_minutes == 5.0
def test_zero_queue_wait(self):
station = QueueStation(id="t", name="T", category="food", zone_id="z",
current_length=0)
assert station.estimated_wait_minutes == 0.0
def test_closed_station_wait(self):
station = QueueStation(id="t", name="T", category="food", zone_id="z",
current_length=10, is_open=False)
assert station.estimated_wait_minutes == 0.0
def test_wait_levels(self):
short = QueueStation(id="t", name="T", category="food", zone_id="z",
avg_service_time_sec=60, current_length=2)
assert short.wait_level == "short"
moderate = QueueStation(id="t", name="T", category="food", zone_id="z",
avg_service_time_sec=60, current_length=8)
assert moderate.wait_level == "moderate"
long_q = QueueStation(id="t", name="T", category="food", zone_id="z",
avg_service_time_sec=60, current_length=20)
assert long_q.wait_level == "long"
def test_category_icons(self):
assert QueueCategory.FOOD.icon == "🍕"
assert QueueCategory.RESTROOM.icon == "🚻"
assert QueueCategory.MERCHANDISE.icon == "🛍️"
class TestQueueService:
"""Test queue service operations."""
def test_get_all_stations(self):
qs = _make_service()
stations = qs.get_all_stations()
assert len(stations) == 4
def test_filter_by_category(self):
qs = _make_service()
food = qs.get_stations_by_category("food")
assert len(food) == 2
assert all(s["category"] == "food" for s in food)
def test_filter_by_zone(self):
qs = _make_service()
z1 = qs.get_stations_by_zone("z1")
assert len(z1) == 2
def test_update_queue_length(self):
qs = _make_service()
result = qs.update_queue_length("f1", 20)
assert result["current_length"] == 20
def test_update_nonexistent_station(self):
qs = _make_service()
result = qs.update_queue_length("nonexistent", 5)
assert result is None
def test_shortest_queue_overall(self):
qs = _make_service()
shortest = qs.get_shortest_queue()
assert shortest is not None
assert shortest["current_length"] <= 5
def test_shortest_queue_by_category(self):
qs = _make_service()
shortest = qs.get_shortest_queue("food")
assert shortest["category"] == "food"
assert shortest["id"] == "f1" # Pizza Stall has fewer people
def test_queue_summary(self):
qs = _make_service()
summary = qs.get_queue_summary()
assert "total_stations" in summary
assert "total_people_waiting" in summary
assert "categories" in summary
def test_virtual_queue_join(self):
qs = _make_service()
ticket = qs.join_virtual_queue("user1", "f1")
assert ticket is not None
assert ticket["user_id"] == "user1"
assert ticket["station_id"] == "f1"
assert ticket["position"] == 1
assert ticket["status"] == "waiting"
def test_virtual_queue_duplicate(self):
qs = _make_service()
t1 = qs.join_virtual_queue("user1", "f1")
t2 = qs.join_virtual_queue("user1", "f1")
# Should return existing ticket
assert t1["id"] == t2["id"]
def test_virtual_queue_position(self):
qs = _make_service()
t1 = qs.join_virtual_queue("user1", "f1")
t2 = qs.join_virtual_queue("user2", "f1")
assert t1["position"] == 1
assert t2["position"] == 2
def test_get_user_tickets(self):
qs = _make_service()
qs.join_virtual_queue("user1", "f1")
qs.join_virtual_queue("user1", "r1")
tickets = qs.get_user_tickets("user1")
assert len(tickets) == 2
def test_cancel_ticket(self):
qs = _make_service()
ticket = qs.join_virtual_queue("user1", "f1")
result = qs.cancel_ticket(ticket["id"], "user1")
assert result is True
def test_cancel_ticket_wrong_user(self):
qs = _make_service()
ticket = qs.join_virtual_queue("user1", "f1")
result = qs.cancel_ticket(ticket["id"], "user2")
assert result is False
def test_join_closed_station(self):
qs = _make_service()
ticket = qs.join_virtual_queue("user1", "m1") # Merch shop is closed
assert ticket is None
|