Ivo Oskamp
dde2ccbb5d
API returns Settings as list of single-key dicts, not a flat dict. Also fixes AccountId display and status summary parsing. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
311 lines
9.5 KiB
Python
311 lines
9.5 KiB
Python
#!/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, debug: bool = False) -> 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 debug:
|
||
print(f"\n RAW response (first 2000 chars):")
|
||
print(json.dumps(data, indent=2)[:2000])
|
||
|
||
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 – raw response:")
|
||
print(json.dumps(data, indent=2)[:1000])
|
||
sys.exit(0)
|
||
|
||
if debug:
|
||
print(f"\n result type: {type(result).__name__}")
|
||
if isinstance(result, dict):
|
||
print(f" result keys: {list(result.keys())}")
|
||
|
||
# Unwrap possible nested result
|
||
if isinstance(result, dict) and "result" in result:
|
||
result = result["result"]
|
||
|
||
# Result can be a list directly or wrapped in Accounts key
|
||
accounts = result if isinstance(result, list) else result.get("Accounts", []) if isinstance(result, dict) else []
|
||
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):
|
||
# Settings is a list of single-key dicts: [{"D09F00": "5"}, {"I1": "name"}, ...]
|
||
# Flatten to a single dict for easy lookup.
|
||
s: dict = {}
|
||
for item in acc.get("Settings", []):
|
||
s.update(item)
|
||
|
||
account_id = acc.get("AccountId", "?")
|
||
device_name = s.get("I1", "(no name)")
|
||
computer = s.get("I18") or "(M365 tenant)"
|
||
customer = s.get("I8", "")
|
||
active_ds = s.get("I78", "")
|
||
|
||
print(f"\n [{i+1}/{total}] {device_name} (AccountId: {account_id})")
|
||
print(f" Computer : {computer}")
|
||
print(f" Customer : {customer}")
|
||
print(f" Datasrc : {active_ds}")
|
||
|
||
# Total (D09)
|
||
print(f" Total:")
|
||
print(f" Status : {fmt_status(s.get('D09F00'))}")
|
||
print(f" Last session: {fmt_ts(s.get('D09F15'))}")
|
||
print(f" Last success: {fmt_ts(s.get('D09F09'))}")
|
||
print(f" 28-day bar : {fmt_colorbar(s.get('D09F08'))}")
|
||
|
||
# Per-datasource (only if present in response)
|
||
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 = s.get(f00_col)
|
||
f15 = s.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:
|
||
flat: dict = {}
|
||
for item in acc.get("Settings", []):
|
||
flat.update(item)
|
||
raw = flat.get("D09F00")
|
||
bc = STATUS_MAP.get(int(raw), "Unknown") if raw 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")
|
||
parser.add_argument("--debug", action="store_true", help="Print raw API responses")
|
||
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, args.debug)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|