| """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" |
|
|
| 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") |
| |
| 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") |
| assert ticket is None |
|
|
