"""Test supervisor repairs.""" from collections.abc import Generator from http import HTTPStatus import os from unittest.mock import AsyncMock, patch from uuid import uuid4 from aiohasupervisor import SupervisorError from aiohasupervisor.models import ( ContextType, Issue, IssueType, Suggestion, SuggestionType, ) import pytest from homeassistant.components.hassio import DOMAIN from homeassistant.core import HomeAssistant from homeassistant.helpers import issue_registry as ir from homeassistant.setup import async_setup_component from .test_init import MOCK_ENVIRON from .test_issues import mock_resolution_info from tests.typing import ClientSessionGenerator @pytest.fixture(autouse=True) def fixture_supervisor_environ() -> Generator[None]: """Mock os environ for supervisor.""" with patch.dict(os.environ, MOCK_ENVIRON): yield @pytest.mark.usefixtures("all_setup_requests") async def test_supervisor_issue_repair_flow( hass: HomeAssistant, supervisor_client: AsyncMock, hass_client: ClientSessionGenerator, issue_registry: ir.IssueRegistry, ) -> None: """Test fix flow for supervisor issue.""" mock_resolution_info( supervisor_client, issues=[ Issue( type=IssueType.MULTIPLE_DATA_DISKS, context=ContextType.SYSTEM, reference="/dev/sda1", uuid=(issue_uuid := uuid4()), reference_extra=None, ), ], suggestions_by_issue={ issue_uuid: [ Suggestion( type=SuggestionType.RENAME_DATA_DISK, context=ContextType.SYSTEM, reference="/dev/sda1", uuid=(sugg_uuid := uuid4()), auto=False, reference_extra=None, ) ] }, ) assert await async_setup_component(hass, DOMAIN, {}) repair_issue = issue_registry.async_get_issue( domain="hassio", issue_id=issue_uuid.hex ) assert repair_issue client = await hass_client() resp = await client.post( "/api/repairs/issues/fix", json={"handler": "hassio", "issue_id": repair_issue.issue_id}, ) assert resp.status == HTTPStatus.OK data = await resp.json() flow_id = data["flow_id"] assert data == { "type": "form", "flow_id": flow_id, "handler": "hassio", "step_id": "system_rename_data_disk", "data_schema": [], "errors": None, "description_placeholders": {"reference": "/dev/sda1"}, "last_step": True, "preview": None, } resp = await client.post(f"/api/repairs/issues/fix/{flow_id}") assert resp.status == HTTPStatus.OK data = await resp.json() flow_id = data["flow_id"] assert data == { "type": "create_entry", "flow_id": flow_id, "handler": "hassio", "description": None, "description_placeholders": None, } assert not issue_registry.async_get_issue(domain="hassio", issue_id=issue_uuid.hex) supervisor_client.resolution.apply_suggestion.assert_called_once_with(sugg_uuid) @pytest.mark.usefixtures("all_setup_requests") async def test_supervisor_issue_repair_flow_with_multiple_suggestions( hass: HomeAssistant, supervisor_client: AsyncMock, hass_client: ClientSessionGenerator, issue_registry: ir.IssueRegistry, ) -> None: """Test fix flow for supervisor issue with multiple suggestions.""" mock_resolution_info( supervisor_client, issues=[ Issue( type=IssueType.REBOOT_REQUIRED, context=ContextType.SYSTEM, reference="test", uuid=(issue_uuid := uuid4()), reference_extra=None, ), ], suggestions_by_issue={ issue_uuid: [ Suggestion( type=SuggestionType.EXECUTE_REBOOT, context=ContextType.SYSTEM, reference="test", uuid=(sugg_uuid := uuid4()), auto=False, reference_extra=None, ), Suggestion( type="test_type", context=ContextType.SYSTEM, reference="test", uuid=uuid4(), auto=False, reference_extra=None, ), ] }, ) assert await async_setup_component(hass, DOMAIN, {}) repair_issue = issue_registry.async_get_issue( domain="hassio", issue_id=issue_uuid.hex ) assert repair_issue client = await hass_client() resp = await client.post( "/api/repairs/issues/fix", json={"handler": "hassio", "issue_id": repair_issue.issue_id}, ) assert resp.status == HTTPStatus.OK data = await resp.json() flow_id = data["flow_id"] # The unknown test_type suggestion has no translation and is filtered; # the remaining reboot suggestion is shown as a form directly assert data["type"] == "form" assert data["step_id"] == "system_execute_reboot" resp = await client.post(f"/api/repairs/issues/fix/{flow_id}", json={}) assert resp.status == HTTPStatus.OK data = await resp.json() flow_id = data["flow_id"] assert data == { "type": "create_entry", "flow_id": flow_id, "handler": "hassio", "description": None, "description_placeholders": None, } assert not issue_registry.async_get_issue(domain="hassio", issue_id=issue_uuid.hex) supervisor_client.resolution.apply_suggestion.assert_called_once_with(sugg_uuid) @pytest.mark.usefixtures("all_setup_requests") async def test_supervisor_issue_repair_flow_with_multiple_suggestions_and_confirmation( hass: HomeAssistant, supervisor_client: AsyncMock, hass_client: ClientSessionGenerator, issue_registry: ir.IssueRegistry, ) -> None: """Test fix flow for supervisor issue with multiple suggestions. Tests suggestions requiring confirmation. """ mock_resolution_info( supervisor_client, issues=[ Issue( type=IssueType.REBOOT_REQUIRED, context=ContextType.SYSTEM, reference=None, uuid=(issue_uuid := uuid4()), reference_extra=None, ), ], suggestions_by_issue={ issue_uuid: [ Suggestion( type=SuggestionType.EXECUTE_REBOOT, context=ContextType.SYSTEM, reference=None, uuid=(sugg_uuid := uuid4()), auto=False, reference_extra=None, ), Suggestion( type="test_type", context=ContextType.SYSTEM, reference=None, uuid=uuid4(), auto=False, reference_extra=None, ), ] }, ) assert await async_setup_component(hass, DOMAIN, {}) repair_issue = issue_registry.async_get_issue( domain="hassio", issue_id=issue_uuid.hex ) assert repair_issue client = await hass_client() resp = await client.post( "/api/repairs/issues/fix", json={"handler": "hassio", "issue_id": repair_issue.issue_id}, ) assert resp.status == HTTPStatus.OK data = await resp.json() flow_id = data["flow_id"] # The unknown test_type suggestion has no translation and is filtered; # the remaining reboot suggestion is shown as its confirmation form assert data == { "type": "form", "flow_id": flow_id, "handler": "hassio", "step_id": "system_execute_reboot", "data_schema": [], "errors": None, "description_placeholders": None, "last_step": True, "preview": None, } resp = await client.post(f"/api/repairs/issues/fix/{flow_id}") assert resp.status == HTTPStatus.OK data = await resp.json() flow_id = data["flow_id"] assert data == { "type": "create_entry", "flow_id": flow_id, "handler": "hassio", "description": None, "description_placeholders": None, } assert not issue_registry.async_get_issue(domain="hassio", issue_id=issue_uuid.hex) supervisor_client.resolution.apply_suggestion.assert_called_once_with(sugg_uuid) @pytest.mark.usefixtures("all_setup_requests") async def test_supervisor_issue_repair_flow_skip_confirmation( hass: HomeAssistant, supervisor_client: AsyncMock, hass_client: ClientSessionGenerator, issue_registry: ir.IssueRegistry, ) -> None: """Test confirmation skipped for fix flow for supervisor issue.""" mock_resolution_info( supervisor_client, issues=[ Issue( type=IssueType.REBOOT_REQUIRED, context=ContextType.SYSTEM, reference=None, uuid=(issue_uuid := uuid4()), reference_extra=None, ), ], suggestions_by_issue={ issue_uuid: [ Suggestion( type=SuggestionType.EXECUTE_REBOOT, context=ContextType.SYSTEM, reference=None, uuid=(sugg_uuid := uuid4()), auto=False, reference_extra=None, ), ] }, ) assert await async_setup_component(hass, DOMAIN, {}) repair_issue = issue_registry.async_get_issue( domain="hassio", issue_id=issue_uuid.hex ) assert repair_issue client = await hass_client() resp = await client.post( "/api/repairs/issues/fix", json={"handler": "hassio", "issue_id": repair_issue.issue_id}, ) assert resp.status == HTTPStatus.OK data = await resp.json() flow_id = data["flow_id"] assert data == { "type": "form", "flow_id": flow_id, "handler": "hassio", "step_id": "system_execute_reboot", "data_schema": [], "errors": None, "description_placeholders": None, "last_step": True, "preview": None, } resp = await client.post(f"/api/repairs/issues/fix/{flow_id}") assert resp.status == HTTPStatus.OK data = await resp.json() flow_id = data["flow_id"] assert data == { "type": "create_entry", "flow_id": flow_id, "handler": "hassio", "description": None, "description_placeholders": None, } assert not issue_registry.async_get_issue(domain="hassio", issue_id=issue_uuid.hex) supervisor_client.resolution.apply_suggestion.assert_called_once_with(sugg_uuid) @pytest.mark.usefixtures("all_setup_requests") async def test_supervisor_issue_ntp_sync_failed_repair_flow( hass: HomeAssistant, supervisor_client: AsyncMock, hass_client: ClientSessionGenerator, issue_registry: ir.IssueRegistry, ) -> None: """Test fix flow for NTP sync failed supervisor issue.""" mock_resolution_info( supervisor_client, issues=[ Issue( type=IssueType.NTP_SYNC_FAILED, context=ContextType.SYSTEM, reference=None, uuid=(issue_uuid := uuid4()), reference_extra=None, ), ], suggestions_by_issue={ issue_uuid: [ Suggestion( type=SuggestionType.ENABLE_NTP, context=ContextType.SYSTEM, reference=None, uuid=(sugg_uuid := uuid4()), auto=False, reference_extra=None, ), ] }, ) assert await async_setup_component(hass, DOMAIN, {}) repair_issue = issue_registry.async_get_issue( domain="hassio", issue_id=issue_uuid.hex ) assert repair_issue client = await hass_client() resp = await client.post( "/api/repairs/issues/fix", json={"handler": "hassio", "issue_id": repair_issue.issue_id}, ) assert resp.status == HTTPStatus.OK data = await resp.json() flow_id = data["flow_id"] assert data == { "type": "form", "flow_id": flow_id, "handler": "hassio", "step_id": "system_enable_ntp", "data_schema": [], "errors": None, "description_placeholders": None, "last_step": True, "preview": None, } resp = await client.post(f"/api/repairs/issues/fix/{flow_id}") assert resp.status == HTTPStatus.OK data = await resp.json() flow_id = data["flow_id"] assert data == { "type": "create_entry", "flow_id": flow_id, "handler": "hassio", "description": None, "description_placeholders": None, } assert not issue_registry.async_get_issue(domain="hassio", issue_id=issue_uuid.hex) supervisor_client.resolution.apply_suggestion.assert_called_once_with(sugg_uuid) @pytest.mark.usefixtures("all_setup_requests") async def test_supervisor_issue_ntp_sync_failed_repair_flow_error( hass: HomeAssistant, supervisor_client: AsyncMock, hass_client: ClientSessionGenerator, issue_registry: ir.IssueRegistry, ) -> None: """Test fix flow aborts when NTP re-enable fails.""" mock_resolution_info( supervisor_client, issues=[ Issue( type=IssueType.NTP_SYNC_FAILED, context=ContextType.SYSTEM, reference=None, uuid=(issue_uuid := uuid4()), reference_extra=None, ), ], suggestions_by_issue={ issue_uuid: [ Suggestion( type=SuggestionType.ENABLE_NTP, context=ContextType.SYSTEM, reference=None, uuid=uuid4(), auto=False, reference_extra=None, ), ] }, suggestion_result=SupervisorError("boom"), ) assert await async_setup_component(hass, DOMAIN, {}) repair_issue = issue_registry.async_get_issue( domain="hassio", issue_id=issue_uuid.hex ) assert repair_issue client = await hass_client() resp = await client.post( "/api/repairs/issues/fix", json={"handler": "hassio", "issue_id": repair_issue.issue_id}, ) assert resp.status == HTTPStatus.OK data = await resp.json() flow_id = data["flow_id"] resp = await client.post(f"/api/repairs/issues/fix/{flow_id}") assert resp.status == HTTPStatus.OK data = await resp.json() flow_id = data["flow_id"] assert data == { "type": "abort", "flow_id": flow_id, "handler": "hassio", "reason": "apply_suggestion_fail", "description_placeholders": None, } assert issue_registry.async_get_issue(domain="hassio", issue_id=issue_uuid.hex) @pytest.mark.usefixtures("all_setup_requests") async def test_mount_failed_repair_flow_error( hass: HomeAssistant, supervisor_client: AsyncMock, hass_client: ClientSessionGenerator, issue_registry: ir.IssueRegistry, ) -> None: """Test repair flow fails when repair fails to apply.""" mock_resolution_info( supervisor_client, issues=[ Issue( type=IssueType.MOUNT_FAILED, context=ContextType.MOUNT, reference="backup_share", uuid=(issue_uuid := uuid4()), reference_extra=None, ), ], suggestions_by_issue={ issue_uuid: [ Suggestion( type=SuggestionType.EXECUTE_RELOAD, context=ContextType.MOUNT, reference="backup_share", uuid=uuid4(), auto=False, reference_extra=None, ), Suggestion( type=SuggestionType.EXECUTE_REMOVE, context=ContextType.MOUNT, reference="backup_share", uuid=uuid4(), auto=False, reference_extra=None, ), ] }, suggestion_result=SupervisorError("boom"), ) assert await async_setup_component(hass, DOMAIN, {}) repair_issue = issue_registry.async_get_issue( domain="hassio", issue_id=issue_uuid.hex ) assert repair_issue client = await hass_client() resp = await client.post( "/api/repairs/issues/fix", json={"handler": "hassio", "issue_id": repair_issue.issue_id}, ) assert resp.status == HTTPStatus.OK data = await resp.json() flow_id = data["flow_id"] resp = await client.post( f"/api/repairs/issues/fix/{flow_id}", json={"next_step_id": "mount_execute_reload"}, ) assert resp.status == HTTPStatus.OK data = await resp.json() flow_id = data["flow_id"] assert data == { "type": "abort", "flow_id": flow_id, "handler": "hassio", "reason": "apply_suggestion_fail", "description_placeholders": None, } assert issue_registry.async_get_issue(domain="hassio", issue_id=issue_uuid.hex) @pytest.mark.usefixtures("all_setup_requests") async def test_mount_failed_repair_flow( hass: HomeAssistant, supervisor_client: AsyncMock, hass_client: ClientSessionGenerator, issue_registry: ir.IssueRegistry, ) -> None: """Test repair flow for mount_failed issue.""" mock_resolution_info( supervisor_client, issues=[ Issue( type=IssueType.MOUNT_FAILED, context=ContextType.MOUNT, reference="backup_share", uuid=(issue_uuid := uuid4()), reference_extra=None, ), ], suggestions_by_issue={ issue_uuid: [ Suggestion( type=SuggestionType.EXECUTE_RELOAD, context=ContextType.MOUNT, reference="backup_share", uuid=(sugg_uuid := uuid4()), auto=False, reference_extra=None, ), Suggestion( type=SuggestionType.EXECUTE_REMOVE, context=ContextType.MOUNT, reference="backup_share", uuid=uuid4(), auto=False, reference_extra=None, ), ] }, ) assert await async_setup_component(hass, DOMAIN, {}) repair_issue = issue_registry.async_get_issue( domain="hassio", issue_id=issue_uuid.hex ) assert repair_issue client = await hass_client() resp = await client.post( "/api/repairs/issues/fix", json={"handler": "hassio", "issue_id": repair_issue.issue_id}, ) assert resp.status == HTTPStatus.OK data = await resp.json() flow_id = data["flow_id"] assert data == { "type": "menu", "flow_id": flow_id, "handler": "hassio", "step_id": "fix_menu", "data_schema": [ { "type": "select", "options": [ ["mount_execute_reload", "mount_execute_reload"], ["mount_execute_remove", "mount_execute_remove"], ], "required": False, "name": "next_step_id", } ], "menu_options": ["mount_execute_reload", "mount_execute_remove"], "description_placeholders": { "reference": "backup_share", "storage_url": "/config/storage", }, } resp = await client.post( f"/api/repairs/issues/fix/{flow_id}", json={"next_step_id": "mount_execute_reload"}, ) assert resp.status == HTTPStatus.OK data = await resp.json() flow_id = data["flow_id"] assert data == { "type": "create_entry", "flow_id": flow_id, "handler": "hassio", "description": None, "description_placeholders": None, } assert not issue_registry.async_get_issue(domain="hassio", issue_id=issue_uuid.hex) supervisor_client.resolution.apply_suggestion.assert_called_once_with(sugg_uuid) @pytest.mark.usefixtures("all_setup_requests") async def test_mount_failed_remove_repair_flow( hass: HomeAssistant, supervisor_client: AsyncMock, hass_client: ClientSessionGenerator, issue_registry: ir.IssueRegistry, ) -> None: """Test removing the mount from the mount_failed repair requires confirmation.""" mock_resolution_info( supervisor_client, issues=[ Issue( type=IssueType.MOUNT_FAILED, context=ContextType.MOUNT, reference="backup_share", uuid=(issue_uuid := uuid4()), reference_extra=None, ), ], suggestions_by_issue={ issue_uuid: [ Suggestion( type=SuggestionType.EXECUTE_RELOAD, context=ContextType.MOUNT, reference="backup_share", uuid=uuid4(), auto=False, reference_extra=None, ), Suggestion( type=SuggestionType.EXECUTE_REMOVE, context=ContextType.MOUNT, reference="backup_share", uuid=(sugg_uuid := uuid4()), auto=False, reference_extra=None, ), ] }, ) assert await async_setup_component(hass, DOMAIN, {}) repair_issue = issue_registry.async_get_issue( domain="hassio", issue_id=issue_uuid.hex ) assert repair_issue client = await hass_client() resp = await client.post( "/api/repairs/issues/fix", json={"handler": "hassio", "issue_id": repair_issue.issue_id}, ) assert resp.status == HTTPStatus.OK data = await resp.json() flow_id = data["flow_id"] assert data["type"] == "menu" resp = await client.post( f"/api/repairs/issues/fix/{flow_id}", json={"next_step_id": "mount_execute_remove"}, ) assert resp.status == HTTPStatus.OK data = await resp.json() flow_id = data["flow_id"] assert data["type"] == "form" assert data["step_id"] == "mount_execute_remove" supervisor_client.resolution.apply_suggestion.assert_not_called() resp = await client.post(f"/api/repairs/issues/fix/{flow_id}", json={}) assert resp.status == HTTPStatus.OK data = await resp.json() assert data["type"] == "create_entry" assert not issue_registry.async_get_issue(domain="hassio", issue_id=issue_uuid.hex) supervisor_client.resolution.apply_suggestion.assert_called_once_with(sugg_uuid) @pytest.mark.usefixtures("all_setup_requests") async def test_mount_failed_move_local_data_repair_flow( hass: HomeAssistant, supervisor_client: AsyncMock, hass_client: ClientSessionGenerator, issue_registry: ir.IssueRegistry, ) -> None: """Test moving blocking local data from the mount_failed repair.""" mock_resolution_info( supervisor_client, issues=[ Issue( type=IssueType.MOUNT_FAILED, context=ContextType.MOUNT, reference="media_share", uuid=(issue_uuid := uuid4()), reference_extra=None, ), ], suggestions_by_issue={ issue_uuid: [ Suggestion( # Not in aiohasupervisor's SuggestionType enum yet, arrives # as a plain string like any newer supervisor suggestion type="move_local_data", context=ContextType.MOUNT, reference="media_share", uuid=(sugg_uuid := uuid4()), auto=False, reference_extra=None, ), Suggestion( type=SuggestionType.EXECUTE_RELOAD, context=ContextType.MOUNT, reference="media_share", uuid=uuid4(), auto=False, reference_extra=None, ), Suggestion( type=SuggestionType.EXECUTE_REMOVE, context=ContextType.MOUNT, reference="media_share", uuid=uuid4(), auto=False, reference_extra=None, ), ] }, ) assert await async_setup_component(hass, DOMAIN, {}) repair_issue = issue_registry.async_get_issue( domain="hassio", issue_id=issue_uuid.hex ) assert repair_issue client = await hass_client() resp = await client.post( "/api/repairs/issues/fix", json={"handler": "hassio", "issue_id": repair_issue.issue_id}, ) assert resp.status == HTTPStatus.OK data = await resp.json() flow_id = data["flow_id"] assert data["type"] == "menu" assert data["menu_options"] == [ "mount_move_local_data", "mount_execute_reload", "mount_execute_remove", ] # Moving data aside requires a confirmation step resp = await client.post( f"/api/repairs/issues/fix/{flow_id}", json={"next_step_id": "mount_move_local_data"}, ) assert resp.status == HTTPStatus.OK data = await resp.json() flow_id = data["flow_id"] assert data["type"] == "form" assert data["step_id"] == "mount_move_local_data" supervisor_client.resolution.apply_suggestion.assert_not_called() resp = await client.post(f"/api/repairs/issues/fix/{flow_id}", json={}) assert resp.status == HTTPStatus.OK data = await resp.json() assert data["type"] == "create_entry" assert not issue_registry.async_get_issue(domain="hassio", issue_id=issue_uuid.hex) supervisor_client.resolution.apply_suggestion.assert_called_once_with(sugg_uuid) @pytest.mark.usefixtures("all_setup_requests") async def test_mount_failed_repair_flow_hides_untranslated_suggestion( hass: HomeAssistant, supervisor_client: AsyncMock, hass_client: ClientSessionGenerator, issue_registry: ir.IssueRegistry, ) -> None: """Test a suggestion without fix flow translation is left out of the menu.""" mock_resolution_info( supervisor_client, issues=[ Issue( type=IssueType.MOUNT_FAILED, context=ContextType.MOUNT, reference="backup_share", uuid=(issue_uuid := uuid4()), reference_extra=None, ), ], suggestions_by_issue={ issue_uuid: [ Suggestion( # A suggestion from a newer Supervisor unknown to this Core type="suggestion_from_the_future", context=ContextType.MOUNT, reference="backup_share", uuid=uuid4(), auto=False, reference_extra=None, ), Suggestion( type=SuggestionType.EXECUTE_RELOAD, context=ContextType.MOUNT, reference="backup_share", uuid=uuid4(), auto=False, reference_extra=None, ), Suggestion( type=SuggestionType.EXECUTE_REMOVE, context=ContextType.MOUNT, reference="backup_share", uuid=uuid4(), auto=False, reference_extra=None, ), ] }, ) assert await async_setup_component(hass, DOMAIN, {}) repair_issue = issue_registry.async_get_issue( domain="hassio", issue_id=issue_uuid.hex ) assert repair_issue client = await hass_client() resp = await client.post( "/api/repairs/issues/fix", json={"handler": "hassio", "issue_id": repair_issue.issue_id}, ) assert resp.status == HTTPStatus.OK data = await resp.json() assert repair_issue.is_fixable is True assert data["type"] == "menu" assert data["menu_options"] == ["mount_execute_reload", "mount_execute_remove"] @pytest.mark.usefixtures("all_setup_requests") async def test_unfixable_issue_with_new_suggestion_stays_unfixable( hass: HomeAssistant, supervisor_client: AsyncMock, issue_registry: ir.IssueRegistry, ) -> None: """Test a suggestion on an issue without fix flow strings is dropped. This Core version shipped the issue as a never-fixable repair; a newer Supervisor adding suggestions must not turn it into a fixable repair that has no fix flow translations. """ mock_resolution_info( supervisor_client, issues=[ Issue( type=IssueType.DISK_LIFETIME, context=ContextType.SYSTEM, reference=None, uuid=(issue_uuid := uuid4()), reference_extra=None, ), ], suggestions_by_issue={ issue_uuid: [ Suggestion( type="suggestion_from_the_future", context=ContextType.SYSTEM, reference=None, uuid=uuid4(), auto=False, reference_extra=None, ), ] }, ) assert await async_setup_component(hass, DOMAIN, {}) repair_issue = issue_registry.async_get_issue( domain="hassio", issue_id=issue_uuid.hex ) assert repair_issue assert repair_issue.is_fixable is False @pytest.mark.usefixtures("all_setup_requests") async def test_mount_failed_repair_all_untranslated_suggestions_kept( hass: HomeAssistant, supervisor_client: AsyncMock, hass_client: ClientSessionGenerator, issue_registry: ir.IssueRegistry, ) -> None: """Test the menu keeps all suggestions if none has a translation. A repair is either always fixable or never fixable per issue key, so an unfixable state cannot be expressed for a key with a fix flow — the raw option keys are shown instead of a dead end. """ mock_resolution_info( supervisor_client, issues=[ Issue( type=IssueType.MOUNT_FAILED, context=ContextType.MOUNT, reference="backup_share", uuid=(issue_uuid := uuid4()), reference_extra=None, ), ], suggestions_by_issue={ issue_uuid: [ Suggestion( type="suggestion_from_the_future", context=ContextType.MOUNT, reference="backup_share", uuid=uuid4(), auto=False, reference_extra=None, ), Suggestion( type="other_future_suggestion", context=ContextType.MOUNT, reference="backup_share", uuid=uuid4(), auto=False, reference_extra=None, ), ] }, ) assert await async_setup_component(hass, DOMAIN, {}) repair_issue = issue_registry.async_get_issue( domain="hassio", issue_id=issue_uuid.hex ) assert repair_issue assert repair_issue.is_fixable is True client = await hass_client() resp = await client.post( "/api/repairs/issues/fix", json={"handler": "hassio", "issue_id": repair_issue.issue_id}, ) assert resp.status == HTTPStatus.OK data = await resp.json() assert data["type"] == "menu" assert data["menu_options"] == [ "mount_suggestion_from_the_future", "mount_other_future_suggestion", ] @pytest.mark.parametrize( "all_setup_requests", [{"include_addons": True}], indirect=True ) @pytest.mark.usefixtures("all_setup_requests") async def test_supervisor_issue_docker_config_repair_flow( hass: HomeAssistant, supervisor_client: AsyncMock, hass_client: ClientSessionGenerator, issue_registry: ir.IssueRegistry, ) -> None: """Test fix flow for supervisor issue.""" mock_resolution_info( supervisor_client, issues=[ Issue( type=IssueType.DOCKER_CONFIG, context=ContextType.SYSTEM, reference=None, uuid=(issue1_uuid := uuid4()), reference_extra=None, ), Issue( type=IssueType.DOCKER_CONFIG, context=ContextType.CORE, reference=None, uuid=(issue2_uuid := uuid4()), reference_extra=None, ), Issue( type=IssueType.DOCKER_CONFIG, context=ContextType.ADDON, reference="test", uuid=(issue3_uuid := uuid4()), reference_extra=None, ), ], suggestions_by_issue={ issue1_uuid: [ Suggestion( type=SuggestionType.EXECUTE_REBUILD, context=ContextType.SYSTEM, reference=None, uuid=(sugg_uuid := uuid4()), auto=False, reference_extra=None, ), ], issue2_uuid: [ Suggestion( type=SuggestionType.EXECUTE_REBUILD, context=ContextType.CORE, reference=None, uuid=uuid4(), auto=False, reference_extra=None, ), ], issue3_uuid: [ Suggestion( type=SuggestionType.EXECUTE_REBUILD, context=ContextType.ADDON, reference="test", uuid=uuid4(), auto=False, reference_extra=None, ), ], }, ) assert await async_setup_component(hass, DOMAIN, {}) repair_issue = issue_registry.async_get_issue( domain="hassio", issue_id=issue1_uuid.hex ) assert repair_issue client = await hass_client() resp = await client.post( "/api/repairs/issues/fix", json={"handler": "hassio", "issue_id": repair_issue.issue_id}, ) assert resp.status == HTTPStatus.OK data = await resp.json() flow_id = data["flow_id"] assert data == { "type": "form", "flow_id": flow_id, "handler": "hassio", "step_id": "system_execute_rebuild", "data_schema": [], "errors": None, "description_placeholders": {"components": "Home Assistant\n- test"}, "last_step": True, "preview": None, } resp = await client.post(f"/api/repairs/issues/fix/{flow_id}") assert resp.status == HTTPStatus.OK data = await resp.json() flow_id = data["flow_id"] assert data == { "type": "create_entry", "flow_id": flow_id, "handler": "hassio", "description": None, "description_placeholders": None, } assert not issue_registry.async_get_issue(domain="hassio", issue_id=issue1_uuid.hex) supervisor_client.resolution.apply_suggestion.assert_called_once_with(sugg_uuid) @pytest.mark.usefixtures("all_setup_requests") async def test_supervisor_issue_repair_flow_multiple_data_disks( hass: HomeAssistant, supervisor_client: AsyncMock, hass_client: ClientSessionGenerator, issue_registry: ir.IssueRegistry, ) -> None: """Test fix flow for multiple data disks supervisor issue.""" mock_resolution_info( supervisor_client, issues=[ Issue( type=IssueType.MULTIPLE_DATA_DISKS, context=ContextType.SYSTEM, reference="/dev/sda1", uuid=(issue_uuid := uuid4()), reference_extra=None, ), ], suggestions_by_issue={ issue_uuid: [ Suggestion( type=SuggestionType.RENAME_DATA_DISK, context=ContextType.SYSTEM, reference="/dev/sda1", uuid=uuid4(), auto=False, reference_extra=None, ), Suggestion( type=SuggestionType.ADOPT_DATA_DISK, context=ContextType.SYSTEM, reference="/dev/sda1", uuid=(sugg_uuid := uuid4()), auto=False, reference_extra=None, ), ] }, ) assert await async_setup_component(hass, DOMAIN, {}) repair_issue = issue_registry.async_get_issue( domain="hassio", issue_id=issue_uuid.hex ) assert repair_issue client = await hass_client() resp = await client.post( "/api/repairs/issues/fix", json={"handler": "hassio", "issue_id": repair_issue.issue_id}, ) assert resp.status == HTTPStatus.OK data = await resp.json() flow_id = data["flow_id"] assert data == { "type": "menu", "flow_id": flow_id, "handler": "hassio", "step_id": "fix_menu", "data_schema": [ { "type": "select", "options": [ ["system_rename_data_disk", "system_rename_data_disk"], ["system_adopt_data_disk", "system_adopt_data_disk"], ], "required": False, "name": "next_step_id", } ], "menu_options": ["system_rename_data_disk", "system_adopt_data_disk"], "description_placeholders": {"reference": "/dev/sda1"}, } resp = await client.post( f"/api/repairs/issues/fix/{flow_id}", json={"next_step_id": "system_adopt_data_disk"}, ) assert resp.status == HTTPStatus.OK data = await resp.json() flow_id = data["flow_id"] assert data == { "type": "form", "flow_id": flow_id, "handler": "hassio", "step_id": "system_adopt_data_disk", "data_schema": [], "errors": None, "description_placeholders": {"reference": "/dev/sda1"}, "last_step": True, "preview": None, } resp = await client.post(f"/api/repairs/issues/fix/{flow_id}") assert resp.status == HTTPStatus.OK data = await resp.json() flow_id = data["flow_id"] assert data == { "type": "create_entry", "flow_id": flow_id, "handler": "hassio", "description": None, "description_placeholders": None, } assert not issue_registry.async_get_issue(domain="hassio", issue_id=issue_uuid.hex) supervisor_client.resolution.apply_suggestion.assert_called_once_with(sugg_uuid) @pytest.mark.parametrize( "all_setup_requests", [{"include_addons": True}], indirect=True ) @pytest.mark.usefixtures("all_setup_requests") async def test_supervisor_issue_app_port_conflict_repair_flow_execute_start( hass: HomeAssistant, supervisor_client: AsyncMock, hass_client: ClientSessionGenerator, issue_registry: ir.IssueRegistry, ) -> None: """Test fix flow for app port conflict with single execute_start suggestion.""" mock_resolution_info( supervisor_client, issues=[ Issue( type="app_port_conflict", context=ContextType.ADDON, reference="test", uuid=(issue_uuid := uuid4()), reference_extra={"port": 11443}, ), ], suggestions_by_issue={ issue_uuid: [ Suggestion( type="execute_start", context=ContextType.ADDON, reference="test", uuid=(sugg_uuid := uuid4()), auto=False, reference_extra={"port": 11443}, ), ] }, ) assert await async_setup_component(hass, DOMAIN, {}) repair_issue = issue_registry.async_get_issue( domain="hassio", issue_id=issue_uuid.hex ) assert repair_issue client = await hass_client() resp = await client.post( "/api/repairs/issues/fix", json={"handler": "hassio", "issue_id": repair_issue.issue_id}, ) assert resp.status == HTTPStatus.OK data = await resp.json() flow_id = data["flow_id"] assert data == { "type": "form", "flow_id": flow_id, "handler": "hassio", "step_id": "addon_execute_start", "data_schema": [], "errors": None, "description_placeholders": { "reference": "test", "addon": "test", "port": "11443", }, "last_step": True, "preview": None, } resp = await client.post(f"/api/repairs/issues/fix/{flow_id}") assert resp.status == HTTPStatus.OK data = await resp.json() flow_id = data["flow_id"] assert data == { "type": "create_entry", "flow_id": flow_id, "handler": "hassio", "description": None, "description_placeholders": None, } assert not issue_registry.async_get_issue(domain="hassio", issue_id=issue_uuid.hex) supervisor_client.resolution.apply_suggestion.assert_called_once_with(sugg_uuid) @pytest.mark.parametrize( "all_setup_requests", [{"include_addons": True}], indirect=True ) @pytest.mark.usefixtures("all_setup_requests") async def test_supervisor_issue_app_port_conflict_repair_flow_menu( hass: HomeAssistant, supervisor_client: AsyncMock, hass_client: ClientSessionGenerator, issue_registry: ir.IssueRegistry, ) -> None: """Test fix flow for app port conflict with two suggestions showing menu.""" mock_resolution_info( supervisor_client, issues=[ Issue( type="app_port_conflict", context=ContextType.ADDON, reference="test", uuid=(issue_uuid := uuid4()), reference_extra={"port": 11443}, ), ], suggestions_by_issue={ issue_uuid: [ Suggestion( type="execute_start", context=ContextType.ADDON, reference="test", uuid=uuid4(), auto=False, reference_extra={"port": 11443}, ), Suggestion( type="clear_port_config", context=ContextType.ADDON, reference="test", uuid=(clear_config_uuid := uuid4()), auto=False, reference_extra={"port": 11443}, ), ] }, ) assert await async_setup_component(hass, DOMAIN, {}) repair_issue = issue_registry.async_get_issue( domain="hassio", issue_id=issue_uuid.hex ) assert repair_issue client = await hass_client() resp = await client.post( "/api/repairs/issues/fix", json={"handler": "hassio", "issue_id": repair_issue.issue_id}, ) assert resp.status == HTTPStatus.OK data = await resp.json() flow_id = data["flow_id"] assert data == { "type": "menu", "flow_id": flow_id, "handler": "hassio", "step_id": "fix_menu", "data_schema": [ { "type": "select", "options": [ ["addon_execute_start", "addon_execute_start"], ["addon_clear_port_config", "addon_clear_port_config"], ], "required": False, "name": "next_step_id", } ], "menu_options": ["addon_execute_start", "addon_clear_port_config"], "description_placeholders": { "reference": "test", "addon": "test", "port": "11443", }, } # Test clear_port_config path - automatically applies without confirmation resp = await client.post( f"/api/repairs/issues/fix/{flow_id}", json={"next_step_id": "addon_clear_port_config"}, ) assert resp.status == HTTPStatus.OK data = await resp.json() flow_id = data["flow_id"] # Since addon_clear_port_config does not require confirmation, it applies immediately assert data == { "type": "create_entry", "flow_id": flow_id, "handler": "hassio", "description": None, "description_placeholders": None, } assert not issue_registry.async_get_issue(domain="hassio", issue_id=issue_uuid.hex) supervisor_client.resolution.apply_suggestion.assert_called_once_with( clear_config_uuid ) @pytest.mark.parametrize( "all_setup_requests", [{"include_addons": True}], indirect=True ) @pytest.mark.usefixtures("all_setup_requests") async def test_supervisor_issue_detached_addon_removed( hass: HomeAssistant, supervisor_client: AsyncMock, hass_client: ClientSessionGenerator, issue_registry: ir.IssueRegistry, ) -> None: """Test fix flow for supervisor issue.""" mock_resolution_info( supervisor_client, issues=[ Issue( type=IssueType.DETACHED_ADDON_REMOVED, context=ContextType.ADDON, reference="test", uuid=(issue_uuid := uuid4()), reference_extra=None, ), ], suggestions_by_issue={ issue_uuid: [ Suggestion( type=SuggestionType.EXECUTE_REMOVE, context=ContextType.ADDON, reference="test", uuid=(sugg_uuid := uuid4()), auto=False, reference_extra=None, ), ] }, ) assert await async_setup_component(hass, DOMAIN, {}) repair_issue = issue_registry.async_get_issue( domain="hassio", issue_id=issue_uuid.hex ) assert repair_issue client = await hass_client() resp = await client.post( "/api/repairs/issues/fix", json={"handler": "hassio", "issue_id": repair_issue.issue_id}, ) assert resp.status == HTTPStatus.OK data = await resp.json() flow_id = data["flow_id"] assert data == { "type": "form", "flow_id": flow_id, "handler": "hassio", "step_id": "addon_execute_remove", "data_schema": [], "errors": None, "description_placeholders": { "reference": "test", "addon": "test", "help_url": "https://www.home-assistant.io/help/", "community_url": "https://community.home-assistant.io/", }, "last_step": True, "preview": None, } resp = await client.post(f"/api/repairs/issues/fix/{flow_id}") assert resp.status == HTTPStatus.OK data = await resp.json() flow_id = data["flow_id"] assert data == { "type": "create_entry", "flow_id": flow_id, "handler": "hassio", "description": None, "description_placeholders": None, } assert not issue_registry.async_get_issue(domain="hassio", issue_id=issue_uuid.hex) supervisor_client.resolution.apply_suggestion.assert_called_once_with(sugg_uuid) @pytest.mark.parametrize( "all_setup_requests", [{"include_addons": True}], indirect=True ) @pytest.mark.usefixtures("all_setup_requests") async def test_supervisor_issue_addon_boot_fail( hass: HomeAssistant, supervisor_client: AsyncMock, hass_client: ClientSessionGenerator, issue_registry: ir.IssueRegistry, ) -> None: """Test fix flow for supervisor issue.""" mock_resolution_info( supervisor_client, issues=[ Issue( type="boot_fail", context=ContextType.ADDON, reference="test", uuid=(issue_uuid := uuid4()), reference_extra=None, ), ], suggestions_by_issue={ issue_uuid: [ Suggestion( type="execute_start", context=ContextType.ADDON, reference="test", uuid=(sugg_uuid := uuid4()), auto=False, reference_extra=None, ), Suggestion( type="disable_boot", context=ContextType.ADDON, reference="test", uuid=uuid4(), auto=False, reference_extra=None, ), ] }, ) assert await async_setup_component(hass, DOMAIN, {}) repair_issue = issue_registry.async_get_issue( domain="hassio", issue_id=issue_uuid.hex ) assert repair_issue client = await hass_client() resp = await client.post( "/api/repairs/issues/fix", json={"handler": "hassio", "issue_id": repair_issue.issue_id}, ) assert resp.status == HTTPStatus.OK data = await resp.json() flow_id = data["flow_id"] assert data == { "type": "menu", "flow_id": flow_id, "handler": "hassio", "step_id": "fix_menu", "data_schema": [ { "type": "select", "options": [ ["addon_execute_start", "addon_execute_start"], ["addon_disable_boot", "addon_disable_boot"], ], "required": False, "name": "next_step_id", } ], "menu_options": ["addon_execute_start", "addon_disable_boot"], "description_placeholders": { "reference": "test", "addon": "test", }, } resp = await client.post( f"/api/repairs/issues/fix/{flow_id}", json={"next_step_id": "addon_execute_start"}, ) assert resp.status == HTTPStatus.OK data = await resp.json() flow_id = data["flow_id"] assert data == { "type": "create_entry", "flow_id": flow_id, "handler": "hassio", "description": None, "description_placeholders": None, } assert not issue_registry.async_get_issue(domain="hassio", issue_id=issue_uuid.hex) supervisor_client.resolution.apply_suggestion.assert_called_once_with(sugg_uuid) # Test disabled for now until repair can be re-enabled. First we need a repair # specifically for the OTBR add-on to make migration to ZHA easy rather then # having this repair encourage uninstall of that add-on and make migration hard. @pytest.mark.parametrize( "all_setup_requests", [{"include_addons": True}], indirect=True ) @pytest.mark.usefixtures("all_setup_requests") @pytest.mark.skip async def test_supervisor_issue_deprecated_addon( hass: HomeAssistant, supervisor_client: AsyncMock, hass_client: ClientSessionGenerator, issue_registry: ir.IssueRegistry, ) -> None: """Test fix flow for supervisor issue for deprecated add-on.""" mock_resolution_info( supervisor_client, issues=[ Issue( type=IssueType.DEPRECATED_ADDON, context=ContextType.ADDON, reference="test", uuid=(issue_uuid := uuid4()), reference_extra=None, ), ], suggestions_by_issue={ issue_uuid: [ Suggestion( type=SuggestionType.EXECUTE_REMOVE, context=ContextType.ADDON, reference="test", uuid=(sugg_uuid := uuid4()), auto=False, reference_extra=None, ), ] }, ) assert await async_setup_component(hass, DOMAIN, {}) repair_issue = issue_registry.async_get_issue( domain="hassio", issue_id=issue_uuid.hex ) assert repair_issue client = await hass_client() resp = await client.post( "/api/repairs/issues/fix", json={"handler": "hassio", "issue_id": repair_issue.issue_id}, ) assert resp.status == HTTPStatus.OK data = await resp.json() flow_id = data["flow_id"] assert data == { "type": "form", "flow_id": flow_id, "handler": "hassio", "step_id": "addon_execute_remove", "data_schema": [], "errors": None, "description_placeholders": { "reference": "test", "addon": "test", "help_url": "https://www.home-assistant.io/help/", "community_url": "https://community.home-assistant.io/", "addon_info": "homeassistant://hassio/addon/test/info", "addon_documentation": "homeassistant://hassio/addon/test/documentation", }, "last_step": True, "preview": None, } resp = await client.post(f"/api/repairs/issues/fix/{flow_id}") assert resp.status == HTTPStatus.OK data = await resp.json() flow_id = data["flow_id"] assert data == { "type": "create_entry", "flow_id": flow_id, "handler": "hassio", "description": None, "description_placeholders": None, } assert not issue_registry.async_get_issue(domain="hassio", issue_id=issue_uuid.hex) supervisor_client.resolution.apply_suggestion.assert_called_once_with(sugg_uuid) @pytest.mark.parametrize( "all_setup_requests", [{"include_addons": True}], indirect=True ) @pytest.mark.usefixtures("all_setup_requests") async def test_supervisor_issue_deprecated_arch_addon( hass: HomeAssistant, supervisor_client: AsyncMock, hass_client: ClientSessionGenerator, issue_registry: ir.IssueRegistry, ) -> None: """Test fix flow for supervisor issue for add-on with deprecated arch.""" mock_resolution_info( supervisor_client, issues=[ Issue( type=IssueType.DEPRECATED_ARCH_ADDON, context=ContextType.ADDON, reference="test", uuid=(issue_uuid := uuid4()), reference_extra=None, ), ], suggestions_by_issue={ issue_uuid: [ Suggestion( type=SuggestionType.EXECUTE_REMOVE, context=ContextType.ADDON, reference="test", uuid=(sugg_uuid := uuid4()), auto=False, reference_extra=None, ), ] }, ) assert await async_setup_component(hass, DOMAIN, {}) repair_issue = issue_registry.async_get_issue( domain="hassio", issue_id=issue_uuid.hex ) assert repair_issue client = await hass_client() resp = await client.post( "/api/repairs/issues/fix", json={"handler": "hassio", "issue_id": repair_issue.issue_id}, ) assert resp.status == HTTPStatus.OK data = await resp.json() flow_id = data["flow_id"] assert data == { "type": "form", "flow_id": flow_id, "handler": "hassio", "step_id": "addon_execute_remove", "data_schema": [], "errors": None, "description_placeholders": { "reference": "test", "addon": "test", "help_url": "https://www.home-assistant.io/help/", "community_url": "https://community.home-assistant.io/", }, "last_step": True, "preview": None, } resp = await client.post(f"/api/repairs/issues/fix/{flow_id}") assert resp.status == HTTPStatus.OK data = await resp.json() flow_id = data["flow_id"] assert data == { "type": "create_entry", "flow_id": flow_id, "handler": "hassio", "description": None, "description_placeholders": None, } assert not issue_registry.async_get_issue(domain="hassio", issue_id=issue_uuid.hex) supervisor_client.resolution.apply_suggestion.assert_called_once_with(sugg_uuid)