mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-10-02 20:22:15 +02:00
maziggy
b99536cc33
- Remove 28 unused imports across 22 test files - Prefix 4 unused local variables with _ in app code (archives, bambu_mqtt, main) and remove 1 dead store - Consolidate import/import-from in test_plate_detection.py - Fix unreachable statement in test_archive_service.py - Simplify redundant comparison in timelapse_processor.py Resolves ~50 CodeQL py/unused-import, py/unused-local-variable, py/import-and-import-from, py/unreachable-statement, and py/redundant-comparison findings.
143 lines
5.3 KiB
Python
143 lines
5.3 KiB
Python
"""Integration tests for Discovery API endpoints.
|
|
|
|
Tests the full request/response cycle for /api/v1/discovery/ endpoints.
|
|
"""
|
|
|
|
import pytest
|
|
from httpx import AsyncClient
|
|
|
|
|
|
class TestDiscoveryAPI:
|
|
"""Integration tests for /api/v1/discovery/ endpoints."""
|
|
|
|
# ========================================================================
|
|
# Info endpoint
|
|
# ========================================================================
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_get_discovery_info(self, async_client: AsyncClient):
|
|
"""Verify discovery info endpoint returns expected fields."""
|
|
response = await async_client.get("/api/v1/discovery/info")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "is_docker" in data
|
|
assert "ssdp_running" in data
|
|
assert "scan_running" in data
|
|
assert isinstance(data["is_docker"], bool)
|
|
assert isinstance(data["ssdp_running"], bool)
|
|
assert isinstance(data["scan_running"], bool)
|
|
|
|
# ========================================================================
|
|
# SSDP Discovery endpoints
|
|
# ========================================================================
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_get_discovery_status(self, async_client: AsyncClient):
|
|
"""Verify SSDP discovery status endpoint works."""
|
|
response = await async_client.get("/api/v1/discovery/status")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "running" in data
|
|
assert isinstance(data["running"], bool)
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_start_discovery(self, async_client: AsyncClient):
|
|
"""Verify SSDP discovery can be started."""
|
|
response = await async_client.post("/api/v1/discovery/start?duration=1.0")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "running" in data
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_stop_discovery(self, async_client: AsyncClient):
|
|
"""Verify SSDP discovery can be stopped."""
|
|
response = await async_client.post("/api/v1/discovery/stop")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "running" in data
|
|
assert data["running"] is False
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_get_discovered_printers_empty(self, async_client: AsyncClient):
|
|
"""Verify empty list when no printers discovered."""
|
|
response = await async_client.get("/api/v1/discovery/printers")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert isinstance(data, list)
|
|
|
|
# ========================================================================
|
|
# Subnet scanning endpoints
|
|
# ========================================================================
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_start_subnet_scan(self, async_client: AsyncClient):
|
|
"""Verify subnet scan can be started."""
|
|
response = await async_client.post(
|
|
"/api/v1/discovery/scan",
|
|
json={"subnet": "192.168.1.0/30", "timeout": 0.1}, # Small subnet for testing
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "running" in data
|
|
assert "scanned" in data
|
|
assert "total" in data
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_get_scan_status(self, async_client: AsyncClient):
|
|
"""Verify subnet scan status endpoint works."""
|
|
response = await async_client.get("/api/v1/discovery/scan/status")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "running" in data
|
|
assert "scanned" in data
|
|
assert "total" in data
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_stop_subnet_scan(self, async_client: AsyncClient):
|
|
"""Verify subnet scan can be stopped."""
|
|
response = await async_client.post("/api/v1/discovery/scan/stop")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "running" in data
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_subnet_scan_invalid_subnet(self, async_client: AsyncClient):
|
|
"""Verify invalid subnet format is rejected."""
|
|
response = await async_client.post("/api/v1/discovery/scan", json={"subnet": "invalid-subnet", "timeout": 1.0})
|
|
|
|
# Should return 422 validation error or 200 with empty results
|
|
assert response.status_code in [200, 422]
|
|
|
|
|
|
class TestDiscoveryService:
|
|
"""Unit tests for discovery service functionality."""
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_docker_detection_fields(self, async_client: AsyncClient):
|
|
"""Verify Docker detection returns consistent response."""
|
|
# Call multiple times to ensure consistency
|
|
response1 = await async_client.get("/api/v1/discovery/info")
|
|
response2 = await async_client.get("/api/v1/discovery/info")
|
|
|
|
assert response1.status_code == 200
|
|
assert response2.status_code == 200
|
|
assert response1.json()["is_docker"] == response2.json()["is_docker"]
|