mirror of
https://github.com/esphome/esphome.git
synced 2025-10-30 06:33:51 +00:00
Add some more coverage for dashboard web_server (#10682)
This commit is contained in:
J. Nick Koston
GitHub
@@ -1,13 +1,16 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
from collections.abc import Generator
|
||||
import gzip
|
||||
import json
|
||||
import os
|
||||
from unittest.mock import Mock
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock, Mock, patch
|
||||
|
||||
import pytest
|
||||
import pytest_asyncio
|
||||
from tornado.httpclient import AsyncHTTPClient, HTTPResponse
|
||||
from tornado.httpclient import AsyncHTTPClient, HTTPClientError, HTTPResponse
|
||||
from tornado.httpserver import HTTPServer
|
||||
from tornado.ioloop import IOLoop
|
||||
from tornado.testing import bind_unused_port
|
||||
@@ -34,6 +37,66 @@ class DashboardTestHelper:
|
||||
return await future
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_async_run_system_command() -> Generator[MagicMock]:
|
||||
"""Fixture to mock async_run_system_command."""
|
||||
with patch("esphome.dashboard.web_server.async_run_system_command") as mock:
|
||||
yield mock
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_trash_storage_path(tmp_path: Path) -> Generator[MagicMock]:
|
||||
"""Fixture to mock trash_storage_path."""
|
||||
trash_dir = tmp_path / "trash"
|
||||
with patch(
|
||||
"esphome.dashboard.web_server.trash_storage_path", return_value=str(trash_dir)
|
||||
) as mock:
|
||||
yield mock
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_archive_storage_path(tmp_path: Path) -> Generator[MagicMock]:
|
||||
"""Fixture to mock archive_storage_path."""
|
||||
archive_dir = tmp_path / "archive"
|
||||
with patch(
|
||||
"esphome.dashboard.web_server.archive_storage_path",
|
||||
return_value=str(archive_dir),
|
||||
) as mock:
|
||||
yield mock
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_dashboard_settings() -> Generator[MagicMock]:
|
||||
"""Fixture to mock dashboard settings."""
|
||||
with patch("esphome.dashboard.web_server.settings") as mock_settings:
|
||||
# Set default auth settings to avoid authentication issues
|
||||
mock_settings.using_auth = False
|
||||
mock_settings.on_ha_addon = False
|
||||
yield mock_settings
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_ext_storage_path(tmp_path: Path) -> Generator[MagicMock]:
|
||||
"""Fixture to mock ext_storage_path."""
|
||||
with patch("esphome.dashboard.web_server.ext_storage_path") as mock:
|
||||
mock.return_value = str(tmp_path / "storage.json")
|
||||
yield mock
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_storage_json() -> Generator[MagicMock]:
|
||||
"""Fixture to mock StorageJSON."""
|
||||
with patch("esphome.dashboard.web_server.StorageJSON") as mock:
|
||||
yield mock
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_idedata() -> Generator[MagicMock]:
|
||||
"""Fixture to mock platformio_api.IDEData."""
|
||||
with patch("esphome.dashboard.web_server.platformio_api.IDEData") as mock:
|
||||
yield mock
|
||||
|
||||
|
||||
@pytest_asyncio.fixture()
|
||||
async def dashboard() -> DashboardTestHelper:
|
||||
sock, port = bind_unused_port()
|
||||
@@ -80,3 +143,439 @@ async def test_devices_page(dashboard: DashboardTestHelper) -> None:
|
||||
first_device = configured_devices[0]
|
||||
assert first_device["name"] == "pico"
|
||||
assert first_device["configuration"] == "pico.yaml"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_wizard_handler_invalid_input(dashboard: DashboardTestHelper) -> None:
|
||||
"""Test the WizardRequestHandler.post method with invalid inputs."""
|
||||
# Test with missing name (should fail with 422)
|
||||
body_no_name = json.dumps(
|
||||
{
|
||||
"name": "", # Empty name
|
||||
"platform": "ESP32",
|
||||
"board": "esp32dev",
|
||||
}
|
||||
)
|
||||
with pytest.raises(HTTPClientError) as exc_info:
|
||||
await dashboard.fetch(
|
||||
"/wizard",
|
||||
method="POST",
|
||||
body=body_no_name,
|
||||
headers={"Content-Type": "application/json"},
|
||||
)
|
||||
assert exc_info.value.code == 422
|
||||
|
||||
# Test with invalid wizard type (should fail with 422)
|
||||
body_invalid_type = json.dumps(
|
||||
{
|
||||
"name": "test_device",
|
||||
"type": "invalid_type",
|
||||
"platform": "ESP32",
|
||||
"board": "esp32dev",
|
||||
}
|
||||
)
|
||||
with pytest.raises(HTTPClientError) as exc_info:
|
||||
await dashboard.fetch(
|
||||
"/wizard",
|
||||
method="POST",
|
||||
body=body_invalid_type,
|
||||
headers={"Content-Type": "application/json"},
|
||||
)
|
||||
assert exc_info.value.code == 422
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_wizard_handler_conflict(dashboard: DashboardTestHelper) -> None:
|
||||
"""Test the WizardRequestHandler.post when config already exists."""
|
||||
# Try to create a wizard for existing pico.yaml (should conflict)
|
||||
body = json.dumps(
|
||||
{
|
||||
"name": "pico", # This already exists in fixtures
|
||||
"platform": "ESP32",
|
||||
"board": "esp32dev",
|
||||
}
|
||||
)
|
||||
with pytest.raises(HTTPClientError) as exc_info:
|
||||
await dashboard.fetch(
|
||||
"/wizard",
|
||||
method="POST",
|
||||
body=body,
|
||||
headers={"Content-Type": "application/json"},
|
||||
)
|
||||
assert exc_info.value.code == 409
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_download_binary_handler_not_found(
|
||||
dashboard: DashboardTestHelper,
|
||||
) -> None:
|
||||
"""Test the DownloadBinaryRequestHandler.get with non-existent config."""
|
||||
with pytest.raises(HTTPClientError) as exc_info:
|
||||
await dashboard.fetch(
|
||||
"/download.bin?configuration=nonexistent.yaml",
|
||||
method="GET",
|
||||
)
|
||||
assert exc_info.value.code == 404
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.usefixtures("mock_ext_storage_path")
|
||||
async def test_download_binary_handler_no_file_param(
|
||||
dashboard: DashboardTestHelper,
|
||||
tmp_path: Path,
|
||||
mock_storage_json: MagicMock,
|
||||
) -> None:
|
||||
"""Test the DownloadBinaryRequestHandler.get without file parameter."""
|
||||
# Mock storage to exist, but still should fail without file param
|
||||
mock_storage = Mock()
|
||||
mock_storage.name = "test_device"
|
||||
mock_storage.firmware_bin_path = str(tmp_path / "firmware.bin")
|
||||
mock_storage_json.load.return_value = mock_storage
|
||||
|
||||
with pytest.raises(HTTPClientError) as exc_info:
|
||||
await dashboard.fetch(
|
||||
"/download.bin?configuration=pico.yaml",
|
||||
method="GET",
|
||||
)
|
||||
assert exc_info.value.code == 400
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.usefixtures("mock_ext_storage_path")
|
||||
async def test_download_binary_handler_with_file(
|
||||
dashboard: DashboardTestHelper,
|
||||
tmp_path: Path,
|
||||
mock_storage_json: MagicMock,
|
||||
) -> None:
|
||||
"""Test the DownloadBinaryRequestHandler.get with existing binary file."""
|
||||
# Create a fake binary file
|
||||
build_dir = tmp_path / ".esphome" / "build" / "test"
|
||||
build_dir.mkdir(parents=True)
|
||||
firmware_file = build_dir / "firmware.bin"
|
||||
firmware_file.write_bytes(b"fake firmware content")
|
||||
|
||||
# Mock storage JSON
|
||||
mock_storage = Mock()
|
||||
mock_storage.name = "test_device"
|
||||
mock_storage.firmware_bin_path = str(firmware_file)
|
||||
mock_storage_json.load.return_value = mock_storage
|
||||
|
||||
response = await dashboard.fetch(
|
||||
"/download.bin?configuration=test.yaml&file=firmware.bin",
|
||||
method="GET",
|
||||
)
|
||||
assert response.code == 200
|
||||
assert response.body == b"fake firmware content"
|
||||
assert response.headers["Content-Type"] == "application/octet-stream"
|
||||
assert "attachment" in response.headers["Content-Disposition"]
|
||||
assert "test_device-firmware.bin" in response.headers["Content-Disposition"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.usefixtures("mock_ext_storage_path")
|
||||
async def test_download_binary_handler_compressed(
|
||||
dashboard: DashboardTestHelper,
|
||||
tmp_path: Path,
|
||||
mock_storage_json: MagicMock,
|
||||
) -> None:
|
||||
"""Test the DownloadBinaryRequestHandler.get with compression."""
|
||||
# Create a fake binary file
|
||||
build_dir = tmp_path / ".esphome" / "build" / "test"
|
||||
build_dir.mkdir(parents=True)
|
||||
firmware_file = build_dir / "firmware.bin"
|
||||
original_content = b"fake firmware content for compression test"
|
||||
firmware_file.write_bytes(original_content)
|
||||
|
||||
# Mock storage JSON
|
||||
mock_storage = Mock()
|
||||
mock_storage.name = "test_device"
|
||||
mock_storage.firmware_bin_path = str(firmware_file)
|
||||
mock_storage_json.load.return_value = mock_storage
|
||||
|
||||
response = await dashboard.fetch(
|
||||
"/download.bin?configuration=test.yaml&file=firmware.bin&compressed=1",
|
||||
method="GET",
|
||||
)
|
||||
assert response.code == 200
|
||||
# Decompress and verify content
|
||||
decompressed = gzip.decompress(response.body)
|
||||
assert decompressed == original_content
|
||||
assert response.headers["Content-Type"] == "application/octet-stream"
|
||||
assert "firmware.bin.gz" in response.headers["Content-Disposition"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.usefixtures("mock_ext_storage_path")
|
||||
async def test_download_binary_handler_custom_download_name(
|
||||
dashboard: DashboardTestHelper,
|
||||
tmp_path: Path,
|
||||
mock_storage_json: MagicMock,
|
||||
) -> None:
|
||||
"""Test the DownloadBinaryRequestHandler.get with custom download name."""
|
||||
# Create a fake binary file
|
||||
build_dir = tmp_path / ".esphome" / "build" / "test"
|
||||
build_dir.mkdir(parents=True)
|
||||
firmware_file = build_dir / "firmware.bin"
|
||||
firmware_file.write_bytes(b"content")
|
||||
|
||||
# Mock storage JSON
|
||||
mock_storage = Mock()
|
||||
mock_storage.name = "test_device"
|
||||
mock_storage.firmware_bin_path = str(firmware_file)
|
||||
mock_storage_json.load.return_value = mock_storage
|
||||
|
||||
response = await dashboard.fetch(
|
||||
"/download.bin?configuration=test.yaml&file=firmware.bin&download=custom_name.bin",
|
||||
method="GET",
|
||||
)
|
||||
assert response.code == 200
|
||||
assert "custom_name.bin" in response.headers["Content-Disposition"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.usefixtures("mock_ext_storage_path")
|
||||
async def test_download_binary_handler_idedata_fallback(
|
||||
dashboard: DashboardTestHelper,
|
||||
tmp_path: Path,
|
||||
mock_async_run_system_command: MagicMock,
|
||||
mock_storage_json: MagicMock,
|
||||
mock_idedata: MagicMock,
|
||||
) -> None:
|
||||
"""Test the DownloadBinaryRequestHandler.get falling back to idedata for extra images."""
|
||||
# Create build directory but no bootloader file initially
|
||||
build_dir = tmp_path / ".esphome" / "build" / "test"
|
||||
build_dir.mkdir(parents=True)
|
||||
firmware_file = build_dir / "firmware.bin"
|
||||
firmware_file.write_bytes(b"firmware")
|
||||
|
||||
# Create bootloader file that idedata will find
|
||||
bootloader_file = tmp_path / "bootloader.bin"
|
||||
bootloader_file.write_bytes(b"bootloader content")
|
||||
|
||||
# Mock storage JSON
|
||||
mock_storage = Mock()
|
||||
mock_storage.name = "test_device"
|
||||
mock_storage.firmware_bin_path = str(firmware_file)
|
||||
mock_storage_json.load.return_value = mock_storage
|
||||
|
||||
# Mock idedata response
|
||||
mock_image = Mock()
|
||||
mock_image.path = str(bootloader_file)
|
||||
mock_idedata_instance = Mock()
|
||||
mock_idedata_instance.extra_flash_images = [mock_image]
|
||||
mock_idedata.return_value = mock_idedata_instance
|
||||
|
||||
# Mock async_run_system_command to return idedata JSON
|
||||
mock_async_run_system_command.return_value = (0, '{"extra_flash_images": []}', "")
|
||||
|
||||
response = await dashboard.fetch(
|
||||
"/download.bin?configuration=test.yaml&file=bootloader.bin",
|
||||
method="GET",
|
||||
)
|
||||
assert response.code == 200
|
||||
assert response.body == b"bootloader content"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_edit_request_handler_post_invalid_file(
|
||||
dashboard: DashboardTestHelper,
|
||||
) -> None:
|
||||
"""Test the EditRequestHandler.post with non-yaml file."""
|
||||
with pytest.raises(HTTPClientError) as exc_info:
|
||||
await dashboard.fetch(
|
||||
"/edit?configuration=test.txt",
|
||||
method="POST",
|
||||
body=b"content",
|
||||
)
|
||||
assert exc_info.value.code == 404
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_edit_request_handler_post_existing(
|
||||
dashboard: DashboardTestHelper,
|
||||
tmp_path: Path,
|
||||
mock_dashboard_settings: MagicMock,
|
||||
) -> None:
|
||||
"""Test the EditRequestHandler.post with existing yaml file."""
|
||||
# Create a temporary yaml file to edit (don't modify fixtures)
|
||||
test_file = tmp_path / "test_edit.yaml"
|
||||
test_file.write_text("esphome:\n name: original\n")
|
||||
|
||||
# Configure the mock settings
|
||||
mock_dashboard_settings.rel_path.return_value = str(test_file)
|
||||
mock_dashboard_settings.absolute_config_dir = test_file.parent
|
||||
|
||||
new_content = "esphome:\n name: modified\n"
|
||||
response = await dashboard.fetch(
|
||||
"/edit?configuration=test_edit.yaml",
|
||||
method="POST",
|
||||
body=new_content.encode(),
|
||||
)
|
||||
assert response.code == 200
|
||||
|
||||
# Verify the file was actually modified
|
||||
assert test_file.read_text() == new_content
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_unarchive_request_handler(
|
||||
dashboard: DashboardTestHelper,
|
||||
mock_archive_storage_path: MagicMock,
|
||||
mock_dashboard_settings: MagicMock,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
"""Test the UnArchiveRequestHandler.post method."""
|
||||
# Set up an archived file
|
||||
archive_dir = Path(mock_archive_storage_path.return_value)
|
||||
archive_dir.mkdir(parents=True, exist_ok=True)
|
||||
archived_file = archive_dir / "archived.yaml"
|
||||
archived_file.write_text("test content")
|
||||
|
||||
# Set up the destination path where the file should be moved
|
||||
config_dir = tmp_path / "config"
|
||||
config_dir.mkdir(parents=True, exist_ok=True)
|
||||
destination_file = config_dir / "archived.yaml"
|
||||
mock_dashboard_settings.rel_path.return_value = str(destination_file)
|
||||
|
||||
response = await dashboard.fetch(
|
||||
"/unarchive?configuration=archived.yaml",
|
||||
method="POST",
|
||||
body=b"",
|
||||
)
|
||||
assert response.code == 200
|
||||
|
||||
# Verify the file was actually moved from archive to config
|
||||
assert not archived_file.exists() # File should be gone from archive
|
||||
assert destination_file.exists() # File should now be in config
|
||||
assert destination_file.read_text() == "test content" # Content preserved
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_secret_keys_handler_no_file(dashboard: DashboardTestHelper) -> None:
|
||||
"""Test the SecretKeysRequestHandler.get when no secrets file exists."""
|
||||
# By default, there's no secrets file in the test fixtures
|
||||
with pytest.raises(HTTPClientError) as exc_info:
|
||||
await dashboard.fetch("/secret_keys", method="GET")
|
||||
assert exc_info.value.code == 404
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_secret_keys_handler_with_file(
|
||||
dashboard: DashboardTestHelper,
|
||||
tmp_path: Path,
|
||||
mock_dashboard_settings: MagicMock,
|
||||
) -> None:
|
||||
"""Test the SecretKeysRequestHandler.get when secrets file exists."""
|
||||
# Create a secrets file in temp directory
|
||||
secrets_file = tmp_path / "secrets.yaml"
|
||||
secrets_file.write_text(
|
||||
"wifi_ssid: TestNetwork\nwifi_password: TestPass123\napi_key: test_key\n"
|
||||
)
|
||||
|
||||
# Configure mock to return our temp secrets file
|
||||
# Since the file actually exists, os.path.isfile will return True naturally
|
||||
mock_dashboard_settings.rel_path.return_value = str(secrets_file)
|
||||
|
||||
response = await dashboard.fetch("/secret_keys", method="GET")
|
||||
assert response.code == 200
|
||||
data = json.loads(response.body.decode())
|
||||
assert "wifi_ssid" in data
|
||||
assert "wifi_password" in data
|
||||
assert "api_key" in data
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_json_config_handler(
|
||||
dashboard: DashboardTestHelper,
|
||||
mock_async_run_system_command: MagicMock,
|
||||
) -> None:
|
||||
"""Test the JsonConfigRequestHandler.get method."""
|
||||
# This will actually run the esphome config command on pico.yaml
|
||||
mock_output = json.dumps(
|
||||
{
|
||||
"esphome": {"name": "pico"},
|
||||
"esp32": {"board": "esp32dev"},
|
||||
}
|
||||
)
|
||||
mock_async_run_system_command.return_value = (0, mock_output, "")
|
||||
|
||||
response = await dashboard.fetch(
|
||||
"/json-config?configuration=pico.yaml", method="GET"
|
||||
)
|
||||
assert response.code == 200
|
||||
data = json.loads(response.body.decode())
|
||||
assert data["esphome"]["name"] == "pico"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_json_config_handler_invalid_config(
|
||||
dashboard: DashboardTestHelper,
|
||||
mock_async_run_system_command: MagicMock,
|
||||
) -> None:
|
||||
"""Test the JsonConfigRequestHandler.get with invalid config."""
|
||||
# Simulate esphome config command failure
|
||||
mock_async_run_system_command.return_value = (1, "", "Error: Invalid configuration")
|
||||
|
||||
with pytest.raises(HTTPClientError) as exc_info:
|
||||
await dashboard.fetch("/json-config?configuration=pico.yaml", method="GET")
|
||||
assert exc_info.value.code == 422
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_json_config_handler_not_found(dashboard: DashboardTestHelper) -> None:
|
||||
"""Test the JsonConfigRequestHandler.get with non-existent file."""
|
||||
with pytest.raises(HTTPClientError) as exc_info:
|
||||
await dashboard.fetch(
|
||||
"/json-config?configuration=nonexistent.yaml", method="GET"
|
||||
)
|
||||
assert exc_info.value.code == 404
|
||||
|
||||
|
||||
def test_start_web_server_with_address_port(
|
||||
tmp_path: Path,
|
||||
mock_trash_storage_path: MagicMock,
|
||||
mock_archive_storage_path: MagicMock,
|
||||
) -> None:
|
||||
"""Test the start_web_server function with address and port."""
|
||||
app = Mock()
|
||||
trash_dir = Path(mock_trash_storage_path.return_value)
|
||||
archive_dir = Path(mock_archive_storage_path.return_value)
|
||||
|
||||
# Create trash dir to test migration
|
||||
trash_dir.mkdir()
|
||||
(trash_dir / "old.yaml").write_text("old")
|
||||
|
||||
web_server.start_web_server(app, None, "127.0.0.1", 6052, str(tmp_path / "config"))
|
||||
|
||||
# The function calls app.listen directly for non-socket mode
|
||||
app.listen.assert_called_once_with(6052, "127.0.0.1")
|
||||
|
||||
# Verify trash was moved to archive
|
||||
assert not trash_dir.exists()
|
||||
assert archive_dir.exists()
|
||||
assert (archive_dir / "old.yaml").exists()
|
||||
|
||||
|
||||
@pytest.mark.skipif(os.name == "nt", reason="Unix sockets are not supported on Windows")
|
||||
@pytest.mark.usefixtures("mock_trash_storage_path", "mock_archive_storage_path")
|
||||
def test_start_web_server_with_unix_socket(tmp_path: Path) -> None:
|
||||
"""Test the start_web_server function with unix socket."""
|
||||
app = Mock()
|
||||
socket_path = tmp_path / "test.sock"
|
||||
|
||||
# Don't create trash_dir - it doesn't exist, so no migration needed
|
||||
with (
|
||||
patch("tornado.httpserver.HTTPServer") as mock_server_class,
|
||||
patch("tornado.netutil.bind_unix_socket") as mock_bind,
|
||||
):
|
||||
server = Mock()
|
||||
mock_server_class.return_value = server
|
||||
mock_bind.return_value = Mock()
|
||||
|
||||
web_server.start_web_server(
|
||||
app, str(socket_path), None, None, str(tmp_path / "config")
|
||||
)
|
||||
|
||||
mock_server_class.assert_called_once_with(app)
|
||||
mock_bind.assert_called_once_with(str(socket_path), mode=0o666)
|
||||
server.add_socket.assert_called_once()
|
||||
|
||||
Reference in New Issue
Block a user