Add VedAstro gateway status contract
This commit is contained in:
@@ -0,0 +1,70 @@
|
||||
#!/usr/bin/env python3
|
||||
"""China-friendly VedAstro-compatible gateway orchestration."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from typing import Any
|
||||
|
||||
|
||||
BACKEND_PRIORITY = ["self_host", "official", "cache", "queue", "local_fallback"]
|
||||
BOUNDARY_TEXT = "Users never call VedAstro directly; backend gateway owns cache, queue, and fallback."
|
||||
|
||||
|
||||
def _bool_env(name: str) -> bool:
|
||||
return os.environ.get(name, "").strip().lower() in {"1", "true", "yes", "on"}
|
||||
|
||||
|
||||
def _int_env(name: str, default: int = 0) -> int:
|
||||
raw = os.environ.get(name, "").strip()
|
||||
if not raw:
|
||||
return default
|
||||
try:
|
||||
return int(float(raw))
|
||||
except ValueError:
|
||||
return default
|
||||
|
||||
|
||||
def build_gateway_config() -> dict[str, Any]:
|
||||
mode = os.environ.get("VEDASTRO_GATEWAY_MODE", "local_first").strip() or "local_first"
|
||||
self_host = os.environ.get("VEDASTRO_SELF_HOST_ENDPOINT", "").strip()
|
||||
official = os.environ.get("VEDASTRO_API_ENDPOINT", "").strip()
|
||||
return {
|
||||
"mode": mode,
|
||||
"self_host_endpoint_configured": bool(self_host),
|
||||
"official_endpoint_configured": bool(official),
|
||||
"cache_ttl_seconds": _int_env("VEDASTRO_CACHE_TTL_SECONDS", 0),
|
||||
"queue_enabled": _bool_env("VEDASTRO_GATEWAY_QUEUE_ENABLED") or _bool_env("VEDASTRO_QUEUE_ENABLED"),
|
||||
"fail_open_local": os.environ.get("VEDASTRO_FAIL_OPEN_LOCAL", "1").strip().lower()
|
||||
not in {"0", "false", "no"},
|
||||
}
|
||||
|
||||
|
||||
def _active_backend(config: dict[str, Any]) -> str:
|
||||
if config["self_host_endpoint_configured"]:
|
||||
return "self_host"
|
||||
if config["official_endpoint_configured"] and _bool_env("VEDASTRO_ENABLE_NETWORK"):
|
||||
return "official"
|
||||
if config["cache_ttl_seconds"] > 0:
|
||||
return "cache"
|
||||
if config["queue_enabled"]:
|
||||
return "queue"
|
||||
return "local_fallback"
|
||||
|
||||
|
||||
def gateway_status() -> dict[str, Any]:
|
||||
config = build_gateway_config()
|
||||
return {
|
||||
"scope": "vedastro_gateway",
|
||||
"mode": config["mode"],
|
||||
"backend_priority": BACKEND_PRIORITY,
|
||||
"active_backend": _active_backend(config),
|
||||
"self_host_configured": config["self_host_endpoint_configured"],
|
||||
"official_configured": config["official_endpoint_configured"],
|
||||
"cache_ttl_seconds": config["cache_ttl_seconds"],
|
||||
"queue_enabled": config["queue_enabled"],
|
||||
"fail_open_local": config["fail_open_local"],
|
||||
"direct_browser_access_allowed": False,
|
||||
"frontend_secret_safe": True,
|
||||
"boundary": BOUNDARY_TEXT,
|
||||
}
|
||||
@@ -0,0 +1,39 @@
|
||||
from __future__ import annotations
|
||||
|
||||
|
||||
def test_gateway_status_defaults_to_local_first_cn_safe(monkeypatch):
|
||||
from scripts import vedastro_gateway
|
||||
|
||||
monkeypatch.delenv("VEDASTRO_GATEWAY_MODE", raising=False)
|
||||
monkeypatch.delenv("VEDASTRO_SELF_HOST_ENDPOINT", raising=False)
|
||||
monkeypatch.delenv("VEDASTRO_API_ENDPOINT", raising=False)
|
||||
monkeypatch.delenv("VEDASTRO_CACHE_TTL_SECONDS", raising=False)
|
||||
monkeypatch.delenv("VEDASTRO_GATEWAY_QUEUE_ENABLED", raising=False)
|
||||
monkeypatch.delenv("VEDASTRO_QUEUE_ENABLED", raising=False)
|
||||
|
||||
status = vedastro_gateway.gateway_status()
|
||||
|
||||
assert status["scope"] == "vedastro_gateway"
|
||||
assert status["mode"] == "local_first"
|
||||
assert status["direct_browser_access_allowed"] is False
|
||||
assert status["frontend_secret_safe"] is True
|
||||
assert status["backend_priority"] == ["self_host", "official", "cache", "queue", "local_fallback"]
|
||||
assert status["active_backend"] == "local_fallback"
|
||||
assert status["boundary"] == "Users never call VedAstro directly; backend gateway owns cache, queue, and fallback."
|
||||
|
||||
|
||||
def test_gateway_status_reports_cn_gateway_self_host(monkeypatch):
|
||||
from scripts import vedastro_gateway
|
||||
|
||||
monkeypatch.setenv("VEDASTRO_GATEWAY_MODE", "cn_gateway")
|
||||
monkeypatch.setenv("VEDASTRO_SELF_HOST_ENDPOINT", "https://jyotish-gateway.example.com/vedastro")
|
||||
monkeypatch.setenv("VEDASTRO_CACHE_TTL_SECONDS", "604800")
|
||||
monkeypatch.setenv("VEDASTRO_GATEWAY_QUEUE_ENABLED", "1")
|
||||
|
||||
status = vedastro_gateway.gateway_status()
|
||||
|
||||
assert status["mode"] == "cn_gateway"
|
||||
assert status["self_host_configured"] is True
|
||||
assert status["active_backend"] == "self_host"
|
||||
assert status["cache_ttl_seconds"] == 604800
|
||||
assert status["queue_enabled"] is True
|
||||
Reference in New Issue
Block a user