47 lines
2.0 KiB
Python
47 lines
2.0 KiB
Python
from unittest.mock import MagicMock, patch
|
|
|
|
from media_library_viewer_api.services.monitoring_actions import poll_machine_snapshot
|
|
|
|
|
|
def test_poll_machine_snapshot_records_history_entries():
|
|
store = MagicMock()
|
|
machine = {
|
|
"id": "local",
|
|
"name": "This machine",
|
|
"mode": "local",
|
|
"media_root": "/srv/media",
|
|
}
|
|
|
|
with (
|
|
patch(
|
|
"media_library_viewer_api.services.monitoring_actions.build_machine_client",
|
|
return_value=object(),
|
|
) as build_client,
|
|
patch(
|
|
"media_library_viewer_api.services.monitoring_actions.resource_collector_status",
|
|
return_value="running pid=123",
|
|
) as status_fn,
|
|
patch(
|
|
"media_library_viewer_api.services.monitoring_actions.read_resource_metrics",
|
|
return_value=[{"ts": 1.0}, {"ts": 2.0}],
|
|
) as metrics_fn,
|
|
patch(
|
|
"media_library_viewer_api.services.monitoring_actions.disk_space",
|
|
return_value={"mount": "/srv/media", "used_pct": "12.5%"},
|
|
) as disk_fn,
|
|
):
|
|
result = poll_machine_snapshot(machine, store, metrics_limit=123, request_id="poll:test")
|
|
|
|
assert result["request_id"] == "poll:test"
|
|
assert result["metrics_samples"] == 2
|
|
assert result["disk_mount"] == "/srv/media"
|
|
build_client.assert_called_once_with(machine)
|
|
status_fn.assert_called_once()
|
|
metrics_fn.assert_called_once_with(build_client.return_value, max_lines=123)
|
|
disk_fn.assert_called_once_with(build_client.return_value, "/srv/media")
|
|
assert store.record_machine_action.call_count == 3
|
|
recorded_actions = [call.args[1] for call in store.record_machine_action.call_args_list]
|
|
assert recorded_actions == ["status lookup", "metrics read", "disk lookup for /srv/media"]
|
|
assert all(call.kwargs["request_id"] == "poll:test" for call in store.record_machine_action.call_args_list)
|
|
assert all(call.args[2] == "ok" for call in store.record_machine_action.call_args_list)
|