From a6fb22c390817bd609e5425d2cb34bd95e318b66 Mon Sep 17 00:00:00 2001 From: Elliot Mackenzie <6545046+barfle@users.noreply.github.com> Date: Mon, 29 Jun 2026 13:38:35 +1000 Subject: [PATCH] Improve CLI reliability and feature-gate messaging. This expands CLI coverage with focused tests and improves command error handling so feature-gated incident-automation endpoints return clear org-specific guidance instead of opaque 404 responses. --- .github/workflows/ci.yml | 20 + dataplicity_cli/cli.py | 548 ++++++++++++++++++++++++++- pyproject.toml | 9 + tests/test_cli_core_helpers.py | 120 ++++++ tests/test_config.py | 77 ++++ tests/test_devices_history_cli.py | 228 +++++++++++ tests/test_entrypoint.py | 16 + tests/test_feature_gated_commands.py | 132 +++++++ tests/test_m2m.py | 82 ++++ tests/test_remote_access_helpers.py | 85 +++++ 10 files changed, 1297 insertions(+), 20 deletions(-) create mode 100644 tests/test_cli_core_helpers.py create mode 100644 tests/test_config.py create mode 100644 tests/test_devices_history_cli.py create mode 100644 tests/test_entrypoint.py create mode 100644 tests/test_feature_gated_commands.py create mode 100644 tests/test_m2m.py create mode 100644 tests/test_remote_access_helpers.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 9091e27..4972fbe 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -6,6 +6,26 @@ on: pull_request: jobs: + unit-tests: + name: Unit tests + runs-on: ubuntu-latest + strategy: + fail-fast: false + matrix: + python-version: ["3.11", "3.12"] + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python-version }} + - name: Install + run: | + python -m pip install --upgrade pip + pip install ".[test]" + - name: Run unit tests + run: | + pytest -q --maxfail=1 --cov=dataplicity_cli --cov-report=term-missing + lint-and-smoke: name: Compile + help smoke runs-on: ${{ matrix.os }} diff --git a/dataplicity_cli/cli.py b/dataplicity_cli/cli.py index 3ece333..b759c41 100644 --- a/dataplicity_cli/cli.py +++ b/dataplicity_cli/cli.py @@ -49,6 +49,7 @@ auth_app = typer.Typer(help="Authentication commands") orgs_app = typer.Typer(help="Organisation commands") devices_app = typer.Typer(help="Device commands") +devices_history_app = typer.Typer(help="Device history commands", no_args_is_help=True) config_app = typer.Typer(help="Configuration commands") api_app = typer.Typer(help="Raw API commands") endpoint_monitors_app = typer.Typer(help="Endpoint monitor commands") @@ -64,6 +65,7 @@ app.add_typer(auth_app, name="auth") app.add_typer(orgs_app, name="org") app.add_typer(devices_app, name="devices") +devices_app.add_typer(devices_history_app, name="history") app.add_typer(config_app, name="config") app.add_typer(api_app, name="api") app.add_typer(endpoint_monitors_app, name="endpoint-monitors") @@ -603,6 +605,57 @@ def _friendly_response_message(default_message: str, response_data: Any, respons return message +def _incident_automation_org_hash(endpoint: str) -> str: + match = re.search(r"/api/organisations/([^/]+)/", endpoint) + if not match: + return "" + return str(match.group(1) or "").strip() + + +def _is_incident_automation_endpoint(endpoint: str) -> bool: + return "/incident-automation/" in endpoint + + +def _is_incident_automation_collection_endpoint(endpoint: str) -> bool: + trimmed = endpoint.strip().rstrip("/") + match = re.search(r"/incident-automation/([^/]+)$", trimmed) + return match is not None + + +def _incident_automation_collection_endpoint(endpoint: str) -> str: + if not _is_incident_automation_endpoint(endpoint): + return endpoint + base = endpoint.split("/incident-automation/", 1)[0] + suffix = endpoint.split("/incident-automation/", 1)[1] + resource = suffix.strip("/").split("/", 1)[0] + return f"{base}/incident-automation/{resource}/" + + +def _incident_automation_feature_unavailable_message(endpoint: str, *, feature_name: str) -> str: + org_hash = _incident_automation_org_hash(endpoint) + if org_hash: + return f"{feature_name} are not available for organisation `{org_hash}`." + return f"{feature_name} are not available for this organisation." + + +def _endpoint_error_message( + state: AppContext, + *, + endpoint: str, + response: ApiResponse, + default_message: str, + feature_name: Optional[str] = None, +) -> str: + if response.status_code == 404 and feature_name and _is_incident_automation_endpoint(endpoint): + if _is_incident_automation_collection_endpoint(endpoint): + return _incident_automation_feature_unavailable_message(endpoint, feature_name=feature_name) + probe_endpoint = _incident_automation_collection_endpoint(endpoint) + probe = state.api.get(probe_endpoint) + if probe.status_code == 404: + return _incident_automation_feature_unavailable_message(endpoint, feature_name=feature_name) + return _friendly_response_message(default_message, response.data, response.text) + + def _normalize_email(value: Optional[str]) -> str: return str(value or "").strip().lower() @@ -1216,6 +1269,63 @@ def _resolve_device_hash_interactive( return _device_hash(devices[idx - 1]) +def _extract_timeline_rows(payload: Any) -> List[Dict[str, Any]]: + if isinstance(payload, dict): + rows = payload.get("results") + if isinstance(rows, list): + return [item for item in rows if isinstance(item, dict)] + if isinstance(payload, list): + return [item for item in payload if isinstance(item, dict)] + return [] + + +def _timeline_event_id(row: Dict[str, Any]) -> Optional[int]: + raw = row.get("id") + if isinstance(raw, int): + return raw + if isinstance(raw, str): + with suppress(ValueError): + return int(raw) + return None + + +def _timeline_entry_not_found_message(event_id: int) -> str: + return ( + f"History entry `{event_id}` was not found in the latest device timeline window. " + "Use `devices history list --limit 500 --include-deleted` to inspect more rows." + ) + + +DEVICE_TIMELINE_ALLOWED_TYPES = { + "general", + "lifecycle", + "service", + "logistics", + "warranty", + "rma", + "shipping", + "diagnostics", + "customer", +} +DEVICE_TIMELINE_OPERATOR_RESERVED_TYPES = {"lifecycle"} +DEVICE_TIMELINE_OPERATOR_ALLOWED_TYPES = tuple( + sorted(DEVICE_TIMELINE_ALLOWED_TYPES - DEVICE_TIMELINE_OPERATOR_RESERVED_TYPES) +) + + +def _normalize_timeline_type(raw_value: Optional[str]) -> str: + candidate = str(raw_value or "").strip().lower() + if candidate in DEVICE_TIMELINE_ALLOWED_TYPES: + return candidate + return "general" + + +def _normalize_timeline_type_filters(raw_value: Optional[str]) -> List[str]: + pieces = [part.strip().lower() for part in str(raw_value or "").split(",")] + cleaned = [piece for piece in pieces if piece in DEVICE_TIMELINE_ALLOWED_TYPES] + return list(dict.fromkeys(cleaned)) + + @app.callback() def main( ctx: typer.Context, @@ -2052,6 +2162,291 @@ def devices_show(ctx: typer.Context, device_hash: Optional[str] = typer.Argument state.console.print(table) +@devices_history_app.command("list") +def devices_history_list( + ctx: typer.Context, + device_hash: Optional[str] = typer.Argument(None), + limit: int = typer.Option(120, "--limit", help="Maximum number of history entries (1-500)"), + source: Optional[str] = typer.Option(None, "--source", help="Filter by source: operator or system"), + timeline_type: Optional[str] = typer.Option( + None, + "--type", + help="Filter by timeline type (comma-separated values allowed)", + ), + search: Optional[str] = typer.Option(None, "--search", help="Search in message/event type"), + include_deleted: bool = typer.Option(False, "--include-deleted", help="Include soft-deleted operator comments"), +) -> None: + """List history entries for a device. + + Examples: + dataplicity devices history list + dataplicity devices history list --source operator --type logistics + dataplicity devices history list --search reboot --include-deleted + """ + state = _ctx(ctx) + _require_auth(state) + resolved_hash = _resolve_device_hash_interactive(state, device_hash, action_name="history list") + params: Dict[str, Any] = {"limit": max(1, min(limit, 500))} + if source: + source_value = source.strip().lower() + if source_value not in {"operator", "system"}: + message = "`--source` must be either `operator` or `system`." + if state.json_output: + _print_json({"ok": False, "detail": message}) + else: + _show_error(state.console, message) + raise typer.Exit(code=2) + params["source"] = source_value + if timeline_type: + filters = _normalize_timeline_type_filters(timeline_type) + if filters: + params["types"] = ",".join(filters) + if search: + params["q"] = search + if include_deleted: + params["include_deleted"] = "1" + + response = state.api.get(f"/api/developer/devices/{resolved_hash}/timeline/", params=params) + if not response.ok: + message = _friendly_response_message("Unable to load device history.", response.data, response.text) + if state.json_output: + _print_json({"ok": False, "detail": message}) + else: + _show_error(state.console, message) + raise typer.Exit(code=1) + + payload = response.data or {} + if state.json_output: + _print_json(payload) + return + + rows = _extract_timeline_rows(payload) + table = Table(title=f"Device history: {resolved_hash}") + table.add_column("ID", style="cyan") + table.add_column("Timestamp") + table.add_column("Source") + table.add_column("Type") + table.add_column("Author") + table.add_column("Message", overflow="fold") + + for row in rows: + event_id = _timeline_event_id(row) + timestamp = str(row.get("timestamp") or "") + entry_source = str(row.get("source") or "") + entry_type = str(row.get("type") or "") + author = str(row.get("author_display") or row.get("author_username") or "") + message = str(row.get("message") or "") + deleted = bool(row.get("is_deleted")) + table.add_row( + str(event_id or ""), + timestamp, + entry_source, + entry_type, + author, + message, + style="dim" if deleted else "", + ) + + state.console.print(table) + state.console.print(f"Showing {len(rows)} history entries.") + + +@devices_history_app.command("show") +def devices_history_show( + ctx: typer.Context, + device_hash: Optional[str] = typer.Argument(None), + event_id: int = typer.Option(..., "--event-id", "-e", help="Timeline event ID"), + include_deleted: bool = typer.Option(True, "--include-deleted/--exclude-deleted", help="Include deleted comments"), +) -> None: + """Show one history entry for a device. + + Examples: + dataplicity devices history show --event-id 12345 + dataplicity devices history show --event-id 12345 + """ + state = _ctx(ctx) + _require_auth(state) + resolved_hash = _resolve_device_hash_interactive(state, device_hash, action_name="history show") + params: Dict[str, Any] = {"limit": 500} + if include_deleted: + params["include_deleted"] = "1" + response = state.api.get(f"/api/developer/devices/{resolved_hash}/timeline/", params=params) + if not response.ok: + message = _friendly_response_message("Unable to load device history.", response.data, response.text) + if state.json_output: + _print_json({"ok": False, "detail": message}) + else: + _show_error(state.console, message) + raise typer.Exit(code=1) + + rows = _extract_timeline_rows(response.data or {}) + row = next((item for item in rows if _timeline_event_id(item) == event_id), None) + if not row: + message = _timeline_entry_not_found_message(event_id) + if state.json_output: + _print_json({"ok": False, "detail": message, "event_id": event_id}) + else: + _show_error(state.console, message) + raise typer.Exit(code=2) + + if state.json_output: + _print_json(row) + return + + table = Table(title=f"Device history entry {event_id}") + table.add_column("Key") + table.add_column("Value") + preferred_keys = [ + "id", + "timestamp", + "source", + "type", + "event_type", + "author_display", + "author_username", + "is_deleted", + "message", + ] + seen: Set[str] = set() + for key in preferred_keys: + if key not in row: + continue + table.add_row(key, json.dumps(_sanitize_payload(row.get(key)))) + seen.add(key) + for key in sorted(row.keys()): + if key in seen: + continue + table.add_row(str(key), json.dumps(_sanitize_payload(row.get(key)))) + state.console.print(table) + + +@devices_history_app.command("comment") +def devices_history_comment( + ctx: typer.Context, + device_hash: Optional[str] = typer.Argument(None), + message: Optional[str] = typer.Option(None, "--message", "-m", help="Comment text"), + timeline_type: str = typer.Option("general", "--type", help="Comment type/category"), +) -> None: + """Add a comment to device history. + + Examples: + dataplicity devices history comment --message "Replaced PSU" --type service + dataplicity devices history comment --message "Moved to rack B" + """ + state = _ctx(ctx) + _require_auth(state) + resolved_hash = _resolve_device_hash_interactive(state, device_hash, action_name="history comment") + comment = (message or "").strip() + if not comment: + if state.json_output: + _print_json({"ok": False, "detail": "`--message` is required in --json mode."}) + raise typer.Exit(code=2) + comment = Prompt.ask("Comment") + comment = comment.strip() + if not comment: + _show_error(state.console, "Comment cannot be empty.") + raise typer.Exit(code=2) + + payload = {"message": comment} + normalized_type = _normalize_timeline_type(timeline_type) + if normalized_type in DEVICE_TIMELINE_OPERATOR_RESERVED_TYPES: + allowed_types = ", ".join(DEVICE_TIMELINE_OPERATOR_ALLOWED_TYPES) + message = ( + f"`--type {normalized_type}` is reserved for automatic system entries. " + f"Allowed manual comment types: {allowed_types}." + ) + if state.json_output: + _print_json({"ok": False, "detail": message}) + else: + _show_error(state.console, message) + raise typer.Exit(code=2) + payload["type"] = normalized_type + response = state.api.post(f"/api/developer/devices/{resolved_hash}/timeline/", json_data=payload) + if not response.ok: + message = _friendly_response_message("Unable to add device history comment.", response.data, response.text) + if state.json_output: + _print_json({"ok": False, "detail": message}) + else: + _show_error(state.console, message) + raise typer.Exit(code=1) + + row = response.data if isinstance(response.data, dict) else {"detail": "Comment created."} + if state.json_output: + _print_json(row) + return + + table = Table(title=f"Comment added to {resolved_hash}") + table.add_column("Key") + table.add_column("Value") + for key in ["id", "timestamp", "type", "author_display", "author_username", "message"]: + if key in row: + table.add_row(key, json.dumps(_sanitize_payload(row.get(key)))) + state.console.print(table) + + +@devices_history_app.command("delete") +def devices_history_delete( + ctx: typer.Context, + device_hash: Optional[str] = typer.Argument(None), + event_id: int = typer.Option(..., "--event-id", "-e", help="Timeline event ID"), +) -> None: + """Mark a history comment as deleted (soft delete). + + Examples: + dataplicity devices history delete --event-id 12345 + dataplicity devices history delete --event-id 12345 + """ + state = _ctx(ctx) + _require_auth(state) + resolved_hash = _resolve_device_hash_interactive(state, device_hash, action_name="history delete") + response = state.api.request( + "DELETE", + f"/api/developer/devices/{resolved_hash}/timeline/{event_id}/", + ) + if not response.ok: + message = _friendly_response_message("Unable to mark history entry as deleted.", response.data, response.text) + if state.json_output: + _print_json({"ok": False, "detail": message, "event_id": event_id}) + else: + _show_error(state.console, message) + raise typer.Exit(code=1) + + row = response.data if isinstance(response.data, dict) else {"id": event_id, "is_deleted": True} + if state.json_output: + _print_json(row) + return + + table = Table(title=f"History entry {event_id} marked deleted") + table.add_column("Key") + table.add_column("Value") + for key in ["id", "timestamp", "type", "author_display", "author_username", "is_deleted", "message"]: + if key in row: + table.add_row(key, json.dumps(_sanitize_payload(row.get(key)))) + state.console.print(table) + state.console.print("[dim]Deleted entries are retained in history and can be shown with --include-deleted.[/dim]") + + +@devices_history_app.command("mark-deleted") +def devices_history_mark_deleted( + ctx: typer.Context, + device_hash: Optional[str] = typer.Argument(None), + event_id: int = typer.Option(..., "--event-id", "-e", help="Timeline event ID"), +) -> None: + """Alias for `devices history delete`.""" + devices_history_delete(ctx, device_hash=device_hash, event_id=event_id) + + +@devices_history_app.command("add-comment") +def devices_history_add_comment( + ctx: typer.Context, + device_hash: Optional[str] = typer.Argument(None), + message: Optional[str] = typer.Option(None, "--message", "-m", help="Comment text"), + timeline_type: str = typer.Option("general", "--type", help="Comment type/category"), +) -> None: + """Alias for `devices history comment`.""" + devices_history_comment(ctx, device_hash=device_hash, message=message, timeline_type=timeline_type) + + @devices_app.command("wormhole-status") def devices_wormhole_status(ctx: typer.Context, device_hash: Optional[str] = typer.Argument(None)) -> None: """Show wormhole status and recent usage for a device. @@ -2900,11 +3295,18 @@ def _list_resource( endpoint: str, title: str, params: Optional[Dict[str, Any]] = None, + feature_name: Optional[str] = None, ) -> None: _require_auth(state) response = state.api.get(endpoint, params=params) if not response.ok: - message = _friendly_response_message(f"Unable to list {title.lower()}.", response.data, response.text) + message = _endpoint_error_message( + state, + endpoint=endpoint, + response=response, + default_message=f"Unable to list {title.lower()}.", + feature_name=feature_name, + ) if state.json_output: _print_json({"ok": False, "detail": message}) else: @@ -2922,11 +3324,19 @@ def _show_resource( endpoint: str, resource_id: str, not_found_label: str, + feature_name: Optional[str] = None, ) -> None: _require_auth(state) - response = state.api.get(f"{endpoint}{resource_id}/") + request_endpoint = f"{endpoint}{resource_id}/" + response = state.api.get(request_endpoint) if not response.ok: - message = _friendly_response_message(f"Unable to fetch {not_found_label}.", response.data, response.text) + message = _endpoint_error_message( + state, + endpoint=request_endpoint, + response=response, + default_message=f"Unable to fetch {not_found_label}.", + feature_name=feature_name, + ) if state.json_output: _print_json({"ok": False, "detail": message}) else: @@ -2944,12 +3354,25 @@ def _show_resource( state.console.print(table) -def _create_resource(state: AppContext, *, endpoint: str, data: str, label: str) -> None: +def _create_resource( + state: AppContext, + *, + endpoint: str, + data: str, + label: str, + feature_name: Optional[str] = None, +) -> None: _require_auth(state) payload = _parse_json_payload_or_exit(state, data) response = state.api.post(endpoint, json_data=payload) if not response.ok: - message = _friendly_response_message(f"Unable to create {label}.", response.data, response.text) + message = _endpoint_error_message( + state, + endpoint=endpoint, + response=response, + default_message=f"Unable to create {label}.", + feature_name=feature_name, + ) if state.json_output: _print_json({"ok": False, "detail": message}) else: @@ -2961,11 +3384,25 @@ def _create_resource(state: AppContext, *, endpoint: str, data: str, label: str) state.console.print(f"[green]{label.title()} created.[/green]") -def _delete_resource(state: AppContext, *, endpoint: str, resource_id: str, label: str) -> None: +def _delete_resource( + state: AppContext, + *, + endpoint: str, + resource_id: str, + label: str, + feature_name: Optional[str] = None, +) -> None: _require_auth(state) - response = state.api.request("DELETE", f"{endpoint}{resource_id}/") + request_endpoint = f"{endpoint}{resource_id}/" + response = state.api.request("DELETE", request_endpoint) if not response.ok: - message = _friendly_response_message(f"Unable to delete {label}.", response.data, response.text) + message = _endpoint_error_message( + state, + endpoint=request_endpoint, + response=response, + default_message=f"Unable to delete {label}.", + feature_name=feature_name, + ) if state.json_output: _print_json({"ok": False, "detail": message}) else: @@ -2994,7 +3431,13 @@ def endpoint_monitors_list( endpoint = _incident_automation_endpoint(state, "signals") response = state.api.get(endpoint) if not response.ok: - message = _friendly_response_message("Unable to list endpoint monitors.", response.data, response.text) + message = _endpoint_error_message( + state, + endpoint=endpoint, + response=response, + default_message="Unable to list endpoint monitors.", + feature_name="Endpoint monitors", + ) if state.json_output: _print_json({"ok": False, "detail": message}) else: @@ -3033,7 +3476,13 @@ def endpoint_monitors_show(ctx: typer.Context, monitor_id: str = typer.Argument( endpoint = _incident_automation_endpoint(state, "signals") response = state.api.get(endpoint) if not response.ok: - message = _friendly_response_message("Unable to fetch endpoint monitor.", response.data, response.text) + message = _endpoint_error_message( + state, + endpoint=endpoint, + response=response, + default_message="Unable to fetch endpoint monitor.", + feature_name="Endpoint monitors", + ) if state.json_output: _print_json({"ok": False, "detail": message}) else: @@ -3066,7 +3515,13 @@ def endpoint_monitors_create( endpoint = _incident_automation_endpoint(state, "rules") response = state.api.post(endpoint, json_data=payload) if not response.ok: - message = _friendly_response_message("Unable to create endpoint monitor.", response.data, response.text) + message = _endpoint_error_message( + state, + endpoint=endpoint, + response=response, + default_message="Unable to create endpoint monitor.", + feature_name="Endpoint monitors", + ) if state.json_output: _print_json({"ok": False, "detail": message}) else: @@ -3091,6 +3546,7 @@ def endpoint_monitors_delete(ctx: typer.Context, monitor_id: str = typer.Argumen endpoint=_incident_automation_endpoint(state, "rules"), resource_id=monitor_id, label="endpoint monitor", + feature_name="Endpoint monitors", ) @@ -3111,7 +3567,13 @@ def user_impact_list( endpoint = _incident_automation_endpoint(state, "signals") response = state.api.get(endpoint) if not response.ok: - message = _friendly_response_message("Unable to list user impact.", response.data, response.text) + message = _endpoint_error_message( + state, + endpoint=endpoint, + response=response, + default_message="Unable to list user impact.", + feature_name="User impact signals", + ) if state.json_output: _print_json({"ok": False, "detail": message}) else: @@ -3149,7 +3611,13 @@ def user_impact_show(ctx: typer.Context, impact_id: str = typer.Argument(...)) - endpoint = _incident_automation_endpoint(state, "signals") response = state.api.get(endpoint) if not response.ok: - message = _friendly_response_message("Unable to fetch user impact item.", response.data, response.text) + message = _endpoint_error_message( + state, + endpoint=endpoint, + response=response, + default_message="Unable to fetch user impact item.", + feature_name="User impact signals", + ) if state.json_output: _print_json({"ok": False, "detail": message}) else: @@ -3182,7 +3650,13 @@ def heartbeat_monitors_list( endpoint = _incident_automation_endpoint(state, "signals") response = state.api.get(endpoint) if not response.ok: - message = _friendly_response_message("Unable to list heartbeat monitors.", response.data, response.text) + message = _endpoint_error_message( + state, + endpoint=endpoint, + response=response, + default_message="Unable to list heartbeat monitors.", + feature_name="Heartbeat monitors", + ) if state.json_output: _print_json({"ok": False, "detail": message}) else: @@ -3221,7 +3695,13 @@ def heartbeat_monitors_show(ctx: typer.Context, monitor_id: str = typer.Argument endpoint = _incident_automation_endpoint(state, "signals") response = state.api.get(endpoint) if not response.ok: - message = _friendly_response_message("Unable to fetch heartbeat monitor.", response.data, response.text) + message = _endpoint_error_message( + state, + endpoint=endpoint, + response=response, + default_message="Unable to fetch heartbeat monitor.", + feature_name="Heartbeat monitors", + ) if state.json_output: _print_json({"ok": False, "detail": message}) else: @@ -3254,7 +3734,13 @@ def heartbeat_monitors_create( endpoint = _incident_automation_endpoint(state, "rules") response = state.api.post(endpoint, json_data=payload) if not response.ok: - message = _friendly_response_message("Unable to create heartbeat monitor.", response.data, response.text) + message = _endpoint_error_message( + state, + endpoint=endpoint, + response=response, + default_message="Unable to create heartbeat monitor.", + feature_name="Heartbeat monitors", + ) if state.json_output: _print_json({"ok": False, "detail": message}) else: @@ -3279,6 +3765,7 @@ def heartbeat_monitors_delete(ctx: typer.Context, monitor_id: str = typer.Argume endpoint=_incident_automation_endpoint(state, "rules"), resource_id=monitor_id, label="heartbeat monitor", + feature_name="Heartbeat monitors", ) @@ -3298,7 +3785,13 @@ def fleet_jobs_list( params: Dict[str, Any] = {"page_size": page_size} if status: params["status"] = status - _list_resource(state, endpoint=_incident_automation_endpoint(state, "rules"), title="Fleet jobs", params=params) + _list_resource( + state, + endpoint=_incident_automation_endpoint(state, "rules"), + title="Fleet jobs", + params=params, + feature_name="Fleet jobs", + ) @fleet_jobs_app.command("ls") @@ -3324,6 +3817,7 @@ def fleet_jobs_show(ctx: typer.Context, job_id: str = typer.Argument(...)) -> No endpoint=_incident_automation_endpoint(state, "rules"), resource_id=job_id, not_found_label="fleet job", + feature_name="Fleet jobs", ) @@ -3345,7 +3839,13 @@ def fleet_jobs_run( """ state = _ctx(ctx) endpoint = _incident_automation_endpoint(state, "rule-simulations") - _create_resource(state, endpoint=endpoint, data=data, label="fleet job simulation") + _create_resource( + state, + endpoint=endpoint, + data=data, + label="fleet job simulation", + feature_name="Fleet jobs", + ) @fleet_jobs_app.command("create") @@ -3366,13 +3866,21 @@ def fleet_jobs_cancel(ctx: typer.Context, job_id: str = typer.Argument(...)) -> """ state = _ctx(ctx) _require_auth(state) + rules_endpoint = _incident_automation_endpoint(state, "rules") + job_endpoint = f"{rules_endpoint}{job_id}/" response = state.api.request( "PATCH", - f"{_incident_automation_endpoint(state, 'rules')}{job_id}/", + job_endpoint, json_data={"enabled": False}, ) if not response.ok: - message = _friendly_response_message("Unable to cancel fleet job.", response.data, response.text) + message = _endpoint_error_message( + state, + endpoint=job_endpoint, + response=response, + default_message="Unable to cancel fleet job.", + feature_name="Fleet jobs", + ) if state.json_output: _print_json({"ok": False, "detail": message}) else: diff --git a/pyproject.toml b/pyproject.toml index c2db352..4563d67 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -20,6 +20,12 @@ dependencies = [ "platformdirs>=4.0.0", ] +[project.optional-dependencies] +test = [ + "pytest>=9.0.0", + "pytest-cov>=7.0.0", +] + [project.scripts] dataplicity = "dataplicity_cli.cli:app" @@ -29,3 +35,6 @@ package-dir = { "" = "." } [tool.setuptools.packages.find] where = ["."] include = ["dataplicity_cli*"] + +[tool.pytest.ini_options] +testpaths = ["tests"] diff --git a/tests/test_cli_core_helpers.py b/tests/test_cli_core_helpers.py new file mode 100644 index 0000000..77fbcc9 --- /dev/null +++ b/tests/test_cli_core_helpers.py @@ -0,0 +1,120 @@ +from __future__ import annotations + +import datetime as dt +import unittest + +from dataplicity_cli.cli import ( + LOGGING_MAX_OUTPUT_ITEMS, + _extract_log_device, + _extract_objects, + _extract_port_forwarding_ports, + _extract_port_numbers, + _find_allowed_ports, + _format_log_ts, + _friendly_response_message, + _org_logs_params_from_developer_params, + _parse_kv_pairs, + _parse_log_time_expr, + _resource_key, + _resource_name, + _resource_status, + _sanitize_payload, + _truncate_logging_payload, +) + + +class CliCoreHelpersTest(unittest.TestCase): + def test_sanitize_payload_recursively_strips_sensitive_fields(self) -> None: + payload = { + "safe": "ok", + "token": "secret", + "nested": {"api_key": "hide", "keep": 1}, + "list": [{"refresh": "hide"}, {"value": 2}], + } + sanitized = _sanitize_payload(payload) + self.assertEqual(sanitized, {"safe": "ok", "nested": {"keep": 1}, "list": [{}, {"value": 2}]}) + + def test_friendly_response_message_uses_detail_and_invalid_auth_override(self) -> None: + message = _friendly_response_message("fallback", {"detail": "Token has expired"}, "") + self.assertIn("Saved login appears expired", message) + + message = _friendly_response_message("fallback", {"non_field_errors": ["bad request"]}, "") + self.assertEqual(message, "bad request") + + def test_extract_port_numbers_supports_int_ranges_and_dict_ranges(self) -> None: + self.assertEqual(_extract_port_numbers(443), {443}) + self.assertEqual(_extract_port_numbers("22, 80 443"), {22, 80, 443}) + self.assertEqual(_extract_port_numbers("1000-1002"), {1000, 1001, 1002}) + self.assertEqual(_extract_port_numbers({"start": 9000, "end": 9002}), {9000, 9001, 9002}) + self.assertEqual(_extract_port_numbers("70000"), set()) + + def test_extract_port_forwarding_ports_handles_none_and_mixed_list(self) -> None: + self.assertIsNone(_extract_port_forwarding_ports({"other": "value"})) + ports = _extract_port_forwarding_ports({"port_forwarding_ports": [22, "80-81", {"start": 9000, "end": 9001}]}) + self.assertEqual(ports, [22, 80, 81, 9000, 9001]) + + def test_find_allowed_ports_walks_nested_payload(self) -> None: + payload = { + "metadata": { + "port_whitelist": ["22", "80-81"], + "nested": [{"supported_ports": [443]}], + } + } + self.assertEqual(_find_allowed_ports(payload), {22, 80, 81, 443}) + + def test_org_logs_params_mapping_prefers_explicit_search(self) -> None: + params = { + "page_size": 55, + "device": "abc", + "path": "/devices/abc", + "search": "timeout", + "level": "error", + "since": "2026-06-01T00:00:00Z", + "until": "2026-06-01T01:00:00Z", + } + mapped = _org_logs_params_from_developer_params(params) + self.assertEqual( + mapped, + { + "limit": 55, + "device_hash": "abc", + "search": "timeout", + "level": "error", + "after": "2026-06-01T00:00:00Z", + "before": "2026-06-01T01:00:00Z", + }, + ) + + def test_parse_log_time_expr_handles_relative_and_iso_inputs(self) -> None: + now = dt.datetime(2026, 6, 29, 0, 0, tzinfo=dt.timezone.utc) + self.assertEqual(_parse_log_time_expr("15m", now=now), "2026-06-28T23:45:00Z") + self.assertEqual(_parse_log_time_expr("2026-06-01T00:00:00", now=now), "2026-06-01T00:00:00Z") + with self.assertRaises(ValueError): + _parse_log_time_expr("not-a-time", now=now) + + def test_truncate_logging_payload_for_large_result_sets(self) -> None: + payload = {"results": [{"message": "x" * 3000}] * (LOGGING_MAX_OUTPUT_ITEMS + 5)} + safe_payload, truncated, dropped = _truncate_logging_payload(payload) + self.assertTrue(truncated) + self.assertEqual(dropped, 5) + self.assertIn("__cli_truncated__", safe_payload) + first_message = safe_payload["results"][0]["message"] + self.assertIn("[truncated", first_message) + + def test_format_log_timestamp_and_extract_device(self) -> None: + self.assertEqual(_format_log_ts(0), "1970-01-01T00:00:00Z") + self.assertEqual(_format_log_ts("bad"), "") + self.assertEqual(_extract_log_device({"device_hash": "hash-a"}), "hash-a") + self.assertEqual(_extract_log_device({"message": "connection device_hash=hash-b"}), "hash-b") + + def test_parse_and_extract_resource_helpers(self) -> None: + self.assertEqual(_parse_kv_pairs(["a=1", "bad", "b=2"]), {"a": "1", "b": "2"}) + items = _extract_objects({"results": [{"id": 1}, "skip", {"id": 2}]}) + self.assertEqual(items, [{"id": 1}, {"id": 2}]) + self.assertEqual(_resource_key({"uuid": "abc"}), "abc") + self.assertEqual(_resource_name({"title": "My Monitor"}), "My Monitor") + self.assertEqual(_resource_status({"metadata": {"health": "healthy"}}), "healthy") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_config.py b/tests/test_config.py new file mode 100644 index 0000000..7afe810 --- /dev/null +++ b/tests/test_config.py @@ -0,0 +1,77 @@ +from __future__ import annotations + +import json +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +from dataplicity_cli.config import ( + DEFAULT_BASE_URL, + LEGACY_DEFAULT_BASE_URLS, + Config, + default_config_path, +) + + +class ConfigTest(unittest.TestCase): + def test_load_missing_file_returns_defaults(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + cfg = Config.load(Path(temp_dir) / "missing.json") + self.assertEqual(cfg.base_url, DEFAULT_BASE_URL) + self.assertIsNone(cfg.auth_method) + self.assertIsNone(cfg.access_token) + + def test_load_invalid_json_returns_defaults(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + path = Path(temp_dir) / "cli.json" + path.write_text("{not-valid-json", encoding="utf-8") + cfg = Config.load(path) + self.assertEqual(cfg.base_url, DEFAULT_BASE_URL) + self.assertIsNone(cfg.api_key) + + def test_load_upgrades_legacy_default_base_url(self) -> None: + legacy_url = next(iter(LEGACY_DEFAULT_BASE_URLS)) + with tempfile.TemporaryDirectory() as temp_dir: + path = Path(temp_dir) / "cli.json" + path.write_text(json.dumps({"base_url": legacy_url}), encoding="utf-8") + cfg = Config.load(path) + self.assertEqual(cfg.base_url, DEFAULT_BASE_URL) + + def test_save_and_load_round_trip(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + path = Path(temp_dir) / "nested" / "cli.json" + cfg = Config( + base_url="https://example.test", + auth_method="jwt", + access_token="a", + refresh_token="r", + api_key="k", + last_email="test@example.com", + preferred_login_method="sso", + ) + cfg.save(path) + loaded = Config.load(path) + self.assertEqual(loaded.to_dict(), cfg.to_dict()) + + def test_clear_tokens_resets_jwt_mode_only(self) -> None: + cfg = Config(auth_method="jwt", access_token="a", refresh_token="r") + cfg.clear_tokens() + self.assertIsNone(cfg.auth_method) + self.assertIsNone(cfg.access_token) + self.assertIsNone(cfg.refresh_token) + + def test_clear_api_key_resets_api_key_mode_only(self) -> None: + cfg = Config(auth_method="api_key", api_key="secret") + cfg.clear_api_key() + self.assertIsNone(cfg.auth_method) + self.assertIsNone(cfg.api_key) + + def test_default_config_path_uses_env_override(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + with patch.dict("os.environ", {"DATAPLICITY_CONFIG_DIR": temp_dir}, clear=False): + self.assertEqual(default_config_path(), Path(temp_dir) / "cli.json") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_devices_history_cli.py b/tests/test_devices_history_cli.py new file mode 100644 index 0000000..b1c8c8b --- /dev/null +++ b/tests/test_devices_history_cli.py @@ -0,0 +1,228 @@ +from __future__ import annotations + +import json +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +from typer.testing import CliRunner + +from dataplicity_cli.api import ApiResponse +from dataplicity_cli.cli import app + + +class DevicesHistoryCliTest(unittest.TestCase): + def setUp(self) -> None: + self.runner = CliRunner() + self.device_hash = "0bbaf1d7919ef6ffb9734b237d0a6477b8a4fe54e1c6705fc032e094a81d21f6" + + @staticmethod + def _write_authed_config(path: Path) -> None: + path.write_text( + json.dumps( + { + "base_url": "https://gateway.dataplicity.com", + "auth_method": "api_key", + "api_key": "test-key", + } + ), + encoding="utf-8", + ) + + def test_history_list_returns_timeline_rows_in_json(self) -> None: + timeline_payload = { + "results": [ + { + "id": 321, + "timestamp": "2026-06-29T02:00:00Z", + "source": "operator", + "type": "service", + "author_display": "ops@example.com", + "message": "Power-cycled after maintenance window.", + "is_deleted": False, + } + ], + "count": 1, + } + + def fake_get(_self, path: str, *, params=None): # type: ignore[no-untyped-def] + self.assertEqual(path, f"/api/developer/devices/{self.device_hash}/timeline/") + self.assertEqual((params or {}).get("limit"), 120) + return ApiResponse(True, 200, timeline_payload, json.dumps(timeline_payload)) + + with tempfile.TemporaryDirectory() as temp_dir: + config_path = Path(temp_dir) / "cli.json" + self._write_authed_config(config_path) + with patch("dataplicity_cli.cli.ApiClient.get", new=fake_get): + result = self.runner.invoke( + app, + ["--json", "--config", str(config_path), "devices", "history", "list", self.device_hash], + ) + + self.assertEqual(result.exit_code, 0, msg=result.output) + payload = json.loads(result.output) + self.assertEqual(payload.get("count"), 1) + self.assertEqual((payload.get("results") or [{}])[0].get("id"), 321) + + def test_history_show_returns_selected_row(self) -> None: + timeline_payload = { + "results": [ + {"id": 100, "message": "Older event"}, + {"id": 555, "message": "Target event", "type": "logistics"}, + ], + "count": 2, + } + + def fake_get(_self, path: str, *, params=None): # type: ignore[no-untyped-def] + self.assertEqual(path, f"/api/developer/devices/{self.device_hash}/timeline/") + self.assertEqual((params or {}).get("limit"), 500) + return ApiResponse(True, 200, timeline_payload, json.dumps(timeline_payload)) + + with tempfile.TemporaryDirectory() as temp_dir: + config_path = Path(temp_dir) / "cli.json" + self._write_authed_config(config_path) + with patch("dataplicity_cli.cli.ApiClient.get", new=fake_get): + result = self.runner.invoke( + app, + [ + "--json", + "--config", + str(config_path), + "devices", + "history", + "show", + self.device_hash, + "--event-id", + "555", + ], + ) + + self.assertEqual(result.exit_code, 0, msg=result.output) + payload = json.loads(result.output) + self.assertEqual(payload.get("id"), 555) + self.assertEqual(payload.get("message"), "Target event") + + def test_history_comment_posts_message_and_type(self) -> None: + created_payload = { + "id": 991, + "timestamp": "2026-06-29T03:15:00Z", + "source": "operator", + "type": "service", + "author_username": "timeline-admin@example.com", + "message": "Replaced sensor board and re-seated cable.", + } + + def fake_post(_self, path: str, *, json_data=None, data=None): # type: ignore[no-untyped-def] + self.assertEqual(path, f"/api/developer/devices/{self.device_hash}/timeline/") + self.assertIsNone(data) + self.assertEqual((json_data or {}).get("message"), "Replaced sensor board and re-seated cable.") + self.assertEqual((json_data or {}).get("type"), "service") + return ApiResponse(True, 201, created_payload, json.dumps(created_payload)) + + with tempfile.TemporaryDirectory() as temp_dir: + config_path = Path(temp_dir) / "cli.json" + self._write_authed_config(config_path) + with patch("dataplicity_cli.cli.ApiClient.post", new=fake_post): + result = self.runner.invoke( + app, + [ + "--json", + "--config", + str(config_path), + "devices", + "history", + "comment", + self.device_hash, + "--message", + "Replaced sensor board and re-seated cable.", + "--type", + "service", + ], + ) + + self.assertEqual(result.exit_code, 0, msg=result.output) + payload = json.loads(result.output) + self.assertEqual(payload.get("id"), 991) + self.assertEqual(payload.get("type"), "service") + + def test_history_comment_rejects_system_reserved_type(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + config_path = Path(temp_dir) / "cli.json" + self._write_authed_config(config_path) + result = self.runner.invoke( + app, + [ + "--json", + "--config", + str(config_path), + "devices", + "history", + "comment", + self.device_hash, + "--message", + "Trying reserved type", + "--type", + "lifecycle", + ], + ) + + self.assertEqual(result.exit_code, 2, msg=result.output) + payload = json.loads(result.output) + self.assertEqual(payload.get("ok"), False) + self.assertIn("reserved for automatic system entries", str(payload.get("detail"))) + + def test_history_delete_soft_deletes_entry(self) -> None: + deleted_payload = { + "id": 777, + "timestamp": "2026-06-29T04:40:00Z", + "source": "operator", + "type": "service", + "is_deleted": True, + "message": "[deleted]", + } + + def fake_request( + _self, + method: str, + path: str, + *, + params=None, + json_data=None, + data=None, + headers=None, + timeout=20, + allow_refresh=True, + ): # type: ignore[no-untyped-def] + _ = (params, json_data, data, headers, timeout, allow_refresh) + self.assertEqual(method, "DELETE") + self.assertEqual(path, f"/api/developer/devices/{self.device_hash}/timeline/777/") + return ApiResponse(True, 200, deleted_payload, json.dumps(deleted_payload)) + + with tempfile.TemporaryDirectory() as temp_dir: + config_path = Path(temp_dir) / "cli.json" + self._write_authed_config(config_path) + with patch("dataplicity_cli.cli.ApiClient.request", new=fake_request): + result = self.runner.invoke( + app, + [ + "--json", + "--config", + str(config_path), + "devices", + "history", + "delete", + self.device_hash, + "--event-id", + "777", + ], + ) + + self.assertEqual(result.exit_code, 0, msg=result.output) + payload = json.loads(result.output) + self.assertEqual(payload.get("id"), 777) + self.assertEqual(payload.get("is_deleted"), True) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_entrypoint.py b/tests/test_entrypoint.py new file mode 100644 index 0000000..5ed1848 --- /dev/null +++ b/tests/test_entrypoint.py @@ -0,0 +1,16 @@ +from __future__ import annotations + +import runpy +import unittest +from unittest.mock import patch + + +class EntrypointTest(unittest.TestCase): + def test_main_module_invokes_cli_app(self) -> None: + with patch("dataplicity_cli.cli.app") as mock_app: + runpy.run_module("dataplicity_cli.__main__", run_name="__main__") + mock_app.assert_called_once_with() + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_feature_gated_commands.py b/tests/test_feature_gated_commands.py new file mode 100644 index 0000000..66fe5a7 --- /dev/null +++ b/tests/test_feature_gated_commands.py @@ -0,0 +1,132 @@ +from __future__ import annotations + +import json +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +from typer.testing import CliRunner + +from dataplicity_cli.api import ApiResponse +from dataplicity_cli.cli import app + + +class FeatureGatedCommandsTest(unittest.TestCase): + def setUp(self) -> None: + self.runner = CliRunner() + self.org_hash = "org-feature-off" + + @staticmethod + def _write_authed_config(path: Path) -> None: + path.write_text( + json.dumps( + { + "base_url": "https://gateway.dataplicity.com", + "auth_method": "api_key", + "api_key": "test-key", + } + ), + encoding="utf-8", + ) + + def _devices_payload(self) -> list[dict[str, str]]: + return [{"hash_id": "dev-1", "organisation_hash": self.org_hash, "name": "device-1"}] + + def test_fleet_jobs_list_reports_feature_unavailable_for_org(self) -> None: + rules_endpoint = f"/api/organisations/{self.org_hash}/incident-automation/rules/" + + def fake_get(_self, path: str, *, params=None): # type: ignore[no-untyped-def] + _ = params + if path == "/api/developer/devices/": + return ApiResponse(True, 200, self._devices_payload(), "") + if path == rules_endpoint: + return ApiResponse(False, 404, {"detail": "Not found."}, '{"detail":"Not found."}') + raise AssertionError(f"unexpected GET path: {path}") + + with tempfile.TemporaryDirectory() as temp_dir: + config_path = Path(temp_dir) / "cli.json" + self._write_authed_config(config_path) + with patch("dataplicity_cli.cli.ApiClient.get", new=fake_get): + result = self.runner.invoke( + app, + ["--json", "--config", str(config_path), "fleet-jobs", "list"], + ) + + self.assertEqual(result.exit_code, 1, msg=result.output) + payload = json.loads(result.output) + self.assertEqual( + payload.get("detail"), + f"Fleet jobs are not available for organisation `{self.org_hash}`.", + ) + + def test_fleet_jobs_show_keeps_not_found_when_feature_exists(self) -> None: + rules_endpoint = f"/api/organisations/{self.org_hash}/incident-automation/rules/" + job_endpoint = f"{rules_endpoint}job-123/" + + def fake_get(_self, path: str, *, params=None): # type: ignore[no-untyped-def] + _ = params + if path == "/api/developer/devices/": + return ApiResponse(True, 200, self._devices_payload(), "") + if path == job_endpoint: + return ApiResponse(False, 404, {"detail": "Not found."}, '{"detail":"Not found."}') + if path == rules_endpoint: + return ApiResponse(True, 200, [], "[]") + raise AssertionError(f"unexpected GET path: {path}") + + with tempfile.TemporaryDirectory() as temp_dir: + config_path = Path(temp_dir) / "cli.json" + self._write_authed_config(config_path) + with patch("dataplicity_cli.cli.ApiClient.get", new=fake_get): + result = self.runner.invoke( + app, + ["--json", "--config", str(config_path), "fleet-jobs", "show", "job-123"], + ) + + self.assertEqual(result.exit_code, 1, msg=result.output) + payload = json.loads(result.output) + self.assertEqual(payload.get("detail"), "Not found.") + + def test_fleet_jobs_run_reports_feature_unavailable_for_org(self) -> None: + simulation_endpoint = f"/api/organisations/{self.org_hash}/incident-automation/rule-simulations/" + + def fake_get(_self, path: str, *, params=None): # type: ignore[no-untyped-def] + _ = params + if path == "/api/developer/devices/": + return ApiResponse(True, 200, self._devices_payload(), "") + raise AssertionError(f"unexpected GET path: {path}") + + def fake_post(_self, path: str, *, json_data=None, data=None): # type: ignore[no-untyped-def] + _ = (json_data, data) + if path == simulation_endpoint: + return ApiResponse(False, 404, {"detail": "Not found."}, '{"detail":"Not found."}') + raise AssertionError(f"unexpected POST path: {path}") + + with tempfile.TemporaryDirectory() as temp_dir: + config_path = Path(temp_dir) / "cli.json" + self._write_authed_config(config_path) + with patch("dataplicity_cli.cli.ApiClient.get", new=fake_get): + with patch("dataplicity_cli.cli.ApiClient.post", new=fake_post): + result = self.runner.invoke( + app, + [ + "--json", + "--config", + str(config_path), + "fleet-jobs", + "run", + "--data", + '{"name":"restart-edge","device_hashes":["abc123"],"command":"restart"}', + ], + ) + + self.assertEqual(result.exit_code, 1, msg=result.output) + payload = json.loads(result.output) + self.assertEqual( + payload.get("detail"), + f"Fleet jobs are not available for organisation `{self.org_hash}`.", + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_m2m.py b/tests/test_m2m.py new file mode 100644 index 0000000..b2db375 --- /dev/null +++ b/tests/test_m2m.py @@ -0,0 +1,82 @@ +from __future__ import annotations + +import asyncio +import unittest +from unittest.mock import AsyncMock + +from dataplicity_cli.m2m import BencodeError, M2MClient, bencode_decode, bencode_encode + + +class BencodeTest(unittest.TestCase): + def test_bencode_round_trip_for_nested_values(self) -> None: + payload = [1, b"abc", "hello", [2, b"x"], {b"k": b"v"}] + encoded = bencode_encode(payload) + decoded = bencode_decode(encoded) + self.assertEqual(decoded, [1, b"abc", b"hello", [2, b"x"], {b"k": b"v"}]) + + def test_bencode_encode_requires_bytes_dict_keys(self) -> None: + with self.assertRaises(BencodeError): + bencode_encode({"not-bytes": "value"}) + + def test_bencode_decode_requires_bytes(self) -> None: + with self.assertRaises(BencodeError): + bencode_decode("not-bytes") # type: ignore[arg-type] + + +class M2MClientTest(unittest.IsolatedAsyncioTestCase): + async def test_send_packet_requires_connected_socket(self) -> None: + client = M2MClient("wss://example.test/m2m/") + with self.assertRaises(RuntimeError): + await client.send_packet("ping", [b"abc"]) + + async def test_handle_packet_ping_responds_with_pong(self) -> None: + client = M2MClient("wss://example.test/m2m/") + client.send_packet = AsyncMock() + + await client._handle_packet(7, [b"nonce"]) + client.send_packet.assert_awaited_once_with("pong", [b"nonce"]) + + async def test_handle_packet_ping_without_payload_uses_empty_bytes(self) -> None: + client = M2MClient("wss://example.test/m2m/") + client.send_packet = AsyncMock() + + await client._handle_packet(7, []) + client.send_packet.assert_awaited_once_with("pong", [b""]) + + async def test_handle_packet_set_identity_decodes_utf8_bytes(self) -> None: + client = M2MClient("wss://example.test/m2m/") + self.assertFalse(client._identity_event.is_set()) + + await client._handle_packet(9, [b"device-123"]) + self.assertEqual(client.identity, "device-123") + self.assertTrue(client._identity_event.is_set()) + + async def test_handle_packet_notify_open_enqueues_port(self) -> None: + client = M2MClient("wss://example.test/m2m/") + await client._handle_packet(14, [12345]) + + port = await asyncio.wait_for(client.wait_for_channel_open(timeout=0.1), timeout=0.2) + self.assertEqual(port, 12345) + + async def test_handle_packet_route_normalizes_to_bytes(self) -> None: + client = M2MClient("wss://example.test/m2m/") + port = 9000 + queue = client.channel_queue(port) + + await client._handle_packet(6, [port, "hello"]) + self.assertEqual(await asyncio.wait_for(queue.get(), timeout=0.1), b"hello") + + await client._handle_packet(6, [port, b"world"]) + self.assertEqual(await asyncio.wait_for(queue.get(), timeout=0.1), b"world") + + async def test_handle_packet_notify_close_pushes_none(self) -> None: + client = M2MClient("wss://example.test/m2m/") + port = 7777 + queue = client.channel_queue(port) + + await client._handle_packet(19, [port]) + self.assertIsNone(await asyncio.wait_for(queue.get(), timeout=0.1)) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_remote_access_helpers.py b/tests/test_remote_access_helpers.py new file mode 100644 index 0000000..574905a --- /dev/null +++ b/tests/test_remote_access_helpers.py @@ -0,0 +1,85 @@ +from __future__ import annotations + +import asyncio +import io +import tempfile +import unittest +from pathlib import Path +from typing import Dict, Optional +from unittest.mock import patch + +from dataplicity_cli.remote_access import _detect_protocol, run_remote_file, run_single_command + + +class _FakeM2M: + def __init__(self) -> None: + self.queues: Dict[int, asyncio.Queue[Optional[bytes]]] = {} + self.sent_payloads: list[bytes] = [] + + def channel_queue(self, port: int) -> asyncio.Queue[Optional[bytes]]: + if port not in self.queues: + self.queues[port] = asyncio.Queue() + return self.queues[port] + + async def send_route(self, port: int, data: bytes) -> None: + _ = port + self.sent_payloads.append(data) + + async def close_channel(self, port: int) -> None: + _ = port + + +class _StdoutWithBuffer: + def __init__(self) -> None: + self.buffer = io.BytesIO() + + +class RemoteAccessHelpersTest(unittest.IsolatedAsyncioTestCase): + async def test_run_single_command_rejects_empty_command(self) -> None: + fake = _FakeM2M() + with self.assertRaises(RuntimeError): + await run_single_command(fake, 99, "") + + async def test_run_remote_file_requires_output_or_stdout(self) -> None: + fake = _FakeM2M() + queue = fake.channel_queue(1) + await queue.put(None) + with self.assertRaises(RuntimeError): + await run_remote_file(fake, 1, output_path=None, allow_stdout=False) + + async def test_run_remote_file_writes_binary_output(self) -> None: + fake = _FakeM2M() + queue = fake.channel_queue(2) + await queue.put(b"hello") + await queue.put(b" world") + await queue.put(None) + + with tempfile.TemporaryDirectory() as temp_dir: + output_path = Path(temp_dir) / "out.bin" + count = await run_remote_file(fake, 2, output_path=str(output_path), allow_stdout=False) + self.assertEqual(count, 11) + self.assertEqual(output_path.read_bytes(), b"hello world") + + async def test_run_remote_file_writes_to_stdout_buffer(self) -> None: + fake = _FakeM2M() + queue = fake.channel_queue(3) + await queue.put(b"abc") + await queue.put(None) + fake_stdout = _StdoutWithBuffer() + + with patch("dataplicity_cli.remote_access.sys.stdout", fake_stdout): + count = await run_remote_file(fake, 3, output_path=None, allow_stdout=True) + + self.assertEqual(count, 3) + self.assertEqual(fake_stdout.buffer.getvalue(), b"abc") + + async def test_detect_protocol_classifies_known_signatures(self) -> None: + self.assertEqual(_detect_protocol(b"GET / HTTP/1.1"), "HTTP request") + self.assertEqual(_detect_protocol(b"HTTP/1.1 200 OK"), "HTTP response") + self.assertEqual(_detect_protocol(b"SSH-2.0-OpenSSH_9.0"), "SSH") + self.assertEqual(_detect_protocol(bytes([0x16, 0x03, 0x03, 0x00])), "TLS") + self.assertIsNone(_detect_protocol(b"\x01\x02\x03")) + + +if __name__ == "__main__": + unittest.main()