diff --git a/.github/workflows/schedule-tests.yml b/.github/workflows/schedule-tests.yml new file mode 100644 index 00000000..00bf72b0 --- /dev/null +++ b/.github/workflows/schedule-tests.yml @@ -0,0 +1,52 @@ +name: Schedule support checks + +on: + push: + branches: [upstream/schedule-support, upstream/admin-schedules] + pull_request: + workflow_dispatch: + +permissions: + contents: read + +concurrency: + group: schedule-tests-${{ github.ref }} + cancel-in-progress: true + +jobs: + test: + runs-on: ubuntu-latest + strategy: + fail-fast: false + matrix: + python: ['3.11', '3.13'] + steps: + - uses: actions/checkout@v6 + - uses: actions/setup-python@v6 + with: + python-version: ${{ matrix.python }} + - uses: astral-sh/setup-uv@v7 + with: + enable-cache: true + - run: uv sync --locked --extra dev + - run: uv run pytest -q + + docker: + needs: test + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v6 + - uses: docker/setup-buildx-action@v4 + - uses: docker/build-push-action@v7 + with: + context: . + platforms: linux/amd64 + push: false + load: true + tags: local/roborock-server:schedule-ci + cache-from: type=gha + cache-to: type=gha,mode=max + - name: Verify schedule module in the built image + run: >- + docker run --rm --network none local/roborock-server:schedule-ci + python -c 'import roborock_local_server.bundled_backend.shared.schedule_runner' diff --git a/docs/known_limitations.md b/docs/known_limitations.md index d81d24b0..c4c897a5 100644 --- a/docs/known_limitations.md +++ b/docs/known_limitations.md @@ -4,6 +4,6 @@ I have tried to copy over most functionality into this server, but some is not 1 - Vacuum camera live streaming — supported with a TURN relay and a short camera bootstrap hostname; see [camera.md](camera.md). Verified on the Saros 10R and Qrevo MaxV; other camera models may bootstrap differently and need validation. - Firmware Updates -- Schedules (Routines/Scenes do work though) +- Schedules are experimental. A scheduled start has been reported on an S7; broader hardware compatibility, DST parity, and expired one-time jobs still need validation. See [schedule support](schedules.md). Routines/Scenes are separate. - China-market devices do not work outside of China. Their firmware checks the region they are used in (via `rrcheck`), and after local onboarding they reject ordinary commands such as `get_status` with `-10002 / rrcheck access denied`. See [#81](https://github.com/Python-roborock/local_roborock_server/issues/81). - Some users have gotten China-market vacuums working outside China on the official cloud by routing the robot's traffic through a VPN that exits in China (the robot's WAN IP is what matters, not the phone's). See [this comment](https://github.com/iobroker-community-adapters/ioBroker.mihome-vacuum/issues/842#issuecomment-1736230673). This has not been tested with this local server yet, so it may or may not help. diff --git a/docs/schedules.md b/docs/schedules.md new file mode 100644 index 00000000..25208eef --- /dev/null +++ b/docs/schedules.md @@ -0,0 +1,154 @@ +# Experimental schedule support + +This implementation provides schedule storage, job APIs, time calculation, and dispatch. It does +not establish complete Roborock cloud compatibility on every model. + +This work integrates [upstream PR #179 by virtitnerd](https://github.com/Python-roborock/local_roborock_server/pull/179) +at `59bf2e6de0a0f62b371ac58a41fb44bec5ddc603`, with additional changes based on +cloud captures from `roborock.vacuum.a15` and `roborock.vacuum.a27`. Those captures +verify protocol shapes and client behavior; a tester subsequently confirmed a scheduled start on an S7 (a15, firmware +02.16.12) using this local server, with LocalRock and the OEM Roborock app used +for schedule control. This confirms one scheduled-start scenario, not all +cleaning settings, failure cases, or other models. Raw captures and credentials are excluded. + +## Responsibilities and compatibility + +Homebridge retains schedule switches, pause/delay decisions, original-time +restoration, daily reset, and recovery. The server provides the underlying APIs. + +| Operation | Behavior in this branch | +| --- | --- | +| GET `/user/devices/{duid}/jobs` | Stored jobs with calculated UTC `nextFireTime` for enabled, valid jobs | +| POST to the collection | Persist a new job with a unique numeric ID | +| PUT/PATCH `/user/devices/{duid}/jobs/{id}` | Verify native enablement when supplied, then save; preserve omitted fields and unknown model fields | +| DELETE an item | Remove the stored job and its occurrence record | +| `get/set/upd/del_server_timer` | Forward unchanged to the robot, which owns these timer entries | +| Legacy `get/set/upd/del_timer` | Local compatibility view for jobs without a native timer name | +| Native scheduled execution | Verify the robot timer is on, then send `server_scheduled_start` with the complete settings entry | + +A job's numeric `id` is different from the robot timer's string ID in +`param.params[0].name`. Imported or previously unsynchronized jobs can disagree with the native timer. +The server now reconciles enablement and still checks both before execution. +Native jobs are not duplicated in the legacy `get_timer` view; both captured +robots returned an empty legacy list while holding native timers. + +Time updates accept the existing-job JSON PUT used by Homebridge, without +requiring `id` or `nextFireTime` in the body. If `param` is supplied, it replaces +the previous object exactly, so app removal of an optional field is preserved. +Settings such as `clean_order_mode`, `mop_template_id`, `map_index`, room-list +strings, and unknown future options survive storage and native dispatch. +Invalid cron/timezone/body values are rejected. A persistence failure does not +receive a successful write acknowledgement. + +## Enablement synchronization across clients + +Native schedule on/off changes are synchronized by the server, independent of +whether Homebridge, LocalRock, the OEM app, or the dashboard made the change. +A forwarded `upd/set/del_server_timer` request prompts a fresh native timer read. +The request and the robot's reply remain unchanged; sending a request or receiving +an `ok` is not used as proof of the final state. No client-specific HTTP write is +required after a native toggle. + +A background read also runs at startup and approximately every 30 seconds to +recover missed traffic, reconnects, and changes made directly over the LAN. +Successful reconciliation changes only `enabled` on existing, uniquely matched +jobs. It preserves IDs, cron/timezone, settings, unknown fields, and occurrence +claims. The dashboard's normal refresh shows the updated saved state. Other +clients see it on their next normal read; this is eventual synchronization, +not an unsolicited push into every app. + +An HTTP job PUT/PATCH that supplies `enabled` verifies the matching native timer +before saving. A timer already in the requested state needs no extra write, +so the OEM app's native-update-then-PUT sequence remains compatible. A lost +acknowledgement is checked with one read-back, without repeating the write. +Time-only requests that omit `enabled` remain store-only edits. + +Background reads and server edits share a per-vacuum lock. A native write that +overlaps a background read invalidates that read; edits/deletions during a read +are checked again before persistence. Missing, malformed, ambiguous or unreachable +timers leave the last saved state intact and never create replacement timers. +A normally polled, still-on timer does not re-enable a consumed one-time job. +An explicit native enable request can do so only after a fresh matching read; +that intent expires after two minutes. Unlinked server-only jobs retain their +existing behavior. Native timer time/settings edits are not imported into job +cron/settings by this enablement reconciliation. + +## Execution and missed runs + +The scheduler evaluates five-field cron expressions in each job's IANA timezone. +It checks approximately every 20 seconds, allowing less than 60 seconds of normal +dispatch latency. Startup establishes a fresh baseline; suspended/late ticks do +not drain historical jobs. Recurring jobs remain enabled for future occurrences. + +Before dispatch, the server saves an occurrence claim in the persistent inventory +under `schedule_occurrences`. A repeated tick, restart, or backward clock change +cannot dispatch the same claimed occurrence again. If that write fails, no start +is attempted. This is an at-most-once policy: a crash after claiming but before +sending can miss a clean. Run only one server process against an inventory; the +inventory lock does not coordinate multiple containers or external file editors. +Cloud re-import replaces schedule definitions, so back up local edits before +refreshing. Occurrence claims are retained across that refresh. + +A fresh native timer read must find exactly one matching `on` entry. Off, missing, +ambiguous, malformed, or unreachable timer state prevents a start. After that +network wait, dispatch rechecks the stored job and its deadline, so a concurrent +time edit or deletion cancels stale work. The remaining read-to-start race is +resolved by the robot's native command; real firmware behavior needs testing. + +A refused or timed-out native scheduled start is never retried as `app_start` or +`app_segment_clean`, and action-locked native starts are not automatically retried. +A robot refusal can mean a disabled/deleted timer; substituting an ordinary clean +could bypass that state. Legacy jobs retain upstream's separate cleaning-command +path; they do not establish native timer support on other firmware. + +In the observed cloud offline experiment, a weekday 14:09 occurrence did not run +while the robot's internet access was blocked. The cloud advanced `nextFireTime` +to the following Monday, and no catch-up run was observed through 14:17 after +access was restored. This supports skipping that missed occurrence. It does not +prove all outage durations, server-restart semantics, or the exact cloud start +packet. Homebridge cannot observe another MQTT client's private command stream. + +## Provisional behavior and remaining validation + +- Native dispatch is based on the captured job's method/settings and upstream + hardware reports. The exact cloud-to-robot start packet was not captured. + Validate room selection, map, cleaning order, mop mode, and history on a15/a27. +- DST policy currently skips nonexistent spring-forward times and runs only the + first occurrence of an ambiguous fall-back time. Automated tests cover this + policy; matching cloud behavior has not been established. +- One-time jobs are consumed when evaluated as due, including when dispatch + subsequently fails. Expired imported one-time jobs and extended downtime need + further investigation: a five-field cron has no year and can match again in a + later year. Begin hardware testing with recurring jobs. +- Test create/edit/delete through the actual app, disabled/missing native timers, + robot reboot, server/container restart, offline/reconnect, and busy/DND refusal. +- B01 and other unobserved scheduling protocols are not established by these + captures. Scene/routine scheduling is separate from these device job routes. +- A startup log saying a job task was accepted is not proof of a clean; check + subsequent timer-read/start replies, robot status, and cleaning history. + +## First hardware test + +Build the selected Git commit into a versioned Docker image. The CI runs the full Python suite on 3.11 and +3.13 and builds an amd64 Docker image without publishing or deploying it. + +1. Prepare the runtime configuration, DNS/certificate path, data backup, and + explicit cloud-to-local onboarding procedure. Start with one robot. +2. Import current jobs and verify every numeric job ID maps to the expected + native timer name. Confirm local jobs reads and robot timer reads separately. +3. Disable a recurring test schedule through Homebridge. Confirm the robot timer + reads off and the saved job becomes disabled without a separate HTTP update. + Re-enable it and confirm both states agree and a later occurrence runs once. +4. Test a room schedule and a whole-home schedule. Compare settings, rooms, map, + and history against the app. A refusal should stop dispatch, with no fallback + ordinary clean. +5. Exercise Homebridge delay and cancellation/restoration. Confirm only intended + times change, settings survive, and each resulting occurrence runs once. +6. Disconnect the robot across a due time and reconnect. Verify no replay, then + a future occurrence. Repeat with a server restart across the due time and + after a run. Test pending edits while a timer read is delayed. + +Record image tag/commit, model and firmware, timezone, expected time, actual time, +timer state, settings, and relevant sanitized replies for each result. Returning +to the cloud requires the appropriate onboarding transition; stopping Docker is +not an automatic cloud/local switch. diff --git a/mkdocs.yml b/mkdocs.yml index 9bcc2567..d0bbc454 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -60,6 +60,7 @@ nav: - Reference: - Log Retention: logging.md - Known Limitations: known_limitations.md + - Schedule Support (Experimental): schedules.md - Scene Validity Reports: scene_validity.md - Tested Vacuums: tested_vacuums.md - Reverse Proxy: reverse_proxy.md diff --git a/src/roborock_local_server/bundled_backend/https_server/endpoint_rules.py b/src/roborock_local_server/bundled_backend/https_server/endpoint_rules.py index 7aa3c233..e616716b 100644 --- a/src/roborock_local_server/bundled_backend/https_server/endpoint_rules.py +++ b/src/roborock_local_server/bundled_backend/https_server/endpoint_rules.py @@ -114,7 +114,13 @@ from .routes.user.deviceshare import match_received_devices as _match_get_received_devices from .routes.user.deviceshare import match_rooms as _match_get_shared_device_rooms from .routes.user.devices.jobs import build as _build_get_schedules +from .routes.user.devices.jobs import build_create as _build_create_schedule +from .routes.user.devices.jobs import build_delete as _build_delete_schedule +from .routes.user.devices.jobs import build_update as _build_update_schedule from .routes.user.devices.jobs import match as _match_get_schedules +from .routes.user.devices.jobs import match_create as _match_create_schedule +from .routes.user.devices.jobs import match_delete as _match_delete_schedule +from .routes.user.devices.jobs import match_update as _match_update_schedule from .routes.user.devices.newadd import build as _build_add_device from .routes.user.devices.newadd import match as _match_add_device from .routes.user.homes.item import build as _build_get_home_data @@ -596,6 +602,9 @@ def default_endpoint_rules() -> Sequence[EndpointRule]: EndpointRule("get_device", _match_get_device, _build_get_device), EndpointRule("get_device_extra", _match_get_device_extra, _build_get_device_extra), EndpointRule("get_schedules", _match_get_schedules, _build_get_schedules), + EndpointRule("create_schedule", _match_create_schedule, _build_create_schedule), + EndpointRule("update_schedule", _match_update_schedule, _build_update_schedule), + EndpointRule("delete_schedule", _match_delete_schedule, _build_delete_schedule), EndpointRule("post_app_info", _match_post_app_info, _build_post_app_info), EndpointRule("get_inbox_latest", _match_get_inbox_latest, _build_get_inbox_latest), EndpointRule("get_products_v4", _match_get_products_v4, _build_get_products_v4), diff --git a/src/roborock_local_server/bundled_backend/https_server/routes/user/devices/jobs.py b/src/roborock_local_server/bundled_backend/https_server/routes/user/devices/jobs.py index d0ab3239..20291c82 100644 --- a/src/roborock_local_server/bundled_backend/https_server/routes/user/devices/jobs.py +++ b/src/roborock_local_server/bundled_backend/https_server/routes/user/devices/jobs.py @@ -1,17 +1,26 @@ from __future__ import annotations +import json import re from typing import Any +from zoneinfo import ZoneInfo, ZoneInfoNotFoundError +from shared.constants import DEFAULT_TIMEZONE from shared.context import ServerContext +from shared.data_helpers import as_bool from shared.http_helpers import wrap_response +from shared.schedule_runner import parse_cron +from shared.schedule_store import SettingsError, delete_job, list_jobs, native_timer_name, upsert_job +from shared.native_schedule_sync import TimerSyncError, sync_timer from .service import device_jobs_payload +_COLLECTION_RE = re.compile(r"/user/devices/([^/]+)/jobs") +_ITEM_RE = re.compile(r"/user/devices/([^/]+)/jobs/([^/]+)") -def match(path: str) -> bool: - clean = path.rstrip("/") - return bool(re.fullmatch(r"/user/devices/[^/]+/jobs", clean)) + +def match(path: str, method: str = "GET") -> bool: + return method.upper() == "GET" and bool(_COLLECTION_RE.fullmatch(path.rstrip("/"))) def build( @@ -24,3 +33,182 @@ def build( device_id = parts[-2] if len(parts) >= 2 else "" return wrap_response(device_jobs_payload(ctx, device_id)) + +def _error(code: int, message: str) -> dict[str, Any]: + return {"success": False, "code": code, "msg": message, "data": None, "result": None} + + +def _json_body(body_params: dict[str, list[str]]) -> dict[str, Any]: + for raw in body_params.get("__json") or []: + try: + parsed = json.loads(raw) + except (TypeError, json.JSONDecodeError): + continue + if isinstance(parsed, dict): + return parsed + raise ValueError("request body must be a JSON object") + + +def _validate_zone(value: Any) -> str: + name = str(value or "").strip() + try: + ZoneInfo(name) + except (ZoneInfoNotFoundError, ValueError) as exc: + raise ValueError("timeZoneId must be a valid IANA timezone") from exc + return name + + +def _parse_param(value: Any) -> dict[str, Any] | None: + if isinstance(value, str) and value.strip(): + try: + value = json.loads(value) + except json.JSONDecodeError: + return None + return value if isinstance(value, dict) else None + + +def match_create(path: str, method: str = "GET") -> bool: + return method.upper() == "POST" and bool(_COLLECTION_RE.fullmatch(path.rstrip("/"))) + + +def build_create( + ctx: ServerContext, + _query_params: dict[str, list[str]], + body_params: dict[str, list[str]], + clean_path: str, +) -> dict[str, Any]: + duid = _COLLECTION_RE.fullmatch(clean_path.rstrip("/")).group(1) # type: ignore[union-attr] + try: + body = _json_body(body_params) + cron = str(body.get("cron") or "").strip() + parse_cron(cron) + zone = _validate_zone(body.get("timeZoneId", getattr(ctx, "timezone", "") or DEFAULT_TIMEZONE)) + except ValueError as exc: + return _error(400, str(exc)) + param = _parse_param(body.get("param")) + if body.get("param") is not None and param is None: + return _error(400, "param must be a JSON object") + try: + created = upsert_job( + ctx, duid, job_id=body.get("id"), cron=cron, + enabled=as_bool(body.get("enabled", True), True), time_zone=zone, + repeated=as_bool(body.get("repeated", True), True), param=param, + extra_fields=body, create_only=True, + ) + except ValueError as exc: + return _error(409, str(exc)) + except OSError: + return _error(500, "unable to persist schedule") + return wrap_response(created) + + +def match_update(path: str, method: str = "GET") -> bool: + return method.upper() in ("PUT", "PATCH") and bool(_ITEM_RE.fullmatch(path.rstrip("/"))) + + +class ScheduleUpdateError(ValueError): + def __init__(self, code: int, message: str) -> None: + super().__init__(message) + self.code = code + + +def _prepare_update(ctx, duid, job_id, body_params): + matches = [job for job in list_jobs(ctx, duid) if str(job.get("id")) == job_id] + if not matches: + raise ScheduleUpdateError(404, "schedule not found") + if len(matches) != 1: + raise ScheduleUpdateError(409, "duplicate schedule IDs") + existing = matches[0] + try: + body = _json_body(body_params) + cron = str(body.get("cron", existing.get("cron")) or "").strip() + parse_cron(cron) + zone = _validate_zone(body["timeZoneId"]) if "timeZoneId" in body else "" + except ValueError as exc: + raise ScheduleUpdateError(400, str(exc)) from exc + param = _parse_param(body.get("param")) if "param" in body else None + if "param" in body and param is None: + raise ScheduleUpdateError(400, "param must be a JSON object") + values = dict( + job_id=job_id, cron=cron, + enabled=as_bool(body.get("enabled", existing.get("enabled", True)), True), + time_zone=zone, + repeated=as_bool(body["repeated"], True) if "repeated" in body else None, + param=param, override_time_zone=bool(zone), extra_fields=body, expected_job=existing, + ) + return existing, body, values + + +def _save_update(ctx, duid, values): + try: + return wrap_response(upsert_job(ctx, duid, **values)) + except ValueError as exc: + return _error(409, str(exc)) + except OSError: + return _error(500, "unable to persist schedule") + + +def build_update(ctx, _query_params, body_params, clean_path): + match_ = _ITEM_RE.fullmatch(clean_path.rstrip("/")) + duid, job_id = match_.group(1), match_.group(2) + try: + _, _, values = _prepare_update(ctx, duid, job_id, body_params) + except ScheduleUpdateError as exc: + return _error(exc.code, str(exc)) + return _save_update(ctx, duid, values) + + +async def build_update_synced(ctx, _query_params, body_params, clean_path, coordinator): + """Authenticated HTTP edits share the same native state as every MQTT client.""" + match_ = _ITEM_RE.fullmatch(clean_path.rstrip("/")) + duid, job_id = match_.group(1), match_.group(2) + async with coordinator.lock(duid): + try: + existing, body, values = _prepare_update(ctx, duid, job_id, body_params) + candidate = {**existing} + if values["param"] is not None: + candidate["param"] = values["param"] + name = native_timer_name(candidate) if "enabled" in body else None + + def check_current(): + jobs = list_jobs(ctx, duid) + if [job for job in jobs if str(job.get("id")) == job_id] != [existing]: + raise TimerSyncError("Schedule changed during timer verification; read it again", 409) + for other in jobs: + if str(other.get("id")) != job_id and native_timer_name(other) == name: + raise TimerSyncError("Multiple schedules reference this native timer", 409) + + if name is not None: + check_current() + await sync_timer(ctx, duid, name, values["enabled"], check_current=check_current) + check_current() + result = _save_update(ctx, duid, values) + if name is not None and not result.get("success"): + result["msg"] = "Robot timer was verified, but the schedule could not be saved; refresh before retrying" + return result + except ScheduleUpdateError as exc: + return _error(exc.code, str(exc)) + except SettingsError as exc: + return _error(409, str(exc)) + except TimerSyncError as exc: + return _error(exc.status, str(exc)) + + +def match_delete(path: str, method: str = "GET") -> bool: + return method.upper() == "DELETE" and bool(_ITEM_RE.fullmatch(path.rstrip("/"))) + + +def build_delete( + ctx: ServerContext, + _query_params: dict[str, list[str]], + _body_params: dict[str, list[str]], + clean_path: str, +) -> dict[str, Any]: + match_ = _ITEM_RE.fullmatch(clean_path.rstrip("/")) + duid, job_id = match_.group(1), match_.group(2) # type: ignore[union-attr] + try: + if not delete_job(ctx, duid, job_id): + return _error(404, "schedule not found") + except OSError: + return _error(500, "unable to persist schedule") + return wrap_response({"id": job_id}) diff --git a/src/roborock_local_server/bundled_backend/https_server/routes/user/devices/service.py b/src/roborock_local_server/bundled_backend/https_server/routes/user/devices/service.py index 0f6990c2..ff4e837c 100644 --- a/src/roborock_local_server/bundled_backend/https_server/routes/user/devices/service.py +++ b/src/roborock_local_server/bundled_backend/https_server/routes/user/devices/service.py @@ -13,6 +13,7 @@ from shared.context import ServerContext from shared.data_helpers import as_bool, as_int, default_product_name, get_value from shared.inventory_io import WEB_API_INVENTORY_FILE, load_inventory +from shared.schedule_runner import next_fire_time_utc _DEFAULT_DEVICE_NAME = "Vacuum 1" @@ -652,6 +653,11 @@ def device_jobs_payload(ctx: ServerContext, device_id: str) -> list[dict[str, An payload["repeated"] = as_bool(get_value(raw, "repeated", default=True), True) payload["enabled"] = as_bool(get_value(raw, "enabled", default=True), True) payload["param"] = get_value(raw, "param", default={}) or {} + upcoming = next_fire_time_utc(payload, default_timezone=str(getattr(ctx, "timezone", "") or "")) + if upcoming is not None: + payload["nextFireTime"] = upcoming.strftime("%Y-%m-%dT%H:%M:%S.000+00:00") + else: + payload.pop("nextFireTime", None) schedules.append(payload) return schedules diff --git a/src/roborock_local_server/bundled_backend/import_web_inventory_from_cloud.py b/src/roborock_local_server/bundled_backend/import_web_inventory_from_cloud.py index b79dfd03..f97bb291 100644 --- a/src/roborock_local_server/bundled_backend/import_web_inventory_from_cloud.py +++ b/src/roborock_local_server/bundled_backend/import_web_inventory_from_cloud.py @@ -9,6 +9,7 @@ import argparse import asyncio +import copy from dataclasses import asdict, is_dataclass from datetime import datetime, timezone import json @@ -757,6 +758,12 @@ def _merge_existing_inventory_mutations( merged = dict(cloud_inventory) + # Refreshing cloud definitions must not erase the local scheduler's record + # of occurrences already attempted (including a clean with a lost reply). + occurrences = existing_inventory.get("schedule_occurrences") + if isinstance(occurrences, dict): + merged["schedule_occurrences"] = copy.deepcopy(occurrences) + rooms = _merge_record_lists_by_identity( _inventory_rooms(merged), _inventory_rooms(existing_inventory), diff --git a/src/roborock_local_server/bundled_backend/mqtt_broker_server/topic_bridge.py b/src/roborock_local_server/bundled_backend/mqtt_broker_server/topic_bridge.py index 7bc087a0..a21b890f 100644 --- a/src/roborock_local_server/bundled_backend/mqtt_broker_server/topic_bridge.py +++ b/src/roborock_local_server/bundled_backend/mqtt_broker_server/topic_bridge.py @@ -9,7 +9,7 @@ from pathlib import Path import re import time -from typing import Any +from typing import Any, Callable import aiomqtt @@ -73,7 +73,13 @@ def __init__( fixed_device_mqtt_usr: str = "", runtime_state: Any | None = None, inventory_path: Path | None = None, + request_interceptor: Callable[[str, bytes], bytes | None] | None = None, + request_observer: Callable[[str, bytes], None] | None = None, ) -> None: + # Given (duid, payload) for each rr/m request; returns an encoded reply to + # publish back to the requester instead of forwarding, or None to forward. + self._request_interceptor = request_interceptor + self._request_observer = request_observer self._host = host self._port = port self._logger = logger @@ -389,6 +395,19 @@ async def _handle_cloud_message( cloud_topic: CloudTopicKey, src_topic: str, ) -> None: + if self._request_interceptor is not None: + payload = message.payload + raw = payload if isinstance(payload, (bytes, bytearray)) else str(payload).encode() + try: + reply = self._request_interceptor(cloud_topic.duid, bytes(raw)) + except Exception as exc: # noqa: BLE001 + self._logger.warning("MQTT request interceptor failed for %s: %s", src_topic, exc) + reply = None + if reply is not None: + await client.publish(cloud_topic.topic_out, reply, qos=_extract_qos(message)) + self._logger.info("MQTT bridge answered %s locally -> %s", src_topic, cloud_topic.topic_out) + return + device_topic = self._resolve_device_for_cloud(cloud_topic) if device_topic is None: if cloud_topic.duid not in self._warned_unmapped_duids: @@ -416,6 +435,11 @@ async def _handle_cloud_message( self._d_to_m[device_topic] = cloud_topic target_topic = device_topic.topic_out + if self._request_observer is not None and not bool(getattr(message, "retain", False)): + try: + self._request_observer(cloud_topic.duid, bytes(message.payload)) + except Exception: # Observation must never prevent forwarding. + self._logger.warning("MQTT request observer failed for %s", src_topic, exc_info=True) await self._publish_mirror(client, message, target_topic) self._logger.info( "MQTT bridge m->d %s -> %s bytes=%d", diff --git a/src/roborock_local_server/bundled_backend/shared/native_schedule_sync.py b/src/roborock_local_server/bundled_backend/shared/native_schedule_sync.py new file mode 100644 index 00000000..324e37dc --- /dev/null +++ b/src/roborock_local_server/bundled_backend/shared/native_schedule_sync.py @@ -0,0 +1,268 @@ +"""Reconcile native schedule enablement independently of the client making a change.""" + +from __future__ import annotations + +import asyncio +from collections.abc import Callable +from enum import StrEnum +import logging +import time +from typing import Any + +from .inventory_io import load_inventory +from .schedule_store import SettingsError, list_jobs, native_timer_name, reconcile_native_enabled + +_LOGGER = logging.getLogger(__name__) + + +class TimerSyncError(RuntimeError): + def __init__(self, message: str, status: int = 503) -> None: + super().__init__(message) + self.status = status + + +class TimerCommand(StrEnum): + GET = "get_server_timer" + UPDATE = "upd_server_timer" + + +def _timer_state(timers: Any, name: str) -> str: + if not isinstance(timers, list): + raise TimerSyncError("The vacuum did not return a usable timer list. Refresh schedules and retry.") + matches = [entry for entry in timers if isinstance(entry, list) and entry and entry[0] == name] + if not matches: + raise TimerSyncError("The linked robot timer is missing. Check this schedule in the Roborock app before retrying.", 409) + if len(matches) != 1 or len(matches[0]) < 2 or matches[0][1] not in ("on", "off"): + raise TimerSyncError("The linked robot timer is ambiguous. Check this schedule in the Roborock app before retrying.", 409) + return matches[0][1] + + +async def sync_timer( + context: Any, duid: str, name: str, enabled: bool, *, check_current: Callable[[], None], +) -> str: + """Update only the named timer; a fresh read is the success criterion. + + The app sends upd_server_timer before its HTTP job update. A timed-out write + may still have reached the vacuum, so read it back without repeating the + write. Never create a timer or infer one from a cloud job's numeric ID. + """ + from shared.routine_runner import _RoutineMqttClient + + device = context.runtime_credentials.resolve_device(duid=duid) + if device is None or not device.get("localkey"): + raise TimerSyncError("Connect this vacuum to the local server before changing its schedule.") + desired = "on" if enabled else "off" + client = None + write_attempted = False + try: + async with asyncio.timeout(45): + client = _RoutineMqttClient(context, device, logging.getLogger(__name__)) + await client.connect() + current = _timer_state(await client.send_command(TimerCommand.GET, []), name) + # Recheck the job after the network read, before changing the robot. + check_current() + if current != desired: + write_attempted = True + try: + await client.send_command(TimerCommand.UPDATE, [[name, desired]]) + except Exception: + # A read-back can confirm a write whose acknowledgement was + # lost. It cannot confirm a refused or unapplied write. + pass + current = _timer_state(await client.send_command(TimerCommand.GET, []), name) + if current != desired: + raise TimerSyncError("The vacuum did not confirm the requested timer setting. Refresh schedules and retry.") + return current + except TimerSyncError: + raise + except Exception as exc: + message = ( + "The robot timer may have changed, but verification failed. Refresh schedules and retry." + if write_attempted else + "Could not read this vacuum's timer. Make sure it is connected locally, then retry." + ) + raise TimerSyncError(message) from exc + finally: + if client is not None: + try: + async with asyncio.timeout(5): + await client.close() + except Exception: + logging.getLogger(__name__).warning("Unable to close schedule timer connection") + + +def timer_states(timers: Any) -> dict[str, bool]: + """Ignore ambiguous/malformed entries; absence is not proof of disablement.""" + if not isinstance(timers, list): + raise TimerSyncError("The vacuum did not return a usable timer list") + states: dict[str, bool] = {} + seen: set[str] = set() + invalid: set[str] = set() + for entry in timers: + if not isinstance(entry, list) or not entry or not isinstance(entry[0], str): + continue + name = entry[0] + if name in seen or len(entry) < 2 or entry[1] not in ("on", "off"): + invalid.add(name) + else: + states[name] = entry[1] == "on" + seen.add(name) + return {name: enabled for name, enabled in states.items() if name not in invalid} + + +class NativeScheduleSync: + """One coordinator for API edits and read-only reconciliation from any client. + + Native RPCs still reach the robot unchanged. Observing a write only schedules + a fresh read; no optimistic request or generic 'ok' reply mutates inventory. + Periodic reads also catch LAN changes and recover after missed MQTT traffic. + """ + + INTERVAL = 30.0 + DEBOUNCE = 0.3 + + def __init__(self, context: Any) -> None: + self.context = context + self._locks: dict[str, asyncio.Lock] = {} + self._generation: dict[str, int] = {} + self._enable_intent: dict[str, dict[str, tuple[float, Any]]] = {} + self._dirty: set[str] = set() + self._wake = asyncio.Event() + self._slots = asyncio.Semaphore(2) + + def lock(self, duid: str) -> asyncio.Lock: + return self._locks.setdefault(duid, asyncio.Lock()) + + def changed(self, duid: str, *, enable_name: str | None = None) -> None: + self._generation[duid] = self._generation.get(duid, 0) + 1 + if enable_name is not None: + # Tie one-time re-enablement to the occurrence already consumed when + # the request was observed, never one consumed after this request. + claims = load_inventory(self.context).get("schedule_occurrences", {}).get(duid, {}) + for job in list_jobs(self.context, duid): + try: + matches = native_timer_name(job) == enable_name + except SettingsError: + continue + if matches: + self._enable_intent.setdefault(duid, {})[enable_name] = ( + time.monotonic() + 120, claims.get(str(job.get("id"))), + ) + self._dirty.add(duid) + self._wake.set() + + def observe_request(self, duid: str, payload: bytes) -> None: + from roborock.protocol import create_mqtt_decoder + from .timer_rpc import _request_body + + device = self.context.runtime_credentials.resolve_device(duid=duid) + if not device or not device.get("localkey"): + return + try: + messages = create_mqtt_decoder(device["localkey"])(payload) + for message in messages: + if int(getattr(message.protocol, "value", message.protocol)) != 101: + continue + request = _request_body(message) + if not request or request.get("method") not in { + "upd_server_timer", "set_server_timer", "del_server_timer", + }: + continue + enable_name = None + params = request.get("params") + if (request.get("method") == "upd_server_timer" and isinstance(params, list) + and len(params) == 1 and isinstance(params[0], list) + and len(params[0]) == 2 and isinstance(params[0][0], str)): + name, state = params[0] + # A later off must invalidate an earlier on intent, too. + self._enable_intent.setdefault(duid, {}).pop(name, None) + if state == "on": + enable_name = name + self.changed(duid, enable_name=enable_name) + except Exception: # Unrelated/unsupported frames must always pass through. + return + + async def refresh(self, duid: str) -> list[Any]: + from .routine_runner import _RoutineMqttClient + + async with self._slots, self.lock(duid): + snapshot = list_jobs(self.context, duid) + names = [] + for job in snapshot: + try: + name = native_timer_name(job) + except SettingsError: + continue + if name is not None: + names.append(name) + if not names: + return [] + device = self.context.runtime_credentials.resolve_device(duid=duid) + if not device or not device.get("localkey"): + return [] + generation = self._generation.get(duid, 0) + intents = {name: intent for name, intent in self._enable_intent.get(duid, {}).items() + if intent[0] > time.monotonic()} + self._enable_intent[duid] = intents + explicit_enable = {name: intent[1] for name, intent in intents.items()} + client = None + try: + async with asyncio.timeout(15): + client = _RoutineMqttClient(self.context, device, _LOGGER) + await client.connect() + states = timer_states(await client.send_command(TimerCommand.GET, [])) + if self._generation.get(duid, 0) != generation: + # A newer native write overlapped the read. A later pass + # will confirm it; do not apply this older snapshot. + self._dirty.add(duid) + self._wake.set() + return [] + changed = reconcile_native_enabled( + self.context, duid, snapshot, states, explicit_enable=explicit_enable, + ) + # A read can arrive before an outstanding command is applied. + # Keep a bounded enable intent until it is actually observed on. + self._enable_intent[duid] = {name: until for name, until in intents.items() + if states.get(name) is not True} + if changed: + _LOGGER.info("Synchronized native schedule enablement for %s: %d changed", duid, len(changed)) + return changed + except (OSError, ValueError): + _LOGGER.warning("Could not save native schedule reconciliation for %s", duid, exc_info=True) + return [] + except Exception: + # An offline/unsupported robot keeps its last saved state. + _LOGGER.debug("Native schedule read unavailable for %s", duid, exc_info=True) + return [] + finally: + if client is not None: + try: + async with asyncio.timeout(5): + await client.close() + except Exception: + _LOGGER.debug("Could not close native schedule read for %s", duid, exc_info=True) + + async def run(self, stop_event: asyncio.Event) -> None: + next_scan = 0.0 + while not stop_event.is_set(): + now = time.monotonic() + due = set(self._dirty) + self._dirty.clear() + self._wake.clear() + if now >= next_scan: + schedules = load_inventory(self.context).get("schedules", {}) + if isinstance(schedules, dict): + due.update(str(duid) for duid in schedules) + next_scan = now + self.INTERVAL + if due: + await asyncio.sleep(self.DEBOUNCE) + results = await asyncio.gather(*(self.refresh(duid) for duid in sorted(due)), return_exceptions=True) + for duid, result in zip(sorted(due), results): + if isinstance(result, Exception): + _LOGGER.warning("Native schedule reconciliation unavailable for %s: %s", duid, type(result).__name__) + if self._dirty: + continue + try: + await asyncio.wait_for(self._wake.wait(), timeout=max(0.1, next_scan - time.monotonic())) + except TimeoutError: + pass diff --git a/src/roborock_local_server/bundled_backend/shared/routine_runner.py b/src/roborock_local_server/bundled_backend/shared/routine_runner.py index 71fcca8e..a4791910 100644 --- a/src/roborock_local_server/bundled_backend/shared/routine_runner.py +++ b/src/roborock_local_server/bundled_backend/shared/routine_runner.py @@ -3,7 +3,10 @@ from __future__ import annotations import asyncio -from dataclasses import dataclass +import copy +from dataclasses import dataclass, replace +from datetime import datetime, timezone +from enum import StrEnum import json import logging from pathlib import Path @@ -32,6 +35,7 @@ def _ensure_local_python_roborock_on_path() -> None: from roborock.roborock_typing import RoborockCommand from .context import ServerContext +from .schedule_store import claimed_job_is_current __all__ = [ "RoutineCommand", @@ -58,6 +62,22 @@ def _ensure_local_python_roborock_on_path() -> None: _ACTION_LOCK_RETRY_DELAY_SECONDS = 5.0 _ACTION_LOCK_RETRY_ATTEMPTS = 6 from .inventory_io import WEB_API_INVENTORY_FILE +class SchedulerCommand(StrEnum): + """A robot command python-roborock's RoborockCommand does not list.""" + + # The cloud's own way to run a schedule: the robot answers it with the schedule's settings entry + # (the same one the cloud job carries, including "name", the robot-side timer id) and records the + # clean as "Schedules" in its history, for whole-home and room schedules alike. Confirmed live; + # without the entry it answers -10005 "Invalid timer", and with a bare id "Param should be + # name + fan + segment". + SERVER_SCHEDULED_START = "server_scheduled_start" + GET_SERVER_TIMER = "get_server_timer" + + +# A whole-home clean started by a schedule rather than a scene; see commands_for_step. +SCHEDULE_APP_START = "schedule_app_start" +# Any schedule that knows its robot-side timer id; the robot runs it itself when asked this way. +SCHEDULE_SERVER_START = "schedule_server_start" _SUPPORTED_METHODS = { "do_scenes_app_start", "do_scenes_segments", @@ -67,8 +87,20 @@ def _ensure_local_python_roborock_on_path() -> None: RoborockCommand.APP_START, RoborockCommand.APP_SEGMENT_CLEAN, RoborockCommand.APP_ZONED_CLEAN, + SchedulerCommand.SERVER_SCHEDULED_START, } _ACTION_LOCK_ERROR_CODES = {-10003, -10007} +_PARAM_ERROR_CODE = -10005 +# What app_start takes in one command, as the official app's own schedule entries carry it. A +# robot accepted every one of these together and applied the modes (confirmed live); it answers +# -10005 "Param error" to a value it does not support, such as a route its model lacks. +_APP_START_KEYS = ("repeat", "water_box_mode", "segments", "fan_power", "mop_mode", "clean_mop", "map_index") +_APP_START_DEFAULTS = {"repeat": 1, "segments": "0", "clean_mop": 1, "map_index": -1} +# app_start's "source" field is how the robot learns who started a clean, for its history. 3 is +# python-roborock's RoborockStartType.schedule, and a robot recorded the clean as "Schedule" in the +# official app's history; an app start is 2. "start_type", "startType", "start_from", "from" and +# "trigger" were all accepted but ignored, so the clean was recorded as "App". +_SCHEDULE_START_SOURCE = 3 _RESUME_COMMAND_BY_IN_CLEANING: dict[int, RoborockCommand] = { RoborockInCleaning.global_clean_not_complete.value: RoborockCommand.APP_START, RoborockInCleaning.zone_clean_not_complete.value: RoborockCommand.RESUME_ZONED_CLEAN, @@ -82,7 +114,7 @@ class RoutineExecutionError(RuntimeError): @dataclass(frozen=True) class RoutineCommand: - command: RoborockCommand + command: RoborockCommand | SchedulerCommand params: dict[str, Any] | list[Any] | None = None @@ -376,9 +408,30 @@ def commands_for_step(step: RoutineStep) -> list[RoutineCommand]: ), RoutineCommand(RoborockCommand.APP_START, None), ] + if step.method == SCHEDULE_SERVER_START: + entry = _single_start_entry(step) + # Preserve the captured payload, including settings introduced by newer + # models. Filtering through app_start's keys loses schedule semantics. + return [RoutineCommand(SchedulerCommand.SERVER_SCHEDULED_START, [copy.deepcopy(entry)])] + if step.method == SCHEDULE_APP_START: + entry = _single_start_entry(step) + body = {**_APP_START_DEFAULTS, **{key: entry[key] for key in _APP_START_KEYS if key in entry}} + body["source"] = _SCHEDULE_START_SOURCE + return [RoutineCommand(RoborockCommand.APP_START, [body])] raise RoutineExecutionError(f"Unsupported step method {step.method!r}") +def fallback_commands_for_step(step: RoutineStep) -> list[RoutineCommand] | None: + """Alternative settings transport for a legacy schedule, never a native timer. + + A refused server_scheduled_start can mean the timer is disabled or gone. + An ordinary clean would bypass that decision, so native starts never fall back. + """ + if step.method != SCHEDULE_APP_START: + return None + return commands_for_step(replace(step, method="do_scenes_app_start")) + + def _is_optional_unsupported_command(command: RoborockCommand, exc: Exception) -> bool: return command == RoborockCommand.SET_MOP_TEMPLATE_ID and isinstance(exc, RoborockUnsupportedFeature) @@ -534,7 +587,7 @@ def _on_message(self, payload: bytes) -> None: async def send_command( self, - command: RoborockCommand, + command: RoborockCommand | SchedulerCommand, params: dict[str, Any] | list[Any] | None = None, ) -> Any: if self._session is None: @@ -801,6 +854,36 @@ def start_scene(self, scene: dict[str, Any]) -> dict[str, Any]: "stepCount": len(steps), } + def start_job( + self, + *, + device_id: str, + job_id: int, + name: str, + steps: list[RoutineStep], + expected_job: dict[str, Any] | None = None, + deadline: datetime | None = None, + ) -> dict[str, Any]: + """Run a scheduled job's steps; never interrupts a routine already running.""" + existing = self._tasks_by_device.get(device_id) + if existing is not None and not existing.task.done(): + return { + "accepted": False, + "status": "routine_in_progress", + "jobId": job_id, + "deviceId": device_id, + "activeSceneName": existing.scene_name, + } + scene = {"id": job_id, "name": name, "device_id": device_id} + task = asyncio.get_running_loop().create_task( + self._run_steps(scene=scene, steps=steps, sync_scene_tids=False, + schedule_job=copy.deepcopy(expected_job), schedule_deadline=deadline), + name=f"routine-job-{job_id}-{device_id}", + ) + self._tasks_by_device[device_id] = _ActiveRoutine(task=task, scene_id=job_id, scene_name=name) + task.add_done_callback(lambda finished: self._on_scene_done(device_id, finished)) + return {"accepted": True, "status": "started", "jobId": job_id, "deviceId": device_id, "stepCount": len(steps)} + def _on_scene_done(self, device_id: str, task: asyncio.Task[None]) -> None: current = self._tasks_by_device.get(device_id) if current is not None and current.task is task: @@ -947,6 +1030,7 @@ async def _send_step_command( except Exception as exc: # noqa: BLE001 if ( routine_command.command in _STEP_START_COMMANDS + and step.method != SCHEDULE_SERVER_START and attempts < _ACTION_LOCK_RETRY_ATTEMPTS and _is_retryable_action_locked_error(exc) ): @@ -966,7 +1050,41 @@ async def _send_step_command( continue raise + async def _send_step_commands( + self, + *, + client: _RoutineMqttClient, + logger: logging.LoggerAdapter, + step: RoutineStep, + commands: list[RoutineCommand], + ) -> None: + for routine_command in commands: + logger.info( + "Routine step=%s method=%s command=%s params=%s", + step.step_id, + step.method, + routine_command.command.value, + routine_command.params, + ) + await self._send_step_command( + client=client, + logger=logger, + step=step, + routine_command=routine_command, + ) + async def _run_scene(self, *, scene: dict[str, Any], steps: list[RoutineStep]) -> None: + await self._run_steps(scene=scene, steps=steps, sync_scene_tids=True) + + async def _run_steps( + self, + *, + scene: dict[str, Any], + steps: list[RoutineStep], + sync_scene_tids: bool, + schedule_job: dict[str, Any] | None = None, + schedule_deadline: datetime | None = None, + ) -> None: device_id = scene_device_id(scene) device = self._device_record(device_id) logger = logging.LoggerAdapter( @@ -988,24 +1106,47 @@ async def _run_scene(self, *, scene: dict[str, Any], steps: list[RoutineStep]) - client = _RoutineMqttClient(self._context, device, logger) await client.connect() try: - await self._sync_scene_tids(client=client, scene=scene, device_id=device_id, logger=logger) + if sync_scene_tids: + await self._sync_scene_tids(client=client, scene=scene, device_id=device_id, logger=logger) for step_index, step in enumerate(steps): - commands = commands_for_step(step) + if step.method == SCHEDULE_SERVER_START: + timer_name = _single_start_entry(step).get("name") + timers = await client.send_command(SchedulerCommand.GET_SERVER_TIMER, []) + # The cloud job's enabled field can remain true when the + # robot timer is off. A failed/malformed read must not start + # cleaning, nor enqueue a start for an offline robot. + if not isinstance(timers, list): + raise RoutineExecutionError("Cannot verify native schedule timer state") + matching = [entry for entry in timers if isinstance(entry, list) and len(entry) >= 2 + and entry[0] == timer_name] + if len(matching) != 1 or matching[0][1] != "on": + logger.info("Skipping schedule: robot timer is off, missing, or ambiguous") + continue + elif schedule_job is not None: + # Legacy jobs have no native flag to read, but still require + # a live robot before sending cleaning commands. + await client.get_status() + if schedule_job is not None and not claimed_job_is_current(self._context, device_id, schedule_job): + logger.info("Skipping schedule: job changed during dispatch") + return + if schedule_deadline is not None and datetime.now(timezone.utc) >= schedule_deadline: + logger.info("Skipping schedule: dispatch window expired") + return waits_for_step_complete = RoborockDataProtocol.TASK_COMPLETE.value in step.finish_dp_ids - for routine_command in commands: - logger.info( - "Routine step=%s method=%s command=%s params=%s", - step.step_id, - step.method, - routine_command.command.value, - routine_command.params, + try: + await self._send_step_commands( + client=client, logger=logger, step=step, commands=commands_for_step(step) ) - await self._send_step_command( - client=client, - logger=logger, - step=step, - routine_command=routine_command, + except Exception as exc: # noqa: BLE001 + fallback = fallback_commands_for_step(step) + if fallback is None or _exception_error_code(exc) != _PARAM_ERROR_CODE: + raise + logger.warning( + "Robot refused the single start step=%s (%s); sending the settings as separate commands", + step.step_id, + _exception_error_message(exc), ) + await self._send_step_commands(client=client, logger=logger, step=step, commands=fallback) if waits_for_step_complete: logger.info("Waiting for ready state step=%s scene=%s", step.step_id, _scene_name(scene)) await client.wait_for_step_complete() @@ -1015,3 +1156,11 @@ async def _run_scene(self, *, scene: dict[str, Any], steps: list[RoutineStep]) - await client.close() logger.info("Completed routine scene=%s device=%s", _scene_name(scene), device_id) + + +def get_routine_runner(context: ServerContext) -> RoutineRunner: + runner = getattr(context, "_routine_runner", None) + if runner is None: + runner = RoutineRunner(context) + setattr(context, "_routine_runner", runner) + return runner diff --git a/src/roborock_local_server/bundled_backend/shared/schedule_runner.py b/src/roborock_local_server/bundled_backend/shared/schedule_runner.py new file mode 100644 index 00000000..84c99440 --- /dev/null +++ b/src/roborock_local_server/bundled_backend/shared/schedule_runner.py @@ -0,0 +1,341 @@ +"""Local executor for cloud-style schedules (``server_scheduled_start`` jobs). + +Roborock's cloud fires these at ``nextFireTime``. With the vacuum pointed at this +server, nothing else would, so the server evaluates the stored cron entries +itself and hands each due job to the routine runner. +""" + +from __future__ import annotations + +import asyncio +import copy +from dataclasses import dataclass +from datetime import date, datetime, timedelta, timezone, tzinfo +import json +import logging +from typing import TYPE_CHECKING, Any +from zoneinfo import ZoneInfo, ZoneInfoNotFoundError + +from .constants import DEFAULT_TIMEZONE +from .data_helpers import as_bool +from .inventory_io import load_inventory +from .routine_runner import ( + SCHEDULE_APP_START, + SCHEDULE_SERVER_START, + RoutineExecutionError, + RoutineStep, + get_routine_runner, +) +from .schedule_store import claim_occurrence + +if TYPE_CHECKING: + from .context import ServerContext + +_LOGGER = logging.getLogger(__name__) + +TICK_SECONDS = 20 +MAX_LATENESS_SECONDS = 60 +_SUPPORTED_JOB_METHOD = "server_scheduled_start" +_CARRIED_ENTRY_KEYS = ("repeat", "fan_power", "water_box_mode", "mop_mode", "mop_template_id") +_START_ENTRY_KEYS = (*_CARRIED_ENTRY_KEYS, "segments", "clean_mop", "map_index") + + +class CronError(ValueError): + """Raised for a cron expression this scheduler cannot evaluate.""" + + +@dataclass(frozen=True) +class CronSpec: + minutes: tuple[int, ...] + hours: tuple[int, ...] + days_of_month: frozenset[int] + months: frozenset[int] + days_of_week: frozenset[int] + dom_restricted: bool + dow_restricted: bool + + def day_matches(self, day: date) -> bool: + if day.month not in self.months: + return False + dom_ok = day.day in self.days_of_month + # cron weekday 0 = Sunday; python's isoweekday() has Sunday as 7. + dow_ok = (day.isoweekday() % 7) in self.days_of_week + if self.dom_restricted and self.dow_restricted: + return dom_ok or dow_ok + return dom_ok and dow_ok + + +def _parse_field(text: str, low: int, high: int, *, name: str) -> tuple[frozenset[int], bool]: + text = text.strip() + if text in ("*", "?"): + return frozenset(range(low, high + 1)), False + values: set[int] = set() + for part in text.split(","): + part = part.strip() + if not part: + raise CronError(f"empty {name} entry in {text!r}") + step = 1 + if "/" in part: + part, _, step_text = part.partition("/") + try: + step = int(step_text) + except ValueError as exc: + raise CronError(f"bad {name} step {step_text!r}") from exc + if step <= 0: + raise CronError(f"{name} step must be positive") + if part in ("*", "?", ""): + start, end = low, high + elif "-" in part: + first, _, last = part.partition("-") + try: + start, end = int(first), int(last) + except ValueError as exc: + raise CronError(f"bad {name} range {part!r}") from exc + else: + try: + start = int(part) + except ValueError as exc: + raise CronError(f"bad {name} value {part!r}") from exc + end = high if step != 1 else start + if start < low or end > high or start > end: + raise CronError(f"{name} value out of range in {text!r}") + values.update(range(start, end + 1, step)) + return frozenset(values), True + + +def parse_cron(expression: str) -> CronSpec: + fields = str(expression or "").split() + if len(fields) != 5: + raise CronError(f"expected 5 cron fields, got {len(fields)}: {expression!r}") + minutes, _ = _parse_field(fields[0], 0, 59, name="minute") + hours, _ = _parse_field(fields[1], 0, 23, name="hour") + dom, dom_restricted = _parse_field(fields[2], 1, 31, name="day-of-month") + months, _ = _parse_field(fields[3], 1, 12, name="month") + raw_dow, dow_restricted = _parse_field(fields[4], 0, 7, name="day-of-week") + dow = frozenset(0 if value == 7 else value for value in raw_dow) + return CronSpec( + minutes=tuple(sorted(minutes)), + hours=tuple(sorted(hours)), + days_of_month=dom, + months=months, + days_of_week=dow, + dom_restricted=dom_restricted, + dow_restricted=dow_restricted, + ) + + +def next_fire(spec: CronSpec, after: datetime) -> datetime | None: + """Next wall-clock match; skip DST gaps and use only the first repeated hour. + + This explicit DST policy is provisional until compared with cloud behavior. + Compare UTC instants because local datetime arithmetic can hide DST folds. + """ + zone = after.tzinfo + if zone is None: + raise ValueError("after must be timezone-aware") + after_utc = after.astimezone(timezone.utc) + first_day = after.date() + # Eight years also covers the leap-day gap across a non-leap century. + for offset in range(8 * 366): + day = first_day + timedelta(days=offset) + if not spec.day_matches(day): + continue + for hour in spec.hours: + for minute in spec.minutes: + candidate = datetime(day.year, day.month, day.day, hour, minute, tzinfo=zone) + candidate_utc = candidate.astimezone(timezone.utc) + if candidate_utc.astimezone(zone).replace(tzinfo=None) != candidate.replace(tzinfo=None): + continue # Nonexistent wall-clock time during spring forward. + if candidate_utc > after_utc: + return candidate + return None + + +def _zone(name: str, fallback: str) -> tzinfo: + # DEFAULT_TIMEZONE is what /location tells the robot when none is configured. + candidate = str(name or fallback or DEFAULT_TIMEZONE).strip() + return ZoneInfo(candidate) + + +def _segment_ids(value: Any) -> list[int]: + if isinstance(value, str): + items: list[Any] = [part for part in value.replace(";", ",").split(",") if part.strip()] + elif isinstance(value, (list, tuple)): + items = list(value) + elif value is None: + items = [] + else: + items = [value] + segments: list[int] = [] + for item in items: + try: + number = int(str(item).strip()) + except ValueError: + continue + if number > 0: + segments.append(number) + return segments + + +def job_param(job: dict[str, Any]) -> dict[str, Any]: + param = job.get("param") + if isinstance(param, str): + try: + param = json.loads(param) + except json.JSONDecodeError as exc: + raise RoutineExecutionError("schedule param is not valid JSON") from exc + if not isinstance(param, dict): + raise RoutineExecutionError("schedule is missing its param payload") + return param + + +def steps_for_job(job: dict[str, Any]) -> list[RoutineStep]: + param = job_param(job) + method = str(param.get("method") or "").strip() + if method != _SUPPORTED_JOB_METHOD: + raise RoutineExecutionError(f"unsupported schedule method {method!r}") + params = param.get("params") + if isinstance(params, list) and len(params) != 1: + raise RoutineExecutionError("schedule params must contain exactly one settings object") + entry = params[0] if isinstance(params, list) and params else params + if not isinstance(entry, dict): + raise RoutineExecutionError("schedule params must contain one settings object") + + settings = {key: entry[key] for key in _CARRIED_ENTRY_KEYS if key in entry} + # A whole-home start also carries the rest of the entry, so one app_start can take it all. + start_settings = {key: entry[key] for key in _START_ENTRY_KEYS if key in entry} + # The robot-side timer id the official app (or LocalRock) saved this schedule under. A schedule + # that has one is run the way the cloud runs it: the robot is asked to start that timer. + timer_id = entry.get("name") + if "name" in entry and (not isinstance(timer_id, str) or not timer_id.strip()): + raise RoutineExecutionError("native schedule timer name must be a nonempty string") + name = str(entry.get("name") or job.get("name") or "").strip() + if not name or name.isdigit(): + name = f"Schedule {job.get('id', '')}".strip() + if isinstance(timer_id, str) and timer_id.strip(): + return [ + RoutineStep( + step_id=1, + name=name, + method=SCHEDULE_SERVER_START, + params=[copy.deepcopy(entry)], + finish_dp_ids=(), + ) + ] + segments = _segment_ids(entry.get("segments")) + if segments: + return [ + RoutineStep( + step_id=1, + name=name, + method="do_scenes_segments", + params={"data": [{**settings, "segs": segments}]}, + finish_dp_ids=(), + ) + ] + return [ + RoutineStep( + step_id=1, + name=name, + method=SCHEDULE_APP_START, + params=[start_settings], + finish_dp_ids=(), + ) + ] + + +def next_fire_time_utc(job: dict[str, Any], *, default_timezone: str = "", now: datetime | None = None) -> datetime | None: + if not as_bool(job.get("enabled", True), True): + return None + try: + spec = parse_cron(str(job.get("cron") or "")) + zone = _zone(str(job.get("timeZoneId") or ""), default_timezone) + except (CronError, ZoneInfoNotFoundError, ValueError): + return None + current = (now or datetime.now(timezone.utc)).astimezone(zone) + upcoming = next_fire(spec, current) + return upcoming.astimezone(timezone.utc) if upcoming is not None else None + + +class ScheduleRunner: + def __init__(self, context: ServerContext, *, default_timezone: str = "") -> None: + self._context = context + self._default_timezone = default_timezone + self._last_tick: datetime | None = None + self._logger = context.loggers.get("api") or context.loggers.get("unknown") or _LOGGER + + def due_jobs(self, since: datetime, until: datetime) -> list[tuple[str, dict[str, Any], datetime]]: + inventory = load_inventory(self._context) + schedules = inventory.get("schedules") + if not isinstance(schedules, dict): + return [] + due: list[tuple[str, dict[str, Any], datetime]] = [] + for device_id, jobs in schedules.items(): + if not isinstance(jobs, list): + continue + for job in jobs: + if not isinstance(job, dict) or not as_bool(job.get("enabled", True), True): + continue + if job.get("id") in (None, ""): + self._logger.warning("Skipping schedule without an id on %s", device_id) + continue + try: + spec = parse_cron(str(job.get("cron") or "")) + zone = _zone(str(job.get("timeZoneId") or ""), self._default_timezone) + except (CronError, ZoneInfoNotFoundError, ValueError) as exc: + self._logger.warning("Skipping schedule %s on %s: %s", job.get("id"), device_id, exc) + continue + upcoming = next_fire(spec, since.astimezone(zone)) + if upcoming is not None and upcoming.astimezone(timezone.utc) <= until: + due.append((str(device_id), job, upcoming)) + return due + + def tick(self, now: datetime | None = None) -> list[dict[str, Any]]: + """Fire anything that came due since the previous tick. Needs a running event loop.""" + current = (now or datetime.now(timezone.utc)).astimezone(timezone.utc) + previous = self._last_tick + if previous is not None and current <= previous: + return [] + self._last_tick = current + if previous is None: + return [] + results: list[dict[str, Any]] = [] + runner = get_routine_runner(self._context) + # Never drain a backlog after suspension or an event-loop stall. Startup + # establishes a fresh baseline; normal ticks get a bounded grace period. + previous = max(previous, current - timedelta(seconds=MAX_LATENESS_SECONDS)) + for device_id, job, due in self.due_jobs(previous, current): + job_id = job.get("id") + try: + steps = steps_for_job(job) + if not claim_occurrence(self._context, device_id, job, due): + continue + result = runner.start_job( + device_id=device_id, + job_id=int(job_id) if str(job_id).isdigit() else 0, + name=steps[0].name, + steps=steps, + expected_job=job, + deadline=due.astimezone(timezone.utc) + timedelta(seconds=MAX_LATENESS_SECONDS), + ) + except (RoutineExecutionError, ValueError, OSError) as exc: + self._logger.warning("Schedule %s on %s not started: %s", job_id, device_id, exc) + result = {"accepted": False, "status": "error", "error": str(exc), "deviceId": device_id} + else: + self._logger.info("Schedule %s on %s fired: %s", job_id, device_id, result.get("status")) + results.append({"jobId": job_id, **result}) + return results + + async def run(self, stop_event: asyncio.Event, *, tick_seconds: float = TICK_SECONDS) -> None: + self.tick() + while not stop_event.is_set(): + try: + await asyncio.wait_for(stop_event.wait(), timeout=tick_seconds) + except TimeoutError: + pass + if stop_event.is_set(): + return + try: + self.tick() + except Exception as exc: # noqa: BLE001 + self._logger.warning("Schedule tick failed: %s", exc) diff --git a/src/roborock_local_server/bundled_backend/shared/schedule_store.py b/src/roborock_local_server/bundled_backend/shared/schedule_store.py new file mode 100644 index 00000000..e59a09f6 --- /dev/null +++ b/src/roborock_local_server/bundled_backend/shared/schedule_store.py @@ -0,0 +1,337 @@ +"""Create, update and delete the per-device schedules kept in the inventory.""" + +from __future__ import annotations + +import copy +from datetime import datetime +import json +import time +from typing import TYPE_CHECKING, Any + +from .data_helpers import as_bool +from .inventory_io import inventory_mutation, load_inventory, write_inventory + +if TYPE_CHECKING: + from .context import ServerContext + +DEFAULT_JOB_PARAM: dict[str, Any] = { + "id": 1, + "method": "server_scheduled_start", + "params": [{"repeat": 1, "segments": "0"}], +} + + +JOB_SETTING_KEYS = ( + "repeat", + "fan_power", + "water_box_mode", + "mop_mode", + "mop_template_id", + "segments", + "clean_mop", +) +_INT_SETTING_KEYS = frozenset(JOB_SETTING_KEYS) - {"segments"} + + +class SettingsError(ValueError): + """A schedule setting had a value this server cannot store.""" + + +def _normalize_setting(key: str, value: Any) -> Any: + if key == "segments": + items = value.replace(";", ",").split(",") if isinstance(value, str) else list(value) + try: + ids = [int(str(item).strip()) for item in items if str(item).strip()] + except ValueError as exc: + raise SettingsError(f"segments must be room ids, got {value!r}") from exc + ids = [room for room in ids if room > 0] + return ",".join(str(room) for room in ids) if ids else "0" + if isinstance(value, bool) or not isinstance(value, (int, str)): + raise SettingsError(f"{key} must be a number, got {value!r}") + try: + return int(value) + except ValueError as exc: + raise SettingsError(f"{key} must be a number, got {value!r}") from exc + + +def _settings_entry(job: dict[str, Any]) -> dict[str, Any]: + param = job.get("param") + if isinstance(param, str): + try: + param = json.loads(param) + except json.JSONDecodeError: + return {} + params = param.get("params") if isinstance(param, dict) else None + return params[0] if isinstance(params, list) and params and isinstance(params[0], dict) else {} + + +def has_native_timer(job: dict[str, Any]) -> bool: + return "name" in _settings_entry(job) + + +def native_timer_name(job: dict[str, Any]) -> str | None: + """Return an exact native reference, refusing malformed or multiple references.""" + param = job.get("param") + if isinstance(param, str): + try: + param = json.loads(param) + except (ValueError, TypeError): + return None + if not isinstance(param, dict) or param.get("method") != "server_scheduled_start": + return None + params = param.get("params") + if isinstance(params, list): + if len(params) != 1: + if any(isinstance(entry, dict) and "name" in entry for entry in params): + raise SettingsError("Schedule has multiple or ambiguous native timer references") + return None + params = params[0] + if not isinstance(params, dict) or "name" not in params: + return None + name = params["name"] + if not isinstance(name, str) or not name.strip(): + raise SettingsError("Schedule has an invalid native timer reference") + return name + + +@inventory_mutation +def reconcile_native_enabled( + ctx: ServerContext, duid: str, snapshot: list[dict[str, Any]], states: dict[str, bool], + *, explicit_enable: dict[str, Any] | None = None, +) -> list[Any]: + """Merge a fresh robot read only into unchanged, uniquely linked jobs. + + Preserve complete records and occurrence claims. A still-on robot timer + must not resurrect a consumed one-time job after a restart or routine poll. + """ + inventory = load_inventory(ctx) + jobs = _jobs_for(inventory, duid, create=False) or [] + by_name: dict[str, list[dict[str, Any]]] = {} + for job in jobs: + if not isinstance(job, dict): + continue + try: + name = native_timer_name(job) + except SettingsError: + continue + if name is not None: + by_name.setdefault(name, []).append(job) + changed = [] + claims = inventory.get("schedule_occurrences", {}).get(duid, {}) + for name, linked in by_name.items(): + if len(linked) != 1 or name not in states: + continue + job = linked[0] + before = [entry for entry in snapshot if _same_id(entry, job.get("id"))] + if before != [job] or sum(_same_id(entry, job.get("id")) for entry in jobs) != 1: + continue + enabled = states[name] + if as_bool(job.get("enabled", True), True) == enabled: + continue + if (enabled and not as_bool(job.get("repeated", True), True) + and str(job.get("id")) in claims + and (name not in (explicit_enable or {}) + or claims[str(job.get("id"))] != explicit_enable[name])): + continue + job["enabled"] = enabled + job.pop("nextFireTime", None) + changed.append(job["id"]) + if changed: + _persist(ctx, inventory) + return changed + + +def job_settings(job: dict[str, Any]) -> dict[str, Any]: + """The vac/mop settings a job carries, as a flat dict. + + ``name``, when the job has one, rides along too, outside ``JOB_SETTING_KEYS``: it is not a + vac/mop setting a client edits, it is the official app's own link from this job to the + robot's server_timer entry it saved alongside it (POST/PUT /user/devices/{duid}/jobs carries + it in exactly this spot, confirmed live). Native timer entries themselves are + read from the robot; this helper does not turn them into legacy timers. + """ + entry = _settings_entry(job) + settings = {key: entry[key] for key in JOB_SETTING_KEYS if key in entry} + name = entry.get("name") + if isinstance(name, str) and name: + settings["name"] = name + return settings + + +def merge_job_settings(param: dict[str, Any] | None, settings: dict[str, Any]) -> dict[str, Any]: + """Copy of ``param`` with ``settings`` merged in; a None value clears that setting. + + Settings not mentioned are left alone, so editing one option never resets the rest. + """ + merged = copy.deepcopy(param) if isinstance(param, dict) else copy.deepcopy(DEFAULT_JOB_PARAM) + params = merged.get("params") + if not isinstance(params, list) or not params or not isinstance(params[0], dict): + params = copy.deepcopy(DEFAULT_JOB_PARAM["params"]) + merged["params"] = params + entry = params[0] + for key, value in settings.items(): + if key not in JOB_SETTING_KEYS: + raise SettingsError(f"unknown schedule setting {key!r}") + if value is None: + entry.pop(key, None) + else: + entry[key] = _normalize_setting(key, value) + return merged + + +def _same_id(job: Any, job_id: Any) -> bool: + return isinstance(job, dict) and str(job.get("id")) == str(job_id) + + +def _jobs_for(inventory: dict[str, Any], duid: str, *, create: bool) -> list[Any] | None: + schedules = inventory.get("schedules") + if not isinstance(schedules, dict): + if not create: + return None + schedules = {} + inventory["schedules"] = schedules + jobs = schedules.get(duid) + if not isinstance(jobs, list): + if not create: + return None + jobs = [] + schedules[duid] = jobs + return jobs + + +def _coerce_id(job_id: Any) -> Any: + text = str(job_id).strip() + return int(text) if text.isdigit() else text + + +def list_jobs(ctx: ServerContext, duid: str) -> list[dict[str, Any]]: + jobs = _jobs_for(load_inventory(ctx), duid, create=False) or [] + return [copy.deepcopy(job) for job in jobs if isinstance(job, dict)] + + +def _persist(ctx: ServerContext, inventory: dict[str, Any]) -> None: + if not write_inventory(ctx, inventory): + raise OSError("Unable to persist schedule inventory") + + +@inventory_mutation +def upsert_job( + ctx: ServerContext, + duid: str, + *, + job_id: Any, + cron: str, + enabled: bool, + time_zone: str = "", + repeated: bool | None = None, + param: dict[str, Any] | None = None, + override_time_zone: bool = False, + extra_fields: dict[str, Any] | None = None, + create_only: bool = False, + expected_job: dict[str, Any] | None = None, +) -> dict[str, Any]: + """Create the job, or update cron/enabled on an existing one while keeping its settings. + + An existing job keeps its own timezone unless ``override_time_zone`` is set. + """ + inventory = load_inventory(ctx) + jobs = _jobs_for(inventory, duid, create=True) + assert jobs is not None + if job_id is None: + job_id = time.time_ns() // 1_000_000 + while any(_same_id(job, job_id) for job in jobs): + job_id += 1 + existing = next((job for job in jobs if _same_id(job, job_id)), None) + if create_only and existing is not None: + raise ValueError("schedule id already exists") + if expected_job is not None and existing != expected_job: + raise ValueError("schedule changed or was deleted; read it again before updating") + if existing is None: + existing = { + "id": _coerce_id(job_id), + "repeated": True if repeated is None else repeated, + "type": "clean", + "param": copy.deepcopy(param) if param is not None else copy.deepcopy(DEFAULT_JOB_PARAM), + } + if time_zone: + existing["timeZoneId"] = time_zone + jobs.append(existing) + else: + if repeated is not None: + existing["repeated"] = repeated + if param is not None: + existing["param"] = copy.deepcopy(param) + if override_time_zone and time_zone: + existing["timeZoneId"] = time_zone + existing["cron"] = cron + existing["enabled"] = enabled + # Preserve model-specific job fields without allowing a body to replace the + # path ID, validated fields, or the server's calculated next-fire value. + reserved = {"id", "nextFireTime", "cron", "enabled", "timeZoneId", "repeated", "param"} + existing.update(copy.deepcopy({key: value for key, value in (extra_fields or {}).items() if key not in reserved})) + existing.pop("nextFireTime", None) + _persist(ctx, inventory) + return copy.deepcopy(existing) + + +@inventory_mutation +def set_enabled(ctx: ServerContext, duid: str, job_id: Any, enabled: bool) -> bool: + inventory = load_inventory(ctx) + jobs = _jobs_for(inventory, duid, create=False) or [] + found = False + for job in jobs: + if _same_id(job, job_id): + job["enabled"] = enabled + found = True + if found: + _persist(ctx, inventory) + return found + + +@inventory_mutation +def delete_job(ctx: ServerContext, duid: str, job_id: Any) -> bool: + inventory = load_inventory(ctx) + jobs = _jobs_for(inventory, duid, create=False) + if not jobs: + return False + remaining = [job for job in jobs if not _same_id(job, job_id)] + if len(remaining) == len(jobs): + return False + jobs[:] = remaining + occurrences = inventory.get("schedule_occurrences", {}).get(duid, {}) + occurrences.pop(str(job_id), None) + _persist(ctx, inventory) + return True + + +@inventory_mutation +def claim_occurrence(ctx: ServerContext, duid: str, job: dict[str, Any], due: datetime) -> bool: + """Persist an at-most-once decision before any robot command is sent. + + A crash after this write can skip a clean; replaying a clean after restart is + deliberately avoided. Compare the snapshot under the same lock as CRUD so a + disabled, edited, or deleted job cannot be claimed from a stale tick. + """ + inventory = load_inventory(ctx) + jobs = _jobs_for(inventory, duid, create=False) or [] + current = next((item for item in jobs if _same_id(item, job.get("id"))), None) + if current != job or not as_bool(job.get("enabled", True), True): + return False + occurrences = inventory.setdefault("schedule_occurrences", {}).setdefault(duid, {}) + key = str(job["id"]) + instant = int(due.timestamp()) + if occurrences.get(key, -1) >= instant: + return False + occurrences[key] = instant + if not as_bool(job.get("repeated", True), True): + current["enabled"] = False + _persist(ctx, inventory) + return True + + +def claimed_job_is_current(ctx: ServerContext, duid: str, job: dict[str, Any]) -> bool: + """Recheck after network awaits; a time edit must cancel stale queued work.""" + expected = copy.deepcopy(job) + if not as_bool(job.get("repeated", True), True): + expected["enabled"] = False # Consumed atomically by claim_occurrence. + return any(item == expected for item in list_jobs(ctx, duid)) diff --git a/src/roborock_local_server/bundled_backend/shared/timer_rpc.py b/src/roborock_local_server/bundled_backend/shared/timer_rpc.py new file mode 100644 index 00000000..5129dac2 --- /dev/null +++ b/src/roborock_local_server/bundled_backend/shared/timer_rpc.py @@ -0,0 +1,238 @@ +"""Answer the vacuum's legacy timer RPCs (get/set/upd/del_timer) from the schedule store. + +On cloud-managed robots the schedules live server-side, not on the robot, so +clients such as LocalRock that ask the robot for its timers would otherwise see +nothing and could never change the schedules the cloud imported. The topic +bridge hands matching requests here instead of forwarding them to the robot. + +Deliberately not handled here: get/set/upd/del_server_timer, the newer sibling +protocol some firmware speaks instead of (or alongside) this one. The robot +stores those entries itself and the official app reads them back from it, so +they must always forward to the robot untouched. Answering them from this store +instead, tried once, hid the robot's real schedules from every client reading +through the server, and the official app then deleted them (it removes +server_timer entries it does not recognize when its schedule screen opens). + +What runs a server_timer schedule is the cloud's own record of it, the job the +official app also writes to /user/devices/{duid}/jobs, which ScheduleRunner +fires like any other. The robot was not seen to run them itself. +""" + +from __future__ import annotations + +import json +import logging +import time +from typing import TYPE_CHECKING, Any + +from roborock.protocol import create_mqtt_decoder, create_mqtt_encoder +from roborock.roborock_message import RoborockMessage, RoborockMessageProtocol + +from .constants import DEFAULT_TIMEZONE +from .schedule_runner import CronError, parse_cron +from .schedule_store import ( + JOB_SETTING_KEYS, + SettingsError, + delete_job, + has_native_timer, + job_settings, + list_jobs, + merge_job_settings, + set_enabled, + upsert_job, +) + +if TYPE_CHECKING: + from .context import ServerContext + +_LOGGER = logging.getLogger(__name__) + +TIMER_METHODS = frozenset({"get_timer", "set_timer", "upd_timer", "del_timer"}) +_OK = ["ok"] + + +class TimerRequestError(ValueError): + """The request used a shape this handler cannot interpret.""" + + +def device_cron(cron: str) -> str | None: + """Cron in the robot's own 'm h dom month dow' form, with '?' written as '*'. + + A one-time schedule pins the day of the month and the month ('15 23 18 10 *'), exactly as + the robot and the official app write it (confirmed live), so it is shown like any other. + A job that pins only one of the two has no such form and is left out of get_timer rather + than shown wrong. + """ + fields = str(cron or "").split() + if len(fields) != 5: + return None + minute, hour, dom, month, dow = (field if field != "?" else "*" for field in fields) + if (dom == "*") != (month == "*"): + return None + return f"{minute} {hour} {dom} {month} {dow}" + + +def _timer_entry(job: dict[str, Any]) -> list[Any] | None: + # Captured a15/a27 cloud jobs have native server timers while get_timer is + # empty. Do not fabricate a second legacy timer for the same native job. + if has_native_timer(job): + return None + cron = device_cron(str(job.get("cron") or "")) + if cron is None: + return None + enabled = job.get("enabled", True) + on = enabled not in (False, 0, "0", "false", "False") + return [str(job.get("id")), "on" if on else "off", [cron, ["start_clean", job_settings(job)]]] + + +def _parse_set_timer(params: Any) -> tuple[str, bool, str, dict[str, Any]]: + entry = params[0] if isinstance(params, list) and params and isinstance(params[0], list) else params + if not isinstance(entry, list) or len(entry) < 2: + raise TimerRequestError("set_timer needs [id, [status, [cron, action]]]") + job_id = str(entry[0]) + if len(entry) >= 3 and isinstance(entry[1], str): + status, detail = entry[1], entry[2] + else: + spec = entry[1] + if not isinstance(spec, list) or len(spec) < 2: + raise TimerRequestError("set_timer spec must be [status, [cron, action]]") + status, detail = spec[0], spec[1] + if not isinstance(detail, list) or not detail: + raise TimerRequestError("set_timer is missing its cron") + action = detail[1] if len(detail) > 1 else None + settings = action[1] if isinstance(action, list) and len(action) > 1 and isinstance(action[1], dict) else {} + return job_id, str(status).lower() == "on", str(detail[0]), settings + + +def _id_and_status(params: Any) -> tuple[str, str]: + entry = params[0] if isinstance(params, list) and params and isinstance(params[0], list) else params + if not isinstance(entry, list) or len(entry) < 2: + raise TimerRequestError("upd_timer needs [id, status]") + return str(entry[0]), str(entry[1]).lower() + + +def _single_id(params: Any) -> str: + if isinstance(params, list) and params: + first = params[0] + return str(first[0]) if isinstance(first, list) and first else str(first) + if params is None or params == []: + raise TimerRequestError("del_timer needs a timer id") + return str(params) + + +def handle_timer_request( + ctx: ServerContext, + duid: str, + method: str, + params: Any, + *, + default_timezone: str = "", +) -> Any: + """Apply one timer RPC to the store and return the RPC ``result`` value.""" + if method == "get_timer": + entries = (_timer_entry(job) for job in list_jobs(ctx, duid)) + return [entry for entry in entries if entry is not None] + + if method == "set_timer": + job_id, enabled, cron, action_settings = _parse_set_timer(params) + try: + parse_cron(cron) + except CronError as exc: + raise TimerRequestError(str(exc)) from exc + # An empty action object (what older clients send) leaves the job's settings alone; + # keys it does carry are merged in, so a client can change one option at a time. + settings = {key: value for key, value in action_settings.items() if key in JOB_SETTING_KEYS} + param = None + if settings: + existing = next((job for job in list_jobs(ctx, duid) if str(job.get("id")) == job_id), None) + try: + param = merge_job_settings(existing.get("param") if existing else None, settings) + except SettingsError as exc: + raise TimerRequestError(str(exc)) from exc + zone = str(getattr(ctx, "timezone", "") or default_timezone or DEFAULT_TIMEZONE) + upsert_job(ctx, duid, job_id=job_id, cron=cron, enabled=enabled, time_zone=zone, param=param) + return list(_OK) + + if method == "upd_timer": + job_id, status = _id_and_status(params) + if status not in ("on", "off"): + raise TimerRequestError(f"unknown timer status {status!r}") + set_enabled(ctx, duid, job_id, status == "on") + return list(_OK) + + if method == "del_timer": + delete_job(ctx, duid, _single_id(params)) + return list(_OK) + + raise TimerRequestError(f"not a timer method: {method!r}") + + +class TimerRpcInterceptor: + """Bridge hook: turn an encrypted timer request into an encrypted reply, or None to forward.""" + + def __init__(self, context: ServerContext, *, default_timezone: str = "") -> None: + self._context = context + self._default_timezone = default_timezone + + def _localkey(self, duid: str) -> str: + creds = self._context.runtime_credentials + device = creds.resolve_device(duid=duid) if creds is not None else None + key = str((device or {}).get("localkey") or "").strip() + return key or str(getattr(self._context, "localkey", "") or "").strip() + + def __call__(self, duid: str, payload: bytes) -> bytes | None: + localkey = self._localkey(duid) + if not localkey: + return None + try: + messages = create_mqtt_decoder(localkey)(payload) + except Exception: # noqa: BLE001 + return None + if len(messages) != 1: + return None + message = messages[0] + protocol = int(getattr(message.protocol, "value", message.protocol)) + if protocol != RoborockMessageProtocol.RPC_REQUEST.value or not message.payload: + return None + request = _request_body(message) + if request is None or request.get("method") not in TIMER_METHODS: + return None + + request_id = request.get("id") + try: + body: dict[str, Any] = { + "id": request_id, + "result": handle_timer_request( + self._context, + duid, + str(request["method"]), + request.get("params"), + default_timezone=self._default_timezone, + ), + } + except TimerRequestError as exc: + _LOGGER.warning("Rejected %s for %s: %s", request.get("method"), duid, exc) + body = {"id": request_id, "error": {"code": -1, "message": str(exc)}} + reply = RoborockMessage( + protocol=RoborockMessageProtocol.RPC_RESPONSE, + payload=json.dumps( + { + "t": int(time.time()), + "dps": {str(RoborockMessageProtocol.RPC_RESPONSE.value): json.dumps(body)}, + }, + separators=(",", ":"), + ).encode(), + seq=message.seq, + version=message.version, + ) + return create_mqtt_encoder(localkey)(reply) + + +def _request_body(message: RoborockMessage) -> dict[str, Any] | None: + try: + outer = json.loads(message.payload) + inner = outer["dps"][str(RoborockMessageProtocol.RPC_REQUEST.value)] + body = json.loads(inner) if isinstance(inner, str) else inner + except (ValueError, KeyError, TypeError): + return None + return body if isinstance(body, dict) else None diff --git a/src/roborock_local_server/server.py b/src/roborock_local_server/server.py index 105a2aa3..bf0e5818 100644 --- a/src/roborock_local_server/server.py +++ b/src/roborock_local_server/server.py @@ -54,6 +54,10 @@ ) from shared.inventory_io import atomic_write_inventory, inventory_transaction from shared.protocol_auth import ProtocolAuthStore +from shared.schedule_runner import ScheduleRunner +from shared.native_schedule_sync import NativeScheduleSync +from shared.timer_rpc import TimerRpcInterceptor +from https_server.routes.user.devices import jobs as schedule_jobs from https_server.routes.user.scene import validity as scene_validity from https_server.routes.auth.service import ( build_login_data_response, @@ -362,6 +366,8 @@ def __init__( self._mqtt_proxy: MqttTlsProxy | None = None self._http_server: ManagedFastApiServer | None = None self._renew_task: asyncio.Task[None] | None = None + self._schedule_task: asyncio.Task[None] | None = None + self._schedule_sync_task: asyncio.Task[None] | None = None self._stop_event = asyncio.Event() self.certificate_manager = CertificateManager(config=config, paths=paths) @@ -485,6 +491,7 @@ def __init__( turn_ttl=self.config.turn.ttl, ) self.endpoint_rules = default_endpoint_rules() + self.schedule_sync = NativeScheduleSync(self.context) self.app = self._create_app() def _init_zone_ranges_store(self) -> ZoneRangesStore: @@ -1351,14 +1358,20 @@ async def _handle_roborock_request(self, request: Request) -> Response: ) return response - route_name, response_payload = resolve_route( - rules=self.endpoint_rules, - context=self.context, - clean_path=clean_path, - query_params=query_params, - body_params=body_params, - method=request.method, - ) + if schedule_jobs.match_update(clean_path, request.method): + route_name = "update_schedule" + response_payload = await schedule_jobs.build_update_synced( + self.context, query_params, body_params, clean_path, self.schedule_sync, + ) + else: + route_name, response_payload = resolve_route( + rules=self.endpoint_rules, + context=self.context, + clean_path=clean_path, + query_params=query_params, + body_params=body_params, + method=request.method, + ) entry["route"] = route_name if route_name == "fw_createca": # Persist metadata, but retain the exact credential response sent to firmware. @@ -1968,6 +1981,11 @@ async def start(self) -> None: logger=self.loggers["mqtt"], runtime_state=self.runtime_state, inventory_path=self.paths.inventory_path, + request_observer=self.schedule_sync.observe_request, + request_interceptor=TimerRpcInterceptor( + self.context, + default_timezone=self.config.network.timezone, + ), ) await self._topic_bridge.start() self.runtime_state.set_service("mqtt_topic_bridge", running=True, required=False, enabled=True) @@ -1997,10 +2015,32 @@ async def start(self) -> None: if self._uses_local_tls() and self.config.tls.mode == "cloudflare_acme": self._renew_task = asyncio.create_task(self._renew_loop(), name="tls-renew-loop") + schedule_runner = ScheduleRunner(self.context, default_timezone=self.config.network.timezone) + self._schedule_sync_task = asyncio.create_task( + self.schedule_sync.run(self._stop_event), name="native-schedule-sync", + ) + self._schedule_task = asyncio.create_task( + schedule_runner.run(self._stop_event), + name="schedule-runner", + ) + async def stop(self) -> None: self._stop_event.set() self._turn_relay.stop() self._refresh_turn_health() + if self._schedule_sync_task is not None: + self._schedule_sync_task.cancel() + try: + await self._schedule_sync_task + except asyncio.CancelledError: + pass + self._schedule_sync_task = None + if self._schedule_task is not None: + self._schedule_task.cancel() + try: + await self._schedule_task + except asyncio.CancelledError: + pass if self._renew_task is not None: self._renew_task.cancel() try: diff --git a/tests/test_cloud.py b/tests/test_cloud.py index 98be0b26..dc2f9c56 100644 --- a/tests/test_cloud.py +++ b/tests/test_cloud.py @@ -182,6 +182,16 @@ def test_cloud_inventory_merge_preserves_local_rooms_and_scenes() -> None: assert merged["scene_order"] == [90, 10, 20] +def test_cloud_refresh_preserves_schedule_occurrence_claims(): + existing = {"schedule_occurrences": {"robot-test": {"42001": 1792789740}}} + cloud = {"schedules": {"robot-test": [{"id": 42001, "enabled": True}]}} + merged = _merge_existing_inventory_mutations(cloud, existing) + assert merged["schedule_occurrences"] == existing["schedule_occurrences"] + assert merged["schedules"] == cloud["schedules"] + merged["schedule_occurrences"]["robot-test"]["42001"] += 60 + assert existing["schedule_occurrences"]["robot-test"]["42001"] == 1792789740 + + def _make_manager(tmp_path: Path) -> CloudImportManager: return CloudImportManager( inventory_path=tmp_path / "inventory.json", diff --git a/tests/test_native_schedule_sync.py b/tests/test_native_schedule_sync.py new file mode 100644 index 00000000..3a7a53a2 --- /dev/null +++ b/tests/test_native_schedule_sync.py @@ -0,0 +1,235 @@ +"""Native state converges across clients without treating a sent command as success.""" + +import asyncio +import copy +import json +from pathlib import Path +import tempfile +from types import SimpleNamespace +import unittest +from unittest.mock import patch + +import roborock_local_server.backend # noqa: F401 (makes bundled shared modules importable) + +from shared import native_schedule_sync as sync +from shared import schedule_store as store + + +def job(enabled=True, *, job_id=701, name="native-1", repeated=True): + return { + "id": job_id, "enabled": enabled, "repeated": repeated, "cron": "0 9 * * *", + "timeZoneId": "America/Los_Angeles", "future": {"keep": [1, 2]}, + "param": {"method": "server_scheduled_start", "params": [{ + "name": name, "segments": "16,18", "map_index": 2, "future": {"keep": True}, + }]}, + } + + +class NativeStoreTests(unittest.TestCase): + def setUp(self): + self.temp = tempfile.TemporaryDirectory() + self.addCleanup(self.temp.cleanup) + self.path = Path(self.temp.name) / "web_api_inventory.json" + self.context = SimpleNamespace(http_jsonl=self.path.parent / "http.jsonl") + self.seed([job()]) + + def seed(self, jobs, claims=None): + self.path.write_text(json.dumps({ + "schedules": {"robot": jobs, "other": [job()]}, + "schedule_occurrences": {"robot": claims or {}}, "keep": {"metadata": 1}, + })) + + def test_reconciles_both_directions_and_preserves_identity_settings_and_other_devices(self): + for enabled in (False, True): + before = json.loads(self.path.read_text()) + snapshot = store.list_jobs(self.context, "robot") + self.assertEqual(store.reconcile_native_enabled(self.context, "robot", snapshot, {"native-1": enabled}), [701]) + before["schedules"]["robot"][0]["enabled"] = enabled + self.assertEqual(json.loads(self.path.read_text()), before) + + def test_noop_does_not_write_inventory(self): + with patch.object(store, "write_inventory", side_effect=AssertionError("unnecessary write")): + self.assertEqual(store.reconcile_native_enabled(self.context, "robot", [job()], {"native-1": True}), []) + + def test_stale_read_cannot_overwrite_job_edited_while_reading_robot(self): + snapshot = store.list_jobs(self.context, "robot") + changed = job() + changed["cron"] = "30 9 * * *" + self.seed([changed]) + self.assertEqual(store.reconcile_native_enabled(self.context, "robot", snapshot, {"native-1": False}), []) + self.assertEqual(store.list_jobs(self.context, "robot"), [changed]) + + def test_deleted_job_is_not_recreated(self): + snapshot = store.list_jobs(self.context, "robot") + self.seed([]) + self.assertEqual(store.reconcile_native_enabled(self.context, "robot", snapshot, {"native-1": False}), []) + self.assertEqual(store.list_jobs(self.context, "robot"), []) + + def test_duplicate_timer_or_job_ids_are_never_guessed(self): + for jobs in ([job(), job(job_id=702)], [job(), job(name="native-2")]): + self.seed(jobs) + self.assertEqual(store.reconcile_native_enabled(self.context, "robot", jobs, {"native-1": False}), []) + self.assertEqual(store.list_jobs(self.context, "robot"), jobs) + + def test_missing_and_malformed_timer_entries_do_not_disable_jobs(self): + for timers in ([], [["native-1", "bad"]], [["native-1", "off"], ["native-1", "on"]], [["native-1"]]): + states = sync.timer_states(timers) + self.assertEqual(store.reconcile_native_enabled(self.context, "robot", [job()], states), []) + + def test_serialized_and_object_params_keep_their_original_form(self): + for serialized in (False, True): + original = job() + if serialized: + original["param"] = json.dumps(original["param"]) + else: + original["param"]["params"] = original["param"]["params"][0] + self.seed([original]) + self.assertEqual(store.reconcile_native_enabled(self.context, "robot", [original], {"native-1": False}), [701]) + self.assertEqual(store.list_jobs(self.context, "robot")[0]["param"], original["param"]) + + def test_consumed_one_time_job_is_not_resurrected_by_periodic_read(self): + original = job(False, repeated=False) + self.seed([original], {"701": 12345}) + self.assertEqual(store.reconcile_native_enabled(self.context, "robot", [original], {"native-1": True}), []) + self.assertEqual(store.reconcile_native_enabled( + self.context, "robot", [original], {"native-1": True}, explicit_enable={"native-1": 12345}, + ), [701]) + self.assertEqual(json.loads(self.path.read_text())["schedule_occurrences"], {"robot": {"701": 12345}}) + + def test_earlier_enable_intent_does_not_resurrect_a_newly_consumed_occurrence(self): + original = job(False, repeated=False) + self.seed([original], {"701": 23456}) + self.assertEqual(store.reconcile_native_enabled( + self.context, "robot", [original], {"native-1": True}, explicit_enable={"native-1": 12345}, + ), []) + + def test_unlinked_server_only_jobs_are_unchanged(self): + original = job() + del original["param"]["params"][0]["name"] + self.seed([original]) + self.assertEqual(store.reconcile_native_enabled(self.context, "robot", [original], {"701": False}), []) + + def test_persistence_failure_does_not_lose_inventory(self): + before = self.path.read_bytes() + with patch.object(store, "write_inventory", return_value=False): + with self.assertRaises(OSError): + store.reconcile_native_enabled(self.context, "robot", [job()], {"native-1": False}) + self.assertEqual(self.path.read_bytes(), before) + + +class NativeCoordinatorTests(unittest.IsolatedAsyncioTestCase): + async def asyncSetUp(self): + self.temp = tempfile.TemporaryDirectory() + self.addCleanup(self.temp.cleanup) + self.path = Path(self.temp.name) / "web_api_inventory.json" + self.path.write_text(json.dumps({"schedules": {"robot": [job()]}})) + self.context = SimpleNamespace( + http_jsonl=self.path.parent / "http.jsonl", + runtime_credentials=SimpleNamespace(resolve_device=lambda **kw: {"duid": "robot", "localkey": "test"}), + ) + self.coordinator = sync.NativeScheduleSync(self.context) + self.robot = SimpleNamespace(timers=[["native-1", "off"]], calls=[], after_read=None, write_error=None, apply=True) + robot = self.robot + + class Client: + def __init__(self, *args): + pass + + async def connect(self): + robot.calls.append("connect") + + async def send_command(self, command, params): + robot.calls.append(str(command)) + if command == "get_server_timer": + result = copy.deepcopy(robot.timers) + if robot.after_read: + await robot.after_read() + return result + if robot.apply: + robot.timers[0][1] = params[0][1] + if robot.write_error: + raise robot.write_error + return ["ok"] + + async def close(self): + robot.calls.append("close") + + self.network = patch.dict("sys.modules", {"shared.routine_runner": SimpleNamespace(_RoutineMqttClient=Client)}) + self.network.start() + self.addCleanup(self.network.stop) + + async def test_external_toggle_reconciliation_reads_only_and_closes_connection(self): + self.assertEqual(await self.coordinator.refresh("robot"), [701]) + self.assertFalse(store.list_jobs(self.context, "robot")[0]["enabled"]) + self.assertEqual(self.robot.calls, ["connect", "get_server_timer", "close"]) + + async def test_later_native_write_invalidates_an_in_flight_read(self): + async def later(): + self.coordinator.changed("robot") + self.robot.timers[0][1] = "on" + self.robot.after_read = later + self.assertEqual(await self.coordinator.refresh("robot"), []) + self.assertTrue(store.list_jobs(self.context, "robot")[0]["enabled"]) + self.robot.after_read = None + self.assertEqual(await self.coordinator.refresh("robot"), []) + + async def test_dashboard_or_http_edit_lock_prevents_background_race(self): + async with self.coordinator.lock("robot"): + pending = asyncio.create_task(self.coordinator.refresh("robot")) + await asyncio.sleep(0) + self.assertEqual(self.robot.calls, []) + self.assertFalse(pending.done()) + self.assertEqual(await pending, [701]) + + async def test_lost_write_ack_requires_a_fresh_matching_read(self): + self.robot.write_error = TimeoutError() + self.assertEqual(await sync.sync_timer(self.context, "robot", "native-1", True, check_current=lambda: None), "on") + self.assertEqual(self.robot.calls.count("upd_server_timer"), 1) + self.assertEqual(self.robot.calls.count("get_server_timer"), 2) + + async def test_unapplied_write_is_not_success(self): + self.robot.apply = False + with self.assertRaises(sync.TimerSyncError): + await sync.sync_timer(self.context, "robot", "native-1", True, check_current=lambda: None) + self.assertEqual(self.robot.calls.count("upd_server_timer"), 1) + + async def test_matching_timer_needs_no_second_write(self): + self.assertEqual(await sync.sync_timer(self.context, "robot", "native-1", False, check_current=lambda: None), "off") + self.assertNotIn("upd_server_timer", self.robot.calls) + + async def test_startup_and_periodic_reads_recover_changes_without_observed_requests(self): + self.coordinator.INTERVAL = 0.02 + self.coordinator.DEBOUNCE = 0 + stop = asyncio.Event() + task = asyncio.create_task(self.coordinator.run(stop)) + try: + for enabled in (False, True): + self.robot.timers[0][1] = "on" if enabled else "off" + async with asyncio.timeout(1): + while store.list_jobs(self.context, "robot")[0]["enabled"] != enabled: + await asyncio.sleep(0.001) + finally: + stop.set() + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + self.assertNotIn("upd_server_timer", self.robot.calls) + + async def test_cancellation_closes_an_in_flight_read(self): + waiting = asyncio.Event() + async def pause(): + waiting.set() + await asyncio.Event().wait() + self.robot.after_read = pause + task = asyncio.create_task(self.coordinator.refresh("robot")) + await waiting.wait() + task.cancel() + with self.assertRaises(asyncio.CancelledError): + await task + self.assertEqual(self.robot.calls[-1], "close") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_routine_runner.py b/tests/test_routine_runner.py index 40110c01..96648763 100644 --- a/tests/test_routine_runner.py +++ b/tests/test_routine_runner.py @@ -9,7 +9,14 @@ from roborock.exceptions import RoborockException from roborock_local_server.bundled_backend.shared.context import ServerContext import roborock_local_server.bundled_backend.shared.routine_runner as routine_runner_module -from roborock_local_server.bundled_backend.shared.routine_runner import RoutineRunner, parse_scene_steps +from roborock_local_server.bundled_backend.shared.routine_runner import ( + SCHEDULE_APP_START, + SCHEDULE_SERVER_START, + RoutineRunner, + RoutineStep, + RoutineExecutionError, + parse_scene_steps, +) from roborock.roborock_typing import RoborockCommand @@ -711,3 +718,225 @@ async def exercise() -> None: await client.wait_for_dock_settle(initial_delay_seconds=0.0) asyncio.run(exercise()) + + +def _schedule_step(**entry) -> RoutineStep: + return RoutineStep(step_id=1, name="Schedule 1", method=SCHEDULE_APP_START, params=[entry], finish_dp_ids=()) + + +def _run_schedule_step(tmp_path: Path, monkeypatch, step: RoutineStep, *, refuse_single_start_with: dict | None): + """Run [step] against a fake robot; returns every command it was sent, refused ones included.""" + sent_commands: list[tuple[RoborockCommand, object]] = [] + + class FakeRoutineClient: + def __init__(self, context, device, logger) -> None: + _ = context, device, logger + + async def connect(self) -> None: + return None + + async def close(self) -> None: + return None + + async def send_command(self, command, params=None): + sent_commands.append((command, params)) + if command == routine_runner_module.SchedulerCommand.GET_SERVER_TIMER: + return [[step.params[0]["name"], "on", 1]] + if refuse_single_start_with is not None and command == RoborockCommand.APP_START and params is not None: + raise RoborockException(refuse_single_start_with) + return ["ok"] + + async def wait_for_step_complete(self) -> None: + return None + + monkeypatch.setattr(routine_runner_module, "_RoutineMqttClient", FakeRoutineClient) + scene = _scene(scene_id=1, device_id="6HL2zfniaoYYV01CkVuhkO", name="Schedule 1") + runner = RoutineRunner(_test_context(tmp_path)) + return sent_commands, asyncio.run(runner._run_steps(scene=scene, steps=[step], sync_scene_tids=False)) + + +def test_a_schedule_start_is_one_app_start_carrying_the_whole_entry(tmp_path: Path, monkeypatch) -> None: + # A robot accepted app_start with all of these together and applied the modes (confirmed live). + step = _schedule_step(fan_power=104, water_box_mode=203, mop_mode=300) + + sent, _ = _run_schedule_step(tmp_path, monkeypatch, step, refuse_single_start_with=None) + + assert sent == [ + ( + RoborockCommand.APP_START, + [{"fan_power": 104, "water_box_mode": 203, "mop_mode": 300, + "repeat": 1, "segments": "0", "clean_mop": 1, "map_index": -1, + # "source" 3 is what makes the robot's history say "Schedule" instead of "App". + "source": 3}], + ) + ] + + +def test_a_schedule_start_falls_back_to_separate_commands_on_a_param_error(tmp_path: Path, monkeypatch) -> None: + # A route the robot's model lacks makes it answer -10005 "Param error", and a refused command + # changes nothing, so the same settings go out the longer way. + step = _schedule_step(fan_power=104, water_box_mode=203, mop_mode=300) + + sent, _ = _run_schedule_step( + tmp_path, monkeypatch, step, refuse_single_start_with={"code": -10005, "message": "Param error"} + ) + + assert [command for command, _ in sent] == [ + RoborockCommand.APP_START, + RoborockCommand.SET_CUSTOM_MODE, + RoborockCommand.SET_WATER_BOX_CUSTOM_MODE, + RoborockCommand.SET_MOP_MODE, + RoborockCommand.SET_CLEAN_REPEAT_TIMES, + RoborockCommand.APP_START, + ] + assert sent[-1] == (RoborockCommand.APP_START, None) + + +def test_a_schedule_start_does_not_fall_back_on_any_other_refusal(tmp_path: Path, monkeypatch) -> None: + step = _schedule_step(fan_power=104) + + with pytest.raises(RoborockException): + _run_schedule_step( + tmp_path, monkeypatch, step, refuse_single_start_with={"code": -10001, "message": "Busy"} + ) + + +def _run_schedule_step_refusing(tmp_path: Path, monkeypatch, step: RoutineStep, error: dict): + """Like _run_schedule_step, but the robot refuses server_scheduled_start and accepts the rest.""" + sent_commands: list[tuple[object, object]] = [] + + class FakeRoutineClient: + def __init__(self, context, device, logger) -> None: + _ = context, device, logger + + async def connect(self) -> None: + return None + + async def close(self) -> None: + return None + + async def send_command(self, command, params=None): + sent_commands.append((command, params)) + if command == routine_runner_module.SchedulerCommand.GET_SERVER_TIMER: + return [[step.params[0]["name"], "on", 1]] + if command == routine_runner_module.SchedulerCommand.SERVER_SCHEDULED_START: + raise RoborockException(error) + return ["ok"] + + async def wait_for_step_complete(self) -> None: + return None + + monkeypatch.setattr(routine_runner_module, "_RoutineMqttClient", FakeRoutineClient) + scene = _scene(scene_id=1, device_id="6HL2zfniaoYYV01CkVuhkO", name="Schedule 1") + runner = RoutineRunner(_test_context(tmp_path)) + with pytest.raises(RoborockException): + asyncio.run(runner._run_steps(scene=scene, steps=[step], sync_scene_tids=False)) + return sent_commands + + +def _server_start_step(**entry) -> RoutineStep: + return RoutineStep(step_id=1, name="Schedule 1", method=SCHEDULE_SERVER_START, params=[entry], finish_dp_ids=()) + + +def test_a_server_start_asks_the_robot_to_run_the_schedule_with_its_whole_entry(tmp_path: Path, monkeypatch) -> None: + # Sent exactly as the cloud job's settings entry, "name" (the robot-side timer id) included; + # confirmed live to start a clean the robot records as "Schedules | Rooms". + step = _server_start_step( + repeat=1, water_box_mode=202, segments="16,17", fan_power=102, mop_mode=300, + clean_mop=1, map_index=0, name="1792336598179", clean_order_mode=0, + mop_template_id=42, future_model_option={"mode": 7}, + ) + + sent, _ = _run_schedule_step(tmp_path, monkeypatch, step, refuse_single_start_with=None) + + assert [command for command, _ in sent] == [ + routine_runner_module.SchedulerCommand.GET_SERVER_TIMER, + routine_runner_module.SchedulerCommand.SERVER_SCHEDULED_START, + ] + assert sent[1][1] == [ + {"repeat": 1, "water_box_mode": 202, "segments": "16,17", "fan_power": 102, "mop_mode": 300, + "clean_mop": 1, "map_index": 0, "name": "1792336598179", "clean_order_mode": 0, + "mop_template_id": 42, "future_model_option": {"mode": 7}} + ] + + +@pytest.mark.parametrize("segments", ["16,17", "0"]) +@pytest.mark.parametrize("code", [-10005, -10003, -10007]) +def test_a_refused_server_start_never_falls_back_or_retries(tmp_path: Path, monkeypatch, segments, code) -> None: + # "Invalid timer" (the robot no longer holds that timer) is a -10005 like any bad parameter. + step = _server_start_step(fan_power=102, water_box_mode=202, segments=segments, name="gone") + sent = _run_schedule_step_refusing(tmp_path, monkeypatch, step, {"code": code, "message": "Refused"}) + + commands = [command for command, _ in sent] + assert commands == [routine_runner_module.SchedulerCommand.GET_SERVER_TIMER, + routine_runner_module.SchedulerCommand.SERVER_SCHEDULED_START] + + +@pytest.mark.parametrize("timers", [[], [["timer", "off", 0]], [["different", "on", 1]], + [["timer", "on", 1], ["timer", "off", 0]]]) +def test_a_native_schedule_requires_one_enabled_robot_timer( + tmp_path: Path, monkeypatch, timers +) -> None: + sent = [] + class Client: + def __init__(self, *args): pass + async def connect(self): pass + async def close(self): pass + async def send_command(self, command, params=None): + sent.append(command) + return timers + monkeypatch.setattr(routine_runner_module, "_RoutineMqttClient", Client) + runner = RoutineRunner(_test_context(tmp_path)) + scene = _scene(scene_id=1, device_id="6HL2zfniaoYYV01CkVuhkO", name="Schedule 1") + asyncio.run(runner._run_steps(scene=scene, steps=[_server_start_step(name="timer")], sync_scene_tids=False)) + assert sent == [routine_runner_module.SchedulerCommand.GET_SERVER_TIMER] + + +def test_offline_timer_read_never_publishes_a_start(tmp_path, monkeypatch): + sent = [] + closed = [] + class Client: + def __init__(self, *args): pass + async def connect(self): pass + async def close(self): closed.append(True) + async def send_command(self, command, params=None): + sent.append(command) + raise RoutineExecutionError("timed out") + monkeypatch.setattr(routine_runner_module, "_RoutineMqttClient", Client) + runner = RoutineRunner(_test_context(tmp_path)) + scene = _scene(scene_id=1, device_id="6HL2zfniaoYYV01CkVuhkO", name="Schedule 1") + with pytest.raises(RoutineExecutionError, match="timed out"): + asyncio.run(runner._run_steps(scene=scene, steps=[_server_start_step(name="timer")], sync_scene_tids=False)) + assert sent == [routine_runner_module.SchedulerCommand.GET_SERVER_TIMER] + assert closed == [True] + + +@pytest.mark.parametrize("change", ["edit", "disable", "delete", "expire"]) +def test_dispatch_rechecks_job_and_deadline_after_waiting_for_the_robot(tmp_path, monkeypatch, change): + from datetime import datetime, timedelta, timezone + from shared import schedule_store + ctx = _test_context(tmp_path) + duid = "6HL2zfniaoYYV01CkVuhkO" + job = schedule_store.upsert_job(ctx, duid, job_id=1, cron="0 9 * * *", enabled=True, + param={"method": "server_scheduled_start", "params": [{"name": "timer"}]}) + sent = [] + class Client: + def __init__(self, *args): pass + async def connect(self): pass + async def close(self): pass + async def send_command(self, command, params=None): + sent.append(command) + if change == "edit": + schedule_store.upsert_job(ctx, duid, job_id=1, cron="0 10 * * *", enabled=True) + elif change == "disable": + schedule_store.set_enabled(ctx, duid, 1, False) + elif change == "delete": + schedule_store.delete_job(ctx, duid, 1) + return [["timer", "on", 1]] + monkeypatch.setattr(routine_runner_module, "_RoutineMqttClient", Client) + deadline = datetime.now(timezone.utc) + timedelta(seconds=-1 if change == "expire" else 60) + runner = RoutineRunner(ctx) + asyncio.run(runner._run_steps(scene=_scene(scene_id=1, device_id=duid, name="Schedule 1"), + steps=[_server_start_step(name="timer")], sync_scene_tids=False, + schedule_job=job, schedule_deadline=deadline)) + assert sent == [routine_runner_module.SchedulerCommand.GET_SERVER_TIMER] diff --git a/tests/test_schedule_runner.py b/tests/test_schedule_runner.py new file mode 100644 index 00000000..a9331056 --- /dev/null +++ b/tests/test_schedule_runner.py @@ -0,0 +1,287 @@ +import json +from datetime import datetime, timedelta, timezone +from types import SimpleNamespace + +import pytest + +import roborock_local_server.backend # noqa: F401 (puts the bundled `shared` package on sys.path) +from shared import schedule_runner as sr +from shared.routine_runner import RoutineExecutionError + +NEW_YORK = "America/New_York" + + +def _job(**overrides): + job = { + "id": 3549431, + "cron": "00 11 ? * 1,3,5", + "timeZoneId": NEW_YORK, + "repeated": True, + "enabled": True, + "param": { + "id": 1, + "method": "server_scheduled_start", + "params": [ + { + "repeat": 1, + "water_box_mode": 203, + "segments": "0", + "fan_power": 104, + "mop_mode": 300, + "clean_mop": 1, + "map_index": -1, + } + ], + }, + } + job.update(overrides) + return job + + +def _utc(*args): + return datetime(*args, tzinfo=timezone.utc) + + +def test_next_fire_matches_the_cloud_reported_fire_times(): + # Values taken from a real cloud snapshot: Monday 2026-09-28 for MWF 11:00 New York. + spec = sr.parse_cron("00 11 ? * 1,3,5") + zone = sr._zone(NEW_YORK, "") + after = _utc(2026, 9, 28, 12, 0).astimezone(zone) + assert sr.next_fire(spec, after).astimezone(timezone.utc) == _utc(2026, 9, 28, 15, 0) + + tue_thu = sr.parse_cron("00 11 ? * 2,4") + assert sr.next_fire(tue_thu, after).astimezone(timezone.utc) == _utc(2026, 9, 29, 15, 0) + + +def test_next_fire_is_strictly_after_and_skips_to_next_matching_day(): + spec = sr.parse_cron("00 11 ? * 1,3,5") + zone = sr._zone(NEW_YORK, "") + on_the_minute = _utc(2026, 9, 28, 15, 0).astimezone(zone) + assert sr.next_fire(spec, on_the_minute).astimezone(timezone.utc) == _utc(2026, 9, 30, 15, 0) + + +def test_weekend_cron_uses_sunday_zero(): + spec = sr.parse_cron("00 11 ? * 0,6") + zone = sr._zone(NEW_YORK, "") + monday = _utc(2026, 9, 28, 12, 0).astimezone(zone) + assert sr.next_fire(spec, monday).astimezone(timezone.utc) == _utc(2026, 10, 3, 15, 0) + + +@pytest.mark.parametrize("expression", ["", "* * * *", "61 * * * *", "* 25 * * *", "a b c d e", "*/0 * * * *"]) +def test_invalid_cron_is_rejected(expression): + with pytest.raises(sr.CronError): + sr.parse_cron(expression) + + +def test_whole_house_job_becomes_an_app_start_step_carrying_vac_and_mop_settings(): + steps = sr.steps_for_job(_job()) + assert len(steps) == 1 + step = steps[0] + assert step.method == "schedule_app_start" + assert step.params == [ + {"repeat": 1, "water_box_mode": 203, "segments": "0", "fan_power": 104, "mop_mode": 300, "clean_mop": 1, "map_index": -1} + ] + assert step.name == "Schedule 3549431" + + +def test_job_with_a_robot_side_timer_id_is_run_by_asking_the_robot_to_start_it(): + # The official app saves every schedule on the robot too and records its timer id as "name" in + # the cloud job's settings; the cloud runs such a job with server_scheduled_start (confirmed + # live: the robot records "Schedules" for it, for rooms and whole-home alike). + job = _job() + job["param"]["params"][0]["name"] = "1759965417864" + step = sr.steps_for_job(job)[0] + assert step.method == "schedule_server_start" + assert step.params == [ + { + "repeat": 1, "water_box_mode": 203, "segments": "0", "fan_power": 104, "mop_mode": 300, + "clean_mop": 1, "map_index": -1, "name": "1759965417864", + } + ] + + +def test_room_job_with_a_robot_side_timer_id_carries_its_rooms_in_the_same_step(): + job = _job() + job["param"]["params"][0].update({"name": "1759965417864", "segments": "16,17", "map_index": 0}) + step = sr.steps_for_job(job)[0] + assert step.method == "schedule_server_start" + assert step.params[0]["segments"] == "16,17" + assert step.params[0]["map_index"] == 0 + + +def test_segment_job_becomes_a_segment_clean_step(): + job = _job() + job["param"]["params"][0]["segments"] = "16,17" + step = sr.steps_for_job(job)[0] + assert step.method == "do_scenes_segments" + assert step.params["data"][0]["segs"] == [16, 17] + assert step.params["data"][0]["fan_power"] == 104 + + +def test_unsupported_method_is_refused(): + job = _job() + job["param"]["method"] = "something_else" + with pytest.raises(RoutineExecutionError): + sr.steps_for_job(job) + + +class _FakeRunner: + def __init__(self): + self.started = [] + + def start_job(self, **kwargs): + self.started.append(kwargs) + return {"accepted": True, "status": "started", "deviceId": kwargs["device_id"]} + + +def _runner_with_inventory(tmp_path, monkeypatch, jobs): + inventory = {"schedules": {"duid-1": jobs}} + (tmp_path / "web_api_inventory.json").write_text(json.dumps(inventory), encoding="utf-8") + ctx = SimpleNamespace(http_jsonl=tmp_path / "http.jsonl", loggers={}) + fake = _FakeRunner() + monkeypatch.setattr(sr, "get_routine_runner", lambda _ctx: fake) + return sr.ScheduleRunner(ctx), ctx, fake + + +def test_tick_fires_a_job_that_came_due_since_the_previous_tick(tmp_path, monkeypatch): + runner, _ctx, fake = _runner_with_inventory(tmp_path, monkeypatch, [_job()]) + assert runner.tick(_utc(2026, 9, 28, 14, 59, 40)) == [] # first tick only sets the baseline + results = runner.tick(_utc(2026, 9, 28, 15, 0, 5)) + assert [r["status"] for r in results] == ["started"] + assert fake.started[0]["device_id"] == "duid-1" + assert fake.started[0]["steps"][0].method == "schedule_app_start" + + +def test_tick_does_not_fire_twice_or_when_disabled(tmp_path, monkeypatch): + runner, _ctx, fake = _runner_with_inventory( + tmp_path, monkeypatch, [_job(), _job(id=2, enabled=False)] + ) + runner.tick(_utc(2026, 9, 28, 14, 59, 40)) + runner.tick(_utc(2026, 9, 28, 15, 0, 5)) + runner.tick(_utc(2026, 9, 28, 15, 0, 25)) + runner.tick(_utc(2026, 9, 28, 15, 0, 45) + timedelta(minutes=5)) + assert len(fake.started) == 1 + + +def test_one_shot_job_disables_itself_after_firing(tmp_path, monkeypatch): + runner, _ctx, _fake = _runner_with_inventory(tmp_path, monkeypatch, [_job(repeated=False)]) + runner.tick(_utc(2026, 9, 28, 14, 59, 40)) + runner.tick(_utc(2026, 9, 28, 15, 0, 5)) + saved = json.loads((tmp_path / "web_api_inventory.json").read_text(encoding="utf-8")) + assert saved["schedules"]["duid-1"][0]["enabled"] is False + + +@pytest.mark.parametrize("disabled_value", [False, 0, "false", "0"]) +def test_string_typed_disabled_flag_is_respected(tmp_path, monkeypatch, disabled_value): + # Imported inventory stores booleans loosely ("false", 0); "false" must not read as enabled. + runner, _ctx, fake = _runner_with_inventory(tmp_path, monkeypatch, [_job(enabled=disabled_value)]) + runner.tick(_utc(2026, 9, 28, 14, 59, 40)) + runner.tick(_utc(2026, 9, 28, 15, 0, 5)) + assert fake.started == [] + + +def test_next_fire_time_utc_is_none_when_disabled(): + now = _utc(2026, 9, 28, 12, 0) + assert sr.next_fire_time_utc(_job(), now=now) == _utc(2026, 9, 28, 15, 0) + assert sr.next_fire_time_utc(_job(enabled=False), now=now) is None + + +def test_native_settings_survive_dispatch_preparation_without_filtering_or_mutation(): + job = _job() + entry = job["param"]["params"][0] + entry.update(name="timer-42", clean_order_mode=0, mop_template_id=7, future_option={"x": [1, 2]}) + step = sr.steps_for_job(job)[0] + assert step.params == [entry] + step.params[0]["future_option"]["x"].append(3) + assert entry["future_option"] == {"x": [1, 2]} + + +@pytest.mark.parametrize("timer_name", [None, "", " ", 123]) +def test_invalid_native_timer_name_cannot_turn_into_an_ordinary_start(timer_name): + job = _job() + job["param"]["params"][0]["name"] = timer_name + with pytest.raises(RoutineExecutionError, match="timer name"): + sr.steps_for_job(job) + + +def test_observed_weekday_next_fire_advances_to_monday_after_offline_occurrence(): + job = _job(cron="09 14 ? * 1,2,3,4,5", timeZoneId="America/Los_Angeles") + assert sr.next_fire_time_utc(job, now=_utc(2026, 10, 23, 21, 7)) == _utc(2026, 10, 23, 21, 9) + assert sr.next_fire_time_utc(job, now=_utc(2026, 10, 23, 21, 10)) == _utc(2026, 10, 26, 21, 9) + + +@pytest.mark.parametrize("cron,after,expected", [ + ("30 2 * * *", _utc(2026, 3, 8, 9, 59), _utc(2026, 3, 9, 9, 30)), + ("30 1 * * *", _utc(2026, 11, 1, 8, 29), _utc(2026, 11, 1, 8, 30)), + ("30 1 * * *", _utc(2026, 11, 1, 8, 30), _utc(2026, 11, 2, 9, 30)), + ("30 1 * * *", _utc(2026, 11, 1, 9, 20), _utc(2026, 11, 2, 9, 30)), + ("0 0 * * 0", _utc(2026, 10, 25, 6, 59), _utc(2026, 10, 25, 7, 0)), + ("0 0 29 2 ?", _utc(2024, 3, 1), _utc(2028, 2, 29, 8, 0)), +]) +def test_timezone_and_provisional_dst_policy(cron, after, expected): + job = _job(cron=cron, timeZoneId="America/Los_Angeles") + assert sr.next_fire_time_utc(job, now=after) == expected + + +def test_invalid_explicit_timezone_never_runs_in_a_fallback_zone(tmp_path, monkeypatch): + job = _job(timeZoneId="Invalid/Zone") + assert sr.next_fire_time_utc(job, default_timezone="UTC", now=_utc(2026, 9, 28, 14)) is None + runner, _, fake = _runner_with_inventory(tmp_path, monkeypatch, [job]) + runner.tick(_utc(2026, 9, 28, 14, 59, 40)) + runner.tick(_utc(2026, 9, 28, 15, 0, 5)) + assert fake.started == [] + + +def test_missed_occurrence_is_skipped_and_future_schedule_stays_enabled(tmp_path, monkeypatch): + runner, ctx, fake = _runner_with_inventory(tmp_path, monkeypatch, [_job()]) + runner.tick(_utc(2026, 9, 28, 14, 59, 40)) + assert runner.tick(_utc(2026, 9, 28, 15, 2)) == [] + restarted = sr.ScheduleRunner(ctx) + assert restarted.tick(_utc(2026, 9, 28, 15, 3)) == [] + assert fake.started == [] + runner.tick(_utc(2026, 9, 30, 14, 59, 40)) + runner.tick(_utc(2026, 9, 30, 15, 0, 5)) + assert len(fake.started) == 1 + assert sr.load_inventory(ctx)["schedules"]["duid-1"][0]["enabled"] is True + + +def test_persisted_claim_prevents_replay_after_restart_and_clock_rollback(tmp_path, monkeypatch): + runner, ctx, fake = _runner_with_inventory(tmp_path, monkeypatch, [_job()]) + runner.tick(_utc(2026, 9, 28, 14, 59, 40)) + runner.tick(_utc(2026, 9, 28, 15, 0, 5)) + runner.tick(_utc(2026, 9, 28, 14, 59, 40)) + runner.tick(_utc(2026, 9, 28, 15, 0, 10)) + restarted = sr.ScheduleRunner(ctx) + restarted.tick(_utc(2026, 9, 28, 14, 59, 40)) + restarted.tick(_utc(2026, 9, 28, 15, 0, 5)) + assert len(fake.started) == 1 + + +def test_failed_persistence_prevents_robot_dispatch(tmp_path, monkeypatch): + from shared import schedule_store + runner, _, fake = _runner_with_inventory(tmp_path, monkeypatch, [_job()]) + monkeypatch.setattr(schedule_store, "write_inventory", lambda *args: False) + runner.tick(_utc(2026, 9, 28, 14, 59, 40)) + result = runner.tick(_utc(2026, 9, 28, 15, 0, 5)) + assert result[0]["accepted"] is False + assert fake.started == [] + + +def test_job_edit_between_tick_and_claim_cancels_stale_occurrence(tmp_path, monkeypatch): + from shared import schedule_store + runner, ctx, fake = _runner_with_inventory(tmp_path, monkeypatch, [_job()]) + original_steps = sr.steps_for_job + def edit_then_prepare(job): + schedule_store.set_enabled(ctx, "duid-1", job["id"], False) + return original_steps(job) + monkeypatch.setattr(sr, "steps_for_job", edit_then_prepare) + runner.tick(_utc(2026, 9, 28, 14, 59, 40)) + runner.tick(_utc(2026, 9, 28, 15, 0, 5)) + assert fake.started == [] + + +def test_malformed_import_does_not_block_other_jobs(tmp_path, monkeypatch): + runner, _, fake = _runner_with_inventory(tmp_path, monkeypatch, [_job(id=None), _job()]) + runner.tick(_utc(2026, 9, 28, 14, 59, 40)) + runner.tick(_utc(2026, 9, 28, 15, 0, 5)) + assert len(fake.started) == 1 diff --git a/tests/test_schedule_store.py b/tests/test_schedule_store.py new file mode 100644 index 00000000..85d00ca5 --- /dev/null +++ b/tests/test_schedule_store.py @@ -0,0 +1,21 @@ +import roborock_local_server.backend # noqa: F401 (puts the bundled `shared` package on sys.path) +from shared.schedule_store import job_settings + + +def _job(**entry_overrides): + entry = {"repeat": 1, "fan_power": 104, "water_box_mode": 203} + entry.update(entry_overrides) + return {"id": "1", "enabled": True, "param": {"params": [entry]}} + + +def test_job_settings_carries_name_when_present(): + # "name" is the official app's link to the robot's server_timer entry, not a JOB_SETTING_KEYS + # vac/mop setting, confirmed live in its POST/PUT /user/devices/{duid}/jobs body. + job = _job(name="1759965417864") + assert job_settings(job) == {"repeat": 1, "fan_power": 104, "water_box_mode": 203, "name": "1759965417864"} + + +def test_job_settings_omits_name_when_absent_or_blank(): + assert "name" not in job_settings(_job()) + assert "name" not in job_settings(_job(name="")) + assert "name" not in job_settings(_job(name=None)) diff --git a/tests/test_schedules.py b/tests/test_schedules.py index f97b1039..90c27ed3 100644 --- a/tests/test_schedules.py +++ b/tests/test_schedules.py @@ -1,5 +1,10 @@ import json +import copy from pathlib import Path +from types import SimpleNamespace +from uuid import uuid4 + +import pytest from fastapi.testclient import TestClient from roborock.data import HomeData, HomeDataSchedule @@ -104,6 +109,11 @@ def test_device_jobs_endpoint_returns_imported_schedules(tmp_path: Path) -> None payload = response.json() assert payload["success"] is True assert payload["data"] == payload["result"] + # nextFireTime is computed from the cron against "now", so check its shape, not its value. + # No timezone is configured, so the cron runs in the zone the robot is told (America/New_York): + # 13:03 local on Dec 15 is 18:03Z. + next_fire_time = payload["result"][0].pop("nextFireTime") + assert next_fire_time.endswith("T18:03:00.000+00:00") assert payload["result"] == [ { **schedule, @@ -161,3 +171,262 @@ def test_cloud_inventory_build_preserves_device_schedules() -> None: "6HL2zfniaoYYV01CkVuhkO": [schedule], "ignored": [], } + + +def _jobs_client(tmp_path: Path, device_id: str, schedules: list[dict[str, object]]): + config_file = write_release_config(tmp_path) + config = load_config(config_file) + paths = resolve_paths(config_file, config) + _write_json( + paths.inventory_path, + { + "home": {"id": 1233716, "name": "My Home"}, + "devices": [{"duid": device_id, "name": "Roborock Qrevo"}], + "schedules": {device_id: schedules}, + }, + ) + _seed_cloud_snapshot(paths.cloud_snapshot_path) + supervisor = ReleaseSupervisor(config=config, paths=paths) + supervisor.refresh_inventory_state() + return TestClient(supervisor.app), paths + + +def _signed_json(paths, path: str, body: dict[str, object]) -> tuple[bytes, dict[str, str]]: + raw = json.dumps(body, separators=(",", ":")).encode() + user = ProtocolAuthStore(paths.cloud_snapshot_path).availability().user + assert user is not None + headers = { + # Hawk rejects replayed nonces, and _hawk_headers() uses nonce- for plain requests. + "Authorization": build_hawk_authorization( + user=user, path=path, json_body=raw, nonce=f"write-{uuid4().hex}" + ), + "Content-Type": "application/json", + } + return raw, headers + + +def test_jobs_can_be_created_updated_and_deleted_over_http(tmp_path: Path) -> None: + device_id = "6HL2zfniaoYYV01CkVuhkO" + existing = { + "id": 3549431, + "cron": "00 11 ? * 1,3,5", + "timeZoneId": "America/New_York", + "repeated": True, + "enabled": True, + "param": { + "id": 1, + "method": "server_scheduled_start", + "params": [{"repeat": 1, "water_box_mode": 203, "segments": "0", "fan_power": 104, "mop_mode": 300}], + }, + } + client, paths = _jobs_client(tmp_path, device_id, [existing]) + collection = f"/user/devices/{device_id}/jobs" + + body, headers = _signed_json(paths, collection, {"cron": "30 7 * * 1,2", "enabled": True}) + created = client.post(collection, content=body, headers=headers) + assert created.status_code == 200 + assert created.json()["success"] is True + new_id = created.json()["result"]["id"] + + listed = client.get(collection, headers=_hawk_headers(paths.cloud_snapshot_path, collection)).json()["result"] + assert {job["id"] for job in listed} == {3549431, new_id} + assert all(job.get("nextFireTime") for job in listed) + + item = f"{collection}/3549431" + body, headers = _signed_json(paths, item, {"cron": "15 9 ? * 1,3,5", "enabled": False}) + updated = client.put(item, content=body, headers=headers) + assert updated.status_code == 200 + result = updated.json()["result"] + assert (result["cron"], result["enabled"]) == ("15 9 ? * 1,3,5", False) + # Vac/mop settings are untouched by a cron/enabled edit. + assert result["param"]["params"][0]["mop_mode"] == 300 + + deleted = client.delete(item, headers=_hawk_headers(paths.cloud_snapshot_path, item)) + assert deleted.status_code == 200 and deleted.json()["success"] is True + user = ProtocolAuthStore(paths.cloud_snapshot_path).availability().user + assert user is not None + second_list = {"Authorization": build_hawk_authorization(user=user, path=collection, nonce="second-list")} + remaining = client.get(collection, headers=second_list).json()["result"] + assert [job["id"] for job in remaining] == [new_id] + + +def test_creating_a_job_with_a_bad_cron_is_rejected(tmp_path: Path) -> None: + device_id = "6HL2zfniaoYYV01CkVuhkO" + client, paths = _jobs_client(tmp_path, device_id, []) + collection = f"/user/devices/{device_id}/jobs" + body, headers = _signed_json(paths, collection, {"cron": "nonsense"}) + response = client.post(collection, content=body, headers=headers) + assert response.json()["success"] is False + assert response.json()["code"] == 400 + + +def _native_job(): + # Synthetic IDs/settings based on the a15/a27 protocol shape, not a raw capture. + return { + "id": 42001, "cron": "09 14 ? * 1,2,3,4,5", "timeZoneId": "America/Los_Angeles", + "repeated": True, "enabled": True, "type": "clean", "modelOption": {"mode": 1}, + "param": {"id": 1, "method": "server_scheduled_start", "params": [{ + "name": "1800000000001", "repeat": 1, "water_box_mode": 204, "segments": "16,17", + "fan_power": 106, "mop_mode": 302, "clean_mop": 1, "clean_order_mode": 0, + "map_index": 1, "mop_template_id": 7, "futureOption": {"mode": 2}, + }]}, + } + + +@pytest.fixture +def native_robot(monkeypatch): + from shared import routine_runner + robot = SimpleNamespace(timers=[["1800000000001", "on"]], calls=[], apply=True, + error=None, after_read=None) + + class Client: + def __init__(self, context, device, logger): + assert device["duid"] == "robot-test" + + async def connect(self): + pass + + async def send_command(self, command, params): + robot.calls.append((str(command), copy.deepcopy(params))) + if str(command) == "get_server_timer": + if robot.after_read: + robot.after_read() + return copy.deepcopy(robot.timers) + assert str(command) == "upd_server_timer" + if robot.apply: + robot.timers[0][1] = params[0][1] + if robot.error: + raise robot.error + return ["ok"] + + async def close(self): + pass + + monkeypatch.setattr(routine_runner, "_RoutineMqttClient", Client) + return robot + + +def test_homebridge_delay_and_restore_put_round_trip_preserves_native_job(tmp_path, native_robot): + original = _native_job() + client, paths = _jobs_client(tmp_path, "robot-test", [original]) + collection = "/user/devices/robot-test/jobs" + item = f"{collection}/{original['id']}" + for cron in ("39 14 ? * 1,2,3,4,5", original["cron"]): + body = {key: value for key, value in original.items() if key not in ("id", "nextFireTime")} + body["cron"] = cron + raw, headers = _signed_json(paths, item, body) + response = client.put(item, content=raw, headers=headers) + assert response.json()["success"] is True + saved = json.loads(paths.inventory_path.read_text())["schedules"]["robot-test"][0] + assert saved == {**original, "cron": cron} + listed = client.get(collection, headers=_hawk_headers(paths.cloud_snapshot_path, collection)).json()["result"] + assert {k: v for k, v in listed[0].items() if k != "nextFireTime"} == original + assert listed[0]["param"]["params"][0]["name"] != str(listed[0]["id"]) + assert all(method == "get_server_timer" for method, _ in native_robot.calls) + + +def test_direct_http_enablement_updates_robot_then_store_for_any_client(tmp_path, native_robot): + original = _native_job() + client, paths = _jobs_client(tmp_path, "robot-test", [original]) + item = "/user/devices/robot-test/jobs/42001" + for enabled in (False, True): + raw, headers = _signed_json(paths, item, {"enabled": enabled}) + result = client.put(item, content=raw, headers=headers).json() + assert result["success"] is True + assert result["result"] == {**original, "enabled": enabled} + assert native_robot.timers[0][1] == ("on" if enabled else "off") + assert native_robot.calls[-2:] == [ + ("upd_server_timer", [["1800000000001", "on" if enabled else "off"]]), + ("get_server_timer", []), + ] + + +def test_oem_timer_then_job_put_does_not_write_native_timer_again(tmp_path, native_robot): + native_robot.timers[0][1] = "off" + client, paths = _jobs_client(tmp_path, "robot-test", [_native_job()]) + item = "/user/devices/robot-test/jobs/42001" + raw, headers = _signed_json(paths, item, {"enabled": False}) + result = client.put(item, content=raw, headers=headers).json() + assert result["success"] is True + assert native_robot.calls == [("get_server_timer", [])] + + +@pytest.mark.parametrize("lost_ack", [False, True]) +def test_http_requires_actual_confirmation_and_does_not_retry_write(tmp_path, native_robot, lost_ack): + native_robot.error = TimeoutError("lost ack") if lost_ack else None + native_robot.apply = lost_ack + client, paths = _jobs_client(tmp_path, "robot-test", [_native_job()]) + before = paths.inventory_path.read_bytes() + item = "/user/devices/robot-test/jobs/42001" + raw, headers = _signed_json(paths, item, {"enabled": False}) + result = client.put(item, content=raw, headers=headers).json() + assert result["success"] is lost_ack + assert sum(method == "upd_server_timer" for method, _ in native_robot.calls) == 1 + if not lost_ack: + assert paths.inventory_path.read_bytes() == before + + +def test_http_change_during_timer_read_refuses_stale_native_write(tmp_path, native_robot): + client, paths = _jobs_client(tmp_path, "robot-test", [_native_job()]) + def change(): + data = json.loads(paths.inventory_path.read_text()) + data["schedules"]["robot-test"][0]["cron"] = "0 10 * * *" + paths.inventory_path.write_text(json.dumps(data)) + native_robot.after_read = change + item = "/user/devices/robot-test/jobs/42001" + raw, headers = _signed_json(paths, item, {"enabled": False}) + result = client.put(item, content=raw, headers=headers).json() + assert result["success"] is False and result["code"] == 409 + assert native_robot.calls == [("get_server_timer", [])] + + +def test_put_replaces_param_exactly_when_app_removes_optional_fields(tmp_path): + original = _native_job() + client, paths = _jobs_client(tmp_path, "robot-test", [original]) + param = json.loads(json.dumps(original["param"])) + del param["params"][0]["mop_template_id"] + param["params"][0]["segments"] = "0" + item = "/user/devices/robot-test/jobs/42001" + raw, headers = _signed_json(paths, item, {"param": param, "modelOption": {"mode": 5}}) + result = client.put(item, content=raw, headers=headers).json()["result"] + assert result["param"] == param + assert result["modelOption"] == {"mode": 5} + assert result["cron"] == original["cron"] + + +@pytest.mark.parametrize("body", [{"cron": ""}, {"timeZoneId": "Invalid/Zone"}, {"param": []}]) +def test_invalid_update_does_not_mutate_inventory(tmp_path, body): + client, paths = _jobs_client(tmp_path, "robot-test", [_native_job()]) + before = paths.inventory_path.read_bytes() + item = "/user/devices/robot-test/jobs/42001" + raw, headers = _signed_json(paths, item, body) + result = client.put(item, content=raw, headers=headers).json() + assert result["success"] is False and result["code"] == 400 + assert paths.inventory_path.read_bytes() == before + + +def test_creates_in_same_millisecond_get_distinct_ids_and_explicit_collision_is_rejected(tmp_path, monkeypatch): + from shared import schedule_store + monkeypatch.setattr(schedule_store.time, "time_ns", lambda: 1800000000001000000) + client, paths = _jobs_client(tmp_path, "robot-test", []) + collection = "/user/devices/robot-test/jobs" + ids = [] + for _ in range(2): + raw, headers = _signed_json(paths, collection, {"cron": "0 9 * * *"}) + ids.append(client.post(collection, content=raw, headers=headers).json()["result"]["id"]) + assert len(set(ids)) == 2 + raw, headers = _signed_json(paths, collection, {"id": ids[0], "cron": "0 10 * * *"}) + result = client.post(collection, content=raw, headers=headers).json() + assert result["success"] is False and result["code"] == 409 + + +def test_http_schedule_write_failure_is_not_acknowledged_as_success(tmp_path, monkeypatch): + from shared import schedule_store + client, paths = _jobs_client(tmp_path, "robot-test", [_native_job()]) + before = paths.inventory_path.read_bytes() + monkeypatch.setattr(schedule_store, "write_inventory", lambda *args: False) + item = "/user/devices/robot-test/jobs/42001" + raw, headers = _signed_json(paths, item, {"cron": "0 10 * * *"}) + result = client.put(item, content=raw, headers=headers).json() + assert result["success"] is False and result["code"] == 500 + assert paths.inventory_path.read_bytes() == before diff --git a/tests/test_timer_rpc.py b/tests/test_timer_rpc.py new file mode 100644 index 00000000..4e951025 --- /dev/null +++ b/tests/test_timer_rpc.py @@ -0,0 +1,314 @@ +import asyncio +import json +import logging +from types import SimpleNamespace + +import pytest + +import roborock_local_server.backend # noqa: F401 (puts the bundled `shared` package on sys.path) +from roborock.protocol import create_mqtt_decoder, create_mqtt_encoder +from roborock.protocols.v1_protocol import decode_rpc_response +from roborock.roborock_message import RoborockMessage, RoborockMessageProtocol +from roborock_local_server.bundled_backend.mqtt_broker_server.topic_bridge import ( + CloudTopicKey, + MqttTopicBridge, +) +from shared import schedule_store, timer_rpc + +DUID = "2e2txkqKqxVifcIdyOT1lY" +LOCALKEY = "AoYY9mpxTOIUZtg4" + + +def _cloud_job(job_id, cron, *, enabled=True, **extra): + job = { + "id": job_id, + "cron": cron, + "timeZoneId": "America/New_York", + "repeated": True, + "enabled": enabled, + "param": { + "id": 1, + "method": "server_scheduled_start", + "params": [{"repeat": 1, "water_box_mode": 203, "segments": "0", "fan_power": 104, "mop_mode": 300}], + }, + } + job.update(extra) + return job + + +@pytest.fixture +def ctx(tmp_path): + inventory = { + "schedules": { + DUID: [ + _cloud_job(3549431, "00 11 ? * 1,3,5"), + _cloud_job(3549434, "00 11 ? * 0,6", enabled=False), + _cloud_job(3878757, "03 13 15 12 ?", repeated=False), + ] + } + } + (tmp_path / "web_api_inventory.json").write_text(json.dumps(inventory), encoding="utf-8") + creds = SimpleNamespace(resolve_device=lambda duid="": {"localkey": LOCALKEY} if duid == DUID else None) + return SimpleNamespace( + http_jsonl=tmp_path / "http.jsonl", loggers={}, timezone="America/New_York", + runtime_credentials=creds, localkey="", + ) + + +def test_get_timer_lists_cloud_schedules_in_the_shape_localrock_parses(ctx): + result = timer_rpc.handle_timer_request(ctx, DUID, "get_timer", None) + # The one-off (Dec 15) keeps its day and month, the form the robot and the official app use. + settings = {"repeat": 1, "water_box_mode": 203, "segments": "0", "fan_power": 104, "mop_mode": 300} + assert result == [ + ["3549431", "on", ["00 11 * * 1,3,5", ["start_clean", settings]]], + ["3549434", "off", ["00 11 * * 0,6", ["start_clean", settings]]], + ["3878757", "on", ["03 13 15 12 *", ["start_clean", settings]]], + ] + + +@pytest.mark.parametrize("serialized", [False, True]) +def test_native_jobs_are_not_fabricated_as_legacy_timers(ctx, serialized): + for job in schedule_store.list_jobs(ctx, DUID): + param = job["param"] + param["params"][0]["name"] = "native-" + str(job["id"]) + schedule_store.upsert_job(ctx, DUID, job_id=job["id"], cron=job["cron"], + enabled=job["enabled"], param=param) + if serialized: + path = ctx.http_jsonl.parent / "web_api_inventory.json" + inventory = json.loads(path.read_text()) + for job in inventory["schedules"][DUID]: + job["param"] = json.dumps(job["param"]) + path.write_text(json.dumps(inventory)) + assert timer_rpc.handle_timer_request(ctx, DUID, "get_timer", []) == [] + assert len(schedule_store.list_jobs(ctx, DUID)) == 3 + + +@pytest.mark.parametrize( + ("cron", "expected"), + [ + ("15 23 18 10 ?", "15 23 18 10 *"), + ("15 23 18 10 *", "15 23 18 10 *"), + ("00 11 ? * 1,3,5", "00 11 * * 1,3,5"), + ("0 8 * * 0,1,2,3,4,5,6", "0 8 * * 0,1,2,3,4,5,6"), + # Only one of day and month pinned has no one-time form. + ("0 8 15 * *", None), + ("0 8 * 10 *", None), + ("not a cron", None), + ], +) +def test_device_cron_keeps_a_one_time_schedules_date(cron, expected): + assert timer_rpc.device_cron(cron) == expected + + +def test_set_timer_creates_a_job_and_get_timer_returns_it(ctx): + params = [["1790000000", ["on", ["30 7 * * 1,2", ["start_clean", {}]]]]] + assert timer_rpc.handle_timer_request(ctx, DUID, "set_timer", params) == ["ok"] + + job = next(j for j in schedule_store.list_jobs(ctx, DUID) if str(j["id"]) == "1790000000") + assert job["cron"] == "30 7 * * 1,2" + assert job["enabled"] is True + assert job["timeZoneId"] == "America/New_York" + assert job["param"]["method"] == "server_scheduled_start" + + listed = timer_rpc.handle_timer_request(ctx, DUID, "get_timer", None) + default_settings = {"repeat": 1, "segments": "0"} + assert ["1790000000", "on", ["30 7 * * 1,2", ["start_clean", default_settings]]] in listed + + +def test_editing_a_cloud_schedule_keeps_its_vac_and_mop_settings(ctx): + params = [["3549431", ["on", ["15 9 * * 1,3,5", ["start_clean", {}]]]]] + timer_rpc.handle_timer_request(ctx, DUID, "set_timer", params) + + job = next(j for j in schedule_store.list_jobs(ctx, DUID) if j["id"] == 3549431) + assert job["cron"] == "15 9 * * 1,3,5" + entry = job["param"]["params"][0] + assert (entry["fan_power"], entry["water_box_mode"], entry["mop_mode"]) == (104, 203, 300) + + +def test_upd_timer_toggles_and_del_timer_removes_a_cloud_schedule(ctx): + assert timer_rpc.handle_timer_request(ctx, DUID, "upd_timer", [["3549431", "off"]]) == ["ok"] + job = next(j for j in schedule_store.list_jobs(ctx, DUID) if j["id"] == 3549431) + assert job["enabled"] is False + + assert timer_rpc.handle_timer_request(ctx, DUID, "del_timer", ["3549431"]) == ["ok"] + assert all(j["id"] != 3549431 for j in schedule_store.list_jobs(ctx, DUID)) + + +def test_set_timer_rejects_an_unparseable_cron(ctx): + with pytest.raises(timer_rpc.TimerRequestError): + timer_rpc.handle_timer_request(ctx, DUID, "set_timer", [["1", ["on", ["not a cron", []]]]]) + + +def _encode_request(method, params, request_id=4242): + payload = json.dumps( + {"t": 1790000000, "dps": {"101": json.dumps({"id": request_id, "method": method, "params": params})}} + ).encode() + message = RoborockMessage(protocol=RoborockMessageProtocol.RPC_REQUEST, payload=payload) + return create_mqtt_encoder(LOCALKEY)(message) + + +def _decode_reply(reply): + (message,) = create_mqtt_decoder(LOCALKEY)(reply) + return decode_rpc_response(message) + + +def test_interceptor_answers_get_timer_with_an_encrypted_rpc_response(ctx): + reply = timer_rpc.TimerRpcInterceptor(ctx)(DUID, _encode_request("get_timer", [])) + assert reply is not None + response = _decode_reply(reply) + assert response.request_id == 4242 + assert [entry[0] for entry in response.data] == ["3549431", "3549434", "3878757"] + + +def test_interceptor_forwards_everything_that_is_not_a_timer_request(ctx): + interceptor = timer_rpc.TimerRpcInterceptor(ctx) + assert interceptor(DUID, _encode_request("get_status", [])) is None + assert interceptor(DUID, b"not a roborock frame") is None + assert interceptor("unknown-duid", _encode_request("get_timer", [])) is None + + +@pytest.mark.parametrize( + "method", ["get_server_timer", "set_server_timer", "upd_server_timer", "del_server_timer"] +) +def test_interceptor_leaves_server_timer_to_the_robot(ctx, method): + # A robot that speaks server_timer stores those entries itself and the official app reads + # them back from it; answering this locally hides the robot's real schedules from every + # client and has caused real data loss. This must always forward, never answer here. + interceptor = timer_rpc.TimerRpcInterceptor(ctx) + assert interceptor(DUID, _encode_request(method, [])) is None + + +def test_native_write_observation_preserves_wire_payload_and_waits_for_confirmation(ctx): + from shared.native_schedule_sync import NativeScheduleSync + from roborock_local_server.bundled_backend.mqtt_broker_server.topic_bridge import DeviceTopicKey + + coordinator = NativeScheduleSync(ctx) + bridge = MqttTopicBridge( + host="127.0.0.1", port=1883, logger=logging.getLogger("native-sync-test"), + request_interceptor=timer_rpc.TimerRpcInterceptor(ctx), + request_observer=coordinator.observe_request, + ) + device = DeviceTopicKey(did="robot-did", mqtt_usr="robot-user") + bridge._remember_device_seen(device) + before = schedule_store.list_jobs(ctx, DUID) + sent = [] + + class Client: + async def publish(self, topic, payload, **kwargs): + sent.append((topic, payload)) + + async def exercise(): + for session, enabled in (("homebridge", False), ("oem", True), ("localrock", False)): + cloud = CloudTopicKey(rriot_u=session, mqtt_username=session, duid=DUID) + payload = _encode_request("upd_server_timer", [["native-timer", "on" if enabled else "off"]]) + message = SimpleNamespace(payload=payload, qos=0, retain=False) + await bridge._handle_cloud_message(Client(), message, cloud, cloud.topic_in) + assert sent[-1] == (device.topic_out, payload) + assert schedule_store.list_jobs(ctx, DUID) == before + assert coordinator._generation[DUID] == 3 + + asyncio.run(exercise()) + + +def test_retained_native_write_is_not_observed_as_a_new_change(ctx): + from shared.native_schedule_sync import NativeScheduleSync + from roborock_local_server.bundled_backend.mqtt_broker_server.topic_bridge import DeviceTopicKey + coordinator = NativeScheduleSync(ctx) + bridge = MqttTopicBridge(host="127.0.0.1", port=1883, logger=logging.getLogger("native-sync-test"), + request_observer=coordinator.observe_request) + bridge._remember_device_seen(DeviceTopicKey(did="robot-did", mqtt_usr="robot-user")) + cloud = CloudTopicKey(rriot_u="app", mqtt_username="app", duid=DUID) + class Client: + async def publish(self, *args, **kwargs): + pass + message = SimpleNamespace(payload=_encode_request("upd_server_timer", [["native-timer", "on"]]), qos=0, retain=True) + asyncio.run(bridge._handle_cloud_message(Client(), message, cloud, cloud.topic_in)) + assert not coordinator._dirty + + +def test_bridge_answers_timer_requests_locally_and_does_not_forward_them(ctx): + class Client: + def __init__(self): + self.published = [] + + async def publish(self, topic, payload, *, qos=0, retain=False): + self.published.append((topic, payload)) + + bridge = MqttTopicBridge( + host="127.0.0.1", + port=1883, + logger=logging.getLogger("test.timer_bridge"), + request_interceptor=timer_rpc.TimerRpcInterceptor(ctx), + ) + cloud = CloudTopicKey(rriot_u="u", mqtt_username="m", duid=DUID) + message = SimpleNamespace(payload=_encode_request("get_timer", []), qos=0, retain=False) + client = Client() + + asyncio.run(bridge._handle_cloud_message(client, message, cloud, cloud.topic_in)) + + assert [topic for topic, _ in client.published] == [cloud.topic_out] + assert _decode_reply(client.published[0][1]).request_id == 4242 + + +def _job(ctx, job_id): + return next(j for j in schedule_store.list_jobs(ctx, DUID) if str(j["id"]) == str(job_id)) + + +def test_set_timer_carries_vac_and_mop_settings_in_the_action_object(ctx): + action = ["start_clean", {"fan_power": 102, "water_box_mode": 200, "repeat": 2}] + params = [["1790000001", ["on", ["0 9 * * 2,4", action]]]] + assert timer_rpc.handle_timer_request(ctx, DUID, "set_timer", params) == ["ok"] + + entry = _job(ctx, 1790000001)["param"]["params"][0] + assert (entry["fan_power"], entry["water_box_mode"], entry["repeat"]) == (102, 200, 2) + + listed = timer_rpc.handle_timer_request(ctx, DUID, "get_timer", None) + row = next(item for item in listed if item[0] == "1790000001") + assert row[2][1][1]["water_box_mode"] == 200 + + +def test_changing_one_setting_leaves_the_others_alone(ctx): + action = ["start_clean", {"water_box_mode": 200}] # vac only on this cloud schedule + timer_rpc.handle_timer_request(ctx, DUID, "set_timer", [["3549431", ["on", ["00 11 * * 1,3,5", action]]]]) + + entry = _job(ctx, 3549431)["param"]["params"][0] + assert entry["water_box_mode"] == 200 + assert (entry["fan_power"], entry["mop_mode"]) == (104, 300) + + +def test_an_empty_action_object_does_not_reset_existing_settings(ctx): + # What older LocalRock builds send: ["start_clean", {}]. + action = ["start_clean", {}] + timer_rpc.handle_timer_request(ctx, DUID, "set_timer", [["3549431", ["on", ["30 8 * * 1,3,5", action]]]]) + + job = _job(ctx, 3549431) + assert job["cron"] == "30 8 * * 1,3,5" + assert job["param"]["params"][0]["water_box_mode"] == 203 + + +def test_null_clears_a_setting_and_segments_are_normalised(ctx): + action = ["start_clean", {"mop_mode": None, "segments": [16, 17]}] + timer_rpc.handle_timer_request(ctx, DUID, "set_timer", [["3549431", ["on", ["00 11 * * 1,3,5", action]]]]) + + entry = _job(ctx, 3549431)["param"]["params"][0] + assert "mop_mode" not in entry + assert entry["segments"] == "16,17" + + +def test_vac_only_schedule_fires_without_a_mop_template(ctx): + from shared import schedule_runner + + action = ["start_clean", {"water_box_mode": 200, "fan_power": 102}] + timer_rpc.handle_timer_request(ctx, DUID, "set_timer", [["1790000002", ["on", ["0 9 * * *", action]]]]) + + (step,) = schedule_runner.steps_for_job(_job(ctx, 1790000002)) + assert step.params[0]["water_box_mode"] == 200 + assert step.params[0]["fan_power"] == 102 + + +def test_non_numeric_setting_is_rejected_and_nothing_is_saved(ctx): + before = schedule_store.list_jobs(ctx, DUID) + action = ["start_clean", {"fan_power": "loud"}] + with pytest.raises(timer_rpc.TimerRequestError): + timer_rpc.handle_timer_request(ctx, DUID, "set_timer", [["9", ["on", ["0 9 * * *", action]]]]) + assert schedule_store.list_jobs(ctx, DUID) == before