Files
supervisor/tests/dbus/agent/test_datadisk.py
Stefan AgnerandGitHub ed91b18c4b tests: enable flake8-pytest-style (PT) ruff rules (#6857)
* tests: enable flake8-pytest-style (PT) ruff rules

Enable the `PT` ruff rule set and fix the resulting violations across the
test suite:

- PT006: pass parametrize argument names as tuples instead of a single
  comma-separated string.
- PT022: switch fixtures that have no teardown from `yield` to `return`
  so the lack of cleanup is obvious at a glance.
- PT011: add `match=` to broad `pytest.raises(ValueError)` blocks so the
  expected error is anchored to a specific message.
- PT012: hoist setup (patches, branching) out of `pytest.raises()`
  blocks so only the call that is expected to raise remains inside.
- PT013: replace `from pytest import X` with `import pytest` and access
  attributes via the module.
- PT015: replace `try/except` + `assert False` patterns with
  `pytest.raises(...)`.
- PT017: replace `assert` on exceptions inside `except` blocks with
  `pytest.raises(...) as exc_info` and assert on `exc_info.value`.

No behavioral changes to the tests; the full suite still passes.

* tests: address review feedback on PT ruff rule enablement

- Fix fixture return-type annotations after switching `yield` to `return`
  in tests/conftest.py: drop the `Generator[...]`/`AsyncGenerator[...]`
  wrapper for `dns_manager_service`, `supervisor_internet`, `websession`,
  and `mock_update_data` so the annotation matches what the fixture
  actually returns.
- Correct the return-type annotation of `fixture_ip6config_service` from
  `IP4ConfigService` to `IP6ConfigService`.
- Fix recurring "excepiton" typo in tests/utils/test_exception_helper.py.

* tests: verify backup cleanup on permission error

After `test_new_backup_permission_error` raises `BackupPermissionError`,
assert that no tarfile was left behind and `tmp_path` is empty. The
previous version only checked that the exception was raised, which
missed any regression where a partial tarfile would survive the failed
create.

* tests: rename DNS_GOOD_V6 to DNS_V6_UNSUPPORTED

The constant was named "good" but its tests assert that the URLs are
rejected by the DNS validator. The IPv6 URLs are well-formed but
currently rejected because IPv6 doesn't work with the Docker network
(see `dns_url` in supervisor/validate.py). Rename the constant and the
related test to make the intent obvious.
2026-05-20 22:17:54 +02:00

91 lines
2.9 KiB
Python

"""Test Datadisk/Agent dbus interface."""
from pathlib import Path
from dbus_fast.aio.message_bus import MessageBus
import pytest
from supervisor.dbus.agent import OSAgent
from supervisor.exceptions import DBusNotConnectedError
from tests.dbus_service_mocks.agent_datadisk import DataDisk as DataDiskService
from tests.dbus_service_mocks.base import DBusServiceMock
@pytest.fixture(name="datadisk_service", autouse=True)
async def fixture_datadisk_service(
os_agent_services: dict[str, DBusServiceMock],
) -> DataDiskService:
"""Mock DataDisk dbus service."""
return os_agent_services["agent_datadisk"]
async def test_dbus_osagent_datadisk(
datadisk_service: DataDiskService, dbus_session_bus: MessageBus
):
"""Test OS-Agent datadisk properties."""
os_agent = OSAgent()
assert os_agent.datadisk.current_device is None
await os_agent.connect(dbus_session_bus)
assert os_agent.datadisk.current_device.as_posix() == "/dev/mmcblk1"
datadisk_service.emit_properties_changed({"CurrentDevice": "/dev/mmcblk1p1"})
await datadisk_service.ping()
assert os_agent.datadisk.current_device.as_posix() == "/dev/mmcblk1p1"
datadisk_service.emit_properties_changed({}, ["CurrentDevice"])
await datadisk_service.ping()
await datadisk_service.ping()
assert os_agent.datadisk.current_device.as_posix() == "/dev/mmcblk1"
async def test_dbus_osagent_datadisk_change_device(
datadisk_service: DataDiskService, dbus_session_bus: MessageBus
):
"""Change datadisk on device."""
datadisk_service.ChangeDevice.calls.clear()
os_agent = OSAgent()
with pytest.raises(DBusNotConnectedError):
await os_agent.datadisk.change_device(Path("/dev/sdb"))
await os_agent.connect(dbus_session_bus)
assert await os_agent.datadisk.change_device(Path("/dev/sdb")) is None
assert datadisk_service.ChangeDevice.calls == [("/dev/sdb",)]
async def test_dbus_osagent_datadisk_reload_device(
datadisk_service: DataDiskService, dbus_session_bus: MessageBus
):
"""Change datadisk on device."""
datadisk_service.ReloadDevice.calls.clear()
os_agent = OSAgent()
with pytest.raises(DBusNotConnectedError):
await os_agent.datadisk.reload_device()
await os_agent.connect(dbus_session_bus)
assert await os_agent.datadisk.reload_device() is None
assert datadisk_service.ReloadDevice.calls == [()]
async def test_dbus_osagent_datadisk_mark_data_move(
datadisk_service: DataDiskService, dbus_session_bus: MessageBus
):
"""Create data disk migration marker for next reboot."""
datadisk_service.MarkDataMove.calls.clear()
os_agent = OSAgent()
with pytest.raises(DBusNotConnectedError):
await os_agent.datadisk.mark_data_move()
await os_agent.connect(dbus_session_bus)
assert await os_agent.datadisk.mark_data_move() is None
assert datadisk_service.MarkDataMove.calls == [()]