mirror of
				https://github.com/home-assistant/supervisor.git
				synced 2025-10-31 06:29:35 +00:00 
			
		
		
		
	Compare commits
	
		
			3 Commits
		
	
	
		
			copilot/ad
			...
			remove-dep
		
	
	| Author | SHA1 | Date | |
|---|---|---|---|
|   | 66a3766b5a | ||
|   | 7031a58083 | ||
|   | 3c0e62f6ba | 
							
								
								
									
										4
									
								
								.github/workflows/builder.yml
									
									
									
									
										vendored
									
									
								
							
							
						
						
									
										4
									
								
								.github/workflows/builder.yml
									
									
									
									
										vendored
									
									
								
							| @@ -107,7 +107,7 @@ jobs: | ||||
|       # home-assistant/wheels doesn't support sha pinning | ||||
|       - name: Build wheels | ||||
|         if: needs.init.outputs.requirements == 'true' | ||||
|         uses: home-assistant/wheels@2025.10.0 | ||||
|         uses: home-assistant/wheels@2025.09.1 | ||||
|         with: | ||||
|           abi: cp313 | ||||
|           tag: musllinux_1_2 | ||||
| @@ -132,7 +132,7 @@ jobs: | ||||
|  | ||||
|       - name: Install Cosign | ||||
|         if: needs.init.outputs.publish == 'true' | ||||
|         uses: sigstore/cosign-installer@faadad0cce49287aee09b3a48701e75088a2c6ad # v4.0.0 | ||||
|         uses: sigstore/cosign-installer@d7543c93d881b35a8faa02e8e3605f69b7a1ce62 # v3.10.0 | ||||
|         with: | ||||
|           cosign-release: "v2.5.3" | ||||
|  | ||||
|   | ||||
							
								
								
									
										6
									
								
								.github/workflows/ci.yaml
									
									
									
									
										vendored
									
									
								
							
							
						
						
									
										6
									
								
								.github/workflows/ci.yaml
									
									
									
									
										vendored
									
									
								
							| @@ -346,7 +346,7 @@ jobs: | ||||
|         with: | ||||
|           python-version: ${{ needs.prepare.outputs.python-version }} | ||||
|       - name: Install Cosign | ||||
|         uses: sigstore/cosign-installer@faadad0cce49287aee09b3a48701e75088a2c6ad # v4.0.0 | ||||
|         uses: sigstore/cosign-installer@d7543c93d881b35a8faa02e8e3605f69b7a1ce62 # v3.10.0 | ||||
|         with: | ||||
|           cosign-release: "v2.5.3" | ||||
|       - name: Restore Python virtual environment | ||||
| @@ -386,7 +386,7 @@ jobs: | ||||
|             -o console_output_style=count \ | ||||
|             tests | ||||
|       - name: Upload coverage artifact | ||||
|         uses: actions/upload-artifact@330a01c490aca151604b8cf639adc76d48f6c5d4 # v5.0.0 | ||||
|         uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # v4.6.2 | ||||
|         with: | ||||
|           name: coverage | ||||
|           path: .coverage | ||||
| @@ -417,7 +417,7 @@ jobs: | ||||
|           echo "Failed to restore Python virtual environment from cache" | ||||
|           exit 1 | ||||
|       - name: Download all coverage artifacts | ||||
|         uses: actions/download-artifact@018cc2cf5baa6db3ef3c5f8a56943fffe632ef53 # v6.0.0 | ||||
|         uses: actions/download-artifact@634f93cb2916e3fdff6788551b99b062d0335ce0 # v5.0.0 | ||||
|         with: | ||||
|           name: coverage | ||||
|           path: coverage/ | ||||
|   | ||||
							
								
								
									
										1
									
								
								.github/workflows/stale.yml
									
									
									
									
										vendored
									
									
								
							
							
						
						
									
										1
									
								
								.github/workflows/stale.yml
									
									
									
									
										vendored
									
									
								
							| @@ -16,7 +16,6 @@ jobs: | ||||
