diff --git a/esphome/__main__.py b/esphome/__main__.py index 280f491924..bba254436e 100644 --- a/esphome/__main__.py +++ b/esphome/__main__.py @@ -226,7 +226,9 @@ def run_miniterm(config: ConfigType, port: str, args) -> int: .replace(b"\n", b"") .decode("utf8", "backslashreplace") ) - time_str = datetime.now().time().strftime("[%H:%M:%S]") + time_ = datetime.now() + nanoseconds = time_.microsecond // 1000 + time_str = f"[{time_.hour:02}:{time_.minute:02}:{time_.second:02}.{nanoseconds:03}]" safe_print(parser.parse_line(line, time_str)) backtrace_state = platformio_api.process_stacktrace( diff --git a/esphome/components/api/client.py b/esphome/components/api/client.py index ce018b3b98..ca1fc089fa 100644 --- a/esphome/components/api/client.py +++ b/esphome/components/api/client.py @@ -62,9 +62,11 @@ async def async_run_logs(config: dict[str, Any], addresses: list[str]) -> None: time_ = datetime.now() message: bytes = msg.message text = message.decode("utf8", "backslashreplace") - for parsed_msg in parse_log_message( - text, f"[{time_.hour:02}:{time_.minute:02}:{time_.second:02}]" - ): + nanoseconds = time_.microsecond // 1000 + timestamp = ( + f"[{time_.hour:02}:{time_.minute:02}:{time_.second:02}.{nanoseconds:03}]" + ) + for parsed_msg in parse_log_message(text, timestamp): print(parsed_msg.replace("\033", "\\033") if dashboard else parsed_msg) stop = await async_run(cli, on_log, name=name) diff --git a/esphome/components/esp32/__init__.py b/esphome/components/esp32/__init__.py index 12d84dd4b3..50a47765bf 100644 --- a/esphome/components/esp32/__init__.py +++ b/esphome/components/esp32/__init__.py @@ -353,6 +353,7 @@ SUPPORTED_PLATFORMIO_ESP_IDF_5X = [ # pioarduino versions that don't require a release number # List based on https://github.com/pioarduino/esp-idf/releases SUPPORTED_PIOARDUINO_ESP_IDF_5X = [ + cv.Version(5, 5, 1), cv.Version(5, 5, 0), cv.Version(5, 4, 2), cv.Version(5, 4, 1), diff --git a/esphome/components/esp32_ble_client/ble_client_base.cpp b/esphome/components/esp32_ble_client/ble_client_base.cpp index af5162afb0..18321ef91c 100644 --- a/esphome/components/esp32_ble_client/ble_client_base.cpp +++ b/esphome/components/esp32_ble_client/ble_client_base.cpp @@ -43,13 +43,6 @@ void BLEClientBase::setup() { void BLEClientBase::set_state(espbt::ClientState st) { ESP_LOGV(TAG, "[%d] [%s] Set state %d", this->connection_index_, this->address_str_.c_str(), (int) st); ESPBTClient::set_state(st); - - if (st == espbt::ClientState::READY_TO_CONNECT) { - // Enable loop for state processing - this->enable_loop(); - // Connect immediately instead of waiting for next loop - this->connect(); - } } void BLEClientBase::loop() { @@ -65,8 +58,8 @@ void BLEClientBase::loop() { } this->set_state(espbt::ClientState::IDLE); } - // If its idle, we can disable the loop as set_state - // will enable it again when we need to connect. + // If idle, we can disable the loop as connect() + // will enable it again when a connection is needed. else if (this->state_ == espbt::ClientState::IDLE) { this->disable_loop(); } @@ -108,9 +101,20 @@ bool BLEClientBase::parse_device(const espbt::ESPBTDevice &device) { #endif void BLEClientBase::connect() { + // Prevent duplicate connection attempts + if (this->state_ == espbt::ClientState::CONNECTING || this->state_ == espbt::ClientState::CONNECTED || + this->state_ == espbt::ClientState::ESTABLISHED) { + ESP_LOGW(TAG, "[%d] [%s] Connection already in progress, state=%s", this->connection_index_, + this->address_str_.c_str(), espbt::client_state_to_string(this->state_)); + return; + } ESP_LOGI(TAG, "[%d] [%s] 0x%02x Connecting", this->connection_index_, this->address_str_.c_str(), this->remote_addr_type_); this->paired_ = false; + // Enable loop for state processing + this->enable_loop(); + // Immediately transition to CONNECTING to prevent duplicate connection attempts + this->set_state(espbt::ClientState::CONNECTING); // Determine connection parameters based on connection type if (this->connection_type_ == espbt::ConnectionType::V3_WITHOUT_CACHE) { @@ -168,7 +172,7 @@ void BLEClientBase::unconditional_disconnect() { this->log_gattc_warning_("esp_ble_gattc_close", err); } - if (this->state_ == espbt::ClientState::READY_TO_CONNECT || this->state_ == espbt::ClientState::DISCOVERED) { + if (this->state_ == espbt::ClientState::DISCOVERED) { this->set_address(0); this->set_state(espbt::ClientState::IDLE); } else { @@ -212,8 +216,6 @@ void BLEClientBase::handle_connection_result_(esp_err_t ret) { if (ret) { this->log_gattc_warning_("esp_ble_gattc_open", ret); this->set_state(espbt::ClientState::IDLE); - } else { - this->set_state(espbt::ClientState::CONNECTING); } } diff --git a/esphome/components/esp32_ble_tracker/esp32_ble_tracker.cpp b/esphome/components/esp32_ble_tracker/esp32_ble_tracker.cpp index 0edde169eb..bab1dd7c98 100644 --- a/esphome/components/esp32_ble_tracker/esp32_ble_tracker.cpp +++ b/esphome/components/esp32_ble_tracker/esp32_ble_tracker.cpp @@ -51,8 +51,6 @@ const char *client_state_to_string(ClientState state) { return "IDLE"; case ClientState::DISCOVERED: return "DISCOVERED"; - case ClientState::READY_TO_CONNECT: - return "READY_TO_CONNECT"; case ClientState::CONNECTING: return "CONNECTING"; case ClientState::CONNECTED: @@ -795,7 +793,7 @@ void ESP32BLETracker::try_promote_discovered_clients_() { #ifdef USE_ESP32_BLE_SOFTWARE_COEXISTENCE this->update_coex_preference_(true); #endif - client->set_state(ClientState::READY_TO_CONNECT); + client->connect(); break; } } diff --git a/esphome/components/esp32_ble_tracker/esp32_ble_tracker.h b/esphome/components/esp32_ble_tracker/esp32_ble_tracker.h index dd67156108..e53c2ac097 100644 --- a/esphome/components/esp32_ble_tracker/esp32_ble_tracker.h +++ b/esphome/components/esp32_ble_tracker/esp32_ble_tracker.h @@ -159,8 +159,6 @@ enum class ClientState : uint8_t { IDLE, // Device advertisement found. DISCOVERED, - // Device is discovered and the scanner is stopped - READY_TO_CONNECT, // Connection in progress. CONNECTING, // Initial connection established. @@ -313,7 +311,6 @@ class ESP32BLETracker : public Component, counts.discovered++; break; case ClientState::CONNECTING: - case ClientState::READY_TO_CONNECT: counts.connecting++; break; default: diff --git a/tests/dashboard/test_web_server.py b/tests/dashboard/test_web_server.py index b77ab7a7a3..a22f4a8b2a 100644 --- a/tests/dashboard/test_web_server.py +++ b/tests/dashboard/test_web_server.py @@ -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() diff --git a/tests/unit_tests/test_main.py b/tests/unit_tests/test_main.py new file mode 100644 index 0000000000..2c7236c7f8 --- /dev/null +++ b/tests/unit_tests/test_main.py @@ -0,0 +1,512 @@ +"""Unit tests for esphome.__main__ module.""" + +from __future__ import annotations + +from collections.abc import Generator +from dataclasses import dataclass +from typing import Any +from unittest.mock import Mock, patch + +import pytest + +from esphome.__main__ import choose_upload_log_host +from esphome.const import CONF_BROKER, CONF_MQTT, CONF_USE_ADDRESS, CONF_WIFI +from esphome.core import CORE + + +@dataclass +class MockSerialPort: + """Mock serial port for testing. + + Attributes: + path (str): The device path of the mock serial port (e.g., '/dev/ttyUSB0'). + description (str): A human-readable description of the mock serial port. + """ + + path: str + description: str + + +def setup_core( + config: dict[str, Any] | None = None, address: str | None = None +) -> None: + """ + Helper to set up CORE configuration with optional address. + + Args: + config (dict[str, Any] | None): The configuration dictionary to set for CORE. If None, an empty dict is used. + address (str | None): Optional network address to set in the configuration. If provided, it is set under the wifi config. + """ + if config is None: + config = {} + + if address is not None: + # Set address via wifi config (could also use ethernet) + config[CONF_WIFI] = {CONF_USE_ADDRESS: address} + + CORE.config = config + + +@pytest.fixture +def mock_no_serial_ports() -> Generator[Mock]: + """Mock get_serial_ports to return no ports.""" + with patch("esphome.__main__.get_serial_ports", return_value=[]) as mock: + yield mock + + +@pytest.fixture +def mock_serial_ports() -> Generator[Mock]: + """Mock get_serial_ports to return test ports.""" + mock_ports = [ + MockSerialPort("/dev/ttyUSB0", "USB Serial"), + MockSerialPort("/dev/ttyUSB1", "Another USB Serial"), + ] + with patch("esphome.__main__.get_serial_ports", return_value=mock_ports) as mock: + yield mock + + +@pytest.fixture +def mock_choose_prompt() -> Generator[Mock]: + """Mock choose_prompt to return default selection.""" + with patch("esphome.__main__.choose_prompt", return_value="/dev/ttyUSB0") as mock: + yield mock + + +@pytest.fixture +def mock_no_mqtt_logging() -> Generator[Mock]: + """Mock has_mqtt_logging to return False.""" + with patch("esphome.__main__.has_mqtt_logging", return_value=False) as mock: + yield mock + + +@pytest.fixture +def mock_has_mqtt_logging() -> Generator[Mock]: + """Mock has_mqtt_logging to return True.""" + with patch("esphome.__main__.has_mqtt_logging", return_value=True) as mock: + yield mock + + +def test_choose_upload_log_host_with_string_default() -> None: + """Test with a single string default device.""" + result = choose_upload_log_host( + default="192.168.1.100", + check_default=None, + show_ota=False, + show_mqtt=False, + show_api=False, + ) + assert result == ["192.168.1.100"] + + +def test_choose_upload_log_host_with_list_default() -> None: + """Test with a list of default devices.""" + result = choose_upload_log_host( + default=["192.168.1.100", "192.168.1.101"], + check_default=None, + show_ota=False, + show_mqtt=False, + show_api=False, + ) + assert result == ["192.168.1.100", "192.168.1.101"] + + +def test_choose_upload_log_host_with_multiple_ip_addresses() -> None: + """Test with multiple IP addresses as defaults.""" + result = choose_upload_log_host( + default=["1.2.3.4", "4.5.5.6"], + check_default=None, + show_ota=False, + show_mqtt=False, + show_api=False, + ) + assert result == ["1.2.3.4", "4.5.5.6"] + + +def test_choose_upload_log_host_with_mixed_hostnames_and_ips() -> None: + """Test with a mix of hostnames and IP addresses.""" + result = choose_upload_log_host( + default=["host.one", "host.one.local", "1.2.3.4"], + check_default=None, + show_ota=False, + show_mqtt=False, + show_api=False, + ) + assert result == ["host.one", "host.one.local", "1.2.3.4"] + + +def test_choose_upload_log_host_with_ota_list() -> None: + """Test with OTA as the only item in the list.""" + setup_core(config={"ota": {}}, address="192.168.1.100") + + result = choose_upload_log_host( + default=["OTA"], + check_default=None, + show_ota=True, + show_mqtt=False, + show_api=False, + ) + assert result == ["192.168.1.100"] + + +@pytest.mark.usefixtures("mock_has_mqtt_logging") +def test_choose_upload_log_host_with_ota_list_mqtt_fallback() -> None: + """Test with OTA list falling back to MQTT when no address.""" + setup_core() + + result = choose_upload_log_host( + default=["OTA"], + check_default=None, + show_ota=False, + show_mqtt=True, + show_api=False, + ) + assert result == ["MQTT"] + + +@pytest.mark.usefixtures("mock_no_serial_ports") +def test_choose_upload_log_host_with_serial_device_no_ports( + caplog: pytest.LogCaptureFixture, +) -> None: + """Test SERIAL device when no serial ports are found.""" + result = choose_upload_log_host( + default="SERIAL", + check_default=None, + show_ota=False, + show_mqtt=False, + show_api=False, + ) + assert result == [] + assert "No serial ports found, skipping SERIAL device" in caplog.text + + +@pytest.mark.usefixtures("mock_serial_ports") +def test_choose_upload_log_host_with_serial_device_with_ports( + mock_choose_prompt: Mock, +) -> None: + """Test SERIAL device when serial ports are available.""" + result = choose_upload_log_host( + default="SERIAL", + check_default=None, + show_ota=False, + show_mqtt=False, + show_api=False, + purpose="testing", + ) + assert result == ["/dev/ttyUSB0"] + mock_choose_prompt.assert_called_once_with( + [ + ("/dev/ttyUSB0 (USB Serial)", "/dev/ttyUSB0"), + ("/dev/ttyUSB1 (Another USB Serial)", "/dev/ttyUSB1"), + ], + purpose="testing", + ) + + +def test_choose_upload_log_host_with_ota_device_with_ota_config() -> None: + """Test OTA device when OTA is configured.""" + setup_core(config={"ota": {}}, address="192.168.1.100") + + result = choose_upload_log_host( + default="OTA", + check_default=None, + show_ota=True, + show_mqtt=False, + show_api=False, + ) + assert result == ["192.168.1.100"] + + +def test_choose_upload_log_host_with_ota_device_with_api_config() -> None: + """Test OTA device when API is configured.""" + setup_core(config={"api": {}}, address="192.168.1.100") + + result = choose_upload_log_host( + default="OTA", + check_default=None, + show_ota=False, + show_mqtt=False, + show_api=True, + ) + assert result == ["192.168.1.100"] + + +@pytest.mark.usefixtures("mock_has_mqtt_logging") +def test_choose_upload_log_host_with_ota_device_fallback_to_mqtt() -> None: + """Test OTA device fallback to MQTT when no OTA/API config.""" + setup_core() + + result = choose_upload_log_host( + default="OTA", + check_default=None, + show_ota=False, + show_mqtt=True, + show_api=False, + ) + assert result == ["MQTT"] + + +@pytest.mark.usefixtures("mock_no_mqtt_logging") +def test_choose_upload_log_host_with_ota_device_no_fallback() -> None: + """Test OTA device with no valid fallback options.""" + setup_core() + + result = choose_upload_log_host( + default="OTA", + check_default=None, + show_ota=True, + show_mqtt=True, + show_api=False, + ) + assert result == [] + + +@pytest.mark.usefixtures("mock_choose_prompt") +def test_choose_upload_log_host_multiple_devices() -> None: + """Test with multiple devices including special identifiers.""" + setup_core(config={"ota": {}}, address="192.168.1.100") + + mock_ports = [MockSerialPort("/dev/ttyUSB0", "USB Serial")] + + with patch("esphome.__main__.get_serial_ports", return_value=mock_ports): + result = choose_upload_log_host( + default=["192.168.1.50", "OTA", "SERIAL"], + check_default=None, + show_ota=True, + show_mqtt=False, + show_api=False, + ) + assert result == ["192.168.1.50", "192.168.1.100", "/dev/ttyUSB0"] + + +def test_choose_upload_log_host_no_defaults_with_serial_ports( + mock_choose_prompt: Mock, +) -> None: + """Test interactive mode with serial ports available.""" + mock_ports = [ + MockSerialPort("/dev/ttyUSB0", "USB Serial"), + ] + + setup_core() + + with patch("esphome.__main__.get_serial_ports", return_value=mock_ports): + result = choose_upload_log_host( + default=None, + check_default=None, + show_ota=False, + show_mqtt=False, + show_api=False, + purpose="uploading", + ) + assert result == ["/dev/ttyUSB0"] + mock_choose_prompt.assert_called_once_with( + [("/dev/ttyUSB0 (USB Serial)", "/dev/ttyUSB0")], + purpose="uploading", + ) + + +@pytest.mark.usefixtures("mock_no_serial_ports") +def test_choose_upload_log_host_no_defaults_with_ota() -> None: + """Test interactive mode with OTA option.""" + setup_core(config={"ota": {}}, address="192.168.1.100") + + with patch( + "esphome.__main__.choose_prompt", return_value="192.168.1.100" + ) as mock_prompt: + result = choose_upload_log_host( + default=None, + check_default=None, + show_ota=True, + show_mqtt=False, + show_api=False, + ) + assert result == ["192.168.1.100"] + mock_prompt.assert_called_once_with( + [("Over The Air (192.168.1.100)", "192.168.1.100")], + purpose=None, + ) + + +@pytest.mark.usefixtures("mock_no_serial_ports") +def test_choose_upload_log_host_no_defaults_with_api() -> None: + """Test interactive mode with API option.""" + setup_core(config={"api": {}}, address="192.168.1.100") + + with patch( + "esphome.__main__.choose_prompt", return_value="192.168.1.100" + ) as mock_prompt: + result = choose_upload_log_host( + default=None, + check_default=None, + show_ota=False, + show_mqtt=False, + show_api=True, + ) + assert result == ["192.168.1.100"] + mock_prompt.assert_called_once_with( + [("Over The Air (192.168.1.100)", "192.168.1.100")], + purpose=None, + ) + + +@pytest.mark.usefixtures("mock_no_serial_ports", "mock_has_mqtt_logging") +def test_choose_upload_log_host_no_defaults_with_mqtt() -> None: + """Test interactive mode with MQTT option.""" + setup_core(config={CONF_MQTT: {CONF_BROKER: "mqtt.local"}}) + + with patch("esphome.__main__.choose_prompt", return_value="MQTT") as mock_prompt: + result = choose_upload_log_host( + default=None, + check_default=None, + show_ota=False, + show_mqtt=True, + show_api=False, + ) + assert result == ["MQTT"] + mock_prompt.assert_called_once_with( + [("MQTT (mqtt.local)", "MQTT")], + purpose=None, + ) + + +@pytest.mark.usefixtures("mock_has_mqtt_logging") +def test_choose_upload_log_host_no_defaults_with_all_options( + mock_choose_prompt: Mock, +) -> None: + """Test interactive mode with all options available.""" + setup_core( + config={"ota": {}, "api": {}, CONF_MQTT: {CONF_BROKER: "mqtt.local"}}, + address="192.168.1.100", + ) + + mock_ports = [MockSerialPort("/dev/ttyUSB0", "USB Serial")] + + with patch("esphome.__main__.get_serial_ports", return_value=mock_ports): + result = choose_upload_log_host( + default=None, + check_default=None, + show_ota=True, + show_mqtt=True, + show_api=True, + purpose="testing", + ) + assert result == ["/dev/ttyUSB0"] + + expected_options = [ + ("/dev/ttyUSB0 (USB Serial)", "/dev/ttyUSB0"), + ("Over The Air (192.168.1.100)", "192.168.1.100"), + ("MQTT (mqtt.local)", "MQTT"), + ] + mock_choose_prompt.assert_called_once_with(expected_options, purpose="testing") + + +@pytest.mark.usefixtures("mock_no_serial_ports") +def test_choose_upload_log_host_check_default_matches() -> None: + """Test when check_default matches an available option.""" + setup_core(config={"ota": {}}, address="192.168.1.100") + + result = choose_upload_log_host( + default=None, + check_default="192.168.1.100", + show_ota=True, + show_mqtt=False, + show_api=False, + ) + assert result == ["192.168.1.100"] + + +@pytest.mark.usefixtures("mock_no_serial_ports") +def test_choose_upload_log_host_check_default_no_match() -> None: + """Test when check_default doesn't match any available option.""" + setup_core() + + with patch( + "esphome.__main__.choose_prompt", return_value="fallback" + ) as mock_prompt: + result = choose_upload_log_host( + default=None, + check_default="192.168.1.100", + show_ota=False, + show_mqtt=False, + show_api=False, + ) + assert result == ["fallback"] + mock_prompt.assert_called_once() + + +@pytest.mark.usefixtures("mock_no_serial_ports") +def test_choose_upload_log_host_empty_defaults_list() -> None: + """Test with an empty list as default.""" + with patch("esphome.__main__.choose_prompt", return_value="chosen") as mock_prompt: + result = choose_upload_log_host( + default=[], + check_default=None, + show_ota=False, + show_mqtt=False, + show_api=False, + ) + assert result == ["chosen"] + mock_prompt.assert_called_once() + + +@pytest.mark.usefixtures("mock_no_serial_ports", "mock_no_mqtt_logging") +def test_choose_upload_log_host_all_devices_unresolved( + caplog: pytest.LogCaptureFixture, +) -> None: + """Test when all specified devices cannot be resolved.""" + setup_core() + + result = choose_upload_log_host( + default=["SERIAL", "OTA"], + check_default=None, + show_ota=False, + show_mqtt=False, + show_api=False, + ) + assert result == [] + assert ( + "All specified devices: ['SERIAL', 'OTA'] could not be resolved." in caplog.text + ) + + +@pytest.mark.usefixtures("mock_no_serial_ports", "mock_no_mqtt_logging") +def test_choose_upload_log_host_mixed_resolved_unresolved() -> None: + """Test with a mix of resolved and unresolved devices.""" + setup_core() + + result = choose_upload_log_host( + default=["192.168.1.50", "SERIAL", "OTA"], + check_default=None, + show_ota=False, + show_mqtt=False, + show_api=False, + ) + assert result == ["192.168.1.50"] + + +def test_choose_upload_log_host_ota_both_conditions() -> None: + """Test OTA device when both OTA and API are configured and enabled.""" + setup_core(config={"ota": {}, "api": {}}, address="192.168.1.100") + + result = choose_upload_log_host( + default="OTA", + check_default=None, + show_ota=True, + show_mqtt=False, + show_api=True, + ) + assert result == ["192.168.1.100"] + + +@pytest.mark.usefixtures("mock_no_mqtt_logging") +def test_choose_upload_log_host_no_address_with_ota_config() -> None: + """Test OTA device when OTA is configured but no address is set.""" + setup_core(config={"ota": {}}) + + result = choose_upload_log_host( + default="OTA", + check_default=None, + show_ota=True, + show_mqtt=False, + show_api=False, + ) + assert result == []