mirror of
https://github.com/home-assistant/supervisor.git
synced 2025-11-04 16:39:33 +00:00
Compare commits
3 Commits
2025.11.0
...
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,6 +1,6 @@
|
||||
repos:
|
||||
- repo: https://github.com/astral-sh/ruff-pre-commit
|
||||
rev: v0.14.3
|
||||
rev: v0.11.10
|
||||
hooks:
|
||||
- id: ruff
|
||||
args:
|
||||
|
||||
@@ -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.43.0
|
||||
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.3
|
||||
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()
|
||||
|
||||
@@ -327,17 +327,6 @@ class JobManager(FileConfiguration, CoreSysAttributes):
|
||||
if not curr_parent.child_job_syncs:
|
||||
continue
|
||||
|
||||
# HACK: If parent trigger the same child job, we just skip this second
|
||||
# sync. Maybe it would be better to have this reflected in the job stage
|
||||
# and reset progress to 0 instead? There is no support for such stage
|
||||
# information on Core update entities today though.
|
||||
if curr_parent.done is True or curr_parent.progress >= 100:
|
||||
_LOGGER.debug(
|
||||
"Skipping parent job sync for done parent job %s",
|
||||
curr_parent.name,
|
||||
)
|
||||
continue
|
||||
|
||||
# Break after first match at each parent as it doesn't make sense
|
||||
# to match twice. But it could match multiple parents
|
||||
for sync in curr_parent.child_job_syncs:
|
||||
|
||||
@@ -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