|           days-before-close: 7 | ||||
|           stale-issue-label: "stale" | ||||
|           exempt-issue-labels: "no-stale,Help%20wanted,help-wanted,pinned,rfc,security" | ||||
|           only-issue-types: "bug" | ||||
|           stale-issue-message: > | ||||
|             There hasn't been any activity on this issue recently. Due to the | ||||
|             high number of incoming GitHub notifications, we have to clean some | ||||
|   | ||||
| @@ -1,14 +1,14 @@ | ||||
| aiodns==3.5.0 | ||||
| aiohttp==3.13.2 | ||||
| aiohttp==3.13.0 | ||||
| atomicwrites-homeassistant==1.4.1 | ||||
| attrs==25.4.0 | ||||
| awesomeversion==25.8.0 | ||||
| blockbuster==1.5.25 | ||||
| brotli==1.1.0 | ||||
| ciso8601==2.3.3 | ||||
| colorlog==6.10.1 | ||||
| colorlog==6.9.0 | ||||
| cpe==1.3.1 | ||||
| cryptography==46.0.3 | ||||
| cryptography==46.0.2 | ||||
| debugpy==1.8.17 | ||||
| deepmerge==2.0 | ||||
| dirhash==0.5.0 | ||||
| @@ -17,13 +17,13 @@ faust-cchardet==2.1.19 | ||||
| gitpython==3.1.45 | ||||
| jinja2==3.1.6 | ||||
| log-rate-limit==1.4.2 | ||||
| orjson==3.11.4 | ||||
| orjson==3.11.3 | ||||
| pulsectl==24.12.0 | ||||
| pyudev==0.24.4 | ||||
| pyudev==0.24.3 | ||||
| PyYAML==6.0.3 | ||||
| requests==2.32.5 | ||||
| securetar==2025.2.1 | ||||
| sentry-sdk==2.42.1 | ||||
| sentry-sdk==2.40.0 | ||||
| setuptools==80.9.0 | ||||
| voluptuous==0.15.2 | ||||
| dbus-fast==2.44.5 | ||||
|   | ||||
| @@ -1,16 +1,16 @@ | ||||
| astroid==4.0.1 | ||||
| coverage==7.11.0 | ||||
| astroid==3.3.11 | ||||
| coverage==7.10.7 | ||||
| mypy==1.18.2 | ||||
| pre-commit==4.3.0 | ||||
| pylint==4.0.2 | ||||
| pylint==3.3.9 | ||||
| pytest-aiohttp==1.1.0 | ||||
| pytest-asyncio==0.25.2 | ||||
| pytest-cov==7.0.0 | ||||
| pytest-timeout==2.4.0 | ||||
| pytest==8.4.2 | ||||
| ruff==0.14.2 | ||||
| ruff==0.14.0 | ||||
| time-machine==2.19.0 | ||||
| types-docker==7.1.0.20251009 | ||||
| types-docker==7.1.0.20250916 | ||||
| types-pyyaml==6.0.12.20250915 | ||||
| types-requests==2.32.4.20250913 | ||||
| urllib3==2.5.0 | ||||
|   | ||||
| @@ -1562,15 +1562,7 @@ class Addon(AddonModel): | ||||
|                 ) | ||||
|                 break | ||||
|  | ||||
|             # Exponential backoff to spread retries over the throttle window | ||||
|             delay = WATCHDOG_RETRY_SECONDS * (1 << max(attempts - 1, 0)) | ||||
|             _LOGGER.debug( | ||||
|                 "Watchdog will retry addon %s in %s seconds (attempt %s)", | ||||
|                 self.name, | ||||
|                 delay, | ||||
|                 attempts + 1, | ||||
|             ) | ||||
|             await asyncio.sleep(delay) | ||||
|             await asyncio.sleep(WATCHDOG_RETRY_SECONDS) | ||||
|  | ||||
|     async def container_state_changed(self, event: DockerContainerStateEvent) -> None: | ||||
|         """Set addon state from container state.""" | ||||
|   | ||||
| @@ -108,8 +108,7 @@ class APISupervisor(CoreSysAttributes): | ||||
|             ATTR_AUTO_UPDATE: self.sys_updater.auto_update, | ||||
|             ATTR_DETECT_BLOCKING_IO: BlockBusterManager.is_enabled(), | ||||
|             ATTR_COUNTRY: self.sys_config.country, | ||||
|             # Depricated | ||||
|             ATTR_WAIT_BOOT: self.sys_config.wait_boot, | ||||
|             # Deprecated | ||||
|             ATTR_ADDONS: [ | ||||
|                 { | ||||
|                     ATTR_NAME: addon.name, | ||||
| @@ -123,10 +122,6 @@ class APISupervisor(CoreSysAttributes): | ||||
|                 } | ||||
|                 for addon in self.sys_addons.local.values() | ||||
|             ], | ||||
|             ATTR_ADDONS_REPOSITORIES: [ | ||||
|                 {ATTR_NAME: store.name, ATTR_SLUG: store.slug} | ||||
|                 for store in self.sys_store.all | ||||
|             ], | ||||
|         } | ||||
|  | ||||
|     @api_process | ||||
| @@ -182,20 +177,10 @@ class APISupervisor(CoreSysAttributes): | ||||
|                 self.sys_config.detect_blocking_io = False | ||||
|                 BlockBusterManager.deactivate() | ||||
|  | ||||
|         # Deprecated | ||||
|         if ATTR_WAIT_BOOT in body: | ||||
|             self.sys_config.wait_boot = body[ATTR_WAIT_BOOT] | ||||
|  | ||||
|         # Save changes before processing addons in case of errors | ||||
|         await self.sys_updater.save_data() | ||||
|         await self.sys_config.save_data() | ||||
|  | ||||
|         # Remove: 2022.9 | ||||
|         if ATTR_ADDONS_REPOSITORIES in body: | ||||
|             await asyncio.shield( | ||||
|                 self.sys_store.update_repositories(set(body[ATTR_ADDONS_REPOSITORIES])) | ||||
|             ) | ||||
|  | ||||
|         await self.sys_resolution.evaluate.evaluate_system() | ||||
|  | ||||
|     @api_process | ||||
|   | ||||
| @@ -306,8 +306,6 @@ class DockerInterface(JobGroup, ABC): | ||||
|         # Our filters have all passed. Time to update the job | ||||
|         # Only downloading and extracting have progress details. Use that to set extra | ||||
|         # We'll leave it around on later stages as the total bytes may be useful after that stage | ||||
|         # Enforce range to prevent float drift error | ||||
|         progress = max(0, min(progress, 100)) | ||||
|         if ( | ||||
|             stage in {PullImageLayerStage.DOWNLOADING, PullImageLayerStage.EXTRACTING} | ||||
|             and reference.progress_detail | ||||
| @@ -373,7 +371,7 @@ class DockerInterface(JobGroup, ABC): | ||||
|  | ||||
|         # To reduce noise, limit updates to when result has changed by an entire percent or when stage changed | ||||
|         if stage != install_job.stage or progress >= install_job.progress + 1: | ||||
|             install_job.update(stage=stage.status, progress=max(0, min(progress, 100))) | ||||
|             install_job.update(stage=stage.status, progress=progress) | ||||
|  | ||||
|     @Job( | ||||
|         name="docker_interface_install", | ||||
|   | ||||
| @@ -9,12 +9,7 @@ from typing import Any | ||||
| from supervisor.resolution.const import UnhealthyReason | ||||
|  | ||||
| from ..coresys import CoreSys, CoreSysAttributes | ||||
| from ..exceptions import ( | ||||
|     DBusError, | ||||
|     DBusNotConnectedError, | ||||
|     DBusObjectError, | ||||
|     HardwareNotFound, | ||||
| ) | ||||
| from ..exceptions import DBusError, DBusObjectError, HardwareNotFound | ||||
| from .const import UdevSubsystem | ||||
| from .data import Device | ||||
|  | ||||
| @@ -212,8 +207,6 @@ class HwDisk(CoreSysAttributes): | ||||
|         try: | ||||
|             block_device = self.sys_dbus.udisks2.get_block_device_by_path(device_path) | ||||
|             drive = self.sys_dbus.udisks2.get_drive(block_device.drive) | ||||
|         except DBusNotConnectedError: | ||||
|             return None | ||||
|         except DBusObjectError: | ||||
|             _LOGGER.warning( | ||||
|                 "Unable to find UDisks2 drive for device at %s", device_path.as_posix() | ||||
|   | ||||
| @@ -8,7 +8,7 @@ from ..const import UnsupportedReason | ||||
| from .base import EvaluateBase | ||||
|  | ||||
| EXPECTED_LOGGING = "journald" | ||||
| EXPECTED_STORAGE = ("overlay2", "overlayfs") | ||||
| EXPECTED_STORAGE = "overlay2" | ||||
|  | ||||
| _LOGGER: logging.Logger = logging.getLogger(__name__) | ||||
|  | ||||
| @@ -41,18 +41,14 @@ class EvaluateDockerConfiguration(EvaluateBase): | ||||
|         storage_driver = self.sys_docker.info.storage | ||||
|         logging_driver = self.sys_docker.info.logging | ||||
|  | ||||
|         is_unsupported = False | ||||
|  | ||||
|         if storage_driver not in EXPECTED_STORAGE: | ||||
|             is_unsupported = True | ||||
|         if storage_driver != EXPECTED_STORAGE: | ||||
|             _LOGGER.warning( | ||||
|                 "Docker storage driver %s is not supported!", storage_driver | ||||
|             ) | ||||
|  | ||||
|         if logging_driver != EXPECTED_LOGGING: | ||||
|             is_unsupported = True | ||||
|             _LOGGER.warning( | ||||
|                 "Docker logging driver %s is not supported!", logging_driver | ||||
|             ) | ||||
|  | ||||
|         return is_unsupported | ||||
|         return storage_driver != EXPECTED_STORAGE or logging_driver != EXPECTED_LOGGING | ||||
|   | ||||
| @@ -12,9 +12,8 @@ import pytest | ||||
| from supervisor.const import CoreState | ||||
| from supervisor.core import Core | ||||
| from supervisor.coresys import CoreSys | ||||
| from supervisor.exceptions import HassioError, HostNotSupportedError, StoreGitError | ||||
| from supervisor.exceptions import HassioError, HostNotSupportedError | ||||
| from supervisor.homeassistant.const import WSEvent | ||||
| from supervisor.store.repository import Repository | ||||
| from supervisor.supervisor import Supervisor | ||||
| from supervisor.updater import Updater | ||||
|  | ||||
| @@ -35,81 +34,6 @@ async def test_api_supervisor_options_debug(api_client: TestClient, coresys: Cor | ||||
|     assert coresys.config.debug | ||||
|  | ||||
|  | ||||
| async def test_api_supervisor_options_add_repository( | ||||
|     api_client: TestClient, coresys: CoreSys, supervisor_internet: AsyncMock | ||||
| ): | ||||
|     """Test add a repository via POST /supervisor/options REST API.""" | ||||
|     assert REPO_URL not in coresys.store.repository_urls | ||||
|  | ||||
|     with ( | ||||
|         patch("supervisor.store.repository.RepositoryGit.load", return_value=None), | ||||
|         patch("supervisor.store.repository.RepositoryGit.validate", return_value=True), | ||||
|     ): | ||||
|         response = await api_client.post( | ||||
|             "/supervisor/options", json={"addons_repositories": [REPO_URL]} | ||||
|         ) | ||||
|  | ||||
|     assert response.status == 200 | ||||
|     assert REPO_URL in coresys.store.repository_urls | ||||
|  | ||||
|  | ||||
| async def test_api_supervisor_options_remove_repository( | ||||
|     api_client: TestClient, coresys: CoreSys, test_repository: Repository | ||||
| ): | ||||
|     """Test remove a repository via POST /supervisor/options REST API.""" | ||||
|     assert test_repository.source in coresys.store.repository_urls | ||||
|     assert test_repository.slug in coresys.store.repositories | ||||
|  | ||||
|     response = await api_client.post( | ||||
|         "/supervisor/options", json={"addons_repositories": []} | ||||
|     ) | ||||
|  | ||||
|     assert response.status == 200 | ||||
|     assert test_repository.source not in coresys.store.repository_urls | ||||
|     assert test_repository.slug not in coresys.store.repositories | ||||
|  | ||||
|  | ||||
| @pytest.mark.parametrize("git_error", [None, StoreGitError()]) | ||||
| async def test_api_supervisor_options_repositories_skipped_on_error( | ||||
|     api_client: TestClient, coresys: CoreSys, git_error: StoreGitError | ||||
| ): | ||||
|     """Test repositories skipped on error via POST /supervisor/options REST API.""" | ||||
|     with ( | ||||
|         patch("supervisor.store.repository.RepositoryGit.load", side_effect=git_error), | ||||
|         patch("supervisor.store.repository.RepositoryGit.validate", return_value=False), | ||||
|         patch("supervisor.store.repository.RepositoryCustom.remove"), | ||||
|     ): | ||||
|         response = await api_client.post( | ||||
|             "/supervisor/options", json={"addons_repositories": [REPO_URL]} | ||||
|         ) | ||||
|  | ||||
|     assert response.status == 400 | ||||
|     assert len(coresys.resolution.suggestions) == 0 | ||||
|     assert REPO_URL not in coresys.store.repository_urls | ||||
|  | ||||
|  | ||||
| async def test_api_supervisor_options_repo_error_with_config_change( | ||||
|     api_client: TestClient, coresys: CoreSys | ||||
| ): | ||||
|     """Test config change with add repository error via POST /supervisor/options REST API.""" | ||||
|     assert not coresys.config.debug | ||||
|  | ||||
|     with patch( | ||||
|         "supervisor.store.repository.RepositoryGit.load", side_effect=StoreGitError() | ||||
|     ): | ||||
|         response = await api_client.post( | ||||
|             "/supervisor/options", | ||||
|             json={"debug": True, "addons_repositories": [REPO_URL]}, | ||||
|         ) | ||||
|  | ||||
|     assert response.status == 400 | ||||
|     assert REPO_URL not in coresys.store.repository_urls | ||||
|  | ||||
|     assert coresys.config.debug | ||||
|     coresys.updater.save_data.assert_called_once() | ||||
|     coresys.config.save_data.assert_called_once() | ||||
|  | ||||
|  | ||||
| async def test_api_supervisor_options_auto_update( | ||||
|     api_client: TestClient, coresys: CoreSys | ||||
| ): | ||||
|   | ||||
| @@ -376,14 +376,3 @@ async def test_try_get_nvme_life_time_missing_percent_used( | ||||
|         coresys.config.path_supervisor | ||||
|     ) | ||||
|     assert lifetime is None | ||||
|  | ||||
|  | ||||
| async def test_try_get_nvme_life_time_dbus_not_connected(coresys: CoreSys): | ||||
|     """Test getting lifetime info from an NVMe when DBUS is not connected.""" | ||||
|     # Set the dbus for udisks2 bus to be None, to make it forcibly disconnected. | ||||
|     coresys.dbus.udisks2.dbus = None | ||||
|  | ||||
|     lifetime = await coresys.hardware.disk.get_disk_life_time( | ||||
|         coresys.config.path_supervisor | ||||
|     ) | ||||
|     assert lifetime is None | ||||
|   | ||||
| @@ -7,8 +7,8 @@ import pytest | ||||
|  | ||||
| from supervisor.coresys import CoreSys | ||||
| from supervisor.dbus.const import DeviceType | ||||
| from supervisor.host.configuration import Interface, VlanConfig, WifiConfig | ||||
| from supervisor.host.const import AuthMethod, InterfaceType, WifiMode | ||||
| from supervisor.host.configuration import Interface, VlanConfig | ||||
| from supervisor.host.const import InterfaceType | ||||
|  | ||||
| from tests.dbus_service_mocks.base import DBusServiceMock | ||||
| from tests.dbus_service_mocks.network_connection_settings import ( | ||||
| @@ -291,237 +291,3 @@ async def test_equals_dbus_interface_eth0_10_real( | ||||
|  | ||||
|     # Test should pass with matching VLAN config | ||||
|     assert test_vlan_interface.equals_dbus_interface(network_interface) is True | ||||
|  | ||||
|  | ||||
| def test_map_nm_wifi_non_wireless_interface(): | ||||
|     """Test _map_nm_wifi returns None for non-wireless interface.""" | ||||
|     # Mock non-wireless interface | ||||
|     mock_interface = Mock() | ||||
|     mock_interface.type = DeviceType.ETHERNET | ||||
|     mock_interface.settings = Mock() | ||||
|  | ||||
|     result = Interface._map_nm_wifi(mock_interface) | ||||
|     assert result is None | ||||
|  | ||||
|  | ||||
| def test_map_nm_wifi_no_settings(): | ||||
|     """Test _map_nm_wifi returns None when interface has no settings.""" | ||||
|     # Mock wireless interface without settings | ||||
|     mock_interface = Mock() | ||||
|     mock_interface.type = DeviceType.WIRELESS | ||||
|     mock_interface.settings = None | ||||
|  | ||||
|     result = Interface._map_nm_wifi(mock_interface) | ||||
|     assert result is None | ||||
|  | ||||
|  | ||||
| def test_map_nm_wifi_open_authentication(): | ||||
|     """Test _map_nm_wifi with open authentication (no security).""" | ||||
|     # Mock wireless interface with open authentication | ||||
|     mock_interface = Mock() | ||||
|     mock_interface.type = DeviceType.WIRELESS | ||||
|     mock_interface.settings = Mock() | ||||
|     mock_interface.settings.wireless_security = None | ||||
|     mock_interface.settings.wireless = Mock() | ||||
|     mock_interface.settings.wireless.ssid = "TestSSID" | ||||
|     mock_interface.settings.wireless.mode = "infrastructure" | ||||
|     mock_interface.wireless = None | ||||
|     mock_interface.interface_name = "wlan0" | ||||
|  | ||||
|     result = Interface._map_nm_wifi(mock_interface) | ||||
|  | ||||
|     assert result is not None | ||||
|     assert isinstance(result, WifiConfig) | ||||
|     assert result.mode == WifiMode.INFRASTRUCTURE | ||||
|     assert result.ssid == "TestSSID" | ||||
|     assert result.auth == AuthMethod.OPEN | ||||
|     assert result.psk is None | ||||
|     assert result.signal is None | ||||
|  | ||||
|  | ||||
| def test_map_nm_wifi_wep_authentication(): | ||||
|     """Test _map_nm_wifi with WEP authentication.""" | ||||
|     # Mock wireless interface with WEP authentication | ||||
|     mock_interface = Mock() | ||||
|     mock_interface.type = DeviceType.WIRELESS | ||||
|     mock_interface.settings = Mock() | ||||
|     mock_interface.settings.wireless_security = Mock() | ||||
|     mock_interface.settings.wireless_security.key_mgmt = "none" | ||||
|     mock_interface.settings.wireless_security.psk = None | ||||
|     mock_interface.settings.wireless = Mock() | ||||
|     mock_interface.settings.wireless.ssid = "WEPNetwork" | ||||
|     mock_interface.settings.wireless.mode = "infrastructure" | ||||
|     mock_interface.wireless = None | ||||
|     mock_interface.interface_name = "wlan0" | ||||
|  | ||||
|     result = Interface._map_nm_wifi(mock_interface) | ||||
|  | ||||
|     assert result is not None | ||||
|     assert isinstance(result, WifiConfig) | ||||
|     assert result.auth == AuthMethod.WEP | ||||
|     assert result.ssid == "WEPNetwork" | ||||
|     assert result.psk is None | ||||
|  | ||||
|  | ||||
| def test_map_nm_wifi_wpa_psk_authentication(): | ||||
|     """Test _map_nm_wifi with WPA-PSK authentication.""" | ||||
|     # Mock wireless interface with WPA-PSK authentication | ||||
|     mock_interface = Mock() | ||||
|     mock_interface.type = DeviceType.WIRELESS | ||||
|     mock_interface.settings = Mock() | ||||
|     mock_interface.settings.wireless_security = Mock() | ||||
|     mock_interface.settings.wireless_security.key_mgmt = "wpa-psk" | ||||
|     mock_interface.settings.wireless_security.psk = "SecretPassword123" | ||||
|     mock_interface.settings.wireless = Mock() | ||||
|     mock_interface.settings.wireless.ssid = "SecureNetwork" | ||||
|     mock_interface.settings.wireless.mode = "infrastructure" | ||||
|     mock_interface.wireless = None | ||||
|     mock_interface.interface_name = "wlan0" | ||||
|  | ||||
|     result = Interface._map_nm_wifi(mock_interface) | ||||
|  | ||||
|     assert result is not None | ||||
|     assert isinstance(result, WifiConfig) | ||||
|     assert result.auth == AuthMethod.WPA_PSK | ||||
|     assert result.ssid == "SecureNetwork" | ||||
|     assert result.psk == "SecretPassword123" | ||||
|  | ||||
|  | ||||
| def test_map_nm_wifi_unsupported_authentication(): | ||||
|     """Test _map_nm_wifi returns None for unsupported authentication method.""" | ||||
|     # Mock wireless interface with unsupported authentication | ||||
|     mock_interface = Mock() | ||||
|     mock_interface.type = DeviceType.WIRELESS | ||||
|     mock_interface.settings = Mock() | ||||
|     mock_interface.settings.wireless_security = Mock() | ||||
|     mock_interface.settings.wireless_security.key_mgmt = "wpa-eap"  # Unsupported | ||||
|     mock_interface.settings.wireless = Mock() | ||||
|     mock_interface.settings.wireless.ssid = "EnterpriseNetwork" | ||||
|     mock_interface.interface_name = "wlan0" | ||||
|  | ||||
|     result = Interface._map_nm_wifi(mock_interface) | ||||
|  | ||||
|     assert result is None | ||||
|  | ||||
|  | ||||
| def test_map_nm_wifi_different_modes(): | ||||
|     """Test _map_nm_wifi with different wifi modes.""" | ||||
|     modes_to_test = [ | ||||
|         ("infrastructure", WifiMode.INFRASTRUCTURE), | ||||
|         ("mesh", WifiMode.MESH), | ||||
|         ("adhoc", WifiMode.ADHOC), | ||||
|         ("ap", WifiMode.AP), | ||||
|     ] | ||||
|  | ||||
|     for mode_value, expected_mode in modes_to_test: | ||||
|         mock_interface = Mock() | ||||
|         mock_interface.type = DeviceType.WIRELESS | ||||
|         mock_interface.settings = Mock() | ||||
|         mock_interface.settings.wireless_security = None | ||||
|         mock_interface.settings.wireless = Mock() | ||||
|         mock_interface.settings.wireless.ssid = "TestSSID" | ||||
|         mock_interface.settings.wireless.mode = mode_value | ||||
|         mock_interface.wireless = None | ||||
|         mock_interface.interface_name = "wlan0" | ||||
|  | ||||
|         result = Interface._map_nm_wifi(mock_interface) | ||||
|  | ||||
|         assert result is not None | ||||
|         assert result.mode == expected_mode | ||||
|  | ||||
|  | ||||
| def test_map_nm_wifi_with_signal(): | ||||
|     """Test _map_nm_wifi with wireless signal strength.""" | ||||
|     # Mock wireless interface with active connection and signal | ||||
|     mock_interface = Mock() | ||||
|     mock_interface.type = DeviceType.WIRELESS | ||||
|     mock_interface.settings = Mock() | ||||
|     mock_interface.settings.wireless_security = None | ||||
|     mock_interface.settings.wireless = Mock() | ||||
|     mock_interface.settings.wireless.ssid = "TestSSID" | ||||
|     mock_interface.settings.wireless.mode = "infrastructure" | ||||
|     mock_interface.wireless = Mock() | ||||
|     mock_interface.wireless.active = Mock() | ||||
|     mock_interface.wireless.active.strength = 75 | ||||
|     mock_interface.interface_name = "wlan0" | ||||
|  | ||||
|     result = Interface._map_nm_wifi(mock_interface) | ||||
|  | ||||
|     assert result is not None | ||||
|     assert result.signal == 75 | ||||
|  | ||||
|  | ||||
| def test_map_nm_wifi_without_signal(): | ||||
|     """Test _map_nm_wifi without wireless signal (no active connection).""" | ||||
|     # Mock wireless interface without active connection | ||||
|     mock_interface = Mock() | ||||
|     mock_interface.type = DeviceType.WIRELESS | ||||
|     mock_interface.settings = Mock() | ||||
|     mock_interface.settings.wireless_security = None | ||||
|     mock_interface.settings.wireless = Mock() | ||||
|     mock_interface.settings.wireless.ssid = "TestSSID" | ||||
|     mock_interface.settings.wireless.mode = "infrastructure" | ||||
|     mock_interface.wireless = None | ||||
|     mock_interface.interface_name = "wlan0" | ||||
|  | ||||
|     result = Interface._map_nm_wifi(mock_interface) | ||||
|  | ||||
|     assert result is not None | ||||
|     assert result.signal is None | ||||
|  | ||||
|  | ||||
| def test_map_nm_wifi_wireless_no_active_ap(): | ||||
|     """Test _map_nm_wifi with wireless object but no active access point.""" | ||||
|     # Mock wireless interface with wireless object but no active AP | ||||
|     mock_interface = Mock() | ||||
|     mock_interface.type = DeviceType.WIRELESS | ||||
|     mock_interface.settings = Mock() | ||||
|     mock_interface.settings.wireless_security = None | ||||
|     mock_interface.settings.wireless = Mock() | ||||
|     mock_interface.settings.wireless.ssid = "TestSSID" | ||||
|     mock_interface.settings.wireless.mode = "infrastructure" | ||||
|     mock_interface.wireless = Mock() | ||||
|     mock_interface.wireless.active = None | ||||
|     mock_interface.interface_name = "wlan0" | ||||
|  | ||||
|     result = Interface._map_nm_wifi(mock_interface) | ||||
|  | ||||
|     assert result is not None | ||||
|     assert result.signal is None | ||||
|  | ||||
|  | ||||
| def test_map_nm_wifi_no_wireless_settings(): | ||||
|     """Test _map_nm_wifi when wireless settings are missing.""" | ||||
|     # Mock wireless interface without wireless settings | ||||
|     mock_interface = Mock() | ||||
|     mock_interface.type = DeviceType.WIRELESS | ||||
|     mock_interface.settings = Mock() | ||||
|     mock_interface.settings.wireless_security = None | ||||
|     mock_interface.settings.wireless = None | ||||
|     mock_interface.wireless = None | ||||
|     mock_interface.interface_name = "wlan0" | ||||
|  | ||||
|     result = Interface._map_nm_wifi(mock_interface) | ||||
|  | ||||
|     assert result is not None | ||||
|     assert result.ssid == "" | ||||
|     assert result.mode == WifiMode.INFRASTRUCTURE  # Default mode | ||||
|  | ||||
|  | ||||
| def test_map_nm_wifi_no_wireless_mode(): | ||||
|     """Test _map_nm_wifi when wireless mode is not specified.""" | ||||
|     # Mock wireless interface without mode specified | ||||
|     mock_interface = Mock() | ||||
|     mock_interface.type = DeviceType.WIRELESS | ||||
|     mock_interface.settings = Mock() | ||||
|     mock_interface.settings.wireless_security = None | ||||
|     mock_interface.settings.wireless = Mock() | ||||
|     mock_interface.settings.wireless.ssid = "TestSSID" | ||||
|     mock_interface.settings.wireless.mode = None | ||||
|     mock_interface.wireless = None | ||||
|     mock_interface.interface_name = "wlan0" | ||||
|  | ||||
|     result = Interface._map_nm_wifi(mock_interface) | ||||
|  | ||||
|     assert result is not None | ||||
|     assert result.mode == WifiMode.INFRASTRUCTURE  # Default mode | ||||
|   | ||||
| @@ -25,18 +25,13 @@ async def test_evaluation(coresys: CoreSys): | ||||
|     assert docker_configuration.reason in coresys.resolution.unsupported | ||||
|     coresys.resolution.unsupported.clear() | ||||
|  | ||||
|     coresys.docker.info.storage = EXPECTED_STORAGE[0] | ||||
|     coresys.docker.info.storage = EXPECTED_STORAGE | ||||
|     coresys.docker.info.logging = "unsupported" | ||||
|     await docker_configuration() | ||||
|     assert docker_configuration.reason in coresys.resolution.unsupported | ||||
|     coresys.resolution.unsupported.clear() | ||||
|  | ||||
|     coresys.docker.info.storage = "overlay2" | ||||
|     coresys.docker.info.logging = EXPECTED_LOGGING | ||||
|     await docker_configuration() | ||||
|     assert docker_configuration.reason not in coresys.resolution.unsupported | ||||
|  | ||||
|     coresys.docker.info.storage = "overlayfs" | ||||
|     coresys.docker.info.storage = EXPECTED_STORAGE | ||||
|     coresys.docker.info.logging = EXPECTED_LOGGING | ||||
|     await docker_configuration() | ||||
|     assert docker_configuration.reason not in coresys.resolution.unsupported | ||||
|   | ||||
		Reference in New Issue
	
	Block a user