public_git/bambuddy
1
0
Fork 0
mirror of https://github.com/maziggy/bambuddy.git synced 2026-10-22 03:49:18 +02:00
Code Issues Packages Projects Releases Wiki Activity

Replace SpoolBuddy self-update with SSH-based updates from Bambuddy

Browse Source 
The daemon's self-update mechanism (git fetch/reset on its own code) was
  fragile: .git permission errors, self-modifying code mid-run, hardcoded
  main branch. Bambuddy now SSHes into the SpoolBuddy Pi and drives the
  update remotely — matching its own branch, with step-by-step progress
  via WebSocket. Install script updated with SSH access, sudoers entry,
  and --ssh-pubkey flag for pairing.
This commit is contained in:
maziggy
2026-03-23 12:51:21 +01:00
parent 511ece791a
commit bc1bba078b
12 changed files with 454 additions and 187 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/tests/integration/test_spoolbuddy.py
+15 -45
View File
@@ -813,27 +813,19 @@ class TestDisplayEndpoints:
class TestUpdateEndpoints:
@pytest.mark.asyncio
@pytest.mark.integration
async def test_trigger_update_queues_command(self, async_client: AsyncClient, device_factory):
async def test_trigger_update_starts_ssh_update(self, async_client: AsyncClient, device_factory):
await device_factory(device_id="sb-upd")
with patch("backend.app.api.routes.spoolbuddy.ws_manager") as mock_ws:
with (
patch("backend.app.api.routes.spoolbuddy.ws_manager") as mock_ws,
patch("backend.app.services.spoolbuddy_ssh.perform_ssh_update", new_callable=AsyncMock),
):
mock_ws.broadcast = AsyncMock()
resp = await async_client.post(f"{API}/devices/sb-upd/update")
assert resp.status_code == 200
assert resp.json()["status"] == "ok"
# Verify heartbeat returns update command
with patch("backend.app.api.routes.spoolbuddy.ws_manager") as mock_ws:
mock_ws.broadcast = AsyncMock()
hb = await async_client.post(
f"{API}/devices/sb-upd/heartbeat",
json={"nfc_ok": True, "scale_ok": True, "uptime_s": 10},
)
# update command is NOT cleared by heartbeat (cleared by update-status)
assert hb.json()["pending_command"] == "update"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_trigger_update_offline_device_409(self, async_client: AsyncClient, device_factory):
@@ -943,50 +935,25 @@ class TestUpdateEndpoints:
@pytest.mark.asyncio
@pytest.mark.integration
async def test_update_check_returns_version_info(self, async_client: AsyncClient, device_factory):
"""GET /devices/{id}/update-check queries GitHub and returns version comparison."""
"""GET /devices/{id}/update-check compares device version against APP_VERSION."""
await device_factory(device_id="sb-uc", firmware_version="0.1.0")
mock_releases = [{"tag_name": "v0.2.0", "html_url": "https://github.com/test/releases/0.2.0"}]
mock_resp = MagicMock()
mock_resp.status_code = 200
mock_resp.json.return_value = mock_releases
mock_resp.raise_for_status = MagicMock()
mock_client = AsyncMock()
mock_client.get.return_value = mock_resp
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
mock_client.__aexit__ = AsyncMock(return_value=False)
with patch("httpx.AsyncClient", return_value=mock_client):
resp = await async_client.get(f"{API}/devices/sb-uc/update-check")
resp = await async_client.get(f"{API}/devices/sb-uc/update-check")
assert resp.status_code == 200
data = resp.json()
assert data["current_version"] == "0.1.0"
assert data["latest_version"] == "0.2.0"
assert data["latest_version"] is not None
assert data["update_available"] is True
assert data["release_url"] == "https://github.com/test/releases/0.2.0"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_update_check_up_to_date(self, async_client: AsyncClient, device_factory):
await device_factory(device_id="sb-uc2", firmware_version="0.2.0")
from backend.app.core.config import APP_VERSION
mock_releases = [{"tag_name": "v0.2.0", "html_url": "https://github.com/test/releases/0.2.0"}]
await device_factory(device_id="sb-uc2", firmware_version=APP_VERSION)
mock_resp = MagicMock()
mock_resp.status_code = 200
mock_resp.json.return_value = mock_releases
mock_resp.raise_for_status = MagicMock()
mock_client = AsyncMock()
mock_client.get.return_value = mock_resp
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
mock_client.__aexit__ = AsyncMock(return_value=False)
with patch("httpx.AsyncClient", return_value=mock_client):
resp = await async_client.get(f"{API}/devices/sb-uc2/update-check")
resp = await async_client.get(f"{API}/devices/sb-uc2/update-check")
assert resp.status_code == 200
assert resp.json()["update_available"] is False
@@ -1002,7 +969,10 @@ class TestUpdateEndpoints:
async def test_trigger_update_broadcasts_websocket(self, async_client: AsyncClient, device_factory):
await device_factory(device_id="sb-upd-ws")
with patch("backend.app.api.routes.spoolbuddy.ws_manager") as mock_ws:
with (
patch("backend.app.api.routes.spoolbuddy.ws_manager") as mock_ws,
patch("backend.app.services.spoolbuddy_ssh.perform_ssh_update", new_callable=AsyncMock),
):
mock_ws.broadcast = AsyncMock()
await async_client.post(f"{API}/devices/sb-upd-ws/update")