From 1020253036006bfd7b29d9b8bf7aeb96f64325a4 Mon Sep 17 00:00:00 2001 From: Luke Lashley Date: Wed, 21 Oct 2026 21:24:06 -0400 Subject: [PATCH 1/3] feat(web_api): add cloud job updates --- roborock/web_api.py | 25 +++++++++++ tests/test_web_api.py | 99 ++++++++++++++++++++++++++++++++++++++++++- 2 files changed, 123 insertions(+), 1 deletion(-) diff --git a/roborock/web_api.py b/roborock/web_api.py index 88ed2b1c..cdadb904 100644 --- a/roborock/web_api.py +++ b/roborock/web_api.py @@ -768,6 +768,31 @@ async def create_job(self, user_data: UserData, device_id: str, job: dict) -> di raise RoborockException(response) return response + async def update_job(self, user_data: UserData, device_id: str, job: HomeDataSchedule) -> None: + """Replace an existing cloud job with the full supplied schedule. + + Pass a schedule returned by :meth:`get_schedules` with the desired changes. + The job ID identifies the URL; all other fields are sent as the replacement body. + """ + rriot = user_data.rriot + if rriot is None: + raise RoborockException("rriot is none") + if rriot.r.a is None: + raise RoborockException("Missing field 'a' in rriot reference") + path = f"/user/devices/{device_id}/jobs/{job.id}" + body = job.as_dict(exclude={"id"}) + job_request = PreparedRequest( + rriot.r.a, + self.session, + { + "Authorization": _get_hawk_authentication(rriot, path, body=body), + "Content-Type": "application/json", + }, + ) + response = await job_request.request("put", path, data=_compact_json(body).encode()) + if not response or not response.get("success"): + raise RoborockException(response) + async def get_products(self, user_data: UserData) -> ProductResponse: """Gets all products and their schemas, good for determining status codes and model numbers.""" base_url = await self.base_url diff --git a/tests/test_web_api.py b/tests/test_web_api.py index a4e38ec7..72038435 100644 --- a/tests/test_web_api.py +++ b/tests/test_web_api.py @@ -1,3 +1,4 @@ +import json import re from typing import Any from unittest.mock import AsyncMock, Mock @@ -6,7 +7,7 @@ import pytest from aioresponses.compat import normalize_url -from roborock import HomeData, HomeDataRoom, HomeDataScene, UserData +from roborock import HomeData, HomeDataRoom, HomeDataScene, HomeDataSchedule, UserData from roborock.exceptions import RoborockAccountDoesNotExist, RoborockException, RoborockInvalidCredentials from roborock.web_api import ( DEFAULT_AGREEMENT_MAJOR_VERSION, @@ -483,6 +484,102 @@ async def test_create_job(mock_rest) -> None: assert response["result"] == "done" +@pytest.mark.parametrize( + ("enabled", "mac"), + [ + (True, "fCO2ogG/WjxJjWEZw6l3ncC7cFChl/95jjjzu/YNRV0="), + (False, "GsGuHY1XF375VJRdhMmoIYI/AacILSNCFBU1+3G/O0Y="), + ], +) +async def test_update_job(mock_rest, monkeypatch, enabled: bool, mac: str) -> None: + """Updates send the full schedule as compact JSON with a matching body-aware Hawk MAC.""" + api = RoborockApiClient(username="test_user@gmail.com") + ud = UserData.from_dict(USER_DATA) + monkeypatch.setattr("roborock.web_api.time.time", lambda: 1_700_000_000) + monkeypatch.setattr("roborock.web_api.secrets.token_urlsafe", lambda n: "fixednonce") + url = "https://api-us.roborock.com/user/devices/123456/jobs/3878757" + mock_rest.put(url, payload={"result": "done", "status": "ok", "success": True}) + job = HomeDataSchedule( + id=3878757, + cron="05 10 * * ?", + repeated=True, + enabled=enabled, + param={"mapId": 0, "rooms": [1, 2], "roomCount": 2, "custom_key": {"id": 7, "value": None}}, + ) + original_job = job.as_dict() + + await api.update_job(ud, "123456", job) + + enabled_json = b"true" if enabled else b"false" + expected_body = ( + b'{"cron":"05 10 * * ?","repeated":true,"enabled":' + + enabled_json + + b',"param":{"mapId":0,"rooms":[1,2],"roomCount":2,"custom_key":{"id":7,"value":null}}}' + ) + mock_rest.assert_called_once() + request = mock_rest.requests[("put", normalize_url(url))][0] + assert request.kwargs["data"] == expected_body + assert request.kwargs["headers"] == { + "Authorization": f'Hawk id="user123",s="pass123",ts="1700000000",nonce="fixednonce",mac="{mac}"', + "Content-Type": "application/json", + } + assert request.kwargs["params"] is None + assert request.kwargs["json"] is None + assert job.as_dict() == original_job + + +@pytest.mark.parametrize("response", [{"success": False, "error": "Job not found"}, {}, None]) +async def test_update_job_failed_response(mock_rest, response: dict | None) -> None: + """Rejected and missing success responses raise the library's exception.""" + api = RoborockApiClient(username="test_user@gmail.com") + ud = UserData.from_dict(USER_DATA) + mock_rest.put( + "https://api-us.roborock.com/user/devices/123456/jobs/3878757", + body=json.dumps(response), + content_type="application/json", + ) + job = HomeDataSchedule(id=3878757, cron="05 10 * * ?", repeated=True, enabled=False) + + with pytest.raises(RoborockException) as exc_info: + await api.update_job(ud, "123456", job) + + assert exc_info.value.args == (response,) + + +@pytest.mark.parametrize("missing_field", ["rriot", "a"]) +async def test_update_job_missing_credentials(mock_rest, missing_field: str) -> None: + """Missing cloud credentials are rejected before making a request.""" + api = RoborockApiClient(username="test_user@gmail.com") + ud = UserData.from_dict(USER_DATA) + if missing_field == "rriot": + ud.rriot = None + message = "rriot is none" + else: + assert ud.rriot is not None + ud.rriot.r.a = None + message = "Missing field 'a' in rriot reference" + job = HomeDataSchedule(id=3878757, cron="05 10 * * ?", repeated=True, enabled=False) + + with pytest.raises(RoborockException, match=message): + await api.update_job(ud, "123456", job) + + assert not mock_rest.requests + + +async def test_update_job_network_error(mock_rest) -> None: + """Network failures surface as RoborockException through the shared request layer.""" + api = RoborockApiClient(username="test_user@gmail.com") + ud = UserData.from_dict(USER_DATA) + mock_rest.put( + "https://api-us.roborock.com/user/devices/123456/jobs/3878757", + exception=aiohttp.ClientConnectionError("connection lost"), + ) + job = HomeDataSchedule(id=3878757, cron="05 10 * * ?", repeated=True, enabled=False) + + with pytest.raises(RoborockException, match="Network error contacting .*: connection lost"): + await api.update_job(ud, "123456", job) + + def test_hawk_authentication_signs_body(monkeypatch) -> None: """Body-bearing writes sign md5(compact JSON body) in the Hawk payload slot; GET is unchanged.""" from roborock import web_api From a4c3a4071297f68d862b786458c4ce56efbb86bc Mon Sep 17 00:00:00 2001 From: Luke Lashley Date: Wed, 21 Oct 2026 21:33:02 -0400 Subject: [PATCH 2/3] feat(cli): add cloud job update command --- roborock/cli.py | 58 ++++++++++++++- tests/test_cli.py | 181 ++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 238 insertions(+), 1 deletion(-) create mode 100644 tests/test_cli.py diff --git a/roborock/cli.py b/roborock/cli.py index 5d1e1cbc..426621e9 100644 --- a/roborock/cli.py +++ b/roborock/cli.py @@ -50,7 +50,7 @@ ) from err from roborock import RoborockCommand -from roborock.data import RoborockBase, UserData +from roborock.data import HomeDataSchedule, RoborockBase, UserData from roborock.data.b01_q10.b01_q10_code_mappings import ( B01_Q10_DP, YXCleanType, @@ -428,6 +428,61 @@ async def execute_scene(ctx, scene_id): await client.execute_scene(connection_cache.user_data, scene_id) +def _load_job_file(ctx: click.Context, param: click.Parameter, value: Path) -> HomeDataSchedule: + """Read a complete job replacement without coercing user-supplied JSON types.""" + try: + data = json.loads(value.read_text(encoding="utf-8")) + except (OSError, UnicodeError, json.JSONDecodeError) as err: + raise click.BadParameter(f"Could not read job JSON: {err}") from err + if not isinstance(data, dict): + raise click.BadParameter("Expected a JSON object") + required = {"id", "cron", "repeated", "enabled", "param"} + if missing := required - data.keys(): + raise click.BadParameter(f"Complete job required; missing fields: {', '.join(sorted(missing))}") + if unknown := data.keys() - required: + raise click.BadParameter(f"Unknown job fields: {', '.join(sorted(unknown))}") + if type(data["id"]) is not int: + raise click.BadParameter("'id' must be an integer") + if not isinstance(data["cron"], str) or not data["cron"].strip(): + raise click.BadParameter("'cron' must be a nonempty string") + for field in ("repeated", "enabled"): + if not isinstance(data[field], bool): + raise click.BadParameter(f"'{field}' must be a boolean") + if not isinstance(data["param"], dict): + raise click.BadParameter("'param' must be an object") + return HomeDataSchedule( + id=data["id"], cron=data["cron"], repeated=data["repeated"], enabled=data["enabled"], param=data["param"] + ) + + +@click.command() +@click.option("--device_id", required=True) +@click.option( + "--job-file", + "job", + required=True, + type=click.Path(exists=True, dir_okay=False, readable=True, path_type=Path), + callback=_load_job_file, + help="JSON file containing id, cron, repeated, enabled, and param (object).", +) +@click.pass_context +@async_command +async def update_job(ctx: click.Context, device_id: str, job: HomeDataSchedule) -> None: + """Replace a cloud job using a complete schedule JSON file. + + Include all existing settings in the file, including unchanged cleaning parameters. + """ + context: RoborockContext = ctx.obj + try: + connection_cache = context.connection_cache() + client = RoborockApiClient(connection_cache.email) + await client.update_job(connection_cache.user_data, device_id, job) + except RoborockException as err: + click.echo(f"Error: {err}", err=True) + raise SystemExit(1) from err + click.echo(f"Updated job {job.id} for device {device_id}") + + async def _v1_trait(context: RoborockContext, device_id: str, display_func: Callable[[], V1TraitMixin]) -> Trait: device_manager = await context.get_device_manager() device = await device_manager.get_device(device_id) @@ -1366,6 +1421,7 @@ def write_markdown_table(product_features: dict[str, dict[str, any]], all_featur cli.add_command(list_devices) cli.add_command(list_scenes) cli.add_command(execute_scene) +cli.add_command(update_job) cli.add_command(status) cli.add_command(command) cli.add_command(parser) diff --git a/tests/test_cli.py b/tests/test_cli.py new file mode 100644 index 00000000..482ee32f --- /dev/null +++ b/tests/test_cli.py @@ -0,0 +1,181 @@ +"""Tests for the command line interface.""" + +import json +from pathlib import Path +from typing import Any +from unittest.mock import Mock + +import pytest +from click.testing import CliRunner + +from roborock.cli import ConnectionCache, RoborockContext, cli +from roborock.data import HomeDataSchedule, UserData +from roborock.exceptions import RoborockException +from roborock.web_api import RoborockApiClient +from tests.mock_data import USER_DATA, USER_EMAIL + + +@pytest.fixture(autouse=True) +def isolated_login(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + """Use temporary cached credentials so commands never load the user's login.""" + login_file = tmp_path / ".roborock" + monkeypatch.setattr(RoborockContext, "roborock_file", login_file) + monkeypatch.setattr(RoborockContext, "roborock_cache_file", tmp_path / ".roborock.cache") + login_file.write_text( + json.dumps(ConnectionCache(user_data=UserData.from_dict(USER_DATA), email=USER_EMAIL).as_dict()), + encoding="utf-8", + ) + + +@pytest.fixture +def mock_api(monkeypatch: pytest.MonkeyPatch) -> Mock: + """Replace the cloud client, including all asynchronous network operations.""" + factory = Mock(return_value=Mock(spec=RoborockApiClient)) + monkeypatch.setattr("roborock.cli.RoborockApiClient", factory) + return factory + + +@pytest.fixture +def job_data() -> dict[str, Any]: + """A full schedule including model-specific cleaning settings.""" + return { + "id": 3878757, + "cron": "05 10 * * ?", + "repeated": True, + "enabled": False, + "param": {"mapId": 0, "rooms": [1, 2], "custom_key": {"id": 7, "value": None}}, + } + + +@pytest.mark.parametrize("enabled", [False, True]) +def test_update_job(mock_api: Mock, tmp_path: Path, job_data: dict[str, Any], enabled: bool) -> None: + """The CLI passes a complete typed schedule using only the cached account.""" + job_data["enabled"] = enabled + job_file = tmp_path / "job.json" + job_file.write_text(json.dumps(job_data), encoding="utf-8") + + result = CliRunner().invoke(cli, ["update-job", "--device_id", "123456", "--job-file", str(job_file)]) + + assert result.exit_code == 0, result.output + assert result.output == "Updated job 3878757 for device 123456\n" + mock_api.assert_called_once_with(USER_EMAIL) + client = mock_api.return_value + client.update_job.assert_awaited_once_with(UserData.from_dict(USER_DATA), "123456", HomeDataSchedule(**job_data)) + client.get_home_data_v3.assert_not_called() + + +def test_update_job_api_error(mock_api: Mock, tmp_path: Path, job_data: dict[str, Any]) -> None: + """A cloud rejection produces an unsuccessful command exit.""" + mock_api.return_value.update_job.side_effect = RoborockException("Job not found") + job_file = tmp_path / "job.json" + job_file.write_text(json.dumps(job_data), encoding="utf-8") + + result = CliRunner().invoke(cli, ["update-job", "--device_id", "123456", "--job-file", str(job_file)]) + + assert result.exit_code == 1 + assert result.output == "Error: Job not found\n" + mock_api.return_value.update_job.assert_awaited_once() + + +def test_update_job_requires_login(mock_api: Mock, tmp_path: Path, job_data: dict[str, Any]) -> None: + """Missing cached credentials fail without creating a cloud client.""" + (tmp_path / ".roborock").unlink() + job_file = tmp_path / "job.json" + job_file.write_text(json.dumps(job_data), encoding="utf-8") + + result = CliRunner().invoke(cli, ["update-job", "--device_id", "123456", "--job-file", str(job_file)]) + + assert result.exit_code == 1 + assert result.output == "Error: You must login first\n" + mock_api.assert_not_called() + + +@pytest.mark.parametrize("field", ["id", "cron", "repeated", "enabled", "param"]) +def test_update_job_missing_fields(mock_api: Mock, tmp_path: Path, job_data: dict[str, Any], field: str) -> None: + """Partial schedule files cannot reach the update endpoint.""" + del job_data[field] + job_file = tmp_path / "job.json" + job_file.write_text(json.dumps(job_data), encoding="utf-8") + + result = CliRunner().invoke(cli, ["update-job", "--device_id", "123456", "--job-file", str(job_file)]) + + assert result.exit_code == 2 + assert f"Complete job required; missing fields: {field}" in result.output + mock_api.assert_not_called() + + +@pytest.mark.parametrize( + ("field", "value"), + [ + ("id", True), + ("id", "3878757"), + ("id", 1.5), + ("cron", None), + ("cron", " "), + ("repeated", 1), + ("enabled", "false"), + ("param", None), + ("param", []), + ("param", "{}"), + ], +) +def test_update_job_invalid_fields( + mock_api: Mock, tmp_path: Path, job_data: dict[str, Any], field: str, value: Any +) -> None: + """Malformed scalar types are rejected instead of silently coerced.""" + job_data[field] = value + job_file = tmp_path / "job.json" + job_file.write_text(json.dumps(job_data), encoding="utf-8") + + result = CliRunner().invoke(cli, ["update-job", "--device_id", "123456", "--job-file", str(job_file)]) + + assert result.exit_code == 2 + assert f"'{field}' must be" in result.output + mock_api.assert_not_called() + + +def test_update_job_unknown_fields(mock_api: Mock, tmp_path: Path, job_data: dict[str, Any]) -> None: + """Unknown top-level settings cannot be silently discarded during replacement.""" + job_data["additionalSetting"] = {"value": 1} + job_file = tmp_path / "job.json" + job_file.write_text(json.dumps(job_data), encoding="utf-8") + + result = CliRunner().invoke(cli, ["update-job", "--device_id", "123456", "--job-file", str(job_file)]) + + assert result.exit_code == 2 + assert "Unknown job fields: additionalSetting" in result.output + mock_api.assert_not_called() + + +@pytest.mark.parametrize("contents", ["[", "[]", "null", '"schedule"']) +def test_update_job_invalid_json(mock_api: Mock, tmp_path: Path, contents: str) -> None: + """Invalid JSON and non-object roots fail before creating a cloud client.""" + job_file = tmp_path / "job.json" + job_file.write_text(contents, encoding="utf-8") + + result = CliRunner().invoke(cli, ["update-job", "--device_id", "123456", "--job-file", str(job_file)]) + + assert result.exit_code == 2 + assert "Invalid value for '--job-file'" in result.output + mock_api.assert_not_called() + + +def test_update_job_missing_file(mock_api: Mock, tmp_path: Path) -> None: + """A nonexistent input file fails without cloud access.""" + result = CliRunner().invoke( + cli, ["update-job", "--device_id", "123456", "--job-file", str(tmp_path / "missing.json")] + ) + + assert result.exit_code == 2 + assert "does not exist" in result.output + mock_api.assert_not_called() + + +def test_update_job_help(mock_api: Mock) -> None: + """Help explains that this operation replaces all schedule settings.""" + result = CliRunner().invoke(cli, ["update-job", "--help"]) + + assert result.exit_code == 0 + assert "Include all existing settings" in result.output + assert "id, cron, repeated, enabled, and param" in result.output + mock_api.assert_not_called() From 262bc026083e2bc8f275f613a8f7bd3bab36a3a5 Mon Sep 17 00:00:00 2001 From: Luke Lashley Date: Wed, 21 Oct 2026 21:39:23 -0400 Subject: [PATCH 3/3] fix(cli): simplify cloud job JSON parsing --- roborock/cli.py | 40 +++------------ tests/test_cli.py | 126 ++++++---------------------------------------- 2 files changed, 22 insertions(+), 144 deletions(-) diff --git a/roborock/cli.py b/roborock/cli.py index 426621e9..c797213c 100644 --- a/roborock/cli.py +++ b/roborock/cli.py @@ -34,7 +34,7 @@ from collections.abc import Callable from dataclasses import asdict, dataclass from pathlib import Path -from typing import Any, cast +from typing import Any, TextIO, cast try: import click @@ -428,56 +428,28 @@ async def execute_scene(ctx, scene_id): await client.execute_scene(connection_cache.user_data, scene_id) -def _load_job_file(ctx: click.Context, param: click.Parameter, value: Path) -> HomeDataSchedule: - """Read a complete job replacement without coercing user-supplied JSON types.""" - try: - data = json.loads(value.read_text(encoding="utf-8")) - except (OSError, UnicodeError, json.JSONDecodeError) as err: - raise click.BadParameter(f"Could not read job JSON: {err}") from err - if not isinstance(data, dict): - raise click.BadParameter("Expected a JSON object") - required = {"id", "cron", "repeated", "enabled", "param"} - if missing := required - data.keys(): - raise click.BadParameter(f"Complete job required; missing fields: {', '.join(sorted(missing))}") - if unknown := data.keys() - required: - raise click.BadParameter(f"Unknown job fields: {', '.join(sorted(unknown))}") - if type(data["id"]) is not int: - raise click.BadParameter("'id' must be an integer") - if not isinstance(data["cron"], str) or not data["cron"].strip(): - raise click.BadParameter("'cron' must be a nonempty string") - for field in ("repeated", "enabled"): - if not isinstance(data[field], bool): - raise click.BadParameter(f"'{field}' must be a boolean") - if not isinstance(data["param"], dict): - raise click.BadParameter("'param' must be an object") - return HomeDataSchedule( - id=data["id"], cron=data["cron"], repeated=data["repeated"], enabled=data["enabled"], param=data["param"] - ) - - @click.command() @click.option("--device_id", required=True) @click.option( "--job-file", - "job", required=True, - type=click.Path(exists=True, dir_okay=False, readable=True, path_type=Path), - callback=_load_job_file, - help="JSON file containing id, cron, repeated, enabled, and param (object).", + type=click.File("r", encoding="utf-8"), + help="JSON file containing the schedule to update.", ) @click.pass_context @async_command -async def update_job(ctx: click.Context, device_id: str, job: HomeDataSchedule) -> None: +async def update_job(ctx: click.Context, device_id: str, job_file: TextIO) -> None: """Replace a cloud job using a complete schedule JSON file. Include all existing settings in the file, including unchanged cleaning parameters. """ context: RoborockContext = ctx.obj try: + job = HomeDataSchedule.from_dict(json.load(job_file)) connection_cache = context.connection_cache() client = RoborockApiClient(connection_cache.email) await client.update_job(connection_cache.user_data, device_id, job) - except RoborockException as err: + except Exception as err: click.echo(f"Error: {err}", err=True) raise SystemExit(1) from err click.echo(f"Updated job {job.id} for device {device_id}") diff --git a/tests/test_cli.py b/tests/test_cli.py index 482ee32f..2dc63835 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -37,7 +37,7 @@ def mock_api(monkeypatch: pytest.MonkeyPatch) -> Mock: @pytest.fixture def job_data() -> dict[str, Any]: - """A full schedule including model-specific cleaning settings.""" + """A schedule including model-specific cleaning settings.""" return { "id": 3878757, "cron": "05 10 * * ?", @@ -47,10 +47,8 @@ def job_data() -> dict[str, Any]: } -@pytest.mark.parametrize("enabled", [False, True]) -def test_update_job(mock_api: Mock, tmp_path: Path, job_data: dict[str, Any], enabled: bool) -> None: - """The CLI passes a complete typed schedule using only the cached account.""" - job_data["enabled"] = enabled +def test_update_job(mock_api: Mock, tmp_path: Path, job_data: dict[str, Any]) -> None: + """The CLI parses JSON into a schedule and calls the API using cached credentials.""" job_file = tmp_path / "job.json" job_file.write_text(json.dumps(job_data), encoding="utf-8") @@ -59,123 +57,31 @@ def test_update_job(mock_api: Mock, tmp_path: Path, job_data: dict[str, Any], en assert result.exit_code == 0, result.output assert result.output == "Updated job 3878757 for device 123456\n" mock_api.assert_called_once_with(USER_EMAIL) - client = mock_api.return_value - client.update_job.assert_awaited_once_with(UserData.from_dict(USER_DATA), "123456", HomeDataSchedule(**job_data)) - client.get_home_data_v3.assert_not_called() - - -def test_update_job_api_error(mock_api: Mock, tmp_path: Path, job_data: dict[str, Any]) -> None: - """A cloud rejection produces an unsuccessful command exit.""" - mock_api.return_value.update_job.side_effect = RoborockException("Job not found") - job_file = tmp_path / "job.json" - job_file.write_text(json.dumps(job_data), encoding="utf-8") - - result = CliRunner().invoke(cli, ["update-job", "--device_id", "123456", "--job-file", str(job_file)]) - - assert result.exit_code == 1 - assert result.output == "Error: Job not found\n" - mock_api.return_value.update_job.assert_awaited_once() + mock_api.return_value.update_job.assert_awaited_once_with( + UserData.from_dict(USER_DATA), "123456", HomeDataSchedule(**job_data) + ) -def test_update_job_requires_login(mock_api: Mock, tmp_path: Path, job_data: dict[str, Any]) -> None: - """Missing cached credentials fail without creating a cloud client.""" - (tmp_path / ".roborock").unlink() +def test_update_job_invalid_json(mock_api: Mock, tmp_path: Path) -> None: + """A JSON parsing failure is reported without calling the cloud API.""" job_file = tmp_path / "job.json" - job_file.write_text(json.dumps(job_data), encoding="utf-8") + job_file.write_text("[", encoding="utf-8") result = CliRunner().invoke(cli, ["update-job", "--device_id", "123456", "--job-file", str(job_file)]) assert result.exit_code == 1 - assert result.output == "Error: You must login first\n" - mock_api.assert_not_called() - - -@pytest.mark.parametrize("field", ["id", "cron", "repeated", "enabled", "param"]) -def test_update_job_missing_fields(mock_api: Mock, tmp_path: Path, job_data: dict[str, Any], field: str) -> None: - """Partial schedule files cannot reach the update endpoint.""" - del job_data[field] - job_file = tmp_path / "job.json" - job_file.write_text(json.dumps(job_data), encoding="utf-8") - - result = CliRunner().invoke(cli, ["update-job", "--device_id", "123456", "--job-file", str(job_file)]) - - assert result.exit_code == 2 - assert f"Complete job required; missing fields: {field}" in result.output - mock_api.assert_not_called() - - -@pytest.mark.parametrize( - ("field", "value"), - [ - ("id", True), - ("id", "3878757"), - ("id", 1.5), - ("cron", None), - ("cron", " "), - ("repeated", 1), - ("enabled", "false"), - ("param", None), - ("param", []), - ("param", "{}"), - ], -) -def test_update_job_invalid_fields( - mock_api: Mock, tmp_path: Path, job_data: dict[str, Any], field: str, value: Any -) -> None: - """Malformed scalar types are rejected instead of silently coerced.""" - job_data[field] = value - job_file = tmp_path / "job.json" - job_file.write_text(json.dumps(job_data), encoding="utf-8") - - result = CliRunner().invoke(cli, ["update-job", "--device_id", "123456", "--job-file", str(job_file)]) - - assert result.exit_code == 2 - assert f"'{field}' must be" in result.output + assert "Error: Expecting value" in result.output mock_api.assert_not_called() -def test_update_job_unknown_fields(mock_api: Mock, tmp_path: Path, job_data: dict[str, Any]) -> None: - """Unknown top-level settings cannot be silently discarded during replacement.""" - job_data["additionalSetting"] = {"value": 1} +def test_update_job_api_error(mock_api: Mock, tmp_path: Path, job_data: dict[str, Any]) -> None: + """A cloud rejection is reported with an unsuccessful command exit.""" + mock_api.return_value.update_job.side_effect = RoborockException("Job not found") job_file = tmp_path / "job.json" job_file.write_text(json.dumps(job_data), encoding="utf-8") result = CliRunner().invoke(cli, ["update-job", "--device_id", "123456", "--job-file", str(job_file)]) - assert result.exit_code == 2 - assert "Unknown job fields: additionalSetting" in result.output - mock_api.assert_not_called() - - -@pytest.mark.parametrize("contents", ["[", "[]", "null", '"schedule"']) -def test_update_job_invalid_json(mock_api: Mock, tmp_path: Path, contents: str) -> None: - """Invalid JSON and non-object roots fail before creating a cloud client.""" - job_file = tmp_path / "job.json" - job_file.write_text(contents, encoding="utf-8") - - result = CliRunner().invoke(cli, ["update-job", "--device_id", "123456", "--job-file", str(job_file)]) - - assert result.exit_code == 2 - assert "Invalid value for '--job-file'" in result.output - mock_api.assert_not_called() - - -def test_update_job_missing_file(mock_api: Mock, tmp_path: Path) -> None: - """A nonexistent input file fails without cloud access.""" - result = CliRunner().invoke( - cli, ["update-job", "--device_id", "123456", "--job-file", str(tmp_path / "missing.json")] - ) - - assert result.exit_code == 2 - assert "does not exist" in result.output - mock_api.assert_not_called() - - -def test_update_job_help(mock_api: Mock) -> None: - """Help explains that this operation replaces all schedule settings.""" - result = CliRunner().invoke(cli, ["update-job", "--help"]) - - assert result.exit_code == 0 - assert "Include all existing settings" in result.output - assert "id, cron, repeated, enabled, and param" in result.output - mock_api.assert_not_called() + assert result.exit_code == 1 + assert result.output == "Error: Job not found\n" + mock_api.return_value.update_job.assert_awaited_once()