Compare commits

...
Author SHA1 Message Date
Pouzor 322600f1c9 fix(zigbee): stop canvas imports dying on a proxy read timeout
A Zigbee2MQTT networkmap on a 200+ device mesh takes minutes to build.
Two separate failures fell out of that:

- POST /zigbee/import held the HTTP request open for the whole MQTT
  round-trip, so any reverse proxy in front of the API cut it first
  (Cloudflare returns a 524 at 120 s) and the browser never saw the map.
  It now registers a job, fetches in the background and answers 202; the
  client polls GET /zigbee/import/{job_id} until the payload is ready.
  Job results are transient and live in memory with a 15 min TTL — the
  same single-worker assumption the scheduler already makes. A failed
  fetch replays the status the synchronous route used to raise, so a bad
  broker is still a 502 and a slow mesh still a 504.

- The networkmap wait was hard-coded at 300 s with no way to raise it.
  It now reads ZIGBEE_NETWORKMAP_TIMEOUT, and the shared MQTT round-trip
  used by the Z-Wave import reads MQTT_RESPONSE_TIMEOUT. Both default to
  300 s, fall back to that if misconfigured to a non-positive value, and
  name themselves in the timeout message.

Also corrects the route and doc claims that the wait was 60 s.

The /import tests changed with the contract they cover, not to pass.

Fixes #380

