Spaces:
Sleeping
Sleeping
| """Tests for API routes.""" | |
| import pytest | |
| from fastapi.testclient import TestClient | |
| from app.main import app | |
| def client(): | |
| """Create test client.""" | |
| return TestClient(app) | |
| def sample_initial_state(): | |
| """Sample initial state string.""" | |
| return "5;5;2;1;2,2,4,4;;0,0" | |
| def sample_traffic(): | |
| """Sample traffic string for 5x5 grid.""" | |
| segments = [] | |
| # Horizontal segments | |
| for x in range(4): | |
| for y in range(5): | |
| segments.append(f"{x},{y},{x+1},{y},1") | |
| # Vertical segments | |
| for x in range(5): | |
| for y in range(4): | |
| segments.append(f"{x},{y},{x},{y+1},1") | |
| return ";".join(segments) | |
| class TestHealthEndpoint: | |
| """Tests for health check endpoint.""" | |
| def test_health_check(self, client): | |
| """Test health check returns ok.""" | |
| response = client.get("/api/health") | |
| assert response.status_code == 200 | |
| assert response.json()["status"] == "ok" | |
| class TestAlgorithmsEndpoint: | |
| """Tests for algorithms endpoint.""" | |
| def test_list_algorithms(self, client): | |
| """Test listing available algorithms.""" | |
| response = client.get("/api/algorithms") | |
| assert response.status_code == 200 | |
| data = response.json() | |
| assert "algorithms" in data | |
| assert len(data["algorithms"]) == 8 # 8 algorithms | |
| # Check algorithm structure | |
| algo = data["algorithms"][0] | |
| assert "code" in algo | |
| assert "name" in algo | |
| assert "description" in algo | |
| def test_algorithms_codes(self, client): | |
| """Test all expected algorithm codes are present.""" | |
| response = client.get("/api/algorithms") | |
| data = response.json() | |
| codes = [a["code"] for a in data["algorithms"]] | |
| expected = ["BF", "DF", "ID", "UC", "GR1", "GR2", "AS1", "AS2"] | |
| assert set(codes) == set(expected) | |
| class TestGenerateGridEndpoint: | |
| """Tests for grid generation endpoint.""" | |
| def test_generate_default_grid(self, client): | |
| """Test generating grid with default config.""" | |
| response = client.post("/api/grid/generate", json={}) | |
| assert response.status_code == 200 | |
| data = response.json() | |
| assert "initial_state" in data | |
| assert "traffic" in data | |
| assert "parsed" in data | |
| def test_generate_custom_grid(self, client): | |
| """Test generating grid with custom config.""" | |
| response = client.post( | |
| "/api/grid/generate", | |
| json={ | |
| "width": 8, | |
| "height": 8, | |
| "num_stores": 2, | |
| "num_destinations": 3, | |
| "num_tunnels": 1, | |
| "obstacle_density": 0.1, | |
| }, | |
| ) | |
| assert response.status_code == 200 | |
| data = response.json() | |
| parsed = data["parsed"] | |
| assert parsed["width"] == 8 | |
| assert parsed["height"] == 8 | |
| assert len(parsed["stores"]) == 2 | |
| assert len(parsed["destinations"]) == 3 | |
| def test_generate_grid_parsed_structure(self, client): | |
| """Test parsed grid structure.""" | |
| response = client.post("/api/grid/generate", json={"width": 5, "height": 5}) | |
| data = response.json() | |
| parsed = data["parsed"] | |
| assert "width" in parsed | |
| assert "height" in parsed | |
| assert "stores" in parsed | |
| assert "destinations" in parsed | |
| assert "tunnels" in parsed | |
| assert "segments" in parsed | |
| def test_generate_grid_stores_structure(self, client): | |
| """Test store structure in response.""" | |
| response = client.post( | |
| "/api/grid/generate", | |
| json={"width": 5, "height": 5, "num_stores": 1}, | |
| ) | |
| data = response.json() | |
| store = data["parsed"]["stores"][0] | |
| assert "id" in store | |
| assert "position" in store | |
| assert "x" in store["position"] | |
| assert "y" in store["position"] | |
| class TestPathEndpoint: | |
| """Tests for path finding endpoint.""" | |
| def test_find_path(self, client): | |
| """Test finding path between two points.""" | |
| # First generate a grid | |
| gen_response = client.post( | |
| "/api/grid/generate", | |
| json={"width": 5, "height": 5, "num_stores": 1, "num_destinations": 1}, | |
| ) | |
| grid_data = gen_response.json()["parsed"] | |
| # Find path from store to destination | |
| store = grid_data["stores"][0] | |
| dest = grid_data["destinations"][0] | |
| response = client.post( | |
| "/api/search/path", | |
| json={ | |
| "grid_width": 5, | |
| "grid_height": 5, | |
| "start": store["position"], | |
| "goal": dest["position"], | |
| "segments": grid_data["segments"], | |
| "tunnels": grid_data["tunnels"], | |
| "strategy": "BF", | |
| }, | |
| ) | |
| assert response.status_code == 200 | |
| data = response.json() | |
| assert "plan" in data | |
| assert "cost" in data | |
| assert "nodes_expanded" in data | |
| assert "path" in data | |
| assert "runtime_ms" in data | |
| assert "memory_kb" in data | |
| def test_find_path_all_strategies(self, client): | |
| """Test path finding with all strategies.""" | |
| gen_response = client.post( | |
| "/api/grid/generate", | |
| json={"width": 5, "height": 5, "num_stores": 1, "num_destinations": 1}, | |
| ) | |
| grid_data = gen_response.json()["parsed"] | |
| store = grid_data["stores"][0] | |
| dest = grid_data["destinations"][0] | |
| strategies = ["BF", "DF", "ID", "UC", "GR1", "GR2", "AS1", "AS2"] | |
| for strategy in strategies: | |
| response = client.post( | |
| "/api/search/path", | |
| json={ | |
| "grid_width": 5, | |
| "grid_height": 5, | |
| "start": store["position"], | |
| "goal": dest["position"], | |
| "segments": grid_data["segments"], | |
| "tunnels": grid_data["tunnels"], | |
| "strategy": strategy, | |
| }, | |
| ) | |
| assert response.status_code == 200, f"Strategy {strategy} failed" | |
| def test_find_path_with_visualization(self, client): | |
| """Test path finding returns visualization steps.""" | |
| gen_response = client.post( | |
| "/api/grid/generate", | |
| json={"width": 5, "height": 5, "num_stores": 1, "num_destinations": 1}, | |
| ) | |
| grid_data = gen_response.json()["parsed"] | |
| store = grid_data["stores"][0] | |
| dest = grid_data["destinations"][0] | |
| response = client.post( | |
| "/api/search/path", | |
| json={ | |
| "grid_width": 5, | |
| "grid_height": 5, | |
| "start": store["position"], | |
| "goal": dest["position"], | |
| "segments": grid_data["segments"], | |
| "tunnels": grid_data["tunnels"], | |
| "strategy": "BF", | |
| }, | |
| ) | |
| data = response.json() | |
| # Steps should be included | |
| assert "steps" in data | |
| assert data["steps"] is not None | |
| assert len(data["steps"]) > 0 | |
| class TestPlanEndpoint: | |
| """Tests for delivery planning endpoint.""" | |
| def test_create_plan(self, client, sample_initial_state, sample_traffic): | |
| """Test creating delivery plan.""" | |
| response = client.post( | |
| "/api/search/plan", | |
| json={ | |
| "initial_state": sample_initial_state, | |
| "traffic": sample_traffic, | |
| "strategy": "BF", | |
| "visualize": False, | |
| }, | |
| ) | |
| assert response.status_code == 200 | |
| data = response.json() | |
| assert "output" in data | |
| assert "assignments" in data | |
| assert "total_cost" in data | |
| assert "total_nodes_expanded" in data | |
| assert "runtime_ms" in data | |
| assert "memory_kb" in data | |
| def test_create_plan_all_strategies(self, client, sample_initial_state, sample_traffic): | |
| """Test planning with all strategies.""" | |
| strategies = ["BF", "DF", "ID", "UC", "GR1", "GR2", "AS1", "AS2"] | |
| for strategy in strategies: | |
| response = client.post( | |
| "/api/search/plan", | |
| json={ | |
| "initial_state": sample_initial_state, | |
| "traffic": sample_traffic, | |
| "strategy": strategy, | |
| "visualize": False, | |
| }, | |
| ) | |
| assert response.status_code == 200, f"Strategy {strategy} failed" | |
| def test_plan_assignments_structure(self, client, sample_initial_state, sample_traffic): | |
| """Test plan assignments structure.""" | |
| response = client.post( | |
| "/api/search/plan", | |
| json={ | |
| "initial_state": sample_initial_state, | |
| "traffic": sample_traffic, | |
| "strategy": "BF", | |
| "visualize": False, | |
| }, | |
| ) | |
| data = response.json() | |
| # Should have 2 assignments (2 destinations in sample) | |
| assert len(data["assignments"]) == 2 | |
| assignment = data["assignments"][0] | |
| assert "store_id" in assignment | |
| assert "destination_id" in assignment | |
| assert "path" in assignment | |
| class TestCompareEndpoint: | |
| """Tests for algorithm comparison endpoint.""" | |
| def test_compare_algorithms(self, client, sample_initial_state, sample_traffic): | |
| """Test comparing all algorithms.""" | |
| response = client.post( | |
| "/api/search/compare", | |
| json={ | |
| "initial_state": sample_initial_state, | |
| "traffic": sample_traffic, | |
| }, | |
| ) | |
| assert response.status_code == 200 | |
| data = response.json() | |
| assert "comparisons" in data | |
| assert "optimal_cost" in data | |
| assert len(data["comparisons"]) == 8 # 8 algorithms | |
| def test_compare_result_structure(self, client, sample_initial_state, sample_traffic): | |
| """Test comparison result structure.""" | |
| response = client.post( | |
| "/api/search/compare", | |
| json={ | |
| "initial_state": sample_initial_state, | |
| "traffic": sample_traffic, | |
| }, | |
| ) | |
| data = response.json() | |
| result = data["comparisons"][0] | |
| assert "algorithm" in result | |
| assert "name" in result | |
| assert "plan" in result | |
| assert "cost" in result | |
| assert "nodes_expanded" in result | |
| assert "runtime_ms" in result | |
| assert "memory_kb" in result | |
| assert "is_optimal" in result | |
| def test_compare_marks_optimal(self, client, sample_initial_state, sample_traffic): | |
| """Test that optimal solutions are marked.""" | |
| response = client.post( | |
| "/api/search/compare", | |
| json={ | |
| "initial_state": sample_initial_state, | |
| "traffic": sample_traffic, | |
| }, | |
| ) | |
| data = response.json() | |
| # At least one should be marked optimal | |
| optimal_count = sum(1 for r in data["comparisons"] if r["is_optimal"]) | |
| assert optimal_count >= 1 | |
| # Optimal results should have cost equal to optimal_cost | |
| for result in data["comparisons"]: | |
| if result["is_optimal"]: | |
| assert result["cost"] == data["optimal_cost"] | |
| def test_compare_ucs_astar_optimal(self, client, sample_initial_state, sample_traffic): | |
| """Test UCS and A* find optimal solutions.""" | |
| response = client.post( | |
| "/api/search/compare", | |
| json={ | |
| "initial_state": sample_initial_state, | |
| "traffic": sample_traffic, | |
| }, | |
| ) | |
| data = response.json() | |
| ucs = next(r for r in data["comparisons"] if r["algorithm"] == "UC") | |
| astar = next(r for r in data["comparisons"] if r["algorithm"] == "AS1") | |
| # Both should find optimal cost | |
| assert ucs["cost"] == data["optimal_cost"] | |
| assert astar["cost"] == data["optimal_cost"] | |
| class TestErrorHandling: | |
| """Tests for API error handling.""" | |
| def test_invalid_strategy(self, client, sample_initial_state, sample_traffic): | |
| """Test error on invalid strategy.""" | |
| response = client.post( | |
| "/api/search/plan", | |
| json={ | |
| "initial_state": sample_initial_state, | |
| "traffic": sample_traffic, | |
| "strategy": "INVALID", | |
| "visualize": False, | |
| }, | |
| ) | |
| # Should return validation error or internal error | |
| assert response.status_code in [400, 422, 500] | |
| def test_invalid_grid_dimensions(self, client): | |
| """Test error on invalid grid dimensions.""" | |
| response = client.post( | |
| "/api/grid/generate", | |
| json={"width": -1, "height": 5}, | |
| ) | |
| # Should return validation error | |
| assert response.status_code in [400, 422, 500] | |