Jyotisha/tests/test_vedastro_runtime_ops.py

from __future__ import annotations

import importlib.metadata
import importlib.util
import json
from pathlib import Path

import pytest

ROOT = Path(__file__).resolve().parents[1]


def _load_patch_module():
    spec = importlib.util.spec_from_file_location(
        "patch_vedastro_update_check",
        ROOT / "deploy" / "patch_vedastro_update_check.py",
    )
    assert spec is not None and spec.loader is not None
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


def _pin_from_requirements() -> str:
    for line in (ROOT / "requirements.txt").read_text(encoding="utf-8").splitlines():
        if line.startswith("vedastro=="):
            return line.split("==", 1)[1].strip()
    raise AssertionError("requirements.txt does not pin vedastro")


def test_dockerfile_patches_vedastro_update_check_after_install() -> None:
    dockerfile = (ROOT / "deploy" / "railway-api.Dockerfile").read_text(encoding="utf-8")
    assert "python -m pip install -r requirements.txt" in dockerfile
    assert "python patch_vedastro_update_check.py" in dockerfile
    assert dockerfile.index("python -m pip install -r requirements.txt") < dockerfile.index(
        "python patch_vedastro_update_check.py"
    )


def test_patch_neutralizes_check_for_update_without_network(tmp_path: Path) -> None:
    patch_update_check = _load_patch_module().patch_update_check

    target = tmp_path / "update_check.py"
    target.write_text(
        "import requests\n"
        "def check_for_update(package_name='vedastro'):\n"
        "    requests.get('https://pypi.org/pypi/vedastro/json')\n",
        encoding="utf-8",
    )
    patch_update_check(target)
    namespace: dict[str, object] = {}
    exec(target.read_text(encoding="utf-8"), namespace)
    assert namespace["check_for_update"]("vedastro") is None


def test_patch_fails_closed_when_hook_is_missing(tmp_path: Path) -> None:
    patch_update_check = _load_patch_module().patch_update_check

    target = tmp_path / "update_check.py"
    target.write_text("def other():\n    return 1\n", encoding="utf-8")
    with pytest.raises(SystemExit):
        patch_update_check(target)


def test_runtime_vedastro_version_matches_requirements_pin() -> None:
    pin = _pin_from_requirements()
    try:
        installed = importlib.metadata.version("vedastro")
    except importlib.metadata.PackageNotFoundError:
        pytest.skip("vedastro is not installed in this environment")
    assert installed == pin


def test_foreground_free_tier_queue_fail_fast_when_budget_exhausted(monkeypatch) -> None:
    from scripts import vedastro_service_adapter as adapter

    sleep_calls: list[float] = []
    monotonic_values = iter([0.0, 0.0, 0.1, 0.1])

    monkeypatch.delenv("VEDASTRO_API_KEY", raising=False)
    monkeypatch.setenv("VEDASTRO_FREE_TIER_MAX_REQUESTS", "1")
    monkeypatch.setenv("VEDASTRO_FREE_TIER_WINDOW_SECONDS", "60")
    monkeypatch.setenv("VEDASTRO_FREE_TIER_WAIT_BUDGET_SECONDS", "0")
    monkeypatch.setenv("VEDASTRO_TIMEOUT_SECONDS", "20")
    monkeypatch.setenv("VEDASTRO_CACHE_TTL_SECONDS", "0")
    monkeypatch.setattr(adapter.time, "monotonic", lambda: next(monotonic_values))
    monkeypatch.setattr(adapter.time, "sleep", lambda seconds: sleep_calls.append(seconds))
    monkeypatch.setattr(
        adapter,
        "_post_json",
        lambda endpoint, preview: {"Status": "Pass", "Payload": {"ok": True}},
    )
    monkeypatch.setattr(adapter, "_FREE_TIER_REQUEST_TIMESTAMPS", [])

    first, _, _ = adapter._post_json_with_retry(
        "https://api.vedastro.org/api",
        {"operation": "range_scan", "official_request_profile": {"endpoint_path": "/Calculate/SearchEvents", "body": {"n": 1}}},
    )
    second, _, _ = adapter._post_json_with_retry(
        "https://api.vedastro.org/api",
        {"operation": "range_scan", "official_request_profile": {"endpoint_path": "/Calculate/SearchEvents", "body": {"n": 2}}},
    )

    assert first["Status"] == "Pass"
    assert second["Status"] == "Fail"
    assert second["source_metadata"]["degraded_reason"] == "free_tier_rate_limited"
    assert second["source_metadata"]["free_tier_rate_limit"]["mode"] == "free_tier_budget_exceeded"
    assert adapter._payload_status(second) == "rate_limited"
    assert sleep_calls == []


def test_wait_budget_never_exceeds_vedastro_timeout(monkeypatch) -> None:
    from scripts import vedastro_service_adapter as adapter

    monkeypatch.setenv("VEDASTRO_TIMEOUT_SECONDS", "8")
    monkeypatch.setenv("VEDASTRO_FREE_TIER_WAIT_BUDGET_SECONDS", "99")
    assert adapter._free_tier_wait_budget_seconds() == 8.0


def test_gateway_status_exposes_runtime_mode_without_secrets(monkeypatch) -> None:
    from scripts import vedastro_gateway

    monkeypatch.setenv("JYOTISH_SKIP_LOCAL_ENV", "1")
    monkeypatch.delenv("VEDASTRO_API_KEY", raising=False)
    monkeypatch.setenv("VEDASTRO_API_ENDPOINT", "https://api.vedastro.org/api")
    monkeypatch.setenv("VEDASTRO_FULL_SNAPSHOT_FANOUT_ENABLED", "1")
    monkeypatch.setenv("VEDASTRO_RANGE_SCAN_NETWORK_ENABLED", "1")
    monkeypatch.setenv("VEDASTRO_CACHE_TTL_SECONDS", "86400")
    monkeypatch.setenv("VEDASTRO_OFFICIAL_FULL_SNAPSHOT_CACHE_TTL_SECONDS", "120")

    status = vedastro_gateway.gateway_status()
    text = json.dumps(status)
    assert status["endpoint_configured"] is True
    assert status["credential_configured"] is False
    assert status["fanout_enabled"] is True
    assert status["range_scan_network_enabled"] is True
    assert status["free_tier_queue_active"] is True
    assert "sdk_version" in status
    assert status["cache_ttl_seconds"] == 86400
    assert status["official_full_snapshot_cache_ttl_seconds"] == 120
    assert "sk_live" not in text

    monkeypatch.setenv("VEDASTRO_API_KEY", "sk_live_should_never_appear")
    with_key = vedastro_gateway.gateway_status()
    dumped = json.dumps(with_key)
    assert with_key["credential_configured"] is True
    assert with_key["free_tier_queue_active"] is False
    assert "sk_live_should_never_appear" not in dumped