ha-relevant: yes
2026-08-31 11:26:45 +02:00
18 changed files with 959 additions and 106 deletions
+7
View File
@@ -89,6 +89,10 @@ MCP_SERVICE_KEY=svc_changeme
# ZIGBEE_BASE_TOPIC=zigbee2mqtt
# ZIGBEE_MQTT_TLS=false # true for TLS brokers (typically port 8883)
# ZIGBEE_MQTT_TLS_INSECURE=false # skip cert verify (self-signed only; requires TLS)
# Seconds to wait for the Z2M bridge to answer a networkmap request. Applies to
# manual imports too. Raise it on a large mesh (200+ devices can take minutes)
# if an import fails with "Timed out waiting for networkmap response".
# ZIGBEE_NETWORKMAP_TIMEOUT=300
# Z-Wave JS UI (zwavejs2mqtt) auto-sync — same MQTT secret/env rules as Zigbee.
# ZWAVE_MQTT_HOST=192.168.1.20
@@ -99,3 +103,6 @@ MCP_SERVICE_KEY=svc_changeme
# ZWAVE_GATEWAY_NAME=zwavejs2mqtt
# ZWAVE_MQTT_TLS=false # true for TLS brokers (typically port 8883)
# ZWAVE_MQTT_TLS_INSECURE=false # skip cert verify (self-signed only; requires TLS)
# Seconds to wait for the gateway to answer an MQTT request. Same rationale as
# ZIGBEE_NETWORKMAP_TIMEOUT above.
# MQTT_RESPONSE_TIMEOUT=300
+70 -17
View File
@@ -19,6 +19,8 @@ from app.schemas.zigbee import (
ZigbeeConfig,
ZigbeeCoordinatorOut,
ZigbeeEdgeOut,
ZigbeeImportJob,
ZigbeeImportJobResult,
ZigbeeImportPendingResponse,
ZigbeeImportRequest,
ZigbeeImportResponse,
@@ -27,6 +29,7 @@ from app.schemas.zigbee import (
ZigbeeTestConnectionRequest,
ZigbeeTestConnectionResponse,
)
from app.services.import_jobs import create_job, fail_job, finish_job, get_job
from app.services.node_dedupe import dedupe_nodes_by_device
from app.services.zigbee_service import (
build_zigbee_properties,
@@ -48,18 +51,70 @@ async def _is_drawn(db: AsyncSession, device_id: str) -> bool:
router = APIRouter()
@router.post("/import", response_model=ZigbeeImportResponse)
def _import_error(exc: BaseException) -> tuple[int, str]:
"""Map a fetch_networkmap failure to the (status, detail) the API reports."""
if isinstance(exc, ImportError):
return 500, str(exc)
if isinstance(exc, ConnectionError):
return 502, str(exc)
if isinstance(exc, TimeoutError):
return 504, str(exc)
if isinstance(exc, ValueError):
return 422, str(exc)
logger.exception("Unexpected error during Zigbee import", exc_info=exc)
return 500, "Unexpected error during Zigbee import"
@router.post("/import", response_model=ZigbeeImportJob, status_code=202)
async def import_zigbee_network(
payload: ZigbeeImportRequest,
background_tasks: BackgroundTasks,
_: str = Depends(get_current_user),
) -> ZigbeeImportResponse:
"""Fetch the Zigbee2MQTT network map and return nodes + edges ready for canvas drop.
) -> ZigbeeImportJob:
"""Start a canvas import and return a job id to poll.
Connects to the specified MQTT broker, publishes a networkmap request to
``<base_topic>/bridge/request/networkmap``, and waits up to 60 s for the
response (large meshes can take 30 s+). The devices are returned as typed homelable nodes with a
coordinator → router → end-device hierarchy.
``<base_topic>/bridge/request/networkmap`` and waits up to
``ZIGBEE_NETWORKMAP_TIMEOUT`` seconds (default 300) for the response.
The fetch runs in the background and the result is collected from
``GET /zigbee/import/{job_id}``: a 200+ device mesh takes minutes to answer,
which outlives the read timeout of any reverse proxy sitting in front of the
API. Polling keeps each request short.
"""
job = create_job()
background_tasks.add_task(_background_canvas_import, job.id, payload)
return ZigbeeImportJob(job_id=job.id, status=job.status)
@router.get("/import/{job_id}", response_model=ZigbeeImportJobResult)
async def get_zigbee_import_job(
job_id: str,
_: str = Depends(get_current_user),
) -> ZigbeeImportJobResult:
"""Poll a canvas import started by ``POST /zigbee/import``.
While running, returns ``status="running"`` with no payload. On success the
nodes and edges are carried in ``result``. A failed fetch is reported as the
status code the synchronous route used to raise, so the client keeps
distinguishing a bad broker (502) from a slow mesh (504).
"""
job = get_job(job_id)
if job is None:
raise HTTPException(status_code=404, detail="Import job not found or expired")
if job.status == "error":
raise HTTPException(
status_code=job.error_status or 500,
detail=job.error or "Zigbee import failed",
)
result = None
if job.status == "done" and job.result is not None:
result = ZigbeeImportResponse(**job.result)
return ZigbeeImportJobResult(job_id=job.id, status=job.status, result=result)
async def _background_canvas_import(job_id: str, payload: ZigbeeImportRequest) -> None:
"""Fetch the network map and park the canvas-ready payload on the job."""
try:
nodes_raw, edges_raw = await fetch_networkmap(
mqtt_host=payload.mqtt_host,
@@ -70,21 +125,19 @@ async def import_zigbee_network(
tls=payload.mqtt_tls,
tls_insecure=payload.mqtt_tls_insecure,
)
except ImportError as exc:
raise HTTPException(status_code=500, detail=str(exc)) from exc
except ConnectionError as exc:
raise HTTPException(status_code=502, detail=str(exc)) from exc
except TimeoutError as exc:
raise HTTPException(status_code=504, detail=str(exc)) from exc
except ValueError as exc:
raise HTTPException(status_code=422, detail=str(exc)) from exc
except Exception as exc:
logger.exception("Unexpected error during Zigbee import")
raise HTTPException(status_code=500, detail="Unexpected error during Zigbee import") from exc
status, detail = _import_error(exc)
fail_job(job_id, detail, status)
return
nodes = [ZigbeeNodeOut(**n) for n in nodes_raw]
edges = [ZigbeeEdgeOut(**e) for e in edges_raw]
return ZigbeeImportResponse(nodes=nodes, edges=edges, device_count=len(nodes))
finish_job(
job_id,
ZigbeeImportResponse(
nodes=nodes, edges=edges, device_count=len(nodes)
).model_dump(),
)
@router.post("/import-pending", response_model=ScanRunResponse)
+8
View File
@@ -179,6 +179,10 @@ class Settings(BaseSettings):
zigbee_mqtt_tls_insecure: bool = False
zigbee_sync_enabled: bool = False
zigbee_sync_interval: int = 3600 # seconds (floor 300 enforced on write)
# Seconds to wait for the Z2M bridge to answer a networkmap request. A mesh
# of 200+ devices can take several minutes to build the map, so raise this
# if imports fail with "Timed out waiting for networkmap response".
zigbee_networkmap_timeout: int = 300
# Z-Wave JS UI (zwavejs2mqtt) auto-sync import. Same secret/env rules.
zwave_mqtt_host: str = ""
@@ -192,6 +196,10 @@ class Settings(BaseSettings):
zwave_sync_enabled: bool = False
zwave_sync_interval: int = 3600 # seconds (floor 300 enforced on write)
# Seconds to wait for any MQTT gateway request/response round-trip
# (the Z-Wave node dump today). Same rationale as the Zigbee twin above.
mqtt_response_timeout: int = 300
def _override_path(self) -> Path:
return Path(self.sqlite_path).parent / "scan_config.json"
+15
View File
@@ -73,6 +73,21 @@ class ZigbeeImportResponse(BaseModel):
device_count: int
class ZigbeeImportJob(BaseModel):
"""Handle returned when a canvas import is queued."""
job_id: str
status: str # running | done | error
class ZigbeeImportJobResult(BaseModel):
"""Poll response for a canvas import. ``result`` is set once done."""
job_id: str
status: str # running | done
result: ZigbeeImportResponse | None = None
class ZigbeeTestConnectionResponse(BaseModel):
connected: bool
message: str
+92
View File
@@ -0,0 +1,92 @@
"""In-process registry for long-running import jobs that return a payload.
Pending imports already run in the background and report through ``scan_runs``.
Canvas imports are different: the caller wants the fetched map *back* so it can
drop it on the canvas, which kept the request open for the whole MQTT
round-trip. On a large mesh that outlives any reverse proxy's read timeout
(Cloudflare cuts at 100 s and returns a 524), so the browser never sees the
result even though the fetch succeeded server-side.
The fix is to hand the client a job id immediately and let it poll. Results are
transient — a canvas import is meaningless once the user has closed the modal —
so they live in memory rather than the DB, and expire on a TTL. The backend runs
a single uvicorn worker, the same assumption the scheduler and ``BackgroundTasks``
already make.
"""
from __future__ import annotations
import time
import uuid
from dataclasses import dataclass, field
from typing import Any, Literal
JobStatus = Literal["running", "done", "error"]
# How long a finished job stays readable. Long enough for a client that polls
# slowly or reloads mid-import, short enough that a forgotten map is not held
# for the life of the process.
JOB_TTL_SECONDS = 900.0
@dataclass
class ImportJob:
id: str
status: JobStatus = "running"
result: dict[str, Any] | None = None
error: str | None = None
# HTTP status the synchronous route would have raised, so the client can
# keep reacting to failures the way it always has.
error_status: int | None = None
created_at: float = field(default_factory=time.monotonic)
finished_at: float | None = None
_jobs: dict[str, ImportJob] = {}
def _purge_expired(now: float) -> None:
for job_id, job in list(_jobs.items()):
if job.finished_at is not None and now - job.finished_at > JOB_TTL_SECONDS:
del _jobs[job_id]
def create_job() -> ImportJob:
"""Register a new running job and return it."""
now = time.monotonic()
_purge_expired(now)
job = ImportJob(id=str(uuid.uuid4()))
_jobs[job.id] = job
return job
def get_job(job_id: str) -> ImportJob | None:
"""Return a job, or None if it never existed or has expired."""
_purge_expired(time.monotonic())
return _jobs.get(job_id)
def finish_job(job_id: str, result: dict[str, Any]) -> None:
"""Mark a job done and attach its payload. No-op if it expired."""
job = _jobs.get(job_id)
if job is None:
return
job.status = "done"
job.result = result
job.finished_at = time.monotonic()
def fail_job(job_id: str, error: str, status: int) -> None:
"""Mark a job failed with a client-safe message. No-op if it expired."""
job = _jobs.get(job_id)
if job is None:
return
job.status = "error"
job.error = error
job.error_status = status
job.finished_at = time.monotonic()
def reset_jobs() -> None:
"""Drop every job. For tests."""
_jobs.clear()
+23 -4
View File
@@ -13,6 +13,8 @@ import logging
import ssl
from typing import Any
from app.core.config import settings
logger = logging.getLogger(__name__)
try:
@@ -21,7 +23,10 @@ except ImportError: # pragma: no cover
aiomqtt = None # type: ignore[assignment]
_CONNECTION_TIMEOUT = 5.0 # seconds to verify broker reachability
_RESPONSE_TIMEOUT = 300.0 # seconds to wait for a gateway response (large meshes are slow)
# Fallback wait for a gateway response. The effective value comes from
# ``settings.mqtt_response_timeout`` (env MQTT_RESPONSE_TIMEOUT), read at call
# time so an operator with a large mesh can raise it without a code change.
_RESPONSE_TIMEOUT = 300.0
def _sanitize_mqtt_error(exc: BaseException) -> str:
@@ -71,11 +76,14 @@ async def request_response(
password: str | None = None,
tls: bool = False,
tls_insecure: bool = False,
response_timeout: float = _RESPONSE_TIMEOUT,
response_timeout: float | None = None,
) -> dict[str, Any]:
"""Publish ``request_payload`` to ``request_topic`` and return the first
JSON message received on ``response_topic`` as a dict.
``response_timeout`` defaults to ``settings.mqtt_response_timeout``
(env ``MQTT_RESPONSE_TIMEOUT``, 300 s) — a large mesh can take minutes.
Raises:
ImportError: if aiomqtt is not installed.
TimeoutError: if no response arrives in time.
@@ -88,6 +96,14 @@ async def request_response(
"Install it with: pip install aiomqtt"
)
timeout = (
float(settings.mqtt_response_timeout)
if response_timeout is None
else response_timeout
)
if timeout <= 0:
timeout = _RESPONSE_TIMEOUT
response_payload: dict[str, Any] = {}
tls_context = _build_tls_context(tls_insecure) if tls else None
@@ -122,12 +138,15 @@ async def request_response(
raise ValueError(f"Malformed MQTT response: {exc}") from exc
return
await asyncio.wait_for(_wait_for_response(), timeout=response_timeout)
await asyncio.wait_for(_wait_for_response(), timeout=timeout)
except aiomqtt.MqttError as exc:
raise ConnectionError(_sanitize_mqtt_error(exc)) from exc
except asyncio.TimeoutError as exc:
raise TimeoutError("Timed out waiting for MQTT response") from exc
raise TimeoutError(
f"Timed out waiting for MQTT response after {timeout:g}s — "
"raise MQTT_RESPONSE_TIMEOUT if your mesh is large"
) from exc
if not response_payload:
raise ValueError("Empty MQTT response received")
+23 -3
View File
@@ -7,6 +7,7 @@ import json
import logging
from typing import Any
from app.core.config import settings
from app.services.mqtt_common import _build_tls_context, _sanitize_mqtt_error
logger = logging.getLogger(__name__)
@@ -19,7 +20,10 @@ except ImportError: # pragma: no cover
_NETWORKMAP_REQUEST_TOPIC = "{base_topic}/bridge/request/networkmap"
_NETWORKMAP_RESPONSE_TOPIC = "{base_topic}/bridge/response/networkmap"
_CONNECTION_TIMEOUT = 5.0 # seconds to verify broker reachability
_NETWORKMAP_TIMEOUT = 300.0 # seconds to wait for the networkmap response (large meshes can be slow)
# Fallback wait for the networkmap response. The effective value comes from
# ``settings.zigbee_networkmap_timeout`` (env ZIGBEE_NETWORKMAP_TIMEOUT), read at
# call time so an operator with a large mesh can raise it without a code change.
_NETWORKMAP_TIMEOUT = 300.0
# Re-exported for backwards compatibility — these now live in mqtt_common.
__all__ = ["_build_tls_context", "_sanitize_mqtt_error"]
@@ -238,9 +242,14 @@ async def fetch_networkmap(
password: str | None = None,
tls: bool = False,
tls_insecure: bool = False,
response_timeout: float | None = None,
) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
"""Connect to the MQTT broker, request the Z2M networkmap, and return (nodes, edges).
``response_timeout`` defaults to ``settings.zigbee_networkmap_timeout``
(env ``ZIGBEE_NETWORKMAP_TIMEOUT``, 300 s) — a 200+ device mesh can take
minutes to answer.
Raises:
TimeoutError: if the broker does not respond in time.
ConnectionError: if the broker cannot be reached.
@@ -252,6 +261,14 @@ async def fetch_networkmap(
"Install it with: pip install aiomqtt"
)
timeout = (
float(settings.zigbee_networkmap_timeout)
if response_timeout is None
else response_timeout
)
if timeout <= 0:
timeout = _NETWORKMAP_TIMEOUT
request_topic = _NETWORKMAP_REQUEST_TOPIC.format(base_topic=base_topic)
response_topic = _NETWORKMAP_RESPONSE_TOPIC.format(base_topic=base_topic)
@@ -295,12 +312,15 @@ async def fetch_networkmap(
) from exc
return
await asyncio.wait_for(_wait_for_response(), timeout=_NETWORKMAP_TIMEOUT)
await asyncio.wait_for(_wait_for_response(), timeout=timeout)
except aiomqtt.MqttError as exc:
raise ConnectionError(_sanitize_mqtt_error(exc)) from exc
except asyncio.TimeoutError as exc:
raise TimeoutError("Timed out waiting for networkmap response") from exc
raise TimeoutError(
f"Timed out waiting for networkmap response after {timeout:g}s — "
"raise ZIGBEE_NETWORKMAP_TIMEOUT if your mesh is large"
) from exc
if not response_payload:
raise ValueError("Empty networkmap response received")
+90
View File
@@ -0,0 +1,90 @@
"""Unit tests for the in-process canvas-import job registry."""
from __future__ import annotations
import pytest
from app.services import import_jobs
from app.services.import_jobs import (
create_job,
fail_job,
finish_job,
get_job,
reset_jobs,
)
@pytest.fixture(autouse=True)
def _clean() -> None:
reset_jobs()
def test_create_job_starts_running_with_unique_id() -> None:
a = create_job()
b = create_job()
assert a.id != b.id
assert a.status == "running"
assert a.result is None
assert a.error is None
def test_get_job_returns_the_registered_job() -> None:
job = create_job()
assert get_job(job.id) is job
def test_get_unknown_job_returns_none() -> None:
assert get_job("nope") is None
def test_finish_job_records_the_payload() -> None:
job = create_job()
finish_job(job.id, {"device_count": 3})
stored = get_job(job.id)
assert stored is not None
assert stored.status == "done"
assert stored.result == {"device_count": 3}
assert stored.finished_at is not None
def test_fail_job_records_message_and_status() -> None:
job = create_job()
fail_job(job.id, "broker unreachable", 502)
stored = get_job(job.id)
assert stored is not None
assert stored.status == "error"
assert stored.error == "broker unreachable"
assert stored.error_status == 502
def test_finish_and_fail_are_noops_for_unknown_ids() -> None:
finish_job("gone", {"device_count": 0})
fail_job("gone", "boom", 500)
assert get_job("gone") is None
def test_finished_jobs_expire_after_the_ttl(monkeypatch) -> None:
clock = {"now": 0.0}
monkeypatch.setattr(import_jobs.time, "monotonic", lambda: clock["now"])
job = create_job()
finish_job(job.id, {"device_count": 1})
clock["now"] = import_jobs.JOB_TTL_SECONDS + 1
assert get_job(job.id) is None
def test_running_jobs_never_expire(monkeypatch) -> None:
"""A slow mesh must not have its job purged out from under it."""
clock = {"now": 0.0}
monkeypatch.setattr(import_jobs.time, "monotonic", lambda: clock["now"])
job = create_job()
clock["now"] = import_jobs.JOB_TTL_SECONDS * 10
assert get_job(job.id) is not None
def test_reset_jobs_clears_everything() -> None:
job = create_job()
reset_jobs()
assert get_job(job.id) is None
+100
View File
@@ -2,13 +2,16 @@
from __future__ import annotations
import asyncio
import json
import ssl
from unittest.mock import patch
import pytest
from app.core.config import settings
from app.services.mqtt_common import (
_RESPONSE_TIMEOUT,
_build_tls_context,
_sanitize_mqtt_error,
request_response,
@@ -181,3 +184,100 @@ async def test_test_connection_failure() -> None:
mock_aiomqtt.MqttError = Exception
with pytest.raises(ConnectionError):
await _test_connection("bad", 1883)
# ---------------------------------------------------------------------------
# Response timeout — configurable via MQTT_RESPONSE_TIMEOUT (issue #380)
# ---------------------------------------------------------------------------
class _SilentClient:
"""An MQTT client that connects but never delivers the response message."""
async def __aenter__(self):
return self
async def __aexit__(self, *_):
pass
async def subscribe(self, *_a, **_kw) -> None:
pass
async def publish(self, *_a, **_kw) -> None:
pass
@property
def messages(self):
async def _never():
await asyncio.Event().wait()
yield # pragma: no cover
return _never()
class _NeverMqttError(Exception):
"""Stand-in for aiomqtt.MqttError that no test path actually raises."""
async def _round_trip(**kwargs):
return await request_response(
mqtt_host="host",
mqtt_port=1883,
request_topic="req",
response_topic="res",
request_payload={},
**kwargs,
)
@pytest.mark.asyncio
async def test_request_response_uses_settings_timeout(monkeypatch) -> None:
monkeypatch.setattr(settings, "mqtt_response_timeout", 0.01)
with patch("app.services.mqtt_common.aiomqtt") as mock_aiomqtt:
mock_aiomqtt.Client.return_value = _SilentClient()
mock_aiomqtt.MqttError = _NeverMqttError
with pytest.raises(TimeoutError) as ei:
await _round_trip()
msg = str(ei.value)
assert "0.01s" in msg
assert "MQTT_RESPONSE_TIMEOUT" in msg
@pytest.mark.asyncio
async def test_request_response_explicit_timeout_overrides_settings(monkeypatch) -> None:
monkeypatch.setattr(settings, "mqtt_response_timeout", 999)
with patch("app.services.mqtt_common.aiomqtt") as mock_aiomqtt:
mock_aiomqtt.Client.return_value = _SilentClient()
mock_aiomqtt.MqttError = _NeverMqttError
with pytest.raises(TimeoutError) as ei:
await _round_trip(response_timeout=0.01)
assert "0.01s" in str(ei.value)
@pytest.mark.asyncio
async def test_request_response_non_positive_setting_falls_back(monkeypatch) -> None:
"""A misconfigured 0 must not mean 'give up immediately'."""
monkeypatch.setattr(settings, "mqtt_response_timeout", 0)
captured: dict[str, float] = {}
async def _fake_wait_for(awaitable, timeout):
captured["timeout"] = timeout
awaitable.close()
raise asyncio.TimeoutError
with patch("app.services.mqtt_common.aiomqtt") as mock_aiomqtt:
mock_aiomqtt.Client.return_value = _SilentClient()
mock_aiomqtt.MqttError = _NeverMqttError
with (
patch("app.services.mqtt_common.asyncio.wait_for", _fake_wait_for),
pytest.raises(TimeoutError),
):
await _round_trip()
assert captured["timeout"] == _RESPONSE_TIMEOUT
+91 -40
View File
@@ -7,10 +7,34 @@ from unittest.mock import AsyncMock, patch
import pytest
from httpx import AsyncClient
from app.services.import_jobs import reset_jobs
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
@pytest.fixture(autouse=True)
def _clear_import_jobs() -> None:
"""The canvas-import job registry is process-global; isolate each test."""
reset_jobs()
async def _start_canvas_import(
client: AsyncClient, headers: dict, body: dict
) -> str:
"""POST /zigbee/import and return the job id.
The route is fire-and-poll: it answers 202 immediately and the fetch runs as
a Starlette background task, which the ASGI transport completes before this
await returns — so the job is already settled by the time we poll.
"""
res = await client.post("/api/v1/zigbee/import", json=body, headers=headers)
assert res.status_code == 202, res.text
job_id: str = res.json()["job_id"]
assert res.json()["status"] == "running"
return job_id
# ---------------------------------------------------------------------------
# /api/v1/zigbee/test-connection
# ---------------------------------------------------------------------------
@@ -104,18 +128,18 @@ _SAMPLE_EDGES = [
async def test_import_success(client: AsyncClient, headers: dict) -> None:
with patch("app.api.routes.zigbee.fetch_networkmap") as mock_fetch:
mock_fetch.return_value = (_SAMPLE_NODES, _SAMPLE_EDGES)
res = await client.post(
"/api/v1/zigbee/import",
json={
"mqtt_host": "localhost",
"mqtt_port": 1883,
"base_topic": "zigbee2mqtt",
},
headers=headers,
job_id = await _start_canvas_import(
client,
headers,
{"mqtt_host": "localhost", "mqtt_port": 1883, "base_topic": "zigbee2mqtt"},
)
res = await client.get(f"/api/v1/zigbee/import/{job_id}", headers=headers)
assert res.status_code == 200
data = res.json()
body = res.json()
assert body["status"] == "done"
assert body["job_id"] == job_id
data = body["result"]
assert data["device_count"] == 2
assert len(data["nodes"]) == 2
assert len(data["edges"]) == 1
@@ -123,22 +147,57 @@ async def test_import_success(client: AsyncClient, headers: dict) -> None:
assert coordinator["ieee_address"] == "0x00000000"
@pytest.mark.asyncio
async def test_import_post_does_not_block_on_the_fetch(
client: AsyncClient, headers: dict
) -> None:
"""The POST answers 202 without a payload — that is what keeps a slow mesh
from outliving a reverse proxy's read timeout (issue #380)."""
with patch("app.api.routes.zigbee.fetch_networkmap") as mock_fetch:
mock_fetch.return_value = (_SAMPLE_NODES, _SAMPLE_EDGES)
res = await client.post(
"/api/v1/zigbee/import",
json={"mqtt_host": "localhost", "mqtt_port": 1883},
headers=headers,
)
assert res.status_code == 202
assert set(res.json()) == {"job_id", "status"}
@pytest.mark.asyncio
async def test_import_job_unknown_id_returns_404(
client: AsyncClient, headers: dict
) -> None:
res = await client.get("/api/v1/zigbee/import/no-such-job", headers=headers)
assert res.status_code == 404
@pytest.mark.asyncio
async def test_import_job_requires_auth(client: AsyncClient, headers: dict) -> None:
with patch("app.api.routes.zigbee.fetch_networkmap") as mock_fetch:
mock_fetch.return_value = ([], [])
job_id = await _start_canvas_import(
client, headers, {"mqtt_host": "localhost", "mqtt_port": 1883}
)
res = await client.get(f"/api/v1/zigbee/import/{job_id}")
assert res.status_code == 401
@pytest.mark.asyncio
async def test_import_with_credentials(client: AsyncClient, headers: dict) -> None:
with patch("app.api.routes.zigbee.fetch_networkmap") as mock_fetch:
mock_fetch.return_value = ([], [])
res = await client.post(
"/api/v1/zigbee/import",
json={
await _start_canvas_import(
client,
headers,
{
"mqtt_host": "localhost",
"mqtt_port": 1883,
"mqtt_username": "admin",
"mqtt_password": "secret",
"base_topic": "z2m",
},
headers=headers,
)
assert res.status_code == 200
mock_fetch.assert_called_once_with(
mqtt_host="localhost",
mqtt_port=1883,
@@ -154,11 +213,10 @@ async def test_import_with_credentials(client: AsyncClient, headers: dict) -> No
async def test_import_connection_error_returns_502(client: AsyncClient, headers: dict) -> None:
with patch("app.api.routes.zigbee.fetch_networkmap") as mock_fetch:
mock_fetch.side_effect = ConnectionError("broker unreachable")
res = await client.post(
"/api/v1/zigbee/import",
json={"mqtt_host": "bad-host", "mqtt_port": 1883},
headers=headers,
job_id = await _start_canvas_import(
client, headers, {"mqtt_host": "bad-host", "mqtt_port": 1883}
)
res = await client.get(f"/api/v1/zigbee/import/{job_id}", headers=headers)
assert res.status_code == 502
assert "broker unreachable" in res.json()["detail"]
@@ -167,23 +225,22 @@ async def test_import_connection_error_returns_502(client: AsyncClient, headers:
async def test_import_timeout_returns_504(client: AsyncClient, headers: dict) -> None:
with patch("app.api.routes.zigbee.fetch_networkmap") as mock_fetch:
mock_fetch.side_effect = TimeoutError("timed out")
res = await client.post(
"/api/v1/zigbee/import",
json={"mqtt_host": "localhost", "mqtt_port": 1883},
headers=headers,
job_id = await _start_canvas_import(
client, headers, {"mqtt_host": "localhost", "mqtt_port": 1883}
)
res = await client.get(f"/api/v1/zigbee/import/{job_id}", headers=headers)
assert res.status_code == 504
assert "timed out" in res.json()["detail"]
@pytest.mark.asyncio
async def test_import_malformed_payload_returns_422(client: AsyncClient, headers: dict) -> None:
with patch("app.api.routes.zigbee.fetch_networkmap") as mock_fetch:
mock_fetch.side_effect = ValueError("malformed response")
res = await client.post(
"/api/v1/zigbee/import",
json={"mqtt_host": "localhost", "mqtt_port": 1883},
headers=headers,
job_id = await _start_canvas_import(
client, headers, {"mqtt_host": "localhost", "mqtt_port": 1883}
)
res = await client.get(f"/api/v1/zigbee/import/{job_id}", headers=headers)
assert res.status_code == 422
@@ -201,13 +258,12 @@ async def test_import_empty_network(client: AsyncClient, headers: dict) -> None:
"""An empty Zigbee network (coordinator only) is a valid response."""
with patch("app.api.routes.zigbee.fetch_networkmap") as mock_fetch:
mock_fetch.return_value = ([], [])
res = await client.post(
"/api/v1/zigbee/import",
json={"mqtt_host": "localhost", "mqtt_port": 1883},
headers=headers,
job_id = await _start_canvas_import(
client, headers, {"mqtt_host": "localhost", "mqtt_port": 1883}
)
res = await client.get(f"/api/v1/zigbee/import/{job_id}", headers=headers)
assert res.status_code == 200
data = res.json()
data = res.json()["result"]
assert data["device_count"] == 0
assert data["nodes"] == []
assert data["edges"] == []
@@ -227,16 +283,11 @@ async def test_import_missing_mqtt_host(client: AsyncClient, headers: dict) -> N
async def test_import_with_tls_passes_flags(client: AsyncClient, headers: dict) -> None:
with patch("app.api.routes.zigbee.fetch_networkmap") as mock_fetch:
mock_fetch.return_value = ([], [])
res = await client.post(
"/api/v1/zigbee/import",
json={
"mqtt_host": "broker.example.com",
"mqtt_port": 8883,
"mqtt_tls": True,
},
headers=headers,
await _start_canvas_import(
client,
headers,
{"mqtt_host": "broker.example.com", "mqtt_port": 8883, "mqtt_tls": True},
)
assert res.status_code == 200
kwargs = mock_fetch.call_args.kwargs
assert kwargs["tls"] is True
assert kwargs["tls_insecure"] is False
+99
View File
@@ -2,6 +2,7 @@
from __future__ import annotations
import asyncio
import json
from typing import Any
from unittest.mock import patch
@@ -9,7 +10,9 @@ from unittest.mock import patch
import aiomqtt # noqa: F401
import pytest
from app.core.config import settings
from app.services.zigbee_service import (
_NETWORKMAP_TIMEOUT,
_find_parent_router,
_z2m_type_to_homelable,
fetch_networkmap,
@@ -433,6 +436,10 @@ async def test_test_mqtt_connection_failure() -> None:
await _test_mqtt_connection("bad-host", 1883)
class _NeverMqttError(Exception):
"""Stand-in for aiomqtt.MqttError that no test path actually raises."""
# ---------------------------------------------------------------------------
# TLS context
# ---------------------------------------------------------------------------
@@ -571,3 +578,95 @@ async def test_fetch_networkmap_does_not_leak_creds_in_connection_error() -> Non
assert "hunter2" not in msg
assert "admin" not in msg
assert msg == "Authentication failed"
# ---------------------------------------------------------------------------
# Networkmap response timeout — configurable via ZIGBEE_NETWORKMAP_TIMEOUT
# (issue #380: a 200+ device mesh needs longer than the hard-coded 300 s)
# ---------------------------------------------------------------------------
class _SilentClient:
"""An MQTT client that connects but never delivers the response message."""
async def __aenter__(self):
return self
async def __aexit__(self, *_):
pass
async def subscribe(self, *_a, **_kw) -> None:
pass
async def publish(self, *_a, **_kw) -> None:
pass
@property
def messages(self):
async def _never():
await asyncio.Event().wait()
yield # pragma: no cover
return _never()
@pytest.mark.asyncio
async def test_fetch_networkmap_uses_settings_timeout(monkeypatch) -> None:
monkeypatch.setattr(settings, "zigbee_networkmap_timeout", 0.01)
with patch("app.services.zigbee_service.aiomqtt") as mock_aiomqtt:
mock_aiomqtt.Client.return_value = _SilentClient()
mock_aiomqtt.MqttError = _NeverMqttError
with pytest.raises(TimeoutError) as ei:
await fetch_networkmap(
mqtt_host="host", mqtt_port=1883, base_topic="zigbee2mqtt"
)
msg = str(ei.value)
assert "0.01s" in msg
assert "ZIGBEE_NETWORKMAP_TIMEOUT" in msg
@pytest.mark.asyncio
async def test_fetch_networkmap_explicit_timeout_overrides_settings(monkeypatch) -> None:
monkeypatch.setattr(settings, "zigbee_networkmap_timeout", 999)
with patch("app.services.zigbee_service.aiomqtt") as mock_aiomqtt:
mock_aiomqtt.Client.return_value = _SilentClient()
mock_aiomqtt.MqttError = _NeverMqttError
with pytest.raises(TimeoutError) as ei:
await fetch_networkmap(
mqtt_host="host",
mqtt_port=1883,
base_topic="zigbee2mqtt",
response_timeout=0.01,
)
assert "0.01s" in str(ei.value)
@pytest.mark.asyncio
async def test_fetch_networkmap_non_positive_setting_falls_back(monkeypatch) -> None:
"""A misconfigured 0 must not mean 'give up immediately'."""
monkeypatch.setattr(settings, "zigbee_networkmap_timeout", 0)
captured: dict[str, float] = {}
async def _fake_wait_for(awaitable, timeout):
captured["timeout"] = timeout
awaitable.close()
raise asyncio.TimeoutError
with patch("app.services.zigbee_service.aiomqtt") as mock_aiomqtt:
mock_aiomqtt.Client.return_value = _SilentClient()
mock_aiomqtt.MqttError = _NeverMqttError
with (
patch("app.services.zigbee_service.asyncio.wait_for", _fake_wait_for),
pytest.raises(TimeoutError),
):
await fetch_networkmap(
mqtt_host="host", mqtt_port=1883, base_topic="zigbee2mqtt"
)
assert captured["timeout"] == _NETWORKMAP_TIMEOUT
+17 -1
View File
@@ -55,9 +55,25 @@ Click **Fetch Devices**. Homelable will:
1. Connect to the broker
2. Subscribe to the response topic
3. Publish `{"type": "raw", "routes": false}` to the request topic
4. Wait up to 60 seconds for the network map response (large meshes can take 30 s+)
4. Wait up to `ZIGBEE_NETWORKMAP_TIMEOUT` seconds (default 300) for the network map response
5. Parse and group devices by type
The fetch runs server-side and the browser polls for the result, so a mesh that
takes minutes to answer cannot be cut short by a reverse proxy's read timeout.
### Large meshes and timeouts
A network of 200+ devices can take several minutes to build its map. Two knobs:
| Variable | Default | What it bounds |
|---|---|---|
| `ZIGBEE_NETWORKMAP_TIMEOUT` | `300` | Seconds to wait for the Z2M bridge to answer a networkmap request. Raise it if an import fails with *Timed out waiting for networkmap response*. |
| `MQTT_RESPONSE_TIMEOUT` | `300` | The same bound for the Z-Wave MQTT round-trip. |
Both apply to manual imports and to auto-sync. If you reverse-proxy the API,
these are the only timeouts that matter — the import request itself returns
immediately.
### 5. Select and add to canvas
Devices are grouped by type (Coordinator / Router / End Device).
@@ -242,6 +242,11 @@ describe('api/client', () => {
expect(api.post).toHaveBeenCalledWith('/zigbee/import-pending', cfg)
})
it('zigbeeApi.getImportJob polls the job by id', () => {
mod.zigbeeApi.getImportJob('job-1')
expect(api.get).toHaveBeenCalledWith('/zigbee/import/job-1')
})
it('zwaveApi.testConnection/importNetwork/importToPending', () => {
const cfg = { mqtt_host: 'h', mqtt_port: 1883, prefix: 'zwave', gateway_name: 'zwavejs2mqtt' }
mod.zwaveApi.testConnection(cfg)
+16 -5
View File
@@ -305,6 +305,8 @@ export const racksApi = {
}),
}
export type ZigbeeImportJobStatus = 'running' | 'done' | 'error'
export interface ZigbeeConfigData {
mqtt_host: string
mqtt_port: number
@@ -358,11 +360,20 @@ export const zigbeeApi = {
mqtt_tls?: boolean
mqtt_tls_insecure?: boolean
}) =>
api.post<{
nodes: import('@/components/zigbee/types').ZigbeeNode[]
edges: import('@/components/zigbee/types').ZigbeeEdge[]
device_count: number
}>('/zigbee/import', data),
api.post<{ job_id: string; status: ZigbeeImportJobStatus }>('/zigbee/import', data),
// Poll a canvas import started by importNetwork. The fetch runs server-side
// so a slow mesh cannot outlive a reverse proxy's read timeout.
getImportJob: (jobId: string) =>
api.get<{
job_id: string
status: ZigbeeImportJobStatus
result: {
nodes: import('@/components/zigbee/types').ZigbeeNode[]
edges: import('@/components/zigbee/types').ZigbeeEdge[]
device_count: number
} | null
}>(`/zigbee/import/${jobId}`),
importToPending: (data: {
mqtt_host: string
@@ -1,10 +1,11 @@
import { useState } from 'react'
import { useEffect, useRef, useState } from 'react'
import { Network, Router, Cpu, CheckCircle2, XCircle, Loader2, Plus } from 'lucide-react'
import { Dialog, DialogContent, DialogHeader, DialogTitle, DialogFooter } from '@/components/ui/dialog'
import { Button } from '@/components/ui/button'
import { Input } from '@/components/ui/input'
import { Label } from '@/components/ui/label'
import { zigbeeApi } from '@/api/client'
import { pollImportJob, PollAbortedError } from '@/utils/importJobPoll'
import { toast } from 'sonner'
import type { ZigbeeNode, ZigbeeEdge } from './types'
@@ -68,6 +69,10 @@ export function ZigbeeImportModal({ open, onClose, onAddToCanvas, onInventoryImp
const [edges, setEdges] = useState<ZigbeeEdge[]>([])
const [checked, setChecked] = useState<Set<string>>(new Set())
const [importMode, setImportMode] = useState<ImportMode>('pending')
// Aborts the canvas-import poll loop when the modal closes or unmounts.
const pollAbort = useRef<AbortController | null>(null)
useEffect(() => () => pollAbort.current?.abort(), [])
const updateField = (field: keyof ConnectionForm, value: string) =>
setForm((f) => ({
@@ -143,17 +148,27 @@ export function ZigbeeImportModal({ open, onClose, onAddToCanvas, onInventoryImp
onInventoryImported?.(null)
handleClose()
} else {
const res = await zigbeeApi.importNetwork(buildPayload())
setDevices(res.data.nodes)
setEdges(res.data.edges)
setChecked(new Set(res.data.nodes.map((n) => n.id)))
if (res.data.device_count === 0) {
// The backend fetches the network map in the background and we poll for
// it — a large mesh takes minutes, which no reverse proxy will hold open.
pollAbort.current?.abort()
const controller = new AbortController()
pollAbort.current = controller
const start = await zigbeeApi.importNetwork(buildPayload())
const result = await pollImportJob(
async () => (await zigbeeApi.getImportJob(start.data.job_id)).data,
{ signal: controller.signal },
)
setDevices(result.nodes)
setEdges(result.edges)
setChecked(new Set(result.nodes.map((n) => n.id)))
if (result.device_count === 0) {
toast.info('No Zigbee devices found in the network map')
} else {
toast.success(`Found ${res.data.device_count} device${res.data.device_count !== 1 ? 's' : ''}`)
toast.success(`Found ${result.device_count} device${result.device_count !== 1 ? 's' : ''}`)
}
}
} catch (err: unknown) {
if (err instanceof PollAbortedError) return
toast.error(extractError(err) ?? 'Failed to fetch Zigbee devices')
} finally {
setLoading(false)
@@ -181,6 +196,8 @@ export function ZigbeeImportModal({ open, onClose, onAddToCanvas, onInventoryImp
}
const handleClose = () => {
pollAbort.current?.abort()
pollAbort.current = null
setDevices([])
setEdges([])
setChecked(new Set())
@@ -1,11 +1,12 @@
import { describe, it, expect, vi, beforeEach } from 'vitest'
import { render, screen, fireEvent, waitFor } from '@testing-library/react'
import { render, screen, fireEvent, waitFor, act } from '@testing-library/react'
import { ZigbeeImportModal } from '../ZigbeeImportModal'
vi.mock('@/api/client', () => ({
zigbeeApi: {
testConnection: vi.fn(),
importNetwork: vi.fn(),
getImportJob: vi.fn(),
importToPending: vi.fn(),
},
}))
@@ -51,6 +52,7 @@ describe('ZigbeeImportModal', () => {
beforeEach(() => {
vi.mocked(zigbeeApi.testConnection).mockReset()
vi.mocked(zigbeeApi.importNetwork).mockReset()
vi.mocked(zigbeeApi.getImportJob).mockReset()
vi.mocked(zigbeeApi.importToPending).mockReset()
vi.mocked(toast.success).mockReset()
vi.mocked(toast.error).mockReset()
@@ -115,16 +117,33 @@ describe('ZigbeeImportModal', () => {
fireEvent.click(screen.getByRole('radio', { name: /canvas directly/i }))
}
it('fetches devices and renders them grouped by type', async () => {
/** A canvas import now answers with a job id; the payload arrives from a poll. */
const mockCanvasImport = (result: {
nodes: typeof sampleNodes
edges: { source: string; target: string }[]
device_count: number
}) => {
vi.mocked(zigbeeApi.importNetwork).mockResolvedValue({
data: { nodes: sampleNodes, edges: [], device_count: 2 },
data: { job_id: 'job-1', status: 'running' },
} as never)
vi.mocked(zigbeeApi.getImportJob).mockResolvedValue({
data: { job_id: 'job-1', status: 'done', result },
} as never)
}
const startCanvasFetch = () => {
render(<ZigbeeImportModal {...defaultProps} />)
selectCanvasMode()
const hostInput = screen.getByPlaceholderText('192.168.1.x or mqtt.local')
fireEvent.change(hostInput, { target: { value: '192.168.1.100' } })
fireEvent.change(screen.getByPlaceholderText('192.168.1.x or mqtt.local'), {
target: { value: '192.168.1.100' },
})
fireEvent.click(screen.getByRole('button', { name: /fetch devices/i }))
}
it('fetches devices and renders them grouped by type', async () => {
mockCanvasImport({ nodes: sampleNodes, edges: [], device_count: 2 })
startCanvasFetch()
await waitFor(() => {
expect(screen.getByText('Coordinator')).toBeDefined()
@@ -134,15 +153,9 @@ describe('ZigbeeImportModal', () => {
})
it('shows info toast when no devices found', async () => {
vi.mocked(zigbeeApi.importNetwork).mockResolvedValue({
data: { nodes: [], edges: [], device_count: 0 },
} as never)
mockCanvasImport({ nodes: [], edges: [], device_count: 0 })
render(<ZigbeeImportModal {...defaultProps} />)
selectCanvasMode()
const hostInput = screen.getByPlaceholderText('192.168.1.x or mqtt.local')
fireEvent.change(hostInput, { target: { value: '192.168.1.100' } })
fireEvent.click(screen.getByRole('button', { name: /fetch devices/i }))
startCanvasFetch()
await waitFor(() => {
expect(toast.info).toHaveBeenCalledWith('No Zigbee devices found in the network map')
@@ -150,15 +163,13 @@ describe('ZigbeeImportModal', () => {
})
it('calls onAddToCanvas with selected devices and closes modal', async () => {
vi.mocked(zigbeeApi.importNetwork).mockResolvedValue({
data: { nodes: sampleNodes, edges: [{ source: '0x0000', target: '0x0001' }], device_count: 2 },
} as never)
mockCanvasImport({
nodes: sampleNodes,
edges: [{ source: '0x0000', target: '0x0001' }],
device_count: 2,
})
render(<ZigbeeImportModal {...defaultProps} />)
selectCanvasMode()
const hostInput = screen.getByPlaceholderText('192.168.1.x or mqtt.local')
fireEvent.change(hostInput, { target: { value: '192.168.1.100' } })
fireEvent.click(screen.getByRole('button', { name: /fetch devices/i }))
startCanvasFetch()
await waitFor(() => screen.getByText('Coordinator'))
@@ -207,17 +218,81 @@ describe('ZigbeeImportModal', () => {
})
it('switching to canvas mode calls importNetwork and not importToPending', async () => {
vi.mocked(zigbeeApi.importNetwork).mockResolvedValue({
data: { nodes: sampleNodes, edges: [], device_count: 2 },
} as never)
mockCanvasImport({ nodes: sampleNodes, edges: [], device_count: 2 })
render(<ZigbeeImportModal {...defaultProps} />)
fireEvent.click(screen.getByRole('radio', { name: /canvas directly/i }))
const hostInput = screen.getByPlaceholderText('192.168.1.x or mqtt.local')
fireEvent.change(hostInput, { target: { value: '192.168.1.100' } })
fireEvent.click(screen.getByRole('button', { name: /fetch devices/i }))
startCanvasFetch()
await waitFor(() => expect(zigbeeApi.importNetwork).toHaveBeenCalled())
expect(zigbeeApi.importToPending).not.toHaveBeenCalled()
})
it('keeps polling the job until the map is ready (issue #380)', async () => {
vi.useFakeTimers({ shouldAdvanceTime: true })
try {
vi.mocked(zigbeeApi.importNetwork).mockResolvedValue({
data: { job_id: 'job-1', status: 'running' },
} as never)
vi.mocked(zigbeeApi.getImportJob)
.mockResolvedValueOnce({
data: { job_id: 'job-1', status: 'running', result: null },
} as never)
.mockResolvedValueOnce({
data: {
job_id: 'job-1',
status: 'done',
result: { nodes: sampleNodes, edges: [], device_count: 2 },
},
} as never)
startCanvasFetch()
await waitFor(() => expect(zigbeeApi.getImportJob).toHaveBeenCalledTimes(1))
await act(async () => {
await vi.advanceTimersByTimeAsync(2000)
})
await waitFor(() => expect(screen.getByText('router_1')).toBeDefined())
expect(zigbeeApi.getImportJob).toHaveBeenCalledWith('job-1')
} finally {
vi.useRealTimers()
}
})
it('surfaces the backend detail when the job fails', async () => {
vi.mocked(zigbeeApi.importNetwork).mockResolvedValue({
data: { job_id: 'job-1', status: 'running' },
} as never)
vi.mocked(zigbeeApi.getImportJob).mockRejectedValue({
response: { status: 504, data: { detail: 'Timed out waiting for networkmap response' } },
})
startCanvasFetch()
await waitFor(() => {
expect(toast.error).toHaveBeenCalledWith('Timed out waiting for networkmap response')
})
})
it('stops polling and stays quiet when the modal is closed mid-import', async () => {
vi.mocked(zigbeeApi.importNetwork).mockResolvedValue({
data: { job_id: 'job-1', status: 'running' },
} as never)
vi.mocked(zigbeeApi.getImportJob).mockResolvedValue({
data: { job_id: 'job-1', status: 'running', result: null },
} as never)
const { unmount } = render(<ZigbeeImportModal {...defaultProps} />)
selectCanvasMode()
fireEvent.change(screen.getByPlaceholderText('192.168.1.x or mqtt.local'), {
target: { value: '192.168.1.100' },
})
fireEvent.click(screen.getByRole('button', { name: /fetch devices/i }))
await waitFor(() => expect(zigbeeApi.getImportJob).toHaveBeenCalled())
unmount()
const callsAtUnmount = vi.mocked(zigbeeApi.getImportJob).mock.calls.length
await new Promise((r) => setTimeout(r, 50))
expect(vi.mocked(zigbeeApi.getImportJob).mock.calls.length).toBe(callsAtUnmount)
expect(toast.error).not.toHaveBeenCalled()
})
})
@@ -0,0 +1,102 @@
import { describe, it, expect, vi } from 'vitest'
import { pollImportJob, PollAbortedError, type ImportJobState } from '../importJobPoll'
interface Payload { device_count: number }
const done = (device_count: number): ImportJobState<Payload> => ({
status: 'done',
result: { device_count },
})
const running: ImportJobState<Payload> = { status: 'running', result: null }
/** Resolves instantly so no test waits on a real timer. */
const noSleep = () => Promise.resolve()
describe('pollImportJob', () => {
it('returns the result when the first poll is already done', async () => {
const fetchJob = vi.fn().mockResolvedValue(done(2))
await expect(pollImportJob(fetchJob, { sleep: noSleep })).resolves.toEqual({ device_count: 2 })
expect(fetchJob).toHaveBeenCalledTimes(1)
})
it('keeps polling while the job is running', async () => {
const fetchJob = vi
.fn()
.mockResolvedValueOnce(running)
.mockResolvedValueOnce(running)
.mockResolvedValueOnce(done(7))
await expect(pollImportJob(fetchJob, { sleep: noSleep })).resolves.toEqual({ device_count: 7 })
expect(fetchJob).toHaveBeenCalledTimes(3)
})
it('waits the configured interval between polls', async () => {
const sleep = vi.fn().mockResolvedValue(undefined)
const fetchJob = vi.fn().mockResolvedValueOnce(running).mockResolvedValueOnce(done(1))
await pollImportJob(fetchJob, { intervalMs: 1234, sleep })
expect(sleep).toHaveBeenCalledTimes(1)
expect(sleep.mock.calls[0][0]).toBe(1234)
})
it('propagates a rejected poll so the caller sees the backend status', async () => {
const err = Object.assign(new Error('boom'), {
response: { status: 504, data: { detail: 'Timed out' } },
})
const fetchJob = vi.fn().mockResolvedValueOnce(running).mockRejectedValueOnce(err)
await expect(pollImportJob(fetchJob, { sleep: noSleep })).rejects.toBe(err)
})
it('rejects with PollAbortedError when aborted before the first poll', async () => {
const controller = new AbortController()
controller.abort()
const fetchJob = vi.fn().mockResolvedValue(done(1))
await expect(
pollImportJob(fetchJob, { signal: controller.signal, sleep: noSleep }),
).rejects.toBeInstanceOf(PollAbortedError)
expect(fetchJob).not.toHaveBeenCalled()
})
it('rejects with PollAbortedError when aborted while a poll is in flight', async () => {
const controller = new AbortController()
const fetchJob = vi.fn().mockImplementation(async () => {
controller.abort()
return running
})
await expect(
pollImportJob(fetchJob, { signal: controller.signal, sleep: noSleep }),
).rejects.toBeInstanceOf(PollAbortedError)
expect(fetchJob).toHaveBeenCalledTimes(1)
})
it('rejects when the job reports done with no payload', async () => {
const fetchJob = vi.fn().mockResolvedValue({ status: 'done', result: null })
await expect(pollImportJob(fetchJob, { sleep: noSleep })).rejects.toThrow(
'Import finished without a result',
)
})
it('default sleep resolves after the interval and rejects on abort', async () => {
vi.useFakeTimers()
try {
const fetchJob = vi.fn().mockResolvedValueOnce(running).mockResolvedValueOnce(done(1))
const promise = pollImportJob(fetchJob, { intervalMs: 50 })
await vi.advanceTimersByTimeAsync(60)
await expect(promise).resolves.toEqual({ device_count: 1 })
const controller = new AbortController()
const aborting = pollImportJob(
vi.fn().mockResolvedValue(running),
{ intervalMs: 50, signal: controller.signal },
)
await vi.advanceTimersByTimeAsync(1)
controller.abort()
await expect(aborting).rejects.toBeInstanceOf(PollAbortedError)
} finally {
vi.useRealTimers()
}
})
})
+73
View File
@@ -0,0 +1,73 @@
/**
* Polling helper for backend import jobs that answer immediately and finish later.
*
* A Zigbee canvas import has to fetch a Z2M networkmap, which on a large mesh
* takes minutes longer than the read timeout of any reverse proxy in front of
* the API. The backend therefore returns a job id and does the work in the
* background; the client polls short requests until the payload is ready.
*/
export class PollAbortedError extends Error {
constructor() {
super('Import polling aborted')
this.name = 'PollAbortedError'
}
}
export interface ImportJobState<T> {
status: string
result: T | null
}
export interface PollImportJobOptions {
/** Delay between polls, in ms. */
intervalMs?: number
/** Abort the loop (modal closed, component unmounted). */
signal?: AbortSignal
/** Injected in tests so no timer actually runs. */
sleep?: (ms: number, signal?: AbortSignal) => Promise<void>
}
const defaultSleep = (ms: number, signal?: AbortSignal) =>
new Promise<void>((resolve, reject) => {
const timer = setTimeout(() => {
signal?.removeEventListener('abort', onAbort)
resolve()
}, ms)
const onAbort = () => {
clearTimeout(timer)
reject(new PollAbortedError())
}
if (signal?.aborted) {
onAbort()
return
}
signal?.addEventListener('abort', onAbort, { once: true })
})
/**
* Poll `fetchJob` until it reports a terminal status, then resolve its result.
*
* A failed job surfaces as a rejected `fetchJob` call (the backend replays the
* original status code), so errors propagate untouched to the caller. Rejects
* with `PollAbortedError` if the signal fires, and with a plain Error if the
* job reports done without a payload.
*/
export async function pollImportJob<T>(
fetchJob: () => Promise<ImportJobState<T>>,
{ intervalMs = 2000, signal, sleep = defaultSleep }: PollImportJobOptions = {},
): Promise<T> {
for (;;) {
if (signal?.aborted) throw new PollAbortedError()
const job = await fetchJob()
if (signal?.aborted) throw new PollAbortedError()
if (job.status !== 'running') {
if (job.result == null) throw new Error('Import finished without a result')
return job.result
}
await sleep(intervalMs, signal)
}
}