#!/usr/bin/env python3 """ Cove Data Protection API – Test Script ======================================= Verified working via Postman (2026-02-23). Uses confirmed column codes. Usage: python3 cove_api_test.py --username "api-user" --password "secret" Or via environment variables: COVE_USERNAME="api-user" COVE_PASSWORD="secret" python3 cove_api_test.py Optional: --url API endpoint (default: https://api.backup.management/jsonapi) --records Max records to fetch (default: 50) """ import argparse import json import os import sys from datetime import datetime, timezone import requests API_URL = "https://api.backup.management/jsonapi" # Session status codes (F00 / F15 / F09) SESSION_STATUS = { 1: "In process", 2: "Failed", 3: "Aborted", 5: "Completed", 6: "Interrupted", 7: "NotStarted", 8: "CompletedWithErrors", 9: "InProgressWithFaults", 10: "OverQuota", 11: "NoSelection", 12: "Restarted", } # Backupchecks status mapping STATUS_MAP = { 1: "Warning", # In process 2: "Error", # Failed 3: "Error", # Aborted 5: "Success", # Completed 6: "Error", # Interrupted 7: "Warning", # NotStarted 8: "Warning", # CompletedWithErrors 9: "Warning", # InProgressWithFaults 10: "Error", # OverQuota 11: "Warning", # NoSelection 12: "Warning", # Restarted } # Confirmed working columns (verified via Postman 2026-02-23) COLUMNS = [ "I1", "I18", "I8", "I78", "D09F00", "D09F09", "D09F15", "D09F08", "D1F00", "D1F15", "D10F00", "D10F15", "D11F00", "D11F15", "D19F00", "D19F15", "D20F00", "D20F15", "D5F00", "D5F15", "D23F00", "D23F15", ] # Datasource labels DATASOURCE_LABELS = { "D09": "Total", "D1": "Files & Folders", "D2": "System State", "D10": "VssMsSql (SQL Server)", "D11": "VssSharePoint", "D19": "M365 Exchange", "D20": "M365 OneDrive", "D5": "M365 SharePoint", "D23": "M365 Teams", } def _post(url: str, payload: dict, timeout: int = 30) -> dict: headers = {"Content-Type": "application/json"} resp = requests.post(url, json=payload, headers=headers, timeout=timeout) resp.raise_for_status() return resp.json() def login(url: str, username: str, password: str) -> tuple[str, int]: """Authenticate and return (visa, partner_id).""" payload = { "jsonrpc": "2.0", "id": "jsonrpc", "method": "Login", "params": { "username": username, "password": password, }, } data = _post(url, payload) if "error" in data: raise RuntimeError(f"Login failed: {data['error']}") visa = data.get("visa") if not visa: raise RuntimeError(f"No visa token in response: {data}") result = data.get("result", {}) partner_id = result.get("PartnerId") or result.get("result", {}).get("PartnerId") if not partner_id: raise RuntimeError(f"Could not find PartnerId in response: {data}") return visa, int(partner_id) def enumerate_statistics(url: str, visa: str, partner_id: int, columns: list[str], records: int = 50) -> dict: payload = { "jsonrpc": "2.0", "visa": visa, "id": "jsonrpc", "method": "EnumerateAccountStatistics", "params": { "query": { "PartnerId": partner_id, "StartRecordNumber": 0, "RecordsCount": records, "Columns": columns, } }, } return _post(url, payload) def fmt_ts(value) -> str: if not value: return "(none)" try: ts = int(value) if ts == 0: return "(none)" dt = datetime.fromtimestamp(ts, tz=timezone.utc) return dt.strftime("%Y-%m-%d %H:%M UTC") except (ValueError, TypeError, OSError): return str(value) def fmt_status(value) -> str: if value is None: return "(none)" try: code = int(value) bc = STATUS_MAP.get(code, "?") label = SESSION_STATUS.get(code, f"Unknown") return f"{code} ({label}) → {bc}" except (ValueError, TypeError): return str(value) def fmt_colorbar(value: str) -> str: if not value: return "(none)" icons = {"5": "✅", "8": "⚠️", "2": "❌", "1": "🔄", "0": "·"} return "".join(icons.get(c, c) for c in str(value)) def print_header(title: str) -> None: print() print("=" * 70) print(f" {title}") print("=" * 70) def run(url: str, username: str, password: str, records: int) -> None: print_header("Cove Data Protection API – Test") print(f" URL: {url}") print(f" Username: {username}") # Login print_header("Step 1: Login") visa, partner_id = login(url, username, password) print(f" ✅ Login OK") print(f" PartnerId: {partner_id}") print(f" Visa: {visa[:40]}...") # Fetch statistics print_header("Step 2: EnumerateAccountStatistics") print(f" Columns: {', '.join(COLUMNS)}") print(f" Records: max {records}") data = enumerate_statistics(url, visa, partner_id, COLUMNS, records) if "error" in data: err = data["error"] print(f" ❌ FAILED – error {err.get('code')}: {err.get('message')}") print(f" Data: {err.get('data')}") sys.exit(1) result = data.get("result") if result is None: print(" ⚠️ result is null (no accounts?)") sys.exit(0) # Result can be a list directly or wrapped accounts = result if isinstance(result, list) else result.get("Accounts", []) total = len(accounts) print(f" ✅ SUCCESS – {total} account(s) returned") # Per-account output print_header(f"Step 3: Account Details ({total} total)") for i, acc in enumerate(accounts): device_name = acc.get("I1", "(no name)") computer = acc.get("I18") or "(M365 tenant)" customer = acc.get("I8", "") active_ds = acc.get("I78", "") print(f"\n [{i+1}/{total}] {device_name}") print(f" Computer : {computer}") print(f" Customer : {customer}") print(f" Datasrc : {active_ds}") # Total (D09) d9_status = acc.get("D09F00") d9_last_ok = acc.get("D09F09") d9_last = acc.get("D09F15") d9_bar = acc.get("D09F08") print(f" Total:") print(f" Status : {fmt_status(d9_status)}") print(f" Last session: {fmt_ts(d9_last)}") print(f" Last success: {fmt_ts(d9_last_ok)}") print(f" 28-day bar : {fmt_colorbar(d9_bar)}") # Per-datasource (only if active) ds_pairs = [ ("D1", "D1F00", "D1F15"), ("D10", "D10F00", "D10F15"), ("D11", "D11F00", "D11F15"), ("D19", "D19F00", "D19F15"), ("D20", "D20F00", "D20F15"), ("D5", "D5F00", "D5F15"), ("D23", "D23F00", "D23F15"), ] for ds_code, f00_col, f15_col in ds_pairs: f00 = acc.get(f00_col) f15 = acc.get(f15_col) if f00 is None and f15 is None: continue label = DATASOURCE_LABELS.get(ds_code, ds_code) print(f" {label}:") print(f" Status : {fmt_status(f00)}") print(f" Last session: {fmt_ts(f15)}") # Summary print_header("Summary") status_counts: dict[str, int] = {} for acc in accounts: s = acc.get("D09F00") bc = STATUS_MAP.get(int(s), "Unknown") if s is not None else "No data" status_counts[bc] = status_counts.get(bc, 0) + 1 for status, count in sorted(status_counts.items()): icon = {"Success": "✅", "Warning": "⚠️", "Error": "❌"}.get(status, " ") print(f" {icon} {status}: {count}") print(f"\n Total accounts: {total}") print() def main() -> None: parser = argparse.ArgumentParser(description="Test Cove Data Protection API") parser.add_argument("--url", default=os.environ.get("COVE_URL", API_URL)) parser.add_argument("--username", default=os.environ.get("COVE_USERNAME", "")) parser.add_argument("--password", default=os.environ.get("COVE_PASSWORD", "")) parser.add_argument("--records", type=int, default=50, help="Max accounts to fetch") args = parser.parse_args() if not args.username or not args.password: print("Error: --username and --password are required.") print("Or set COVE_USERNAME and COVE_PASSWORD environment variables.") sys.exit(1) run(args.url, args.username, args.password, args.records) if __name__ == "__main__": main()