From 3bd8178464c2b488ac45f59d18c81a50fe4ceea7 Mon Sep 17 00:00:00 2001 From: Ivo Oskamp Date: Mon, 23 Feb 2026 10:21:01 +0100 Subject: [PATCH] Fix Cove importer: correct API payload format and response parsing - Login: use lowercase username/password params and id="jsonrpc" - Login: visa is at top-level of response (not inside result) - EnumerateAccountStatistics: use lowercase query param, RecordsCount instead of RecordCount, remove DisplayColumns (not needed) - _flatten_settings: Settings items are single-key dicts like {"D09F00": "5"}, not {Key: ..., Value: ...} - use dict.update() - _cove_enumerate: unwrap nested result and handle Accounts key - _process_account: AccountId is top-level field, not from Settings Co-Authored-By: Claude Sonnet 4.6 --- .../src/backend/app/cove_importer.py | 85 ++++++++++++------- 1 file changed, 53 insertions(+), 32 deletions(-) diff --git a/containers/backupchecks/src/backend/app/cove_importer.py b/containers/backupchecks/src/backend/app/cove_importer.py index caaf6a4..7b4740f 100644 --- a/containers/backupchecks/src/backend/app/cove_importer.py +++ b/containers/backupchecks/src/backend/app/cove_importer.py @@ -76,15 +76,20 @@ def _cove_login(url: str, username: str, password: str) -> tuple[str, int]: """ payload = { "jsonrpc": "2.0", + "id": "jsonrpc", "method": "Login", - "id": 1, "params": { - "Username": username, - "Password": password, + "username": username, + "password": password, }, } try: - resp = requests.post(url, json=payload, timeout=30) + resp = requests.post( + url, + json=payload, + headers={"Content-Type": "application/json"}, + timeout=30, + ) resp.raise_for_status() data = resp.json() except requests.RequestException as exc: @@ -92,21 +97,22 @@ def _cove_login(url: str, username: str, password: str) -> tuple[str, int]: except ValueError as exc: raise CoveImportError(f"Cove login response is not valid JSON: {exc}") from exc - result = data.get("result") or {} - if not result: - error = data.get("error") or {} - raise CoveImportError(f"Cove login failed: {error.get('message', 'unknown error')}") + if "error" in data and data["error"]: + error = data["error"] + msg = error.get("message") or str(error) if isinstance(error, dict) else str(error) + raise CoveImportError(f"Cove login failed: {msg}") - visa = result.get("Visa") or "" + # Visa is returned at the top level of the response (not inside result) + visa = data.get("visa") or "" if not visa: - raise CoveImportError("Cove login succeeded but no Visa token returned") + raise CoveImportError("Cove login succeeded but no visa token returned") - # Partner/account info may be nested - account_info = result.get("Accounts", [{}])[0] if result.get("Accounts") else {} + # PartnerId is inside result + result = data.get("result") or {} partner_id = ( - account_info.get("PartnerId") + result.get("PartnerId") or result.get("PartnerID") - or result.get("PartnerId") + or result.get("result", {}).get("PartnerId") or 0 ) @@ -126,21 +132,25 @@ def _cove_enumerate( """ payload = { "jsonrpc": "2.0", - "method": "EnumerateAccountStatistics", - "id": 2, "visa": visa, + "id": "jsonrpc", + "method": "EnumerateAccountStatistics", "params": { - "Query": { + "query": { "PartnerId": partner_id, - "DisplayColumns": COVE_COLUMNS, "StartRecordNumber": start, - "RecordCount": count, + "RecordsCount": count, "Columns": COVE_COLUMNS, } }, } try: - resp = requests.post(url, json=payload, timeout=60) + resp = requests.post( + url, + json=payload, + headers={"Content-Type": "application/json"}, + timeout=60, + ) resp.raise_for_status() data = resp.json() except requests.RequestException as exc: @@ -148,18 +158,32 @@ def _cove_enumerate( except ValueError as exc: raise CoveImportError(f"Cove EnumerateAccountStatistics response is not valid JSON: {exc}") from exc - result = data.get("result") or {} - if not result: - error = data.get("error") or {} - raise CoveImportError(f"Cove EnumerateAccountStatistics failed: {error.get('message', 'unknown error')}") + if "error" in data and data["error"]: + error = data["error"] + msg = error.get("message") or str(error) if isinstance(error, dict) else str(error) + raise CoveImportError(f"Cove EnumerateAccountStatistics failed: {msg}") - return result.get("result", []) or [] + result = data.get("result") + if result is None: + return [] + + # Unwrap possible nested result (same as test script) + if isinstance(result, dict) and "result" in result: + result = result["result"] + + # Accounts can be a list directly or wrapped in an "Accounts" key + if isinstance(result, list): + return result + if isinstance(result, dict): + return result.get("Accounts", []) or [] + return [] def _flatten_settings(account: dict) -> dict: """Convert the Settings array in an account dict to a flat key→value dict. - Cove returns settings as a list of {Key, Value} objects. + Cove returns settings as a list of single-key dicts, e.g.: + [{"D09F00": "5"}, {"I1": "device name"}, ...] This flattens them so we can do `flat['D09F00']` instead of iterating. """ flat: dict[str, Any] = {} @@ -167,10 +191,7 @@ def _flatten_settings(account: dict) -> dict: if isinstance(settings_list, list): for item in settings_list: if isinstance(item, dict): - key = item.get("Key") or item.get("key") - value = item.get("Value") if "Value" in item else item.get("value") - if key is not None: - flat[str(key)] = value + flat.update(item) return flat @@ -274,8 +295,8 @@ def _process_account(account: dict, settings) -> bool: flat = _flatten_settings(account) - # AccountId – try both the top-level field and Settings - account_id = account.get("AccountId") or account.get("AccountID") or flat.get("I18") + # AccountId is a top-level field (not in Settings) + account_id = account.get("AccountId") or account.get("AccountID") if not account_id: return False try: