diff --git a/scripts/vedastro_gateway.py b/scripts/vedastro_gateway.py new file mode 100644 index 00000000..9bebacb3 --- /dev/null +++ b/scripts/vedastro_gateway.py @@ -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, + } diff --git a/tests/test_vedastro_gateway.py b/tests/test_vedastro_gateway.py new file mode 100644 index 00000000..df59f203 --- /dev/null +++ b/tests/test_vedastro_gateway.py @@ -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