112 lines
3.7 KiB
Python
112 lines
3.7 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import http.cookiejar
|
|
import json
|
|
import urllib.error
|
|
import urllib.request
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
def request_json(
|
|
opener: urllib.request.OpenerDirector,
|
|
url: str,
|
|
payload: dict[str, Any] | None = None,
|
|
) -> tuple[int, Any]:
|
|
data = None
|
|
headers = {"Accept": "application/json"}
|
|
if payload is not None:
|
|
data = json.dumps(payload, ensure_ascii=False).encode("utf-8")
|
|
headers["Content-Type"] = "application/json"
|
|
request = urllib.request.Request(url, data=data, headers=headers)
|
|
try:
|
|
with opener.open(request, timeout=90) as response:
|
|
return response.status, json.loads(response.read().decode("utf-8"))
|
|
except urllib.error.HTTPError as exc:
|
|
return exc.code, json.loads(exc.read().decode("utf-8"))
|
|
|
|
|
|
def session(base_url: str, username: str, password: str) -> urllib.request.OpenerDirector:
|
|
opener = urllib.request.build_opener(
|
|
urllib.request.HTTPCookieProcessor(http.cookiejar.CookieJar())
|
|
)
|
|
status, body = request_json(
|
|
opener,
|
|
f"{base_url.rstrip('/')}/api/auth/login",
|
|
{"username": username, "password": password},
|
|
)
|
|
if status != 200 or not body.get("ok"):
|
|
raise RuntimeError(f"Login failed for {base_url}: HTTP {status} {body}")
|
|
return opener
|
|
|
|
|
|
def digest(value: Any) -> str:
|
|
content = json.dumps(
|
|
value, ensure_ascii=False, sort_keys=True, separators=(",", ":")
|
|
).encode("utf-8")
|
|
return hashlib.sha256(content).hexdigest()
|
|
|
|
|
|
def comparable(value: Any) -> Any:
|
|
if isinstance(value, dict):
|
|
return {
|
|
key: comparable(item)
|
|
for key, item in value.items()
|
|
if key != "request_id"
|
|
}
|
|
if isinstance(value, list):
|
|
return [comparable(item) for item in value]
|
|
return value
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description="Compare authenticated preservation APIs")
|
|
parser.add_argument("--original", required=True)
|
|
parser.add_argument("--migrated", required=True)
|
|
parser.add_argument("--username", required=True)
|
|
parser.add_argument("--password", required=True)
|
|
parser.add_argument("--output", type=Path, required=True)
|
|
parser.add_argument("endpoints", nargs="+")
|
|
args = parser.parse_args()
|
|
|
|
original = session(args.original, args.username, args.password)
|
|
migrated = session(args.migrated, args.username, args.password)
|
|
rows = []
|
|
all_equal = True
|
|
for endpoint in args.endpoints:
|
|
original_status, original_body = request_json(
|
|
original, f"{args.original.rstrip('/')}{endpoint}"
|
|
)
|
|
migrated_status, migrated_body = request_json(
|
|
migrated, f"{args.migrated.rstrip('/')}{endpoint}"
|
|
)
|
|
original_comparable = comparable(original_body)
|
|
migrated_comparable = comparable(migrated_body)
|
|
equal = original_status == migrated_status and original_comparable == migrated_comparable
|
|
all_equal = all_equal and equal
|
|
rows.append(
|
|
{
|
|
"endpoint": endpoint,
|
|
"original_status": original_status,
|
|
"migrated_status": migrated_status,
|
|
"original_sha256": digest(original_comparable),
|
|
"migrated_sha256": digest(migrated_comparable),
|
|
"equal": equal,
|
|
}
|
|
)
|
|
|
|
result = {"all_equal": all_equal, "endpoints": rows}
|
|
args.output.parent.mkdir(parents=True, exist_ok=True)
|
|
args.output.write_text(
|
|
json.dumps(result, ensure_ascii=False, indent=2) + "\n", encoding="utf-8"
|
|
)
|
|
print(json.dumps(result, ensure_ascii=False, indent=2))
|
|
if not all_equal:
|
|
raise SystemExit(1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|