153 lines
4.7 KiB
Python
153 lines
4.7 KiB
Python
#!/usr/bin/env python3
|
|
"""REST API client tool for ScadaLink test infrastructure."""
|
|
|
|
import argparse
|
|
import json
|
|
import sys
|
|
|
|
import requests
|
|
|
|
|
|
DEFAULT_URL = "http://localhost:5200"
|
|
DEFAULT_API_KEY = "scadalink-test-key-1"
|
|
|
|
|
|
def cmd_check(args):
|
|
"""Test API server connectivity and report status."""
|
|
headers = {"X-API-Key": args.api_key}
|
|
|
|
# Test health endpoint (no auth required)
|
|
try:
|
|
resp = requests.get(f"{args.url}/api/Ping", timeout=5)
|
|
ping_ok = resp.status_code == 200
|
|
except Exception as e:
|
|
ping_ok = False
|
|
ping_err = str(e)
|
|
|
|
print(f"Ping ({args.url}/api/Ping): {'OK' if ping_ok else 'FAILED - ' + ping_err}")
|
|
|
|
if not ping_ok:
|
|
sys.exit(1)
|
|
|
|
# Test authenticated endpoint
|
|
try:
|
|
resp = requests.post(f"{args.url}/api/GetStatus", json={}, headers=headers, timeout=5)
|
|
if resp.status_code == 200:
|
|
data = resp.json()
|
|
print(f"Status: {data.get('status', 'unknown')}")
|
|
print(f"Uptime: {data.get('uptime', 0)}s")
|
|
auth_ok = True
|
|
elif resp.status_code == 401:
|
|
print("Auth: FAILED - invalid API key")
|
|
auth_ok = False
|
|
else:
|
|
print(f"Status check: HTTP {resp.status_code}")
|
|
auth_ok = False
|
|
except Exception as e:
|
|
print(f"Status check: FAILED - {e}")
|
|
auth_ok = False
|
|
|
|
# List available methods
|
|
try:
|
|
resp = requests.get(f"{args.url}/api/methods", headers=headers, timeout=5)
|
|
if resp.status_code == 200:
|
|
methods = resp.json().get("methods", [])
|
|
print(f"Available methods: {len(methods)}")
|
|
except Exception:
|
|
pass
|
|
|
|
if ping_ok and auth_ok:
|
|
print("\nREST API server is healthy.")
|
|
else:
|
|
sys.exit(1)
|
|
|
|
|
|
def cmd_call(args):
|
|
"""Call an API method."""
|
|
headers = {"X-API-Key": args.api_key, "Content-Type": "application/json"}
|
|
url = f"{args.url}/api/{args.method}"
|
|
|
|
params = {}
|
|
if args.params:
|
|
try:
|
|
params = json.loads(args.params)
|
|
except json.JSONDecodeError as e:
|
|
print(f"Error: invalid JSON params: {e}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
try:
|
|
if args.method == "Ping":
|
|
resp = requests.get(url, headers=headers, timeout=args.timeout)
|
|
else:
|
|
resp = requests.post(url, json=params, headers=headers, timeout=args.timeout)
|
|
except requests.exceptions.Timeout:
|
|
print(f"Error: request timed out after {args.timeout}s", file=sys.stderr)
|
|
sys.exit(1)
|
|
except Exception as e:
|
|
print(f"Error: {e}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
print(f"HTTP {resp.status_code}")
|
|
try:
|
|
print(json.dumps(resp.json(), indent=2))
|
|
except Exception:
|
|
print(resp.text)
|
|
|
|
if resp.status_code >= 400:
|
|
sys.exit(1)
|
|
|
|
|
|
def cmd_methods(args):
|
|
"""List all available API methods."""
|
|
headers = {"X-API-Key": args.api_key}
|
|
|
|
try:
|
|
resp = requests.get(f"{args.url}/api/methods", headers=headers, timeout=5)
|
|
if resp.status_code != 200:
|
|
print(f"Error: HTTP {resp.status_code}", file=sys.stderr)
|
|
sys.exit(1)
|
|
except Exception as e:
|
|
print(f"Error: {e}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
methods = resp.json().get("methods", [])
|
|
print(f"{'Method':<25} {'HTTP':<6} {'Path':<30} {'Description'}")
|
|
print("-" * 95)
|
|
|
|
for m in methods:
|
|
print(f"{m['method']:<25} {m['httpMethod']:<6} {m['path']:<30} {m.get('description', '')}")
|
|
if m.get("params"):
|
|
params_str = ", ".join(f"{k}: {v}" for k, v in m["params"].items())
|
|
print(f" {'Params:':<23} {params_str}")
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="REST API client tool for ScadaLink test infrastructure")
|
|
parser.add_argument("--url", default=DEFAULT_URL, help=f"API base URL (default: {DEFAULT_URL})")
|
|
parser.add_argument("--api-key", default=DEFAULT_API_KEY, help=f"API key (default: {DEFAULT_API_KEY})")
|
|
|
|
sub = parser.add_subparsers(dest="command", required=True)
|
|
|
|
sub.add_parser("check", help="Test API connectivity and report status")
|
|
|
|
call_p = sub.add_parser("call", help="Call an API method")
|
|
call_p.add_argument("--method", required=True, help="Method name (e.g. Add, GetProductionReport)")
|
|
call_p.add_argument("--params", help="JSON params (e.g. '{\"a\": 2, \"b\": 3}')")
|
|
call_p.add_argument("--timeout", type=float, default=30, help="Request timeout in seconds (default: 30)")
|
|
|
|
sub.add_parser("methods", help="List all available API methods")
|
|
|
|
args = parser.parse_args()
|
|
|
|
commands = {
|
|
"check": cmd_check,
|
|
"call": cmd_call,
|
|
"methods": cmd_methods,
|
|
}
|
|
|
|
commands[args.command](args)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|