288 lines
9.2 KiB
Python
288 lines
9.2 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timezone
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from urllib.parse import quote
|
|
|
|
import requests
|
|
from dotenv import load_dotenv
|
|
|
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parents[1]
|
|
LOGS_DIR = PROJECT_ROOT / "logs"
|
|
DEFAULT_REPORT_PATH = LOGS_DIR / "foxylink_probe_report.json"
|
|
|
|
|
|
def utc_now_iso() -> str:
|
|
return datetime.now(timezone.utc).isoformat()
|
|
|
|
|
|
def _as_bool(raw: str | None, default: bool) -> bool:
|
|
if raw is None:
|
|
return default
|
|
return raw.strip().lower() in {"1", "true", "yes", "on"}
|
|
|
|
|
|
def _normalize_path(path: str | None, default: str) -> str:
|
|
value = (path or default).strip()
|
|
if not value:
|
|
value = default
|
|
if not value.startswith("/"):
|
|
value = "/" + value
|
|
if not value.endswith("/"):
|
|
value = value + "/"
|
|
return value
|
|
|
|
|
|
def _resolve_path(raw: str) -> Path:
|
|
path = Path(raw)
|
|
if path.is_absolute():
|
|
return path
|
|
return PROJECT_ROOT / path
|
|
|
|
|
|
def _load_payload(payload_file: str | None, fallback_json: str) -> tuple[Any, str, str]:
|
|
payload_source = "env:ONEC_FOXY_PAYLOAD_JSON"
|
|
raw = fallback_json
|
|
if payload_file:
|
|
payload_path = _resolve_path(payload_file)
|
|
payload_source = f"file:{payload_path}"
|
|
raw = payload_path.read_text(encoding="utf-8")
|
|
|
|
try:
|
|
payload_obj = json.loads(raw)
|
|
if isinstance(payload_obj, str):
|
|
body = payload_obj
|
|
else:
|
|
body = json.dumps(payload_obj, ensure_ascii=False)
|
|
return payload_obj, body, payload_source
|
|
except json.JSONDecodeError:
|
|
return raw, raw, payload_source
|
|
|
|
|
|
def _preview(value: str, limit: int = 2000) -> str:
|
|
if len(value) <= limit:
|
|
return value
|
|
return value[:limit] + "...<truncated>"
|
|
|
|
|
|
def _classify(status_code: int | None, error: str | None) -> str:
|
|
if error:
|
|
return "network_error"
|
|
if status_code is None:
|
|
return "unknown_error"
|
|
if status_code == 200:
|
|
return "reachable"
|
|
if status_code in {401, 403}:
|
|
return "auth_failed"
|
|
if status_code == 404:
|
|
return "endpoint_not_found_or_not_published"
|
|
if status_code >= 500:
|
|
return "service_error"
|
|
return "unexpected_status"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class FoxyProbeSettings:
|
|
base_url: str
|
|
infobase: str
|
|
username: str
|
|
password: str
|
|
timeout: int
|
|
verify_tls: bool
|
|
foxy_path: str
|
|
exchange: str
|
|
operation: str
|
|
message_type: str
|
|
payload_json: str
|
|
reply_to: str
|
|
app_id: str
|
|
correlation_id: str
|
|
|
|
@property
|
|
def endpoint_url(self) -> str:
|
|
base = self.base_url.rstrip("/")
|
|
infobase = self.infobase.strip().strip("/")
|
|
path = self.foxy_path
|
|
exchange = quote(self.exchange.strip(), safe="")
|
|
operation = quote(self.operation.strip(), safe="")
|
|
message_type = quote(self.message_type.strip().upper(), safe="")
|
|
suffix = f"{path}v1/{exchange}/{operation}/{message_type}"
|
|
if infobase:
|
|
return f"{base}/{infobase}{suffix}"
|
|
return f"{base}{suffix}"
|
|
|
|
|
|
def _load_settings(args: argparse.Namespace) -> FoxyProbeSettings:
|
|
env_file = PROJECT_ROOT / ".env"
|
|
if env_file.exists():
|
|
load_dotenv(env_file)
|
|
|
|
base_url = os.getenv("ONEC_BASE_URL", "http://localhost").strip()
|
|
infobase = os.getenv("ONEC_INFOBASE", "AccountingBase").strip()
|
|
username = os.getenv("ONEC_USERNAME", "").strip()
|
|
password = os.getenv("ONEC_PASSWORD", "")
|
|
timeout = int(os.getenv("ONEC_TIMEOUT", "30").strip())
|
|
verify_tls = _as_bool(os.getenv("ONEC_VERIFY_TLS"), default=False)
|
|
|
|
foxy_path = _normalize_path(
|
|
args.foxy_path or os.getenv("ONEC_FOXY_PATH", "/hs/AppEndpoint/"),
|
|
default="/hs/AppEndpoint/",
|
|
)
|
|
exchange = (args.exchange or os.getenv("ONEC_FOXY_EXCHANGE", "Self")).strip()
|
|
operation = (args.operation or os.getenv("ONEC_FOXY_OPERATION", "Query")).strip()
|
|
message_type = (args.message_type or os.getenv("ONEC_FOXY_TYPE", "SYNC")).strip()
|
|
payload_json = os.getenv("ONEC_FOXY_PAYLOAD_JSON", "{}")
|
|
reply_to = (args.reply_to or os.getenv("ONEC_FOXY_REPLYTO", "")).strip()
|
|
app_id = (args.app_id or os.getenv("ONEC_FOXY_APPID", "")).strip()
|
|
correlation_id = (args.correlation_id or os.getenv("ONEC_FOXY_CORRELATION_ID", "")).strip()
|
|
|
|
return FoxyProbeSettings(
|
|
base_url=base_url,
|
|
infobase=infobase,
|
|
username=username,
|
|
password=password,
|
|
timeout=timeout,
|
|
verify_tls=verify_tls,
|
|
foxy_path=foxy_path,
|
|
exchange=exchange,
|
|
operation=operation,
|
|
message_type=message_type,
|
|
payload_json=payload_json,
|
|
reply_to=reply_to,
|
|
app_id=app_id,
|
|
correlation_id=correlation_id,
|
|
)
|
|
|
|
|
|
def _parse_args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(
|
|
description="Probe FoxyLink HTTP endpoint and write diagnostic report."
|
|
)
|
|
parser.add_argument("--exchange", help="Exchange description used in URL.")
|
|
parser.add_argument("--operation", help="Operation description used in URL.")
|
|
parser.add_argument("--message-type", help="SYNC or ASYNC.")
|
|
parser.add_argument("--foxy-path", help="HTTP service path, default /hs/AppEndpoint/.")
|
|
parser.add_argument("--payload-file", help="Path to JSON payload file.")
|
|
parser.add_argument("--reply-to", help="Optional REPLYTO header.")
|
|
parser.add_argument("--app-id", help="Optional APPID header.")
|
|
parser.add_argument("--correlation-id", help="Optional CORRELATIONID header.")
|
|
parser.add_argument(
|
|
"--output",
|
|
default=str(DEFAULT_REPORT_PATH),
|
|
help="Output report path (relative to project root or absolute).",
|
|
)
|
|
parser.add_argument(
|
|
"--strict",
|
|
action="store_true",
|
|
help="Return non-zero code when endpoint is not reachable (HTTP 200).",
|
|
)
|
|
return parser.parse_args()
|
|
|
|
|
|
def main() -> int:
|
|
args = _parse_args()
|
|
settings = _load_settings(args)
|
|
|
|
payload_obj, body, payload_source = _load_payload(
|
|
args.payload_file, settings.payload_json
|
|
)
|
|
|
|
headers: dict[str, str] = {
|
|
"Accept": "application/json",
|
|
"Content-Type": "application/json; charset=utf-8",
|
|
}
|
|
if settings.reply_to:
|
|
headers["REPLYTO"] = settings.reply_to
|
|
if settings.app_id:
|
|
headers["APPID"] = settings.app_id
|
|
if settings.correlation_id:
|
|
headers["CORRELATIONID"] = settings.correlation_id
|
|
|
|
response_status_code: int | None = None
|
|
response_headers: dict[str, str] = {}
|
|
response_text = ""
|
|
error_message: str | None = None
|
|
elapsed_ms: int | None = None
|
|
|
|
try:
|
|
session = requests.Session()
|
|
if settings.username:
|
|
session.auth = (settings.username, settings.password)
|
|
|
|
response = session.post(
|
|
settings.endpoint_url,
|
|
data=body.encode("utf-8"),
|
|
headers=headers,
|
|
timeout=settings.timeout,
|
|
verify=settings.verify_tls,
|
|
)
|
|
response_status_code = response.status_code
|
|
response_headers = dict(response.headers)
|
|
response_text = response.text or ""
|
|
elapsed_ms = int(response.elapsed.total_seconds() * 1000)
|
|
except requests.RequestException as exc:
|
|
error_message = f"{exc.__class__.__name__}: {exc}"
|
|
|
|
classification = _classify(response_status_code, error_message)
|
|
success = classification == "reachable"
|
|
|
|
report = {
|
|
"generated_at": utc_now_iso(),
|
|
"request": {
|
|
"url": settings.endpoint_url,
|
|
"method": "POST",
|
|
"timeout_sec": settings.timeout,
|
|
"verify_tls": settings.verify_tls,
|
|
"exchange": settings.exchange,
|
|
"operation": settings.operation,
|
|
"message_type": settings.message_type.upper(),
|
|
"headers": headers,
|
|
"payload_source": payload_source,
|
|
"payload_preview": _preview(body),
|
|
"payload_is_json": isinstance(payload_obj, (dict, list, int, float, bool, type(None))),
|
|
},
|
|
"response": {
|
|
"status_code": response_status_code,
|
|
"elapsed_ms": elapsed_ms,
|
|
"headers": response_headers,
|
|
"body_preview": _preview(response_text),
|
|
"body_length": len(response_text),
|
|
},
|
|
"classification": classification,
|
|
"success": success,
|
|
"error": error_message,
|
|
"notes": {
|
|
"reachable_requires_http_200": True,
|
|
"diagnostic_hint": (
|
|
"HTTP 404 usually means FoxyLink HTTP service is not merged into the "
|
|
"published infobase or not published with HTTP services."
|
|
),
|
|
},
|
|
}
|
|
|
|
output_path = _resolve_path(args.output)
|
|
output_path.parent.mkdir(parents=True, exist_ok=True)
|
|
output_path.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
|
|
print(f"[ok] saved: {output_path}")
|
|
print(
|
|
"[ok] foxylink_probe: "
|
|
f"classification={classification}, "
|
|
f"status={response_status_code}, "
|
|
f"url={settings.endpoint_url}"
|
|
)
|
|
|
|
if args.strict and not success:
|
|
return 2
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|